NIO Selector problem

[Arie Keren]

Hi,

I'm running Diablo Java(TM) SE Runtime Environment (build 1.6.0_07-b02) on
FreeBSD.
I have a problem related to Selector:

After receiving read event, I remove OP_READ from key.interestOps(), then
read the channel and then return the OP_READ to key.interestOps().
After this the selector doesn't wake up on read.

This problem can be reproduced with the attached test.
The test runs OK on Windows but Fails on FreeBSD.
Note that if you comment the lines marked with //1-3, the program runs OK
both on Windows and FreeBSD (but of course this causes to much false read
invocations).

arie

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestSelector {

    static boolean terminate;
    public static void main(String[] args) {
        try {
            new NioSelector();
            Socket socket = new Socket("127.0.0.1", 4000);
            for (int i=0; i<5; i++) {
                socket.getOutputStream().write(1);
                Thread.sleep(500);
            }
            terminate = true;
            socket.close();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println ((NioSelector.numReads == 5 ? "OK:" : "FAIL:") +
NioSelector.numReads);
    }

    public static class NioSelector implements Runnable {
        protected Selector selector;
        ServerSocketChannel tcpListener;
        SocketChannel tcpClient;
        ExecutorService executor;
        static int numReads;

        public NioSelector() throws IOException {
            executor = Executors.newFixedThreadPool(1);
            selector = Selector.open();
            tcpListener = ServerSocketChannel.open();
            tcpListener.configureBlocking(false);
            tcpListener.socket().bind(new InetSocketAddress(4000));
            tcpListener.register(selector, SelectionKey.OP_ACCEPT);
            new Thread(this).start();
        }

        @Override
        public void run() {
            while (!terminate) {
                try {
                    int numKeys = selector.select(1000);
                    if (numKeys == 0) continue;

                    Iterator<SelectionKey> it =
selector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();

                        if (key.isAcceptable()) {
                            tcpClient =
((ServerSocketChannel)key.channel()).accept();
                            tcpClient.configureBlocking(false);
                            tcpClient.register(selector,
SelectionKey.OP_READ);
                        }
                        else if (key.isReadable()) {
                            key.interestOps(key.interestOps() &
~SelectionKey.OP_READ);  //1
                            final SelectionKey fkey = key;
                            executor.execute(new Runnable () {
                                public void run() {
                                    ByteBuffer inBuf =
ByteBuffer.allocate(1);
                                    try {
                                        if (tcpClient.read(inBuf) > 0)
numReads++;
                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    }
                                    fkey.interestOps(fkey.interestOps() |
SelectionKey.OP_READ);  //2
                                    selector.wakeup();   //3
                                }
                            });
                        }
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
            executor.shutdownNow();
        }
    }

}