Java : A callback API for epoll(), building on top of Java NIO

I just finished the first draft of the Java callback API that allows you to do asynchronous socket communication using the scalable epoll() mechanism of Linux. This is built on top of Java NIO, so it will take advantage of the best underlying network mechanism the OS has to offer. In Linux 2.6 + /FreeBSD this will likely be epoll().

Use the API if you need to get data from a large number of hosts quickly without the overhead of a thread for each host/connection. Having a thread per connection has serious issues as the number of simultaneous connections increases, specially if memory is scarce.

The Java NIO api can certainly be used for this purpose. But using that API is somewhat difficult and there are various caveats you have to guard against. You need to understand the Selector class, SelectionKey class, InterestOps class, SocketChannel class and their interplay. You need to know the logic for canceling SelectionKeys and setting InterestOps appropriately. And you need to understand the cryptic ByteBuffer class and its variants, possibly how the ByteArrayOutputStream works as well. You need to figure out how to use these classes to store the data separately for each connection, keeping in mind that the data will likely arrive mixed. You also should not call select if all sockets have been completely read, or select() will block - which means you will have to keep track of pending hosts.

The callback framework I developed aims to relieve you of all these details. It lets you focus on the flow of calls between the client and each host along with the application logic once you get data from the hosts.

Usage:
Extend the abstract class ConnectionParam. (Let's name it ConnectionParamImpl)
Create a ConnectionParamImpl object for each connection and stuff these into an array.
Call ASyncSocketHandler.connect_all(array_of_connections) - the argument is the above array.

Since we extended the ConnectionParam class, we need to implement some methods. These are simply the callback methods that tell us that data has either been written to the network, or read from the network. This is where we get a chance to examine the data and do what we need to do from the application's point of view.

    // returns the next op we're interested in: OP_READ / OP_WRITE / OP_CLOSE
    public abstract int connected_ok();
    public abstract int connect_failed();
    public abstract int write_ok(boolean eod);
    public abstract int read_ok(boolean eod);

Each of the over-ridden methods needs to return an int signaling the next network operation you are interested in. The available options are:

ASyncSocketHandler.OP_READ
ASyncSocketHandler.OP_WRITE
ASyncSocketHandler.OP_CLOSE

This is a simple usage scenario that gets output from a bunch of known time of day servers.

There are two objects you need to implement:

ConnectionParamImpl class and ASyncEchoClient class.
ConnectionParamImpl represents a connection to a single host.
ASyncEchoClient is the application entry point.

import java.io.UnsupportedEncodingException;

/**
 * @author: thushara
 * @version: Nov 8, 2009
 */
public class ConnectionParamImpl extends ConnectionParam {

    public ConnectionParamImpl(byte[] ip, int port) {
        super(ip, port);
    }

    // returns the next op we're interested in: OP_READ / OP_WRITE / OP_CLOSE
    // after connecting, we can proceed to read
    public int connected_ok() {
        return ASyncSocketHandler.OP_READ;
    }

    public int connect_failed() {
        return ASyncSocketHandler.OP_CLOSE;
    }

    // returns the next op we're interested in: OP_READ / OP_WRITE / OP_CLOSE
    // param: eod = true iif if the complete request was written to the socket
    //              if request is not fully written, return signal to write again
    //              else, we can go on to read, return read singnal.
    public int write_ok(boolean eod) {
        return eod ? ASyncSocketHandler.OP_READ : ASyncSocketHandler.OP_WRITE;
    }

    // params: eod = true iif all data was read from the socket, socket has no more data
    //               if all data is read, this socket won't return any more data so
    //               we should close the socket, signal such
    //               else we should continue reading, signal such.
    public int read_ok(boolean eod) {
        if (eod) {
            byte[] resp = getResponse();
            try {
                System.out.println(new String(resp,"UTF-8"));
            } catch (UnsupportedEncodingException e) {
                System.out.println("bad chars - non utf-8");
            }
        }
        return eod ? ASyncSocketHandler.OP_CLOSE : ASyncSocketHandler.OP_READ;
    }

}

/**
 * @author: thushara
 * @version: Nov 8, 2009
 */
public class ASyncEchoClient {
    public static void main(String[] args) {

        ConnectionParam.verbose = true;
        ASyncSocketHandler.verbose = true;

        ASyncSocketHandler hndlr = new  ASyncSocketHandler();
        int port = 13;

        String[] ips = {"64.90.182.55", "129.6.15.28", "129.6.15.29", "206.246.118.250", "64.236.96.53", "68.216.79.113", "208.66.175.36",
                "173.14.47.149", "64.113.32.5", "132.163.4.101", "132.163.4.102", "132.163.4.103", "192.43.244.18", "128.138.140.44",
                "128.138.188.172", "198.60.73.8", "131.107.13.100", "207.200.81.113", "69.25.96.13", "64.125.78.85", "64.147.116.229"};
        ConnectionParam[] params = new ConnectionParamImpl[ips.length];

        for (int j=0; j<ips.length; j++) {
            String ip = ips[j];
            String[] sb = ip.split("\\.");
            byte[] bytes = new byte[sb.length];
            for (int i=0; i<sb.length; i++) {
                bytes[i] = (byte)Integer.parseInt(sb[i]);
            }
            params[j] = new ConnectionParamImpl(bytes, port);
        }

        try {
            hndlr.connect_all(params);
        } catch (Exception e) {
            System.err.println("unexpected :" + e.getMessage());
            e.printStackTrace();
        }
    }
}

Notice that you don't see a single reference to a socket or a byte buffer here. All the communication details are wrapped in ASyncSocketHandler.java and ConnectionParam.java.

This is the output you should see, basically the time of day as reported by each host:

55145 09-11-10 23:13:25 00 0 0 434.7 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 401.7 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 398.1 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 402.8 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 383.4 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 386.1 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 348.4 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 347.4 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 348.2 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 347.9 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 339.2 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 316.3 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 306.9 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 310.7 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 313.4 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 298.7 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 295.4 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 274.6 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 268.1 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 268.0 UTC(NIST) * 


55145 09-11-10 23:13:25 00 0 0 275.9 UTC(NIST) * 

This is the framework implementation. There are two files.

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetEncoder;
import java.nio.charset.CharacterCodingException;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.net.InetSocketAddress;

/**
 * @author: thushara
 * @version: Nov 6, 2009
 */
public abstract class ConnectionParam {
    private byte[] ip;
    private int port;
    private ByteBuffer request;
    private ByteBuffer response;
    private int respBufSize = 1024;
    private ByteArrayOutputStream outStrm;
    private SocketChannel client;

    public static boolean verbose = false;

    public ConnectionParam(byte[] ip, int port) {
        this.ip = ip;
        this.port = port;
        response = ByteBuffer.allocate(respBufSize);
        outStrm = new ByteArrayOutputStream();
    }

    public ConnectionParam(byte[] ip, int port, String request) throws CharacterCodingException {
        this(ip,port);
        setRequest(request);
    }

    public ConnectionParam(byte[] ip, int port, String request, int respBufSize) throws CharacterCodingException {
        this(ip,port);
        setRequest(request);
        this.respBufSize = respBufSize;
    }

    public ConnectionParam(byte[] ip, int port, int respBufSize) {
        this(ip,port);
        this.respBufSize = respBufSize;
    }

    public void setRequest(String request) throws CharacterCodingException {
        Charset charset = Charset.forName( "UTF-8" );
        CharsetEncoder encoder = charset.newEncoder();
        this.request = encoder.encode(CharBuffer.wrap(request));
    }

    public byte[] getResponse() {
        int respSize = outStrm.size() + response.position();
        byte[] out = new byte[respSize];
        System.arraycopy(outStrm.toByteArray(),0,out,0,outStrm.size());

        int prevPos = response.position();
        response.flip();
        response.get(out,outStrm.size(),response.limit());
        response.position(prevPos);
        response.limit(response.capacity());
        
        return out;
    }

    private InetAddress getAddr() {
        try {
            return InetAddress.getByAddress(ip);
        } catch (UnknownHostException e) {
            return null;
        }
    }

    private void store() throws IOException {
        response.flip();
        byte[] bytes = new byte[response.limit()];
        response.get(bytes);
        outStrm.write(bytes, 0, bytes.length);
        response.clear(); //initialize for reading from socket again
    }


    public SocketChannel getSocketChannel() throws IOException {
        if (client != null) return client;

        client = SocketChannel.open();
        client.configureBlocking(false);
        client.connect(new InetSocketAddress(getAddr(),port));
        return client;
    }

    public String toString() {
        InetAddress addr = null;
        try {
            addr = InetAddress.getByAddress(ip);
        } catch (UnknownHostException e) {
            return "";
        }
        return addr.toString() + ":" + port;
        //return String.format("%d.%d.%d.%d:%d", ip[0], ip[1], ip[2], ip[3], port);
    }

    //returns true if all data is written to the socket
    public boolean write() throws IOException {
        int numBytes = getSocketChannel().write(request);
        if (numBytes == 0 && verbose) {
            System.err.println("nothing to write to server");
        }
        return !request.hasRemaining();
    }

    //returns true if all data is read, our end of the socket should be closed
    public boolean read() throws IOException {
        int numBytes = 0;
        try {
            numBytes = getSocketChannel().read(response);
        } catch (IOException e) {
            //System.err.println("end of data");
            store();
            return true;
        }
        if (numBytes == 0 && verbose) {
            System.err.println("nothing to read from server");
        }
        if (numBytes == -1) {
            //System.err.println("end of data");
            store();
            return true;
        }

        while (response.remaining() == 0 && numBytes > 0) {
            store();
            //read again from socket
            numBytes = getSocketChannel().read(response);
        }

        return false;

    }

    // returns the next op we're interested in: OP_READ / OP_WRITE / OP_CLOSE
    public abstract int connected_ok();
    public abstract int connect_failed();
    public abstract int write_ok(boolean eod);
    public abstract int read_ok(boolean eod);
}

import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.SelectionKey;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.Iterator;
import java.net.ConnectException;

/**
 * @author: thushara
 * @version: Nov 6, 2009
 */
public class ASyncSocketHandler {
    private Selector selector;
    private HashSet<ConnectionParam> clients = new HashSet<ConnectionParam>();
    private int timeout;

    public static final int OP_READ = SelectionKey.OP_READ;
    public static final int OP_WRITE = SelectionKey.OP_WRITE;
    public static final int OP_CLOSE = 128; //leave enough room for possible expansion of SelectionKey enum

    public static boolean verbose = false;

    public void connect(ConnectionParam connParm) throws IOException {
        SocketChannel client = connParm.getSocketChannel();
        SelectionKey clientKey = client.register(selector, SelectionKey.OP_CONNECT);
        clientKey.attach(connParm);
        clients.add(connParm);
    }

    public void connect_all(ConnectionParam[] connParms) throws IOException {
        selector = Selector.open();
        for (ConnectionParam connParm : connParms) {
            connect(connParm);
        }

        while (clients.size() > 0) {
            if (verbose) {
                System.out.println("waiting for sockets to be ready. currently waiting for:");
                dumpClients();
            }
            int numReady = 0;

            try {
                numReady = selector.select(timeout);
            } catch (IOException e) {
                if (verbose) {
                    System.err.println("select failed, unrecoverable!" + e.getMessage());
                }
                break;
            }
            if (numReady <= 0) {
                if (verbose) {
                    System.err.println("No sockets ready for " + timeout/1000 + " secs, consider dead remotes, bailing...");
                }
                break;
            }

            Set<SelectionKey> keys = selector.selectedKeys(); // can't recover from exception thrown here
            Iterator i = keys.iterator();
            ConnectionParam connParm = null;

            SelectionKey key = null;
            while (i.hasNext()) {
                try {
                    key = (SelectionKey)i.next();
                    i.remove();
                    SocketChannel channel = (SocketChannel)key.channel();
                    connParm = (ConnectionParam)key.attachment();
                    if (key.isConnectable()) {
                        if (channel.isConnectionPending()) {
                          boolean ok = false;
                          try {
                            ok = channel.finishConnect();
                          } catch (ConnectException e) {
                              if (verbose) {
                                System.err.println("error processing " + connParm.toString());
                                e.printStackTrace();
                              }
                          }

                          if (!ok) {
                              if (verbose) {
                                System.err.println("not connected, giving up");
                              }
                              //cleanup
                              if (key != null) {
                                  try {
                                      key.channel().close();
                                  } catch (IOException x) {}
                                  key.cancel();
                                  clients.remove(key.attachment());
                                  continue;
                              }
                              connParm.connect_failed();
                          } else {
                              int op = connParm.connected_ok();
                              key.interestOps(op);
                          }
                        }
                    }
                    if (key.isWritable()) {
                        boolean completedWrite = connParm.write();
                        int op = connParm.write_ok(completedWrite);
                        if (op == ASyncSocketHandler.OP_CLOSE) {
                            channel.close();
                            key.cancel();
                            clients.remove(connParm);
                        } else {
                            key.interestOps(op);
                        }
                    }
                    if (key.isReadable()) {
                        boolean endOfData = connParm.read();
                        int op = connParm.read_ok(endOfData); 
                        if (op == ASyncSocketHandler.OP_CLOSE || endOfData) {
                            channel.close();
                            key.cancel();
                            clients.remove(connParm);
                        } else {
                            key.interestOps(op);
                        }
                    }

                } catch (IOException e) {

                }
            }
        }
    }

    private void dumpClients() {
        for (ConnectionParam connParm : clients) {
            System.out.println(connParm.toString());
        }
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }

}

I'm making the project available here:

Java : fetching web pages using NIO - timeouts

I'm developing a call back framework on top of Java NIO, that will allow for easier implementations of asynchronous page downloads.

As you start adding sockets to a single selector, you get into request time-out issues. This happens because you normally add all the sockets to the selector and then wait for sockets that are ready. Some sockets, after waiting for a few seconds for the request, will timeout at the remote end.

The HTTP response code for this is 408 - it is easy to handle this by removing the SelectionKey from the selector, and adding the socket back to the selector. Since, the application will now be dealing with fewer sockets, the next time, the socket will most probably not time out.

The latency is mostly in domain name resolution. I will next try doing name resolution totally outside socket registration, so I don't eat into the connection time with the slow DNS lookup calls. This should reduce the number of 408 responses, and I should be able to add more sockets to the selector.

Resolving domain names quickly with Java

Domain name resolution can take upward of 5 seconds depending on your ISP. Generally, when the resolution works, the DNS response comes back quickly, within a second. Any DNS entry not discovered within a second is likely to be dead.

Java provides InetAddress.getByName(domain) that does the DNS resolution for the given domain. There is no timeout you can specify to this function.

In the interest of providing a responsive user interface, as well as improve performance on background tasks, you can hack a timeout using a Thread. Feel free to re-use this code:

    public class DNSResolver implements Runnable {
        private String domain;
        private InetAddress inetAddr;

        public DNSResolver(String domain) {
            this.domain = domain;
        }

        public void run() {
            try {
                InetAddress addr = InetAddress.getByName(domain);
                set(addr);
            } catch (UnknownHostException e) {
                
            }
        }

        public synchronized void set(InetAddress inetAddr) {
            this.inetAddr = inetAddr;
        }
        public synchronized InetAddress get() {
            return inetAddr;
        }
    }

Use this class as follows:

                DNSResolver dnsRes = new DNSResolver(host);
                Thread t = new Thread(dnsRes);
                t.start();
                t.join(1000);
                InetAddress inetAddr = dnsRes.get();

Basically, we start a thread that executes the blocking call to get the domain name. We wait for the thread to exit under 1 second (1000 ms). If the DNS resolution happens under a second, the thread will exit. If not, control will return to the main thread due to the 1000ms timeout. The thread will keep running, and eventually exit. But the main program has stopped caring at that point. It (main thread) will continue after 1 second considering the domain name to be unresolved.

Caveat:
Note that if you resolve a large number of domains that are unreachable, soon you will have lots of threads taking up memory and you could run out of JVM heap. So keep that in mind as you use this class.

Possible usage scenario:
If you are doing socket communication using nio, and you need to resolve DNS for each socket connection attached to the selector, it is important you quickly timeout DNS, as otherwise some sockets might get closed at the remote end. This is specially true for HTTP connections, if you connect to a HTTP server and do not send a request for a few seconds, the server will close the remote end of the socket.

Mac/SnowLeopard and Wireshark

Wireshark has to be tweaked a bit due to permission issues on the Berekley Packet Filters. This has the info.

Since these settings are lost on reboot, a start up item can be created as shown here.

Also, if a module not found dialog box appears, follow these steps (taken from the long thread here):

You need to set the path to the folder that Wireshark looks in for MIBs &c, because the default in version 1.0.6 for Mac OS X is incorrect. In Wireshark, do the following:

• From the Edit menu, select Preferences...
• In the left pane of the Wireshark: Preferences window, click on Name Resolution
• For SMI (MIB and PIB) paths, click the Edit button
• In the SMI Paths window, click the New button
• In the SMI Paths: New window, in the name text box, type /usr/share/snmp/mibs/ and click OK
• Click OK
• Click OK
• From the File menu, select Quit

I found that it was also necessary to quit and restart X11 for the changed Wireshark preferences to take effect:

• From the X11 menu, select Quit X11

This gets rid of the loading the MIBS errors for me. YMMV. By the way, Wireshark is supposed to still be useable, even if you can’t get rid of these errors.

many options of lsof

this is handy reference to the ways of using lsof.

Mac and launchd - removing programs that refuse to be killed

Mac uses the launchd daemon to start processes on boot time, much like the init.d scripts in Linux. Except, launchd has a configuration file (with an extention .plist) for each such process, and these files are in a few different places. So if you want to stop an auto launch, you will be hunting around, which I just did.

The official doc states all the locations a plist file.

In my case, I was using a educational prototype from UW called vanish. This spawns a couple of processes using launchd. So if I kill the processes, they come right back up. I found the plist files under ~/LaunchAgents and removing them did the trick.

Java ByteBuffer : how does this work?

The first time you encounter the ByteBuffer, you may run into some surprises. The function that flips most folks is in fact ByteBuffer.flip(). To understand the flip() and to be not flipped by it and other such idioms, we will look at what this class is and how it should be used.

Basically, a ByteBuffer allows us to read data repeatedly from some input, like a non-blocking socket. ByteBuffer keeps track of the position of the last byte written, so you don't need to. You can keep writing to the same ByteBuffer and rest assured that previous data will not be over-written.

This is pretty handy in asynchronous I/O (using Java NIO package) as data from asynchronous sockets don't always arrive all at once. We need to map buffers to sockets and keep reading until there is no more data from the remote end.

So what about this flip()? Well, the way the ByteBuffer class was designed, data is read to the buffer starting at position and upto limit. Data is written starting at position and upto limit as well. So, if you followed that, after reading some data from a socket, the ByteBuffer position would be advanced, and reading now will not get any data as position is at the end of the buffer. So flip() basically sets the position to 0 (start), and limit to the position (previous position to be exact, or rather the end of useful input).

So think about read and write operations on the ByteBuffer manipulating data within position and limit and you will see more clearly the need to flip once in a while.

Happy flipping!