@carl-mastrangelo, we can't merge this right now, right? Because otherwise we could leak buffers?

I'm pretty sure we already leak buffers, if close / flush is never called.

It is worth noting that reducing the minimum size of the buffer in NettyWritableBufferAllocator to 32 produces a similar effect. It looks like it may be slightly faster for the Flow Control benchmark.

I'm pretty sure we already leak buffers, if close / flush is never called.

Which is not currently possible, with the exception of unary requests that lack a halfClose. It's hard to consider a unary request that lacks a halfClose that isn't a blatant bug in the application and wouldn't be noticed when the unary request is never sent. That leakage is impossible when using our stubs.

With this PR the only way we don't leak is if halfClose/close is called (note that cancel() isn't enough). That impacts every outbound streaming RPC. That concerns me, and doesn't seem like something that can be ignored.

It is worth noting that reducing the minimum size of the buffer in NettyWritableBufferAllocator to 32 produces a similar effect.

That sounds like a much easier solution. If it gets us performance in the same ballpark, I'd be much more eager to take it.

If we're going to retain direct buffers in MessageFramer, we really have to have it thread-safe enough that we can release those buffers when the RPC completes. That would be a pain, but I would understand if we needed to do so.

From skimming over the code it seems like here we are also allocating a 4K buffer for when we know we only need 5 bytes ... https://github.com/grpc/grpc-java/blob/master/core/src/main/java/io/grpc/internal/MessageFramer.java#L204

It is worth noting that reducing the minimum size of the buffer in NettyWritableBufferAllocator to 32 produces a similar effect. It looks like it may be slightly faster for the Flow Control benchmark.

I agree with Eric that this sounds much simpler. Why 32? Is that specific to the benchmark payload size? Do we know how often piggy-backing writes in the message framer actually happens?

32 = 5 bytes of frame which is large enough to hold small messages. Next size down is 16, which my gut says is too small.

Not changing the smallest size was based on a conversion with Louis. I was under the impression that changing the size was a not feasible.

Reusing the buffer also has some benefits, assuming that chunks are returned in powers of two. A stream of large messages that straddle the power of two boundary would still cause a lot of waste (possibly <1/2 the buffer).

If the allocator returns chunks of power size 2^-3 (as in there will never be more than 12.5% waste) then it doesn't make much sense to reuse the buffer. That is a decision of the WritableBufferAllocator and the Writable buffer. They can decide if splitting the buffer makes sense based on decisions the MessageFramer can't which is why it can return null.

As for your question about piggy backed writes: they happen in streaming RPCs, which is why the benchmarks no longer OOM with this change. Another common case is streaming large chunks of data from a database, as Cloud Spanner does.

Having the stream close out the buffer safely is possible.

@carl-mastrangelo

As for your question about piggy backed writes: they happen in streaming RPCs, which is why the benchmarks no longer OOM with this change. Another common case is streaming large chunks of data from a database, as Cloud Spanner does.

I thought on server streaming we call framer.flush() after every message https://github.com/grpc/grpc-java/blob/master/core/src/main/java/io/grpc/internal/ServerCallImpl.java#L147 ? Am I wrong?

@buchgr You are correct, but I don't think that is relevant. The next write after the flush is going to want a buffer.

Just so I don't confuse things. Here is my understanding.

Right now (current master), we don't piggy-back writes for server streaming. We only piggy-back writes for client streaming, as we don't flush until 4K have been written or half close (whatever comes first). So in a sense that 4K minimum size right now is mostly useless, in terms of doing fewer allocations and wastes lots of space for small writes. Correct?

Now in this PR you suggest to slice the buffer after flush and re-use. Which sounds like a good idea, but with the downside that we might leak buffers if halfClose / close never gets called. Did you measure the allocation latency? It might be so fast that slicing vs. allocating doesn't really matter. Note, that small allocations are served from thread caches without synchronization. So doing several small allocations vs. one 4K allocation might be faster. I am not exactly sure how Netty's allocator thead caches work. Real jemalloc does cache allocations up to 32K by default, but with fewer slots for larger buffers. So it might cache e.g. up to 1000 64 byte buffers, but only 10 4K buffers or so. Allocation / Deallocation in a thread cache in jemalloc is just pointer pumping.

Your other suggestion is to do one allocation per write, with that allocation having a minimum size of 32 bytes. Now my question, why have the minimum size if we effectively never piggy-back? On the fast-path (known payload length), we know how much space we need anyway.

Right now (current master), we don't piggy-back writes for server streaming. We only piggy-back writes for client streaming, as we don't flush until 4K have been written or half close (whatever comes first).

Current master does no piggy-backing to my knowledge. I also think any solution we would apply to both client streaming and server streaming.

Your other suggestion is to do one allocation per write, with that allocation having a minimum size of 32 bytes. Now my question, why have the minimum size if we effectively never piggy-back?

I think the minimum was originally put in to avoid fragmentation in the allocator, although it was also written with flush delaying in mind (although we don't do that today).

Yeah, we many not really need the minimum size. We just tell Netty what we need and let Netty's allocator do what it thinks is best.

The only advantage we have vs the allocator is that the allocator API has stricter semantics than we need. When we allocate a size, the size is only a hint. We can handle smaller sizes and sometimes make use of larger. Netty's allocator may allocate something larger, but I haven't seen an API to know how much extra space there is past the capacity we requested.

Right now the min and max in NettyWritableBufferAllocator is semi-arbitrary. I don't believe we've done any real investigation on whether the numbers are appropriate. It was introduced in f7010f2 via #313 and hasn't changed since.

Current master does no piggy-backing to my knowledge. I also think any solution we would apply to both client streaming and server streaming.

You are right. So if we never piggy-back, then there is no point in having a minimum size really?

I think the minimum was originally put in to avoid fragmentation in the allocator, although it was also written with flush delaying in mind (although we don't do that today).

I think we should have confidence in the allocator and let the allocator deal with fragmentation.

Netty's allocator may allocate something larger, but I haven't seen an API to know how much extra space there is past the capacity we requested.

It will fit an allocation to a specific size class. Those size classes were chosen to keep overhead to a minimum, so I think that's fine. I would argue that the allocator would certainly do a better job with keeping that extra capacity to a minimum, than us choosing some fixed size class.

So we get rid of the minimum size in NettyWritableBufferAllocator @carl-mastrangelo @ejona86 ?

While working on #2139, I noticed that a large minimum size does make sense for another reason, that's not immediately obvious.

In the MessageFramer we try to size the buffer to at least the message size, however, that might not always be possible (okhttp, or large messages in netty) and so we might get a smaller buffer than we need. In that case, MessageFramer.writeRaw has logic to allocate additional buffers, if the current buffer is full [1]. It allocates the new buffer according to the length of the byte[]. However, since in our fast path we are zero copy those byte arrays will often be tiny (i.e. less than 10 bytes, as we might call OutputStream.write immediately for each produced "element" without buffering). I think for protobuf, writeRaw is called for each message entry.

That's more a problem of the current code structure though and could be fixed, as we call writeRaw only when we know the length of the input stream. Just wanted to mention it here.

[1] https://github.com/grpc/grpc-java/blob/master/core/src/main/java/io/grpc/internal/MessageFramer.java#L245

But just before writeRaw we will generally allocate an appropriately-sized buffer. (Maybe that allocation in writeRaw should go away? It is probably left-over from an earlier version of the code)

@ejona86

But just before writeRaw we will generally allocate an appropriately-sized buffer.

yep, but the buffer allocator is allowed to return a buffer less than appropriately-sized bytes. Which seems not-ideal. Should rather push that complexity of allocating a big enough buffer further down into the allocator and not have the framer worry about it.

Which seems not-ideal. Should rather push that complexity of allocating a big enough buffer further down into the allocator and not have the framer worry about it.

But if the message is 1 GB, we don't want a contiguous buffer. 1024 1 MB buffers would be fine. We could make writeRaw a bit smarter and inform it how much we expect is remaining? So then if we get a smaller buffer initially we will continue asking for large buffers.

But if the message is 1 GB, we don't want a contiguous buffer. 1024 1 MB buffers would be fine.

In case you are thinking about peak memory usage - we wouldn't be flushing until that 1GB has been framed. So for peak memory usage it should make no difference.

Also I am not saying that a contiguous 1GB buffer should be returned. In fact, the allocator might not be able to do that. It's specific to the transport. However, it could always return a CompositeWritableBuffer that is backed by 1024 1MB buffers?

@buchgr Having to allocate any memory seems like a problem in a near future. I expect that proto will be able to self describe as ByteBuffers (or equiv) and thus not need to be copied by message framer. In that world, does it make sense to move serialization into the transport?

In case you are thinking about peak memory usage - we wouldn't be flushing until that 1GB has been framed. So for peak memory usage it should make no difference.

I was not thinking of peak memory usage. I was thinking more about fragmentation and the effects of large contiguous allocations. For example, it would be virtually impossible to reliably allocate a 1 GB buffer on a 32 bit VM, assuming you have memory available.

The flushing could be changed fairly trivially, but I don't expect much benefit from going down that road. Pretty soon we will want protobuf to stop implementing KnownLength, because that will allow writing backward with only a single pass over the protobuff, instead of two right now: one to get lengths, one to write. Right now protobuf itself requires two passes, so it is "free" for us to get the length. But Nathan's work to write backward will be available eventually.

Also you can't release any of the 1GB buffer until the entire thing is written. Splitting the buffer into smaller pieces is a feature, not a downfall. At some point combining buffers stops being helpful/performance gain. And that makes sense, because otherwise writev/readv wouldn't exist. Only small buffers cause issue.

Also I am not saying that a contiguous 1GB buffer should be returned. In fact, the allocator might not be able to do that. It's specific to the transport. However, it could always return a CompositeWritableBuffer that is backed by 1024 1MB buffers?

I don't see what advantage that provides. Whether it is 1 CompositeWritableBuffer with 1024 ByteBufs or a list of 1024 ByteBufs it seems mostly the same to me. Yes, it might be slightly more efficient than ChannelOutboundBuffer, but I'd have trouble believing it wouldn't be in the noise since it is amortized over 1 MB of data. Basically, if the message is large, then a lot of the costs stop mattering because they are fixed-cost or heavily amortized. As long as you manage the data is reasonably large chunks and don't do needless copies, then it's fine.

I expect that proto will be able to self describe as ByteBuffers (or equiv) and thus not need to be copied by message framer. In that world, does it make sense to move serialization into the transport?

What self describing feature are you talking about? I've not heard anything about that. (Or rather, I've not heard of anything along those lines that isn't a completely different wire format.) Even if we had that though, it seems like it wouldn't matter where we serialize since serializing doesn't actually need to serialize anything, so it is close to free.

I do know there is work to serialize directly to ByteBuffers as an alternative to byte[]. But that optimization isn't fundamentally different.

I don't see what advantage that provides. Whether it is 1 CompositeWritableBuffer with 1024 ByteBufs or a list of 1024 ByteBufs it seems mostly the same to me.

My thinking was that the code in MessageFramer would get much simpler, if the framer can rely on the allocator returning a buffer that's at least the size it needs.

My thinking was that the code in MessageFramer would get much simpler, if the framer can rely on the allocator returning a buffer that's at least the size it needs.

Oh, okay. That's fair. That may be the case. Although I think MessageFramer is probably the most stable class we've had on the critical path of an RPC. And the complicated parts are in the paths that we don't know the message size. We also have 3 implementations of WritableBuffer. If we went down that road, Cronet (which uses plain ByteBuffers) would need a composite, but I guess that isn't too much code.

My instinct tells me that we don't save much though. In that, by trying to simplify we make things more complicated, more opaque, and harder to optimize. Especially if we ever stop flushing after every message.

The following code of allocating 1024 blocks of 1MB each would take about 15 seconds - which could be a bit long - taking into account an insert for a LinkedList is O(1):

import java.util.LinkedList;

public class GBConsecutive
{
  public static void main(String[] args)
  {
    long tick = System.currentTimeMillis();
    LinkedList<byte[]> ll = new LinkedList<byte[]>();
    int ONE_MB = 1048576;
    for ( int i = 0 ; i < 1024 ; i++ ) {
      byte [] ONE_MB_BYTE = new byte[ONE_MB];
      ll.add(ONE_MB_BYTE);
    }
    long elapsed = System.currentTimeMillis() - tick;
    System.out.println("Elapsed milliseconds: " + elapsed );
  }
}

Now the above code runs in 46 seconds, and when I perform it all at once like this, it runs in 28 seconds:

import java.util.LinkedList;

public class GBConsecutiveAll
{
  public static void main(String[] args)
  {
    long tick = System.currentTimeMillis();
    LinkedList<byte[]> ll = new LinkedList<byte[]>();
    int ONE_GB = 1073741824;
    for ( int i = 0 ; i < 1 ; i++ ) {
      byte [] ONE_GB_BYTE = new byte[ONE_GB];
      ll.add(ONE_GB_BYTE);
    }
    long elapsed = System.currentTimeMillis() - tick;
    System.out.println("Elapsed milliseconds:" + elapsed );
  }
}