Skip to content

use libp2p peer events to track peer #1468

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.

Sign up for GitHub

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

Merged
arnetheduck merged 7 commits into from
Aug 10, 2020
Merged

use libp2p peer events to track peer #1468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d88f914
use libp2p peer events to track peer
arnetheduck Aug 7, 2020
d89436a
naming
arnetheduck Aug 7, 2020
90b8cb9
naming
arnetheduck Aug 8, 2020
3d3a7e5
bump
arnetheduck Aug 8, 2020
03c6452
bump
arnetheduck Aug 8, 2020
cd39273
bump
arnetheduck Aug 8, 2020
5e97d45
add todo
arnetheduck Aug 10, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion beacon_chain/beacon_node.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -729,7 +729,6 @@ proc installDebugApiHandlers(rpcServer: RpcServer, node: BeaconNode) =
peers.add(
%(
info: shortLog(peer.info),
wasDialed: peer.wasDialed,
connectionState: $peer.connectionState,
score: peer.score,
)
Expand Down
100 changes: 74 additions & 26 deletions beacon_chain/eth2_network.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ type
connTable: HashSet[PeerID]
forkId: ENRForkID
rng*: ref BrHmacDrbgContext
peers: Table[PeerID, Peer]

EthereumNode = Eth2Node # needed for the definitions in p2p_backends_helpers

Expand All @@ -85,14 +86,15 @@ type
Peer* = ref object
network*: Eth2Node
info*: PeerInfo
wasDialed*: bool
discoveryId*: Eth2DiscoveryId
connectionState*: ConnectionState
protocolStates*: seq[RootRef]
maxInactivityAllowed*: Duration
netThroughput: AverageThroughput
score*: int
lacksSnappy: bool
connections*: int
disconnectedFut: Future[void]

PeerAddr* = object
peerId*: PeerID
Expand Down Expand Up @@ -148,7 +150,7 @@ type

PeerStateInitializer* = proc(peer: Peer): RootRef {.gcsafe.}
NetworkStateInitializer* = proc(network: EthereumNode): RootRef {.gcsafe.}
OnPeerConnectedHandler* = proc(peer: Peer, conn: Connection): Future[void] {.gcsafe.}
OnPeerConnectedHandler* = proc(peer: Peer, incoming: bool): Future[void] {.gcsafe.}
OnPeerDisconnectedHandler* = proc(peer: Peer): Future[void] {.gcsafe.}
ThunkProc* = LPProtoHandler
MounterProc* = proc(network: Eth2Node) {.gcsafe.}
Expand Down Expand Up @@ -295,10 +297,11 @@ proc openStream(node: Eth2Node,
proc init*(T: type Peer, network: Eth2Node, info: PeerInfo): Peer {.gcsafe.}

proc getPeer*(node: Eth2Node, peerId: PeerID): Peer {.gcsafe.} =
result = node.peerPool.getOrDefault(peerId)
if result == nil:
# TODO: We should register this peer in the pool!
result = Peer.init(node, PeerInfo.init(peerId))
node.peers.withValue(peerId, peer) do:

@zah zah Aug 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no benefit of using withValue here. It will only lead to slightly increased code size because it's a template.

A more efficient implementation of the operation needed here is possible right now, but it's slightly awkward:

template mgetOrPutLazy*[A, B](t: Table[A, B], key: A, val: B): var B =
  type R = B

  proc setter(loc: var R): var R =
    if loc == default(R):
      loc = val
    loc

  setter(mgetOrPut(t, key, default(R)))

@zah zah Aug 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Relevant PR in Stew:
status-im/nim-stew#52

@arnetheduck arnetheduck Aug 7, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm mostly interested in have one pattern to use across the board without having to think too much about nuance - withValue comes close to offering a particular set of properties I'm often interested in:

  • no exceptions - it's a table, it's expected that the value is missing sometimes, not exceptional - I don't want to pay a hefty exception tax (specially since nim is mostly exception-unsafe)
  • doesn't require double lookups
  • doesn't require a valid value up-front if it's missing from table
  • maintains these tradeoffs for different types

mgetorputlazy looks pretty complex - there's a closure function, a template, defaults etc - doesn't really look like it'll be that much less bloat.

I dislike withValue for several reasons:

  • requires var - no read-only version
  • is indeed a template - though if it was well made, it wouldn't incur any overhead over an optimal version - I guess it's not well made, judging from your comment
  • introduces a pointer, which is annoying / breaks some safety features such as escapes

mgetorputlazy doesn't seem to hit the sweet spot of a single tool with which I can forget about the other tools (I really don't want to use a different tool just because here I happen to be dealing with a ref whose nil value is free and therefore yadayada - I'm too old to remember so many details)

@arnetheduck arnetheduck Aug 8, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah, one more thing about withValue:

  • the template doesn't have return value - thus it can't be used in expressions let x = y.withValue(xx, yy): do: yy[].field; do: ..

return peer[]
do:
let peer = Peer.init(node, PeerInfo.init(peerId))
return node.peers.mGetOrPut(peerId, peer)

proc peerFromStream(network: Eth2Node, conn: Connection): Peer {.gcsafe.} =
# TODO: Can this be `nil`?
Expand All @@ -308,7 +311,9 @@ proc getKey*(peer: Peer): PeerID {.inline.} =
result = peer.info.peerId

proc getFuture*(peer: Peer): Future[void] {.inline.} =
result = peer.info.lifeFuture()
if peer.disconnectedFut.isNil:
peer.disconnectedFut = newFuture[void]()
result = peer.disconnectedFut

proc getScore*(a: Peer): int =
## Returns current score value for peer ``peer``.
Expand Down Expand Up @@ -390,7 +395,6 @@ proc disconnect*(peer: Peer, reason: DisconnectionReason,
of FaultOrError:
SeemTableTimeFaultOrError
peer.network.addSeen(peer.info.peerId, seenTime)
peer.info.close()

include eth/p2p/p2p_backends_helpers
include eth/p2p/p2p_tracing
Expand Down Expand Up @@ -537,13 +541,12 @@ template send*[M](r: SingleChunkResponse[M], val: auto): untyped =
doAssert UntypedResponse(r).writtenChunks == 0
sendResponseChunkObj(UntypedResponse(r), val)

proc performProtocolHandshakes*(peer: Peer) {.async.} =
var subProtocolsHandshakes = newSeqOfCap[Future[void]](allProtocols.len)
proc performProtocolHandshakes*(peer: Peer, incoming: bool) {.async.} =
# Loop down serially because it's easier to reason about the connection state
# when there are fewer async races, specially during setup
for protocol in allProtocols:

@zah zah Aug 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The code here was prepared for a future where more protocols are added. If you deem this unnecessary, you need to also remove the for loop above.

@arnetheduck arnetheduck Aug 7, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it still supports more than one, just runs the initializers serially in case one fails, which is easier to reason about

@arnetheduck arnetheduck Aug 7, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

basically, when libp2p perform operations, it mutates the state of any connection object they use (for example closing it or setting fields to different values - if they are run in concurrently, there's no locking meaning the state might change in the middle of an operation, causing async races which few understand - it's like threading, without the parallelism.

if protocol.onPeerConnected != nil:
subProtocolsHandshakes.add protocol.onPeerConnected(peer, nil)

await allFuturesThrowing(subProtocolsHandshakes)
await protocol.onPeerConnected(peer, incoming)

proc initProtocol(name: string,
peerInit: PeerStateInitializer,
Expand Down Expand Up @@ -688,7 +691,7 @@ proc handleIncomingStream(network: Eth2Node,
finally:
await safeClose(conn)

proc handleOutgoingPeer*(peer: Peer): Future[bool] {.async.} =
proc handleOutgoingPeer(peer: Peer): Future[bool] {.async.} =
let network = peer.network

proc onPeerClosed(udata: pointer) {.gcsafe.} =
Expand All @@ -704,7 +707,7 @@ proc handleOutgoingPeer*(peer: Peer): Future[bool] {.async.} =

nbc_peers.set int64(len(network.peerPool))

proc handleIncomingPeer*(peer: Peer): Future[bool] {.async.} =
proc handleIncomingPeer(peer: Peer): Future[bool] {.async.} =
let network = peer.network

proc onPeerClosed(udata: pointer) {.gcsafe.} =
Expand Down Expand Up @@ -754,18 +757,18 @@ proc dialPeer*(node: Eth2Node, peerAddr: PeerAddr) {.async.} =
logScope: peer = peerAddr.peerId

debug "Connecting to discovered peer"

# TODO connect is called here, but there's no guarantee that the connection
# we get when using dialPeer later on is the one we just connected
let peer = node.getPeer(peerAddr.peerId)

await node.switch.connect(peerAddr.peerId, peerAddr.addrs)
var peer = node.getPeer(peerAddr.peerId)
peer.wasDialed = true

#let msDial = newMultistream()
#let conn = node.switch.connections.getOrDefault(peerInfo.id)
#let ls = await msDial.list(conn)
#debug "Supported protocols", ls

debug "Initializing connection"
await performProtocolHandshakes(peer)

inc nbc_successful_dials
successfullyDialledAPeer = true
debug "Network handshakes completed"
Expand Down Expand Up @@ -850,6 +853,50 @@ proc getPersistentNetMetadata*(conf: BeaconNodeConf): Eth2Metadata =
else:
result = Json.loadFile(metadataPath, Eth2Metadata)

proc onConnEvent(node: Eth2Node, peerId: PeerID, event: ConnEvent) {.async.} =
let peer = node.getPeer(peerId)
case event.kind
of ConnEventKind.Connected:
inc peer.connections
debug "Peer upgraded", peer = peerId, connections = peer.connections

if peer.connections == 1:
# Libp2p may connect multiple times to the same peer - using different
# transports or both incoming and outgoing. For now, we'll count our
# "fist" encounter with the peer as the true connection, leaving the
# other connections be - libp2p limits the number of concurrent
# connections to the same peer, and only one of these connections will be
# active. Nonetheless, this quirk will cause a number of odd behaviours:
# * For peer limits, we might miscount the incoming vs outgoing quota
# * Protocol handshakes are wonky: we'll not necessarily use the newly
# connected transport - instead we'll just pick a random one!
await performProtocolHandshakes(peer, event.incoming)

# While performing the handshake, the peer might have been disconnected -
# there's still a slim chance of a race condition here if a reconnect
# happens quickly
if peer.connections == 1:

# TODO when the pool is full, adding it will block - this means peers
# will be left in limbo until some other peer makes room for it
let added = if event.incoming:
await handleIncomingPeer(peer)
else:
await handleOutgoingPeer(peer)

if not added:
# We must have hit a limit!
await peer.disconnect(FaultOrError)

of ConnEventKind.Disconnected:
dec peer.connections
debug "Peer disconnected", peer = peerId, connections = peer.connections
if peer.connections == 0:
let fut = peer.disconnectedFut
if fut != nil:
peer.disconnectedFut = nil
fut.complete()

proc init*(T: type Eth2Node, conf: BeaconNodeConf, enrForkId: ENRForkID,
switch: Switch, ip: Option[ValidIpAddress], tcpPort, udpPort: Port,
privKey: keys.PrivateKey, rng: ref BrHmacDrbgContext): T =
Expand Down Expand Up @@ -878,12 +925,16 @@ proc init*(T: type Eth2Node, conf: BeaconNodeConf, enrForkId: ENRForkID,
if msg.protocolMounter != nil:
msg.protocolMounter result

let node = result
proc peerHook(peerId: PeerID, event: ConnEvent): Future[void] {.gcsafe.} =
onConnEvent(node, peerId, event)

switch.addConnEventHandler(peerHook, ConnEventKind.Connected)
switch.addConnEventHandler(peerHook, ConnEventKind.Disconnected)

template publicKey*(node: Eth2Node): keys.PublicKey =
node.discovery.privKey.toPublicKey

template addKnownPeer*(node: Eth2Node, peer: enr.Record) =
node.discovery.addNode peer

proc startListening*(node: Eth2Node) {.async.} =
node.discovery.open()
node.libp2pTransportLoops = await node.switch.start()
Expand Down Expand Up @@ -956,9 +1007,6 @@ proc p2pProtocolBackendImpl*(p: P2PProtocol): Backend =
result.SerializationFormat = Format
result.RequestResultsWrapper = ident "NetRes"

result.afterProtocolInit = proc (p: P2PProtocol) =
p.onPeerConnected.params.add newIdentDefs(streamVar, Connection)

result.implementMsg = proc (msg: p2p_protocol_dsl.Message) =
if msg.kind == msgResponse:
return
Expand Down
59 changes: 24 additions & 35 deletions beacon_chain/sync_protocol.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@ type
forkDigest*: ForkDigest

BeaconSyncPeerState* = ref object
initialStatusReceived*: bool
statusMsg*: StatusMsg

BlockRootSlot* = object
Expand Down Expand Up @@ -89,21 +88,32 @@ p2pProtocol BeaconSync(version = 1,
networkState = BeaconSyncNetworkState,
peerState = BeaconSyncPeerState):

onPeerConnected do (peer: Peer) {.async.}:
onPeerConnected do (peer: Peer, incoming: bool) {.async.}:
debug "Peer connected",
peer, peerInfo = shortLog(peer.info), wasDialed = peer.wasDialed
if peer.wasDialed:
let
ourStatus = peer.networkState.getCurrentStatus()
# TODO: The timeout here is so high only because we fail to
# respond in time due to high CPU load in our single thread.
theirStatus = await peer.status(ourStatus, timeout = 60.seconds)
peer, peerInfo = shortLog(peer.info), incoming
# Per the eth2 protocol, whoever dials must send a status message when
# connected for the first time, but because of how libp2p works, there may
# be a race between incoming and outgoing connections and disconnects that
# makes the incoming flag unreliable / obsolete by the time we get to
# this point - instead of making assumptions, we'll just send a status
# message redundantly.
# TODO the spec does not prohibit sending the extra status message on
# incoming connections, but it should not be necessary - this would
# need a dedicated flow in libp2p that resolves the race conditions -
# this needs more thinking around the ordering of events and the
# given incoming flag
let

@zah zah Aug 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, due of a rare race condition we choose to always violate the spec?

@arnetheduck arnetheduck Aug 7, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not so sure about rare, I've been encountering and reporting it for months now, and it causes peer pool miscounts - I'm not sure I understand your position - are you suggesting we should occasionally allow this to happen?

the spec mandates that the initiating connection sends a status message but does not place limits on the other end - it's free to do as it likes and send as many status messages as it wants (in theory), including this one.

@zah zah Aug 10, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, let's merge this now as it fixes a known issue, but I think we can do better in the long term, so some TODO comments and future directions will be appreciated.

ourStatus = peer.networkState.getCurrentStatus()
# TODO: The timeout here is so high only because we fail to
# respond in time due to high CPU load in our single thread.
theirStatus = await peer.status(ourStatus, timeout = 60.seconds)

if theirStatus.isOk:
await peer.handleStatus(peer.networkState,
ourStatus, theirStatus.get())
else:
warn "Status response not received in time", peer
if theirStatus.isOk:
await peer.handleStatus(peer.networkState,
ourStatus, theirStatus.get())
else:
warn "Status response not received in time",
peer, error = theirStatus.error

proc status(peer: Peer,
theirStatus: StatusMsg,
Expand Down Expand Up @@ -178,7 +188,6 @@ p2pProtocol BeaconSync(version = 1,

proc setStatusMsg(peer: Peer, statusMsg: StatusMsg) =
debug "Peer status", peer, statusMsg
peer.state(BeaconSync).initialStatusReceived = true
peer.state(BeaconSync).statusMsg = statusMsg

proc updateStatus*(peer: Peer): Future[bool] {.async.} =
Expand All @@ -197,10 +206,6 @@ proc updateStatus*(peer: Peer): Future[bool] {.async.} =
peer.setStatusMsg(theirStatus.get)
result = true

proc hasInitialStatus*(peer: Peer): bool {.inline.} =
## Returns head slot for specific peer ``peer``.
peer.state(BeaconSync).initialStatusReceived

proc getHeadSlot*(peer: Peer): Slot {.inline.} =
## Returns head slot for specific peer ``peer``.
result = peer.state(BeaconSync).statusMsg.headSlot
Expand All @@ -213,22 +218,6 @@ proc handleStatus(peer: Peer,
notice "Irrelevant peer", peer, theirStatus, ourStatus
await peer.disconnect(IrrelevantNetwork)
else:
if not peer.state(BeaconSync).initialStatusReceived:
# Initial/handshake status message handling
peer.state(BeaconSync).initialStatusReceived = true
debug "Peer connected", peer, ourStatus = shortLog(ourStatus),
theirStatus = shortLog(theirStatus)
var res: bool
if peer.wasDialed:
res = await handleOutgoingPeer(peer)
else:
res = await handleIncomingPeer(peer)

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

peer.setStatusMsg(theirStatus)

proc initBeaconSync*(network: Eth2Node, chainDag: ChainDAGRef,
Expand Down
22 changes: 10 additions & 12 deletions beacon_chain/sync_protocol.nim.generated.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@

## Generated at line 88
## Generated at line 87
type
BeaconSync* = object
template State*(PROTO: type BeaconSync): type =
Expand Down Expand Up @@ -365,7 +365,7 @@ registerMsg(BeaconSyncProtocol, "beaconBlocksByRoot", beaconBlocksByRootMounter,
"/eth2/beacon_chain/req/beacon_blocks_by_root/1/")
registerMsg(BeaconSyncProtocol, "goodbye", goodbyeMounter,
"/eth2/beacon_chain/req/goodbye/1/")
proc BeaconSyncPeerConnected(peer: Peer; stream: Connection) {.async, gcsafe.} =
proc BeaconSyncPeerConnected(peer: Peer; incoming: bool) {.async, gcsafe.} =
type
CurrentProtocol = BeaconSync
template state(peer: Peer): ref[BeaconSyncPeerState:ObjectType] =
Expand All @@ -375,16 +375,14 @@ proc BeaconSyncPeerConnected(peer: Peer; stream: Connection) {.async, gcsafe.} =
cast[ref[BeaconSyncNetworkState:ObjectType]](getNetworkState(peer.network,
BeaconSyncProtocol))

debug "Peer connected", peer, peerInfo = shortLog(peer.info),
wasDialed = peer.wasDialed
if peer.wasDialed:
let
ourStatus = peer.networkState.getCurrentStatus()
theirStatus = await peer.status(ourStatus, timeout = 60.seconds)
if theirStatus.isOk:
await peer.handleStatus(peer.networkState, ourStatus, theirStatus.get())
else:
warn "Status response not received in time", peer
debug "Peer connected", peer, peerInfo = shortLog(peer.info), incoming
let
ourStatus = peer.networkState.getCurrentStatus()
theirStatus = await peer.status(ourStatus, timeout = 60.seconds)
if theirStatus.isOk:
await peer.handleStatus(peer.networkState, ourStatus, theirStatus.get())
else:
warn "Status response not received in time", peer, error = theirStatus.error

setEventHandlers(BeaconSyncProtocol, BeaconSyncPeerConnected, nil)
registerProtocol(BeaconSyncProtocol)
2 changes: 1 addition & 1 deletion vendor/nim-libp2p
Open in desktop
Submodule nim-libp2p updated from 7c2ab3 to 6ffd5b
2 changes: 1 addition & 1 deletion vendor/nim-stew
Open in desktop
Submodule nim-stew updated 3 files
+45 −0 stew/ctops.nim
+2 −1 tests/all_tests.nim
+105 −0 tests/test_ctops.nim