Serialize outbound messages on netty buffers

[Tim-Brooks]

Currently we use BigArrays for serializing outbound messages. Since
these messages are only handled at the network layer, it is better to
use the Netty allocator when using the netty transport. This commit
implements this serialization method when netty is enabled.

[elasticmachine]

Pinging @elastic/es-distributed (Team:Distributed)

[jpountz]

I only had a high-level look but it looks good. Being closer to the byte[] where the data is written and using the VarHandles will hopefully make writing data faster.

NetworkStreamOutput looks to me as a better BytesStreamOutput in many cases. E.g. I believe it would also be a better fit than BytesStreamOutput in GeometryDocValuesWriter? If so maybe we should give it a more generic name that doesn't sound like it's only a good fit for network messages?

[Tim-Brooks]

@jpountz - I made your changes. As I mentioned offline, the stream I created RecylerBytesStreamOutput could replace BytesStreamOutput. The main scenario where BytesStreamOutput works better currently is for expected values much smaller than the page size.

Currently, RecylerBytesStreamOutput use with the non_recycling_instance will allocate a minimum of 16KB for streams. So probably would over allocate bytes in the case of DimensionalShapeType. If you want me to investigate making RecylerBytesStreamOutput replace all instances of BytesStreamOutput I can improve its performance on streams that are very small.

Or I could look into that in a follow-up.

[jpountz]

Thanks for explaining. Looking into replacing BytesStreamOutput as a follow-up sounds good to me as it will make it easier to evaluate the potential downsides for networking.

[original-brownbear]

LGTM on the code changes this looks really nice and IMO is 100% the right thing to do.
The one thing I'm a little worried about is the change in behavior for very large transport messages.
Previously, the BigArrays would eventually stop pooling bytes and move to un-pooled allocation after going over the configured heap threshold, this isn't the case any longer with this change and we'd potentially pool a lot more memory than we used to after a sporadic spike in required buffers (thinking about things like servicing a CS request for stats with a large cluster state or so). What is Netty's behavior here in terms of downsizing the buffer pool again after dealing with a message of hundreds of MB on a 2G master for example?

[Tim-Brooks]

What is Netty's behavior here in terms of downsizing the buffer pool again after dealing with a message of hundreds of MB on a 2G master for example?

Netty immediately releases PoolChunks when they are no longer utilized. PoolChunks are larger that 16KB. However, in the case of large messages you are discussing, there will be minimal waste.

With that said, we have obviously implemented our own recycler here that is used for inbound decompression and outbound messages. If we are concerned about overuse, we can add a counter and fallback on JVM allocations once it crosses a certain threshold. We can discuss.