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.

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

libp2p lifecycle hooks and pubsub monitor #1360

Closed
wants to merge 14 commits into from
34 changes: 27 additions & 7 deletions beacon_chain/eth2_network.nim
Original file line number Diff line number Diff line change
@@ -93,6 +93,7 @@ type
netThroughput: AverageThroughput
score*: int
lacksSnappy: bool
pubsubFut*: Future[void]

ConnectionState* = enum
None,
@@ -216,9 +217,9 @@ const
## Period of time for dead peers.
SeenTableTimeIrrelevantNetwork* = 24.hours
## Period of time for `IrrelevantNetwork` error reason.
SeenTableTimeClientShutDown* = 10.minutes
SeenTableTimeClientShutDown* = 1.minutes
## Period of time for `ClientShutDown` error reason.
SeemTableTimeFaultOrError* = 10.minutes
SeemTableTimeFaultOrError* = 1.minutes
## Period of time for `FaultOnError` error reason.

var successfullyDialledAPeer = false # used to show a warning
@@ -826,7 +827,7 @@ proc runDiscoveryLoop*(node: Eth2Node) {.async.} =
if peerRecord.isOk:
let peerInfo = peerRecord.value.toPeerInfo
if peerInfo != nil:
if peerInfo.id notin node.switch.connections:
if not node.switch.isConnected(peerInfo):
await node.connQueue.addLast(peerInfo)
else:
peerInfo.close()
@@ -855,8 +856,8 @@ proc init*(T: type Eth2Node, conf: BeaconNodeConf, enrForkId: ENRForkID,
result.switch = switch
result.wantedPeers = conf.maxPeers
result.peerPool = newPeerPool[Peer, PeerID](maxPeers = conf.maxPeers)
result.connectTimeout = 10.seconds
result.seenThreshold = 10.minutes
result.connectTimeout = 1.minutes
result.seenThreshold = 1.minutes
result.seenTable = initTable[PeerID, SeenItem]()
result.connTable = initTable[PeerID, PeerInfo]()
result.connQueue = newAsyncQueue[PeerInfo](ConcurrentConnections)
@@ -1165,10 +1166,29 @@ proc createEth2Node*(rng: ref BrHmacDrbgContext, conf: BeaconNodeConf, enrForkId
SecureProtocol.Noise, # Only noise in ETH2!
],
rng = rng)
result = Eth2Node.init(conf, enrForkId, switch,

let node = Eth2Node.init(conf, enrForkId, switch,
extIp, extTcpPort, extUdpPort,
keys.seckey.asEthKey, rng = rng)

# switch.addHook(
# proc(peer: PeerInfo, cycle: Lifecycle) {.async.} =
# ## trigger every time a new connection
# ## for a peer is upgradded
# ##

# doAssert(cycle == Lifecycle.Upgraded)
# doAssert(isNil(peer) == false)

# let ethPeer = node.getPeer(peer)
# if not(isNil(ethPeer)) and isNil(ethPeer.pubsubFut):
# ethPeer.pubsubFut = switch.subscribePeer(peer)

# , Lifecycle.Upgraded
# )

return node

proc getPersistenBootstrapAddr*(rng: var BrHmacDrbgContext, conf: BeaconNodeConf,
ip: ValidIpAddress, port: Port): EnrResult[enr.Record] =
let pair = getPersistentNetKeys(rng, conf)
@@ -1243,7 +1263,7 @@ proc broadcast*(node: Eth2Node, topic: string, msg: auto) =
inc nbc_gossip_messages_sent
let
data = snappy.encode(SSZ.encode(msg))
var futSnappy = node.switch.publish(topic & "_snappy", data)
var futSnappy = node.switch.publish(topic & "_snappy", data, 1.minutes)
traceMessage(futSnappy, gossipId(data))

# TODO:
2 changes: 1 addition & 1 deletion beacon_chain/inspector.nim
Original file line number Diff line number Diff line change
@@ -570,7 +570,7 @@ proc discoveryLoop(conf: InspectorConf,
if pinfoOpt.isOk():
let pinfo = pinfoOpt.get()
if pinfo.hasTCP():
if pinfo.id() notin switch.connections:
if not switch.isConnected(pinfo):
debug "Discovered new peer", peer = pinfo,
peers_count = len(peers)
await connQueue.addLast(pinfo)
3 changes: 2 additions & 1 deletion beacon_chain/sync_protocol.nim
Original file line number Diff line number Diff line change
@@ -226,7 +226,8 @@ proc handleStatus(peer: Peer,

if not res:
debug "Peer is dead or already in pool", peer
await peer.disconnect(ClientShutDown)
# TODO: DON NOT DROP THE PEER!
# await peer.disconnect(ClientShutDown)

peer.setStatusMsg(theirStatus)

2 changes: 1 addition & 1 deletion vendor/nim-libp2p