Autonat service

libp2p/builders.nim

@@ -61,6 +61,7 @@ type
     autonat: bool
     circuitRelay: Relay
     rdv: RendezVous
+    services: seq[Service]
 
 proc new*(T: type[SwitchBuilder]): T {.public.} =
   ## Creates a SwitchBuilder
@@ -199,6 +200,10 @@ proc withRendezVous*(b: SwitchBuilder, rdv: RendezVous = RendezVous.new()): Swit
   b.rdv = rdv
   b
 
+proc withServices*(b: SwitchBuilder, services: seq[Service]): SwitchBuilder =
+  b.services = services
+  b
+
 proc build*(b: SwitchBuilder): Switch
   {.raises: [Defect, LPError], public.} =
 
@@ -254,7 +259,8 @@ proc build*(b: SwitchBuilder): Switch
     connManager = connManager,
     ms = ms,
     nameResolver = b.nameResolver,
-    peerStore = peerStore)
+    peerStore = peerStore,
+    services = b.services)
 
   if b.autonat:
     let autonat = Autonat.new(switch)

libp2p/connmanager.nim

@@ -110,6 +110,13 @@ proc new*(C: type ConnManager,
 proc connCount*(c: ConnManager, peerId: PeerId): int =
   c.conns.getOrDefault(peerId).len
 
+proc connectedPeers*(c: ConnManager, dir: Direction): seq[PeerId] =
+  var peers = newSeq[PeerId]();
+  for peerId, conns in c.conns:
+    if conns.anyIt(it.dir == dir):
+      peers.add(peerId)
+  return peers
+
 proc addConnEventHandler*(c: ConnManager,
                           handler: ConnEventHandler,
                           kind: ConnEventKind) =
@@ -537,3 +544,4 @@ proc close*(c: ConnManager) {.async.} =
       await conn.close()
 
   trace "Closed ConnManager"
+

libp2p/protocols/connectivity/autonat.nim

@@ -32,6 +32,7 @@ const
 
 type
   AutonatError* = object of LPError
+  AutonatUnreachableError* = object of LPError
 
   MsgType* = enum
     Dial = 0
@@ -203,25 +204,37 @@ type
     sem: AsyncSemaphore
     switch*: Switch
 
-proc dialMe*(a: Autonat, pid: PeerId, ma: MultiAddress|seq[MultiAddress]):
-    Future[MultiAddress] {.async.} =
-  let addrs = when ma is MultiAddress: @[ma] else: ma
-  let conn = await a.switch.dial(pid, addrs, AutonatCodec)
+method dialMe*(a: Autonat, pid: PeerId, addrs: seq[MultiAddress] = newSeq[MultiAddress]()):
+    Future[MultiAddress] {.base, async.} =
+
+  proc getResponseOrRaise(autonatMsg: Option[AutonatMsg]): AutonatDialResponse {.raises: [UnpackError, AutonatError].} =
+    if autonatMsg.isNone() or
+       autonatMsg.get().msgType != DialResponse or
+       autonatMsg.get().response.isNone() or
+       autonatMsg.get().response.get().ma.isNone():
+      raise newException(AutonatError, "Unexpected response")
+    else:
+      autonatMsg.get().response.get()
+
+  let conn =
+    try:
+      if addrs.len == 0:
+        await a.switch.dial(pid, @[AutonatCodec])
+      else:
+        await a.switch.dial(pid, addrs, AutonatCodec)
+    except CatchableError as err:
+      raise newException(AutonatError, "Unexpected error when dialling", err)
+
   defer: await conn.close()
   await conn.sendDial(a.switch.peerInfo.peerId, a.switch.peerInfo.addrs)
-  let msgOpt = AutonatMsg.decode(await conn.readLp(1024))
-  if msgOpt.isNone() or
-     msgOpt.get().msgType != DialResponse or
-     msgOpt.get().response.isNone():
-    raise newException(AutonatError, "Unexpected response")
-  let response = msgOpt.get().response.get()
-  if response.status != ResponseStatus.Ok:
-    raise newException(AutonatError, "Bad status " &
-                                      $response.status & " " &
-                                      response.text.get(""))
-  if response.ma.isNone():
-    raise newException(AutonatError, "Missing address")
-  return response.ma.get()
+  let response = getResponseOrRaise(AutonatMsg.decode(await conn.readLp(1024)))
+  return case response.status:
+    of ResponseStatus.Ok:
+      response.ma.get()
+    of ResponseStatus.DialError:
+      raise newException(AutonatUnreachableError, "Peer could not dial us back")
+    else:
+      raise newException(AutonatError, "Bad status " & $response.status & " " & response.text.get(""))
 
 proc tryDial(a: Autonat, conn: Connection, addrs: seq[MultiAddress]) {.async.} =
   try:

libp2p/protocols/connectivity/relay/client.nim

@@ -25,6 +25,7 @@ import ./relay,
        ../../../multiaddress,
        ../../../stream/connection
 
+export options
 
 logScope:
   topics = "libp2p relay relay-client"

libp2p/services/autonatservice.nim

@@ -0,0 +1,151 @@
+# Nim-LibP2P
+# Copyright (c) 2022 Status Research & Development GmbH
+# 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.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import std/[options, deques, sequtils]
+import chronos, metrics
+import ../switch
+import ../protocols/[connectivity/autonat]
+import ../utils/heartbeat
+import ../crypto/crypto
+
+declarePublicGauge(libp2p_autonat_reachability_confidence, "autonat reachability confidence", labels = ["reachability"])
+
+type
+  AutonatService* = ref object of Service
+    newConnectedPeerHandler: PeerEventHandler
+    scheduleHandle: Future[void]
+    networkReachability: NetworkReachability
+    confidence: Option[float]
+    answers: Deque[NetworkReachability]
+    autonat: Autonat
+    statusAndConfidenceHandler: StatusAndConfidenceHandler
+    rng: ref HmacDrbgContext
+    scheduleInterval: Option[Duration]
+    disableAskNewConnectedPeers: bool
+    numPeersToAsk: int
+    maxQueueSize: int
+    minConfidence: float
+    dialTimeout: Duration
+
+  NetworkReachability* {.pure.} = enum
+    NotReachable, Reachable, Unknown
+
+  StatusAndConfidenceHandler* = proc (networkReachability: NetworkReachability, confidence: Option[float]): Future[void]  {.gcsafe, raises: [Defect].}
+
+proc new*(
+  T: typedesc[AutonatService],
+  autonat: Autonat,
+  rng: ref HmacDrbgContext,
+  scheduleInterval: Option[Duration] = none(Duration),
+  disableAskNewConnectedPeers = false,
+  numPeersToAsk: int = 5,
+  maxQueueSize: int = 10,
+  minConfidence: float = 0.3,
+  dialTimeout = 5.seconds): T =
+  return T(
+    scheduleInterval: scheduleInterval,
+    networkReachability: Unknown,
+    confidence: none(float),
+    answers: initDeque[NetworkReachability](),
+    autonat: autonat,
+    rng: rng,
+    disableAskNewConnectedPeers: disableAskNewConnectedPeers,
+    numPeersToAsk: numPeersToAsk,
+    maxQueueSize: maxQueueSize,
+    minConfidence: minConfidence,
+    dialTimeout: dialTimeout)
+
+proc networkReachability*(self: AutonatService): NetworkReachability {.inline.} =
+  return self.networkReachability
+
+proc handleAnswer(self: AutonatService, ans: NetworkReachability) {.async.} =
+
+  if self.answers.len == self.maxQueueSize:
+    self.answers.popFirst()
+
+  self.answers.addLast(ans)
+
+  self.networkReachability = Unknown
+  self.confidence = none(float)
+  const reachabilityPriority = [Reachable, NotReachable]
+  for reachability in reachabilityPriority:
+    let confidence = self.answers.countIt(it == reachability) / self.maxQueueSize
+    libp2p_autonat_reachability_confidence.set(value = confidence, labelValues = [$reachability])
+    if self.confidence.isNone and confidence >= self.minConfidence:
+      self.networkReachability = reachability
+      self.confidence = some(confidence)
+
+  if not isNil(self.statusAndConfidenceHandler):
+    await self.statusAndConfidenceHandler(self.networkReachability, self.confidence)
+
+  trace "Current status", currentStats = $self.networkReachability, confidence = $self.confidence
+
+proc askPeer(self: AutonatService, s: Switch, peerId: PeerId): Future[NetworkReachability] {.async.} =
+  trace "Asking for reachability", peerId = $peerId
+  let ans =
+    try:
+      discard await self.autonat.dialMe(peerId).wait(self.dialTimeout)
+      Reachable
+    except AutonatUnreachableError:
+      trace "dialMe answer is not reachable", peerId = $peerId
+      NotReachable
+    except AsyncTimeoutError:
+      trace "dialMe timed out", peerId = $peerId
+      Unknown
+    except CatchableError as err:
+      trace "dialMe unexpected error", peerId = $peerId, errMsg = $err.msg
+      Unknown
+  await self.handleAnswer(ans)
+  return ans
+
+proc askConnectedPeers(self: AutonatService, switch: Switch) {.async.} =
+  var peers = switch.connectedPeers(Direction.Out)
+  self.rng.shuffle(peers)
+  var answersFromPeers = 0
+  for peer in peers:
+    if answersFromPeers >= self.numPeersToAsk:
+      break
+    elif (await askPeer(self, switch, peer)) != Unknown:
+      answersFromPeers.inc()
+
+proc schedule(service: AutonatService, switch: Switch, interval: Duration) {.async.} =
+  heartbeat "Schedule AutonatService run", interval:
+    await service.run(switch)
+
+method setup*(self: AutonatService, switch: Switch): Future[bool] {.async.} =
+  let hasBeenSetup = await procCall Service(self).setup(switch)
+  if hasBeenSetup:
+    if not self.disableAskNewConnectedPeers:
+      self.newConnectedPeerHandler = proc (peerId: PeerId, event: PeerEvent): Future[void] {.async.} =
+        discard askPeer(self, switch, peerId)
+      switch.connManager.addPeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
+    if self.scheduleInterval.isSome():
+      self.scheduleHandle = schedule(self, switch, self.scheduleInterval.get())
+  return hasBeenSetup
+
+method run*(self: AutonatService, switch: Switch) {.async, public.} =
+  await askConnectedPeers(self, switch)
+
+method stop*(self: AutonatService, switch: Switch): Future[bool] {.async, public.} =
+  let hasBeenStopped = await procCall Service(self).stop(switch)
+  if hasBeenStopped:
+    if not isNil(self.scheduleHandle):
+      self.scheduleHandle.cancel()
+      self.scheduleHandle = nil
+    if not isNil(self.newConnectedPeerHandler):
+      switch.connManager.removePeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
+  return hasBeenStopped
+
+proc statusAndConfidenceHandler*(self: AutonatService, statusAndConfidenceHandler: StatusAndConfidenceHandler) =
+  self.statusAndConfidenceHandler = statusAndConfidenceHandler

libp2p/switch.nim

@@ -74,6 +74,28 @@ type
       peerStore*: PeerStore
       nameResolver*: NameResolver
       started: bool
+      services*: seq[Service]
+
+    Service* = ref object of RootObj
+      inUse: bool
+
+
+method setup*(self: Service, switch: Switch): Future[bool] {.base, async, gcsafe.} =
+  if self.inUse:
+    warn "service setup has already been called"
+    return false
+  self.inUse = true
+  return true
+
+method run*(self: Service, switch: Switch) {.base, async, gcsafe.} =
+  doAssert(false, "Not implemented!")
+
+method stop*(self: Service, switch: Switch): Future[bool] {.base, async, gcsafe.} =
+  if not self.inUse:
+    warn "service is already stopped"
+    return false
+  self.inUse = false
+  return true
 
 proc addConnEventHandler*(s: Switch,
                           handler: ConnEventHandler,
@@ -108,6 +130,9 @@ method addTransport*(s: Switch, t: Transport) =
   s.transports &= t
   s.dialer.addTransport(t)
 
+proc connectedPeers*(s: Switch, dir: Direction): seq[PeerId] =
+  s.connManager.connectedPeers(dir)
+
 proc isConnected*(s: Switch, peerId: PeerId): bool {.public.} =
   ## returns true if the peer has one or more
   ## associated connections
@@ -294,6 +319,9 @@ proc stop*(s: Switch) {.async, public.} =
     if not a.finished:
       a.cancel()
 
+  for service in s.services:
+    discard await service.stop(s)
+
   await s.ms.stop()
 
   trace "Switch stopped"
@@ -335,6 +363,9 @@ proc start*(s: Switch) {.async, gcsafe, public.} =
 
   await s.ms.start()
 
+  for service in s.services:
+    discard await service.setup(s)
+
   s.started = true
 
   debug "Started libp2p node", peer = s.peerInfo
@@ -346,7 +377,8 @@ proc newSwitch*(peerInfo: PeerInfo,
                 connManager: ConnManager,
                 ms: MultistreamSelect,
                 nameResolver: NameResolver = nil,
-                peerStore = PeerStore.new()): Switch
+                peerStore = PeerStore.new(),
+                services = newSeq[Service]()): Switch
                 {.raises: [Defect, LPError], public.} =
   if secureManagers.len == 0:
     raise newException(LPError, "Provide at least one secure manager")
@@ -358,8 +390,10 @@ proc newSwitch*(peerInfo: PeerInfo,
     connManager: connManager,
     peerStore: peerStore,
     dialer: Dialer.new(peerInfo.peerId, connManager, transports, ms, nameResolver),
-    nameResolver: nameResolver)
+    nameResolver: nameResolver,
+    services: services)
 
   switch.connManager.peerStore = peerStore
   switch.mount(identity)
+
   return switch

tests/config.nims

@@ -1,3 +1,5 @@
 import ../config.nims
 
 --threads:on
+--d:metrics
+--d:withoutPCRE

tests/helpers.nim

@@ -116,4 +116,4 @@ proc checkExpiringInternal(cond: proc(): bool {.raises: [Defect], gcsafe.} ): Fu
       await sleepAsync(1.millis)
 
 template checkExpiring*(code: untyped): untyped =
-  check await checkExpiringInternal(proc(): bool = code)
+  check await checkExpiringInternal(proc(): bool = code)