Jean-Francois Arcand's Blog

Understanding Grizzly part III: Implementing your own protocol on top of the animal

Extending Grizzly to support any tcp protocols is not complex. This time I will explain how to customize the current framework to support other protocols than HTTP. In part I and part II, I've explained how to embed Grizzly and how to configure its thread pool. This time let me give some details about how a request is handled in Grizzly, and how to extends the current implementation.

All tcp requests are accepted by the SelectorThread.doSelect(). Technically, it means the SelectionKey's operation-set bit for socket-accept operations (OP_ACCEPT) is handled with the same thread (see SelectorThread.handleAccept()). Next the operation-set bit for read operations (OP_READ) is delegated to the Pipeline (thread pool). The Pipeline can execute any Task implementation. A Task is a Runnable object augmented with some Grizzly specific methods. For handling OP_READ, the SelectorThread will get from its pool an instance of ReadTask. By default, Grizzly at startup creates a pool of DefaultReadTask instances. This task is responsible for reading the initial bytes, decide if there is enough bytes to continue the processing, and then handle the decision to keep-alive the connection or not. The StreamAlgorithm is where you implement the logic to decide if there is enough bytes available to process the request. Any implementation of a StreamAlgorithm will need to take care of managing its ByteBuffer life cycle (creating, pooling, recycling etc.) and decide if more bytes are needed before continuing the request processing. Hence this is where protocols specifics parsing logic needs to be implemented. The easiest way to start with a StreamAlgorithm implementation is by extending the StreamAlgorithmBase and implement the parse method:

     13     public boolean parse(ByteBuffer byteBuffer) {
     14         boolean isFound = false;
     15
     16         int curPosition = byteBuffer.position();
     17         int curLimit = byteBuffer.limit();
     18         byteBuffer.position(0);
     19         byteBuffer.limit(curPosition);
     20         int state=0;
     21         int start=0;
     22         int end= 0;
     23
     24         try {
     25             byte c;
     26
     27              
     28             while(byteBuffer.hasRemaining()) {
     29                 c = byteBuffer.get();
     30                 // State Machine
     31                 // 0 - Search for the first SPACE ' ' between the method and the
     32                 //     the request URI
     33                 // 1 - Search for the second SPACE ' ' between the request URI
     34                 //     and the method
     35                 switch(state) {
     36                     case 0: // Search for first ' '
     37                         if (c == 0x20){
     38                             state = 1;
     39                             start = byteBuffer.position();
     40                         }
     41                         break;
     42                     case 1: // Look for the request path
     43                         if (c == 0x20){
     44                            state = 2;
     45                            end = byteBuffer.position() - 1; 
     46                            byteBuffer.position(start);
     47                            byte[] requestURI = new byte[end-start];
     48                            byteBuffer.get(requestURI); 
     49                         }
     50                         break;
     51                      case 2: // Look for the protocol
     52                         if (c == 0x2f){
     53                             // ....
     54                             return true;
     55                        }
     56                        break;
     57                     default:
     58                         throw new IllegalArgumentException("Unexpected state");
     59                 }
     60             }

The example above is a parser which seek for the request path (ex: /grizzly/Servlet) and the protocol used (FOO/1.1). This parser will support request that takes the form of "FOO /foo FOO/FOO". The parse() method will return true if the request processing can continue, or false if the StreamAlgorithm needs more bytes to take a decision. If more bytes are needed, the SelectionKey will be registered back to the Selector and will be reprocessed once bytes are available. If there is enough bytes, the ReadTask will delegate the request processing to an instance of ProcessorTask. The ProcessorTask is where the request processing needs to happens. As an example, in GlassFish, this is where the HTTP protocol is parsed and the Servlet Container invoked. Take a look at the DefaultProcessorTask to see how the HTTP protocol is handled.

The main methods to implement is the process() and initialize():

     48     public void initialize() {
     49         requestLine = new byte[8192];
     50     }
     51
     52     public boolean process(InputStream input, OutputStream output) throws Exception {
     53
     54         ByteBufferInputStream bbInputStream = (ByteBufferInputStream)input;
     55         bbInputStream.read(readLine);
     56
     57         // parse the protocol
     58         // invoke some Container
     59         // flush the response
     60         // return true to keep-alive the connection, false to close the connection.
     61
     62     }

If you look at the ProcessorTask interface, you will soon realize there is a lot of methods to implement. The reason is because Grizzly asynchronous request processing ARP mechanism needs all of them to execute an asynchronous request. I will soon describe in this series of blogs how to implement an asynchronous protocol implementation, but if you don't plan to use ARP, just take a look at the Jetty ProcessorTask implementation.

To recap, to implement a tcp protocol with Grizzly you first need to implement a StreamAlgorithm and a ProcessorTask. Next you need to extends the SelectorThread so it can configure those instances and pool them:

     36 public class FooSelectorThread extends SelectorThread {
     37
     38     public FooSelectorThread() {
     39         super();
     40     }
     41
     42     /**
     43      * Force Grizzly to use the FooStreamAlgorithm
     44      */
     45     protected void initAlgorithm() {
     46         setAlgorithmClassName("org.foo.FooStreamAlgorithm");
     47         super.initAlgorithm();
     48     }
     49
     50     /**
     51      * Return a FooProcessorTask implementation.
     52      */
     53     public ProcessorTask newProcessorTask(boolean initialize) {
     54         FooProcessorTask task= new FooProcessorTask();
     55         task.initialize();
     56         task.setPipeline(processorPipeline);
     57         return task;
     58     }

OK this might looks difficult, but it isn't :-).

In case you don't want to implement a StreamAlgorithm, you can always delegate all the processing to the ProcessorTask. The InputStream passed to the ProcessorTask.process() uses a pool of Selector under the hood, so you are guarantee to get all the bytes. But this approach is similar to a socket blocking approach, and may need consume more threads depending on what your protocol is.

Euhhh...enough for this blog :-) Next time I will discuss how you can implement a ReadTask from scratch, and introduce some asynchronous protocol processing concept.

technorati: grizzly web server embedded

Comments

• can implement UDP based protocol on top of the grizzly?

  patrick
  

  Posted by: mk926 on December 13, 2006 at 07:00 PM

• Hi, Yes you can, but it requires more work than tcp. I will soon blog about it :-) -- Jeanfrancois

  Posted by: jfarcand on December 14, 2006 at 07:43 AM

• Can you provide the source for this example? I tried rebuilding it myself but for some reason grizzly never enters the parse method in the algorithm class. Also, any client that connects is immediately disconnected after sending any data. Not specifying the algorithm class name allows me to process the data in the processor task class.

  Posted by: kmk2007 on January 01, 2007 at 05:52 PM

• Hi Jean, We have an existing server application using blocking I/O and thread-per-connection implementation. We want to add NIO layer (build on top of Grizzly) to this app to improve capacity. Our existing app not only handle requests from client connection (request-process-response model), but we also have varies events sent to the client that are initiated by the server. How should we handle those kind of events in grizzly? Is there a way to plug-in a writerTask pipeline?

  Posted by: fzheng1998 on April 11, 2007 at 01:40 PM

• FYI: The answer and discussion can be followed on users@grizzly.dev.java.net.

  Posted by: jfarcand on April 12, 2007 at 04:48 PM

This work is licensed under a Creative Commons License.