merge 1.0

libp2p/protocols/pubsub/gossipsub11.nim

@@ -242,30 +242,31 @@ proc getGossipPeers(g: GossipSub): Table[string, ControlMessage] {.gcsafe.} =
 
 proc heartbeat(g: GossipSub) {.async.} =
   while g.heartbeatRunning:
-    try:
+    withLock g.heartbeatLock:
-      trace "running heartbeat"
+      try:
+        trace "running heartbeat"
 
-      for t in toSeq(g.topics.keys):
+        for t in toSeq(g.topics.keys):
-        await g.rebalanceMesh(t)
+          await g.rebalanceMesh(t)
 
-      await g.dropFanoutPeers()
+        await g.dropFanoutPeers()
 
-      # replenish known topics to the fanout
+        # replenish known topics to the fanout
-      for t in toSeq(g.fanout.keys):
+        for t in toSeq(g.fanout.keys):
-        g.replenishFanout(t)
+          g.replenishFanout(t)
 
-      let peers = g.getGossipPeers()
+        let peers = g.getGossipPeers()
-      var sent: seq[Future[void]]
+        var sent: seq[Future[void]]
-      for peer in peers.keys:
+        for peer in peers.keys:
-        if peer in g.peers:
+          if peer in g.peers:
-          sent &= g.peers[peer].send(@[RPCMsg(control: some(peers[peer]))])
+            sent &= g.peers[peer].send(@[RPCMsg(control: some(peers[peer]))])
-      checkFutures(await allFinished(sent))
+        checkFutures(await allFinished(sent))
 
-      g.mcache.shift() # shift the cache
+        g.mcache.shift() # shift the cache
-    except CancelledError as exc:
+      except CancelledError as exc:
-      raise exc
+        raise exc
-    except CatchableError as exc:
+      except CatchableError as exc:
-      trace "exception ocurred in gossipsub heartbeat", exc = exc.msg
+        trace "exception ocurred in gossipsub heartbeat", exc = exc.msg
 
     await sleepAsync(1.seconds)
 
@@ -275,29 +276,30 @@ method handleDisconnect*(g: GossipSub, peer: PubSubPeer) {.async.} =
 
   await procCall FloodSub(g).handleDisconnect(peer)
 
-  for t in toSeq(g.gossipsub.keys):
+  withLock g.heartbeatLock:
-    if t in g.gossipsub:
+    for t in toSeq(g.gossipsub.keys):
-      g.gossipsub[t].excl(peer.id)
+      if t in g.gossipsub:
+        g.gossipsub[t].excl(peer.id)
 
-    libp2p_gossipsub_peers_per_topic_gossipsub
+      libp2p_gossipsub_peers_per_topic_gossipsub
-      .set(g.gossipsub.getOrDefault(t).len.int64, labelValues = [t])
+        .set(g.gossipsub.getOrDefault(t).len.int64, labelValues = [t])
 
-    # mostly for metrics
+      # mostly for metrics
-    await procCall PubSub(g).subscribeTopic(t, false, peer.id)
+      await procCall PubSub(g).subscribeTopic(t, false, peer.id)
 
-  for t in toSeq(g.mesh.keys):
+    for t in toSeq(g.mesh.keys):
-    if t in g.mesh:
+      if t in g.mesh:
-      g.mesh[t].excl(peer.id)
+        g.mesh[t].excl(peer.id)
 
-    libp2p_gossipsub_peers_per_topic_mesh
+      libp2p_gossipsub_peers_per_topic_mesh
-      .set(g.mesh[t].len.int64, labelValues = [t])
+        .set(g.mesh[t].len.int64, labelValues = [t])
 
-  for t in toSeq(g.fanout.keys):
+    for t in toSeq(g.fanout.keys):
-    if t in g.fanout:
+      if t in g.fanout:
-      g.fanout[t].excl(peer.id)
+        g.fanout[t].excl(peer.id)
 
-    libp2p_gossipsub_peers_per_topic_fanout
+      libp2p_gossipsub_peers_per_topic_fanout
-      .set(g.fanout[t].len.int64, labelValues = [t])
+        .set(g.fanout[t].len.int64, labelValues = [t])
 
 method subscribeToPeer*(p: GossipSub,
                         conn: Connection) {.async.} =
@@ -310,26 +312,27 @@ method subscribeTopic*(g: GossipSub,
                        peerId: string) {.gcsafe, async.} =
   await procCall PubSub(g).subscribeTopic(topic, subscribe, peerId)
 
-  if topic notin g.gossipsub:
+  withLock g.heartbeatLock:
-    g.gossipsub[topic] = initHashSet[string]()
+    if topic notin g.gossipsub:
+      g.gossipsub[topic] = initHashSet[string]()
 
-  if subscribe:
+    if subscribe:
-    trace "adding subscription for topic", peer = peerId, name = topic
+      trace "adding subscription for topic", peer = peerId, name = topic
-    # subscribe remote peer to the topic
+      # subscribe remote peer to the topic
-    g.gossipsub[topic].incl(peerId)
+      g.gossipsub[topic].incl(peerId)
-    if peerId in g.explicit:
+      if peerId in g.explicit:
-      g.explicit[topic].incl(peerId)
+        g.explicit[topic].incl(peerId)
-  else:
+    else:
-    trace "removing subscription for topic", peer = peerId, name = topic
+      trace "removing subscription for topic", peer = peerId, name = topic
-    # unsubscribe remote peer from the topic
+      # unsubscribe remote peer from the topic
-    g.gossipsub[topic].excl(peerId)
+      g.gossipsub[topic].excl(peerId)
-    if peerId in g.explicit:
+      if peerId in g.explicit:
-      g.explicit[topic].excl(peerId)
+        g.explicit[topic].excl(peerId)
 
-  libp2p_gossipsub_peers_per_topic_gossipsub
+    libp2p_gossipsub_peers_per_topic_gossipsub
-    .set(g.gossipsub[topic].len.int64, labelValues = [topic])
+      .set(g.gossipsub[topic].len.int64, labelValues = [topic])
 
-  trace "gossip peers", peers = g.gossipsub[topic].len, topic
+    trace "gossip peers", peers = g.gossipsub[topic].len, topic
 
   # also rebalance current topic if we are subbed to
   if topic in g.topics:
@@ -555,14 +558,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)
@@ -580,14 +580,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"
@@ -595,15 +591,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()