Backpressure in AsyncHandler

[jroper]

AsyncHandler provides no mechanism to send back pressure on receiving the body parts.

Imagine you have a server that stores large files on Amazon S3, and streams them out to clients, using async http client to connect to S3. Now imagine you have a very slow client, that connects and downloads a file. The slow client pushes back on the server via TCP. However, async http client will keep on calling onBodyPartReceived as fast as S3 provides it with data. The AsyncHandler implementation will have three choices:

1. Block. Then it's blocking a worker thread, preventing other concurrent operations from happening. This is not an option.
2. Buffer. Eventually this will cause an OutOfMemoryError. This is not an option.
3. Drop. Then the client gets a corrupted file. This is not an option.

AsyncHandler therefore needs a mechanism to propagate back pressure when handling body parts. One possibility here is to provide a method to say whether you are interested in receiving more data or not. This would correspond to a Channel.setReadable(true/false) in the netty provider, which will push back via TCP flow control. This could either be provided by injecting some sort of "channel" object into the AsyncHandler, or, since HttpResponseBodyPart already provides mechanisms for talking back to the channel (eg closeUnderlyingConnection()), it could be provided there.

[jfarcand]

@jroper I'm tempted to answer you like you did recently with the non sense issue with headers/body in Play. But I won't ;-). Contribution welcomed ;-)

[jroper]

Reading over my that issue again, you're right, I didn't use the best tone, I probably should have just said that it was an architectural limitation of Play - thanks for not completely fobbing me off :)

Do you think the approach of adding a callback onto HttpResponseBodyPart is a good one, or would you suggest something else?

[jfarcand]

A simple solution would be to add a new state on AsyncHandler and return a PUSHBACK (or a better name). That would make it simple for provider to properly invoke the native API, e.g for Netty Channel.setReadable(true/false). That could also be extended to status and headers. What do you think?

[slandelle]

How would you put the channel back in readable state?

[jfarcand]

@slandelle Good point. So we need some kind of suspend/resume (sound like Atmosphere :-)) API on HttpContent. Or have our own notion of Channelpropagated with the AsyncHandler. I'm more for suspend/resume as it won't break the AsyncHandler API. WDYT?

[slandelle]

I don't see how you'd implement resume. What I would do is schedule a suspend timeout that would notify the handler so it can decide if it wants to resume. My only concern is thread safety, as this would execute in the timer thread instead of the eventloop.

[jroper]

The handler will know when it wants to resume because the event loop its using to write data out will let it know that writing is now possible. In the servlet 3.1 API for example, this will be via registering a WriteListener, when onWritePossible is invoked, at that point, it will invoke resume(). In Netty, for example, it would be via a callback attached to the ChannelFuture returned by the write operation. There should be no need for async-http-client to be involved in that.

[slandelle]

I'll be frank with you: I seriously doubt anyone else but you Typesafe guys would implement this...

[jroper]

I think probably the best approach here is to wait for reactive streams, and then add support for using reactive streams instead of AsyncHandler to handle the async streams. This will be really easy to implement from the Netty provider, since Netty I'm pretty sure will provide a reactive streams implementation (the RedHat guys are involved in the reactive streams spec), and the aim of reactive streams in fact is to become a part of the JDK, so Grizzly should some day also follow suite (Oracle also has representation in the reactive streams spec).

This will mean interoperability between the AHC streaming APIs and every other reactive streams implementation out there for free (rxjava, akka streams, play, netty, etc).

[slandelle]

Thanks for the information.

[jroper]

An update here - I've implemented reactive streams support for Netty 4.0 here:

https://github.com/typesafehub/netty-reactive-streams

So, as far as how we'd actually implement support for reactive streams in this, there's two things to consider, request bodies and response bodies. We also need to consider that not every provider is going to provide a reactive streams implementation at the same time, so it needs to be an opt in thing.

Starting with response bodies, I propose that we provide a new sub interface of AsyncHandler called StreamedAsyncHandler, which introduces a new method, onStream(org.reactivestreams.Publisher<HttpResponseBodyPart>). A provider can, if it chooses to, check if the supplied async handler is an instance of StreamedAsyncHandler, and if it is, cast and invoke onStatusReceived, onHeadersReceived, and then onStream. If it invokes onStream, it will never invoke onBodyPartReceived, but will eventually invoke either onThrowable or onCompleted once the body has been consumed (in addition to invoking the onComplete or onError methods on the subscriber). A provider does not have to invoke onStream, even if it supports reactive streams, for example, if the request has no body, or if the request body is small and it has buffered it in memory, it may invoke onBodyPartReceived instead. A provider that does support reactive streams however should invoke onStream when the body is of a size that should not be buffered into memory.

For request bodies, this can be supported by creating a new sub class of BodyGenerator for bodies generated by a reactive streams Publisher. This will have a method to get the publisher that a reactive streams supporting provider will use. The getBody method will return an implementation of Body that uses a crude blocking queue based reactive streams subscriber to implement the read method for those providers that don't support reactive streams.

How does that sound?

[jfarcand]

Non binding (Stephane is the chef), but I like your proposal +1 !