Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

BlockExchange improvement 1 #964

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

BlockExchange improvement 1 #964

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d6b0de0
Cleanup changes
benbierens Oct 22, 2024
2a25460
Fixes issue where only wants of type block are stored in peerContext
benbierens Oct 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 17 additions & 19 deletions codex/blockexchange/engine/engine.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,20 +136,23 @@
for p in peers:
if p notin excluded:
if address notin p.peerHave:
trace "Sending wantHave", address, peer = p.id
await b.network.request.sendWantList(
p.id,
@[address],
wantType = WantType.WantHave) # we only want to know if the peer has the block
codex_block_exchange_want_have_lists_sent.inc()

proc sendWantBlock(
b: BlockExcEngine,
address: BlockAddress, # pluralize this entire call chain, please
blockPeer: BlockExcPeerCtx): Future[void] {.async.} =
trace "Sending wantBlock request to", peer = blockPeer.id, address
trace "Sending wantBlock", address, peer = blockPeer.id
await b.network.request.sendWantList(
blockPeer.id,
@[address],
wantType = WantType.WantBlock) # we want this remote to send us a block
codex_block_exchange_want_block_lists_sent.inc()

proc monitorBlockHandle(
b: BlockExcEngine,
Expand Down Expand Up @@ -183,24 +186,17 @@

if not b.pendingBlocks.isInFlight(address):
let peers = b.peers.selectCheapest(address)
if peers.len == 0:
b.discovery.queueFindBlocksReq(@[address.cidOrTreeCid])

let maybePeer =
if peers.len > 0:
peers[hash(address) mod peers.len].some
elif b.peers.len > 0:
toSeq(b.peers)[hash(address) mod b.peers.len].some
else:
BlockExcPeerCtx.none

if peer =? maybePeer:
if peers.len > 0:
let peer = peers[hash(address) mod peers.len]

Check warning on line 190 in codex/blockexchange/engine/engine.nim

View check run for this annotation

Codecov / codecov/patch

codex/blockexchange/engine/engine.nim#L190 

Added line #L190 was not covered by tests
trace "Existing peers found for address.", address, nPeers = peers.len, selectedPeer = peer.id
asyncSpawn b.monitorBlockHandle(blockFuture, address, peer.id)
b.pendingBlocks.setInFlight(address)
await b.sendWantBlock(address, peer)
codex_block_exchange_want_block_lists_sent.inc()
await b.sendWantHave(address, @[peer], toSeq(b.peers))
codex_block_exchange_want_have_lists_sent.inc()
else:

Check warning on line 196 in codex/blockexchange/engine/engine.nim

View check run for this annotation

Codecov / codecov/patch

codex/blockexchange/engine/engine.nim#L196 

Added line #L196 was not covered by tests
trace "No existing peers found for address.", address
b.discovery.queueFindBlocksReq(@[address.cidOrTreeCid])
await b.sendWantHave(address, @[], toSeq(b.peers))

# Don't let timeouts bubble up. We can't be too broad here or we break
# cancellations.
Expand Down Expand Up @@ -405,7 +401,8 @@
address = e.address
wantType = $e.wantType

if idx < 0: # updating entry
if idx < 0: # new entry
peerCtx.peerWants.add(e)
let
have = await e.address in b.localStore
price = @(
Expand All @@ -428,9 +425,8 @@
`type`: BlockPresenceType.Have,
price: price))
elif e.wantType == WantType.WantBlock:
peerCtx.peerWants.add(e)
codex_block_exchange_want_block_lists_received.inc()
else:
else: # update existing entry
# peer doesn't want this block anymore
if e.cancel:
peerCtx.peerWants.del(idx)
Expand Down Expand Up @@ -494,7 +490,9 @@
trace "Sending our want list to a peer", peer
let cids = toSeq(b.pendingBlocks.wantList)
await b.network.request.sendWantList(
peer, cids, full = true)
peer, cids, full = true,
wantType = WantType.WantHave)
codex_block_exchange_want_have_lists_sent.inc()

if address =? b.pricing.?address:
await b.network.request.sendAccount(peer, Account(address: address))
Expand Down
10 changes: 4 additions & 6 deletions codex/blockexchange/peers/peerctxstore.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,19 +71,17 @@ func peersWant*(self: PeerCtxStore, cid: Cid): seq[BlockExcPeerCtx] =
toSeq(self.peers.values).filterIt( it.peerWants.anyIt( it.address.cidOrTreeCid == cid ) )

func selectCheapest*(self: PeerCtxStore, address: BlockAddress): seq[BlockExcPeerCtx] =
# assume that the price for all leaves in a tree is the same
let rootAddress = BlockAddress(leaf: false, cid: address.cidOrTreeCid)
var peers = self.peersHave(rootAddress)
var peers = self.peersHave(address)

func cmp(a, b: BlockExcPeerCtx): int =
var
priceA = 0.u256
priceB = 0.u256

a.blocks.withValue(rootAddress, precense):
a.blocks.withValue(address, precense):
priceA = precense[].price

b.blocks.withValue(rootAddress, precense):
b.blocks.withValue(address, precense):
priceB = precense[].price

if priceA == priceB:
Expand All @@ -94,7 +92,7 @@ func selectCheapest*(self: PeerCtxStore, address: BlockAddress): seq[BlockExcPee
-1

peers.sort(cmp)
trace "Selected cheapest peers", peers = peers.len
trace "Selected cheapest peers", address = address, peers = peers.len
return peers

proc new*(T: type PeerCtxStore): PeerCtxStore =
Expand Down
Loading