Dmitriy Ryajov 6c92b3dc25
Poc 2 (#7)
* moving protobuf into bitswap

* adding block type

* reworking bitswap

* adding chunker

* adding license header

* use 1.2.6

* adding fixed size chunker

* add blockstore

* add iterator to chunker

* more bitswap changes

* rename ipfs to dagger

* rename to dagger

* blockstore inherits from BlockProvider

* wip - add core block handling logic

* smal changes

* use proper block store methods

* adding asynq heapqueue

* wip prepare for bitswap task runner

* adding `$`

* adding memory store and tests

* fixed chunking

* extracted desicion engine from bitswap

* added helper random funcs

* adding testing helpers

* only handle seqs

* add peer events

* cleanup pending blocks on blockstore event

* allow nil handlers

* move protobuf type helpers

* allow initializing block from Cid

* testing and fixes

* small fixes

* expose `<`

* spelling

* default value

* spelling

* pending blocks manager

* adding stores manager

* more tests a wip around bitswap

* small changes

* merge bitswap and engine for now

* for now run only the new poc's tests

* add a more complete ci setup

* use template in map

* remove p2pd

* remove go

* dont use asyncCheck

* few small changes

* adding ability to update items

* adding multiple task runners

* handle cancelation properly

* use Result instead of throwing

* wip bitswap tests

* moving things around

* split out engine again

* add request and handlers interface

* fix tests

* wip - engine tests

* remove unused imports

* fix tests

* cleanup block requesting logic

* add block request tests

* more block requests

* add support for max heap

* don't use result

* use max heap & send block presence in task handler

* add task handler tests

* rename store to localStore

* cleanup & logging

* cancel task on stop

* don't depend on local store for events

* dont use heap queue for wants

* add chronicles

* fix issue with peer wants

* add test for delayed block sends

* remove obsolete tests

* wip chunker

* run all tests

* add todo

* misc

* remove irrelevant files

* removing more files

* adding helpers for bitswap tests

* moved bitswap file

* misc

* make blocks timeout longer

* adjust block timeout

* speedup test

* compile with threads

* import missing crypto

* misc

* disable threads for now

* fix 32 bit platforms

* re-enable threads support in tests
2021-02-25 18:23:22 -06:00

197 lines
5.1 KiB
Nim

import std/sequtils
import std/tables
import pkg/asynctest
import pkg/chronos
import pkg/libp2p
import pkg/libp2p/standard_setup
import pkg/libp2p/errors
import pkg/protobuf_serialization
import pkg/dagger/stores/memorystore
import pkg/dagger/bitswap/network
import pkg/dagger/p2p/rng
import pkg/dagger/chunker
import pkg/dagger/blocktype as bt
import ../helpers
suite "Bitswap network":
let
rng = Rng.instance()
seckey = PrivateKey.random(rng[]).tryGet()
peerId = PeerID.init(seckey.getKey().tryGet()).tryGet()
chunker = newRandomChunker(Rng.instance(), size = 1024, chunkSize = 256)
blocks = chunker.mapIt( bt.Block.new(it) )
var
network: BitswapNetwork
networkPeer: NetworkPeer
buffer: BufferStream
done: Future[void]
proc getConn(): Future[Connection] {.async.} =
return Connection(buffer)
setup:
done = newFuture[void]()
buffer = newBufferStream()
network = BitswapNetwork.new(
switch = newStandardSwitch(),
connProvider = getConn)
network.setupPeer(peerId)
networkPeer = network.peers[peerId]
discard await networkPeer.connect()
test "Want List handler":
proc wantListHandler(peer: PeerID, wantList: WantList) {.gcsafe.} =
# check that we got the correct amount of entries
check wantList.entries.len == 4
for b in blocks:
check b.cid in wantList.entries
let entry = wantList.entries[wantList.entries.find(b.cid)]
check entry.wantType == WantType.wantHave
check entry.priority == 1
check entry.cancel == true
check entry.sendDontHave == true
done.complete()
network.handlers.onWantList = wantListHandler
let wantList = makeWantList(
blocks.mapIt( it.cid ),
1, true, WantType.wantHave,
true, true)
let msg = Message(wantlist: wantList)
await buffer.pushData(lenPrefix(Protobuf.encode(msg)))
await done.wait(500.millis)
test "Blocks Handler":
proc blocksHandler(peer: PeerID, blks: seq[bt.Block]) {.gcsafe.} =
check blks == blocks
done.complete()
network.handlers.onBlocks = blocksHandler
let msg = Message(payload: makeBlocks(blocks))
await buffer.pushData(lenPrefix(Protobuf.encode(msg)))
await done.wait(500.millis)
test "Precense Handler":
proc presenceHandler(peer: PeerID, precense: seq[BlockPresence]) {.gcsafe.} =
for b in blocks:
check:
b.cid in precense
done.complete()
network.handlers.onPresence = presenceHandler
let msg = Message(
blockPresences: blocks.mapIt(
BlockPresence(
cid: it.cid.data.buffer,
type: BlockPresenceType.presenceHave
)))
await buffer.pushData(lenPrefix(Protobuf.encode(msg)))
await done.wait(500.millis)
suite "Bitswap Network - e2e":
let
chunker = newRandomChunker(Rng.instance(), size = 1024, chunkSize = 256)
blocks = chunker.mapIt( bt.Block.new(it) )
var
switch1, switch2: Switch
network1, network2: BitswapNetwork
awaiters: seq[Future[void]]
done: Future[void]
setup:
done = newFuture[void]()
switch1 = newStandardSwitch()
switch2 = newStandardSwitch()
awaiters.add(await switch1.start())
awaiters.add(await switch2.start())
network1 = BitswapNetwork.new(
switch = switch1)
switch1.mount(network1)
network2 = BitswapNetwork.new(
switch = switch2)
switch2.mount(network2)
await switch1.connect(
switch2.peerInfo.peerId,
switch2.peerInfo.addrs)
teardown:
await allFuturesThrowing(
switch1.stop(),
switch2.stop())
await allFuturesThrowing(awaiters)
test "broadcast want list":
proc wantListHandler(peer: PeerID, wantList: WantList) {.gcsafe.} =
# check that we got the correct amount of entries
check wantList.entries.len == 4
for b in blocks:
check b.cid in wantList.entries
let entry = wantList.entries[wantList.entries.find(b.cid)]
check entry.wantType == WantType.wantHave
check entry.priority == 1
check entry.cancel == true
check entry.sendDontHave == true
done.complete()
network2.handlers.onWantList = wantListHandler
network1.broadcastWantList(
switch2.peerInfo.peerId,
blocks.mapIt( it.cid ),
1, true, WantType.wantHave,
true, true)
await done.wait(500.millis)
test "broadcast blocks":
proc blocksHandler(peer: PeerID, blks: seq[bt.Block]) {.gcsafe.} =
check blks == blocks
done.complete()
network2.handlers.onBlocks = blocksHandler
network1.broadcastBlocks(
switch2.peerInfo.peerId,
blocks)
await done.wait(500.millis)
test "broadcast precense":
proc presenceHandler(peer: PeerID, precense: seq[BlockPresence]) {.gcsafe.} =
for b in blocks:
check:
b.cid in precense
done.complete()
network2.handlers.onPresence = presenceHandler
network1.broadcastBlockPresence(
switch2.peerInfo.peerId,
blocks.mapIt(
BlockPresence(
cid: it.cid.data.buffer,
type: BlockPresenceType.presenceHave
)))
await done.wait(500.millis)