Reactor Pattern Explained - Part 3

This is the 3rd and final post of the topic Reactor Pattern Explained. If you haven't read the previous posts, please read them since they include a great deal of explanations.

Reacter Pattern Explained - Part 1
Reacter Pattern Explained - Part 2

In this post the usage of Thread pools in Handlers is explained. We will create an extended version of Handler class named HandlerWithThreadPool. Check this out.

public class HandlerWithThreadPool extends Handler {

    static ExecutorService pool = Executors.newFixedThreadPool(2);
    static final int PROCESSING = 2;

    public HandlerWithThreadPool(Selector sel, SocketChannel c) throws IOException {
        super(sel, c);
    }

    void read() throws IOException {
        int readCount = socketChannel.read(input);
        if (readCount > 0) {
            state = PROCESSING;
            pool.execute(new Processer(readCount));
        }
        //We are interested in writing back to the client soon after read processing is done.
        selectionKey.interestOps(SelectionKey.OP_WRITE);
    }

    //Start processing in a new Processer Thread and Hand off to the reactor thread.
    synchronized void processAndHandOff(int readCount) {
        readProcess(readCount);
        //Read processing done. Now the server is ready to send a message to the client.
        state = SENDING;
    }

    class Processer implements Runnable {
        int readCount;
        Processer(int readCount) {
            this.readCount =  readCount;
        }
        public void run() {
            processAndHandOff(readCount);
        }
    }
}

Notice that there is a new state PROCESSING introduced and that the read() method is over-ridden. Now when a Read Event is dispatched to this Handler, it will read the data but not change the state to SENDING. It will create a Processer which will process the message and run it in a different Thread in the Thread pool and set the Interested Operation to OP_WRITE. At this point even if the Channel is ready to be written to and the Hander is interested in writing, the Handler will not write since its still in PROCESSING state. See the run() method of the Handler, it will only write when its in SENDING state. Once the Processer is done with its read process, it will change the state to SENDING. Now the Handler can send data to the client.

Lets run the Reactor with the boolean isWithThreadPool set to true.

public static void main(String[] args) throws IOException{

    Reactor reactor  = new Reactor(9900, true);
    new Thread(reactor).start();
}

Notice that the size of the Thread pool is 2 which would limit the number of Handlers to run concurrently to 2. As we already know, when the events are handled concurrenly we can easyly improve the system performance.

Done :). Hope you find this useful. Have fun with coding.
Cheers.

Reactor Pattern Explained - Part 2

This is the continuation of my previous post Reactor Pattern Explained - Part 1 which gives a high-level understanding of the topic.

In this blog post I will explain the implementation of Reactor Pattern with a simple Client - Server system where the server will send Hello messages to each client when their names are told to the server. The server will listen to port 9900 and multiple clients will connect to the server to shout out their names. A thread pool will not be used here. First lets run the server in a single thread. Part 3 of this series will explain how a Thread pool is used.

First lets make the Client to connect to port 9900.

public class Client {
    String hostIp;
    int hostPort;

    public Client(String hostIp, int hostPort) {
        this.hostIp = hostIp;
        this.hostPort = hostPort;
    }

    public void runClient() throws IOException {
        Socket clientSocket = null;
        PrintWriter out = null;
        BufferedReader in = null;

        try {
            clientSocket = new Socket(hostIp, hostPort);
            out = new PrintWriter(clientSocket.getOutputStream(), true);
            in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
        } catch (UnknownHostException e) {
            System.err.println("Unknown host: " + hostIp);
            System.exit(1);
        } catch (IOException e) {
            System.err.println("Couldn't connect to: " + hostIp);
            System.exit(1);
        }

        BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
        String userInput;

        System.out.println("Client connected to host : " + hostIp + " port: " + hostPort);
        System.out.println("Type (\"Bye\" to quit)");
        System.out.println("Tell what your name is to the Server.....");

        while ((userInput = stdIn.readLine()) != null) {

            out.println(userInput);

            // Break when client says Bye.
            if (userInput.equalsIgnoreCase("Bye"))
                break;

            System.out.println("Server says: " + in.readLine());
        }

        out.close();
        in.close();
        stdIn.close();
        clientSocket.close();
    }

    public static void main(String[] args) throws IOException {

        Client client = new Client("127.0.0.1", 9900);
        client.runClient();
    }
}

Notice that the client doesn't use java.nio to create the Socket. It simply uses a java.net.Socket everybody knows about.

Now lets make the Reactor in the Server.

public class Reactor implements Runnable {

    final Selector selector;
    final ServerSocketChannel serverSocketChannel;
    final boolean isWithThreadPool;

    Reactor(int port, boolean isWithThreadPool) throws IOException {

        this.isWithThreadPool = isWithThreadPool;
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        SelectionKey selectionKey0 = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        selectionKey0.attach(new Acceptor());
    }


    public void run() {
        System.out.println("Server listening to port: " + serverSocketChannel.socket().getLocalPort());
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set selected = selector.selectedKeys();
                Iterator it = selected.iterator();
                while (it.hasNext()) {
                    dispatch((SelectionKey) (it.next()));
                }
                selected.clear();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    void dispatch(SelectionKey k) {
        Runnable r = (Runnable) (k.attachment());
        if (r != null) {
            r.run();
        } 
    }

    class Acceptor implements Runnable {
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                if (socketChannel != null) {
                    if (isWithThreadPool)
                        new HandlerWithThreadPool(selector, socketChannel);
                    else
                        new Handler(selector, socketChannel);
                }
                System.out.println("Connection Accepted by Reactor");
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
}

The Reactor is a Runnable. See the while loop in the run() method. It will call selector.select() to get the SelectionKeys which have pending IO events. When the SelectionKeys are selected, they will be dispatched one by one. See the dispatch() method. The SelectionKey will have an attatchment which is also a Runnable. This attatchement will either be an Acceptor or a Handler.
Notice how the Acceptor inner class in the Reactor accepts connections to make SocketChannels. When a SocketChannel is created a new Handler will be created as well. (HandlerWithThreadPool will be discussed in the next section)

public class Handler implements Runnable {

    final SocketChannel socketChannel;
    final SelectionKey selectionKey;
    ByteBuffer input = ByteBuffer.allocate(1024);
    static final int READING = 0, SENDING = 1;
    int state = READING;
    String clientName = "";

    Handler(Selector selector, SocketChannel c) throws IOException {
        socketChannel = c;
        c.configureBlocking(false);
        selectionKey = socketChannel.register(selector, 0);
        selectionKey.attach(this);
        selectionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }


    public void run() {
        try {
            if (state == READING) {
                read();
            } else if (state == SENDING) {
                send();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    void read() throws IOException {
        int readCount = socketChannel.read(input);
        if (readCount > 0) {
            readProcess(readCount);
        }
        state = SENDING;
        // Interested in writing
        selectionKey.interestOps(SelectionKey.OP_WRITE);
    }

    /**
     * Processing of the read message. This only prints the message to stdOut.
     *
     * @param readCount
     */
    synchronized void readProcess(int readCount) {
        StringBuilder sb = new StringBuilder();
        input.flip();
        byte[] subStringBytes = new byte[readCount];
        byte[] array = input.array();
        System.arraycopy(array, 0, subStringBytes, 0, readCount);
        // Assuming ASCII (bad assumption but simplifies the example)
        sb.append(new String(subStringBytes));
        input.clear();
        clientName = sb.toString().trim();
    }

    void send() throws IOException {
        System.out.println("Saying hello to " + clientName);
        ByteBuffer output = ByteBuffer.wrap(("Hello " + clientName + "\n").getBytes());
        socketChannel.write(output);
        selectionKey.interestOps(SelectionKey.OP_READ);
        state = READING;
    }
}

A Handler has 2 states, READING and SENDING. Both cant be handled at the same time because a Channel supports only one operation at one time. Since its the client who speaks first, a server Handler will start with the READING state. Notice how this Handler is attatched to the SelectionKey and how the Interested Operation is set to OP_READ. This means that the Selector should only select this SelectionKey when a Read Event occurs. Once the read process is done, the Handler will change its state to SENDING and will change the Interested Operation to OP_WRITE. Now the Selector will select this SelectionKey only when it gets a Write Event from the Channel when its ready to be written with data. When a Write Event is dispatched to this Handler, it will write the Hello message to the output buffer since now the state is SENDING. Once sending is done, it will change back to READING state with Interested Operation changed to OP_READ again. It should be obvious that since both Handler and Acceptor are Runnables, the dispatch() method of the Reactor can execute the run() method of any attatchment it gets from a selected SelectionKey.

Here is the main method. We will run it without a Thread pool for the moment.

public static void main(String[] args) throws IOException{

    Reactor reactor  = new Reactor(9900, false);
    new Thread(reactor).start();
}

To see how this works first run the server. Then run several clients and see how they get connected to the server. When each client writes a name to standard in of the client, the sever will respond to the client with a Hello message. Notice that the server runs in a single Thread but responds to any number of clients which connect to the server.

Read the next section Reacter Pattern Explained - Part 3 to see how to use a Thread pool to run Handlers.

Reactor Pattern Explained - Part 1

Handling concurrent events a Server receives is often thought of as a use-case for creating a separate thread for each IO event listener. Most programmers are tempted to use the famous socket loop for creating Sockets for every incoming connection.

class Server implements Runnable {
    public void run() {
        try {
            ServerSocket ss = new ServerSocket(PORT);
        while (!Thread.interrupted())
            new Thread(new Handler(ss.accept())).start();
            // or, single-threaded, or a thread pool
        } catch (IOException ex) { }
    }
}

class Handler implements Runnable {
    final Socket socket;
    Handler(Socket s) { socket = s; }
    public void run() {
        try {
            byte[] input = new byte[MAX_INPUT];
            socket.getInputStream().read(input);
            byte[] output = process(input);
            socket.getOutputStream().write(output);
        } catch (IOException ex) { }
    }
    private byte[] process(byte[] cmd) { }
}

The disadvantage of using a separate thread for each event listener is the overhead of context switching. In the worst case, some threads handling event listeners which do not read or write data frequently, will be context switched periodically without doing useful work. Every time such a Thread is dispatched to the CPU by the scheduler, it will be blocked until an IO event occurs, in which case all the time spent waiting for an IO event will be wasted. Note that ss.accept() is a blocking call which blocks the server thread till a client connects. The server thread will not be able to call start() method of the new Handler thread until it is returned from ss.accept(). To reduce the wastage of CPU time by unnecessary context switches, the concept of non blocking IO was invented.

Reactor Pattern is an event handling design pattern used to address this issue. Here, one Reactor will keep looking for events and will inform the corresponding event handler to handle it once the event gets triggered. To explain this I am using some Java code borrowed from some lecture slides by Professor Doug Lea. To see his explanation please go through this set of slides.

Java provides a standard API (java.nio) which could be used to design non-blocking IO systems. I will explain the Reactor pattern with a simple client server model where the clients will shout out their names to the server while the server will respond to the corresponding client with a Hello message.

There are two important participants in the architecture of Reactor Pattern.

1. Reactor  

A Reactor runs in a separate thread and its job is to react to IO events by dispatching the work to the appropriate handler. Its like a telephone operator in a company who answers the calls from clients and transfers the communication line to the appropriate receiver. Don't go too far with the analogy though :).

2. Handlers

A Handler performs the actual work to be done with an IO event similar to the actual officer in the company the client who called wants to speak to.

Since we are using java.nio package, its important to understand some of the classes used to implement the system. I will simply repeat some of the explanations by Doug Lea in his lecture sides to make the readers lives easy :).

Channels

These are connections to files, sockets etc. that support non blocking reads. Just like many TV channels can be watched from one physical connection to the antena, many java.nio.channels.SocketChannels corresponding to each client can be made from a single java.nio.channels.ServerSocketChannel which is bound to a single port.

Buffers

Array-like objects that can be directly read or written to by Channels.

Selectors

Selectors tell which of a set of Channels has IO events.

Selection Keys

Selection Keys maintain IO event status and bindings. Its a representation of the relationship between a Selector and a Channel. By looking at the Selection Key given by the Selector, the Reactor can decide what to do with the IO event which occurs on the Channel.

Now lets try to understand what Reactor Pattern is. Take a look at this diagram.

 
Here, there is a single ServerSocketChannel which is registered with a Selector. The SelectionKey 0 for this registration has information on what to do with the ServerSocketChannel if it gets an event. Obviously the ServerSocketChannel should receive events from incoming connection requests from clients. When a client requests for a connection and wants to have a dedicated SocketChannel, the ServerSocketChannel should get triggered with an IO event. What does the Reactor have to do with this event? It simply has to Accept it to make a SocketChannel. Therefore SelectionKey 0 will be bound to an Acceptor which is a special handler made to accept connections so that the Reactor can figure out that the event should be dispatched to the Acceptor by looking at SelectionKey 0. Notice that ServerSocketChannel, SelectionKey 0 and Acceptor are all in same colour ( Gray I suppose :) )

The Selector is made to keep looking for IO events. When the Reactor calls Selector.select() method, the Selector will provide a set of SelectionKeys for the channels which have pending events. When SelectionKey 0 is selected, it means that an event has occurred on ServerSocketChannel. So the Reactor will dispatch the event to the Acceptor.

When the Acceptor accepts the connection from Client 1, it will create a dedicated SocketChannel 1 for the client. This SocketChannel will be registered with the same Selector with SelectionKey 1. What would the client do with this SocketChannel? It will simply read from and write to the server. The server does not need to accept connections from client 1 any more since it already accepted the connection. Now what the server needs is to Read and Write data to the channel. So SelectionKey 1 will be bound to Handler 1 object which handles reading and writing. Notice that SocketChannel 1, SelectionKey 1 and Handler 1 are all in Green.

The next time the Reactor calles Selector.select(), if the returned SelectionKey Set has SelectionKey 1 in it,  it means that SocketChannel 1 is triggered with an event. Now by looking at SelectionKey 1, the Reactor knows that it has to dispatch the event to Handler 1 since Hander 1 is bound to SelectionKey 1. If the returned SelectionKey Set has SelectionKey 0 in it, it means that ServerSocketChannel has received an event from another client and by looking at the SelectionKey 0 the Reactor knows that it has to dispatch the event to the Acceptor again. When the event is dispatched to the Acceptor it will make SocketChannel 2 for client 2 and register the socket channel with the Selector with SelectionKey 2.

So in this scenario we are interested in 3 types of events.

1. Connection request events which get triggered on the ServerSocketChannel which we need to Accept.
2. Read events which get triggerd on SocketChannels when they have data to be read, from which we need to Read.
3. Write events which get triggered on SocketChannels when they are ready to be written with data, to which we need to Write.

A SelectionKey will have all the information about the relationship with its corresponding Channel and the Selector. It will have information about the corresponding Handler too. Selector will just select the SelectionKeys which have pending IO events. This way the Reactor can decide how to deal with the IO events accordingly. The relationships among the Channels, Selection Keys and Handlers can be put in a table as follows.

Selection Key	Channel	Handler	Interested Operation
SelectionKey 0	ServerSocketChannel	Acceptor	Accept
SelectionKey 1	SocketChannel 1	Handler 1	Read and Write
SelectionKey 2	SocketChannel 2	Handler 2	Read and Write
SelectionKey 3	SocketChannel 3	Handler 3	Read and Write

Now what does a Thread pool has to do with this? Let me explain. The beauty of non blocking architecture is that we can write the server to run in a single Thread while catering all the requests from clients. Just forget about the Thread pool for a while. Naturally when concurrency is not used to design a server it should obviously be less responsive to events. In this scenario when the system runs in a single Thread the Reactor will not respond to other events until the Handler to which the event is dispatched is done with the event. Why? Because we are using one Thread to handle all the events. We naturally have to go one by one.

We can add concurrency to our design to make the system more responsive and faster. When the Reactor dispatches the event to a Handler, it can start the Handler in a new Thread so that the Reactor can happily continue to deal with other events. This will always be a better design when performance is concerned. To limit the number of Threads in the system and to make things more organized, a Thread pool can be used.

I believe this explanation is adequate for us to get our hands dirty with some coding.

Please read Reactor Pattern Explained - Part 2 and Reactor Pattern Explained - Part 3.