Merge pull request #348 from status-im/p2p-chronos-strict

Adjust for chronosStrictException usage in rest of eth/p2p
This commit is contained in:
Kim De Mey 2021-05-11 11:03:43 +02:00 committed by GitHub
commit 2d75e73bab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 195 additions and 122 deletions

View File

@ -44,22 +44,11 @@ task test_discv5, "Run discovery v5 tests":
runTest("tests/p2p/all_discv5_tests")
task test_discv4, "Run discovery v4 tests":
runTest("tests/p2p/test_discovery", chronosStrict = false)
runTest("tests/p2p/test_discovery")
task test_p2p, "Run p2p tests":
runTest("tests/p2p/all_tests")
# Code that still requires chronosStrict = false
for filename in [
"les/test_flow_control",
"test_rlpx_thunk",
"test_shh",
"test_shh_config",
"test_shh_connect",
"test_protocol_handlers"
]:
runTest("tests/p2p/" & filename, chronosStrict = false)
task test_rlp, "Run rlp tests":
# workaround for github action CI
# mysterious crash on windows-2019 64bit mode

View File

@ -368,10 +368,12 @@ template deref*(b: Blob): auto = b
template deref*(o: Option): auto = o.get
template deref*(r: EthResourceRefs): auto = r[]
method genesisHash*(db: AbstractChainDB): KeccakHash {.base, gcsafe.} =
method genesisHash*(db: AbstractChainDB): KeccakHash
{.base, gcsafe, raises: [Defect].} =
notImplemented()
method getBlockHeader*(db: AbstractChainDB, b: HashOrNum, output: var BlockHeader): bool {.base, gcsafe.} =
method getBlockHeader*(db: AbstractChainDB, b: HashOrNum,
output: var BlockHeader): bool {.base, gcsafe, raises: [CatchableError, Defect].} =
notImplemented()
proc getBlockHeader*(db: AbstractChainDB, hash: KeccakHash): BlockHeaderRef {.gcsafe.} =
@ -384,22 +386,29 @@ proc getBlockHeader*(db: AbstractChainDB, b: BlockNumber): BlockHeaderRef {.gcsa
if not db.getBlockHeader(HashOrNum(isHash: false, number: b), result[]):
return nil
method getBestBlockHeader*(self: AbstractChainDB): BlockHeader {.base, gcsafe.} =
method getBestBlockHeader*(self: AbstractChainDB): BlockHeader
{.base, gcsafe, raises: [CatchableError, Defect].} =
notImplemented()
method getSuccessorHeader*(db: AbstractChainDB, h: BlockHeader, output: var BlockHeader, skip = 0'u): bool {.base, gcsafe.} =
method getSuccessorHeader*(db: AbstractChainDB, h: BlockHeader,
output: var BlockHeader, skip = 0'u): bool
{.base, gcsafe, raises: [CatchableError, Defect].} =
notImplemented()
method getAncestorHeader*(db: AbstractChainDB, h: BlockHeader, output: var BlockHeader, skip = 0'u): bool {.base, gcsafe.} =
method getAncestorHeader*(db: AbstractChainDB, h: BlockHeader,
output: var BlockHeader, skip = 0'u): bool
{.base, gcsafe, raises: [CatchableError, Defect].} =
notImplemented()
method getBlockBody*(db: AbstractChainDB, blockHash: KeccakHash): BlockBodyRef {.base, gcsafe.} =
method getBlockBody*(db: AbstractChainDB, blockHash: KeccakHash): BlockBodyRef
{.base, gcsafe, raises: [Defect].} =
notImplemented()
method getReceipt*(db: AbstractChainDB, hash: KeccakHash): ReceiptRef {.base, gcsafe.} =
notImplemented()
method getTrieDB*(db: AbstractChainDB): TrieDatabaseRef {.base, gcsafe.} =
method getTrieDB*(db: AbstractChainDB): TrieDatabaseRef
{.base, gcsafe, raises: [Defect].} =
notImplemented()
method getCodeByHash*(db: AbstractChainDB, hash: KeccakHash): Blob {.base, gcsafe.} =

View File

@ -18,6 +18,7 @@ proc getContractCode*(chain: AbstractChainDB, req: ContractCodeRequest): Blob {.
let acc = getAccount(chain.getTrieDB, b.stateRoot, req.key)
result = chain.getCodeByHash(acc.codeHash)
proc getStorageNode*(chain: AbstractChainDB, hash: KeccakHash): Blob =
proc getStorageNode*(chain: AbstractChainDB, hash: KeccakHash): Blob
{.raises: [CatchableError, Defect].} =
let db = chain.getTrieDB
return db.get(hash.data)

View File

@ -1,12 +1,9 @@
#
# Ethereum P2P
# (c) Copyright 2018
# Status Research & Development GmbH
#
# Licensed under either of
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
#
# nim-eth
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
std/[tables, algorithm, random],
@ -17,7 +14,8 @@ import
export
p2p_types, rlpx, enode, kademlia
proc addCapability*(node: var EthereumNode, p: ProtocolInfo) =
proc addCapability*(node: var EthereumNode, p: ProtocolInfo)
{.raises: [Defect].} =
doAssert node.connectionState == ConnectionState.None
let pos = lowerBound(node.protocols, p, rlpx.cmp)
@ -38,10 +36,10 @@ proc newEthereumNode*(keys: KeyPair,
addAllCapabilities = true,
useCompression: bool = false,
minPeers = 10,
rng = newRng()): EthereumNode =
rng = newRng()): EthereumNode {.raises: [Defect].} =
if rng == nil: # newRng could fail
raise (ref CatchableError)(msg: "Cannot initialize RNG")
raise (ref Defect)(msg: "Cannot initialize RNG")
new result
result.keys = keys

View File

@ -1,3 +1,10 @@
# nim-eth
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
std/[sets, options, random, hashes],
chronos, chronicles,
@ -36,7 +43,7 @@ type
trustedPeers: HashSet[Peer]
hasOutOfOrderBlocks: bool
proc hash*(p: Peer): Hash {.inline.} = hash(cast[pointer](p))
proc hash*(p: Peer): Hash = hash(cast[pointer](p))
proc endIndex(b: WantedBlocks): BlockNumber =
result = b.startIndex

View File

@ -1,10 +1,17 @@
# nim-eth
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
eth/common/[eth_types, state_accessors]
# TODO: Perhaps we can move this to eth-common
proc getBlockHeaders*(db: AbstractChainDB,
req: BlocksRequest): seq[BlockHeader] {.gcsafe.} =
proc getBlockHeaders*(db: AbstractChainDB, req: BlocksRequest): seq[BlockHeader]
{.gcsafe, raises: [CatchableError, Defect].} =
result = newSeqOfCap[BlockHeader](req.maxResults)
var foundBlock: BlockHeader

View File

@ -379,7 +379,7 @@ proc decodeMessage*(body: openarray[byte]): DecodeResult[Message] =
return err("Invalid request-id")
proc decode[T](rlp: var Rlp, v: var T)
{.inline, nimcall, raises:[RlpError, ValueError, Defect].} =
{.nimcall, raises:[RlpError, ValueError, Defect].} =
for k, v in v.fieldPairs:
v = rlp.read(typeof(v))

View File

@ -468,7 +468,7 @@ proc `$`*(r: Record): string =
proc `==`*(a, b: Record): bool = a.raw == b.raw
proc read*(rlp: var Rlp, T: typedesc[Record]):
T {.inline, raises:[RlpError, ValueError, Defect].} =
T {.raises: [RlpError, ValueError, Defect].} =
if not rlp.hasData() or not result.fromBytes(rlp.rawData):
# TODO: This could also just be an invalid signature, would be cleaner to
# split of RLP deserialisation errors from this.

View File

@ -193,7 +193,7 @@ proc neighbours*(d: Protocol, id: NodeId, k: int = BUCKET_SIZE): seq[Node] =
## Return up to k neighbours (closest node ids) of the given node id.
d.routingTable.neighbours(id, k)
proc nodesDiscovered*(d: Protocol): int {.inline.} = d.routingTable.len
proc nodesDiscovered*(d: Protocol): int = d.routingTable.len
func privKey*(d: Protocol): lent PrivateKey =
d.privateKey

View File

@ -132,8 +132,8 @@ proc distanceTo(k: KBucket, id: NodeId): UInt256 = k.midpoint xor id
proc nodesByDistanceTo(k: KBucket, id: NodeId): seq[Node] =
sortedByIt(k.nodes, it.distanceTo(id))
proc len(k: KBucket): int {.inline.} = k.nodes.len
proc tail(k: KBucket): Node {.inline.} = k.nodes[high(k.nodes)]
proc len(k: KBucket): int = k.nodes.len
proc tail(k: KBucket): Node = k.nodes[high(k.nodes)]
proc ipLimitInc(r: var RoutingTable, b: KBucket, n: Node): bool =
## Check if the ip limits of the routing table and the bucket are reached for
@ -194,7 +194,7 @@ proc split(k: KBucket): tuple[lower, upper: KBucket] =
doAssert(bucket.ipLimits.inc(node.address.get().ip),
"IpLimit increment should work as all buckets have the same limits")
proc inRange(k: KBucket, n: Node): bool {.inline.} =
proc inRange(k: KBucket, n: Node): bool =
k.istart <= n.id and n.id <= k.iend
proc contains(k: KBucket, n: Node): bool = n in k.nodes
@ -234,7 +234,7 @@ proc computeSharedPrefixBits(nodes: openarray[NodeId]): int =
doAssert(false, "Unable to calculate number of shared prefix bits")
proc init*(r: var RoutingTable, thisNode: Node, bitsPerHop = DefaultBitsPerHop,
ipLimits = DefaultTableIpLimits, rng: ref BrHmacDrbgContext) {.inline.} =
ipLimits = DefaultTableIpLimits, rng: ref BrHmacDrbgContext) =
## Initialize the routing table for provided `Node` and bitsPerHop value.
## `bitsPerHop` is default set to 5 as recommended by original Kademlia paper.
r.thisNode = thisNode

View File

@ -28,9 +28,12 @@ template networkState*(connection: Peer, Protocol: type): untyped =
## particular connection.
protocolState(connection.network, Protocol)
proc initProtocolState*[T](state: T, x: Peer|EthereumNode) {.gcsafe.} = discard
proc initProtocolState*[T](state: T, x: Peer|EthereumNode)
{.gcsafe, raises: [Defect].} =
discard
proc initProtocolStates(peer: Peer, protocols: openarray[ProtocolInfo]) =
proc initProtocolStates(peer: Peer, protocols: openarray[ProtocolInfo])
{.raises: [Defect].} =
# Initialize all the active protocol states
newSeq(peer.protocolStates, allProtocols.len)
for protocol in protocols:

View File

@ -1,3 +1,10 @@
# nim-eth
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
# PeerPool attempts to keep connections to at least min_peers
# on the given network.
@ -26,9 +33,7 @@ proc newPeerPool*(network: EthereumNode,
result.observers = initTable[int, PeerObserver]()
result.listenPort = listenPort
template ensureFuture(f: untyped) = asyncCheck f
proc nodesToConnect(p: PeerPool): seq[Node] {.inline.} =
proc nodesToConnect(p: PeerPool): seq[Node] =
p.discovery.randomNodes(p.minPeers).filterIt(it notin p.discovery.bootstrapNodes)
proc addObserver(p: PeerPool, observerId: int, observer: PeerObserver) =
@ -42,10 +47,10 @@ proc addObserver(p: PeerPool, observerId: int, observer: PeerObserver) =
proc delObserver(p: PeerPool, observerId: int) =
p.observers.del(observerId)
proc addObserver*(p: PeerPool, observerId: ref, observer: PeerObserver) {.inline.} =
proc addObserver*(p: PeerPool, observerId: ref, observer: PeerObserver) =
p.addObserver(cast[int](observerId), observer)
proc delObserver*(p: PeerPool, observerId: ref) {.inline.} =
proc delObserver*(p: PeerPool, observerId: ref) =
p.delObserver(cast[int](observerId))
template setProtocol*(observer: PeerObserver, Protocol: type) =
@ -95,20 +100,15 @@ proc connect(p: PeerPool, remote: Node): Future[Peer] {.async.} =
# self.logger.exception("Unexpected error during auth/p2p handshake with %s", remote)
# return None
proc lookupRandomNode(p: PeerPool) {.async.} =
# This method runs in the background, so we must catch OperationCancelled
# ere otherwise asyncio will warn that its exception was never retrieved.
try:
discard await p.discovery.lookupRandom()
except: # OperationCancelled
discard
proc lookupRandomNode(p: PeerPool) {.async, raises: [Defect].} =
discard await p.discovery.lookupRandom()
p.lastLookupTime = epochTime()
proc getRandomBootnode(p: PeerPool): Option[Node] =
if p.discovery.bootstrapNodes.len != 0:
result = option(p.discovery.bootstrapNodes.sample())
proc addPeer*(pool: PeerPool, peer: Peer) {.gcsafe.} =
proc addPeer*(pool: PeerPool, peer: Peer) {.gcsafe, raises: [Defect].} =
doAssert(peer.remote notin pool.connectedNodes)
pool.connectedNodes[peer.remote] = peer
connected_peers.inc()
@ -149,7 +149,7 @@ proc maybeConnectToMorePeers(p: PeerPool) {.async.} =
return
if p.lastLookupTime + lookupInterval < epochTime():
ensureFuture p.lookupRandomNode()
asyncSpawn p.lookupRandomNode()
let debugEnode = getEnv("ETH_DEBUG_ENODE")
if debugEnode.len != 0:
@ -163,7 +163,7 @@ proc maybeConnectToMorePeers(p: PeerPool) {.async.} =
if p.connectedNodes.len == 0 and (let n = p.getRandomBootnode(); n.isSome):
await p.connectToNode(n.get())
proc run(p: PeerPool) {.async.} =
proc run(p: PeerPool) {.async, raises: [Defect].} =
trace "Running PeerPool..."
p.running = true
while p.running:
@ -184,7 +184,7 @@ proc run(p: PeerPool) {.async.} =
proc start*(p: PeerPool) =
if not p.running:
asyncCheck p.run()
asyncSpawn p.run()
proc len*(p: PeerPool): int = p.connectedNodes.len
# @property

View File

@ -1,3 +1,10 @@
# nim-eth
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
std/[deques, tables],
bearssl, chronos,
@ -61,8 +68,8 @@ type
observers*: Table[int, PeerObserver]
PeerObserver* = object
onPeerConnected*: proc(p: Peer) {.gcsafe.}
onPeerDisconnected*: proc(p: Peer) {.gcsafe.}
onPeerConnected*: proc(p: Peer) {.gcsafe, raises: [Defect].}
onPeerDisconnected*: proc(p: Peer) {.gcsafe, raises: [Defect].}
protocol*: ProtocolInfo
Capability* = object
@ -138,16 +145,20 @@ type
# Private types:
MessageHandlerDecorator* = proc(msgId: int, n: NimNode): NimNode
ThunkProc* = proc(x: Peer, msgId: int, data: Rlp): Future[void] {.gcsafe.}
MessageContentPrinter* = proc(msg: pointer): string {.gcsafe.}
ThunkProc* = proc(x: Peer, msgId: int, data: Rlp): Future[void]
{.gcsafe, raises: [RlpError, Defect].}
MessageContentPrinter* = proc(msg: pointer): string
{.gcsafe, raises: [Defect].}
RequestResolver* = proc(msg: pointer, future: FutureBase)
{.gcsafe, raises:[Defect].}
NextMsgResolver* = proc(msgData: Rlp, future: FutureBase) {.gcsafe.}
PeerStateInitializer* = proc(peer: Peer): RootRef {.gcsafe.}
NetworkStateInitializer* = proc(network: EthereumNode): RootRef {.gcsafe.}
HandshakeStep* = proc(peer: Peer): Future[void] {.gcsafe.}
{.gcsafe, raises: [Defect].}
NextMsgResolver* = proc(msgData: Rlp, future: FutureBase)
{.gcsafe, raises: [RlpError, Defect].}
PeerStateInitializer* = proc(peer: Peer): RootRef {.gcsafe, raises: [Defect].}
NetworkStateInitializer* = proc(network: EthereumNode): RootRef
{.gcsafe, raises: [Defect].}
HandshakeStep* = proc(peer: Peer): Future[void] {.gcsafe, raises: [Defect].}
DisconnectionHandler* = proc(peer: Peer, reason: DisconnectionReason):
Future[void] {.gcsafe.}
Future[void] {.gcsafe, raises: [Defect].}
ConnectionState* = enum
None,

View File

@ -1,3 +1,10 @@
# nim-eth
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
std/[tables, algorithm, deques, hashes, options, typetraits],
stew/shims/macros, chronicles, nimcrypto, chronos,
@ -108,7 +115,7 @@ proc messagePrinter[MsgType](msg: pointer): string {.gcsafe.} =
# result = $(cast[ptr MsgType](msg)[])
proc disconnect*(peer: Peer, reason: DisconnectionReason,
notifyOtherPeer = false) {.gcsafe, async.}
notifyOtherPeer = false) {.gcsafe, raises: [Defect], async.}
template raisePeerDisconnected(msg: string, r: DisconnectionReason) =
var e = newException(PeerDisconnected, msg)
@ -248,17 +255,17 @@ func nameStr*(p: ProtocolInfo): string =
result = newStringOfCap(3)
for c in p.name: result.add(c)
# XXX: this used to be inline, but inline procs
# cannot be passed to closure params
proc cmp*(lhs, rhs: ProtocolInfo): int =
for i in 0..2:
if lhs.name[i] != rhs.name[i]:
return int16(lhs.name[i]) - int16(rhs.name[i])
return 0
proc nextMsgResolver[MsgType](msgData: Rlp, future: FutureBase) {.gcsafe.} =
proc nextMsgResolver[MsgType](msgData: Rlp, future: FutureBase)
{.gcsafe, raises: [RlpError, Defect].} =
var reader = msgData
Future[MsgType](future).complete reader.readRecordType(MsgType, MsgType.rlpFieldsCount > 1)
Future[MsgType](future).complete reader.readRecordType(MsgType,
MsgType.rlpFieldsCount > 1)
proc registerMsg(protocol: ProtocolInfo,
id: int, name: string,
@ -288,7 +295,7 @@ proc registerProtocol(protocol: ProtocolInfo) =
# Message composition and encryption
#
proc perPeerMsgIdImpl(peer: Peer, proto: ProtocolInfo, msgId: int): int {.inline.} =
proc perPeerMsgIdImpl(peer: Peer, proto: ProtocolInfo, msgId: int): int =
result = msgId
if not peer.dispatcher.isNil:
result += peer.dispatcher.protocolOffsets[proto.index]
@ -297,17 +304,18 @@ template getPeer(peer: Peer): auto = peer
template getPeer(responder: ResponderWithId): auto = responder.peer
template getPeer(responder: ResponderWithoutId): auto = Peer(responder)
proc supports*(peer: Peer, proto: ProtocolInfo): bool {.inline.} =
proc supports*(peer: Peer, proto: ProtocolInfo): bool =
peer.dispatcher.protocolOffsets[proto.index] != -1
proc supports*(peer: Peer, Protocol: type): bool {.inline.} =
proc supports*(peer: Peer, Protocol: type): bool =
## Checks whether a Peer supports a particular protocol
peer.supports(Protocol.protocolInfo)
template perPeerMsgId(peer: Peer, MsgType: type): int =
perPeerMsgIdImpl(peer, MsgType.msgProtocol.protocolInfo, MsgType.msgId)
proc invokeThunk*(peer: Peer, msgId: int, msgData: var Rlp): Future[void] =
proc invokeThunk*(peer: Peer, msgId: int, msgData: var Rlp): Future[void]
{.raises: [UnsupportedMessageError, RlpError, Defect].} =
template invalidIdError: untyped =
raise newException(UnsupportedMessageError,
"RLPx message with an invalid id " & $msgId &
@ -367,7 +375,7 @@ proc registerRequest(peer: Peer,
proc timeoutExpired(udata: pointer) {.gcsafe, raises:[Defect].} =
requestResolver(nil, responseFuture)
addTimer(timeoutAt, timeoutExpired, nil)
discard setTimer(timeoutAt, timeoutExpired, nil)
proc resolveResponseFuture(peer: Peer, msgId: int, msg: pointer, reqId: int) =
logScope:
@ -504,7 +512,7 @@ proc recvMsg*(peer: Peer): Future[tuple[msgId: int, msgData: Rlp]] {.async.} =
await peer.disconnectAndRaise(BreachOfProtocol,
"Cannot read RLPx message id")
proc checkedRlpRead(peer: Peer, r: var Rlp, MsgType: type): auto {.inline.} =
proc checkedRlpRead(peer: Peer, r: var Rlp, MsgType: type): auto =
when defined(release):
return r.read(MsgType)
else:
@ -766,7 +774,10 @@ proc p2pProtocolBackendImpl*(protocol: P2PProtocol): Backend =
thunkName = ident(msgName & "Thunk")
msg.defineThunk quote do:
proc `thunkName`(`peerVar`: `Peer`, _: int, data: Rlp) {.async, gcsafe.} =
proc `thunkName`(`peerVar`: `Peer`, _: int, data: Rlp)
# Fun error if you just use `RlpError` instead of `rlp.RlpError`:
# "Error: type expected, but got symbol 'RlpError' of kind 'EnumField'"
{.async, gcsafe, raises: [rlp.RlpError, Defect].} =
var `receivedRlp` = data
var `receivedMsg` {.noinit.}: `msgRecName`
`readParamsPrelude`
@ -866,12 +877,13 @@ p2pProtocol DevP2P(version = 5, rlpxName = "p2p"):
proc pong(peer: Peer, emptyList: EmptyList) =
discard
proc removePeer(network: EthereumNode, peer: Peer) =
proc removePeer(network: EthereumNode, peer: Peer) {.raises: [Defect].} =
# It is necessary to check if peer.remote still exists. The connection might
# have been dropped already from the peers side.
# E.g. when receiving a p2p.disconnect message from a peer, a race will happen
# between which side disconnects first.
if network.peerPool != nil and not peer.remote.isNil and peer.remote in network.peerPool.connectedNodes:
if network.peerPool != nil and not peer.remote.isNil and
peer.remote in network.peerPool.connectedNodes:
network.peerPool.connectedNodes.del(peer.remote)
connected_peers.dec()
@ -883,16 +895,23 @@ proc removePeer(network: EthereumNode, peer: Peer) =
if observer.protocol.isNil or peer.supports(observer.protocol):
observer.onPeerDisconnected(peer)
proc callDisconnectHandlers(peer: Peer, reason: DisconnectionReason): Future[void] =
proc callDisconnectHandlers(peer: Peer, reason: DisconnectionReason):
Future[void] {.async.} =
var futures = newSeqOfCap[Future[void]](allProtocols.len)
for protocol in peer.dispatcher.activeProtocols:
if protocol.disconnectHandler != nil:
futures.add((protocol.disconnectHandler)(peer, reason))
return all(futures)
await allFutures(futures)
proc disconnect*(peer: Peer, reason: DisconnectionReason, notifyOtherPeer = false) {.async.} =
for f in futures:
doAssert(f.finished())
if f.failed():
trace "Disconnection handler ended with an error", err = f.error.msg
proc disconnect*(peer: Peer, reason: DisconnectionReason,
notifyOtherPeer = false) {.async, raises: [Defect].} =
if peer.connectionState notin {Disconnecting, Disconnected}:
peer.connectionState = Disconnecting
# Do this first so sub-protocols have time to clean up and stop sending
@ -901,7 +920,7 @@ proc disconnect*(peer: Peer, reason: DisconnectionReason, notifyOtherPeer = fals
# In case of `CatchableError` in any of the handlers, this will be logged.
# Other handlers will still execute.
# In case of `Defect` in any of the handlers, program will quit.
traceAwaitErrors callDisconnectHandlers(peer, reason)
await callDisconnectHandlers(peer, reason)
if notifyOtherPeer and not peer.transport.closed:
var fut = peer.sendDisconnectMsg(DisconnectionReasonList(value: reason))
@ -926,12 +945,13 @@ proc validatePubKeyInHello(msg: DevP2P.hello, pubKey: PublicKey): bool =
let pk = PublicKey.fromRaw(msg.nodeId)
pk.isOk and pk[] == pubKey
proc checkUselessPeer(peer: Peer) {.inline.} =
proc checkUselessPeer(peer: Peer) =
if peer.dispatcher.numProtocols == 0:
# XXX: Send disconnect + UselessPeer
raise newException(UselessPeerError, "Useless peer")
proc initPeerState*(peer: Peer, capabilities: openarray[Capability]) =
proc initPeerState*(peer: Peer, capabilities: openarray[Capability])
{.raises: [UselessPeerError, Defect].} =
peer.dispatcher = getDispatcher(peer.network, capabilities)
checkUselessPeer(peer)
@ -979,7 +999,12 @@ proc postHelloSteps(peer: Peer, h: DevP2P.hello) {.async.} =
# The handshake may involve multiple async steps, so we wait
# here for all of them to finish.
#
await all(subProtocolsHandshakes)
await allFutures(subProtocolsHandshakes)
for handshake in subProtocolsHandshakes:
doAssert(handshake.finished())
if handshake.failed():
raise handshake.error
# This is needed as a peer might have already disconnected. In this case
# we need to raise so that rlpxConnect/rlpxAccept fails.

View File

@ -316,7 +316,7 @@ proc acceptRequest*(network: LesNetwork, peer: LesPeer,
network.updateFlowControl t
while not network.canServeRequest:
await sleepAsync(10)
await sleepAsync(10.milliseconds)
if peer notin network.peers:
# The peer was disconnected or the network

View File

@ -1,19 +1,19 @@
#
# Whisper
# (c) Copyright 2018-2019
# Status Research & Development GmbH
#
# Licensed under either of
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
#
# nim-eth - Whisper
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
std/[algorithm, bitops, math, options, tables, times, strutils, hashes],
std/[algorithm, bitops, math, options, tables, times, hashes],
chronicles, stew/[byteutils, endians2], metrics, bearssl,
nimcrypto/[bcmode, hash, keccak, rijndael],
".."/../../[keys, rlp, p2p], ../../ecies
when chronicles.enabledLogLevel == LogLevel.TRACE:
import std/strutils
logScope:
topics = "whisper_types"

View File

@ -1,12 +1,9 @@
#
# Whisper
# (c) Copyright 2018-2019
# Status Research & Development GmbH
#
# Licensed under either of
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
#
# nim-eth - Whisper
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
## Whisper
## *******
@ -96,8 +93,9 @@ proc allowed*(msg: Message, config: WhisperConfig): bool =
return true
proc run(peer: Peer) {.gcsafe, async.}
proc run(node: EthereumNode, network: WhisperNetwork) {.gcsafe, async.}
proc run(peer: Peer) {.gcsafe, async, raises: [Defect].}
proc run(node: EthereumNode, network: WhisperNetwork)
{.gcsafe, async, raises: [Defect].}
proc initProtocolState*(network: WhisperNetwork, node: EthereumNode) {.gcsafe.} =
new(network.queue)
@ -107,7 +105,7 @@ proc initProtocolState*(network: WhisperNetwork, node: EthereumNode) {.gcsafe.}
network.config.powRequirement = defaultMinPow
network.config.isLightNode = false
network.config.maxMsgSize = defaultMaxMsgSize
asyncCheck node.run(network)
asyncSpawn node.run(network)
p2pProtocol Whisper(version = whisperVersion,
rlpxName = "shh",
@ -152,7 +150,7 @@ p2pProtocol Whisper(version = whisperVersion,
whisperPeer.initialized = true
if not whisperNet.config.isLightNode:
traceAsyncErrors peer.run()
asyncSpawn peer.run()
debug "Whisper peer initialized", peer
@ -249,7 +247,7 @@ p2pProtocol Whisper(version = whisperVersion,
# 'Runner' calls ---------------------------------------------------------------
proc processQueue(peer: Peer) =
proc processQueue(peer: Peer) {.raises: [Defect].} =
# Send to peer all valid and previously not send envelopes in the queue.
var
envelopes: seq[Envelope] = @[]
@ -280,7 +278,7 @@ proc processQueue(peer: Peer) =
# gets dropped
traceAsyncErrors peer.messages(envelopes)
proc run(peer: Peer) {.async.} =
proc run(peer: Peer) {.async, raises: [Defect].} =
while peer.connectionState notin {Disconnecting, Disconnected}:
peer.processQueue()
await sleepAsync(messageInterval)
@ -298,7 +296,7 @@ proc pruneReceived(node: EthereumNode) {.raises: [].} =
# the received sets.
peer.received = intersection(peer.received, whisperNet.queue.itemHashes)
proc run(node: EthereumNode, network: WhisperNetwork) {.async.} =
proc run(node: EthereumNode, network: WhisperNetwork) {.async, raises: [Defect].} =
while true:
# prune message queue every second
# TTL unit is in seconds, so this should be sufficient?
@ -313,7 +311,13 @@ proc run(node: EthereumNode, network: WhisperNetwork) {.async.} =
proc sendP2PMessage(node: EthereumNode, peerId: NodeId, env: Envelope): bool =
for peer in node.peers(Whisper):
if peer.remote.id == peerId:
asyncCheck peer.p2pMessage(env)
let f = peer.p2pMessage(env)
# Can't make p2pMessage not raise so this is the "best" option I can think
# of instead of using asyncSpawn and still keeping the call not async.
f.callback = proc(data: pointer) {.gcsafe, raises: [Defect].} =
if f.failed:
warn "P2PMessage send failed", msg = f.readError.msg
return true
proc queueMessage(node: EthereumNode, msg: Message): bool =

View File

@ -4,4 +4,11 @@ import
./test_crypt,
./test_discovery,
./test_ecies,
./test_enode
./test_enode,
./test_rlpx_thunk,
./test_shh,
./test_shh_config,
./test_shh_connect,
./test_protocol_handlers,
./les/test_flow_control

View File

@ -1,3 +1,5 @@
{.used.}
import
../../../eth/p2p/rlpx_protocols/les/flow_control

View File

@ -1,6 +1,6 @@
import
std/[unittest, strutils],
chronos, nimcrypto, bearssl,
std/strutils,
chronos, bearssl,
../../eth/[keys, p2p], ../../eth/p2p/[discovery, enode]
var nextPort = 30303

View File

@ -7,6 +7,8 @@
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
{.used.}
import
std/tables,
chronos, testutils/unittests,

View File

@ -1,3 +1,5 @@
{.used.}
import
std/[json, os, unittest],
chronos, stew/byteutils,

View File

@ -7,6 +7,8 @@
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
{.used.}
import
std/[sequtils, options, unittest, tables],
nimcrypto/hash,

View File

@ -7,6 +7,8 @@
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
{.used.}
import
std/[sequtils, options, unittest, times],
../../eth/p2p/rlpx_protocols/whisper_protocol as whisper

View File

@ -7,6 +7,8 @@
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
{.used.}
import
std/[sequtils, options, tables],
chronos, testutils/unittests, bearssl,