feat: message prioritization with immediate peer-published dispatch and queuing for other msgs (#1015)

Author: diegomrsantos

libp2p/protocols/pubsub/floodsub.nim

@@ -157,7 +157,7 @@ method rpcHandler*(f: FloodSub,
 
     # In theory, if topics are the same in all messages, we could batch - we'd
     # also have to be careful to only include validated messages
-    f.broadcast(toSendPeers, RPCMsg(messages: @[msg]))
+    f.broadcast(toSendPeers, RPCMsg(messages: @[msg]), isHighPriority = false)
     trace "Forwared message to peers", peers = toSendPeers.len
 
   f.updateMetrics(rpcMsg)
@@ -219,7 +219,7 @@ method publish*(f: FloodSub,
     return 0
 
   # Try to send to all peers that are known to be interested
-  f.broadcast(peers, RPCMsg(messages: @[msg]))
+  f.broadcast(peers, RPCMsg(messages: @[msg]), isHighPriority = true)
 
   when defined(libp2p_expensive_metrics):
     libp2p_pubsub_messages_published.inc(labelValues = [topic])

libp2p/protocols/pubsub/gossipsub.nim

@@ -220,6 +220,8 @@ method unsubscribePeer*(g: GossipSub, peer: PeerId) =
     for topic, info in stats[].topicInfos.mpairs:
       info.firstMessageDeliveries = 0
 
+  pubSubPeer.stopSendNonPriorityTask()
+
   procCall FloodSub(g).unsubscribePeer(peer)
 
 proc handleSubscribe*(g: GossipSub,
@@ -279,31 +281,40 @@ proc handleControl(g: GossipSub, peer: PubSubPeer, control: ControlMessage) =
   respControl.prune.add(g.handleGraft(peer, control.graft))
   let messages = g.handleIWant(peer, control.iwant)
 
-  if
-    respControl.prune.len > 0 or
-    respControl.iwant.len > 0 or
-    messages.len > 0:
-    # iwant and prunes from here, also messages
+  let
+    isPruneNotEmpty = respControl.prune.len > 0
+    isIWantNotEmpty = respControl.iwant.len > 0
+
+  if isPruneNotEmpty or isIWantNotEmpty:
+
+    if isIWantNotEmpty:
+      libp2p_pubsub_broadcast_iwant.inc(respControl.iwant.len.int64)
+
+    if isPruneNotEmpty:
+      for prune in respControl.prune:
+        if g.knownTopics.contains(prune.topicId):
+          libp2p_pubsub_broadcast_prune.inc(labelValues = [prune.topicId])
+        else:
+          libp2p_pubsub_broadcast_prune.inc(labelValues = ["generic"])
+
+    trace "sending control message", msg = shortLog(respControl), peer
+    g.send(
+      peer,
+      RPCMsg(control: some(respControl)), isHighPriority = true)
 
+  if messages.len > 0:
     for smsg in messages:
       for topic in smsg.topicIds:
         if g.knownTopics.contains(topic):
           libp2p_pubsub_broadcast_messages.inc(labelValues = [topic])
         else:
           libp2p_pubsub_broadcast_messages.inc(labelValues = ["generic"])
 
-    libp2p_pubsub_broadcast_iwant.inc(respControl.iwant.len.int64)
-
-    for prune in respControl.prune:
-      if g.knownTopics.contains(prune.topicId):
-        libp2p_pubsub_broadcast_prune.inc(labelValues = [prune.topicId])
-      else:
-        libp2p_pubsub_broadcast_prune.inc(labelValues = ["generic"])
-
-    trace "sending control message", msg = shortLog(respControl), peer
+    # iwant replies have lower priority
+    trace "sending iwant reply messages", peer
     g.send(
       peer,
-      RPCMsg(control: some(respControl), messages: messages))
+      RPCMsg(messages: messages), isHighPriority = false)
 
 proc validateAndRelay(g: GossipSub,
                       msg: Message,
@@ -356,7 +367,7 @@ proc validateAndRelay(g: GossipSub,
     if msg.data.len > msgId.len * 10:
       g.broadcast(toSendPeers, RPCMsg(control: some(ControlMessage(
           idontwant: @[ControlIWant(messageIds: @[msgId])]
-        ))))
+        ))), isHighPriority = true)
 
     for peer in toSendPeers:
       for heDontWant in peer.heDontWants:
@@ -370,7 +381,7 @@ proc validateAndRelay(g: GossipSub,
 
     # In theory, if topics are the same in all messages, we could batch - we'd
     # also have to be careful to only include validated messages
-    g.broadcast(toSendPeers, RPCMsg(messages: @[msg]))
+    g.broadcast(toSendPeers, RPCMsg(messages: @[msg]), isHighPriority = false)
     trace "forwarded message to peers", peers = toSendPeers.len, msgId, peer
     for topic in msg.topicIds:
       if topic notin g.topics: continue
@@ -441,7 +452,7 @@ method rpcHandler*(g: GossipSub,
   peer.recvObservers(rpcMsg)
 
   if rpcMsg.ping.len in 1..<64 and peer.pingBudget > 0:
-    g.send(peer, RPCMsg(pong: rpcMsg.ping))
+    g.send(peer, RPCMsg(pong: rpcMsg.ping), isHighPriority = true)
     peer.pingBudget.dec
   for i in 0..<min(g.topicsHigh, rpcMsg.subscriptions.len):
     template sub: untyped = rpcMsg.subscriptions[i]
@@ -551,7 +562,7 @@ method onTopicSubscription*(g: GossipSub, topic: string, subscribed: bool) =
             topicID: topic,
             peers: g.peerExchangeList(topic),
             backoff: g.parameters.unsubscribeBackoff.seconds.uint64)])))
-    g.broadcast(mpeers, msg)
+    g.broadcast(mpeers, msg, isHighPriority = true)
 
     for peer in mpeers:
       g.pruned(peer, topic, backoff = some(g.parameters.unsubscribeBackoff))
@@ -655,7 +666,7 @@ method publish*(g: GossipSub,
 
   g.mcache.put(msgId, msg)
 
-  g.broadcast(peers, RPCMsg(messages: @[msg]))
+  g.broadcast(peers, RPCMsg(messages: @[msg]), isHighPriority = true)
 
   if g.knownTopics.contains(topic):
     libp2p_pubsub_messages_published.inc(peers.len.int64, labelValues = [topic])

libp2p/protocols/pubsub/gossipsub/behavior.nim

@@ -530,14 +530,14 @@ proc rebalanceMesh*(g: GossipSub, topic: string, metrics: ptr MeshMetrics = nil)
   # Send changes to peers after table updates to avoid stale state
   if grafts.len > 0:
     let graft = RPCMsg(control: some(ControlMessage(graft: @[ControlGraft(topicID: topic)])))
-    g.broadcast(grafts, graft)
+    g.broadcast(grafts, graft, isHighPriority = true)
   if prunes.len > 0:
     let prune = RPCMsg(control: some(ControlMessage(
       prune: @[ControlPrune(
         topicID: topic,
         peers: g.peerExchangeList(topic),
         backoff: g.parameters.pruneBackoff.seconds.uint64)])))
-    g.broadcast(prunes, prune)
+    g.broadcast(prunes, prune, isHighPriority = true)
 
 proc dropFanoutPeers*(g: GossipSub) {.raises: [].} =
   # drop peers that we haven't published to in
@@ -669,7 +669,7 @@ proc onHeartbeat(g: GossipSub) {.raises: [].} =
             topicID: t,
             peers: g.peerExchangeList(t),
             backoff: g.parameters.pruneBackoff.seconds.uint64)])))
-        g.broadcast(prunes, prune)
+        g.broadcast(prunes, prune, isHighPriority = true)
 
       # pass by ptr in order to both signal we want to update metrics
       # and as well update the struct for each topic during this iteration
@@ -691,7 +691,7 @@ proc onHeartbeat(g: GossipSub) {.raises: [].} =
           libp2p_pubsub_broadcast_ihave.inc(labelValues = [ihave.topicId])
         else:
           libp2p_pubsub_broadcast_ihave.inc(labelValues = ["generic"])
-      g.send(peer, RPCMsg(control: some(control)))
+      g.send(peer, RPCMsg(control: some(control)), isHighPriority = true)
 
     g.mcache.shift() # shift the cache
 

libp2p/protocols/pubsub/pubsub.nim

@@ -138,18 +138,34 @@ method unsubscribePeer*(p: PubSub, peerId: PeerId) {.base, gcsafe.} =
 
   libp2p_pubsub_peers.set(p.peers.len.int64)
 
-proc send*(p: PubSub, peer: PubSubPeer, msg: RPCMsg) {.raises: [].} =
-  ## Attempt to send `msg` to remote peer
+proc send*(p: PubSub, peer: PubSubPeer, msg: RPCMsg, isHighPriority: bool) {.raises: [].} =
+  ## This procedure attempts to send a `msg` (of type `RPCMsg`) to the specified remote peer in the PubSub network.
   ##
+  ## Parameters:
+  ## - `p`: The `PubSub` instance.
+  ## - `peer`: An instance of `PubSubPeer` representing the peer to whom the message should be sent.
+  ## - `msg`: The `RPCMsg` instance that contains the message to be sent.
+  ## - `isHighPriority`: A boolean indicating whether the message should be treated as high priority.
+  ## High priority messages are sent immediately, while low priority messages are queued and sent only after all high
+  ## priority messages have been sent.
 
   trace "sending pubsub message to peer", peer, msg = shortLog(msg)
-  peer.send(msg, p.anonymize)
+  asyncSpawn peer.send(msg, p.anonymize, isHighPriority)
 
 proc broadcast*(
   p: PubSub,
   sendPeers: auto, # Iteratble[PubSubPeer]
-  msg: RPCMsg) {.raises: [].} =
-  ## Attempt to send `msg` to the given peers
+  msg: RPCMsg,
+  isHighPriority: bool) {.raises: [].} =
+  ## This procedure attempts to send a `msg` (of type `RPCMsg`) to a specified group of peers in the PubSub network.
+  ##
+  ## Parameters:
+  ## - `p`: The `PubSub` instance.
+  ## - `sendPeers`: An iterable of `PubSubPeer` instances representing the peers to whom the message should be sent.
+  ## - `msg`: The `RPCMsg` instance that contains the message to be broadcast.
+  ## - `isHighPriority`: A boolean indicating whether the message should be treated as high priority.
+  ## High priority messages are sent immediately, while low priority messages are queued and sent only after all high
+  ## priority messages have been sent.
 
   let npeers = sendPeers.len.int64
   for sub in msg.subscriptions:
@@ -195,19 +211,19 @@ proc broadcast*(
 
   if anyIt(sendPeers, it.hasObservers):
     for peer in sendPeers:
-      p.send(peer, msg)
+      p.send(peer, msg, isHighPriority)
   else:
     # Fast path that only encodes message once
     let encoded = encodeRpcMsg(msg, p.anonymize)
     for peer in sendPeers:
-      asyncSpawn peer.sendEncoded(encoded)
+      asyncSpawn peer.sendEncoded(encoded, isHighPriority)
 
 proc sendSubs*(p: PubSub,
                peer: PubSubPeer,
                topics: openArray[string],
                subscribe: bool) =
   ## send subscriptions to remote peer
-  p.send(peer, RPCMsg.withSubs(topics, subscribe))
+  p.send(peer, RPCMsg.withSubs(topics, subscribe), isHighPriority = true)
 
   for topic in topics:
     if subscribe: