From 3b2a775175c9e578597ff249d29f88ffdf3d6b1a Mon Sep 17 00:00:00 2001 From: Dmitriy Ryajov Date: Wed, 30 Sep 2020 19:21:34 -0600 Subject: [PATCH] wip --- libp2p/switch.nim | 306 ++------------------------- libp2p/transports/tcptransport.nim | 54 ++++- libp2p/transports/transport.nim | 16 +- libp2p/upgrademngrs/muxedupgrade.nim | 77 ++++--- libp2p/upgrademngrs/upgrade.nim | 4 +- libp2p/utils/semaphore.nim | 1 - 6 files changed, 117 insertions(+), 341 deletions(-) diff --git a/libp2p/switch.nim b/libp2p/switch.nim index 15dafaf61..2f7f07c35 100644 --- a/libp2p/switch.nim +++ b/libp2p/switch.nim @@ -45,24 +45,20 @@ logScope: declareCounter(libp2p_dialed_peers, "dialed peers") declareCounter(libp2p_failed_dials, "failed dials") -declareCounter(libp2p_failed_upgrade, "peers failed upgrade") type - UpgradeFailedError* = object of CatchableError - DialFailedError* = object of CatchableError - - Switch* = ref object of RootObj - peerInfo*: PeerInfo - connManager: ConnManager - transports*: seq[Transport] - protocols*: seq[LPProtocol] - muxers*: Table[string, MuxerProvider] - ms*: MultistreamSelect - identity*: Identify - streamHandler*: StreamHandler - secureManagers*: seq[Secure] - dialLock: Table[PeerID, AsyncLock] - acceptFuts: seq[Future[void]] + Switch* = ref object of RootObj + peerInfo*: PeerInfo + connManager: ConnManager + transports*: seq[Transport] + protocols*: seq[LPProtocol] + muxers*: Table[string, MuxerProvider] + ms*: MultistreamSelect + identity*: Identify + streamHandler*: StreamHandler + secureManagers*: seq[Secure] + dialLock: Table[PeerID, AsyncLock] + acceptFuts: seq[Future[void]] proc addConnEventHandler*(s: Switch, handler: ConnEventHandler, @@ -110,163 +106,9 @@ proc secure(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} = return await secureProtocol[0].secure(conn, true) -proc identify(s: Switch, conn: Connection) {.async, gcsafe.} = - ## identify the connection - - if (await s.ms.select(conn, s.identity.codec)): - let info = await s.identity.identify(conn, conn.peerInfo) - - if info.pubKey.isNone and isNil(conn): - raise newException(UpgradeFailedError, - "no public key provided and no existing peer identity found") - - if isNil(conn.peerInfo): - conn.peerInfo = PeerInfo.init(info.pubKey.get()) - - if info.addrs.len > 0: - conn.peerInfo.addrs = info.addrs - - if info.agentVersion.isSome: - conn.peerInfo.agentVersion = info.agentVersion.get() - - if info.protoVersion.isSome: - conn.peerInfo.protoVersion = info.protoVersion.get() - - if info.protos.len > 0: - conn.peerInfo.protocols = info.protos - - trace "identified remote peer", conn, peerInfo = shortLog(conn.peerInfo) - -proc identify(s: Switch, muxer: Muxer) {.async, gcsafe.} = - # new stream for identify - var stream = await muxer.newStream() - if stream == nil: - return - - try: - await s.identify(stream) - finally: - await stream.closeWithEOF() - -proc mux(s: Switch, conn: Connection): Future[Muxer] {.async, gcsafe.} = - ## mux incoming connection - - trace "Muxing connection", conn - if s.muxers.len == 0: - warn "no muxers registered, skipping upgrade flow", conn - return - - let muxerName = await s.ms.select(conn, toSeq(s.muxers.keys())) - if muxerName.len == 0 or muxerName == "na": - debug "no muxer available, early exit", conn - return - - trace "Found a muxer", conn, muxerName - - # create new muxer for connection - let muxer = s.muxers[muxerName].newMuxer(conn) - - # install stream handler - muxer.streamHandler = s.streamHandler - - s.connManager.storeMuxer(muxer) - - # start muxer read loop - the future will complete when loop ends - let handlerFut = muxer.handle() - - # store it in muxed connections if we have a peer for it - s.connManager.storeMuxer(muxer, handlerFut) # update muxer with handler - - return muxer - proc disconnect*(s: Switch, peerId: PeerID): Future[void] {.gcsafe.} = s.connManager.dropPeer(peerId) -proc upgradeOutgoing(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} = - trace "Upgrading outgoing connection", conn - - let sconn = await s.secure(conn) # secure the connection - if isNil(sconn): - raise newException(UpgradeFailedError, - "unable to secure connection, stopping upgrade") - - if sconn.peerInfo.isNil: - raise newException(UpgradeFailedError, - "current version of nim-libp2p requires that secure protocol negotiates peerid") - - let muxer = await s.mux(sconn) # mux it if possible - if muxer == nil: - # TODO this might be relaxed in the future - raise newException(UpgradeFailedError, - "a muxer is required for outgoing connections") - - try: - await s.identify(muxer) - except CatchableError as exc: - # 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 - # loop - debug "Could not identify connection", conn, msg = exc.msg - - if isNil(sconn.peerInfo): - await sconn.close() - raise newException(UpgradeFailedError, - "No peerInfo for connection, stopping upgrade") - - s.connManager.updateConn(conn, sconn) - trace "Upgraded outgoing connection", conn, sconn - - return sconn - -proc upgradeIncoming(s: Switch, incomingConn: Connection) {.async, gcsafe.} = # noraises - trace "Upgrading incoming connection", incomingConn - let ms = newMultistream() - - # secure incoming connections - proc securedHandler (conn: Connection, - proto: string) - {.async, gcsafe, closure.} = - trace "Starting secure handler", conn - let secure = s.secureManagers.filterIt(it.codec == proto)[0] - - try: - var sconn = await secure.secure(conn, false) - if isNil(sconn): - return - - defer: - await sconn.close() - - s.connManager.updateConn(conn, sconn) - - # add the muxer - for muxer in s.muxers.values: - ms.addHandler(muxer.codecs, muxer) - - # handle subsequent secure requests - await ms.handle(sconn) - - except CancelledError as exc: - raise exc - except CatchableError as exc: - debug "Exception in secure handler", msg = exc.msg, conn - - trace "Stopped secure handler", conn - - try: - if (await ms.select(incomingConn)): # just handshake - # add the secure handlers - for k in s.secureManagers: - ms.addHandler(k.codec, securedHandler) - - # handle un-secured connections - # we handshaked above, set this ms handler as active - await ms.handle(incomingConn, active = true) - except CatchableError as exc: - debug "exception upgrading incoming", exc = exc.msg - finally: - await incomingConn.close() - proc internalConnect(s: Switch, peerId: PeerID, addrs: seq[MultiAddress]): Future[Connection] {.async.} = @@ -279,32 +121,13 @@ proc internalConnect(s: Switch, try: await lock.acquire() - # Check if we have a connection already and try to reuse it - conn = await s.connManager.getMuxedStream(peerId) - if conn != nil: - if conn.atEof or conn.closed: - # This connection should already have been removed from the connection - # manager - it's essentially a bug that we end up here - we'll fail - # for now, hoping that this will clean themselves up later... - warn "dead connection in connection manager", conn - await conn.close() - raise newException(DialFailedError, "Zombie connection encountered") - - trace "Reusing existing connection", conn, direction = $conn.dir - - return conn - trace "Dialing peer", peerId for t in s.transports: # for each transport - for a in addrs: # for each address - if t.handles(a): # check if it can dial it + for a in addrs: # for each address + if t.handles(a): # check if we can dial it trace "Dialing address", address = $a, peerId let dialed = try: - # await a connection slot when the total - # connection count is equal to `maxConns` - await s.connManager.trackOutgoingConn( - () => t.dial(a) - ) + await t.dial(a) except CancelledError as exc: trace "Dialing canceled", msg = exc.msg, peerId raise exc @@ -317,21 +140,6 @@ proc internalConnect(s: Switch, dialed.peerInfo = PeerInfo.init(peerId, addrs) libp2p_dialed_peers.inc() - - let upgraded = try: - await s.upgradeOutgoing(dialed) - except CatchableError as exc: - # If we failed to establish the connection through one transport, - # we won't succeeded through another - no use in trying again - await dialed.close() - debug "Upgrade failed", msg = exc.msg, peerId - if exc isnot CancelledError: - libp2p_failed_upgrade.inc() - raise exc - - doAssert not isNil(upgraded), "connection died after upgradeOutgoing" - - conn = upgraded trace "Dial successful", conn, peerInfo = conn.peerInfo break finally: @@ -441,12 +249,7 @@ proc accept(s: Switch, transport: Transport) {.async.} = while transport.running: try: - trace "About to accept incoming connection" - let conn = await s.connManager.trackIncomingConn( - () => transport.accept() - ) - trace "Accepted an incoming connection", conn - asyncSpawn s.upgradeIncoming(conn) # perform upgrade on incoming connection + await transport.accept() except CancelledError as exc: trace "Canceling accept loop" break @@ -490,66 +293,6 @@ proc stop*(s: Switch) {.async.} = trace "Switch stopped" -proc muxerHandler(s: Switch, muxer: Muxer) {.async, gcsafe.} = - let - conn = muxer.connection - - if conn.peerInfo.isNil: - warn "This version of nim-libp2p requires secure protocol to negotiate peerid" - await muxer.close() - return - - # store muxer and muxed connection - s.connManager.storeMuxer(muxer) - - try: - await s.identify(muxer) - except CatchableError as exc: - # 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 - # loop - debug "Could not identify connection", conn, msg = exc.msg - - try: - let peerId = conn.peerInfo.peerId - - proc peerCleanup() {.async.} = - try: - await muxer.connection.join() - await s.connManager.triggerConnEvent( - peerId, ConnEvent(kind: ConnEventKind.Disconnected)) - await s.connManager.triggerPeerEvents(peerId, PeerEvent.Left) - except CatchableError as exc: - # This is top-level procedure which will work as separate task, so it - # do not need to propagate CancelledError and shouldn't leak others - debug "Unexpected exception in switch muxer cleanup", - conn, msg = exc.msg - - proc peerStartup() {.async.} = - try: - await s.connManager.triggerPeerEvents(peerId, PeerEvent.Joined) - await s.connManager.triggerConnEvent(peerId, - ConnEvent(kind: ConnEventKind.Connected, incoming: true)) - except CatchableError as exc: - # This is top-level procedure which will work as separate task, so it - # do not need to propagate CancelledError and shouldn't leak others - debug "Unexpected exception in switch muxer startup", - conn, msg = exc.msg - - # All the errors are handled inside `peerStartup()` procedure. - asyncSpawn peerStartup() - - # All the errors are handled inside `peerCleanup()` procedure. - asyncSpawn peerCleanup() - - except CancelledError as exc: - await muxer.close() - raise exc - except CatchableError as exc: - await muxer.close() - libp2p_failed_upgrade.inc() - trace "Exception in muxer handler", conn, msg = exc.msg - proc newSwitch*(peerInfo: PeerInfo, transports: seq[Transport], identity: Identify, @@ -569,24 +312,7 @@ proc newSwitch*(peerInfo: PeerInfo, muxers: muxers, secureManagers: @secureManagers) - switch.streamHandler = proc(conn: Connection) {.async, gcsafe.} = # noraises - trace "Starting stream handler", conn - try: - await switch.ms.handle(conn) # handle incoming connection - except CancelledError as exc: - raise exc - except CatchableError as exc: - trace "exception in stream handler", conn, msg = exc.msg - finally: - await conn.closeWithEOF() - trace "Stream handler done", conn - switch.mount(identity) - for key, val in muxers: - val.streamHandler = switch.streamHandler - val.muxerHandler = proc(muxer: Muxer): Future[void] = - switch.muxerHandler(muxer) - return switch proc isConnected*(s: Switch, peerInfo: PeerInfo): bool diff --git a/libp2p/transports/tcptransport.nim b/libp2p/transports/tcptransport.nim index 1fbf5a30d..36835b294 100644 --- a/libp2p/transports/tcptransport.nim +++ b/libp2p/transports/tcptransport.nim @@ -7,7 +7,7 @@ ## This file may not be copied, modified, or distributed except according to ## those terms. -import oids, sequtils +import std/[oids, sequtils, sugar] import chronos, chronicles import transport, ../errors, @@ -60,7 +60,7 @@ proc setupTcpTransportTracker(): TcpTransportTracker = proc connHandler*(t: TcpTransport, client: StreamTransport, - initiator: bool): Connection = + initiator: bool): Future[Connection] {.async, gcsafe.} = debug "Handling tcp connection", address = $client.remoteAddress, initiator = initiator, clients = t.clients[initiator].len @@ -88,11 +88,19 @@ proc connHandler*(t: TcpTransport, return conn -proc init*(T: type TcpTransport, - flags: set[ServerFlags] = {}): T = - result = T(flags: flags) +proc init*( + T: type TcpTransport, + flags: set[ServerFlags] = {}, + connMngr: ConnManager, + upgrade: Upgrade): T = - result.initTransport() + let t = T( + flags: flags, + upgrade: upgrade, + connMngr: connMngr) + + t.initTransport() + return t method initTransport*(t: TcpTransport) = t.multicodec = multiCodec("tcp") @@ -100,6 +108,9 @@ method initTransport*(t: TcpTransport) = method start*(t: TcpTransport, ma: MultiAddress) {.async.} = ## listen on the transport + ## + + trace "Starting TCP transport", address = ma await procCall Transport(t).start(ma) t.server = createStreamServer(t.ma, t.flags, t) @@ -113,7 +124,7 @@ method start*(t: TcpTransport, ma: MultiAddress) {.async.} = method stop*(t: TcpTransport) {.async, gcsafe.} = ## stop the transport try: - trace "stopping transport" + trace "Stopping TCP transport" await procCall Transport(t).stop() # call base checkFutures( @@ -143,13 +154,16 @@ method stop*(t: TcpTransport) {.async, gcsafe.} = finally: t.running = false -method accept*(t: TcpTransport): Future[Connection] {.async, gcsafe.} = +method accept*(t: TcpTransport) {.async, gcsafe.} = try: let transp = await t.server.accept() try: # we don't need result connection in this # case as it's added inside connHandler - return t.connHandler(transp, false) + let conn = await t.connMngr.trackIncomingConn( + () => t.connHandler(transp, false) + ) + await t.upgrade.upgradeIncoming(conn) except CancelledError as exc: raise exc except CatchableError as exc: @@ -173,13 +187,33 @@ method accept*(t: TcpTransport): Future[Connection] {.async, gcsafe.} = raise exc method dial*(t: TcpTransport, + peerId: PeerID, address: MultiAddress): Future[Connection] {.async, gcsafe.} = trace "dialing remote peer", address = $address ## dial a peer + + var conn: Connection try: + # Check if we have a connection already and try to reuse it + conn = await t.connMngr.getMuxedStream(peerId) + if conn != nil: + if conn.atEof or conn.closed: + # This connection should already have been removed from the connection + # manager - it's essentially a bug that we end up here - we'll fail + # for now, hoping that this will clean themselves up later... + warn "dead connection in connection manager", conn + await conn.close() + raise newException(DialFailedError, "Zombie connection encountered") + + trace "Reusing existing connection", conn, direction = $conn.dir + return conn + let transp = await connect(address) - return t.connHandler(transp, true) + conn = await t.connMngr.trackOutgoingConn( + () => t.connHandler(transp, initiator = true) + ) + return await t.upgrade.upgradeOutgoing(conn) except TransportTooManyError as exc: warn "dial: could not create new connection, too many files opened" raise exc diff --git a/libp2p/transports/transport.nim b/libp2p/transports/transport.nim index 507d8aaad..40bdeee02 100644 --- a/libp2p/transports/transport.nim +++ b/libp2p/transports/transport.nim @@ -10,14 +10,22 @@ import sequtils import chronos, chronicles -import ../stream/connection, +import ../upgrademngrs/upgrade, + ../stream/connection, ../multiaddress, - ../multicodec + ../multicodec, + ../connmanager + +export multicodec, connection, upgrade, connmanager, chronos type + DialFailedError* = object of CatchableError + Transport* = ref object of RootObj ma*: Multiaddress multicodec*: MultiCodec + connMngr*: ConnManager + upgrade*: Upgrade running*: bool method initTransport*(t: Transport) {.base, gcsafe, locks: "unknown".} = @@ -38,14 +46,14 @@ method stop*(t: Transport) {.base, async.} = ## including all outstanding connections discard -method accept*(t: Transport): Future[Connection] - {.base, async, gcsafe.} = +method accept*(t: Transport) {.base, async.} = ## accept incoming connections ## discard method dial*(t: Transport, + peerId: PeerID, address: MultiAddress): Future[Connection] {.base, async, gcsafe.} = ## dial a peer diff --git a/libp2p/upgrademngrs/muxedupgrade.nim b/libp2p/upgrademngrs/muxedupgrade.nim index c6da37484..39450f6e4 100644 --- a/libp2p/upgrademngrs/muxedupgrade.nim +++ b/libp2p/upgrademngrs/muxedupgrade.nim @@ -46,6 +46,11 @@ proc init*( await conn.close() trace "Stream handler done", conn + for key, val in muxers: + val.streamHandler = upgrader.streamHandler + val.muxerHandler = proc(muxer: Muxer): Future[void] = + upgrader.muxerHandler(muxer) + return upgrader proc mux(u: MuxedUpgrade, conn: Connection): Future[Muxer] {.async, gcsafe.} = @@ -155,43 +160,47 @@ proc muxerHandler(u: MuxedUpgrade, muxer: Muxer) {.async, gcsafe.} = libp2p_failed_upgrade.inc() trace "Exception in muxer handler", conn, msg = exc.msg -proc upgradeOutgoing(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} = - trace "Upgrading outgoing connection", conn - - let sconn = await s.secure(conn) # secure the connection - if isNil(sconn): - raise newException(UpgradeFailedError, - "unable to secure connection, stopping upgrade") - - if sconn.peerInfo.isNil: - raise newException(UpgradeFailedError, - "current version of nim-libp2p requires that secure protocol negotiates peerid") - - let muxer = await s.mux(sconn) # mux it if possible - if muxer == nil: - # TODO this might be relaxed in the future - raise newException(UpgradeFailedError, - "a muxer is required for outgoing connections") - +method upgradeOutgoing(s: Switch, conn: Connection): Future[Connection] {.async, gcsafe.} = try: - await s.identify(muxer) + trace "Upgrading outgoing connection", conn + + let sconn = await s.secure(conn) # secure the connection + if isNil(sconn): + raise newException(UpgradeFailedError, + "unable to secure connection, stopping upgrade") + + if sconn.peerInfo.isNil: + raise newException(UpgradeFailedError, + "current version of nim-libp2p requires that secure protocol negotiates peerid") + + let muxer = await s.mux(sconn) # mux it if possible + if muxer == nil: + # TODO this might be relaxed in the future + raise newException(UpgradeFailedError, + "a muxer is required for outgoing connections") + + try: + await s.identify(muxer) + except CatchableError as exc: + # 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 + # loop + debug "Could not identify connection", conn, msg = exc.msg + + if isNil(sconn.peerInfo): + await sconn.close() + raise newException(UpgradeFailedError, + "No peerInfo for connection, stopping upgrade") + + s.connManager.updateConn(conn, sconn) + trace "Upgraded outgoing connection", conn, sconn + + return sconn except CatchableError as exc: - # 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 - # loop - debug "Could not identify connection", conn, msg = exc.msg + if not isNil(conn): + await conn.close() - if isNil(sconn.peerInfo): - await sconn.close() - raise newException(UpgradeFailedError, - "No peerInfo for connection, stopping upgrade") - - s.connManager.updateConn(conn, sconn) - trace "Upgraded outgoing connection", conn, sconn - - return sconn - -proc upgradeIncoming(s: Switch, incomingConn: Connection) {.async, gcsafe.} = # noraises +method upgradeIncoming(s: Switch, incomingConn: Connection) {.async, gcsafe.} = # noraises trace "Upgrading incoming connection", incomingConn let ms = newMultistream() diff --git a/libp2p/upgrademngrs/upgrade.nim b/libp2p/upgrademngrs/upgrade.nim index ec8b204f6..c72ce3a9b 100644 --- a/libp2p/upgrademngrs/upgrade.nim +++ b/libp2p/upgrademngrs/upgrade.nim @@ -32,10 +32,10 @@ type connManager*: ConnManager secureManagers*: seq[Secure] -method upgradeIncoming*(u: Upgrade, conn: Connection): Future[void] = +method upgradeIncoming*(u: Upgrade, conn: Connection): Future[void] {.base.} = doAssert(false, "Not implemented!") -method upgradeOutgoing*(u: Upgrade, conn: Connection): Future[void] = +method upgradeOutgoing*(u: Upgrade, conn: Connection): Future[Connection] {.base.} = doAssert(false, "Not implemented!") proc secure*(u: Upgrade, conn: Connection): Future[Connection] {.async, gcsafe.} = diff --git a/libp2p/utils/semaphore.nim b/libp2p/utils/semaphore.nim index f75699185..bd56ce0d5 100644 --- a/libp2p/utils/semaphore.nim +++ b/libp2p/utils/semaphore.nim @@ -91,7 +91,6 @@ proc release*(s: AsyncSemaphore) = when isMainModule: import unittest - import chronos suite "AsyncSemaphore": test "should acquire":