Handling long streams of data, or handling header and body efficiently?

[lukasa (Cory Benfield)]

@jamamp

My question, then, would be if it's possible to temporarily not wait on the channel's socket file descriptor, which the EventLoop/Group internally uses in its epoll/kqueue system call.

Yes. This is known as "backpressure", and NIO supports it via the mechanism of the read function on ChannelInboundHandler.

Under the hood, a NIO channel has a read cycle. The cycle is approximately as follows:

1. The idle state. Here, the socket is not registered for readability notifications with the system selector: we are not asking to read any data. No reads can be delivered.
2. A read() call that has passed through the pipeline reaches the head of the ChannelPipeline and is delivered to the Channel. This triggers NIO to register the socket with the selector for readability notification.
3. Some time later (potentially as soon as the next event loop "tick") the socket is marked readable by the kernel. NIO begins processing reads on that channel.
4. NIO now spins in a loop on the socket calling the read system call.
   1. Each call to read may read some data.
   2. If it does, channelRead is called with the data we read, and we loop around again.
5. NIO runs this loop until either the read system call returns EAGAIN, indicating that there is no more data to be read at this time, or until ChannelOptions.maxMessagesPerRead calls have been made.
6. At the end of the loop, we invoke channelReadComplete.
7. Presuming we're still open and no new calls to Channel.read() came in the meantime, we return to the idle state and the sequence of operations here begins again.

This should make it fairly clear how to stop NIO from reading from the socket: you stop calling Channel.read. But for most users this might be surprising: after all, they have never called Channel.read. They didn't even know it existed!

This is because NIO has a channel option called autoRead which is turned on by default for all NIO channels. This option will automatically call Channel.read if a read sequence completes and no read() call was already made. This means that in practice Channels are constantly having read called on them, but they call it on themselves.

Happily, then, this gives us a good design for how to exert backpressure. To talk about how we do it, I'll use an example from the NIO repo itself: the HTTPServerPipelineHandler. This ChannelHandler does something quite close to what you're discussing: it arranges it so that a HTTP server that uses it never sees pipelined requests, even if a client sends them. To arrange that, it counts the messages passing through, and when it gets a request .end it doesn't allow any further request messages until the user writes their response .end. Instead it just buffers them, and replays them when the response .end comes through. Simple enough.

The wrinkle, as you identified, is that if this handler just buffered in memory then it would be a nasty little DoS vector. Remote users could pipeline a bunch of requests behind a long-running one and we'd hold them all in memory. Not good! So what it wants to do is, whenever it's passed on a full HTTP message and is waiting for the response, it would like to ask NIO to stop reading from the network.

To do that, it implements ChannelOutboundHandler.read. If it's happy to accept more data, it allows the read call straight through. But if it's waiting for the response .end, it will instead just mark that the read call happened, and do nothing. Later, once it has seen the response .end it was waiting for, it will start letting through read() calls and will trigger a read() if it dropped any earlier on.

This works seamlessly regardless of whether autoRead is on or not, by having the ChannelHandler reduce the read() pattern to the one it wants. The ChannelHandler doesn't issue the reads itself: that would prevent the user from managing their own backpressure. Instead, it adds constraints: if the user wanted more data, but we don't, we suppress the read. Otherwise, do what the user wants.

I'd like to add some detail while I'm here. Firstly, read calls are not cumulative: they are not like Combine's Demand. Each read call will trigger zero or one channelReadComplete calls. If two read calls are made before a channelReadComplete, they will trigger only one channelReadComplete: another read is needed. Essentially, reading is a binary state: we're either waiting for data or we aren't, and asking to read when we were already waiting doesn't change anything.

Secondly, even if you exert backpressure you still need buffers. ChannelPipelines are still pipelines, and just because we stopped reading from the socket doesn't mean you'll immediately have no further data delivered to you. In the most straightforward case, stopping read calls will still potentially allow as many as maxMessagePerRead more chunks of data to come your way! Backpressure works in combination with buffers in NIO-land, not instead of them.

Finally, this is an important design principle that all NIO Channels follow, even the ones that don't have sockets behind them. This works on socket channels, pipe channels, NIO Transport Services channels, and HTTP/2 stream channels.

@jamamp

The closest thing I can find to this is the ChannelHandlerContext's fireChannelInactive to temporarily stop SwiftNIO from pushing data into the application when we want to instead wait, and then fireChannelActive to re-activate it once the stream is being read from. I'm not sure if that's what these functions actually do, though. I would fear they would kill the connection, or we aren't really supposed to use them like this, in normal read operations.

This is the wrong thing to do. active/inactive are one-time switches, and they are controlled by the channel, not you. Those messages are notifications, not channel control operations, and merely sending them doesn't change the truth of what is happening.