make relayed messages non priority

[diegomrsantos]

This PR is still in an exploratory phase and the base idea is to give a lower priority to the messages received from other peers that need to be relayed than the messages published by the peer.

[codecov]

Codecov Report

Attention: 3 lines in your changes are missing coverage. Please review.

Comparison is base (5594bcb) 83.13% compared to head (9eb9cca) 83.22%.

Additional details and impacted files

@@             Coverage Diff              @@
##           unstable    #1009      +/-   ##
============================================
+ Coverage     83.13%   83.22%   +0.09%     
============================================
  Files            91       91              
  Lines         15344    15419      +75     
============================================
+ Hits          12756    12833      +77     
+ Misses         2588     2586       -2     

Files	Coverage Δ
libp2p/errors.nim	84.21% <100.00%> (ø)
libp2p/protocols/pubsub/gossipsub.nim	88.16% <100.00%> (+0.08%)	⬆️
libp2p/protocols/pubsub/pubsub.nim	85.44% <100.00%> (+1.53%)	⬆️
libp2p/protocols/pubsub/pubsubpeer.nim	93.93% <95.38%> (+1.43%)	⬆️

... and 1 file with indirect coverage changes

[arnetheduck]

in general, the queue for priority messages is not needed - one can simply send the message and put its future on a deque to achieve the same effect.

For non-priority messages, these are put on a queue like in this PR. The "worker loop" for non-priority messages then becomes:

sendQueue: var Deque[Future[void]]

proc clearSendQueue() =
  while sendQueue.len > 0 and senQueue[0].finished:
    discard sendQueue.popFirst()


proc addSendQueue(msg: Future[void]) = 
  clearSendQueue()
  if not msg.finished: sendQueue.addLast(msg)

proc send(msg): Future[void] = ...

while true:
   let msg = await nonPrioQueue.popFirst()
   while sendQueue.len > 0:
     await sendQueue[^1]
     clearSendQueue()
   addSendQueue(send(msg))

Among other things, this avoids latency for publishing prio message as well as the additional memory traffic resulting from message copies.

I'm also not sure what messageAvailable is doing here - await nonPriorityQueue.popFirst() already wakes up whenever there's a message.

[diegomrsantos]

I think the priority queue is for the case when we're trying to publish messages faster than we can send them. In this case we'll need to either drop or slow down somehow. How are we going to achieve the same without an explicit queue?

[diegomrsantos]

this avoids latency for publishing prio message

Could you please elaborate more on how the current code causes this latency? Currently, we don't send a non-priority message as long as there's a priority message to be sent.

[arnetheduck]

I think the priority queue is for the case when we're trying to publish messages faster than we can send them. In this case we'll need to either drop or slow down somehow. How are we going to achieve the same without an explicit queue?

There is still a queue, but it is implicit: it exists in the form of pending futures - similar to how it happens now - asyncSpawn puts things in a hidden implicit queue

Pending futures can be cancelled - ie the deque I suggest above does exactly that.

[arnetheduck]

Could you please elaborate more on how the current code causes this latency? Currently, we don't send a non-priority message as long as there's a priority message to be sent.

In this PR, you copy the message into priorityQueue which wakes up processMessages which removes the message from the priorityQueue and adds it to the (hidden) send queue of the connection - the time it takes to wake up processMessages is added latency (and additional copies of the message) - it is at least one iteration of the event poll loop which can take several milliseconds (or worse, if there's a lot of work).

[diegomrsantos]

ok, thanks for the explanation. How will the new proposed solution prioritize the published msgs over the relayed ones? Will we check the sendQueue for priority messages and send non-priority ones only when this queue is empty?

[arnetheduck]

yes, that's how it could work - not added in my snippet is that priority messages also get tracked via sendQueue

[diegomrsantos]

This has been replaced by #1015