diff --git a/nim-test-node/regression/main.nim b/nim-test-node/regression/main.nim index 8199fdb..5accef7 100644 --- a/nim-test-node/regression/main.nim +++ b/nim-test-node/regression/main.nim @@ -35,6 +35,8 @@ proc createMessageHandler(): proc(topic: string, data: seq[byte]) {.async, gcsaf # warm-up if timestampNs < 1000000: return + notePublishingStarted() + # Log received message info "Received message", msgId = msgId, @@ -202,6 +204,12 @@ proc main {.async.} = await switch.start() + # Before the staggered bootstrap dial below, not after. A node's mesh arrives on + # inbound dials from nodes that bootstrapped earlier, so it can publish long before + # its own dial returns; gating :8645 on that left half the fleet refusing publishes. + info "Starting listening endpoint for publish controller" + discard gossipSub.startHttpServer(myId) + info "Starting metrics server" let metricsServer = startMetricsServer(parseIpAddress("0.0.0.0"), prometheusPort) if metricsServer.isErr: @@ -233,11 +241,9 @@ proc main {.async.} = info "Mesh details ", meshSize = gossipSub.mesh.getOrDefault("test").len, peersConnected = gossipSub.gossipsub.getOrDefault("test").len - # Periodic mesh-only pings - asyncSpawn pingMeshLoop(switch, pingProtocol, gossipSub, "test") + # Hold connections open until gossipsub traffic takes over + asyncSpawn pingLoop(switch, pingProtocol) - info "Starting listening endpoint for publish controller" - discard gossipSub.startHttpServer(myId) await sleepAsync(2.days) diff --git a/nim-test-node/regression/ping_utils.nim b/nim-test-node/regression/ping_utils.nim index befb9a5..6003f8e 100644 --- a/nim-test-node/regression/ping_utils.nim +++ b/nim-test-node/regression/ping_utils.nim @@ -1,26 +1,38 @@ import chronos -import std/[tables, sets] +import std/[random, tables] import libp2p import libp2p/protocols/ping -import libp2p/protocols/pubsub/gossipsub const - MeshPingInterval = 45.seconds - MeshPingTimeout = 4.seconds - MeshDialTimeout = 4.seconds - MeshCloseTimeout = 2.seconds - SlowDialLog = 500.milliseconds - SlowCloseLog = 500.milliseconds + PingInterval = 12.seconds + ## A full sweep in batches takes a few seconds, so the gap between two pings of the + ## same peer is the interval plus a sweep. Keep that comfortably under quic's 30s + ## idle timeout. + PingBatch = 16 + ## Dials in flight at once. All ~250 together stalls the node's own publish + ## endpoint, which cost 34 injections in the first minute of a 1000-node run. + PingBatchGap = 200.milliseconds + PingTimeout = 4.seconds + DialTimeout = 4.seconds + CloseTimeout = 2.seconds + SlowDialLog = 500.milliseconds + SlowCloseLog = 500.milliseconds -proc meshPeerIds*(gossipSub: GossipSub, topic: string): seq[PeerId] = - ## Only peers in the mesh for `topic` (convert HashSet[PubSubPeer] -> seq[PeerId]) - let peers: HashSet[PubSubPeer] = gossipSub.mesh.getOrDefault(topic, initHashSet[PubSubPeer]()) - result = newSeqOfCap[PeerId](peers.len) - for p in peers: - result.add(p.peerId) +var messagesStarted = false -proc pingMeshPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.} = +proc notePublishingStarted*() = + ## Called from the message handler; stops the ping loop. + messagesStarted = true + +proc connectedPeerIds*(switch: Switch): seq[PeerId] = + ## Every peer we hold a connection to, not just the gossipsub mesh. + let conns = switch.connManager.getConnections() + result = newSeqOfCap[PeerId](conns.len) + for peerId in conns.keys: + result.add(peerId) + +proc pingPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.} = let book = switch.peerStore[AddressBook] if not book.book.hasKey(peerId): @@ -30,20 +42,20 @@ proc pingMeshPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.} if addrs.len == 0: return - var stream: Stream + var stream: Connection let dialStart = Moment.now() try: - stream = await switch.dial(peerId, addrs, PingCodec).wait(MeshDialTimeout) + stream = await switch.dial(peerId, addrs, PingCodec).wait(DialTimeout) let dialDur = Moment.now() - dialStart if dialDur >= SlowDialLog: - warn "mesh ping: slow dial", peerId = peerId, dialMs = dialDur.milliseconds + warn "keepalive ping: slow dial", peerId = peerId, dialMs = dialDur.milliseconds let pingStart = Moment.now() - let latency = await pingProtocol.ping(stream).wait(MeshPingTimeout) + let latency = await pingProtocol.ping(stream).wait(PingTimeout) let pingDur = Moment.now() - pingStart - info "mesh ping", + trace "keepalive ping", peerId = peerId, latency = latency, dialMs = dialDur.milliseconds, @@ -52,36 +64,52 @@ proc pingMeshPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.} raise exc except CatchableError as exc: let dialDur = Moment.now() - dialStart - warn "mesh ping failed", peerId = peerId, error = exc.msg, dialMs = dialDur.milliseconds + warn "keepalive ping failed", peerId = peerId, error = exc.msg, dialMs = dialDur.milliseconds finally: if not stream.isNil and not stream.closed: let closeStart = Moment.now() try: # Make close observable: if it never completes, you'll never see "slow close" today. - await stream.closeWithEOF().wait(MeshCloseTimeout) + await stream.closeWithEOF().wait(CloseTimeout) except CancelledError as exc: raise exc except CatchableError as exc: - warn "mesh ping: stream close failed", peerId = peerId, error = exc.msg + warn "keepalive ping: stream close failed", peerId = peerId, error = exc.msg finally: let closeDur = Moment.now() - closeStart if closeDur >= SlowCloseLog: - warn "mesh ping: slow close", peerId = peerId, closeMs = closeDur.milliseconds + warn "keepalive ping: slow close", peerId = peerId, closeMs = closeDur.milliseconds -proc pingMeshOnce*(switch: Switch, pingProtocol: Ping, gossipSub: GossipSub, topic: string) {.async.} = - let peers = gossipSub.meshPeerIds(topic) +proc pingAllOnce*(switch: Switch, pingProtocol: Ping) {.async.} = + var peers: seq[PeerId] = @[] + for pid in switch.connectedPeerIds(): + if pid != switch.peerInfo.peerId: + peers.add(pid) if peers.len == 0: return - var futs: seq[Future[void]] = @[] - for pid in peers: - if pid == switch.peerInfo.peerId: continue - futs.add(switch.pingMeshPeer(pingProtocol, pid)) - - if futs.len > 0: + var i = 0 + while i < peers.len: + if messagesStarted: + return + var futs: seq[Future[void]] = @[] + for pid in peers[i ..< min(i + PingBatch, peers.len)]: + futs.add(switch.pingPeer(pingProtocol, pid)) await allFutures(futs) + i += PingBatch + if i < peers.len: + await sleepAsync(PingBatchGap) -proc pingMeshLoop*(switch: Switch, pingProtocol: Ping, gossipSub: GossipSub, topic: string) {.async.} = - while true: - await switch.pingMeshOnce(pingProtocol, gossipSub, topic) - await sleepAsync(MeshPingInterval) +proc pingLoop*(switch: Switch, pingProtocol: Ping) {.async.} = + ## Hold every connection open until traffic starts doing it for us. + ## + ## quic closes a connection after 30s of silence, and before the first message there + ## is nothing to gossip about, so an idle cold start drops every connection that is + ## not in the mesh. Once messages flow, gossip reaches each peer every few seconds and + ## the ping is redundant, so it stops. + # Stagger the start so a thousand nodes do not ping in lockstep. + await sleepAsync(rand(PingInterval.milliseconds.int).milliseconds) + while not messagesStarted: + await switch.pingAllOnce(pingProtocol) + await sleepAsync(PingInterval) + info "keepalive ping stopped, gossipsub traffic now keeps connections open" diff --git a/nim-test-node/regression/test_node.nimble b/nim-test-node/regression/test_node.nimble index 925612b..a3e2de6 100644 --- a/nim-test-node/regression/test_node.nimble +++ b/nim-test-node/regression/test_node.nimble @@ -11,6 +11,6 @@ skipDirs = @[] requires "nim >= 2.2.0", "nimcrypto 0.6.4", - "https://github.com/vacp2p/nim-libp2p#26e181e4dd65188051ab4783bcc538ed9579644f", # 2.0.0 + "https://github.com/vacp2p/nim-libp2p#d721f9cb13c36b844957556ac04fecade371b059", # release/v2.2.0 # The shadow image's tick-floor patch targets lsquic 0.5.4's layout. "lsquic >= 0.5.4 & < 0.5.5" \ No newline at end of file