nim-libp2p/libp2p/dialer.nim

347 lines
10 KiB
Nim
Raw Normal View History

2022-07-01 18:19:57 +00:00
# Nim-LibP2P
2023-01-20 14:47:40 +00:00
# Copyright (c) 2023 Status Research & Development GmbH
2022-07-01 18:19:57 +00:00
# Licensed under either of
# * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
# * MIT license ([LICENSE-MIT](LICENSE-MIT))
# at your option.
# This file may not be copied, modified, or distributed except according to
# those terms.
import std/tables
import stew/results
import pkg/[chronos,
chronicles,
metrics]
import dial,
peerid,
peerinfo,
2023-03-08 11:30:19 +00:00
peerstore,
2022-09-26 09:03:24 +00:00
multicodec,
2023-03-08 11:30:19 +00:00
muxers/muxer,
multistream,
connmanager,
stream/connection,
2021-05-24 17:55:33 +00:00
transports/transport,
nameresolving/nameresolver,
2022-08-23 15:49:07 +00:00
upgrademngrs/upgrade,
2021-05-24 17:55:33 +00:00
errors
export dial, errors, results
logScope:
topics = "libp2p dialer"
declareCounter(libp2p_total_dial_attempts, "total attempted dials")
declareCounter(libp2p_successful_dials, "dialed successful peers")
declareCounter(libp2p_failed_dials, "failed dials")
type
2021-05-24 17:55:33 +00:00
DialFailedError* = object of LPError
Dialer* = ref object of Dial
localPeerId*: PeerId
connManager: ConnManager
2021-12-16 10:05:20 +00:00
dialLock: Table[PeerId, AsyncLock]
transports: seq[Transport]
2023-03-08 11:30:19 +00:00
peerStore: PeerStore
nameResolver: NameResolver
2022-09-26 09:03:24 +00:00
proc dialAndUpgrade(
self: Dialer,
peerId: Opt[PeerId],
hostname: string,
address: MultiAddress,
upgradeDir = Direction.Out):
2023-03-08 11:30:19 +00:00
Future[Muxer] {.async.} =
2022-09-26 09:03:24 +00:00
for transport in self.transports: # for each transport
if transport.handles(address): # check if it can dial it
2023-06-28 15:00:33 +00:00
trace "Dialing address", address, peerId = peerId.get(default(PeerId)), hostname
2022-09-26 09:03:24 +00:00
let dialed =
try:
libp2p_total_dial_attempts.inc()
await transport.dial(hostname, address, peerId)
2022-09-26 09:03:24 +00:00
except CancelledError as exc:
2023-06-28 15:00:33 +00:00
debug "Dialing canceled", err = exc.msg, peerId = peerId.get(default(PeerId))
2022-09-26 09:03:24 +00:00
raise exc
except CatchableError as exc:
2023-06-28 15:00:33 +00:00
debug "Dialing failed", err = exc.msg, peerId = peerId.get(default(PeerId))
2022-09-26 09:03:24 +00:00
libp2p_failed_dials.inc()
return nil # Try the next address
libp2p_successful_dials.inc()
2023-03-08 11:30:19 +00:00
let mux =
2022-09-26 09:03:24 +00:00
try:
dialed.transportDir = upgradeDir
await transport.upgrade(dialed, upgradeDir, peerId)
2022-09-26 09:03:24 +00:00
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()
2023-06-28 15:00:33 +00:00
debug "Upgrade failed", err = exc.msg, peerId = peerId.get(default(PeerId))
2022-09-26 09:03:24 +00:00
if exc isnot CancelledError:
if upgradeDir == Direction.Out:
libp2p_failed_upgrades_outgoing.inc()
else:
libp2p_failed_upgrades_incoming.inc()
2022-09-26 09:03:24 +00:00
# Try other address
return nil
doAssert not isNil(mux), "connection died after upgrade " & $upgradeDir
2023-03-08 11:30:19 +00:00
debug "Dial successful", peerId = mux.connection.peerId
return mux
2022-09-26 09:03:24 +00:00
return nil
proc expandDnsAddr(
self: Dialer,
peerId: Opt[PeerId],
address: MultiAddress): Future[seq[(MultiAddress, Opt[PeerId])]] {.async.} =
if not DNSADDR.matchPartial(address): return @[(address, peerId)]
if isNil(self.nameResolver):
info "Can't resolve DNSADDR without NameResolver", ma=address
return @[]
let
toResolve =
if peerId.isSome:
address & MultiAddress.init(multiCodec("p2p"), peerId.tryGet()).tryGet()
else:
address
resolved = await self.nameResolver.resolveDnsAddr(toResolve)
for resolvedAddress in resolved:
let lastPart = resolvedAddress[^1].tryGet()
if lastPart.protoCode == Result[MultiCodec, string].ok(multiCodec("p2p")):
let
peerIdBytes = lastPart.protoArgument().tryGet()
addrPeerId = PeerId.init(peerIdBytes).tryGet()
result.add((resolvedAddress[0..^2].tryGet(), Opt.some(addrPeerId)))
else:
result.add((resolvedAddress, peerId))
proc dialAndUpgrade(
self: Dialer,
peerId: Opt[PeerId],
addrs: seq[MultiAddress],
upgradeDir = Direction.Out):
2023-03-08 11:30:19 +00:00
Future[Muxer] {.async.} =
2022-09-26 09:03:24 +00:00
2023-06-28 15:00:33 +00:00
debug "Dialing peer", peerId = peerId.get(default(PeerId))
2022-09-26 09:03:24 +00:00
for rawAddress in addrs:
# resolve potential dnsaddr
let addresses = await self.expandDnsAddr(peerId, rawAddress)
for (expandedAddress, addrPeerId) in addresses:
# DNS resolution
let
hostname = expandedAddress.getHostname()
resolvedAddresses =
if isNil(self.nameResolver): @[expandedAddress]
else: await self.nameResolver.resolveMAddress(expandedAddress)
for resolvedAddress in resolvedAddresses:
result = await self.dialAndUpgrade(addrPeerId, hostname, resolvedAddress, upgradeDir)
2022-09-26 09:03:24 +00:00
if not isNil(result):
return result
proc tryReusingConnection(self: Dialer, peerId: PeerId): Opt[Muxer] =
2023-03-08 11:30:19 +00:00
let muxer = self.connManager.selectMuxer(peerId)
if muxer == nil:
return Opt.none(Muxer)
2023-03-08 11:30:19 +00:00
trace "Reusing existing connection", muxer, direction = $muxer.connection.dir
return Opt.some(muxer)
proc internalConnect(
self: Dialer,
peerId: Opt[PeerId],
2022-02-24 16:31:47 +00:00
addrs: seq[MultiAddress],
forceDial: bool,
reuseConnection = true,
upgradeDir = Direction.Out):
2023-03-08 11:30:19 +00:00
Future[Muxer] {.async.} =
if Opt.some(self.localPeerId) == peerId:
raise newException(CatchableError, "can't dial self!")
# Ensure there's only one in-flight attempt per peer
let lock = self.dialLock.mgetOrPut(peerId.get(default(PeerId)), newAsyncLock())
try:
await lock.acquire()
if reuseConnection:
peerId.withValue(peerId):
self.tryReusingConnection(peerId).withValue(mux):
return mux
2023-01-05 14:02:52 +00:00
let slot = self.connManager.getOutgoingSlot(forceDial)
2023-03-08 11:30:19 +00:00
let muxed =
try:
await self.dialAndUpgrade(peerId, addrs, upgradeDir)
except CatchableError as exc:
slot.release()
raise exc
2023-03-08 11:30:19 +00:00
slot.trackMuxer(muxed)
if isNil(muxed): # None of the addresses connected
raise newException(DialFailedError, "Unable to establish outgoing link")
2023-03-08 11:30:19 +00:00
try:
self.connManager.storeMuxer(muxed)
await self.peerStore.identify(muxed)
except CatchableError as exc:
trace "Failed to finish outgoung upgrade", err=exc.msg
await muxed.close()
raise exc
return muxed
finally:
if lock.locked():
lock.release()
method connect*(
self: Dialer,
2021-12-16 10:05:20 +00:00
peerId: PeerId,
2022-02-24 16:31:47 +00:00
addrs: seq[MultiAddress],
forceDial = false,
reuseConnection = true,
upgradeDir = Direction.Out) {.async.} =
## connect remote peer without negotiating
## a protocol
##
if self.connManager.connCount(peerId) > 0 and reuseConnection:
return
discard await self.internalConnect(Opt.some(peerId), addrs, forceDial, reuseConnection, upgradeDir)
method connect*(
self: Dialer,
address: MultiAddress,
allowUnknownPeerId = false): Future[PeerId] {.async.} =
## Connects to a peer and retrieve its PeerId
parseFullAddress(address).toOpt().withValue(fullAddress):
return (await self.internalConnect(
Opt.some(fullAddress[0]),
@[fullAddress[1]],
2023-03-08 11:30:19 +00:00
false)).connection.peerId
if allowUnknownPeerId == false:
raise newException(DialFailedError, "Address without PeerID and unknown peer id disabled!")
return (await self.internalConnect(
Opt.none(PeerId),
@[address],
false)).connection.peerId
proc negotiateStream(
self: Dialer,
conn: Connection,
protos: seq[string]): Future[Connection] {.async.} =
trace "Negotiating stream", conn, protos
2023-03-08 11:30:19 +00:00
let selected = await MultistreamSelect.select(conn, protos)
if not protos.contains(selected):
await conn.closeWithEOF()
raise newException(DialFailedError, "Unable to select sub-protocol " & $protos)
return conn
2022-08-23 15:49:07 +00:00
method tryDial*(
self: Dialer,
peerId: PeerId,
addrs: seq[MultiAddress]): Future[Opt[MultiAddress]] {.async.} =
2022-09-02 18:40:14 +00:00
## Create a protocol stream in order to check
2022-08-23 15:49:07 +00:00
## if a connection is possible.
## Doesn't use the Connection Manager to save it.
##
trace "Check if it can dial", peerId, addrs
try:
2023-03-08 11:30:19 +00:00
let mux = await self.dialAndUpgrade(Opt.some(peerId), addrs)
if mux.isNil():
2022-08-23 15:49:07 +00:00
raise newException(DialFailedError, "No valid multiaddress")
2023-03-08 11:30:19 +00:00
await mux.close()
return mux.connection.observedAddr
2022-08-23 15:49:07 +00:00
except CancelledError as exc:
raise exc
except CatchableError as exc:
raise newException(DialFailedError, exc.msg)
method dial*(
self: Dialer,
2021-12-16 10:05:20 +00:00
peerId: PeerId,
protos: seq[string]): Future[Connection] {.async.} =
## create a protocol stream over an
## existing connection
##
trace "Dialing (existing)", peerId, protos
let stream = await self.connManager.getStream(peerId)
if stream.isNil:
raise newException(DialFailedError, "Couldn't get muxed stream")
return await self.negotiateStream(stream, protos)
method dial*(
self: Dialer,
2021-12-16 10:05:20 +00:00
peerId: PeerId,
addrs: seq[MultiAddress],
2022-02-24 16:31:47 +00:00
protos: seq[string],
forceDial = false): Future[Connection] {.async.} =
## create a protocol stream and establish
## a connection if one doesn't exist already
##
var
2023-03-08 11:30:19 +00:00
conn: Muxer
stream: Connection
proc cleanup() {.async.} =
if not(isNil(stream)):
await stream.closeWithEOF()
if not(isNil(conn)):
await conn.close()
try:
trace "Dialing (new)", peerId, protos
conn = await self.internalConnect(Opt.some(peerId), addrs, forceDial)
trace "Opening stream", conn
stream = await self.connManager.getStream(conn)
if isNil(stream):
raise newException(DialFailedError,
"Couldn't get muxed stream")
return await self.negotiateStream(stream, protos)
except CancelledError as exc:
trace "Dial canceled", conn
await cleanup()
raise exc
except CatchableError as exc:
2023-06-28 15:00:33 +00:00
debug "Error dialing", conn, err = exc.msg
await cleanup()
raise exc
method addTransport*(self: Dialer, t: Transport) =
self.transports &= t
proc new*(
T: type Dialer,
localPeerId: PeerId,
connManager: ConnManager,
2023-03-08 11:30:19 +00:00
peerStore: PeerStore,
transports: seq[Transport],
nameResolver: NameResolver = nil): Dialer =
T(localPeerId: localPeerId,
connManager: connManager,
transports: transports,
2023-03-08 11:30:19 +00:00
peerStore: peerStore,
nameResolver: nameResolver)