Exchange pricing when connecting to new peer

This commit is contained in:
Mark Spanbroek 2021-04-08 14:27:49 +02:00 committed by markspanbroek
parent 0897e615e6
commit 3c4d4dc29f
5 changed files with 49 additions and 0 deletions

View File

@ -174,10 +174,14 @@ proc new*(
blocks: seq[bt.Block]) {.gcsafe.} = blocks: seq[bt.Block]) {.gcsafe.} =
engine.blocksHandler(peer, blocks) engine.blocksHandler(peer, blocks)
proc pricingHandler(peer: PeerId, pricing: Pricing) =
engine.pricingHandler(peer, pricing)
network.handlers = BitswapHandlers( network.handlers = BitswapHandlers(
onWantList: blockWantListHandler, onWantList: blockWantListHandler,
onBlocks: blocksHandler, onBlocks: blocksHandler,
onPresence: blockPresenceHandler, onPresence: blockPresenceHandler,
onPricing: pricingHandler
) )
return b return b

View File

@ -42,6 +42,7 @@ type
bytesRecv*: int # bytes received from remote bytesRecv*: int # bytes received from remote
exchanged*: int # times peer has exchanged with us exchanged*: int # times peer has exchanged with us
lastExchange*: Moment # last time peer has exchanged with us lastExchange*: Moment # last time peer has exchanged with us
pricing*: ?Pricing # optional bandwidth price for this peer
BitswapEngine* = ref object of RootObj BitswapEngine* = ref object of RootObj
localStore*: BlockStore # where we localStore blocks for this instance localStore*: BlockStore # where we localStore blocks for this instance
@ -51,6 +52,7 @@ type
peersPerRequest: int # max number of peers to request from peersPerRequest: int # max number of peers to request from
scheduleTask*: TaskScheduler # schedule a new task with the task runner scheduleTask*: TaskScheduler # schedule a new task with the task runner
request*: BitswapRequest # bitswap network requests request*: BitswapRequest # bitswap network requests
pricing*: ?Pricing # optional bandwidth pricing
proc contains*(a: AsyncHeapQueue[Entry], b: Cid): bool = proc contains*(a: AsyncHeapQueue[Entry], b: Cid): bool =
## Convenience method to check for entry prepense ## Convenience method to check for entry prepense
@ -259,6 +261,13 @@ proc wantListHandler*(
if not b.scheduleTask(peerCtx): if not b.scheduleTask(peerCtx):
trace "Unable to schedule task for peer", peer trace "Unable to schedule task for peer", peer
proc pricingHandler*(engine: BitswapEngine, peer: PeerID, pricing: Pricing) =
let context = engine.getPeerCtx(peer)
if context.isNil:
return
context.pricing = pricing.some
proc setupPeer*(b: BitswapEngine, peer: PeerID) = proc setupPeer*(b: BitswapEngine, peer: PeerID) =
## Perform initial setup, such as want ## Perform initial setup, such as want
## list exchange ## list exchange
@ -274,6 +283,9 @@ proc setupPeer*(b: BitswapEngine, peer: PeerID) =
if b.wantList.len > 0: if b.wantList.len > 0:
b.request.sendWantList(peer, b.wantList, full = true) b.request.sendWantList(peer, b.wantList, full = true)
if b.pricing.isSome:
b.request.sendPricing(peer, b.pricing.get)
proc dropPeer*(b: BitswapEngine, peer: PeerID) = proc dropPeer*(b: BitswapEngine, peer: PeerID) =
## Cleanup disconnected peer ## Cleanup disconnected peer
## ##

View File

@ -54,11 +54,13 @@ type
BlocksBroadcaster* = proc(peer: PeerID, presence: seq[bt.Block]) {.gcsafe.} BlocksBroadcaster* = proc(peer: PeerID, presence: seq[bt.Block]) {.gcsafe.}
PresenceBroadcaster* = proc(peer: PeerID, presence: seq[BlockPresence]) {.gcsafe.} PresenceBroadcaster* = proc(peer: PeerID, presence: seq[BlockPresence]) {.gcsafe.}
PricingBroadcaster* = proc(peer: PeerID, pricing: Pricing) {.gcsafe.}
BitswapRequest* = object BitswapRequest* = object
sendWantList*: WantListBroadcaster sendWantList*: WantListBroadcaster
sendBlocks*: BlocksBroadcaster sendBlocks*: BlocksBroadcaster
sendPresence*: PresenceBroadcaster sendPresence*: PresenceBroadcaster
sendPricing*: PricingBroadcaster
BitswapNetwork* = ref object of LPProtocol BitswapNetwork* = ref object of LPProtocol
peers*: Table[PeerID, NetworkPeer] peers*: Table[PeerID, NetworkPeer]
@ -348,10 +350,14 @@ proc new*(
proc sendPresence(id: PeerID, presence: seq[BlockPresence]) {.gcsafe.} = proc sendPresence(id: PeerID, presence: seq[BlockPresence]) {.gcsafe.} =
b.broadcastBlockPresence(id, presence) b.broadcastBlockPresence(id, presence)
proc sendPricing(id: PeerID, pricing: Pricing) =
b.broadcastPricing(id, pricing)
b.request = BitswapRequest( b.request = BitswapRequest(
sendWantList: sendWantList, sendWantList: sendWantList,
sendBlocks: sendBlocks, sendBlocks: sendBlocks,
sendPresence: sendPresence, sendPresence: sendPresence,
sendPricing: sendPricing
) )
b.init() b.init()

View File

@ -16,6 +16,7 @@ import pkg/dagger/utils/asyncheapqueue
import ./utils import ./utils
import ../helpers import ../helpers
import ../examples
suite "Bitswap engine - 2 nodes": suite "Bitswap engine - 2 nodes":
@ -28,6 +29,7 @@ suite "Bitswap engine - 2 nodes":
var var
switch1, switch2: Switch switch1, switch2: Switch
wallet1, wallet2: Wallet wallet1, wallet2: Wallet
pricing1, pricing2: Pricing
network1, network2: BitswapNetwork network1, network2: BitswapNetwork
bitswap1, bitswap2: Bitswap bitswap1, bitswap2: Bitswap
awaiters: seq[Future[void]] awaiters: seq[Future[void]]
@ -42,6 +44,8 @@ suite "Bitswap engine - 2 nodes":
switch2 = newStandardSwitch() switch2 = newStandardSwitch()
wallet1 = Wallet.init(EthPrivateKey.random()) wallet1 = Wallet.init(EthPrivateKey.random())
wallet2 = Wallet.init(EthPrivateKey.random()) wallet2 = Wallet.init(EthPrivateKey.random())
pricing1 = Pricing.example
pricing2 = Pricing.example
awaiters.add(await switch1.start()) awaiters.add(await switch1.start())
awaiters.add(await switch2.start()) awaiters.add(await switch2.start())
@ -65,6 +69,9 @@ suite "Bitswap engine - 2 nodes":
bitswap1.engine.wantList = blocks2.mapIt( it.cid ) bitswap1.engine.wantList = blocks2.mapIt( it.cid )
bitswap2.engine.wantList = blocks1.mapIt( it.cid ) bitswap2.engine.wantList = blocks1.mapIt( it.cid )
bitswap1.engine.pricing = pricing1.some
bitswap2.engine.pricing = pricing2.some
await switch1.connect( await switch1.connect(
switch2.peerInfo.peerId, switch2.peerInfo.peerId,
switch2.peerInfo.addrs) switch2.peerInfo.addrs)
@ -93,6 +100,10 @@ suite "Bitswap engine - 2 nodes":
peerCtx2.peerHave.mapIt( $it ).sorted(cmp[string]) == peerCtx2.peerHave.mapIt( $it ).sorted(cmp[string]) ==
bitswap1.engine.wantList.mapIt( $it ).sorted(cmp[string]) bitswap1.engine.wantList.mapIt( $it ).sorted(cmp[string])
test "exchanges pricing on connect":
check peerCtx1.pricing == pricing1.some
check peerCtx2.pricing == pricing2.some
test "should send want-have for block": test "should send want-have for block":
let blk = bt.Block.new("Block 1".toBytes) let blk = bt.Block.new("Block 1".toBytes)
bitswap2.engine.localStore.putBlocks(@[blk]) bitswap2.engine.localStore.putBlocks(@[blk])

View File

@ -14,6 +14,7 @@ import pkg/dagger/chunker
import pkg/dagger/blocktype as bt import pkg/dagger/blocktype as bt
import ../helpers import ../helpers
import ../examples
suite "Bitswap engine basic": suite "Bitswap engine basic":
let let
@ -52,6 +53,21 @@ suite "Bitswap engine basic":
await done await done
test "sends pricing to new peers":
let pricing = Pricing.example
proc sendPricing(peer: PeerID, toBeSent: Pricing) =
check toBeSent == pricing
done.complete()
let request = BitswapRequest(sendPricing: sendPricing)
let engine = BitswapEngine.new(MemoryStore.new, request)
engine.pricing = pricing.some
engine.setupPeer(peerId)
await done.wait(100.millis)
suite "Bitswap engine handlers": suite "Bitswap engine handlers":
let let
rng = Rng.instance() rng = Rng.instance()