High-Performance I/O with Java NIO

To watch over the Channels, the NIOWorker thread needs to perform "selects" by calling the selectNow method on the Selector. Performing "selects" tells the Selector to update all of its keys to reflect which Channels are ready and which are not, allowing the NIOWorker thread to stack the deck by only working on the Channels that are ready. The number of ready keys is returned from this method. Listing Eight shows the main run method for the NIOWorker thread that performs the select.

Listing Eight

public void run() {
    while (true) {
        try {
            int num = selector.selectNow();
            if (num > 0) {
                processKeys();
            } else {
                Thread.yield();
            }
        } catch (IOException ioe) {
            System.err.println("Unable to select: " + ioe.toString());
        } catch (InterruptedException ie) {
            // Continue processing
        }
    }
}

The processKeys method handles all of the Channels that signal the Selector that they are ready to perform I/O. Because the Selector handles multiple Channels, the processKeys method performs the I/O for all Channels that are ready. Listing Nine shows the processKeys method that is responsible for iterating over all the SelectionKeys and handling each one.

Listing Nine

protected void processKeys() {
    Set keys = selector.selectedKeys();
    for (Iterator iter = keys.iterator(); iter.hasNext();) {
        SelectionKey key = (SelectionKey) iter.next();
        iter.remove();
        WorkState state = (WorkState) key.attachment();
        SocketChannel channel = (SocketChannel) key.channel();
        try {
            if (state.isTimedOut()) {
                finished(channel, key, state);
                continue;
            }
            boolean connectable = key.isConnectable();
            if (connectable && channel.finishConnect()) {
                // If the Channel is connected, setup the Channel to 
                // write the HTTP message to the remote server
                key.interestOps(SelectionKey.OP_WRITE);
            } else if (key.isWritable()) {
                // If the Channel is finished writing, setup the
                // Channel to read the HTTP response
                if (doWrite(channel, state)) {
                    key.interestOps(SelectionKey.OP_READ);
                }
            } else if (key.isReadable()) {
                // If the Channel is finished reading, call finished
                // to complete the work
                if (doRead(channel, state)) {
                    finished(channel, key, state);
                }
            }
        } catch (IOException ioe) {
            System.err.println("Failure during IO operation");
            finished(channel, key, state);
        }
    }
}

The Selector object provides the ability to retrieve all the keys that are ready to perform I/O using the selectedKeys() method. This method returns a Set containing all the keys that are ready to perform I/O. This Set is populated with the ready keys whenever a select operation is performed on the Selector.

One important detail about the processKeys method is that during the loop iteration, the WorkState attachment is retrieved from the SelectionKey and used to verify that the work has not timed out. If it has timed out, the finished method is called to signal that the work is complete, which in this case is due to a timeout rather than success.

The doWrite, doRead, and finished methods are the last methods that complete the NIOWorker. The doWrite method writes the HTTP-request String to the channel. Listing Ten shows the doWrite method.

Listing Ten

protected boolean doWrite(SocketChannel channel, WorkState state)
throws IOException {
    int rem = state.work.httpRequest.remaining();
    int num = channel.write(state.work.httpRequest);
    // Continue writing until everything has been written
    return (num == rem);
}

The doRead method is responsible for performing a read from a Channel and signaling if everything has been read from the Channel or not. Listing Eleven is the doRead method.

Listing Eleven

protected boolean doRead(SocketChannel channel, WorkState state)
throws IOException {
    buffer.clear();
    decoder.reset();
    boolean done = false;
    int num = channel.read(buffer);
    if (num == -1) {
        state.success = true;
        done = true;
    } else if (num > 0) {
        buffer.flip();
        String data = decoder.decode(buffer).toString();
        state.buffer.append(data);
    }
    return done;
}

The doRead method is a standard NIO implementation of reading from a Channel. It reads zero or more bytes from the Channel and also determines if the Channel has any more to read. Once this code is certain that there is nothing more to read, it returns True to signal to the processKeys method that it should call the finished method. This code goes one step further and sets the success flag on the WorkState object. This is used in the finished method to determine whether the HTTP-response should be parsed.

The final finished method notifies the execute thread that its work was complete, whether successfully or not. The success flag set in the doRead method is used to determine if the HTTP-response is parsed out. Listing Twelve is the finished method.

Listing Twelve

protected void finished(SocketChannel channel, 
                                  SelectionKey key, WorkState state) {
    key.cancel();
    try {
        channel.close();
    } catch (IOException ioe) {
        System.err.println("Failed to close socket: " + ioe.toString());
    } finally {
        Work work = state.work;
        synchronized (work) {
            // Only if the Work was successful, parse out the HTTP response
            if (state.success) {
                String result = state.buffer.toString();
                work.out = parseHTTPResponse(result);
            }
            work.notify();
        }
    }
}

NIO does not handle the HTTP-request generation and HTTP-response parsing automatically. That work is left up to the NIO implementation or the client. This design handles the generation of the HTTP-request and parsing of the HTTP-response messages. The buildHTTPRequest and parseHTTPResponse methods were added here for illustration. The work of generating and parsing the HTTP messages is slightly more complex but is available in the full version of the code, within the HTTPParser class and the classes that implement the HTTPMethod interface.

Listing Thirteen is an example of an execute thread calling the NIOWorker to perform some work.

Listing Thirteen

public class NewIO {
    // Assume someone else setup the NIOWorker and passed it in
    public String execute(NIOWorker worker) {
        try {
            // Construct the URL and pass it to the worker
            URL url = new URL("http://example.orbitz.com");
            return worker.doWork(url, 15000); // Wait 15 seconds
        } catch (IOException ioe) {
            System.err.println(ioe.toString());
        }
    }
}

Performance

To fully appreciate the benefits gained between standard I/O and NIO, I collected statistics using each design employed. These statistics were collected by running each of the actual implementations through a series of load tests in a fixed environment. Table 1 presents the results from those tests. All values are the number of transactions completed by an implementation per minute.

Table 1: Load test results.

In addition to a load test, the standard I/O and the NIO solutions were both used in the Orbitz production environment for some period of time. I was able to monitor each one and collect statistics for load and latency. Table 2 shows the statistics from the Orbitz.com production environment.

Table 2: Statistics from the Orbitz.com production environment.

Additional Thoughts

The code I present here is only part of the complete solution used in the Orbitz.com application. I simplified some of the code, but have retained the major work required to implement a multiplexed NIO solution.

NIO is not a simple API to use and I encountered many issues during testing that required fine tuning. A few things I found that should be kept in mind are:

• Prevent adding too much work to the Selector. This can cause memory leaks that result in a Java VM crash. The Selector can be queried to determine how many keys it currently has and that number can be compared to a configurable threshold.
• Determine if your run method runs in a tight loop. The earlier code runs in a tight loop and this may be undesirable for low- to medium-volume applications. To cause a looser loop, one of the Selector methods with a timeout can be used. This will cause the call to that method to block until a Channel is ready.
• You may want to use a tight loop but wait when the Selector is empty inside the run method. Inside the add methods, the NIOWorker thread can then be notified when work is successfully added.
• Increase performance by moving the ByteBuffer and CharsetDecoder from the doRead method to instance variables on the NIOWorker object.
• Check the Selector for each loop iteration in the main run method to determine if any work has timed out. This check will prevent the Selector from becoming full of invalid connections due to network failures.
• The full version of the code, available at the top of the first page, includes all of these implementation details.

Pitfalls

One major NIO pitfall to be aware of is that you should always use multiplexed NIO. Putting NIO code into multiple threads reduces overall performance and can cause excessive load. I experimented with using NIO in multiple threads and observed that due to enormous amounts of context switching the load nearly tripled, while latency and throughput remained unchanged.

Conclusion

A multiplexed NIO solution is an ideal solution for the Orbitz application requirements because of the efficiencies of multiplexed I/O operations. Multiplexed I/O means that there is a single thread doing many I/O operations by leveraging a Selector to perform the multiplexing. Using a Selector, multiple Channels executing I/O operations are managed and watched over concurrently. This means that at no point is the Java VM ever waiting on Channels that are not ready to perform I/O; because the Selector knows precisely the Channels that are ready to perform I/O and those that are not. Because there is only a single thread performing I/O, the problem of context switching between multiple threads performing I/O operations has been eliminated. All of these conditions greatly reduce overhead and latency while increasing total throughput over a standard I/O implementation.

Acknowledgments

Thanks to David P. Thomas of Orbitz for reviewing this article.

Brian is a senior architect with Orbitz.com. He can be contacted at [email protected].