Optimize SimpleSubscriber for Netty

[kciesielski]

Fixes #3548

This PR updates the SimpleSubscriber with following improvements:

1. If there is only one chunk of data (by default for body length < 8192), read it from the buffer and return immediately, requiring only one array allocation
2. If there are multiple chunks, collect all Netty ByteBufs without rewriting them into arrays, then copy them into the final array in onComplete. Disclaimer: initially I wanted to cover this case with Netty's CompositeByteBuf, but it turned out to be bad idea. It does some reallocations to resize internal representation underneath, resulting in making the overall performance significantly worse.

Results:
PostBytes Simulation

	Before	After
CPU samples	30.86%	29.73%
throughput	44738	56955

Latency improvement:

PostLongBytes Simulation

	Before	After
CPU samples	48.8%	38.69%
throughput	363	454

Latency improvement:

I haven't measured it, but there should be also noteworthy gains in memory allocations.

[adamw]

shouldn't the subscriber enter into some state, where another arriving content would report an exception? it's possible that the incoming header is malformed

[kciesielski]

I added a handler for this case - now if the result queue.offer call returns false, we follow up with canceling the subscription - as far as I checked that's the recommended way to 'abort'.

[adamw]

actually ... what are the concurrency guarantees for a subscriber - can multiple onNext be called concurrently? or maybe onNext + onError? I'm wondering if we (a) need a concurrent data structure here at all and (b) if concurrency is allowed, is the impl safe

[kciesielski]

The onNext, onError, and onComplete operations are guaranteed to be called sequentially without concurrency, so we are safe to replace the concurrent data structure with something simpler. I tried with a ListBuffer and it actually gave another noticeable boost to throughput and latency for PostLongBytes.

[adamw]

Nice results! :)

[adamw]

follow-up question (sorry ;) ) - onNext/onComplete/onError are guaranteed to be called from one thread, but is it going to be the same thread? that is, does buffers need to be volatile?

[kciesielski]

We don't have such a guarantee, and maybe that was why I had a ConcurrentLinkedQueue for byte arrays in the previous implementation, but I forgot :) This means I either fall back to it, or use a volatile ListBuffer.
I guess var totalLength is also unsafe and should be converted to an AtomicInteger, right?

[kciesielski]

Update: You can't have a volatile val, so I fell back to ConcurrentLinkedQueue. For totalLength I chose @volatile instead of AtomicInteger, because we are using this variable sequentially, our only scenario is increasing it in onNext and reading in onComplete.

[adamw]

but we can have a volatile var with an immutable list - always less synchronisations

[kciesielski]

Ok, I replaced the ConcurrentLinkedQueue with a volatile var ListBuffer. In theory, ListBuffer should have slightly cheaper append time than a Vector, and we don't need Vector's fast random access, which takes additional cost of maintaining more complex underlying structure.
I don't see any improvement of throughput, but there's a slight improvement in latency.

[adamw]

I'm not sure this properly protects the value - ListBuffer is mutable, and the volatile only ensures there's a memory barrier before reading the buffers reference (not references inside the ListBuffer). But maybe since we have a memory barrier, everything will be synchronized correctly ... as you can't really access the inner references before first reading the buffers reference (which creates the barrier).

Anyway, I though about a simpler design, using immutable data structures, where you don't have to think that much ;-) But maybe this one works as well :)