[jetty-dev] Jetty IO changes from 9.0 to 9.1

[Greg Wilkins <gregw@xxxxxxxxxxx>]

This is a high level overview (brain dump) of the changes made to the jetty IO infrastructure from Jetty 9.0 to 9.1 to support the servlet 3.1 asynchronous IO API.

The main state machine involved with handling HTTP requests (from both HTTP connections and SPDY connections) is the HttpChannelState.    Previously this had a single state machine that encapsulated both the state of handling (dispatch or not) plus the state in the asynchronous life cycle (initial, suspended, dispatching, completing etc.):

    public enum State  // Jetty-9.0
    {
        IDLE,          // Idle request
        DISPATCHED,    // Request dispatched to filter/servlet
        ASYNCSTARTED,  // Suspend called, but not yet returned to container
        REDISPATCHING, // resumed while dispatched
        ASYNCWAIT,     // Suspended and parked
        REDISPATCH,    // Has been scheduled
        REDISPATCHED,  // Request redispatched to filter/servlet
        COMPLETECALLED,// complete called
        COMPLETING,    // Request is completable
        COMPLETED      // Request is complete
    }

When moving the HttpChannel through this state machine, the following actions could be return to the caller to instruct them what to do next:

    public enum Next
    {
        CONTINUE,       // Continue handling the channel       
        WAIT,           // Wait for further events
        COMPLETE        // Complete the channel
    }

However, the servlet 3.1 API introduces two additional types of dispatch to the application: ReadListener.onDataAvailable and WriteListener.onWritePossible.    Importantly these dispatches have to be done in exclusion to any normal dispatch to the servlets and with threads anointed with the context (classloaders, JNDI etc.).    So the extra dispatch states and mutual exclusion made it far to complex to continue to have a single state machine for both request lifecycle and dispatch state.   Thus in 9.1 they have been split out into 2 state machines and 2 state booleans:

    public enum State
    {
        IDLE,          // Idle request
        DISPATCHED,    // Request dispatched to filter/servlet
        ASYNCWAIT,     // Suspended and parked
        ASYNCIO,       // Has been dispatched for async IO
        COMPLETING,    // Request is completable
        COMPLETED      // Request is complete
    }

    public enum Async
    {
        STARTED,
        DISPATCH,
        COMPLETE,
        EXPIRING,
        EXPIRED
    }   

    private boolean _asyncRead;
    private boolean _asyncWrite;

And methods that mutate these states now return more possible actions that need to be performed:

    public enum Action
    {
        REQUEST_DISPATCH, // handle a normal request dispatch 
        ASYNC_DISPATCH,   // handle an async request dispatch
        ASYNC_EXPIRED,    // handle an async timeout
        WRITE_CALLBACK,   // handle an IO write callback
        READ_CALLBACK,    // handle an IO read callback
        WAIT,             // Wait for further events
        COMPLETE          // Complete the channel
    }

This has resulted in much simpler state machine code in the HttpChannelState class.  However, the downside is that having two state machines means that we definitely have to do synchronize style state changes and cannot consider a lock free solution.   But that is what we already do, so it is not extra cost - just a no longer possible future optimisation.

When a connection wants to signal that a read or write is possible, it calls HttpChannelState.onReadPossible or HttpChannelState.onWritePossible().     If the channel is already dispatched - either in the normal request handling or in a read or write callback, then this call simply flags that an IO callback is needed and this Action is performed when the dispatch in progress completes.  If the channel is not already dispatched, then it is dispatched to do the actual io handling.

As well as changing the channel state, their is a need to change the state machines inside the Input and Output streams used by the handlers/servlets - well to be more precise there is now a need to make them stateful when before they pretty much just had a boolean closed.

HttpOutput now implements a non locking statemachine with:

    enum State { OPEN, ASYNC, READY, PENDING, UNREADY, CLOSED }
    private final AtomicReference<State> _state=new AtomicReference<>(State.OPEN);

and this drives it around either a blocking write conversation or an asynchronous write conversation.   Both blocking and non blocking writes make use of the HttpChannel API

   write(ByteBuffer content, boolean complete, Callback callback)

with the main difference being what callback is used.   If it is a blocking write, then a simple blocking callback is used:

                        BlockingCallback callback = _channel.getWriteBlockingCallback();
                        _channel.write(_aggregate, complete, callback);
                        callback.block();

if it is an asynchronous write then an AsyncWrite IteratingCallback instance is created.  This is a callback that continues to call write asynchronously  until all the content is written and then moves the state machine from UNREADY to READY before calling onWritePossible()

Note that the HttpChannel.write() API is really just a facade over HttpTransport methods:

    void send(HttpGenerator.ResponseInfo info, ByteBuffer content, boolean lastContent, Callback callback);
    void send(ByteBuffer content, boolean lastContent, Callback callback);
   

which are implemented either by the HTTP connection or the SPDY connection.    I have modified the HTTPConnection so that these asynchronous methods simply wrap the callback and call asynchronously all the way down to the AbstractChannel WriteFlusher.      The SPDY implementation needs to be updated here as it may still be simulating async with some blocking implementations of these calls.

Similarly HttpInput now has a state machine to control blocking or asynchronous writes bases around state classes with behaviour rather than just an enum:

    protected static class State
    {
        public void waitForContent(HttpInput<?> in) throws IOException
        {
        }
       
        public int noContent() throws IOException
        {
            return -1;
        }
       
        public boolean isEOF()
        {
            return false;
        }
    }

    protected static final State BLOCKING= new State()
    protected static final State ASYNC= new State()
    protected static final State EARLY_EOF= new State()
    protected static final State EOF= new State()

This is intended to eventually be a lock free implementation, but it currently uses synchronise to maintain compatibility with the content queuing implementation.    The content queue which used to be part of the base HttpInput has now been moved to a QueuedHttpInput derivation that is used by test harnesses and SPDY.    This implementation extends blockForContent to wait on the queue and also uses any calls to add content to either wake up a blocked reader or call channel.onReadPossible() for asynchronous readers.

This queued implementation (which still could do with some modernisation) works fine for SPDY where different threads do the protocol parsing and request handling.  The calls from the protocol thread to content, eof etc. are sufficient to drive the state machine and back pressure from slow readers is achieve via protocol mechanisms that monitor the size of the content queue.

However, for HTTP, a different solution is required because of HTTP pipelining, all reading/parsing of IO bytes is suspended while the request is dispatched, until such time as a read is called.   HTTP cannot read ahead else there is no TCP/IP back pressure exerted by a slow reader. Thus there is a separated HttpInputOverHTTP class that provides an alternative non queuing implementation.

With HttpInput, all calls to read, isReady() and available() essentially call nextContent().  In the queuing application this is implemented to pop the next content off the queue, but in the HTTP impl, this actually does a IO read and parse looking for a call to content(ByteBuffer).  If there is no content, then it calls _httpConnection.fillInterested(Callback) to wait for ome IO activity before nextContent is called again.   If it is a blocking  read, then the fill interest called like:

            _httpConnection.fillInterested(_readBlocker);
            _readBlocker.block();

otherwise for async reads it is called from unready() with the input as a callback and which has:

    public void succeeded()
    {
        _httpConnection.getHttpChannel().getState().onReadPossible();
    }

that triggers the async lifecycle.

So the final piece of this puzzle is the HttpConnection.fillInterested(Callback call).    This was not in 9.0 but is derived from the AbstractConnection.block(BlockingCallback callback) method which was used to do blocking reads for HTTP.

AbstractConnection in Jetty-9 has a state machine that tracks dispatches to onFillable() together with calls for interest either from fillInterested() or block(BlockingCallback).

    private enum State
    {
        IDLE, INTERESTED, FILLING, FILLING_INTERESTED, FILLING_BLOCKED, BLOCKED, FILLING_BLOCKED_INTERESTED, BLOCKED_INTERESTED
    }

The complexity here is that the interest can be registered from within or outside of a FILLING state (a call to onFillable) and if we are already calling onFillable() then we cannot register for that additional interest and queue it up until the return.

In 9.1, we have simplified this somewhat by having a lock free State class bases state machine that implements the main onFillable states:

    public static final State IDLE=new State("IDLE")
    public static final State FILL_INTERESTED=new State("FILL_INTERESTED")
    public static final State FILLING=new State("FILLING")
    public static final State FILLING_FILL_INTERESTED=new State("FILLING_FILL_INTERESTED")

The in addition to that statemachine, a call to fillInterested(Callback) will establish a nested state machine with the normal state machine wrapped in a FillingInterestedCallback state that holds the passed callback.       The wrapped statemachine can still progress (eg if the onFillable call returns it may move from FILLING to IDLE).  While the state machine is wrapped, it has it's own callback registered with connection.getEndPoint().fillInterested(callback); to process IO read interest.  Only when that succeeds or fails will the nested state machine be unwrapped and normal processing continue.

This approach is still a little complex and creates a little more garbage than desirable, but as it is for only called when input is not available, this is not a prime use-case for most request handling.  

While making these changes work, there were a few other cleanups that went through the code.  Primarily in the HttpConnection.onFillable implementation that still had a few too many special cases for closing.  These have been removed and replaced by a policy of always sending a BadMessage event if a connection closes during a request header.  This is a simpler approach but is currently producing a few unwarranted exceptions when a connection closes before ever sending a request (mostly a problem for test harnesses).

cheers

--
Greg Wilkins <gregw@xxxxxxxxxxx>
http://www.webtide.com
Developer advice and support from the Jetty & CometD experts.
Intalio, the modern way to build business applications.