-
Notifications
You must be signed in to change notification settings - Fork 54
/
autorelayservice.nim
155 lines (134 loc) · 5.29 KB
/
autorelayservice.nim
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
# Nim-LibP2P
# Copyright (c) 2023 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.
{.push raises: [].}
import chronos, chronicles, times, tables, sequtils
import ../switch, ../protocols/connectivity/relay/[client, utils]
logScope:
topics = "libp2p autorelay"
type
OnReservationHandler = proc(addresses: seq[MultiAddress]) {.gcsafe, raises: [].}
AutoRelayService* = ref object of Service
running: bool
runner: Future[void]
client: RelayClient
maxNumRelays: int # maximum number of relays we can reserve at the same time
relayPeers: Table[PeerId, Future[void]]
relayAddresses: Table[PeerId, seq[MultiAddress]]
backingOff: seq[PeerId]
peerAvailable: AsyncEvent
onReservation: OnReservationHandler
addressMapper: AddressMapper
rng: ref HmacDrbgContext
proc isRunning*(self: AutoRelayService): bool =
return self.running
proc addressMapper(
self: AutoRelayService, listenAddrs: seq[MultiAddress]
): Future[seq[MultiAddress]] {.async.} =
return concat(toSeq(self.relayAddresses.values)) & listenAddrs
proc reserveAndUpdate(
self: AutoRelayService, relayPid: PeerId, switch: Switch
) {.async.} =
while self.running:
let
rsvp = await self.client.reserve(relayPid).wait(chronos.seconds(5))
relayedAddr = rsvp.addrs.mapIt(MultiAddress.init($it & "/p2p-circuit").tryGet())
ttl = rsvp.expire.int64 - times.now().utc.toTime.toUnix
if ttl <= 60:
# A reservation under a minute is basically useless
break
if relayPid notin self.relayAddresses or self.relayAddresses[relayPid] != relayedAddr:
self.relayAddresses[relayPid] = relayedAddr
await switch.peerInfo.update()
debug "Updated relay addresses", relayPid, relayedAddr
if not self.onReservation.isNil():
self.onReservation(concat(toSeq(self.relayAddresses.values)))
await sleepAsync chronos.seconds(ttl - 30)
method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async.} =
self.addressMapper = proc(
listenAddrs: seq[MultiAddress]
): Future[seq[MultiAddress]] {.async.} =
return await addressMapper(self, listenAddrs)
let hasBeenSetUp = await procCall Service(self).setup(switch)
if hasBeenSetUp:
proc handlePeerIdentified(peerId: PeerId, event: PeerEvent) {.async.} =
trace "Peer Identified", peerId
if self.relayPeers.len < self.maxNumRelays:
self.peerAvailable.fire()
proc handlePeerLeft(peerId: PeerId, event: PeerEvent) {.async.} =
trace "Peer Left", peerId
self.relayPeers.withValue(peerId, future):
future[].cancel()
switch.addPeerEventHandler(handlePeerIdentified, Identified)
switch.addPeerEventHandler(handlePeerLeft, Left)
switch.peerInfo.addressMappers.add(self.addressMapper)
await self.run(switch)
return hasBeenSetUp
proc manageBackedOff(self: AutoRelayService, pid: PeerId) {.async.} =
await sleepAsync(chronos.seconds(5))
self.backingOff.keepItIf(it != pid)
self.peerAvailable.fire()
proc innerRun(self: AutoRelayService, switch: Switch) {.async.} =
while true:
# Remove relayPeers that failed
let peers = toSeq(self.relayPeers.keys())
for k in peers:
if self.relayPeers[k].finished():
self.relayPeers.del(k)
self.relayAddresses.del(k)
if not self.onReservation.isNil():
self.onReservation(concat(toSeq(self.relayAddresses.values)))
# To avoid ddosing our peers in certain conditions
self.backingOff.add(k)
asyncSpawn self.manageBackedOff(k)
# Get all connected relayPeers
self.peerAvailable.clear()
var connectedPeers = switch.connectedPeers(Direction.Out)
connectedPeers.keepItIf(
RelayV2HopCodec in switch.peerStore[ProtoBook][it] and it notin self.relayPeers and
it notin self.backingOff
)
self.rng.shuffle(connectedPeers)
for relayPid in connectedPeers:
if self.relayPeers.len() >= self.maxNumRelays:
break
self.relayPeers[relayPid] = self.reserveAndUpdate(relayPid, switch)
if self.relayPeers.len() > 0:
await one(toSeq(self.relayPeers.values())) or self.peerAvailable.wait()
else:
await self.peerAvailable.wait()
method run*(self: AutoRelayService, switch: Switch) {.async.} =
if self.running:
trace "Autorelay is already running"
return
self.running = true
self.runner = self.innerRun(switch)
method stop*(self: AutoRelayService, switch: Switch): Future[bool] {.async.} =
let hasBeenStopped = await procCall Service(self).stop(switch)
if hasBeenStopped:
self.running = false
self.runner.cancel()
switch.peerInfo.addressMappers.keepItIf(it != self.addressMapper)
await switch.peerInfo.update()
return hasBeenStopped
proc getAddresses*(self: AutoRelayService): seq[MultiAddress] =
result = concat(toSeq(self.relayAddresses.values))
proc new*(
T: typedesc[AutoRelayService],
maxNumRelays: int,
client: RelayClient,
onReservation: OnReservationHandler,
rng: ref HmacDrbgContext,
): T =
T(
maxNumRelays: maxNumRelays,
client: client,
onReservation: onReservation,
peerAvailable: newAsyncEvent(),
rng: rng,
)