Reactor Pattern and Non-blocking IO

README

Reactor Pattern and Non-blocking IO

http://www.cs.bgu.ac.il/~spl051/Personal_material/Practical_sessions/Ps_12/ps12.html

ConnectionAcceptor.java

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

/**
 * Handles new client connections.
 * An Acceptor is bound on a ServerSocketChannel objects, which can produce new
 * SocketChannels for new clients using its <CODE>accept</CODE> method.
 */
public class ConnectionAcceptor {
    protected Selector _selector;
    protected ServerSocketChannel _ssChannel;
    protected ThreadPool _pool;

    /**
     * Creates a new ConnectionAcceptor
     * @param selector the Selector to which new SocketChannels will be registered
     * @param ssChannel the ServerSocketChannel which can accept new connections
     * @param pool the thread pool, which is needed by the new ConnectionReaders
     */
    public ConnectionAcceptor(Selector selector, ServerSocketChannel ssChannel, ThreadPool pool) {
        _selector = selector;
        _ssChannel = ssChannel;
        _pool = pool;
    }

    /**
     * Accepts a connection:
     * <UL>
     * <LI>Creates a SocketChannel for it
     * <LI>Creates a ConnectionReader for it
     * <LI>Registers the SocketChannel and the ConnectionReader to the Selector
     * </UL>
     * @throws IOException in case of an IOException during the acceptance of a new connection
     */
    public void accept() throws IOException {
        // Get a new channel for the connection request
        SocketChannel sChannel = _ssChannel.accept();

        // If serverSocketChannel is non-blocking, sChannel may be null
        if (sChannel != null) {
            SocketAddress address = sChannel.socket().getRemoteSocketAddress();
            System.out.println("Accepting connection from " + address);
            sChannel.configureBlocking(false);
            sChannel.register(_selector, SelectionKey.OP_READ, new ConnectionReader(sChannel, _pool));
        }
    }
}

ConnectionReader.java

import java.nio.channels.SocketChannel;
import java.nio.ByteBuffer;
import java.io.IOException;
import java.net.SocketAddress;

/**
 * Handles messages from clients
 */
public class ConnectionReader {
    public static final int BUFFER_SIZE = 256;
    public static final char MESSAGE_END = ';';

    protected SocketChannel _sChannel;
    protected String _incomingData;
    protected ThreadPool _pool;

    /**
     * Creates a new ConnectionReader object
     * @param sChannel the SocketChannel of the client
     * @param pool the ThreadPool to which new Tasks should be inserted
     */
    public ConnectionReader(SocketChannel sChannel, ThreadPool pool) {
        _sChannel = sChannel;
        _pool = pool;
        _incomingData = "";
    }

    /**
     * Reads messages from the client:
     * <UL>
     * <LI>Reads the entire SocketChannel's buffer
     * <LI>Separate the information into messges
     * <LI>For each message:
     * <UL>Creates a Task for the message
     * <LI>Inserts the Task to the ThreadPool
     * </UL>
     * </UL>
     * @throws IOException in case of an IOException during reading
     */
    public void read() throws IOException {

        SocketAddress address = _sChannel.socket().getRemoteSocketAddress();
        System.out.println("Reading from " + address);

        //ByteBuffer buf = ByteBuffer.allocateDirect(BUFFER_SIZE);
        ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE);

        // Read the entire content of the socket
        while (true) {
            buf.clear();
            int numBytesRead = _sChannel.read(buf);

            // Closed channel
            if (numBytesRead == -1) {
                // No more bytes can be read from the channel
                System.out.println("client on " + address + " has disconnected");
                _sChannel.close();
                break;
            }

            // Read the buffer
            if (numBytesRead > 0) {
                //read the data
                buf.flip();
                String str = new String(buf.array(), 0, numBytesRead);
                _incomingData = _incomingData + str;
            }

            //end of message
            if (numBytesRead < BUFFER_SIZE) {
                break;
            }
        }

        // Parse the incoming data into buffer separate messages
        // and handle them
        while (true) {
            int pos = _incomingData.indexOf(MESSAGE_END);

            // No message end mark in the incoming data buffer
            if (pos==-1) {
                break;
            }

            // Extract one message, omit it from the incoming data buffer
            String message = _incomingData.substring(0, pos);
            _incomingData = pos==_incomingData.length()-1 ? "" : _incomingData.substring(pos+1);

            // Do something with the message
            System.out.println("Message " + message + " added to the pool");
            _pool.addTask(new MessageProcessorTask(message, _sChannel));
        }
    }
}

MessageProcessorTask.java

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

/**
 * A sample Message Processor
 * The Message Processor holds a received message, and a SocketChannel to the message sender;
 * this socket enables replying back to the sender.
 * This sample processor sends a simple reply back to the message sender, regardless of the message's content.
 * <B>You should either extend or rewrite this Message Processor to work properly with the assignment definition.</B>
 */
class MessageProcessorTask implements Task {
    protected String _message;
    protected SocketChannel _channel;

    /**
     * Creates a new MessageProcessorTask
     * @param message the messge, as receieved from the sender
     * @param channel a channel which will be used to reply to the message sender
     */
    public MessageProcessorTask(String message, SocketChannel channel) {
        _message = message;
        _channel = channel;
    }


    /**
     * Executes the task
     * This simple implementation simply replies the sender with a general reply.
     * @throws TaskFailedException in case of a failure while executing the task 
     */
    public void executeTask() throws TaskFailedException {
        String response = "Got your message!;";
        try {
            _channel.write(ByteBuffer.wrap(response.getBytes()));
        }
        catch (IOException io) {
            throw new TaskFailedException("I/O exception while processing the message: " + _message, io);
        }
    }
}

Reactor.java

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;

/**
 * An implementation of the Reactor pattern
 */
public class Reactor extends Thread {

    protected int _port;
    protected int _poolSize;
    protected ThreadPool _pool;
    protected volatile boolean _shouldRun = true;

    /**
     * Creates a new Reactor
     * @param poolSize the number of WorkerThreads to include in the ThreadPool
     * @param port the port to bind the Reactor to
     * @throws IOException if some I/O problems arise during connection
     */
    public Reactor(int poolSize, int port) throws IOException {
        _port = port;
        _poolSize = poolSize;
    }

    /**
     * Main operation of the Reactor:
     * <UL>
     * <LI>Uses the <CODE>Selector.select()</CODE> method to find new requests from clients
     * <LI>For each request in the selection set:
     * <UL>If it is <B>acceptable</B>, use the ConnectionAcceptor to accept it,
     * create a new ConnectionReader for it register it to the Selector
     * <LI>If it is <B>readable</B>, use the ConnectionReader to read it,
     * extract messages and insert them to the ThreadPool
     * </UL>
     */
    public void run() {
        try {
            // Create & start the ThreadPool
            _pool = new ThreadPool(_poolSize);
            _pool.startPool();

            // Create a non-blocking server socket channel and bind to to the Reactor port
            ServerSocketChannel ssChannel = ServerSocketChannel.open();
            ssChannel.configureBlocking(false);
            ssChannel.socket().bind(new InetSocketAddress(_port));

            // Create the selector and bind the server socket to it
            Selector selector = Selector.open();
            ssChannel.register(selector, SelectionKey.OP_ACCEPT, new ConnectionAcceptor(selector, ssChannel, _pool));

            while (_shouldRun) {
                // Wait for an event
                selector.select();

                // Get list of selection keys with pending events
                Iterator it = selector.selectedKeys().iterator();

                // Process each key
                while (it.hasNext()) {
                    // Get the selection key
                    SelectionKey selKey = (SelectionKey)it.next();

                    // Remove it from the list to indicate that it is being processed
                    it.remove();

                    // Check if it's a connection request
                    if (selKey.isValid() && selKey.isAcceptable()) {
                        ConnectionAcceptor connectionAcceptor = (ConnectionAcceptor)selKey.attachment();
                        connectionAcceptor.accept();
                    }
                    // Check if a message has been sent
                    if (selKey.isValid() && selKey.isReadable()) {
                        ConnectionReader connectionReader = (ConnectionReader)selKey.attachment();
                        connectionReader.read();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace(System.err);
            stopReactor();
        }
        stopReactor();
    }

    /**
     * Returns the listening port of the Reactor
     * @return the listening port of the Reactor
     */
    public int getPort() {
        return _port;
    }

    /**
     * Stops the Reactor activity, including the Reactor thread and the Worker
     * Threads in the Thread Pool.
     */
    public void stopReactor(){
        _shouldRun = false;
        _pool.stopPool();
    }

    public static void main(String args[]) {
        if (args.length!=2) {
            System.err.println("Usage: java Reactor <thread pool size> <port>");
            System.exit(1);
        }
        
        try {
            Reactor reactor = new Reactor(Integer.parseInt(args[0]), Integer.parseInt(args[1]));
            reactor.start();
            System.out.println("Reactor is ready on port " + reactor.getPort());
            reactor.join();
        }
        catch (Exception e) {
            e.printStackTrace();
        }

    }
}

SimpleClient.java

import java.net.Socket;
import java.io.IOException;

/**
 * A simple client implementation
 */
public class SimpleClient {

    protected String _host;
    protected int _port;
    protected Socket _socket;

    /**
     * Creates a new SimpleClient object
     * @param host the server's host
     * @param port the server's port
     */
    public SimpleClient(String host, int port) {
        _host = host;
        _port = port;
    }

    /**
     * Connects to the server
     * @throws IOException in case of a connection failure
     */
    public void connect() throws IOException {
        _socket = new Socket(_host, _port);
    }

    /**
     * Disconnects from the server
     * @throws IOException in the case of a disconnection failure
     */
    public void disconnect() throws IOException {
        _socket.close();
    }

    /**
     * Sends a message to the server
     * @param message the message to send
     * @throws IOException in the case of sending failure
     */
    public void send(String message) throws IOException {
        _socket.getOutputStream().write(message.getBytes());
        _socket.getOutputStream().flush();
    }

    /**
     * Receives information from the server
     * @return the received message, or null of no message received
     * @throws IOException in the case of reception failure
     */
    public String receive() throws IOException {
        byte []buff = new byte[8192];
        int nBytes = _socket.getInputStream().read(buff);
        if (nBytes>0) {
            return new String(buff, 0, nBytes);
        }
        else {
            return null;
        }
    }

    public static void main(String args[]) {

        if (args.length!=2) {
            System.err.println("Usage: java SimpleClient <host> <port>");
            System.exit(1);
        }

        try {
            SimpleClient client = new SimpleClient(args[0], Integer.parseInt(args[1]));
            client.connect();
            for (int i=0; i<10; i++) {
                String message = "Hey!;";
                System.out.println("Sent \"" + message + "\"");
                client.send(message);

                String res = client.receive();
                System.out.println("Got  \"" + res + "\"");

                try {
                    Thread.sleep(500);
                }
                catch (InterruptedException ie) {
                    ie.printStackTrace(System.err);
                }
            }
            client.disconnect();
        } catch (IOException io) {
            io.printStackTrace(System.err);
        }
    }
}

ThreadPool.java

/**
 * A Pool where the Worker Threads are placed.
 * This pool contains an inner TaskQueue, from which the Worker Threads extract Tasks and
 * executeTask them.
 * The Thread Pool provides functionaliry for inserting Tasks into the Task Queue, starting
 * and stopping the Worker Threads.
 */
public class ThreadPool {
    protected Thread[] _pool;
    protected TaskQueue _taskQueue;
    protected volatile boolean _shouldRun;
    protected boolean _started;

    /**
     * Implementation of the Worker Thread.
     * These threads, when activated, constantly trying to extract Tasks from
     * the TaskQueue; when a Task is extracted, they executeTask it by invoking Task.executeTask();
     */
    private class WorkerThread extends Thread {
        private WorkerThread(String name) {
            //@todo setName(name);
            super(name);
        }

        public void run() {
            while (_shouldRun) {
                try {
                    Task task = _taskQueue.getTask();
                    task.executeTask();
                    System.out.println(getName() + ": executed");
                } catch (InterruptedException i) {
                    _shouldRun = false;
                } catch (TaskFailedException tf) {
                    _shouldRun = false;
                } catch (Exception e) {
                    e.printStackTrace(System.err);
                }
            }
        }
    }

    /**
     * Creates a new ThreadPool object
     * @param size the number of Worker Threads in the ThreadPool
     */
    public ThreadPool(int size) {
        _taskQueue = new TaskQueue();
        _pool = new WorkerThread[size];
        _shouldRun = true;
        _started = false;
    }

    /**
     * Starts all the Worker Threads in the ThreadPool.
     * If the Task Queue is empty, the Worker Threads will be waiting for Tasks to be entered.
     */
    public void startPool() {
        if (!_started) {
            _started = true;

            for (int i = 0; i < _pool.length; i++) {
                _pool[i] = new WorkerThread("WorkerThread_" + i);
                _pool[i].start();
            }
        }
    }

    /**
     * Causes the Worker Threads to stop
     */
    public void stopPool() {
        _shouldRun = false;

        for (int i = 0; i < _pool.length; i++) {
            _pool[i].interrupt();
        }
    }

    /**
     * Adds a Task to the ThreadPool's TaskQueue
     * @param task the Task to be added
     */
    public void addTask(Task task) {
        _taskQueue.addTask(task);
    }

    public int size() {
        int retVal = 0;

        for (int i = 0; i < _pool.length; i++) {
            WorkerThread worker = (WorkerThread) _pool[i];

            if ((worker != null) && worker.isAlive()) {
                retVal++;
            }
        }

        return retVal;
    }

    public void join(int millisec) {
        if (millisec > 0) {
            for (int i = 0; i < _pool.length; i++) {
                WorkerThread worker = (WorkerThread) _pool[i];

                if (worker != null) {
                    try {
                        worker.join(millisec);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        } else {
            while (!_taskQueue.isEmpty()) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}