Registering a SocketChannel on Multiple Selectors

We want to hear from you! Please send us your FEEDBACK.

The following sample program may contain actual software programs in source code form. This source code is made available for developers to use as needed, pursuant to the terms and conditions of this license.

INTRODUCTION
The Java[tm] 2 Standard Edition (J2SE[tm]) platform, version 1.4 (Merlin) introduced non-blocking input/output (IO) based on the Reactor Pattern. NIO (or New IO) is built around Selectors and Channels. In this document, we will discuss a sample Server program to show how to register a single SocketChannel with multiple Selectors running in seperate threads.

SAMPLE PROGRAM
Here we assume that you are familiar with the NIO programming in J2SE, version 1.4. Please refer to the NIO guide to learn about NIO.

Let's look at our example code. You can download the source here.

1. Import the necessary NIO classes:

import java.net.ServerSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.io.IOException;
import java.util.Iterator;
import java.util.Set;

2. Review the definitions for two Selectors() (readSelector and writeSelector) which will run in two different threads (readThread and writeThread respectively). A single ServerSocketChannel will be registered to the readSelector and writeSelector.

public class SelectorTestServer {

  /** Single selector for accepts, reads */
  private Selector readSelector = null;

  /** Single selector for writes */
  private Selector writeSelector = null;

  /** ServerSocketChannel which listens for client connections*/
  private ServerSocketChannel ssch = null;

  /** The thread that waits for ready Channels - accept / read*/
  private SelectorThread readThread = null;

  /** The thread that waits for ready Channels - write*/
  private WriteThread writeThread = null;

3.  Set up the listeners and start listening.

  protected void startListening () {
        try{
        // create the selector and the ServerSocket
                readSelector = SelectorProvider.provider().openSelector();
                writeSelector = SelectorProvider.provider().openSelector();

                //Create the ServerSocketChannel

                ssch = ServerSocketChannel.open();

                //Configure the ServerSocketChannel as Non-Blocking

                ssch.configureBlocking(false);

                //Bind the socket to the localhost at port 2000

                InetSocketAddress isa = new InetSocketAddress(InetAddress.getLocalHost(), 2000);
                ssch.socket().bind(isa);

                //register the ServerSocketChannel with the Selector for accept

                ssch.register(readSelector, SelectionKey.OP_ACCEPT);
        }
        catch (Exception e){
                System.out.println("Error starting listening"+e.toString());
        }

        //Start the Selector threads

        this.writeThread = new WriteThread();
        this.writeThread.setDaemon(true);
        this.writeThread.start();
        this.readThread = new SelectorThread();
        this.readThread.setDaemon(true);
        this.readThread.start();
  }
 

4. Look at the SelectorThread which will run the Selector() for accepts and reads (OP_ACCEPT and OP_READ).

  private class SelectorThread extends Thread {
        public SelectorThread() {
                super("SelectorThread");
        }

    public void run () {
        boolean running = true;
        try{

        // block until a Channel is ready for I/O

                while(readSelector.select() > 0){
                        System.out.println("Read Thread while loop");
                        Set readyKeys = readSelector.selectedKeys();
                        Iterator i = readyKeys.iterator();
                        while (i.hasNext()){
                                SelectionKey sk = (SelectionKey) i.next();
                                i.remove();
                                if (sk.isAcceptable()) {

                                // new client connection. Accept the connection.

                                        ServerSocketChannel nextReady = (ServerSocketChannel) sk.channel();
                                        SocketChannel channel = nextReady.accept();
                                        channel.configureBlocking(false);

                                        //Register for Reads and writes with the two Selectors

                                        channel.register(readSelector, SelectionKey.OP_READ);
                                        writeSelector.wakeup(); // Added becose required by specs
                                        channel.register(writeSelector, SelectionKey.OP_WRITE);
                                }
                                else if (sk.isReadable()) {
                                        System.out.println("****** readable");
                                }
                        }
                }
        }
        catch (Exception ex){
                System.out.println("Exception in selector loop: "+ex.toString());
                running = false;
        }
    } //end run()
  } //end class

In SelectorThread it is important to call wakeup() on one of the Selectors, else you may not recieve any events on any of the Selectors. According to the specifications of the register() method,

    "This method may be invoked at any time. If this method is invoked while another invocation of
     this method or of the configureBlocking method is in progress then it will first block until
     the other operation is complete. This method will then synchronize on the selector's key set
     and therefore may block if invoked concurrently with another registration or selection
     operation involving the same selector."
 
 

5. Look at WriteThread, which will run the Selector() for writes (OP_WRITE).

  private class WriteThread extends Thread {
        public WriteThread() {
        super("WriteThread");
        }

    public void run () {
        boolean running = true;
        try{
        // block until a Channel is ready for I/O
                while(writeSelector.select() > 0){
                        System.out.println("Write Thread while loop");
                        Set readyKeys = writeSelector.selectedKeys();
                        Iterator i = readyKeys.iterator();
                        while (i.hasNext()){
                                SelectionKey sk = (SelectionKey) i.next();
                                i.remove();
                                if (sk.isWritable()) {
                                        System.out.println("******* writable");
                                         // Do more here
                                }
                        }
                }
        }
      catch (Exception ex){
        System.out.println("Exception in write selector loop: "+ex.toString());
        running = false;
      }
    } //end run()
  } //end class

6. Provide a method to stop Selector threads.

  public synchronized void stop () {
        try {
                this.ssch.close();
        }
        catch (Exception e) {
                System.err.println("Exception in stop()"+e.toString());
        }
        this.ssch = null;
  }

7. Write a main() method to show that this works.
 

 public static void main (String args[]){
    SelectorTestServer s = new SelectorTestServer();
    s.startListening();
    try{
        //Keep main thread alive
        Thread.currentThread().sleep(500000);
    }
    catch (Exception e){
        System.err.println(e.toString());
    }
  }

The server code shown above only prints out "****readable" or "****writable", depending on the state of the client connection. It can be modified to show appropriate actions as required.

When you start the server, two threads are started. These threads run readSelector and writeSelector to listen for readable or writable connections to the client. Let's look at the readSelector code to see what happens when the client request comes in.

The server waits until a client makes a connection:

 while(readSelector.select() > 0){
                        System.out.println("Read Thread while loop");

Let's have a Set hold the keys returned from readSelector.selectedKeys();

Set readyKeys = readSelector.selectedKeys();
                        Iterator i = readyKeys.iterator();
                        while (i.hasNext()){
                                SelectionKey sk = (SelectionKey) i.next();
                                i.remove();

Use the Iterator to check if there is a new client connection or if this is a current client available for read.

If we have a new client connection, sk.isAcceptable() will return true. In that case, we accept the connection and set up the channel for Non-blocking IO:

                                        if (sk.isAcceptable()){
                                        // new client connection
                                        ServerSocketChannel nextReady = (ServerSocketChannel) sk.channel(
);
                                        SocketChannel channel = nextReady.accept();
                                        channel.configureBlocking(false);

Now let's register the channel with read and write selectors:

                                        channel.register(readSelector, SelectionKey.OP_READ);
                                        writeSelector.wakeup(); // Added as required by register() specifications.
                                        channel.register(writeSelector, SelectionKey.OP_WRITE);

Note that we have to call wakeup() on the writeSelector thread, else no write events will be detected by the selector.

If the client is not a new client, then check if it is still ready for reads. If yes, then we print "****readable":

                                         if (sk.isReadable()){
                                                System.out.println("****** readable");
                                }
Similar actions occur in the writeselector thread, which detects for write events from the clients.

This sample Server has been tested to work with a telnet client (telnet localhost 2000).
 

Thanks to Tarik Tutill of Isocra for his contributions to this article.