Java VM embodies continuations now (not in production, in a hacking mode :), This post shows how to write a non-blocking server with continuations.

Why using continuation with non blocking IO

There are two models when you deals with IO:

1. the thread model: read and write calls block until they at least read one caracter or write the whole buffer, so one use thread to be able to process several requests at the same time.
   Pro: it's easy to write code with blocking IO.
   Cons: you need one thread by connection, but one thread means one stack (*) and one entry in the scheduler, not very scalable.
2. the event model (**): read and write are non blocking and you can ask the OS using a Selector if you can read or write. Or register a callback using asynchronous IO that will be called when read or write is done
   Pro: you can have more clients that the number of threads.
   Cons: painful to implement because you have to write a state machine and Selector design/implementation doesn't match well with concurrency *** so you have to code carefully (and ask your mother to verify the code).

JDK 7 also provides classes to use asynchronous IOs instead of Selector but you can't used them with continuations at least those implemented in the VM. The asynchronous callback of an asynchronous read/write can be called in another thread and continuations implemented in the VM are restricted to one thread (if yield occurs in one thread, resume must be called by the same thread).

The big advantage of using continuations in this context is that you can code as if you were using threads, i.e in a blocking IO style, but the runtime will use non-blocking IO under the hood.

Continuation with NIO

Here is an example of an echo server using non blocking IO and continuation, if you want another service, create a class that extends AbstractNIOServer and overrides method handle of the request processor, don't forget to tag it with annotation @Continuable. A request processor is an object that will be created to handle one connection of one client, it creates and manage the underlying continuation.

  public class EchoNIOServer extends AbstractNIOServer {
  @Override
  protected RequestProcessor createProcessor(Selector selector, SocketChannel channel) {
    return new RequestProcessor(selector, channel) {
      @Override
      @Continuable
      protected void handle(ByteBuffer buffer) throws IOException {
        while (read(buffer) != -1) {
          buffer.flip();
          
          write(buffer);
          
          if (buffer.hasRemaining())
            buffer.compact();
          else
            buffer.clear();
        }
        
        close();
      }
    };
  }
  
  public static void main(String[] args) throws IOException {
    new EchoNIOServer().start();
  }
}

As you notice, it's a simple while loop that read data and write the same data. The code is as you will write it if you use blocking IO.

Under the hood

When the server receive a new connection, it first accept it and then delegate to a thread the processing of the request, this is done by posting the client channel in a queue and wake up the selector of one worker thread. Then the worker thread executes the last part of the above snippet, it allocates a new byte buffer, creates a request processor and starts a fiber that executes the method handle of the RequestProcessor.
When method handle calls read or write, the server try to read/write in non-blocking mode, if it doesn't succeed, the client channel is registered in the selector of the thread with the request processor stored as attachment and the the fiber calls yield. So the call to start returns and the worker thread go to waiting in select until at least one channel is readable or writeable.
When one channel is selected, the request processor resume the fiber that will restart the execution of the method handle by trying to read or write again. If method handle issues another non blocking read or write, the method handle will call yield again and the next selected channel will be processed.

  [...]
  for(;;) {
    int select = selector.select();
    if (select != 0) {
      for(Iterator it=selectedKeys.iterator(); it.hasNext();) {
        SelectionKey key = it.next();

        if (!key.isValid()) {
          it.remove();
          continue;
        }
                  
        RequestProcessor processor = (RequestProcessor)key.attachment();
        processor.resume();
                  
        it.remove();
      }
    }

    SocketChannel channel;
    while((channel = queue.poll()) != null) {
      ByteBuffer buffer = ByteBuffer.allocate(8192);
                
      RequestProcessor processor = createProcessor(selector, channel);
      processor.start(buffer);
    }
  }
  [...]

The whole code is available here.
To run it, you can:

1. Build the Da Vinci VM with only enabling callcc patch (comment all other patches). Don't forget that only C1 works.
2. If you use a Linux (I use a Fedora 11), I've already compiled a VM with callcc patch. So download jdk7-b75 binaries, and unzip coroutine-VM.zip in directory jre/lib/i386. You also need to download coroutine.jar that contains Java classes that provide support for fibers.

What's next

This webserver is a toy, there is lots of rooms of improvement (use more than one thread to do the accept, pipeline the write, pool buffers and fibers, etc.) I haven't benchmark this implementation because I have no gigabit switch available to do it. If you test it, don't forget to drop me a comment on this post. By the way, don't forget to increase the number of threads in AbstractNIOServer.

See you (*) soon,
cheers,
Rémi

* With google analytics, this is almost true.