Allow for batching writes to the framer

[louiscryan]

Need to allow applications to perform a sequence of writes that cause a single flush in the framer to improve throughput.

The typical example for this would be an application thats wants to write messages until isReady() is false and to do more writes when onReady() is called. Even if the application is not flow-control aware (i.e is not using isReady) it would still be useful to allow write batching for bursty streams

One simple option for doing this would be to delay the outbound framer flush while executing onPayload/isReady callbacks though this would only help cases where sends are done inside these callbacks by the same thread. A more thorough API change is probably warranted.

To give some performance context the change described above allows the FlowControlledMessagesPerSecondBenchmark to go from ~700kqps to ~4Mqps on my box

[ejona86]

Let's brainstorm various options.

Add flush to Call:

public abstract void setAutoFlush(boolean);
public abstract void flush();

Add an inhibit (a la TCP_CORK) to Call:

public abstract void inhibitFlush(boolean);

On the Channel, provide a way to inhibit. This could be a inhibit/allow pair of methods. It must be more powerful than a boolean, though, since more than one could be active at a time. Or it could be we return a special object that is completed when the application no longer wants to inhibit:

public abstract Inhibit inhibit();

public abstract class Inhibit {
  public abstract void complete();
}

Note that adding an API to the Channel is not really generalizable to Server-side, unless we add a way to get the Server instance from a ServerCall.

[louiscryan]

Other options...

1. Make Call a 'batch' factory ...

public abstract Batch batch();

interface Batch extends AutoCloseable {
  public boolean sendMessage(...);  // return value is equivalent to isRead() call
}

The quirk here is the semantics of 'close', i.e. if a batch is not closed is it not written or just not flushed. This would generalize to the server side.

---

I like explicit flush calls better than inhibit. If we go with that model I suspect it would actually make the transport stream implementations a little simpler than they are today as they don't have to make their own flush decisions.

[ejona86]

How would Batch interact with interceptors? It seems like it would be important to make batching orthogonal to the rest of the APIs; it seems important to be able to use the same interceptors, stubs, and reuse your own code independent of batching.

[louiscryan]

If batch is simply a deferred sequence of writes then it really has little
impact on interceptors. The primary concern is to ensure a proper closure
of message production so that flush can be sent reliably.

I don't really like

sendMessage(Collection messages)

though it's similar to Batch in some ways it doesn't allow for feedback
from isReady().

sendMessage(Iterator messages)

would provide the closure and allow for the lazy production of messages and
the isReady() check could be done implicitly inside our impl (I kinda like
this one)

Then there's also

boolean sendMessage(message, flush)

where the return value is simply isReady() which would allow for writing
loops like

do (

} while (sendMessage(msg, false));
stream.flush();

The use of onReady (and by implication onPayload etc.) to implement the
closure over flush works reasonably well though this interface pattern
would need propagate higher to be effective.

[ejona86]

I don't like sendMessage(Collection messages) and sendMessage(Iterator messages). I agree using a Collection prevents any flow control feedback. Iterator turns message creation into a callback, which seems unnecessary and harder to use. It would also require having a different method on Call, and so would be a burden to interceptors.

boolean sendMessage(message, flush) could work, although it is little different than having a flush() method that is required to be called, with the exception of the possibility of less synchronization. Since there would be many callers over time, seeing the true/false could easily be a readability issue (not knowing if it is flushing or not). Granted, not having the parameter could be a readability issue by omission (not noticing it isn't flushing), but people are pretty familiar with the need to flush on things like OutputStreams. I guess we could have sendMessage(message) always call sendMessage(message, true), and so the second parameter is typically missing, explicit flushing unnecessary, and there would be few calls with the boolean argument. We could make an enum FLUSH/NO_FLUSH instead of boolean if we really wanted.

[louiscryan]

OK. I think we're largely in agreement on

void sendMessage(msg) --> which calls sendMessage(msg, true)
boolean sendMessage(msg, flush)

I'll create a PR for that and see how it feels

@benjchristensen - Not sure if this has come up in discussions in reactive streams io land, i.e. how a variable rate message producer reacts to changes in transport flow-control when the writeable window is expressed as bytes not messages.

[ejona86]

@louiscryan, earlier you were considering channel-wide flush reduction. Are you still interested in that?

[louiscryan]

channel-wide flush reduction is largely handled by the shared write-queue. The proposed change here only affects transport flushes in so much as fewer flushes emanate from the framer.

This change ensures that message sequences are batched into single DATA frames so that the receiving side can parse as a unit and wake the application layer less often. An alternate way to achieve this effect is to have the transport layer flow-controller coalesce writes into a single DATA frame under flush. It also prevents the framer from having to allocate another buffer.

@nmittler

I can try to implement handling for that at the transport layer by
(a) - Essentially making the flow-controller writes deferred so that they pack into DATA frames
(b) - Having the framer slice the buffer it reports to the transport so it can retain the unused portion

In the case of Netty this would largely mean making sure that we can properly write out a CompositeBuffer into the flow-controller and have its chunks be released properly.

I suspect this will probably not perform as well as explicit batching on an individual stream as the write queue would have to compose individual writes across streams.

[ejona86]

Ah @louiscryan, that makes sense, that it is only for the framer's benefit. Gotcha.

[louiscryan]

Ill test out the DATA frame merging in Netty and see how it stacks up

[nmittler]

@louiscryan I think having the outbound flow controller coalesce data frames makes sense in general. The only sticky bit that comes to mind is what to do about padding if it's used. For gRPC we're not using it, but a general solution would have to take padding into consideration. Maybe it's simple: we just coalesce adjacent frames if they have no padding.

[benjchristensen]

Not sure if this has come up in discussions in reactive streams io land, i.e. how a variable rate message producer reacts to changes in transport flow-control when the writeable window is expressed as bytes not messages.

Yes the discussion came up a few times and was specifically discussed here: reactive-streams/reactive-streams-jvm#47

We recently implemented a solution with Netty here: https://github.com/ReactiveX/RxNetty/blob/79cd68e01c8fc1139c71ec44b73d53f6b571ec3f/rxnetty/src/main/java/io/reactivex/netty/protocol/tcp/BackpressureManagingHandler.java This correctly stops the upstream Observable from producing and then resumes based on HTTP/TCP network backpressure.

Another example with Aeron is here: https://gist.github.com/benjchristensen/48f17210a674bf38329d#file-rxaeronexample-java-L114 This is somewhat of a toy example but shows how it stops production when backpressure from the network occurs and resumes upstream production when it is able to emit. It also shows the conversion from application object to bytes.

With Reactive Streams implementations over networks the general idea is that the application level requests n items, and the network level converts an item into bytes and then deals with the network level backpressure. This means we can have bounded buffers for the number of items, but need at least one unbounded buffer (or the ability to partially serialize the item with pause/resume etc) to store a single serialized item while writing and/or waiting to write over the network.

Another way of considering this is that the "async writer" is an asynchronous boundary which has its own internal bounded buffer of items and invokes request(n) upstream to allow production as it is capable of receiving. Then it writes out and deals with the conversion to bytes as the network flow-control permits it. As it drains its bounded buffer of items it requests more upstream. With this pattern it naturally bridges the two and correctly stops production if the network can't keep up.

[louiscryan]

@benjchristensen thanks for the pointers. I note the issue with an unbounded buffer vs partial serialization (which is probably rare for anything other than byte-sequence types). Or put another way its hard for the transport layer to do anything other than

request(1)

as it has no idea how large the next produced message will be (hinting and heuristics aside) e.g.
https://gist.github.com/benjchristensen/48f17210a674bf38329d#file-rxaeronexample-java-L89

GRPC chose to model this as onReady() rather than request(n) for this reason. In either case you have a cycle which looks like

request(1) --> onValue --> transport write msg --> request(1) if more window available --> ...

which on the read side leads to

transport read msg --> onValue --> transport read next msg --> onValue --> ...

This can perform pretty well but theres a problem if the transport can't coalesce message writes in such a way that the read-side can take advantage of the fact the more than one message is coming off the wire at a time so it can do

transport read many messages --> loop over onValue

In short, don't write in such a way that the read side can't dispatch batches. This is particularly important if the --> implies a thread boundary being crossed

See netty/netty#3852 for just such an issue in Netty's HTTP2 stack.