Prevent OOM when receiving large request streams

[davidlar]

This PR aims to fix 3173 by making sure we don't read more data than we can handle.
Now ctx.read() is called after the queue is drained by the consuming end of the stream. The call to ctx.read() will fill the queue with more data asynchronously.
Nettys AUTO_READ flag is already turned off in zio-http and ctx.read should be called when we are ready for more data, not before.

I'm not sure if we should keep the unbounded queue or not. I guess it will only contain one chunk at most.

[kyri-petrou]

Thanks for the fix! AFAICT the fix works, although the part that worries me a bit is that with these changes we're delegating ctx.read() to a different threadpool which is will add a fair bit of overhead due to context switching.

I'm wondering whether we should add a continueReading: () => Boolean argument to UnsafeAsync.Streaming. This way we can decide in AsyncBodyReader whether we'll invoke a ctx.read() or defer it. The part that worries me with this is that we might be adding too much complexity. What do you think?

[kyri-petrou]

I'm wondering whether we should buffer reads here, e.g., nettyRead.whenZIO(queue.size.map(_ < XXX))

Previously this was set to 4096 (using a bounded queue which we can't use anymore since it was blocking Netty's threads) and there weren't any issues right?

[kyri-petrou]

Also, 2 notes on performance.

1. Use whenZIODiscard instead of whenZIO
2. I think it might be better to store this ZChannel in a val to avoid creating it on each loop:

val maybeRead = ZChannel.fromZIO(nettyRead.whenZIODiscard(???))

// And then
a => ZChannel.write(a) *> maybeRead *> loop,

[davidlar]

I'll fix the suggested changes in NettyBody!

[davidlar]

Not sure I understand how ctx.read() can be called from AsyncBodyReader as it has be called from the consuming side of the queue. So it's probably to complex for me at least :-)

[davidlar]

I tried locally with queue.size.map(_ < 4096) and it can be a lot of data since the chunks in the queue can be big in some cases.
Maybe we should count the queued bytes instead in a ref or something. Or lower that limit to 128 or so.
What do you think @kyri-petrou ?

[davidlar]

It feels like the buffering before #3060 was meant to be maximum 4096 bytes, but was really max 4096 chunks which is not what we want, I guess...
On the other hand I haven't seen smaller chunks than 8kb when testing this, so maybe queue.isEmpty is a good choice after all.
I leave it like that if you want to approve! I will go away for the weekend...