Non-Blocking I/O Made Possible in Java

Non-Blocking I/O Made Possible in Java

In this article we will review non-blocking IO, a feature of java.nio (New I/O) package that is a part of Java v1.4, v1.5 and v1.6 and introduce the java.nio.file (NIO.2) package. NIO.2 will be included in the upcoming Java SE7 ("Dolphin") release.

Prior to JDK1.4, threads engaged in network I/O exhibited the following characteristics:

• " multiple threads for polling devices for readiness
• required dedicating a thread to each socket connection
• blocking until data is available

  class Server implements Runnable {
    public void run() {
    try {
       ServerSocket ss = new ServerSocket(PORT);
       while (!Thread.interrupted())
        new Thread(new Handler(ss.accept())).start();
        // one thread per socket connection
   // every thread created this way will essentially block for I/O
    } catch (IOException ex) { /* ... */ }
  }

With functionality introduced in JSR-51, it is possible to set Channels in a non-blocking mode, under the watch of a Selector that has the ability to recognize/sense when one or more channels become available for data transfer. This frees the application from having to dedicate threads for blocking on devices awaiting data or implementing expensive polling processes. The following UML diagram shows the significant methods of Selector, SelectableChannel and SelectionKey.

Figure 3. Selector Topology

Some notes about non-blocking I/O:

• In non-blocking mode, an I/O operation never blocks (methods return immediately),
• In non -blocking mode, an I/O operation may transfer fewer bytes than were requested (partial read or write) or possibly no bytes at all.
• FileChannel does not support non-blocking file operations.
• Non-Blocking Polling API (JSR-51)is not the same as asynchronous API (JSR-203 - part of JDK 1.7).

The Selector for Single-Thread Multi-Socket I/O

A Selector allows processing of multiple sockets I/O read & write operations in a single thread, also known as multiplexing, solving the limitation of 'one dedicated thread per socket connection.' It manages SelectableChannels (gateway for registering/deregistering channels), SelectionKeys (glue that associates Selectors, Channels and their I/O events) and monitors concurrent thread executions involved in network operations.

Figure 5. Selector maintains SelectionKeys & Channel Bindings

A Selector is modeled after 'Reactor' role in the Reactor pattern--it responds to I/O events by dispatching the appropriate handler. The channels declare interest in I/O events with the Selector to be indicated when they are ready for I/O. Selector does not initiate the actual I/O operation, the channels do.

A Selector maintains three kinds of key-sets.

1. key-set : keys representing the current channel registrations of this selector
2. selected-key set: each key's channel was detected to be ready for at least one of the operations identified in the key's interest set during a prior selection operation.
3. cancelled-key set: keys that have been cancelled but whose channels have not yet been deregistered

Using Selectors

When a thread is ready for I/O, use one of the selection methods--select(), select(timeout), or selectNow() to identify the channels that are ready to perform one or more of the operations previously declared in the interest set. The host operating system is queried for this status. Keys may be added to and removed from the key sets using interestOps(int). After a key change, a selection method should be invoked again to obtain status on the changed channels. While Selectors are safe for use by multiple concurrent threads, their key sets are not.

A more detailed discussion of keys and interest sets follows below.

Let's look at some important operations of a Selector.

open() creates a Selector object by calling SelectorProvider. openSelector()

select()/select(timeout) blocks until at least one channel is selected, or selector's wakeup method is invoked, or the current thread is interrupted or the given timeout period expires.

selectNow() is non-blocking but may return 0 if no channels are ready at the time of the call.

selectedKeys() fetches the Set of selected-keys. You could remove the keys from the Set by invoking the remove method of an iterator but should not add to the Set. Employ proper synchronization when using the selected-key set since it is not thread safe.

keys() fetches the keys containing the current channel registrations. Employ proper synchronization when using the key set since it is not thread safe.

SelectionKey

When a Channel is registered with a Selector, a SelectionKey object is created to represent the registration, which includes the channel, the operational sets and the Selector that was provided during the registration process.

SelectionKey carries the event information necessary for event handler to process the event. As mentioned before, a Selector decouples event arrival from event handling, for example when a channel is ready to perform an I/O operation the Selector records the necessary information related to the event and it's bindings (channel, it's most recent interest set, ready-set etc.) in the SelectionKey and makes it available in the selection-set. The actual handling of the event starts when a SelectionKey is processed based on the caller's interest and ready sets.

Let's look at some important operations of a SelectionKey.

cancel() cancels the registration of this key's channel with its selector. It synchronizes on the selector's cancelled-key set, and therefore may block briefly if invoked concurrently with a cancellation or selection operation involving the same selector.

isValid() checks if the key is valid. A key is invalid if it is cancelled, its channel is closed, or its selector is closed

isReadable() / isWritable() / isConnectable() / isAcceptable() Each of the operations test the readiness of the channel. Call these methods (based on the type of operation you are interested in) soon after the selection process is completed. You could also bit-mask a specific type against the ready-set. For example, you could use the following expression similar to isConnectable().

key.readyOps() & SelectionKey.OP_CONNECT != 0

Operational Sets

A SelectionKey contains two operational sets (interest and ready), each represented as an integer bit-map. Each bit of an operation set denotes a category of selectable operations that are supported by the key's channel:

• reading - OP_READ
• writing - OP_WRITE
• socket-accept - OP_ACCEPT
• socket-connect - OP_CONNECT

The interest set identifies the operations for which the key's channel is monitored for by the Selector. This can be changed after the registration process using interestOps(int).

The ready-set identifies the operations the channel is ready to perform.

The current operational sets are available using interestOps() / readyOps().

Implementing Scalable Servers using Selectors

Here is a common way of implementing Servers using Selectors to manage I/O.

1. Create the ServerSocketChannel and the associated Selector

  public class Reactor implements Runnable{
   final ServerSocketChannel server; 
   final Selector selector; 
  
  Reactor(int port) throws IOException
  {
      server = ServerSocketChannel.open();
      selector = Selector.open(); 
  }

2. Enable non-blocking mode, bind the socket and register the ServerSocketChannel

  Reactor
  {
   …..        
   server.configureBlocking(false);
   server.socket().bind(new InetSocketAddress(remoteHost, port));
   SelectionKey sk = server.register(selector, SelectionKey.OP_ACCEPT);
   //attach any object if you would like with the key
  }

3. Initiate the selection process by calling one of select operations on the selector.
This one thread blocks for all configured I/O!

  public void run() { 
  try {
  while (!Thread.interrupted()) {
     selector.select(); //only this one thread blocks 
            //freeing other threads from blocking
   ……..
  }
  }catch( IOException ex)

4. When Key sets are updated by selector, it is time to grab the keys
Selector detects/discovers channels that are ready to initiate specific I/O operations.

  while(..)
  {
   ………..
   Set selected = selector.selectedKeys();
   Iterator itr = selected.iterator();
   while (itr.hasNext())
        dispatch((SelectionKey)(itr.next()); //starts separate threads 
   selected.clear(); //clear the keys from the set since they are already processed
  }

5. Process the key based on the operation the channel is ready to perform A) Key.isAcceptable()

• Accept the incoming connection
• Turn on the non-blocking mode of the channels
• Register the client channels with the Selector based on ability of the server to read and write

//dispatch method spawns new threads based on the type of the key.
// none of the threads inside this method should block for I/O. 
void dispatch(SelectionKey key) 
{
  ………………
 if(key.isAcceptable()) // server is ready to accept a client channel
  { 
     //start a new thread that run the following code snippet
     SocketChannel clientChannel = server.accept(); 
     //key.channel() gives ServerSocketChannel
     clientChannel.configureBlocking(false);
     clientChannel.register(selector, SelectionKey.OP_READ | OP_WRITE);
     selector.wakeUp(); 

B) Key.isReadable()

   void dispatch(SelectionKey key) 
    {
     ………………
     if(key.isReadable())
    {
       //start reading from the channel in a separate thread
      performRead( (SocketChannel)key.channel() );
    }
 

C) Key.isWritable()

  void dispatch(SelectionKey key) 
   {
      ………………
     //start writing to the channel in a separate thread
     peformWrite ( SocketChannel)key.channel() );
  }

NIO.2 Features in JDK 1.7

In this section we will cover some of the features of NIO.2 from the perspective of this article:

• An abstraction of a file system - java.nio.file
• Enhancements to java.io.channels
• asynchronous I/O

java.nio.file

Typical operations such as deleting, copying and moving files are supported, as well as file access permission retrieval and existence checking. Metadata about the files themselves and the file stores in which they reside are provided. Improved file I/O capabilities include channel- based I/O and directory operations (including create, delete and read). Where supported by the host operating system, the creation of UNIX-style symbolic and hard links is supported. An interesting capability to recursively walk a directory tree is provided as well as locating sets of files matching a specific filename pattern specified by a UNIX shell-style glob or regular expression.

The capabilities of this package are dependent upon the semantics of the host operating system.

Below is an example of using java.nio.file.Path to open a file as a Channel:

  FileSystem fileSystem = FileSystems.getDefault();
  Path file1 = fileSystem.getPath("C:/TestFileChannel.java");
  FileChannel fcin = FileChannel.open(file1);

An Updated java.nio.channels

SeekableByteChannel (New Addition)

• Maintains a current file position.
• Factory methods such as newByteChannel(…)
• Methods to read and write to the channel using ByteBuffer

FileChannel (Updated)

• Implements SeekableByteChannel
• New open() methods to open a file with Channel semantics.

NetworkChannel (New Addition)

• Represents a network socket and defines methods for binding, setting socket options and returning local addresses.

ServerSocketChannel / SocketChannel / DatagramChannel (Updated)

• Each of the network oriented channels implements NetworkChannel
• Methods related to binding and connecting have been added without requiring socket()
• Multicast support has been added to DatagramChannel

Asynchronous I/O for Sockets and Files

AsynchronousChannel

A channel that supports asynchronous I/O operations.

AsynchronousByteChannel

An asynchronous channel that can read and write bytes.

AsynchronousFileChannel

An asynchronous channel for reading, writing, and manipulating a file without support for a current file position (i.e. not seekable). You could specify the file position to each read and write operation.

AsynchronousSocketChannel

An asynchronous channel for connection oriented network client-sockets (TCP) with timeout on read/write operations.

AsynchronousDatagramChannel

An asynchronous channel for datagram-oriented sockets with a support for multicasting.

AsynchronousServerSocketChannel

An asynchronous channel for stream-oriented listening sockets.

CompletionHandler

A handler for consuming the result of an asynchronous I/O operation as call back methods: cancelled(), completed(), failed(). The caller supplies the implementation for each call back (avoid long lived and blocking operations inside these methods).

AsynchronousChannelGroup

A grouping of asynchronous channels for the purpose of resource sharing (i.e. thread pools). Asynchronous channels that do not specify a group at construction time are bound to the default group maintained by JVM.

Using Asynchronous IO API

The Asynchronous IO is initiated with java.util.concurrent specific classes to implement robust and scalable Network IO components. Following code snippet demonstrates how to create an AsychronousChannelGroup using java.util.concurrent.ExecutorService that leverages N number of threads (fixed thread pool).

  //Use java.util.concurrent.Executor to run the tasks submitted  
  //creating an an asynchronous channel group with a fixed thread pool.
  java.util.concurrent.ExecutorService executor =    Executors.newFixedThreadPool(poolSize);
  AsynchronousChannelGroup group =  AsynchronousChannelGroup.withThreadPool(executor); 

An Asynchronous channel could be a part of a group of could exist independently (part of default group maintained by JVM). Following code snippet demonstrates how to use create an AsynchronousChannel both ways - part of a group or part of a default group.

  //create channels as a part of a group
  AsynchronousServerSocketChannel asynServer =  AsynchronousServerSocketChannel.open(group);
  AsynchronousSocketChannel asyChannel = AsynchronousSocketChannel.open(group);
    
  //create channels as part of a default group
  final AsynchronousServerSocketChannel listener = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(5000));

There are two ways you could accomplish or initiate non-blocking asynchronous IO: Using java.util.concurrent package or using java.nio.channels.CompletionHandler.

Following code snippet demonstrates use of java.util.concurrent specific classes.

  //Use java.util.concurrent package to accomplish non-blocking I/O 
  ByteBuffer buffer = ByteBuffer.allocate(1024);
  java.util.concurrent.Future<Integer> pendingResult = asyChannel.read(buffer);
  int nReads  = pendingResult.get(); //wait for the I/O to complete
  boolean isDone = pendingResult.isDone();

Following code snippet demonstrates use of java.nio.channels.CompletionHandler.

  //Use completion Handler to accomplish non-blocking I/O
  listener.accept(null, new      CompletionHandler<ASYNCHRONOUSSOCKETCHANNEL,VOID>() {
  
   public void completed(AsynchronousSocketChannel ch, Void att) {
    handleAccept(ch); 
    // handle next connection
    Future<ASYNCHRONOUSSOCKETCHANNEL> asySocCh =  listener.accept(null, this);
   }
   public void failed(Throwable exc, Void att) {
     // handleFailed(); 
   }
   public void cancelled(Void att) {
      // handleCancelled(); 
   }
  ……..
  }

Resources

NIO Java Specification
NIO.2 Java Specification
Reactor Pattern - Implementing Selector
Proactor Pattern - Basis for Asynchronous IO
NIO.2 Examples
Article on Concurrent processing
An Introduction to Java NIO and NIO.2
Java One Presentation on Asynchronous IO
JSR 203 Spec Discussion