vacp2p/nim-libp2p
1
0
Fork 0
mirror of https://github.com/vacp2p/nim-libp2p.git synced 2025-03-01 08:30:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

create pubsubpeer-queue-metrics

Browse Source
This commit is contained in:
Diego 2024-02-22 19:24:02 +01:00
parent e6be3c0ae6
commit b8c7b347c3
No known key found for this signature in database
GPG Key ID: C9DAC9BF68D1F806
1 changed files with 6 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -31,6 +31,7 @@ when defined(libp2p_expensive_metrics):
declareCounter(libp2p_pubsub_skipped_received_messages, "number of received skipped messages", labels = ["id"])
declareCounter(libp2p_pubsub_skipped_sent_messages, "number of sent skipped messages", labels = ["id"])
when defined(pubsubpeer_queue_metrics):
declareGauge(libp2p_gossipsub_priority_queue_size, "the number of messages in the priority queue", labels = ["id"])
declareGauge(libp2p_gossipsub_non_priority_queue_size, "the number of messages in the non-priority queue", labels = ["id"])
@ -161,7 +162,7 @@ proc stopSendNonPriorityTask*(p: PubSubPeer) =
f.cancel()
p.rpcmessagequeue.sendPriorityQueue.clear()
p.rpcmessagequeue.nonPriorityQueue.clear()
when defined(libp2p_expensive_metrics):
when defined(pubsubpeer_queue_metrics):
libp2p_gossipsub_priority_queue_size.set(labelValues = [$p.peerId], value = 0)
libp2p_gossipsub_non_priority_queue_size.set(labelValues = [$p.peerId], value = 0)
@ -274,7 +275,7 @@ proc clearSendPriorityQueue(p: PubSubPeer) =
discard p.rpcmessagequeue.sendPriorityQueue.popLast()
else:
break
when defined(libp2p_expensive_metrics):
when defined(pubsubpeer_queue_metrics):
libp2p_gossipsub_priority_queue_size.dec(labelValues = [$p.peerId])
proc sendMsg(p: PubSubPeer, msg: seq[byte]) {.async.} =
@ -326,11 +327,11 @@ proc sendEncoded*(p: PubSubPeer, msg: seq[byte], isHighPriority: bool) {.async.}
let f = p.sendMsg(msg)
if not f.finished:
p.rpcmessagequeue.sendPriorityQueue.addLast(f)
when defined(libp2p_expensive_metrics):
when defined(pubsubpeer_queue_metrics):
libp2p_gossipsub_priority_queue_size.inc(labelValues = [$p.peerId])
else:
await p.rpcmessagequeue.nonPriorityQueue.addLast(msg)
when defined(libp2p_expensive_metrics):
when defined(pubsubpeer_queue_metrics):
libp2p_gossipsub_non_priority_queue_size.inc(labelValues = [$p.peerId])
trace "message queued", p, msg = shortLog(msg)
@ -422,7 +423,7 @@ proc sendNonPriorityTask(p: PubSubPeer) {.async.} =
# to be finished already (since sends are processed in order).
if p.rpcmessagequeue.sendPriorityQueue.len > 0:
await p.rpcmessagequeue.sendPriorityQueue[^1]
when defined(libp2p_expensive_metrics):
when defined(pubsubpeer_queue_metrics):
libp2p_gossipsub_non_priority_queue_size.dec(labelValues = [$p.peerId])
await p.sendMsg(msg)