codex-storage
/
nim-libp2p
mirror of https://github.com/codex-storage/nim-libp2p.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

a few more log fixes (#355)

This commit is contained in:
Jacek Sieka 2020-09-07 14:15:11 +02:00 committed by GitHub
parent c1856fda53
commit 2b72d485a3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 63 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
libp2p/switch.nim
View File

@ -44,7 +44,8 @@ declareCounter(libp2p_failed_dials, "failed dials")
declareCounter(libp2p_failed_upgrade, "peers failed upgrade") declareCounter(libp2p_failed_upgrade, "peers failed upgrade")
type type
NoPubSubException* = object of CatchableError UpgradeFailedError* = object of CatchableError
DialFailedError* = object of CatchableError
ConnEventKind* {.pure.} = enum ConnEventKind* {.pure.} = enum
Connected, # A connection was made and securely upgraded - there may be Connected, # A connection was made and securely upgraded - there may be
@ -111,13 +112,13 @@ proc isConnected*(s: Switch, peerId: PeerID): bool =
proc secure(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} = proc secure(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} =
if s.secureManagers.len <= 0: if s.secureManagers.len <= 0:
raise newException(CatchableError, "No secure managers registered!") raise newException(UpgradeFailedError, "No secure managers registered!")
let manager = await s.ms.select(conn, s.secureManagers.mapIt(it.codec)) let manager = await s.ms.select(conn, s.secureManagers.mapIt(it.codec))
if manager.len == 0: if manager.len == 0:
raise newException(CatchableError, "Unable to negotiate a secure channel!") raise newException(UpgradeFailedError, "Unable to negotiate a secure channel!")
trace "securing connection", codec = manager, conn trace "Securing connection", codec = manager, conn
let secureProtocol = s.secureManagers.filterIt(it.codec == manager) let secureProtocol = s.secureManagers.filterIt(it.codec == manager)
# ms.select should deal with the correctness of this # ms.select should deal with the correctness of this
@ -133,7 +134,7 @@ proc identify(s: Switch, conn: Connection) {.async, gcsafe.} =
let info = await s.identity.identify(conn, conn.peerInfo) let info = await s.identity.identify(conn, conn.peerInfo)
if info.pubKey.isNone and isNil(conn): if info.pubKey.isNone and isNil(conn):
raise newException(CatchableError, raise newException(UpgradeFailedError,
"no public key provided and no existing peer identity found") "no public key provided and no existing peer identity found")
if isNil(conn.peerInfo): if isNil(conn.peerInfo):
@ -151,7 +152,7 @@ proc identify(s: Switch, conn: Connection) {.async, gcsafe.} =
if info.protos.len > 0: if info.protos.len > 0:
conn.peerInfo.protocols = info.protos conn.peerInfo.protocols = info.protos
trace "identify: identified remote peer", conn trace "identified remote peer", conn, peerInfo = shortLog(conn.peerInfo)
proc identify(s: Switch, muxer: Muxer) {.async, gcsafe.} = proc identify(s: Switch, muxer: Muxer) {.async, gcsafe.} =
# new stream for identify # new stream for identify
@ -168,7 +169,7 @@ proc identify(s: Switch, muxer: Muxer) {.async, gcsafe.} =
proc mux(s: Switch, conn: Connection): Future[Muxer] {.async, gcsafe.} = proc mux(s: Switch, conn: Connection): Future[Muxer] {.async, gcsafe.} =
## mux incoming connection ## mux incoming connection
trace "muxing connection", conn trace "Muxing connection", conn
if s.muxers.len == 0: if s.muxers.len == 0:
warn "no muxers registered, skipping upgrade flow", conn warn "no muxers registered, skipping upgrade flow", conn
return return
@ -184,7 +185,8 @@ proc mux(s: Switch, conn: Connection): Future[Muxer] {.async, gcsafe.} =
# install stream handler # install stream handler
muxer.streamHandler = s.streamHandler muxer.streamHandler = s.streamHandler
s.connManager.storeOutgoing(muxer.connection) s.connManager.storeOutgoing(conn)
trace "Storing muxer", conn
s.connManager.storeMuxer(muxer) s.connManager.storeMuxer(muxer)
trace "found a muxer", name = muxerName, conn trace "found a muxer", name = muxerName, conn
@ -193,7 +195,7 @@ proc mux(s: Switch, conn: Connection): Future[Muxer] {.async, gcsafe.} =
let handlerFut = muxer.handle() let handlerFut = muxer.handle()
# store it in muxed connections if we have a peer for it # store it in muxed connections if we have a peer for it
trace "adding muxer for peer", conn trace "Storing muxer with handler", conn
s.connManager.storeMuxer(muxer, handlerFut) # update muxer with handler s.connManager.storeMuxer(muxer, handlerFut) # update muxer with handler
return muxer return muxer
@ -202,20 +204,21 @@ proc disconnect*(s: Switch, peerId: PeerID): Future[void] {.gcsafe.} =
s.connManager.dropPeer(peerId) s.connManager.dropPeer(peerId)
proc upgradeOutgoing(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} = proc upgradeOutgoing(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} =
trace "Upgrading outgoing connection", conn
let sconn = await s.secure(conn) # secure the connection let sconn = await s.secure(conn) # secure the connection
if isNil(sconn): if isNil(sconn):
raise newException(CatchableError, raise newException(UpgradeFailedError,
"unable to secure connection, stopping upgrade") "unable to secure connection, stopping upgrade")
if sconn.peerInfo.isNil: if sconn.peerInfo.isNil:
raise newException(CatchableError, raise newException(UpgradeFailedError,
"current version of nim-libp2p requires that secure protocol negotiates peerid") "current version of nim-libp2p requires that secure protocol negotiates peerid")
trace "upgrading connection", conn
let muxer = await s.mux(sconn) # mux it if possible let muxer = await s.mux(sconn) # mux it if possible
if muxer == nil: if muxer == nil:
# TODO this might be relaxed in the future # TODO this might be relaxed in the future
raise newException(CatchableError, raise newException(UpgradeFailedError,
"a muxer is required for outgoing connections") "a muxer is required for outgoing connections")
try: try:
@ -228,28 +231,26 @@ proc upgradeOutgoing(s: Switch, conn: Connection): Future[Connection] {.async, g
if isNil(sconn.peerInfo): if isNil(sconn.peerInfo):
await sconn.close() await sconn.close()
raise newException(CatchableError, raise newException(UpgradeFailedError,
"unable to identify connection, stopping upgrade") "No peerInfo for connection, stopping upgrade")
trace "successfully upgraded outgoing connection", conn, sconn trace "Upgraded outgoing connection", conn, sconn
return sconn return sconn
proc upgradeIncoming(s: Switch, conn: Connection) {.async, gcsafe.} = proc upgradeIncoming(s: Switch, conn: Connection) {.async, gcsafe.} =
trace "upgrading incoming connection", conn trace "Upgrading incoming connection", conn
let ms = newMultistream() let ms = newMultistream()
# secure incoming connections # secure incoming connections
proc securedHandler (conn: Connection, proc securedHandler (conn: Connection,
proto: string) proto: string)
{.async, gcsafe, closure.} = {.async, gcsafe, closure.} =
var sconn: Connection
trace "Securing connection", conn trace "Securing connection", conn
let secure = s.secureManagers.filterIt(it.codec == proto)[0] let secure = s.secureManagers.filterIt(it.codec == proto)[0]
try: try:
sconn = await secure.secure(conn, false) var sconn = await secure.secure(conn, false)
if isNil(sconn): if isNil(sconn):
return return
@ -266,7 +267,9 @@ proc upgradeIncoming(s: Switch, conn: Connection) {.async, gcsafe.} =
except CancelledError as exc: except CancelledError as exc:
raise exc raise exc
except CatchableError as exc: except CatchableError as exc:
debug "ending secured handler", err = exc.msg, conn debug "Exception in secure handler", err = exc.msg, conn
trace "Ending secured handler", conn
if (await ms.select(conn)): # just handshake if (await ms.select(conn)): # just handshake
# add the secure handlers # add the secure handlers
@ -296,11 +299,12 @@ proc internalConnect(s: Switch,
# This connection should already have been removed from the connection # This connection should already have been removed from the connection
# manager - it's essentially a bug that we end up here - we'll fail # manager - it's essentially a bug that we end up here - we'll fail
# for now, hoping that this will clean themselves up later... # for now, hoping that this will clean themselves up later...
warn "dead connection in connection manager", peerId warn "dead connection in connection manager", conn
await conn.close() await conn.close()
raise newException(CatchableError, "Zombie connection encountered") raise newException(DialFailedError, "Zombie connection encountered")
trace "Reusing existing connection", conn, direction = $conn.dir
trace "Reusing existing connection", conn
return conn return conn
trace "Dialing peer", peerId trace "Dialing peer", peerId
@ -311,10 +315,10 @@ proc internalConnect(s: Switch,
let dialed = try: let dialed = try:
await t.dial(a) await t.dial(a)
except CancelledError as exc: except CancelledError as exc:
trace "dialing canceled", exc = exc.msg, peerId trace "Dialing canceled", exc = exc.msg, peerId
raise exc raise exc
except CatchableError as exc: except CatchableError as exc:
trace "dialing failed", exc = exc.msg, peerId trace "Dialing failed", exc = exc.msg, peerId
libp2p_failed_dials.inc() libp2p_failed_dials.inc()
continue # Try the next address continue # Try the next address
@ -337,7 +341,7 @@ proc internalConnect(s: Switch,
doAssert not isNil(upgraded), "connection died after upgradeOutgoing" doAssert not isNil(upgraded), "connection died after upgradeOutgoing"
conn = upgraded conn = upgraded
trace "dial successful", conn, peerInfo = upgraded.peerInfo trace "Dial successful", conn, peerInfo = conn.peerInfo
break break
finally: finally:
if lock.locked(): if lock.locked():
@ -371,27 +375,25 @@ proc internalConnect(s: Switch,
# All the errors are handled inside `cleanup()` procedure. # All the errors are handled inside `cleanup()` procedure.
asyncSpawn peerCleanup() asyncSpawn peerCleanup()
return conn return conn
proc connect*(s: Switch, peerId: PeerID, addrs: seq[MultiAddress]) {.async.} = proc connect*(s: Switch, peerId: PeerID, addrs: seq[MultiAddress]) {.async.} =
discard await s.internalConnect(peerId, addrs) discard await s.internalConnect(peerId, addrs)
proc negotiateStream(s: Switch, stream: Connection, proto: string): Future[Connection] {.async.} = proc negotiateStream(s: Switch, conn: Connection, proto: string): Future[Connection] {.async.} =
trace "Attempting to select remote", proto = proto, stream trace "Negotiating stream", proto = proto, conn
if not await s.ms.select(conn, proto):
await conn.close()
raise newException(DialFailedError, "Unable to select sub-protocol " & proto)
if not await s.ms.select(stream, proto): return conn
await stream.close()
raise newException(CatchableError, "Unable to select sub-protocol" & proto)
return stream
proc dial*(s: Switch, proc dial*(s: Switch,
peerId: PeerID, peerId: PeerID,
proto: string): Future[Connection] {.async.} = proto: string): Future[Connection] {.async.} =
let stream = await s.connmanager.getMuxedStream(peerId) let stream = await s.connmanager.getMuxedStream(peerId)
if stream.isNil: if stream.isNil:
raise newException(CatchableError, "Couldn't get muxed stream") raise newException(DialFailedError, "Couldn't get muxed stream")
return await s.negotiateStream(stream, proto) return await s.negotiateStream(stream, proto)
@ -413,7 +415,7 @@ proc dial*(s: Switch,
try: try:
if isNil(stream): if isNil(stream):
await conn.close() await conn.close()
raise newException(CatchableError, "Couldn't get muxed stream") raise newException(DialFailedError, "Couldn't get muxed stream")
return await s.negotiateStream(stream, proto) return await s.negotiateStream(stream, proto)
except CancelledError as exc: except CancelledError as exc:
@ -421,7 +423,7 @@ proc dial*(s: Switch,
await cleanup() await cleanup()
raise exc raise exc
except CatchableError as exc: except CatchableError as exc:
trace "error dialing", exc = exc.msg, conn trace "Error dialing", exc = exc.msg, conn
await cleanup() await cleanup()
raise exc raise exc
@ -440,14 +442,16 @@ proc start*(s: Switch): Future[seq[Future[void]]] {.async, gcsafe.} =
trace "starting switch for peer", peerInfo = s.peerInfo trace "starting switch for peer", peerInfo = s.peerInfo
proc handle(conn: Connection): Future[void] {.async, closure, gcsafe.} = proc handle(conn: Connection): Future[void] {.async, closure, gcsafe.} =
trace "Incoming connection", conn
try: try:
await s.upgradeIncoming(conn) # perform upgrade on incoming connection await s.upgradeIncoming(conn) # perform upgrade on incoming connection
except CancelledError as exc: except CancelledError as exc:
raise exc raise exc
except CatchableError as exc: except CatchableError as exc:
trace "Error in connection handler", exc = exc.msg, conn trace "Exception occurred in incoming handler", exc = exc.msg, conn
finally: finally:
await conn.close() await conn.close()
trace "Connection handler done", conn
var startFuts: seq[Future[void]] var startFuts: seq[Future[void]]
for t in s.transports: # for each transport for t in s.transports: # for each transport
@ -457,7 +461,7 @@ proc start*(s: Switch): Future[seq[Future[void]]] {.async, gcsafe.} =
s.peerInfo.addrs[i] = t.ma # update peer's address s.peerInfo.addrs[i] = t.ma # update peer's address
startFuts.add(server) startFuts.add(server)
debug "started libp2p node", peerInfo = s.peerInfo debug "Started libp2p node", peer = s.peerInfo
result = startFuts # listen for incoming connections result = startFuts # listen for incoming connections
proc stop*(s: Switch) {.async.} = proc stop*(s: Switch) {.async.} =
@ -477,13 +481,16 @@ proc stop*(s: Switch) {.async.} =
trace "switch stopped" trace "switch stopped"
proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} = proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} =
if muxer.connection.peerInfo.isNil: let
conn = muxer.connection
if conn.peerInfo.isNil:
warn "This version of nim-libp2p requires secure protocol to negotiate peerid" warn "This version of nim-libp2p requires secure protocol to negotiate peerid"
await muxer.close() await muxer.close()
return return
# store incoming connection # store incoming connection
s.connManager.storeIncoming(muxer.connection) s.connManager.storeIncoming(conn)
# store muxer and muxed connection # store muxer and muxed connection
s.connManager.storeMuxer(muxer) s.connManager.storeMuxer(muxer)
@ -494,10 +501,10 @@ proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} =
# Identify is non-essential, though if it fails, it might indicate that # Identify is non-essential, though if it fails, it might indicate that
# the connection was closed already - this will be picked up by the read # the connection was closed already - this will be picked up by the read
# loop # loop
debug "Could not identify connection", err = exc.msg, muxer debug "Could not identify connection", err = exc.msg, conn
try: try:
let peerId = muxer.connection.peerInfo.peerId let peerId = conn.peerInfo.peerId
proc peerCleanup() {.async.} = proc peerCleanup() {.async.} =
try: try:
@ -507,10 +514,10 @@ proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} =
except CancelledError: except CancelledError:
# This is top-level procedure which will work as separate task, so it # This is top-level procedure which will work as separate task, so it
# do not need to propogate CancelledError. # do not need to propogate CancelledError.
debug "Unexpected cancellation in switch muxer cleanup" debug "Unexpected cancellation in switch muxer cleanup", conn
except CatchableError as exc: except CatchableError as exc:
debug "Unexpected exception in switch muxer cleanup", debug "Unexpected exception in switch muxer cleanup",
errMsg = exc.msg, muxer err = exc.msg, conn
proc peerStartup() {.async.} = proc peerStartup() {.async.} =
try: try:
@ -520,10 +527,10 @@ proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} =
except CancelledError: except CancelledError:
# This is top-level procedure which will work as separate task, so it # This is top-level procedure which will work as separate task, so it
# do not need to propogate CancelledError. # do not need to propogate CancelledError.
debug "Unexpected cancellation in switch muxer startup", muxer debug "Unexpected cancellation in switch muxer startup", conn
except CatchableError as exc: except CatchableError as exc:
debug "Unexpected exception in switch muxer startup", debug "Unexpected exception in switch muxer startup",
errMsg = exc.msg, muxer err = exc.msg, conn
# All the errors are handled inside `peerStartup()` procedure. # All the errors are handled inside `peerStartup()` procedure.
asyncSpawn peerStartup() asyncSpawn peerStartup()
@ -537,7 +544,7 @@ proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} =
except CatchableError as exc: except CatchableError as exc:
await muxer.close() await muxer.close()
libp2p_failed_upgrade.inc() libp2p_failed_upgrade.inc()
trace "exception in muxer handler", exc = exc.msg, muxer trace "Exception in muxer handler", exc = exc.msg, conn
proc newSwitch*(peerInfo: PeerInfo, proc newSwitch*(peerInfo: PeerInfo,
transports: seq[Transport], transports: seq[Transport],
@ -558,17 +565,17 @@ proc newSwitch*(peerInfo: PeerInfo,
) )
let s = result # can't capture result let s = result # can't capture result
result.streamHandler = proc(stream: Connection) {.async, gcsafe.} = result.streamHandler = proc(conn: Connection) {.async, gcsafe.} = # noraises
trace "Incoming muxed connection", conn
try: try:
trace "handling connection for", stream await s.ms.handle(conn) # handle incoming connection
defer:
if not(isNil(stream)):
await stream.close()
await s.ms.handle(stream) # handle incoming connection
except CancelledError as exc: except CancelledError as exc:
raise exc raise exc
except CatchableError as exc: except CatchableError as exc:
trace "exception in stream handler", exc = exc.msg, stream trace "exception in stream handler", exc = exc.msg, conn
finally:
await conn.close()
trace "Muxed connection done", conn
result.mount(identity) result.mount(identity)
for key, val in muxers: for key, val in muxers: