Gossipsub Message Ordering

[erhant]

In our p2p network where there are 2 types of nodes: RPC and Compute.

• RPC basically connects this network to outside, and published messages in the form of tasks.
• Compute nodes handle these tasks and reply with the results back to the RPC.

We make several measurements throughout the lifecycle of a message, as follows:

sequenceDiagram
  actor R as RPC
  actor C as Compute

  note over R: R_p = publish timestamp
  R ->> C: task
  note over C: C_r = receive timestamp

  activate C
    note right of C: C_e = execution duration
  deactivate C

  note over C: C_p = publish timestamp
  C ->> R: result

  note over R: R_r = receive timestamp

Loading

In the diagram above we have the following:

• $R_p$ is the timestamp when RPC publishes the task message
• $C_r$ is the timestamp when Compute node receives this task message
• $C_e$ is the execution time of the task, its a duration; not a timestamp.
• $C_p$ is the timestamp when Compute node published the result message
• $R_r$ is the timestamp when RPC receives this result message

-- | ------------- | ============== | --------- | --
  R_p             C_r      C_e     C_p         R_r

From these, we define 4 values:

• publish_latency: $C_r - R_p$ i.e. the time it took the message to go from RPC to Compute node
• receive_latency: $R_r - C_p$ i.e. the time it took the message to go from Compute node to RPC
• execution_time: time spent during the task execution (in our case this is LLM requests, some via API and some using locally hosted models, irrelevant for this discussion though)
• roundtrip: this is the sum of all 3 values above, which is roughly equivalent to $R_r - R_p$.

Note that the timestamps are given at application level, e.g. $C_p$ is recorded when we call publish method of GossipSub.

Problem

Assume that 3 messages are sent for the same task, so that execution times are roughly the same for each task. What we have observed is that with consecutive messages, publish_latency increases and receive_latency decreases, here is an example to give an idea:

{'publish_latency': 2, 'execution_time': 10, 'receive_latency': 8, 'roundtrip': 20}
{'publish_latency': 5, 'execution_time': 10, 'receive_latency': 5, 'roundtrip': 20}
{'publish_latency': 9, 'execution_time': 10, 'receive_latency': 1, 'roundtrip': 20}

What we understand from this is that the messages are first queued in the receipt order, and the messages to be sent are queued as well. Once all tasks are finished, the receive queue is finished and send queue is consumed, sending all these messages at once.

How can we prioritize publish messages to be consumed before the received messages, so that roundtrip times are slower?

For the record, we are using libp2p with GossipSub, Kademlia DHT, DCuTR. (https://github.com/firstbatchxyz/dkn-compute-node). Here is our main loop that handles swarm events.

[elenaf9]

Hi @erhant, three questions for clarification:

• Where is C_r measured? After you polled the gossisub message event from the swarm?
• Is this code the "task handling" in the compute node that you are referring to?
• What's the frequency of the publish messages?

One thing I noticed in your code is that you stop polling the swarm while handling an gossipsub message. When the swarm is not polled, it won't make any progress. So if you receive a second publish message while the first one is still being handled, it will affect the publish latency. A solution for that would be e.g. to poll the swarm in a separate task continuously and forward the received gossipsub messages in a channel.

[erhant]

by the way even with the multi-threaded approach, gossipsub queue will still prioritize the received messages BEFORE the messages to be published, right?

[elenaf9]

How can we prioritize publish messages to be consumed before the received messages, so that roundtrip times are slower?

by the way even with the multi-threaded approach, gossipsub queue will still prioritize the received messages BEFORE the messages to be published, right?

I am a bit confused with what you mean by this. From the perspective of the Compute Client, the Publish message from the RPC Client is a received (i.e. incoming) message.

I am not super familiar with the gossipsub internals, but generally the swarm prioritizes outgoing messages over incoming ones:
It first poll events from the network behavior and sends them to the connection handler (e.g. outgoing events like gossipsub publish messages) before processing incoming events from the connection handler (See this code). This is done on purpose to prioritize local work over remote work.

So yes the compute client will prioritize sending of the task result (i.e. an outgoing message) before handling the next received publish message, if that's what you mean.

[erhant]

So yes the compute client will prioritize sending of the task result (i.e. an outgoing message) before handling the next received publish message, if that's what you mean.

Ah yes, this was exactly what I sought for actually.

So my problem is probably doing the whole process in a single thread, e.g. handle-event --> process-task --> handle-event --> ... whereas handle-event should be a thread on its own, sending messages via a channel, and process-task is that other thread processing tasks and publishing messages.

[elenaf9]

and process-task is that other thread processing tasks and publishing messages.

Small suggestion: you might still want to do the publishing of messages in the same task that polls the swarm, because otherwise you need to share the swarm in a Mutex.
So basically you p2p client would be doing all the swarm interaction:

• forwarding incoming events to the process-task via channel
• polling a second channel where it receives the validation results from the process-task and calling report_message_validation_result

examples/file-sharing might be a good starting point.

[erhant]

I am not super familiar with the gossipsub internals, but generally the swarm prioritizes outgoing messages over incoming ones:
It first poll events from the network behavior and sends them to the connection handler (e.g. outgoing events like gossipsub publish messages) before processing incoming events from the connection handler (See this code). This is done on purpose to prioritize local work over remote work.

about this, I was looking a bit deeper into the code and I think the problem with gossipsub messages being processed such that publish events take place after receive events still persists.

• gossipsub.publish calls send_message
• send_message puts an event to the events queue, which has type VecDeque<ToSwarm<Event, HandlerIn>>
• from the docs of ToSwarm we see that it is:

A command issued from a [NetworkBehaviour] for the [Swarm].

• so publish messages are also considered a NetworkBehaviour, which would mean that they are queued all the way in the back so again we have to consume all received messages before publish messages (i.e. HandlerIn::Message(RpcOut) events) can be consumed.

nevertheless, looks like we need separate threads so that received messages are handled and another thread handles the message processing that the queue can continue doing its work while task execution takes place

examples/file-sharing might be a good starting point.

thanks for the example, we were looking into mpsc channel as well so this is spot on!

[elenaf9]

Well yes, if an incoming message was already bubbled all the way up to the Behavior before Behavior::publish was called, it will be in in the events queue in front of the publish event.

But this behavior will actually change with #5595. #5595 introduces direct channels from behavior to connection handle for sending outgoing events like publish messages, instead of enqueuing them in the events queue.

[erhant]

I am in the process of rewriting our p2p client to use threading, as in the example above, I will close this discussion with this answer once I confirm it works! 🙏🏻

Thanks again for your help.

[erhant]

Update: having a separate thread DID solve our problems :)