Zero-copy* write combining

[buchgr]

TL;DR;
I propose to defer the serialization of protobuf, thrift, ... from the MessageFramer to the WriteCombiningHandler.

PR #1996 shows that combining small buffers into one large buffer can substantially improve throughput. However, it's very hard to come up with a one-size-fits all algorithm that decides on when to combine buffers. It's probably safe to assume that all buffers (HTTP/2 frames) except for DATA frames are small enough to be combined (copied) without much overhead. HEADERS could be large, but probably aren't. Say we would expect headers to be in the 100 - 300 byte range.

For example, a RPC response will typically write 3 HTTP/2 frames

HEADERS + DATA + HEADERS (trailers)

In Netty's HTTP/2 codec that would typically translate to the following buffers being written

HEADERS(FrameHeader) + HEADERS(HeaderBlock) + DATA(FrameHeader) + 
DATA(Payload) + HEADERS(FrameHeader) + HEADERS(Payload)

When using write combining as proposed in #1996, then depending on the size of the DATA(Payload) buffer, we would either pass one buffer or 3 buffers to write/writev. As mentioned before, coming up with a "magic" number N bytes that for all machines and workloads tells us when to combine buffers is hard. It would be much easier if we would simply always combine all writes until flush, and I believe we can.

If we change the MessageFramer internals and the WritableBuffer API, so that instead of always copying the InputStream into a buffer, it's up to the WritableBuffer implementation of when that copy happens. We could then create a simple, internal ByteBuf implementation that effectively just wraps an InputStream. The copying into a "real" ByteBuf could then be done in the WriteCombiningHandler. So we would still only do one copy, but instead of copying in the MessageFramer we do it in the WriteCombiningHandler.

The logic of the WriteCombiningHandler would become very streamlined. It simply stores a list of ByteBufs and Promises, and on flush it allocates a large buffer with exactly the number of bytes it needs, copies all buffers from the list into the large buffer, and does a ctx.writeAndFlush(largeBuffer, combinedPromise).

A nice sideeffect of this change would be that buffering / piggy-backing messages would no longer be a concern of the MessageFramer, but be a responsibility of the transport only.

Some questions:

• This moves the protobuf serialization from the application thread to the Netty thread, is this a problem?
• Right now the InputStream is immediately consumed when passed to the MessageFramer. With this change it would be consumed at some later point in time on the Netty thread. Could that be a problem in terms of leaks or so?

Thoughts @louiscryan @ejona86 @nmittler @carl-mastrangelo ?

* Zero-copy for DATA frames only, but assuming most of the time DATA frames make up the majority of the bytes on the wire. Certainly true for streaming :-).

[buchgr]

Some rough sketch of how to do this

public class MessageFramer {
  public interface Sink {
    void deliverFrame(@Nullable Frame frame, boolean endOfStream, boolean flush);
  }

  public abstract class Frame {

    protected final InputStream message;
    protected final boolean compressed;

    protected Frame(boolean compressed, InputStream message) {
      this.message = message;
      this.compressed = compressed;
    }

    protected WritableBuffer write(WritableBufferAllocator allocator) {
       // Implementation similar to MessageFramer.writePayload(InputStream).
       // Do the actual framing and request buffers via the allocator.
    }
  }
}

class NettyFrame {

  ByteBuf bytebuf() {
    int messageLen = getKnownLength(message);
    if (messageLen > 0) {
      return new WrappedByteBuf(message, messageLen);
    } else {
      // If the length of the stream is not known, then
      // invoke super.write(allocator) and return a normal ByteBuf.
    }
  }

  class WrappedByteBuf extends AbstractByteBuf {

    @Override
    public int readableBytes() {
      return grpcFrameHeaderLen + messageLen;
    }

    @Override
    public int writableBytes() {
      return 0;
    }

    @Override
    public ByteBuf getBytes(int index, ByteBuf dst, int length) {
      // Invokes NettyFrame.this.write(allocator) with an allocator
      // that returns a buffer backed by dst
    }
  }
}

[nmittler]

I suspect that due to multiplexing many streams over a single channel, we can't move the serialization logic to the event loop thread until we have HTTP/2 child channels in Netty. Maybe more incentive for someone (cough) to pick that work back up? :)

[ejona86]

As mentioned before, coming up with a "magic" number N bytes that for all machines and workloads tells us when to combine buffers is hard.

It is an optimization. If the magic number is 1 then it is disabled. If it is ∞ then it is always enabled. I feel like we should be able to find a conservative number that is a benefit to all workloads. Although a higher number may benefit some workloads, choosing a smaller number doesn't hurt any workloads.

This moves the protobuf serialization from the application thread to the Netty thread, is this a problem?

I do believe that to be a problem. Protobuf is where most of the time is spent now; it seems like it would substantially reduce throughput per Channel.

we can't move the serialization logic to the event loop thread until we have HTTP/2 child channels in Netty

Except that doesn't solve anything, because WriteCombiningHandler would still be on the main Channel.

[buchgr]

Not that I am crazy enough to do it, but we could enable it for servers when direct executor is used 😅

Except that doesn't solve anything, because WriteCombiningHandler would still be on the main Channel.

Well the WriteCombiningHandler could be per child channel, and the child channel could use a different executor?

[ejona86]

Not that I am crazy enough to do it, but we could enable it for servers when direct executor is used 😅

Yeah, I considered that. That would be fine, but also a lot of work. Direct would already probably be benefiting from releasing the buffer on the same thread, so I'd expect fewer gains.

Well the WriteCombiningHandler could be per child channel, and the child channel could use a different executor?

Isn't that effectively what we have now with MessageFramer? The only difference is that the child channel has a well-defined lifecycle so we could easily retain buffers after a write. But I already think we don't want to do that because it could waste a lot of memory.

[buchgr]

Isn't that effectively what we have now with MessageFramer? The only difference is that the child channel has a well-defined lifecycle so we could easily retain buffers after a write. But I already think we don't want to do that because it could waste a lot of memory.

I don't think so. E.g. for a unary RPC response, at least 6 buffers are created in the HTTP/2 codec (as mentioned in my first post). All those would be combined into one. Also a streaming RPC would all be buffered up until flush.

I am not convinced it wastes memory. I think it might actually use less memory, as keeping the writes in an ArrayList in the WriteCombiningHandler rather than the ChannelOutboundBuffer (massive linked data structure) might use less space. We would only allocate an appropriately-sized buffer and combine the buffers when flush is called.

Direct would already probably be benefiting from releasing the buffer on the same thread, so I'd expect fewer gains.

I am not sure I understand. Can you please elaborate?

[ejona86]

I don't think so. E.g. for a unary RPC response, at least 6 buffers are created in the HTTP/2 codec (as mentioned in my first post). All those would be combined into one.

But that combining is mostly the same, independent of child channel; most of those 6 buffers will still be created with child channel. We only get to remove 1 buffer, and that's the buffer that's most likely to be large.

Also a streaming RPC would all be buffered up until flush.

That's true, although I'd also expect it will be a bit before we elide flushes for child channels.

I am not convinced it wastes memory. I think it might actually use less memory, as keeping the writes in an ArrayList in the WriteCombiningHandler rather than the ChannelOutboundBuffer (massive linked data structure) might use less space.

I'm not talking about management overhead. I'm talking about the buffers themselves. If we have 16K per idle stream, that adds up quickly for a lot of cases. When a stream is idle, we really don't want it to maintain any buffers.

We would only allocate an appropriately-sized buffer and combine the buffers when flush is called.

And that will happen in either case. I'm not comparing with/without WriteCombiningHandler. I'm comparing with WriteCombiningHandler while protobuf serializing on/off the event loop.

Direct would already probably be benefiting from releasing the buffer on the same thread, so I'd expect fewer gains.
I am not sure I understand. Can you please elaborate?

Basically, direct is a different beast, and fits much better with Netty optimizations. The allocation cost itself is reduced because 1) there are much fewer buffer pools, so less memory overhead and 2) it avoids the release queue and stays on the allocator's fast-path. If each unary RPC is 2 buffers (HTTP/2 framing overhead + DATA body) instead of 1 I'd not expect many gains.

[buchgr]

If each unary RPC is 2 buffers (HTTP/2 framing overhead + DATA body) instead of 1 I'd not expect many gains.

But there are many RPCs on a channel, not just one. Say we flush every 100 RPCs, then we are talking 200 buffers vs 1 buffer.

[ejona86]

But there are many RPCs on a channel, not just one. Say we flush every 100 RPCs, then we are talking 200 buffers vs 1 buffer.

If the buffers are small, then they get copied and it can be 1 buffer in either case. If the protobuf is large and we get 2 buffers instead of 1, then I don't see much to be gained. The gain of moving to the transport thread allows avoiding an allocation+copy. And I think most people would rather pay the extra CPU cost for the increased parallelism. Although if the application is using direct then we could use the WrappedByteBuf trick.

[buchgr]

If the protobuf is large and we get 2 buffers instead of 1, then I don't see much to be gained.

My understanding is that a large portion of the optimization is to be able to not use vectorized I/O. That is, always call write instead of writev.

Shall we leave this open, as a low-priority performance optimization?

[pgrosu]

@ejona86 , @buchgr Could you please diagram the flow of data and the edge cases, so that things can be optimized properly. I feel we're now getting into the "I'm right", "No wait, I'm right" territory and optimizing properly using the data and a diagram - with test validation - will get us to the correct implementation.