Cleanup blockexchange blockAddress handling

codex/blockexchange/engine/discovery.nim

@@ -58,7 +58,8 @@ type DiscoveryEngine* = ref object of RootObj
 
 proc discoveryQueueLoop(b: DiscoveryEngine) {.async: (raises: []).} =
   while b.discEngineRunning:
-    for cid in toSeq(b.pendingBlocks.wantListBlockCids):
+    for address in toSeq(b.pendingBlocks.wantList):
+      let cid = address.cidOrTreeCid
       try:
         await b.discoveryQueue.put(cid)
       except CancelledError:
@@ -87,9 +88,7 @@ proc discoveryTaskLoop(b: DiscoveryEngine) {.async: (raises: []).} =
         trace "Discovery request already in progress", cid
         continue
 
-      let haves = b.peers.peersHave(cid)
-
-      if haves.len < b.minPeersPerBlock:
+      if b.peers.countPeersWhoHave(cid) < b.minPeersPerBlock:
         try:
           let request = b.discovery.find(cid).wait(DefaultDiscoveryTimeout)
 
@@ -126,6 +125,9 @@ proc queueFindBlocksReq*(b: DiscoveryEngine, cids: seq[Cid]) {.inline.} =
       except CatchableError as exc:
         warn "Exception queueing discovery request", exc = exc.msg
 
+proc queueFindBlocksReq*(b: DiscoveryEngine, addresses: seq[BlockAddress]) {.inline.} =
+  b.queueFindBlocksReq(addresses.mapIt(it.cidOrTreeCid))
+
 proc start*(b: DiscoveryEngine) {.async.} =
   ## Start the discengine task
   ##

codex/blockexchange/engine/engine.nim

@@ -177,7 +177,7 @@
 
     # drop unresponsive peer
     await b.network.switch.disconnect(peerId)
-    b.discovery.queueFindBlocksReq(@[address.cidOrTreeCid])
+    b.discovery.queueFindBlocksReq(@[address])
 
 proc pickPseudoRandom(
     address: BlockAddress, peers: seq[BlockExcPeerCtx]
@@ -193,7 +193,7 @@
     let peers = b.peers.getPeersForBlock(address)
 
     if peers.with.len == 0:
-      b.discovery.queueFindBlocksReq(@[address.cidOrTreeCid])
+      b.discovery.queueFindBlocksReq(@[address])
     else:
       let selected = pickPseudoRandom(address, peers.with)
       asyncSpawn b.monitorBlockHandle(blockFuture, address, selected.id)
@@ -242,25 +242,21 @@
   # if none of the connected peers report our wants in their have list,
   # fire up discovery
   b.discovery.queueFindBlocksReq(
-    toSeq(b.pendingBlocks.wantListCids).filter do(cid: Cid) -> bool:
-      not b.peers.anyIt(cid in it.peerHaveCids)
+    toSeq(b.pendingBlocks.wantList).filterIt(b.peers.peersHave(it).len == 0)
   )
 
 proc scheduleTasks(b: BlockExcEngine, blocksDelivery: seq[BlockDelivery]) {.async.} =
-  let cids = blocksDelivery.mapIt(it.blk.cid)
-
+  let addresses = blocksDelivery.mapIt(it.address)
   # schedule any new peers to provide blocks to
   for p in b.peers:
-    for c in cids: # for each cid
-      # schedule a peer if it wants at least one cid
-      # and we have it in our local store
-      if c in p.peerWantsCids:
-        if await (c in b.localStore):
+    for address in addresses:
+      # schedule a peer if it wants at least one
+      if address in p.peerWants:
+        if await (address in b.localStore):
           if b.scheduleTask(p):
             trace "Task scheduled for peer", peer = p.id
           else:
             warn "Unable to schedule task for peer", peer = p.id
-
           break # do next peer
 
 proc cancelBlocks(b: BlockExcEngine, addrs: seq[BlockAddress]) {.async.} =

codex/blockexchange/engine/pendingblocks.nim

@@ -130,19 +130,6 @@ iterator wantList*(p: PendingBlocksManager): BlockAddress =
   for a in p.blocks.keys:
     yield a
 
-iterator wantListBlockCids*(p: PendingBlocksManager): Cid =
-  for a in p.blocks.keys:
-    if not a.leaf:
-      yield a.cid
-
-iterator wantListCids*(p: PendingBlocksManager): Cid =
-  var yieldedCids = initHashSet[Cid]()
-  for a in p.blocks.keys:
-    let cid = a.cidOrTreeCid
-    if cid notin yieldedCids:
-      yieldedCids.incl(cid)
-      yield cid
-
 iterator wantHandles*(p: PendingBlocksManager): Future[Block] =
   for v in p.blocks.values:
     yield v.handle

codex/blockexchange/peers/peercontext.nim

@@ -37,12 +37,6 @@ type BlockExcPeerCtx* = ref object of RootObj
 proc peerHave*(self: BlockExcPeerCtx): seq[BlockAddress] =
   toSeq(self.blocks.keys)
 
-proc peerHaveCids*(self: BlockExcPeerCtx): HashSet[Cid] =
-  self.blocks.keys.toSeq.mapIt(it.cidOrTreeCid).toHashSet
-
-proc peerWantsCids*(self: BlockExcPeerCtx): HashSet[Cid] =
-  self.peerWants.mapIt(it.address.cidOrTreeCid).toHashSet
-
 proc contains*(self: BlockExcPeerCtx, address: BlockAddress): bool =
   address in self.blocks
 

codex/blockexchange/peers/peerctxstore.nim

@@ -65,15 +65,12 @@ func len*(self: PeerCtxStore): int =
 func peersHave*(self: PeerCtxStore, address: BlockAddress): seq[BlockExcPeerCtx] =
   toSeq(self.peers.values).filterIt(it.peerHave.anyIt(it == address))
 
-func peersHave*(self: PeerCtxStore, cid: Cid): seq[BlockExcPeerCtx] =
-  toSeq(self.peers.values).filterIt(it.peerHave.anyIt(it.cidOrTreeCid == cid))
+func countPeersWhoHave*(self: PeerCtxStore, cid: Cid): int =
+  self.peers.values.countIt(it.peerHave.anyIt(it.cidOrTreeCid == cid))
 
 func peersWant*(self: PeerCtxStore, address: BlockAddress): seq[BlockExcPeerCtx] =
   toSeq(self.peers.values).filterIt(it.peerWants.anyIt(it == address))
 
-func peersWant*(self: PeerCtxStore, cid: Cid): seq[BlockExcPeerCtx] =
-  toSeq(self.peers.values).filterIt(it.peerWants.anyIt(it.address.cidOrTreeCid == cid))
-
 proc getPeersForBlock*(self: PeerCtxStore, address: BlockAddress): PeersForBlock =
   var res = PeersForBlock()
   for peer in self:

tests/codex/blockexchange/testpendingblocks.nim

@@ -63,8 +63,8 @@ checksuite "Pending Blocks":
     discard blks.mapIt(pendingBlocks.getWantHandle(it.cid))
 
     check:
-      blks.mapIt($it.cid).sorted(cmp[string]) ==
-        toSeq(pendingBlocks.wantListBlockCids).mapIt($it).sorted(cmp[string])
+      blks.mapIt($BlockAddress.init(it.cid)).sorted(cmp[string]) ==
+        toSeq(pendingBlocks.wantList).mapIt($it).sorted(cmp[string])
 
   test "Should get want handles list":
     let