Multiple fixes

This commit is contained in:
Ludovic Chenut 2022-12-14 16:32:10 +01:00
parent bc0aa57a6b
commit 8ddc636fa7
No known key found for this signature in database
GPG Key ID: D9A59B1907F1D50C
4 changed files with 54 additions and 45 deletions

View File

@ -22,89 +22,96 @@ logScope:
type type
OnReservationHandler = proc (ma: MultiAddress): Future[void] {.gcsafe, raises: [Defect].} OnReservationHandler = proc (ma: MultiAddress): Future[void] {.gcsafe, raises: [Defect].}
AutoRelay* = ref object of Service AutoRelayService* = ref object of Service
running: bool running: bool
runner: Future[void] runner: Future[void]
client: RelayClient client: RelayClient
npeers: int numRelays: int
peers: Table[PeerId, Future[void]] relayPeers: Table[PeerId, Future[void]]
relayAddresses: Table[PeerId, MultiAddress]
peerJoined: AsyncEvent peerJoined: AsyncEvent
onReservation: OnReservationHandler onReservation: OnReservationHandler
proc reserveAndUpdate(self: AutoRelay, relayPid: PeerId, selfPid: PeerId) {.async.} = proc reserveAndUpdate(self: AutoRelayService, relayPid: PeerId, selfPid: PeerId) {.async.} =
while self.running: while self.running:
let let
rsvp = await self.client.reserve(relayPid).wait(chronos.seconds(5)) rsvp = await self.client.reserve(relayPid).wait(chronos.seconds(5))
relayedAddr = MultiAddress.init($(rsvp.addrs[0]) & relayedAddr = MultiAddress.init($(rsvp.addrs[0]) &
"/p2p-circuit/p2p/" & "/p2p-circuit/p2p/" &
$selfPid).tryGet() $selfPid).tryGet()
await self.onReservation(relayedAddr) if not self.onReservation.isNil():
await self.onReservation(relayedAddr)
self.relayAddresses[relayPid] = relayedAddr
await sleepAsync chronos.seconds(rsvp.expire.int64 - times.now().utc.toTime.toUnix) await sleepAsync chronos.seconds(rsvp.expire.int64 - times.now().utc.toTime.toUnix)
method setup*(self: AutoRelay, switch: Switch) {.async, gcsafe.} = method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcsafe.} =
if self.inUse: let hasBeenSetUp = await procCall Service(self).setup(switch)
warn "Autorelay setup has already been called" if hasBeenSetUp:
return proc handlePeerJoined(peerId: PeerId, event: PeerEvent) {.async.} =
self.inUse = true if self.relayPeers.len < self.numRelays:
proc handlePeer(peerId: PeerId, event: PeerEvent) {.async.} = self.peerJoined.fire()
if event.kind == Left and peerId in self.peers: proc handlePeerLeft(peerId: PeerId, event: PeerEvent) {.async.} =
self.peers[peerId].cancel() if peerId in self.relayPeers:
elif event.kind == Joined and self.peers.len < self.npeers: self.relayPeers[peerId].cancel()
self.peerJoined.fire() switch.addPeerEventHandler(handlePeerJoined, Joined)
switch.addPeerEventHandler(handlePeer, Joined) switch.addPeerEventHandler(handlePeerLeft, Left)
switch.addPeerEventHandler(handlePeer, Left) return hasBeenSetUp
method innerRun(self: AutoRelay, switch: Switch) {.async, gcsafe.} = method innerRun(self: AutoRelayService, switch: Switch) {.async, gcsafe.} =
while true: while true:
# Remove peers that failed # Remove relayPeers that failed
var peersToRemove: seq[PeerId] var peersToRemove: seq[PeerId]
for k, v in self.peers: for k, v in self.relayPeers:
if v.finished(): if v.finished():
peersToRemove.add(k) peersToRemove.add(k)
for k in peersToRemove: for k in peersToRemove:
self.peers.del(k) self.relayPeers.del(k)
self.relayAddresses.del(k)
if peersToRemove.len() > 0: if peersToRemove.len() > 0:
await sleepAsync(500.millis) # To avoid ddosing our peers in certain condition await sleepAsync(500.millis) # To avoid ddosing our relayPeers in certain condition
# Get all connected peers # Get all connected relayPeers
let rng = newRng() let rng = newRng()
var connectedPeers = switch.connectedPeers(Direction.Out) var connectedPeers = switch.connectedPeers(Direction.Out)
connectedPeers.keepItIf(RelayV2HopCodec in switch.peerStore[ProtoBook][it] or connectedPeers.keepItIf(RelayV2HopCodec in switch.peerStore[ProtoBook][it] or
it notin self.peers.keys()) it notin self.relayPeers)
rng.shuffle(connectedPeers) rng.shuffle(connectedPeers)
for relayPid in connectedPeers: for relayPid in connectedPeers:
if self.peers.len() >= self.npeers: if self.relayPeers.len() >= self.numRelays:
break break
if RelayV2HopCodec in switch.peerStore[ProtoBook][relayPid]: if RelayV2HopCodec in switch.peerStore[ProtoBook][relayPid]:
self.peers[relayPid] = self.reserveAndUpdate(relayPid, switch.peerInfo.peerId) self.relayPeers[relayPid] = self.reserveAndUpdate(relayPid, switch.peerInfo.peerId)
let peersFutures = toSeq(self.peers.values()) let peersFutures = toSeq(self.relayPeers.values())
if self.peers.len() < self.npeers: if self.relayPeers.len() < self.numRelays:
self.peerJoined.clear() self.peerJoined.clear()
await one(peersFutures) or self.peerJoined.wait() await one(peersFutures) or self.peerJoined.wait()
else: else:
discard await one(peersFutures) discard await one(peersFutures)
method run*(self: AutoRelay, switch: Switch) {.async, gcsafe.} = method run*(self: AutoRelayService, switch: Switch) {.async, gcsafe.} =
if self.running: if self.running:
trace "Autorelay is already running" trace "Autorelay is already running"
return return
self.running = true self.running = true
self.runner = self.innerRun(switch) self.runner = self.innerRun(switch)
method stop*(self: AutoRelay, switch: Switch) {.async, gcsafe.} = method stop*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcsafe.} =
if not self.inUse: let hasBeenStopped = await procCall Service(self).stop(switch)
warn "service is already stopped" if hasBeenStopped:
self.inUse = false self.running = false
self.running = false self.runner.cancel()
self.runner.cancel() return hasBeenStopped
proc new*(T: typedesc[AutoRelay], method getAddresses*(self: AutoRelayService): seq[MultiAddress] =
npeers: int, result = toSeq(self.relayAddresses.values)
proc new*(T: typedesc[AutoRelayService],
numRelays: int,
client: RelayClient, client: RelayClient,
onReservation: OnReservationHandler): T = onReservation: OnReservationHandler): T =
T(npeers: npeers, T(numRelays: numRelays,
client: client, client: client,
onReservation: onReservation, onReservation: onReservation,
peerJoined: newAsyncEvent()) peerJoined: newAsyncEvent())

View File

@ -79,20 +79,22 @@ type
Service* = ref object of RootObj Service* = ref object of RootObj
inUse*: bool inUse*: bool
method setup*(self: Service, switch: Switch) {.base, async, gcsafe, public.} = method setup*(self: Service, switch: Switch): Future[bool] {.base, async, gcsafe, public.} =
if self.inUse: if self.inUse:
warn "service setup has already been called" warn "service setup has already been called"
return return false
self.inUse = true self.inUse = true
return true
method run*(self: Service, switch: Switch) {.base, async, gcsafe, public.} = method run*(self: Service, switch: Switch) {.base, async, gcsafe, public.} =
doAssert(false, "Not implemented!") doAssert(false, "Not implemented!")
method stop*(self: Service, switch: Switch) {.base, async, gcsafe, public.} = method stop*(self: Service, switch: Switch): Future[bool] {.base, async, gcsafe, public.} =
if not self.inUse: if not self.inUse:
warn "service is already stopped" warn "service is already stopped"
return return false
self.inUse = false self.inUse = false
return true
proc addConnEventHandler*(s: Switch, proc addConnEventHandler*(s: Switch,
handler: ConnEventHandler, handler: ConnEventHandler,

View File

@ -35,7 +35,7 @@ suite "Autorelay":
$relay.peerInfo.peerId & "/p2p-circuit/p2p/" & $relay.peerInfo.peerId & "/p2p-circuit/p2p/" &
$switch.peerInfo.peerId).get() $switch.peerInfo.peerId).get()
fut.complete() fut.complete()
let autorelay = AutoRelay.new(3, client, checkMA) let autorelay = AutoRelayService.new(3, client, checkMA)
switch.addService(autorelay) switch.addService(autorelay)
await switch.start() await switch.start()
await relay.start() await relay.start()
@ -46,4 +46,3 @@ suite "Autorelay":
await fut await fut
await switch.stop() await switch.stop()
await relay.stop() await relay.stop()

View File

@ -40,4 +40,5 @@ import testtcptransport,
testrendezvous, testrendezvous,
testdiscovery, testdiscovery,
testyamux, testyamux,
testautonat testautonat,
testautorelay