참고

jdk 1.6.0_22 의 release note(http://www.oracle.com/technetwork/java/javase/6u22releasenotes-176121.html)를 봐도 특별한 내용은 없다.  그러나, 1.6.0_23의 버그 fix (http://www.oracle.com/technetwork/java/javase/2col/6u23bugfixes-191074.html) 의 내용 중 “6728542”은 관련 내용이 있다.  그 내용(http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6728542)은 다음과 같다. 아마도 이 부분을 수정한 것이 아닌가 싶다.

Epoll implementation of java.nio.channels.Selector
 (EPollSelectorProvider) doesn't correctly packs
 epoll_event struct for all platforms.

abstract class SelectorImpl   extends AbstractSelector {
…
    public int select(long timeout)  throws IOException
    {
        if (timeout < 0)
            throw new IllegalArgumentException("Negative timeout");
         return lockAndDoSelect((timeout == 0) ? -1 : timeout);
    }

    private int lockAndDoSelect(long timeout) throws IOException {
    synchronized (this) {
        if (!isOpen())
        throw new ClosedSelectorException();
        synchronized (publicKeys) {
        synchronized (publicSelectedKeys) {
            return doSelect(timeout);
        }
        }
        }
    }

   protected abstract int doSelect(long timeout) throws IOException;
}

// The poll object
EPollArrayWrapper pollWrapper;

protected int doSelect(long timeout)  throws IOException
    {
        if (closed)
            throw new ClosedSelectorException();
        processDeregisterQueue();
        try {
            begin();
           pollWrapper.poll(timeout);
        } finally {
            end();
        }
        processDeregisterQueue();
        int numKeysUpdated = updateSelectedKeys();
        if (pollWrapper.interrupted()) {
            // Clear the wakeup pipe
            pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
            synchronized (interruptLock) {
                pollWrapper.clearInterrupted();
                IOUtil.drain(fd0);
                interruptTriggered = false;
            }
        }
        return numKeysUpdated;
    }

}

/**
* Manipulates a native array of epoll_event structs on Linux:
*
* typedef union epoll_data {
*     void *ptr;
*     int fd;
*     __uint32_t u32;
*     __uint64_t u64;
*  } epoll_data_t;
*
* struct epoll_event {
*     __uint32_t events; 
*     epoll_data_t data;
* };
*/

int poll(long timeout) throws IOException {
    updateRegistrations();
    updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
    for (int i=0; i<updated; i++) {
        if (getDescriptor(i) == incomingInterruptFD) {
            interruptedIndex = i;
            interrupted = true;
            break;
        }
    }
    return updated;
}

private native int epollWait(long pollAddress, int numfds, long timeout,
                             int epfd) throws IOException;
}

/* epoll_wait(2) man page */

typedef union epoll_data {
    void *ptr;
    int fd;
    __uint32_t u32;
    __uint64_t u64;
} epoll_data_t;

struct epoll_event {
    __uint32_t events;  /* Epoll events */
    epoll_data_t data;  /* User data variable */
} __attribute__ ((__packed__));

JNIEXPORT jint JNICALL
Java_sun_nio_ch_EPollArrayWrapper_epollWait(JNIEnv *env, jobject this,
                                            jlong address, jint numfds,
                                            jlong timeout, jint epfd)
{
    struct epoll_event *events = jlong_to_ptr(address);
    int res;

    if (timeout <= 0) {           /* Indefinite or no wait */
        RESTARTABLE((*epoll_wait_func)(epfd, events, numfds, timeout), res);
    } else {                      /* Bounded wait; bounded restarts */
        res = iepoll(epfd, events, numfds, timeout);
    }
   
    if (res < 0) {
        JNU_ThrowIOExceptionWithLastError(env, "epoll_wait failed");
    }
    return res;
}

static int iepoll(int epfd, struct epoll_event *events, int numfds, jlong timeout)
{
    jlong start, now;
    int remaining = timeout;
    struct timeval t;
    int diff;

    gettimeofday(&t, NULL);
    start = t.tv_sec * 1000 + t.tv_usec / 1000;

    for (;;) {
        int res = (*epoll_wait_func)(epfd, events, numfds, timeout);
        if (res < 0 && errno == EINTR) {
            if (remaining >= 0) {
                gettimeofday(&t, NULL);
                now = t.tv_sec * 1000 + t.tv_usec / 1000;
                diff = now - start;
                remaining -= diff;
                if (diff < 0 || remaining <= 0) {
                    return 0;
                }
                start = now;
            }
        } else {
            return res;
        }
    }
}

public abstract class AbstractPollingIoProcessor<T extends AbstractIoSession> implements IoProcessor<T> {

private static final long SELECT_TIMEOUT = 1000L;

private class Processor implements Runnable {
        public void run() {
            int nSessions = 0;
            lastIdleCheckTime = System.currentTimeMillis();

            for (;;) {
                try {
                    int selected = select(SELECT_TIMEOUT);

                    nSessions += handleNewSessions();
                    updateTrafficMask();

                     if (selected > 0) {
                        process();
                    }

                    long currentTime = System.currentTimeMillis();
                    flush(currentTime);
                    nSessions -= removeSessions();
                    notifyIdleSessions(currentTime);

                    if (nSessions == 0) {
                        synchronized (lock) {
                            if (newSessions.isEmpty() && isSelectorEmpty()) {
                                processor = null;
                                break;
                            }
                        }
                    }

                     if (isDisposing()) {
                        for (Iterator<T> i = allSessions(); i.hasNext();) {
                            scheduleRemove(i.next());
                        }
                        wakeup();
                     }
                } catch (Throwable t) {
                   …..

                }
            }

/**
* poll those sessions for the given timeout
* @param timeout milliseconds before the call timeout if no event appear
* @return The number of session ready for read or for write
* @throws Exception if some low level IO error occurs
*/
protected abstract int select(long timeout) throws Exception;

}

public final class NioProcessor extends AbstractPollingIoProcessor<NioSession> {
    private final Selector selector;

@Override
protected int select() throws Exception {
    return selector.select();
}

}

public abstract class AbstractPollingIoProcessor<T extends AbstractIoSession> implements IoProcessor<T> {

 protected AtomicBoolean wakeupCalled = new AtomicBoolean(false);

private class Processor implements Runnable {
        public void run() {
            assert (processorRef.get() == this);

            int nSessions = 0;
            lastIdleCheckTime = System.currentTimeMillis();

            for (;;) {
                try {
                    // This select has a timeout so that we can manage
                    // idle session when we get out of the select every
                    // second. (note : this is a hack to avoid creating
                    // a dedicated thread).
                   long t0 = System.currentTimeMillis();
                    int selected = select(SELECT_TIMEOUT);
                    long t1 = System.currentTimeMillis();
                    long delta = (t1 - t0);

                    if ((selected == 0) && !wakeupCalled.get() && (delta < 100)) {
                        // Last chance : the select() may have been
                        // interrupted because we have had an closed channel.
                        if (isBrokenConnection()) {
                            LOG.warn("Broken connection");

                            // we can reselect immediately
                            // set back the flag to false
                            wakeupCalled.getAndSet(false);

                            continue;
                        } else {
                            LOG.warn("Create a new selector. Selected is 0, delta = "
                                            + (t1 - t0));
                            // Ok, we are hit by the nasty epoll
                            // spinning.
                            // Basically, there is a race condition
                            // which causes a closing file descriptor not to be
                            // considered as available as a selected channel, but
                            // it stopped the select. The next time we will
                            // call select(), it will exit immediately for the same
                            // reason, and do so forever, consuming 100%
                            // CPU.
                            // We have to destroy the selector, and
                            // register all the socket on a new one.
                           registerNewSelector();
                        }

                       // Set back the flag to false
                        wakeupCalled.getAndSet(false);

                        // and continue the loop
                        continue;
                    }

                    // Manage newly created session first
                    nSessions += handleNewSessions();

                    updateTrafficMask();

                    // Now, if we have had some incoming or outgoing events,
                    // deal with them
                    if (selected > 0) {
                        //LOG.debug("Processing ..."); // This log hurts one of the MDCFilter test...
                        process();
                    }

                    // Write the pending requests
                    long currentTime = System.currentTimeMillis();
                    flush(currentTime);

                    // And manage removed sessions
                    nSessions -= removeSessions();

                    // Last, not least, send Idle events to the idle sessions
                    notifyIdleSessions(currentTime);

                    // Get a chance to exit the infinite loop if there are no
                    // more sessions on this Processor
                    if (nSessions == 0) {
                        processorRef.set(null);
                       
                        if (newSessions.isEmpty() && isSelectorEmpty()) {
                            // newSessions.add() precedes startupProcessor
                            assert (processorRef.get() != this);
                            break;
                        }
                       
                        assert (processorRef.get() != this);
                       
                        if (!processorRef.compareAndSet(null, this)) {
                            // startupProcessor won race, so must exit processor
                            assert (processorRef.get() != this);
                            break;
                        }
                       
                        assert (processorRef.get() == this);
                    }

                    // Disconnect all sessions immediately if disposal has been
                    // requested so that we exit this loop eventually.
                    if (isDisposing()) {
                        for (Iterator<S> i = allSessions(); i.hasNext();) {
                            scheduleRemove(i.next());
                        }

                        wakeup();
                    }
                } catch (ClosedSelectorException cse) {
                    // If the selector has been closed, we can exit the loop
                    break;
                } catch (Throwable t) {
                    ExceptionMonitor.getInstance().exceptionCaught(t);

                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e1) {
                        ExceptionMonitor.getInstance().exceptionCaught(e1);
                    }
                }
            }

public final class NioProcessor extends AbstractPollingIoProcessor<NioSession> {
    /** The selector associated with this processor */
    private Selector selector;

@Override
protected void registerNewSelector() throws IOException {
    synchronized (selector) {
        Set<SelectionKey> keys = selector.keys();

        // Open a new selector
        Selector newSelector = Selector.open();

        // Loop on all the registered keys, and register them on the new selector
        for (SelectionKey key : keys) {
            SelectableChannel ch = key.channel();

            // Don't forget to attache the session, and back !
            NioSession session = (NioSession)key.attachment();
            SelectionKey newKey = ch.register(newSelector, key.interestOps(), session);
            session.setSelectionKey( newKey );
        }

        // Now we can close the old selector and switch it
        selector.close();
        selector = newSelector;
    }
}

class RecvWorker extends Thread {

    Long sid;

    SocketChannel channel;

    volatile boolean running = true;

     final SendWorker sw;

    RecvWorker(SocketChannel channel, Long sid, SendWorker sw) {

        this.sid = sid;

        this.channel = channel;

        this.sw = sw;

    }

    public void run() {

        threadCnt.incrementAndGet();

        try {

        byte[] size = new byte[4];

        ByteBuffer msgLength = ByteBuffer.wrap(size);

        while (running && !shutdown && channel != null) {

            while (msgLength.hasRemaining()) {

-                channel.read(msgLength);  // 기존 코드

+                if (channel.read(msgLength) < 0) { // 수정된 코드

+                      throw new IOException("Channel eof");

+                }

       }

       msgLength.position(0);

      int length = msgLength.getInt();

      if(length <= 0) {

         throw new IOException("Invalid packet length:" + length);

      }

}