status-im/nimbus-eth2
2
0
Fork 0
mirror of https://github.com/status-im/nimbus-eth2.git synced 2025-01-09 13:56:23 +00:00
Code Issues Projects Releases Wiki Activity

Clean up some obsolete code

Browse Source
This commit is contained in:
Zahary Karadjov 2020-03-23 01:23:21 +02:00 committed by zah
parent d5e4e640b4
commit 4623aa81ec
6 changed files with 82 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
beacon_chain/beacon_node.nim
View File

@ -190,17 +190,15 @@ proc init*(T: type BeaconNode, conf: BeaconNodeConf): Future[BeaconNode] {.async
var bootNodes: seq[ENode] var bootNodes: seq[ENode]
var bootEnrs: seq[enr.Record] var bootEnrs: seq[enr.Record]
for node in conf.bootstrapNodes: addBootstrapNode(node, bootNodes, bootEnrs, ourPubKey) for node in conf.bootstrapNodes:
addBootstrapNode(node, bootNodes, bootEnrs, ourPubKey)
loadBootstrapFile(string conf.bootstrapNodesFile, bootNodes, bootEnrs, ourPubKey) loadBootstrapFile(string conf.bootstrapNodesFile, bootNodes, bootEnrs, ourPubKey)
let persistentBootstrapFile = conf.dataDir / "bootstrap_nodes.txt" let persistentBootstrapFile = conf.dataDir / "bootstrap_nodes.txt"
if fileExists(persistentBootstrapFile): if fileExists(persistentBootstrapFile):
loadBootstrapFile(persistentBootstrapFile, bootNodes, bootEnrs, ourPubKey) loadBootstrapFile(persistentBootstrapFile, bootNodes, bootEnrs, ourPubKey)
let let network = await createEth2Node(conf)
network = await createEth2Node(conf, bootNodes)
addressFile = string(conf.dataDir) / "beacon_node.address"
network.saveConnectionAddressFile(addressFile)
let rpcServer = if conf.rpcEnabled: let rpcServer = if conf.rpcEnabled:
RpcServer.init(conf.rpcAddress, conf.rpcPort) RpcServer.init(conf.rpcAddress, conf.rpcPort)
@ -252,13 +250,12 @@ proc init*(T: type BeaconNode, conf: BeaconNodeConf): Future[BeaconNode] {.async
return res return res
proc connectToNetwork(node: BeaconNode) {.async.} = proc connectToNetwork(node: BeaconNode) {.async.} =
if node.bootstrapNodes.len > 0: if node.bootstrapEnrs.len > 0:
info "Connecting to bootstrap nodes", bootstrapNodes = node.bootstrapNodes info "Connecting to bootstrap nodes", bootstrapEnrs = node.bootstrapEnrs
else: else:
info "Waiting for connections" info "Waiting for connections"
await node.network.connectToNetwork(node.bootstrapNodes, await node.network.connectToNetwork(node.bootstrapEnrs)
node.bootstrapEnrs)
template findIt(s: openarray, predicate: untyped): int = template findIt(s: openarray, predicate: untyped): int =
var res = -1 var res = -1

156
beacon_chain/eth2_network.nim
View File

@ -38,7 +38,6 @@ type
PrivateKey* = crypto.PrivateKey PrivateKey* = crypto.PrivateKey
Bytes = seq[byte] Bytes = seq[byte]
P2PStream = Connection
# TODO Is this really needed? # TODO Is this really needed?
Eth2Node* = ref object of RootObj Eth2Node* = ref object of RootObj
@ -70,7 +69,7 @@ type
UntypedResponder = object UntypedResponder = object
peer*: Peer peer*: Peer
stream*: P2PStream stream*: Connection
Responder*[MsgType] = distinct UntypedResponder Responder*[MsgType] = distinct UntypedResponder
@ -80,8 +79,6 @@ type
# Private fields: # Private fields:
libp2pCodecName: string libp2pCodecName: string
protocolMounter*: MounterProc protocolMounter*: MounterProc
printer*: MessageContentPrinter
nextMsgResolver*: NextMsgResolver
ProtocolInfoObj* = object ProtocolInfoObj* = object
name*: string name*: string
@ -104,12 +101,11 @@ type
PeerStateInitializer* = proc(peer: Peer): RootRef {.gcsafe.} PeerStateInitializer* = proc(peer: Peer): RootRef {.gcsafe.}
NetworkStateInitializer* = proc(network: EthereumNode): RootRef {.gcsafe.} NetworkStateInitializer* = proc(network: EthereumNode): RootRef {.gcsafe.}
HandshakeStep* = proc(peer: Peer, stream: P2PStream): Future[void] {.gcsafe.} HandshakeStep* = proc(peer: Peer, conn: Connection): Future[void] {.gcsafe.}
DisconnectionHandler* = proc(peer: Peer): Future[void] {.gcsafe.} DisconnectionHandler* = proc(peer: Peer): Future[void] {.gcsafe.}
ThunkProc* = LPProtoHandler ThunkProc* = LPProtoHandler
MounterProc* = proc(network: Eth2Node) {.gcsafe.} MounterProc* = proc(network: Eth2Node) {.gcsafe.}
MessageContentPrinter* = proc(msg: pointer): string {.gcsafe.} MessageContentPrinter* = proc(msg: pointer): string {.gcsafe.}
NextMsgResolver* = proc(msgData: SszReader, future: FutureBase) {.gcsafe.}
DisconnectionReason* = enum DisconnectionReason* = enum
ClientShutDown ClientShutDown
@ -148,10 +144,10 @@ declareCounter gossip_messages_received,
"Number of gossip messages received by this peer" "Number of gossip messages received by this peer"
declarePublicGauge libp2p_successful_dials, declarePublicGauge libp2p_successful_dials,
"Number of successfully dialed peers" "Number of successfully dialed peers"
declarePublicGauge libp2p_peers, declarePublicGauge libp2p_peers,
"Number of active libp2p peers" "Number of active libp2p peers"
template libp2pProtocol*(name: string, version: int) {.pragma.} template libp2pProtocol*(name: string, version: int) {.pragma.}
@ -161,21 +157,12 @@ chronicles.formatIt(Peer): $it
template remote*(peer: Peer): untyped = template remote*(peer: Peer): untyped =
peer.info.peerId peer.info.peerId
# TODO: This exists only as a compatibility layer between the daemon
# APIs and the native LibP2P ones. It won't be necessary once the
# daemon is removed.
#
template writeAllBytes(stream: P2PStream, bytes: seq[byte]): untyped =
write(stream, bytes)
template openStream(node: Eth2Node, peer: Peer, protocolId: string): untyped = template openStream(node: Eth2Node, peer: Peer, protocolId: string): untyped =
dial(node.switch, peer.info, protocolId) dial(node.switch, peer.info, protocolId)
proc peer(stream: P2PStream): PeerID = func peerId(conn: Connection): PeerID =
# TODO: Can this be `nil`? # TODO: Can this be `nil`?
stream.peerInfo.peerId conn.peerInfo.peerId
#
# End of compatibility layer
proc init*(T: type Peer, network: Eth2Node, info: PeerInfo): Peer {.gcsafe.} proc init*(T: type Peer, network: Eth2Node, info: PeerInfo): Peer {.gcsafe.}
@ -183,11 +170,12 @@ proc getPeer*(node: Eth2Node, peerInfo: PeerInfo): Peer {.gcsafe.} =
let peerId = peerInfo.peerId let peerId = peerInfo.peerId
result = node.peerPool.getOrDefault(peerId) result = node.peerPool.getOrDefault(peerId)
if result == nil: if result == nil:
# TODO: We should register this peer in the pool!
result = Peer.init(node, peerInfo) result = Peer.init(node, peerInfo)
proc peerFromStream(network: Eth2Node, stream: P2PStream): Peer {.gcsafe.} = proc peerFromStream(network: Eth2Node, conn: Connection): Peer {.gcsafe.} =
# TODO: Can this be `nil`? # TODO: Can this be `nil`?
return network.getPeer(stream.peerInfo) return network.getPeer(conn.peerInfo)
proc getKey*(peer: Peer): PeerID {.inline.} = proc getKey*(peer: Peer): PeerID {.inline.} =
result = peer.info.peerId result = peer.info.peerId
@ -208,9 +196,9 @@ proc disconnect*(peer: Peer, reason: DisconnectionReason,
peer.network.peerPool.release(peer) peer.network.peerPool.release(peer)
peer.info.close() peer.info.close()
proc safeClose(stream: P2PStream) {.async.} = proc safeClose(conn: Connection) {.async.} =
if not stream.closed: if not conn.closed:
await close(stream) await close(conn)
proc handleIncomingPeer*(peer: Peer) proc handleIncomingPeer*(peer: Peer)
@ -243,18 +231,18 @@ proc disconnectAndRaise(peer: Peer,
await peer.disconnect(r) await peer.disconnect(r)
raisePeerDisconnected(msg, r) raisePeerDisconnected(msg, r)
proc readChunk(stream: P2PStream, proc readChunk(conn: Connection,
MsgType: type, MsgType: type,
withResponseCode: bool, withResponseCode: bool,
deadline: Future[void]): Future[Option[MsgType]] {.gcsafe.} deadline: Future[void]): Future[Option[MsgType]] {.gcsafe.}
proc readSizePrefix(stream: P2PStream, proc readSizePrefix(conn: Connection,
deadline: Future[void]): Future[int] {.async.} = deadline: Future[void]): Future[int] {.async.} =
trace "about to read msg size prefix" trace "about to read msg size prefix"
var parser: VarintParser[uint64, ProtoBuf] var parser: VarintParser[uint64, ProtoBuf]
while true: while true:
var nextByte: byte var nextByte: byte
var readNextByte = stream.readExactly(addr nextByte, 1) var readNextByte = conn.readExactly(addr nextByte, 1)
await readNextByte or deadline await readNextByte or deadline
if not readNextByte.finished: if not readNextByte.finished:
trace "size prefix byte not received in time" trace "size prefix byte not received in time"
@ -274,7 +262,7 @@ proc readSizePrefix(stream: P2PStream,
of Incomplete: of Incomplete:
continue continue
proc readMsgBytes(stream: P2PStream, proc readMsgBytes(conn: Connection,
withResponseCode: bool, withResponseCode: bool,
deadline: Future[void]): Future[Bytes] {.async.} = deadline: Future[void]): Future[Bytes] {.async.} =
trace "about to read message bytes", withResponseCode trace "about to read message bytes", withResponseCode
@ -283,7 +271,7 @@ proc readMsgBytes(stream: P2PStream,
if withResponseCode: if withResponseCode:
var responseCode: byte var responseCode: byte
trace "about to read response code" trace "about to read response code"
var readResponseCode = stream.readExactly(addr responseCode, 1) var readResponseCode = conn.readExactly(addr responseCode, 1)
await readResponseCode or deadline await readResponseCode or deadline
if not readResponseCode.finished: if not readResponseCode.finished:
@ -299,7 +287,7 @@ proc readMsgBytes(stream: P2PStream,
case ResponseCode(responseCode) case ResponseCode(responseCode)
of InvalidRequest, ServerError: of InvalidRequest, ServerError:
let responseErrMsg = await readChunk(stream, string, false, deadline) let responseErrMsg = await conn.readChunk(string, false, deadline)
debug "P2P request resulted in error", responseErrMsg debug "P2P request resulted in error", responseErrMsg
return return
@ -307,20 +295,20 @@ proc readMsgBytes(stream: P2PStream,
# The response is OK, the execution continues below # The response is OK, the execution continues below
discard discard
var sizePrefix = await readSizePrefix(stream, deadline) var sizePrefix = await conn.readSizePrefix(deadline)
trace "got msg size prefix", sizePrefix trace "got msg size prefix", sizePrefix
if sizePrefix == -1: if sizePrefix == -1:
debug "Failed to read an incoming message size prefix", peer = stream.peer debug "Failed to read an incoming message size prefix", peer = conn.peerId
return return
if sizePrefix == 0: if sizePrefix == 0:
debug "Received SSZ with zero size", peer = stream.peer debug "Received SSZ with zero size", peer = conn.peerId
return return
trace "about to read msg bytes", len = sizePrefix trace "about to read msg bytes", len = sizePrefix
var msgBytes = newSeq[byte](sizePrefix) var msgBytes = newSeq[byte](sizePrefix)
var readBody = stream.readExactly(addr msgBytes[0], sizePrefix) var readBody = conn.readExactly(addr msgBytes[0], sizePrefix)
await readBody or deadline await readBody or deadline
if not readBody.finished: if not readBody.finished:
trace "msg bytes not received in time" trace "msg bytes not received in time"
@ -332,11 +320,11 @@ proc readMsgBytes(stream: P2PStream,
except TransportIncompleteError: except TransportIncompleteError:
return @[] return @[]
proc readChunk(stream: P2PStream, proc readChunk(conn: Connection,
MsgType: type, MsgType: type,
withResponseCode: bool, withResponseCode: bool,
deadline: Future[void]): Future[Option[MsgType]] {.gcsafe, async.} = deadline: Future[void]): Future[Option[MsgType]] {.gcsafe, async.} =
var msgBytes = await stream.readMsgBytes(withResponseCode, deadline) var msgBytes = await conn.readMsgBytes(withResponseCode, deadline)
try: try:
if msgBytes.len > 0: if msgBytes.len > 0:
return some SSZ.decode(msgBytes, MsgType) return some SSZ.decode(msgBytes, MsgType)
@ -346,7 +334,7 @@ proc readChunk(stream: P2PStream,
return return
proc readResponse( proc readResponse(
stream: P2PStream, conn: Connection,
MsgType: type, MsgType: type,
deadline: Future[void]): Future[Option[MsgType]] {.gcsafe, async.} = deadline: Future[void]): Future[Option[MsgType]] {.gcsafe, async.} =
@ -354,13 +342,13 @@ proc readResponse(
type E = ElemType(MsgType) type E = ElemType(MsgType)
var results: MsgType var results: MsgType
while true: while true:
let nextRes = await readChunk(stream, E, true, deadline) let nextRes = await conn.readChunk(E, true, deadline)
if nextRes.isNone: break if nextRes.isNone: break
results.add nextRes.get results.add nextRes.get
if results.len > 0: if results.len > 0:
return some(results) return some(results)
else: else:
return await readChunk(stream, MsgType, true, deadline) return await conn.readChunk(MsgType, true, deadline)
proc encodeErrorMsg(responseCode: ResponseCode, errMsg: string): Bytes = proc encodeErrorMsg(responseCode: ResponseCode, errMsg: string): Bytes =
var s = init OutputStream var s = init OutputStream
@ -370,7 +358,7 @@ proc encodeErrorMsg(responseCode: ResponseCode, errMsg: string): Bytes =
s.getOutput s.getOutput
proc sendErrorResponse(peer: Peer, proc sendErrorResponse(peer: Peer,
stream: P2PStream, conn: Connection,
err: ref SerializationError, err: ref SerializationError,
msgName: string, msgName: string,
msgBytes: Bytes) {.async.} = msgBytes: Bytes) {.async.} =
@ -378,18 +366,18 @@ proc sendErrorResponse(peer: Peer,
peer, msgName, msgBytes, errMsg = err.formatMsg("<msg>") peer, msgName, msgBytes, errMsg = err.formatMsg("<msg>")
let responseBytes = encodeErrorMsg(InvalidRequest, err.formatMsg("msg")) let responseBytes = encodeErrorMsg(InvalidRequest, err.formatMsg("msg"))
await stream.writeAllBytes(responseBytes) await conn.write(responseBytes)
await stream.close() await conn.close()
proc sendErrorResponse(peer: Peer, proc sendErrorResponse(peer: Peer,
stream: P2PStream, conn: Connection,
responseCode: ResponseCode, responseCode: ResponseCode,
errMsg: string) {.async.} = errMsg: string) {.async.} =
debug "Error processing request", peer, responseCode, errMsg debug "Error processing request", peer, responseCode, errMsg
let responseBytes = encodeErrorMsg(ServerError, errMsg) let responseBytes = encodeErrorMsg(ServerError, errMsg)
await stream.writeAllBytes(responseBytes) await conn.write(responseBytes)
await stream.close() await conn.close()
proc sendNotificationMsg(peer: Peer, protocolId: string, requestBytes: Bytes) {.async} = proc sendNotificationMsg(peer: Peer, protocolId: string, requestBytes: Bytes) {.async} =
var deadline = sleepAsync RESP_TIMEOUT var deadline = sleepAsync RESP_TIMEOUT
@ -409,7 +397,7 @@ proc sendNotificationMsg(peer: Peer, protocolId: string, requestBytes: Bytes) {.
s.appendVarint requestBytes.len.uint64 s.appendVarint requestBytes.len.uint64
s.append requestBytes s.append requestBytes
let bytes = s.getOutput let bytes = s.getOutput
await stream.writeAllBytes(bytes) await stream.write(bytes)
# TODO There is too much duplication in the responder functions, but # TODO There is too much duplication in the responder functions, but
# I hope to reduce this when I increse the reliance on output streams. # I hope to reduce this when I increse the reliance on output streams.
@ -419,14 +407,14 @@ proc sendResponseChunkBytes(responder: UntypedResponder, payload: Bytes) {.async
s.appendVarint payload.len.uint64 s.appendVarint payload.len.uint64
s.append payload s.append payload
let bytes = s.getOutput let bytes = s.getOutput
await responder.stream.writeAllBytes(bytes) await responder.stream.write(bytes)
proc sendResponseChunkObj(responder: UntypedResponder, val: auto) {.async.} = proc sendResponseChunkObj(responder: UntypedResponder, val: auto) {.async.} =
var s = init OutputStream var s = init OutputStream
s.append byte(Success) s.append byte(Success)
s.appendValue SSZ, sizePrefixed(val) s.appendValue SSZ, sizePrefixed(val)
let bytes = s.getOutput let bytes = s.getOutput
await responder.stream.writeAllBytes(bytes) await responder.stream.write(bytes)
proc sendResponseChunks[T](responder: UntypedResponder, chunks: seq[T]) {.async.} = proc sendResponseChunks[T](responder: UntypedResponder, chunks: seq[T]) {.async.} =
var s = init OutputStream var s = init OutputStream
@ -435,7 +423,7 @@ proc sendResponseChunks[T](responder: UntypedResponder, chunks: seq[T]) {.async.
s.appendValue SSZ, sizePrefixed(chunk) s.appendValue SSZ, sizePrefixed(chunk)
let bytes = s.getOutput let bytes = s.getOutput
await responder.stream.writeAllBytes(bytes) await responder.stream.write(bytes)
proc makeEth2Request(peer: Peer, protocolId: string, requestBytes: Bytes, proc makeEth2Request(peer: Peer, protocolId: string, requestBytes: Bytes,
ResponseMsg: type, ResponseMsg: type,
@ -460,14 +448,14 @@ proc makeEth2Request(peer: Peer, protocolId: string, requestBytes: Bytes,
s.appendVarint requestBytes.len.uint64 s.appendVarint requestBytes.len.uint64
s.append requestBytes s.append requestBytes
let bytes = s.getOutput let bytes = s.getOutput
await stream.writeAllBytes(bytes) await stream.write(bytes)
# Read the response # Read the response
return await stream.readResponse(ResponseMsg, deadline) return await stream.readResponse(ResponseMsg, deadline)
proc init*[MsgType](T: type Responder[MsgType], proc init*[MsgType](T: type Responder[MsgType],
peer: Peer, stream: P2PStream): T = peer: Peer, conn: Connection): T =
T(UntypedResponder(peer: peer, stream: stream)) T(UntypedResponder(peer: peer, stream: conn))
template write*[M](r: var Responder[M], val: auto): auto = template write*[M](r: var Responder[M], val: auto): auto =
mixin send mixin send
@ -539,7 +527,7 @@ proc implementSendProcBody(sendProc: SendProc) =
sendProc.useStandardBody(nil, nil, sendCallGenerator) sendProc.useStandardBody(nil, nil, sendCallGenerator)
proc handleIncomingStream(network: Eth2Node, stream: P2PStream, proc handleIncomingStream(network: Eth2Node, conn: Connection,
MsgType, Format: distinct type) {.async, gcsafe.} = MsgType, Format: distinct type) {.async, gcsafe.} =
mixin callUserHandler, RecType mixin callUserHandler, RecType
const msgName = typetraits.name(MsgType) const msgName = typetraits.name(MsgType)
@ -550,21 +538,21 @@ proc handleIncomingStream(network: Eth2Node, stream: P2PStream,
# when chronicles.runtimeFilteringEnabled: # when chronicles.runtimeFilteringEnabled:
# setLogLevel(LogLevel.TRACE) # setLogLevel(LogLevel.TRACE)
# defer: setLogLevel(LogLevel.DEBUG) # defer: setLogLevel(LogLevel.DEBUG)
# trace "incoming " & `msgNameLit` & " stream" # trace "incoming " & `msgNameLit` & " conn"
let peer = peerFromStream(network, stream) let peer = peerFromStream(network, conn)
handleIncomingPeer(peer) handleIncomingPeer(peer)
defer: defer:
await safeClose(stream) await safeClose(conn)
let let
deadline = sleepAsync RESP_TIMEOUT deadline = sleepAsync RESP_TIMEOUT
msgBytes = await readMsgBytes(stream, false, deadline) msgBytes = await readMsgBytes(conn, false, deadline)
if msgBytes.len == 0: if msgBytes.len == 0:
await sendErrorResponse(peer, stream, ServerError, readTimeoutErrorMsg) await sendErrorResponse(peer, conn, ServerError, readTimeoutErrorMsg)
return return
type MsgRec = RecType(MsgType) type MsgRec = RecType(MsgType)
@ -572,21 +560,21 @@ proc handleIncomingStream(network: Eth2Node, stream: P2PStream,
try: try:
msg = decode(Format, msgBytes, MsgRec) msg = decode(Format, msgBytes, MsgRec)
except SerializationError as err: except SerializationError as err:
await sendErrorResponse(peer, stream, err, msgName, msgBytes) await sendErrorResponse(peer, conn, err, msgName, msgBytes)
return return
except Exception as err: except Exception as err:
# TODO. This is temporary code that should be removed after interop. # TODO. This is temporary code that should be removed after interop.
# It can be enabled only in certain diagnostic builds where it should # It can be enabled only in certain diagnostic builds where it should
# re-raise the exception. # re-raise the exception.
debug "Crash during serialization", inputBytes = toHex(msgBytes), msgName debug "Crash during serialization", inputBytes = toHex(msgBytes), msgName
await sendErrorResponse(peer, stream, ServerError, err.msg) await sendErrorResponse(peer, conn, ServerError, err.msg)
raise err raise err
try: try:
logReceivedMsg(peer, MsgType(msg)) logReceivedMsg(peer, MsgType(msg))
await callUserHandler(peer, stream, msg) await callUserHandler(peer, conn, msg)
except CatchableError as err: except CatchableError as err:
await sendErrorResponse(peer, stream, ServerError, err.msg) await sendErrorResponse(peer, conn, ServerError, err.msg)
proc handleOutgoingPeer*(peer: Peer): Future[void] {.async.} = proc handleOutgoingPeer*(peer: Peer): Future[void] {.async.} =
let network = peer.network let network = peer.network
@ -727,21 +715,18 @@ proc init*(T: type Peer, network: Eth2Node, info: PeerInfo): Peer =
proc registerMsg(protocol: ProtocolInfo, proc registerMsg(protocol: ProtocolInfo,
name: string, name: string,
mounter: MounterProc, mounter: MounterProc,
libp2pCodecName: string, libp2pCodecName: string) =
printer: MessageContentPrinter) =
protocol.messages.add MessageInfo(name: name, protocol.messages.add MessageInfo(name: name,
protocolMounter: mounter, protocolMounter: mounter,
libp2pCodecName: libp2pCodecName, libp2pCodecName: libp2pCodecName)
printer: printer)
proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend = proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend =
var var
Format = ident "SSZ" Format = ident "SSZ"
Responder = bindSym "Responder" Responder = bindSym "Responder"
P2PStream = bindSym "P2PStream" Connection = bindSym "Connection"
Peer = bindSym "Peer" Peer = bindSym "Peer"
Eth2Node = bindSym "Eth2Node" Eth2Node = bindSym "Eth2Node"
messagePrinter = bindSym "messagePrinter"
registerMsg = bindSym "registerMsg" registerMsg = bindSym "registerMsg"
initProtocol = bindSym "initProtocol" initProtocol = bindSym "initProtocol"
msgVar = ident "msg" msgVar = ident "msg"
@ -761,7 +746,7 @@ proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend =
result.ResponderType = Responder result.ResponderType = Responder
result.afterProtocolInit = proc (p: P2PProtocol) = result.afterProtocolInit = proc (p: P2PProtocol) =
p.onPeerConnected.params.add newIdentDefs(streamVar, P2PStream) p.onPeerConnected.params.add newIdentDefs(streamVar, Connection)
result.implementMsg = proc (msg: Message) = result.implementMsg = proc (msg: Message) =
let let
@ -775,13 +760,13 @@ proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend =
if msg.procDef.body.kind != nnkEmpty and msg.kind == msgRequest: if msg.procDef.body.kind != nnkEmpty and msg.kind == msgRequest:
# Request procs need an extra param - the stream where the response # Request procs need an extra param - the stream where the response
# should be written: # should be written:
msg.userHandler.params.insert(2, newIdentDefs(streamVar, P2PStream)) msg.userHandler.params.insert(2, newIdentDefs(streamVar, Connection))
msg.initResponderCall.add streamVar msg.initResponderCall.add streamVar
## ##
## Implement the Thunk: ## Implement the Thunk:
## ##
## The protocol handlers in nim-libp2p receive only a `P2PStream` ## The protocol handlers in nim-libp2p receive only a `Connection`
## parameter and there is no way to access the wider context (such ## parameter and there is no way to access the wider context (such
## as the current `Switch`). In our handlers, we may need to list all ## as the current `Switch`). In our handlers, we may need to list all
## peers in the current network, so we must keep a reference to the ## peers in the current network, so we must keep a reference to the
@ -799,12 +784,12 @@ proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend =
if msg.userHandler != nil: if msg.userHandler != nil:
protocol.outRecvProcs.add quote do: protocol.outRecvProcs.add quote do:
template `callUserHandler`(`peerVar`: `Peer`, template `callUserHandler`(`peerVar`: `Peer`,
`streamVar`: `P2PStream`, `streamVar`: `Connection`,
`msgVar`: `MsgRecName`): untyped = `msgVar`: `MsgRecName`): untyped =
`userHandlerCall` `userHandlerCall`
proc `protocolMounterName`(`networkVar`: `Eth2Node`) = proc `protocolMounterName`(`networkVar`: `Eth2Node`) =
proc thunk(`streamVar`: `P2PStream`, proc thunk(`streamVar`: `Connection`,
proto: string): Future[void] {.gcsafe.} = proto: string): Future[void] {.gcsafe.} =
return handleIncomingStream(`networkVar`, `streamVar`, return handleIncomingStream(`networkVar`, `streamVar`,
`MsgStrongRecName`, `Format`) `MsgStrongRecName`, `Format`)
@ -830,8 +815,7 @@ proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend =
protocol.protocolInfoVar, protocol.protocolInfoVar,
msgNameLit, msgNameLit,
mounter, mounter,
codecNameLit, codecNameLit))
newTree(nnkBracketExpr, messagePrinter, MsgRecName)))
result.implementProtocolInit = proc (p: P2PProtocol): NimNode = result.implementProtocolInit = proc (p: P2PProtocol): NimNode =
return newCall(initProtocol, newLit(p.name), p.peerInit, p.netInit) return newCall(initProtocol, newLit(p.name), p.peerInit, p.netInit)
@ -903,8 +887,7 @@ proc getPersistentNetKeys*(conf: BeaconNodeConf): KeyPair =
KeyPair(seckey: privKey, pubkey: privKey.getKey()) KeyPair(seckey: privKey, pubkey: privKey.getKey())
proc createEth2Node*(conf: BeaconNodeConf, proc createEth2Node*(conf: BeaconNodeConf): Future[Eth2Node] {.async.} =
bootstrapNodes: seq[ENode]): Future[Eth2Node] {.async.} =
var var
(extIp, extTcpPort, _) = setupNat(conf) (extIp, extTcpPort, _) = setupNat(conf)
hostAddress = tcpEndPoint(conf.libp2pAddress, conf.tcpPort) hostAddress = tcpEndPoint(conf.libp2pAddress, conf.tcpPort)
@ -912,8 +895,7 @@ proc createEth2Node*(conf: BeaconNodeConf,
else: @[tcpEndPoint(extIp, extTcpPort)] else: @[tcpEndPoint(extIp, extTcpPort)]
info "Initializing networking", hostAddress, info "Initializing networking", hostAddress,
announcedAddresses, announcedAddresses
bootstrapNodes
let keys = conf.getPersistentNetKeys let keys = conf.getPersistentNetKeys
# TODO nim-libp2p still doesn't have support for announcing addresses # TODO nim-libp2p still doesn't have support for announcing addresses
@ -924,9 +906,14 @@ proc createEth2Node*(conf: BeaconNodeConf,
result = Eth2Node.init(conf, switch, extIp, keys.seckey.asEthKey) result = Eth2Node.init(conf, switch, extIp, keys.seckey.asEthKey)
proc getPersistenBootstrapAddr*(conf: BeaconNodeConf, proc getPersistenBootstrapAddr*(conf: BeaconNodeConf,
ip: IpAddress, port: Port): ENode = ip: IpAddress, port: Port): enr.Record =
let pair = getPersistentNetKeys(conf) let
initENode(pair.pubkey.skkey, Address(ip: ip, udpPort: port)) pair = getPersistentNetKeys(conf)
enode = initENode(pair.pubkey.skkey, Address(ip: ip, udpPort: port))
return enr.Record.init(1'u64, # sequence number
pair.seckey.asEthKey,
enode.address)
proc shortForm*(id: KeyPair): string = proc shortForm*(id: KeyPair): string =
$PeerID.init(id.pubkey) $PeerID.init(id.pubkey)
@ -938,7 +925,6 @@ proc toPeerInfo(enode: ENode): PeerInfo =
return PeerInfo.init(peerId, addresses) return PeerInfo.init(peerId, addresses)
proc connectToNetwork*(node: Eth2Node, proc connectToNetwork*(node: Eth2Node,
bootstrapNodes: seq[ENode],
bootstrapEnrs: seq[enr.Record]) {.async.} = bootstrapEnrs: seq[enr.Record]) {.async.} =
for bootstrapNode in bootstrapEnrs: for bootstrapNode in bootstrapEnrs:
debug "Adding known peer", peer = bootstrapNode debug "Adding known peer", peer = bootstrapNode
@ -954,10 +940,6 @@ proc connectToNetwork*(node: Eth2Node,
traceAsyncErrors checkIfConnectedToBootstrapNode() traceAsyncErrors checkIfConnectedToBootstrapNode()
proc saveConnectionAddressFile*(node: Eth2Node, filename: string) =
writeFile(filename, $node.switch.peerInfo.addrs[0] & "/p2p/" &
node.switch.peerInfo.id)
func peersCount*(node: Eth2Node): int = func peersCount*(node: Eth2Node): int =
len(node.peerPool) len(node.peerPool)

6
tests/simulation/start.sh
View File

@ -76,8 +76,8 @@ fi
rm -f beacon_node.log rm -f beacon_node.log
# Delete any leftover address files from a previous session # Delete any leftover address files from a previous session
if [ -f "${MASTER_NODE_ADDRESS_FILE}" ]; then if [ -f "${MASTER_NODE_PID_FILE}" ]; then
rm "${MASTER_NODE_ADDRESS_FILE}" rm "${MASTER_NODE_PID_FILE}"
fi fi
# to allow overriding the program names # to allow overriding the program names
@ -137,7 +137,7 @@ fi
for i in $(seq $MASTER_NODE -1 $TOTAL_USER_NODES); do for i in $(seq $MASTER_NODE -1 $TOTAL_USER_NODES); do
if [[ "$i" != "$MASTER_NODE" && "$USE_MULTITAIL" == "no" ]]; then if [[ "$i" != "$MASTER_NODE" && "$USE_MULTITAIL" == "no" ]]; then
# Wait for the master node to write out its address file # Wait for the master node to write out its address file
while [ ! -f "${MASTER_NODE_ADDRESS_FILE}" ]; do while [ ! -f "${MASTER_NODE_PID_FILE}" ]; do
sleep 0.1 sleep 0.1
done done
fi fi

2
tests/simulation/vars.sh
View File

@ -33,7 +33,7 @@ NETWORK_BOOTSTRAP_FILE="${SIMULATION_DIR}/bootstrap_nodes.txt"
BEACON_NODE_BIN="${SIMULATION_DIR}/beacon_node" BEACON_NODE_BIN="${SIMULATION_DIR}/beacon_node"
BOOTSTRAP_NODE_BIN="${SIMULATION_DIR}/bootstrap_node" BOOTSTRAP_NODE_BIN="${SIMULATION_DIR}/bootstrap_node"
DEPLOY_DEPOSIT_CONTRACT_BIN="${SIMULATION_DIR}/deploy_deposit_contract" DEPLOY_DEPOSIT_CONTRACT_BIN="${SIMULATION_DIR}/deploy_deposit_contract"
MASTER_NODE_ADDRESS_FILE="${SIMULATION_DIR}/node-${MASTER_NODE}/beacon_node.address" MASTER_NODE_PID_FILE="${SIMULATION_DIR}/node-${MASTER_NODE}/beacon_node.pid"
BASE_P2P_PORT=30000 BASE_P2P_PORT=30000
BASE_RPC_PORT=7000 BASE_RPC_PORT=7000

4
tests/simulation/wait_master_node.sh
View File

@ -1,8 +1,8 @@
#!/bin/bash #!/bin/bash
if [ ! -f "${MASTER_NODE_ADDRESS_FILE}" ]; then if [ ! -f "${MASTER_NODE_PID_FILE}" ]; then
echo Waiting for master node... echo Waiting for master node...
while [ ! -f "${MASTER_NODE_ADDRESS_FILE}" ]; do while [ ! -f "${MASTER_NODE_PID_FILE}" ]; do
sleep 0.1 sleep 0.1
done done
fi fi

2
tests/test_peer_connection.nim
View File

@ -35,5 +35,5 @@ asyncTest "connect two nodes":
c2.nat = "none" c2.nat = "none"
var n2 = await createEth2Node(c2) var n2 = await createEth2Node(c2)
await n2.connectToNetwork(bootstrapNodes = @[n1PersistentAddress]) await n2.connectToNetwork(@[n1PersistentAddress])