The NIO library, a feature added to Java as part of JDK 1.4, offers an alternative, high-performance technique for handling input/output (I/O) operations. This library supports working directly with native channels, sockets, and direct memory buffers to increase the performance and throughput of I/O. In addition, it offers the ability to do true nonblocking I/O operations. However, if used incorrectly, NIO can yield little performance increase and can cause an increase in the load and reduce stability of an application. In this article, I discuss how Orbitz (the company I work for) used NIO to increase performance and stability by replacing an older standard I/O implementation. Orbitz communicates with various third-party backend systems to retrieve data for its customers. During the course of handling a single customer request, the Orbitz.com application performs time-sensitive I/O to fetch the customer's response. With a large number of concurrent users, the application can be performing hundreds of I/O operations at the same time.
The Orbitz.com application handles concurrent user requests by allowing each request to be performed in a separate thread. These threads, called "execute threads," can be configured such that each request is handled by a new thread or by a thread from a thread pool. In either case, it is vital to ensure that no thread is running for an extended period of time. When threads take too long processing a single user request and a new thread is used for each request, the Java VM has the possibility of crashing when excessive threads are created. Likewise, if a pool is being used and threads from that pool take too long while executing, other users are forced to wait until a thread has finished and can be reused.
Standard I/O
The Orbitz.com application was initially implemented with a standard I/O design using Java 1.3. To guarantee timeouts for I/O operations, it employed a separate thread to perform the I/O operations. The execute threads are then joined on this new thread using a timeout. With this approach, it prevented the execute thread from hanging when I/O operations encountered problems. Listing One shows the original thread that performed the I/O operations. It has been simplified from its original version.
Listing One
public class IOThread extends Thread {
private String result;
public synchronized String getResult() {
return result;
}
public void run() {
try {
// Connect to the remote host and read in the data
URL url = new URL("http://example.orbitz.com");
URLConnection connection = url.openConnection();
String localResult = doRead(connection);
synchronized (this) {
result = localResult;
}
} catch (IOException ioe) {
System.err.println(ioe.toString());
}
}
private String doRead(URLConnection connection) throws IOException {
StringBuffer buf = new StringBuffer();
InputStream is = connection.getInputStream();
BufferedReader br = new BufferedReader(new InputStreamReader(is));
char[] cbuf = new char[4096];
int n;
do {
n = br.read(cbuf);
if (n > 0) {
buf.append(cbuf, 0, n);
}
} while (n >= 0);
return buf.toString();
}
}
By calling the join method on the IOThread, one of the execute threads waits until the IOThread has finished running. A timeout is specified when the join method is called to ensure that the execute threads do not wait forever for an IOThread to complete. Listing Two shows an example of how an execute thread used an IOThread.
Listing Two
public class OldIO {
public String execute() {
IOThread thread = new IOThread();
thread.start();
try {
thread.join(15000); // Wait for 15 seconds
} catch (InterruptedException ie) {}
return thread.getResult();
}
}
The first version of the execute threads created new instances of the IOThread class for each I/O operation. Using the initial implementation it was observed that the servers were periodically experiencing high thread counts. It had initially been assumed that the maximum number of threads that could be created in a Java VM running this code would be 2N, where N was the number of execute threads. After some investigation, I found that an instance of the IOThread would remain in memory if it experienced network problems and was hung while performing an I/O operation. In addition, the execute thread correctly timed-out and the next execute thread attempted the same operation, with the same results. The net result was that the Java VM created an enormous number of threads and eventually ran out of memory.
To reduce the thread count, I added a thread pool that stored instances of the IOThread class. I also changed the execute thread code so that it checked out an IOThread instance from the pool, used it, and then checked it back in. This not only reduced the high thread count, but also reduced the overhead of thread creation. Listing Three shows the second version of the execute thread that used the pool of IOThreads (the implementation of a thread pool is out of scope of this article).
Listing Three
public class PoolOldIO {
private static ThreadPool pool = new ThreadPool();
public String execute() {
IOThread thread = pool.checkOut();
String result;
try {
thread.execute(15000); // Wait for 15 seconds
} catch (InterruptedException ie) {
// Smother, okay to be interrupted
} finally {
result = thread.getResult();
pool.checkIn(thread); // Must always check-in the thread
}
return result;
}
}
This improvement still has an underlying problem. If the network experiences major failures, all instances of IOThread in the pool quickly become stalled and execute threads begin to fail during check-out because there are no more IOThreads in the pool. Additionally, when I began testing this new design under heavy load, I found that it still had problems that reduced overall performance. The issue was excessive context switching.
All the implementations thus far force the Java VM to perform increased context switching because either the execute threads or the IOThreads are all performing I/O operations concurrently. The largest problem with this concurrency is that the Java VM doesn't know which threads are ready to perform I/O operations and which are not. Therefore, the Java VM might put a thread onto the processor that isn't ready to perform I/O operations. Meanwhile another thread that is waiting to run is ready to perform I/O operations. Because Java does not provide the ability to "stack the deck" by moving threads that are ready to perform I/O to the top of the list, it is possible that a thread, which is ready to perform I/O, might wait long enough that it times out.
NIO
The NIO package added to Java 1.4 offers Java developers the ability to perform true nonblocking I/O and guarantee timeouts for I/O operations. It also works with low-level operating-system constructs to increase performance and reduce latency. In addition, it provides the ability to monitor multiple I/O operations concurrently, also known as "multiplexing."
Multiplexed NIO is a technique that moves all the I/O work into a single thread that watches over many I/O operations executing concurrently. A multiplexed NIO design uses the java.nio.channels.Selector class in conjunction with subclasses of the java.nio.channels.SelectableChannel class. The Selector class lets the application watch many I/O connections (subclasses of the SelectableChannel class and also known as "Channels") and organize these Channels into those that are ready to perform I/O and those that are not.
I decided to redesign to use NIO and a single thread to perform all the I/O operations. Listing Four shows the skeleton body of the NIO thread.
Listing Four
public class NIOWorker extends Thread {
private Selector selector;
public NIOWorker () throws IOException {
selector = Selector.open();
}
...
}
As with the threaded standard I/O design, the execute threads need a way to supply work to the NIOWorker in the form of a URL they want to contact. The NIOWorker then connects to the URL, sends an HTTP message to the server, and then reads the HTTP-response. After it completes that work, it passes back the result to the execute thread. One unit of work is designated using an inner class within the NIOWorker thread that lets a URL be passed in and the contents to be passed back. Listing Five shows the inner class of the NIOWorker that represents one unit of work.
Listing Five
public static class Work {
public URL in;
public String out;
public ByteBuffer httpRequest;
}
Supplying work to the NIOWorker is not a simple process and requires two methods. Because the NIOWorker thread is a separate thread that handles all the I/O operations, execute threads must add their work and then wait for the NIOWorker thread to process it. Therefore, the NIOWorker class has one method that forces the execute threads to wait, and a second method that actually adds the work. This design waits and notifies using the Work object as the monitor. Listing Six shows the two add methods that force the execute threads to wait and add the work to the Selector.
Listing Six
public String doWork(URL url, long timeout) {
Work work = new Work();
work.in = url;
String result = null;
synchronized (work) {
work.httpRequest = buildHTTPRequest(work);
// If the work was added successfully, call wait on the work object
// which forces the calling thread to wait (i.e. the execute thread)
if (add(work, timeout)) {
try {
work.wait(timeout);
} catch (InterruptedException e) {
System.err.println("NIO operation interrupted");
}
result = work.out;
}
}
return result;
}
protected boolean add(Work work, long timeout) {
SocketChannel channel = null;
try {
URL url = work.in;
InetSocketAddress addr = new InetSocketAddress(url.getHost(), 80);
channel = SocketChannel.open();
channel.configureBlocking(false);
channel.connect(addr);
WorkState state=new WorkState(System.currentTimeMillis(),timeout,work);
channel.register(selector, SelectionKey.OP_CONNECT, state);
} catch (IOException ioe) {
if (channel != null) {
try {
channel.close();
} catch (IOException ioe2) {
System.err.println("Unable to close channel: " + ioe2);
}
}
System.err.println("Channel creation or registration failed: " + ioe);
return false;
}
return true;
}
This code creates a SocketChannel, which is a Channel that extends SelectableChannel, and registers the Selector with it. This Channel is set to nonblocking so that operations performed on Channel do not block until complete but rather return as quickly as possible. Registering a Selector with a Channel informs the Channel to begin a specific I/O operation and also tells the Channel to signal the Selector once the operation is complete. Here the code is setting the Channel to begin an OP_CONNECT operation, which forces the Channel to connect to a remote socket.
During the registration, this code also constructs a small class that holds some state information for the I/O operation. This is called an "attachment" and can be any object. The attachment is associated with a specific Selector/Channel relationship. This Selector/Channel relationship is called a "SelectionKey" (also known as "key"). It stores the type of operation the Channel is performing as well as the attachment. The WorkState object is what the NIOWorker uses as an attachment. WorkState lets the results that are read from the Channel be stored in a StringBuffer. It also provides a mechanism to monitor timeouts so that the NIOWorker thread does not continue working on something that has timed out. Listing Seven shows the WorkState class.
Listing Seven
public static class WorkState {
public final StringBuffer buffer = new StringBuffer();
public final Work work;
public final long start;
public final long timeout;
public boolean success = false;
public WorkState(long start, long timeout, Work work) {
this.start = start;
this.timeout = timeout;
this.work = work;
}
public boolean isTimedOut() {
return (System.currentTimeMillis() - start >= timeout);
}
}
