status-im
/
nim-libp2p
mirror of https://github.com/status-im/nim-libp2p.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

clear prio queue before adding to it

This commit is contained in:
Diego 2024-02-09 16:15:39 +01:00
parent df0b98bfdd
commit b16ec00327
No known key found for this signature in database
GPG Key ID: C9DAC9BF68D1F806
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
libp2p/protocols/pubsub/pubsubpeer.nim
View File

@ -250,6 +250,12 @@ template sendMetrics(msg: RPCMsg): untyped =
# metrics # metrics
libp2p_pubsub_sent_messages.inc(labelValues = [$p.peerId, t]) libp2p_pubsub_sent_messages.inc(labelValues = [$p.peerId, t])
proc clearSendPriorityQueue(p: PubSubPeer) =
while p.rpcmessagequeue.sendPriorityQueue.len > 0 and p.rpcmessagequeue.sendPriorityQueue[0].finished:
when defined(libp2p_expensive_metrics):
libp2p_gossipsub_priority_queue_size.dec(labelValues = [$p.peerId])
discard p.rpcmessagequeue.sendPriorityQueue.popFirst()
proc sendMsg(p: PubSubPeer, msg: seq[byte]) {.async.} = proc sendMsg(p: PubSubPeer, msg: seq[byte]) {.async.} =
if p.sendConn == nil: if p.sendConn == nil:
# Wait for a send conn to be setup. `connectOnce` will # Wait for a send conn to be setup. `connectOnce` will
@ -287,6 +293,7 @@ proc sendEncoded*(p: PubSubPeer, msg: seq[byte], isHighPriority: bool = false) {
return return
if isHighPriority: if isHighPriority:
p.clearSendPriorityQueue()
let f = p.sendMsg(msg) let f = p.sendMsg(msg)
if not f.finished: if not f.finished:
p.rpcmessagequeue.sendPriorityQueue.addLast(f) p.rpcmessagequeue.sendPriorityQueue.addLast(f)
@ -366,12 +373,6 @@ proc canAskIWant*(p: PubSubPeer, msgId: MessageId): bool =
return true return true
return false return false
proc clearSendPriorityQueue(p: PubSubPeer) =
while p.rpcmessagequeue.sendPriorityQueue.len > 0 and p.rpcmessagequeue.sendPriorityQueue[0].finished:
when defined(libp2p_expensive_metrics):
libp2p_gossipsub_priority_queue_size.dec(labelValues = [$p.peerId])
discard p.rpcmessagequeue.sendPriorityQueue.popFirst()
proc sendNonPriorityTask(p: PubSubPeer) {.async.} = proc sendNonPriorityTask(p: PubSubPeer) {.async.} =
while true: while true:
# we send non-priority messages only if there are no pending priority messages # we send non-priority messages only if there are no pending priority messages