with sugar

libp2p/protocols/pubsub/gossipsub.nim

@@ -189,34 +189,31 @@ proc getGossipPeers(g: GossipSub): Table[string, ControlMessage] {.gcsafe.} =
 
 proc heartbeat(g: GossipSub) {.async.} =
   while g.heartbeatRunning:
-    await g.heartbeatLock.acquire()
+    withLock g.heartbeatLock:
-    try:
+      try:
+        trace "running heartbeat"
 
-      trace "running heartbeat"
+        for t in toSeq(g.topics.keys):
+          await g.rebalanceMesh(t)
 
-      for t in toSeq(g.topics.keys):
+        await g.dropFanoutPeers()
-        await g.rebalanceMesh(t)
 
-      await g.dropFanoutPeers()
+        # replenish known topics to the fanout
+        for t in toSeq(g.fanout.keys):
+          g.replenishFanout(t)
 
-      # replenish known topics to the fanout
+        let peers = g.getGossipPeers()
-      for t in toSeq(g.fanout.keys):
+        var sent: seq[Future[void]]
-        g.replenishFanout(t)
+        for peer in peers.keys:
+          if peer in g.peers:
+            sent &= g.peers[peer].send(@[RPCMsg(control: some(peers[peer]))])
+        checkFutures(await allFinished(sent))
 
-      let peers = g.getGossipPeers()
+        g.mcache.shift() # shift the cache
-      var sent: seq[Future[void]]
+      except CancelledError as exc:
-      for peer in peers.keys:
+        raise exc
-        if peer in g.peers:
+      except CatchableError as exc:
-          sent &= g.peers[peer].send(@[RPCMsg(control: some(peers[peer]))])
+        trace "exception ocurred in gossipsub heartbeat", exc = exc.msg
-      checkFutures(await allFinished(sent))
-
-      g.mcache.shift() # shift the cache
-    except CancelledError as exc:
-      raise exc
-    except CatchableError as exc:
-      trace "exception ocurred in gossipsub heartbeat", exc = exc.msg
-    finally:
-      g.heartbeatLock.release()
 
     await sleepAsync(1.seconds)
 
@@ -227,8 +224,7 @@ method handleDisconnect*(g: GossipSub, peer: PubSubPeer) {.async.} =
   await procCall FloodSub(g).handleDisconnect(peer)
 
   # must avoid running this while manipulating mesh/gossip tables
-  await g.heartbeatLock.acquire()
+  withLock g.heartbeatLock:
-  try:
     for t in toSeq(g.gossipsub.keys):
       g.gossipsub[t].excl(peer.id)
 
@@ -249,8 +245,6 @@ method handleDisconnect*(g: GossipSub, peer: PubSubPeer) {.async.} =
 
       libp2p_gossipsub_peers_per_topic_fanout
         .set(g.fanout[t].len.int64, labelValues = [t])
-  finally:
-    g.heartbeatLock.release()
 
 method subscribeToPeer*(p: GossipSub,
                         conn: Connection) {.async.} =
@@ -264,8 +258,7 @@ method subscribeTopic*(g: GossipSub,
   await procCall PubSub(g).subscribeTopic(topic, subscribe, peerId)
   
   # must avoid running this while manipulating mesh/gossip tables
-  await g.heartbeatLock.acquire()
+  withLock g.heartbeatLock:
-  try:
     if topic notin g.gossipsub:
       g.gossipsub[topic] = initHashSet[string]()
 
@@ -282,8 +275,6 @@ method subscribeTopic*(g: GossipSub,
       .set(g.gossipsub[topic].len.int64, labelValues = [topic])
 
     trace "gossip peers", peers = g.gossipsub[topic].len, topic
-  finally:
-    g.heartbeatLock.release()
 
 proc handleGraft(g: GossipSub,
                  peer: PubSubPeer,
@@ -486,14 +477,11 @@ method publish*(g: GossipSub,
         trace "publish: sending message to peer", peer = p
         sent.add(peer.send(@[RPCMsg(messages: @[msg])]))
       else:
-        # Notice this needs a better fix! for now it's a hack
+        # this absolutely should not happen
-        error "publish: peer or peerInfo was nil", missing = p
+        # if it happens there is a bug that needs fixing asap
-        if topic in g.mesh:
+        # this ain't no place to manage connections
-          g.mesh[topic].excl(p)
+        fatal "publish: peer or peerInfo was nil", missing = p
-        if topic in g.fanout:
+        doAssert(false, "publish: peer or peerInfo was nil")
-          g.fanout[topic].excl(p)
-        if topic in g.gossipsub:
-          g.gossipsub[topic].excl(p)
     
     sent = await allFinished(sent)
     checkFutures(sent)
@@ -511,14 +499,10 @@ method start*(g: GossipSub) {.async.} =
   ## start pubsub
   ## start long running/repeating procedures
 
-  # interlock start to to avoid overlapping to stops
+  withLock g.heartbeatLock:
-  await g.heartbeatLock.acquire()
+    # setup the heartbeat interval
-
+    g.heartbeatRunning = true
-  # setup the heartbeat interval
+    g.heartbeatFut = g.heartbeat()
-  g.heartbeatRunning = true
-  g.heartbeatFut = g.heartbeat()
-
-  g.heartbeatLock.release()
 
 method stop*(g: GossipSub) {.async.} =
   trace "gossipsub stop"
@@ -526,15 +510,12 @@ method stop*(g: GossipSub) {.async.} =
   ## stop pubsub
   ## stop long running tasks
 
-  await g.heartbeatLock.acquire()
+  withLock g.heartbeatLock:
-
+    # stop heartbeat interval
-  # stop heartbeat interval
+    g.heartbeatRunning = false
-  g.heartbeatRunning = false
+    if not g.heartbeatFut.finished:
-  if not g.heartbeatFut.finished:
+      trace "awaiting last heartbeat"
-    trace "awaiting last heartbeat"
+      await g.heartbeatFut
-    await g.heartbeatFut
-
-  g.heartbeatLock.release()
 
 method initPubSub*(g: GossipSub) =
   procCall FloodSub(g).initPubSub()

libp2p/utility.nim

@@ -35,3 +35,10 @@ func shortLog*(item: string): string =
     result &= item[0..<split]
     result &= "..."
     result &= item[(item.len - split)..item.high]
+
+template withLock*(l: auto, body: untyped): untyped =
+  await l.acquire()
+  try:
+    body
+  finally:
+    l.release()