Increase network buffer sizes even more

[tomaka]

cc #6009

After #6064, it's pretty clear that the spikes in the network worker taking a long time are caused by the fact that the network worker is sleeping while waiting for a libp2p node task to empty the messages in its channel, and that during this sleep the events accumulate in the queues.

We still have spikes, so this PR increases these channel sizes even more, allowing for more parallelism between the network worker and the node tasks.

[arkpar]

Could you give an estimate on how this will increase memory usage?

[tomaka]

That's 16 more "worker to node" events per node task, plus 896 more "node to worker" events. Events are enums and all their fields are small, maybe around 32 bytes? So that'd be around 284 kiB more memory consumed.

[arkpar]

So events do not contain actual network messages or any kinds of heap allocated buffers?

[tomaka]

Well, the number that I gave is what will be the fixed memory overhead compared to today.

Events themselves contain heap-allocated messages, but it's impossible to estimate how that will impact memory usage.

The main point of this change is to prevent the network worker from sleeping, sleeps during which events accumulate. The grandpa messages, for instance, use an unbounded channel (#5481), and I expect to see this channel usage actually be consistantly lower.

[tomaka]

To give an example, let's say GrandPa suddenly starts sending 25 messages to 50 different nodes (so 25 times 50 messages).

Before this PR, the network worker would queue 16 of these messages to the first node, then sleep. During this sleep, the (25 * 50 - 16) other messages would accumulate in the unbounded channel.
Then once the task of the first node starts processing its queue, the network worker will wake up, pull the next messages from the unbounded channel, enqueue these, go to sleep again, and so on.

After this PR, the intention is that the network worker would be able to continuously pull from the unbounded channel and queue all 25 messages to all 50 nodes without going to sleep. If, in parallel, nodes tasks wake up and drain the queue, the peak of memory will actually be lower than before.

The example number of 25 messages is totally crafted, but as long as these spikes show up in the graphs, that means that the channel sizes are not big enough.

[arkpar]

The main point of this change is to prevent the network worker from sleeping, sleeps during which events accumulate.

Why would the network worker "sleep" and for how long?

During this sleep, the (25 * 50 - 16) other messages would accumulate in the unbounded channel.

What unbounded channel are you referring too? I thought the gossip queue is bound and network behaviour skips sending messages if it grows too large.

As far as I understand this is a workaround for the fact that libp2p hash a single choke point. An event stream for a single connection blocks all of the network worker and effectively other connections if it grows too large. Would it be possible to fix that fundamental design flaw instead? E.g. by not dispatching everything through the network worker/swarm, but pushing these events directly to node tasks?

[arkpar]

Also I've just noticed that gossip message batching that I've added in #4055 has been silently removed by a later refactoring. I think that was a much more efficient way to reduce the number of network messages and consequently queued events.
Can we bring it back instead?

[tomaka]

Why would the network worker "sleep" and for how long?

It is sleeping because the queue of messages is full, and the only alternative to sleeping would be to not put a limit to the size of the channel, which is notoriously a bad thing (#5106).

What unbounded channel are you referring too? I thought the gossip queue is bound and network behaviour skips sending messages if it grows too large.

This channel: #5481

substrate/client/network/src/service.rs

Lines 95 to 96 in 812d94d

	/// Channel that sends messages to the actual worker.
	to_worker: TracingUnboundedSender<ServiceToWorkerMsg<B, H>>,

The channel between the gossip machine and the network is unbounded, and making it bounded would likely require an epic refactoring.
It is unfortunately a logic error to drop messages, and the skipping of messages that you're talking about is a hack in order to not have a memory leak. But in order to function correctly, GrandPa at the moment expects no message to ever be dropped.

Would it be possible to fix that fundamental design flaw instead? E.g. by not dispatching everything through the network worker/swarm, but pushing these events directly to node tasks?

While sending messages directly to the node task could be a positive optimization, having a single choke point is done by design. The single choke point is how we avoid getting completely crazy with having different parts of the code having different views of the state of a connection.

Considering that GrandPa is single-threaded anyway (it is a "choke point" by itself), and provided that buffers are large enough (which is what this PR is for), the only drawback of that choke point right now is some added latency when delivering messages (which should be in the order of a few microseconds if things work well).

[arkpar]

I don't see how this has anything to do with the state of the connection. The problem is this weird polling model adopted all over the networking code. Let's say a protocol behaviour needs to send 10 messages for each of the 10 peers we are connected to. So the current design of libp2p forces it to be implemented in the following way:

fn send_a_bunch_of_messages(&self, messages: Vec<Message>) {
   self.some_queue.extend(messages)
}

fn poll(&self) -> MaybeMessage {
   // return a single message here
    return self.some_queue.pop();
}

Later, libp2p calls poll in a loop and distributes messages to handlers. Except if one of them is full it simply stops and all others have to wait. That's what I mean by the choke point. It has nothing to do with grandpa being single-threaded, as it is a design issue with libp2p itself.
This leads to a lot of queues, such as some_queue all over the codebase that are hard to reason about and follow.

I would propose that instead of returning messages that need to be send out from poll one by one, there would be an object that dispatches messages internally to handlers right away, without the poll loop.

fn send_a_bunch_of_messages(&self, messages: Vec<Message>) {
   for m in messages {
      if self.swarm.try_send_to(m.peer, m.data) == Full {
        // handle individual peer connection being to slow here
      }
   }
}

This does not introduce any new state, but simply changes how messages are dispatched to node tasks, does it?

[tomaka]

I don't see how this has anything to do with the state of the connection.

The network worker (which includes the libp2p Swarm) exists because it is the single-threaded authority which decides which connections are opening, open, used, not used, and so on. That's the reason why all the messages go through it, as its job is to relay these messages to the right connection.

This by itself doesn't mean that we can't bypass that blocking behaviour, but it is definitely not unrelated.

Later, libp2p calls poll in a loop and distributes messages to handlers. Except if one of them is full it simply stops and all others have to wait. That's what I mean by the choke point. It has nothing to do with grandpa being single-threaded, as it is a design issue with libp2p itself.

I mention GrandPa being single-threaded, because how it works now is:

Node <-\
Node <-|
Node <--> Network Worker <-> GrandPa
Node <-|
Node <-/

If GrandPa was multithreaded, yes the choke point would be very negative. At the moment, the only drawback of the choke point is the increased latency.

I would propose that instead of returning messages that need to be send out from poll one by one, there would be an object that dispatches messages internally to handlers right away, without the poll loop.

That would indeed make the code more readable, but it is just moving the problem and not solving it.
You can't just call send_message() an arbitrary number of times and expect the code to:

• Deliver all messages successfully.
• Deliver all messages without sleeping on the sending side.
• Have a cap on the memory usage.

Having all three constraints at the same time is physically impossible.

// handle individual peer connection being to slow here

That is exactly the root of the problem. How do you handle an individual node task being slow? Putting a comment won't magically solve that problem.
As a reminder, this has nothing to do with the connection to the peer itself being slow. Node tasks have to accept all events unconditionally and as soon as possible, and when events are accepted by the node task is purely a matter of when the tokio scheduler will schedule the task.

The point of the buffer size increase that this PR does is to not reach a full buffer while tokio schedules the task.

[tomaka]

To give a comparison: TCP connections use a window system for proper back-pressure.
If the window size increase was too low, then you could only transfer a few hundred bytes before waiting for an ACK, and TCP connections in general would be very slow.
But you solve that by increasing the window size until proper speed is achieved, and not by removing congestion control altogether.
The same applies here, but for messages rather than packets of data.

[arkpar]

As a reminder, this has nothing to do with the connection to the peer itself being slow. Node tasks have to accept all events unconditionally and as soon as possible

Why? That does not make any sense. If The peer connection is only 1kbps and we arer trying to send messages at the rate of 1mbps the outbound backpressure needs to be propagated to the code that actually tries to send the message. Because this is where we know how to handle the slow connection. In case of grandpa this is where it may decide what to do with a slow peer. E.g. Only send important round messages, skip a round or simply drop the connection.

The approach that I've suggested gives the protocol behaviour the flexibility to handle outbound connection backpressure and prevents the swarm::poll blocking on a single task.

[arkpar]

To give a comparison: TCP connections use a window system for proper back-pressure.
If the window size increase was too low, then you could only transfer a few hundred bytes before waiting for an ACK, and TCP connections in general would be very slow.
But you solve that by increasing the window size until proper speed is achieved, and not by removing congestion control altogether.
The same applies here, but for messages rather than packets of data.

This is a false analogy. TCP connections do not block on each other. Libp2p task handlers do. If network behaviour generates a lot of events for connection 1, connection 2 (and all of the networking) will be starved.

[tomaka]

Why? That does not make any sense.

It is the node task that must decide what to do with a flow of events that is too high.
The events are not opaque to the node task, in other words it knows which events are sync, which are grandpa, and so on, and can take the appropriate action.

In case of grandpa this is where it may decide what to do with a slow peer. E.g. Only send important round messages, skip a round or simply drop the connection.

That's what I mentioned above: GrandPa is not ready to deal with this, at all. Fixing it would require an epic ten-thousand-lines-of-code refactoring that we clearly can't do, and we have to deal with this constraint.

As I mentioned, we could in theory totally do some changes in the networking code to make GrandPa directly send events to node tasks, but for the foreseeable future we will have this constraint that GrandPa can't communicate with the network without introducing unbounded channels, and doing this change won't bring anything more than what this PR does.

[arkpar]

It is the node task that must decide what to do with a flow of events that is too high.
The events are not opaque to the node task, in other words it knows which events are sync, which are grandpa, and so on, and can take the appropriate action.

Does not really explain why it has to be this way. To make it makes much more sense to handle this in the code that actually decides what to send and to whom.

That's what I mentioned above: GrandPa is not ready to deal with this, at all. Fixing it would require an epic ten-thousand-lines-of-code refactoring that we clearly can't do, and we have to deal with this constraint.

Not convinced this is really that complicated. @mxinden @romanb @twittner Would be good to get your opinion on this.

[tomaka]

I'm going to try to summarize:

• GrandPa sends an unbounded number of messages to the network worker destined to various nodes.
• The network worker (which is normally supposed to be lightweight) is responsible for dispatching these messages to the correct task.
• The tasks discard GrandPa messages if there are too many, which is a violation of the expectations of GrandPa, but is the pragmatic option right now.

The network worker is not supposed to be "back-pressured" (i.e. sleep) itself because one of the tasks is slow. It normally only exists to do the relay between GrandPa and the tasks. That's why node tasks have to accept all events unconditionally.

While we could make GrandPa bypass the network worker altogether, it doesn't solve the problem of GrandPa sending an unbounded number of messages and the task potentially discarding them.

Supposing that GrandPa becomes ready to handle back-pressure, we would indeed have to modify the network worker and introduce new mechanisms in order to make it work. But in the meanwhile, I believe that introducing such mechanism wouldn't bring anything more than what this PR does.

[arkpar]

But in the meanwhile, I believe that introducing such mechanism wouldn't bring anything more than what this PR does.

I can see the following benefits:

1. Get rid of "blocking on a single task" behaviour

2. Remove notification queues all over the networking code. They've already proven to be a source of numerous hard to spot issues.

3. Make handling backpressure more straightforward.

4. Make the code more simple and easier to follow.

The way I see it can be done in 3 steps

1. Release libp2p version that allow sending events directly through swarm reference (or an object that encapsulates it). Sending is allowed to fail if an internal bounded event queue for the node task is full.

2. Gradually use it in substrate. Not only in gossip, but all protocols.

3. Eventually drop NetworkBehaviourAction

[tomaka]

Release libp2p version that allow sending events directly through swarm reference (or an object that encapsulates it).

I do believe that the way libp2p works is fundamentally correct. Your proposal can be implemented by adding a channel within Substrate and does not require changes in libp2p.

I also in general don't think that it's a good idea to even consider refactoring half of libp2p and Substrate as an alternative to a two lines of code PR.

Remove notification queues all over the networking code. They've already proven to be a source of numerous hard to spot issues.

I'm only aware of one such hard to spot issue, and it is caused by the sync state machine working differently than the rest of the networking state machine.

[twittner]

That's what I mentioned above: GrandPa is not ready to deal with this, at all. Fixing it would require an epic ten-thousand-lines-of-code refactoring that we clearly can't do, and we have to deal with this constraint.

Not convinced this is really that complicated. @mxinden @romanb @twittner Would be good to get your opinion on this.

If we replace the unbounded channel with a bounded one in substrate's NetworkWorker, then finality-grandpa must be able to handle the back-pressure. See https://github.com/twittner/substrate/tree/issue-5481 for an initial attempt to replace the channel.

How do you handle an individual node task being slow? Putting a comment won't magically solve that problem. As a reminder, this has nothing to do with the connection to the peer itself being slow. Node tasks have to accept all events unconditionally and as soon as possible, and when events are accepted by the node task is purely a matter of when the tokio scheduler will schedule the task.

[...]

It is the node task that must decide what to do with a flow of events that is too high.
The events are not opaque to the node task, in other words it knows which events are sync, which are grandpa, and so on, and can take the appropriate action.

When you say "node task", what are you referring to exactly? Surely it can not be libp2p's task.rs.

[romanb]

Please treat all my comments below with a grain of salt, as I may not have the full picture of all the details that are being discussed here.

After #6064, it's pretty clear that the spikes in the network worker taking a long time are caused by the fact that the network worker is sleeping while waiting for a libp2p node task to empty the messages in its channel, and that during this sleep the events accumulate in the queues.

This is due to the single pending_event in the Swarm, right? And increasing the buffer sizes in libp2p-core is basically a countermeasure to reduce the likelihood of the pending_event in the Swarm not being able to be delivered? So this is the "outgoing" direction, e.g. for sending something to a peer (via the behaviour emitting an event). And since the Swarm drives the underlying Network and NetworkBehaviour in tandem, a single connection being slow to send stuff even results in all connection handlers receiving events slowly?

I'd just like to differentiate the problem, because I don't currently think that the buffers themselves in libp2p-core are a problem, since using libp2p-core directly, you can send events to specific connections directly and so one connection's background task being slow to make progress does not need to stop the user from sending events to other connections. In other words, this seems to be a problem of libp2p-swarm. I would tend to agree that the current design of libp2p-swarm in such a way introduces what seems to be an artificial bottleneck, and as I can see it caused primarily by the inability of a Swarm to communicate back-pressure between itself and the NetworkBehaviour, e.g. when the behaviour emits an event for a connection, the Swarm must deliver or buffer it, but cannot "give it back" to the behaviour if the connection is busy.

As far as this PR is concerned and if my understanding of the problem(s) is correct, increasing the buffer sizes seems like a short-term stop-gap measure that may be fine for now, but certainly not a desirable long-term solution, but as stated initially, I may not have the full picture. I opened libp2p/rust-libp2p#1585 for some related discussion.

[tomaka]

When you say "node task", what are you referring to exactly? Surely it can not be libp2p's task.rs.

I'm indeed referring to the tasks spawned in task.rs. By "node task", I'm also including the code in sc_network/src/protocol/generic_proto/handler.rs, which is where we know which event corresponds to what.

This is due to the single pending_event in the Swarm, right? And increasing the buffer sizes in libp2p-core is basically a countermeasure to reduce the likelihood of the pending_event in the Swarm not being able to be delivered? So this is the "outgoing" direction, e.g. for sending something to a peer (via the behaviour emitting an event). And since the Swarm drives the underlying Network and NetworkBehaviour in tandem, a single connection being slow to send stuff even results in all connection handlers receiving events slowly?

Correct.

I would tend to agree that the current design of libp2p-swarm in such a way introduces what seems to be an artificial bottleneck, and as I can see it caused primarily by the inability of a Swarm to communicate back-pressure between itself and the NetworkBehaviour, e.g. when the behaviour emits an event for a connection, the Swarm must deliver or buffer it, but cannot "give it back" to the behaviour if the connection is busy.

The reason why ProtocolsHandler::inject_event is a synchronous method is that we want to make it mandatory for a ProtocolsHandler to immediately accept events that are being sent to it.
If the channel that transmits messages to the node task is full, the Swarm only has to sleep until the executor (e.g. tokio) schedules the node task.

(the exception to that is if there's some CPU intensive action in a node task, or if it is stuck in an infinite loop, which thankfully doesn't happen, but we should eventually have some protection measure here)

It is also pretty important for the ProtocolsHandler to immediately accept all events to prevent a potential deadlock where the node task is blocked trying to deliver an event to the Swarm while the Swarm is blocked trying to deliver an event to that node task.

While giving it back to the NetworkBehaviour could be a solution, it moves the complexity from the Swarm to the NetworkBehaviour.

[romanb]

The reason why ProtocolsHandler::inject_event is a synchronous method [..]

I don't actually think this is part of the problem, is it? As these calls are scoped to a particular connection (and background task). As far as I currently understand, the problem is rather that NetworkBehaviour::poll and NetworkBehaviour::inject_event do not permit communicating (connection-specific) busyness / back-pressure between the network and behaviour via the swarm, and so essentially the Swarm can only progress as fast as the slowest connection (+ leeway due to buffers) if it does not wish to offload the problem of back-pressure onto the behaviour.

[tomaka]

I don't actually think this is part of the problem, is it?

I do think it is important. The fact that ProtocolsHandler::inject_event is synchronous means that there is a finite time before the Swarm will be able to wake up.

and so essentially the Swarm can only progress as fast as the slowest connection (+ leeway due to buffers) if it does not wish to offload the problem of back-pressure onto the behaviour.

Yes, but an important point is that it is not the slowest connection, as this word gives the impression that this is dependant on the bandwidth of the connection, which is not the case at all. It is rather based on the CPU usage of the node task. Node tasks are supposed to be idle 99% of the time.
If the Swarm goes to sleep, it is only to wait until tokio (or whatever executor we use) schedules the relevant node task.

[arkpar]

Yes, but an important point is that it is not the slowest connection, as this word gives the impression that this is dependant on the bandwidth of the connection, which is not the case at all. It is rather based on the CPU usage of the node task.

Where is send called on the TCP socket? Does this happen in the node task? And if so, is the socket open in blocking mode?

Regardless, the only reason to have a queue in the task handler is for packets that wait for their turn to be sent, isn't it? What other possible causes for anything to be queued there? All other types of events could simply be processed synchronously.

[tomaka]

Where is send called on the TCP socket? Does this happen in the node task? And if so, is the socket open in blocking mode?

That's indeed in the node task, and done by mio, the non-blocking sockets library.

[arkpar]

Where is send called on the TCP socket? Does this happen in the node task? And if so, is the socket open in blocking mode?

That's indeed in the node task, and done by mio, the non-blocking sockets library.

It's non-blocking for receiving, sure. But I could not confirm that writes are non-blocking too. Anyway, the queue is the node handler grows when the socket can't send messages fast enough, doesn't it? Or are there any other reasons for a handler not to accept an event?

[arkpar]

Let's test this on a sentry node with ~150 connections for a couple of days and see if the memory usage increase is significant

[tomaka]

This PR got deployed on a sentry node around midnight.

The CPU and memory usage look the same before and after the deployment (they vary a lot over time under normal circumstances, so any difference before and after this PR is completely lost in the background noise).

Here is the CPU load of the network worker specifically, in blue:
One can see that indeed the spikes are lower.

These are the messages that the GrandPa messages processing task receives:

Again, it's smoother. The spikes have disappeared.

To summarize the situation, I totally agree that we should make notifications go directly to the background task (and I'm slowly working towards this), but I do believe that this PR improves the situation at basically no cost.

Most likely the additional memory consumption is compensated to some degree by the fact that the unbounded channels, for example the one towards GrandPa, consume less memory in the absence of spikes.

[gnunicorn]

bot merge

[ghost]

Trying merge.