From a380e49d2ca689b41460125e1f9464dd694c6259 Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Sat, 16 May 2026 16:41:33 +0000 Subject: [PATCH 1/8] remove mix and wait 15s instead 5s for mesh creation --- nim-test-node/gossipsub-queues/.gitignore | 3 + .../gossipsub-queues/Dockerfile_amd64 | 65 +++ .../gossipsub-queues/Dockerfile_arm64 | 49 ++ nim-test-node/gossipsub-queues/config.nims | 4 + nim-test-node/gossipsub-queues/env.nim | 77 +++ nim-test-node/gossipsub-queues/libp2p_version | 1 + nim-test-node/gossipsub-queues/main.nim | 497 ++++++++++++++++++ .../gossipsub-queues/test_node.nimble | 15 + nim-test-node/nimble.paths | 54 ++ 9 files changed, 765 insertions(+) create mode 100644 nim-test-node/gossipsub-queues/.gitignore create mode 100644 nim-test-node/gossipsub-queues/Dockerfile_amd64 create mode 100644 nim-test-node/gossipsub-queues/Dockerfile_arm64 create mode 100644 nim-test-node/gossipsub-queues/config.nims create mode 100644 nim-test-node/gossipsub-queues/env.nim create mode 100644 nim-test-node/gossipsub-queues/libp2p_version create mode 100644 nim-test-node/gossipsub-queues/main.nim create mode 100644 nim-test-node/gossipsub-queues/test_node.nimble create mode 100644 nim-test-node/nimble.paths diff --git a/nim-test-node/gossipsub-queues/.gitignore b/nim-test-node/gossipsub-queues/.gitignore new file mode 100644 index 0000000..f3685e2 --- /dev/null +++ b/nim-test-node/gossipsub-queues/.gitignore @@ -0,0 +1,3 @@ +nimble.develop +nimble.paths +nimbledeps diff --git a/nim-test-node/gossipsub-queues/Dockerfile_amd64 b/nim-test-node/gossipsub-queues/Dockerfile_amd64 new file mode 100644 index 0000000..ab9752d --- /dev/null +++ b/nim-test-node/gossipsub-queues/Dockerfile_amd64 @@ -0,0 +1,65 @@ +FROM debian:bookworm-slim AS builder + +WORKDIR /node + +COPY . . + +RUN apt-get update && apt-get install -y \ + curl git build-essential ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +RUN apt-get update && apt-get install -y \ + gcc-multilib g++-multilib libc6-dev-i386 \ + && rm -rf /var/lib/apt/lists/* + +RUN apt-get update && apt-get install -y \ + curl git build-essential ca-certificates \ + gcc libc6-dev-i386 \ + libssl3 iproute2 procps \ + && rm -rf /var/lib/apt/lists/* \ + && apt-get clean + +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + ca-certificates \ + libssl3 \ + iproute2 \ + curl \ + procps \ + && rm -rf /var/lib/apt/lists/* \ + && apt-get clean + +RUN git config --global http.sslVerify false + +RUN curl https://nim-lang.org/download/nim-2.2.10-linux_x64.tar.xz -o /tmp/nim.tar.xz \ + && tar -xf /tmp/nim.tar.xz -C /tmp \ + && cp -r /tmp/nim-2.2.10/* /usr/local \ + && rm -rf /usr/local/nim* && mv /tmp/nim-2.2.10 /usr/local/nim-2.2.10 + +RUN nimble refresh + +RUN ln -sf /usr/local/nim-2.2.10/bin/nim /usr/local/bin/nim +RUN ln -sf /usr/local/nim-2.2.10/bin/nimble /usr/local/bin/nimble +RUN ln -sf /usr/local/nim-2.2.10/bin/nim /usr/local/bin/nimcheck + +RUN echo 'import strutils; echo "OK"' > temp.nim && nim r temp.nim && rm temp.nim + +RUN nim --version +RUN nimble --version +RUN nimble install -dy --verbose | tee /tmp/install.log && grep -A 20 "libp2p" /tmp/install.log | grep -oP '\b\d+\.\d+\.\d+\b' | head -n1 > ./libp2p_version + +RUN nimble c \ + --os:linux --cpu:amd64 --passL:"-static -mmusl" \ + -d:chronicles_colors=None --threads:on --mm:refc \ + -d:metrics -d:libp2p_network_protocols_metrics -d:release -d:pubsubpeer_queue_metrics \ + -d:libp2p_expensive_metrics -d:libp2p_agents_metrics \ + --passL:"-static-libgcc -static-libstdc++" \ + main + +RUN chmod +x /node/main +RUN nim --version > nim_version +RUN nimble --version > nimble_version + +EXPOSE 5000 8008 8645 + +ENTRYPOINT ["/node/main"] \ No newline at end of file diff --git a/nim-test-node/gossipsub-queues/Dockerfile_arm64 b/nim-test-node/gossipsub-queues/Dockerfile_arm64 new file mode 100644 index 0000000..79ee333 --- /dev/null +++ b/nim-test-node/gossipsub-queues/Dockerfile_arm64 @@ -0,0 +1,49 @@ +FROM nimlang/nim:latest AS builder + +WORKDIR /node + +COPY . . + +RUN apt-get update && apt-get install -y \ + curl git build-essential ca-certificates \ + && rm -rf /var/lib/apt/lists/* +RUN git config --global http.sslVerify false +RUN git config --global --add safe.directory /node + +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + ca-certificates \ + libssl3 \ + iproute2 \ + curl \ + procps \ + && rm -rf /var/lib/apt/lists/* \ + && apt-get clean + +RUN git config --global http.sslVerify false + +RUN nimble refresh +RUN nim --version +RUN nimble --version + +RUN echo 'import strutils; echo "OK"' > temp.nim && nim r temp.nim && rm temp.nim + +RUN nimble install -dy --verbose | tee /tmp/install.log && grep -A 20 "libp2p" /tmp/install.log | grep -oP '\b\d+\.\d+\.\d+\b' | head -n1 > ./libp2p_version + + +RUN nimble c \ + --os:linux --cpu:arm64 --passL:"-static -mmusl" \ + -d:chronicles_colors=None --threads:on --mm:refc \ + -d:metrics -d:libp2p_network_protocols_metrics -d:release -d:pubsubpeer_queue_metrics \ + -d:libp2p_expensive_metrics -d:libp2p_agents_metrics \ + --passL:"-static-libgcc -static-libstdc++" \ + main + + +RUN chmod +x /node/main +RUN nim --version > nim_version +RUN nimble --version > nimble_version + +EXPOSE 5000 8008 8645 + +ENTRYPOINT ["/node/main"] \ No newline at end of file diff --git a/nim-test-node/gossipsub-queues/config.nims b/nim-test-node/gossipsub-queues/config.nims new file mode 100644 index 0000000..8ee48d2 --- /dev/null +++ b/nim-test-node/gossipsub-queues/config.nims @@ -0,0 +1,4 @@ +# begin Nimble config (version 2) +when withDir(thisDir(), system.fileExists("nimble.paths")): + include "nimble.paths" +# end Nimble config diff --git a/nim-test-node/gossipsub-queues/env.nim b/nim-test-node/gossipsub-queues/env.nim new file mode 100644 index 0000000..a0a653e --- /dev/null +++ b/nim-test-node/gossipsub-queues/env.nim @@ -0,0 +1,77 @@ +import strutils, os, osproc +import chronos, metrics/chronos_httpserver, chronicles +from nativesockets import getHostname + +let + mountsMix* = existsEnv("MOUNTSMIX") #Full mix-net peer + usesMix* = existsEnv("USESMIX") #Supports sending mix messages + mixCount* = parseInt(getEnv("NUMMIX", "0")) #Number of mix peers (mountsMix + usesMix) + inShadow* = getEnv("SHADOWENV").cmpIgnoreCase("true") == 0 #If Running for shadow simulator + httpPublishPort* = Port(8645) + prometheusPort* = Port(8008) + myPort* = Port(5000) + chunks* = parseInt(getEnv("FRAGMENTS", "1")) #No. of fragments for each message + mix_D* = parseInt(getEnv("MIXD", "4")) #No. of mix tunnels + + +proc getPeerDetails*(): Result[(int, int, int, string, string, string), string] = + let + hostname = getHostname() + ordinal = parseInt(hostname.split('-')[^1]) + peerIdOffset = parseInt(getEnv("PEER_ID_OFFSET", "0")) + myId = peerIdOffset + ordinal + networkSize = parseInt(getEnv("PEERS", "100")) + connectTo = parseInt(getEnv("CONNECTTO", "10")) + muxer = getEnv("MUXER", "yamux") + filePath = if inShadow: "../" else: getEnv("FILEPATH", "./") + address = if muxer.toLowerAscii() == "quic": + "/ip4/0.0.0.0/udp/" & $myPort & "/quic-v1" + else: + "/ip4/0.0.0.0/tcp/" & $myPort + + if muxer.toLowerAscii() notin ["quic", "yamux", "mplex"]: + return err("Unknown muxer type : " & muxer) + + if connectTo >= networkSize: + return err("Not enough peers to make target connections. Network size : " & $networkSize) + + info "Host info ", hostname = hostname, peer = myId, muxer = muxer, mountsMix = mountsMix, usesMix = usesMix, mixCount = mixCount, inShadow = inShadow, address = address + + return ok((myId, networkSize, connectTo, muxer, filePath, address)) + +#Prometheus metrics +proc startMetricsServer*( + serverIp: IpAddress, serverPort: Port +): Result[MetricsHttpServerRef, string] = + info "Starting metrics HTTP server", serverIp = $serverIp, serverPort = $serverPort + + let metricsServerRes = MetricsHttpServerRef.new($serverIp, serverPort) + if metricsServerRes.isErr(): + return err("metrics HTTP server start failed: " & $metricsServerRes.error) + + let server = metricsServerRes.value + try: + waitFor server.start() + except CatchableError: + return err("metrics HTTP server start failed: " & getCurrentExceptionMsg()) + + info "Metrics HTTP server started", serverIp = $serverIp, serverPort = $serverPort + ok(metricsServerRes.value) + +#log metrics if needed (useful for shadow simulations) +proc storeMetrics*(myId: int) {.async.} = + await sleepAsync((myId*60).milliseconds) + while true: + try: + let cmd = "curl -s --connect-timeout 5 --max-time 5 http://localhost:" & + $prometheusPort & "/metrics >> metrics_pod-" & $myId & ".txt" + + let exitCode = execCmd(cmd) + if exitCode == 0: + info "Metrics saved for peer ", pod = myId + else: + info "Failed to fetch metrics for peer ", pod = myId, curlExitCode = $exitCode + except CatchableError as e: + info "Error storing metrics: ", error = e.msg + return + await sleepAsync(5.minutes) \ No newline at end of file diff --git a/nim-test-node/gossipsub-queues/libp2p_version b/nim-test-node/gossipsub-queues/libp2p_version new file mode 100644 index 0000000..3eefcb9 --- /dev/null +++ b/nim-test-node/gossipsub-queues/libp2p_version @@ -0,0 +1 @@ +1.0.0 diff --git a/nim-test-node/gossipsub-queues/main.nim b/nim-test-node/gossipsub-queues/main.nim new file mode 100644 index 0000000..cdd854b --- /dev/null +++ b/nim-test-node/gossipsub-queues/main.nim @@ -0,0 +1,497 @@ +import stew/endians2, stew/byteutils, tables, strutils, os, json +import chronos, chronos/apps/http/httpserver +import env +import std/[strformat, random, hashes] +import libp2p, libp2p/[muxers/mplex/lpchannel, stream/connection, crypto/secp, multiaddress] +import libp2p/protocols/[pubsub/pubsubpeer, pubsub/rpc/messages, ping] +# Mix protocol not available in this libp2p version +# import libp2p/protocols/[mix, mix/mix_protocol] + +import sequtils, math, metrics, metrics/chronos_httpserver +from times import getTime, Time, toUnix, fromUnix, `-`, initTime, `$`, inMilliseconds +from times import getTime, toUnixFloat, `-`, initTime, `$`, inMilliseconds, Time +from nativesockets import getHostname + + +template toUnixNanoseconds(t: times.Time): int64 = + (t.toUnixFloat() * 1_000_000_000).int64 + +template fromUnixNanoseconds(ns: int64): times.Time = + initTime(ns div 1_000_000_000, ns mod 1_000_000_000) + +declareCounter( + dst_testnode_publish_requests_total, + "number of /publish requests accepted by the test node" +) + +declareCounter( + dst_testnode_publish_failures_total, + "number of failed local publish attempts" +) + +declareCounter( + dst_testnode_received_chunks_total, + "number of application-level message chunks received" +) + +declareCounter( + dst_testnode_completed_messages_total, + "number of application-level messages fully received" +) + +declareGauge( + dst_testnode_last_message_delay_ms, + "last observed application-level end-to-end message delay in milliseconds" +) + +declareGauge( + dst_testnode_mesh_size, + "current GossipSub mesh size for the test topic" +) + +declareGauge( + dst_testnode_topic_peers, + "current number of GossipSub peers for the test topic" +) +proc getEnvInt(name: string, defaultValue: int): int = + let value = getEnv(name, "") + if value.len == 0: + return defaultValue + + try: + return parseInt(value) + except ValueError: + warn "Invalid integer ENV value, using default", + name = name, + value = value, + defaultValue = defaultValue + return defaultValue + + +proc getEnvFloat(name: string, defaultValue: float): float = + let value = getEnv(name, "") + if value.len == 0: + return defaultValue + + try: + return parseFloat(value) + except ValueError: + warn "Invalid float ENV value, using default", + name = name, + value = value, + defaultValue = defaultValue + return defaultValue + + +proc getEnvBool(name: string, defaultValue: bool): bool = + let value = getEnv(name, "") + if value.len == 0: + return defaultValue + + try: + return parseBool(value) + except ValueError: + warn "Invalid bool ENV value, using default", + name = name, + value = value, + defaultValue = defaultValue + return defaultValue + +proc msgIdProvider(m: Message): Result[MessageId, ValidationResult] = + return ok(($m.data.hash).toBytes()) + +proc createMessageHandler(): proc(topic: string, data: seq[byte]) {.async, gcsafe.} = + var messagesChunks: CountTable[uint64] + + return proc(topic: string, data: seq[byte]) {.async, gcsafe.} = + let + timestampNs = uint64.fromBytesLE(data[0 ..< 8]).int64 + sendTime = fromUnixNanoseconds(timestampNs) + msgId = uint64.fromBytesLE(data[8 ..< 16]) + recvTime = getTime() + delay = recvTime - sendTime + + # warm-up + if timestampNs < 1000000: return + + # Log received message + info "Received message", + msgId = msgId, + sentAt = timestampNs, + current = recvTime.toUnixNanoseconds(), + delayMs = delay.inMilliseconds() + + messagesChunks.inc(msgId) # Use msgId instead of timestamp for tracking + if messagesChunks[msgId] < chunks: return + + echo msgId, " milliseconds: ", delay.inMilliseconds() + dst_testnode_completed_messages_total.inc() + dst_testnode_last_message_delay_ms.set(delay.inMilliseconds().int64) + +proc messageValidator(topic: string, msg: Message): Future[ValidationResult] {.async.} = + return ValidationResult.Accept + + +proc publishNewMessage(gossipSub: GossipSub, msgSize: int, topic: string): Future[(Time, int)] {.async.} = + dst_testnode_publish_requests_total.inc() + let + now = getTime() + nowInt = now.toUnixFloat() * 1_000_000_000.0 # seconds + nanoseconds as float + msgId = uint64(rand(high(int64))) # Safe 0..<2^63 range + + var + res = 0 + nowBytes = @(toBytesLE(uint64(nowInt))) & @(toBytesLE(msgId)) & + newSeq[byte](msgSize div chunks - 16) + + info "Sent message", + msgId = msgId, + timestamp = getTime().toUnixNanoseconds() + + #To support message fragmentation, we add fragment #. Each fragment (chunk) differs by one byte + for chunk in 0.. 0: + let responseJson = """{"status":"success","message":"Message published at time """ & $publishTime & "}" + return await req.respond(Http200, responseJson, HttpTable.init([("Content-Type", "application/json")])) + else: + let responseJson = """{"status":"error","message":"Failed to publist at time """ & $publishTime & "}" + return await req.respond(Http500, responseJson, HttpTable.init([("Content-Type", "application/json")])) + else: + return await req.respond(Http404, "Not Found") + else: + return await req.respond(Http405, "Method Not Supported") + + except CatchableError as e: + info "Error handling http request: ", error = e.msg + let responseJson = """{"status":"error","message":"""" & e.msg.replace("\"", "\\\"") & """"}""" + return await req.respond(Http400, responseJson, HttpTable.init([("Content-Type", "application/json")])) + + # http endpoint for publish controller + info "starting http server", httpPort = $httpPublishPort + let serverAddress = initTAddress("0.0.0.0:" & $httpPublishPort) + let serverRes = HttpServerRef.new(serverAddress, processRequests) + + if serverRes.isErr(): + raise newException(CatchableError, "Failed to create HTTP server: " & $serverRes.error) + + let server = serverRes.get() + server.start() + info "http server started ", httpPort = $httpPublishPort + return server + +# Mix protocol not available in this libp2p version +# proc initializeGossipsub(switch: Switch, anonymize: bool, mixProto: Opt[MixProtocol] = Opt.none(MixProtocol)): GossipSub = +proc initializeGossipsub(switch: Switch, anonymize: bool): GossipSub = + return GossipSub.init( + switch = switch, + triggerSelf = parseBool(getEnv("SELFTRIGGER", "true")), + msgIdProvider = msgIdProvider, + verifySignature = false, + anonymize = anonymize, + rng = libp2p.newRng(), + # Mix callbacks disabled - mix protocol not available in this libp2p version + # customConnCallbacks = if mountsMix and mixProto.isSome: + # Opt.some(CustomConnectionCallbacks( + # customConnCreationCB: makeMixConnCb(mixProto.get()), + # customPeerSelectionCB: makeMixPeerSelectCb() + # )) + # else: + # Opt.none(CustomConnectionCallbacks) + ) + +proc configureGossipsubParams(gossipSub: GossipSub) = + let + d = getEnvInt("GOSSIPSUB_D", 6) + dLow = getEnvInt("GOSSIPSUB_D_LOW", 4) + dHigh = getEnvInt("GOSSIPSUB_D_HIGH", 8) + dScore = getEnvInt("GOSSIPSUB_D_SCORE", dLow) + dOut = getEnvInt("GOSSIPSUB_D_OUT", d div 2) + dLazy = getEnvInt("GOSSIPSUB_D_LAZY", d) + + heartbeatMs = getEnvInt("GOSSIPSUB_HEARTBEAT_MS", 1000) + pruneBackoffSec = getEnvInt("GOSSIPSUB_PRUNE_BACKOFF_SEC", 60) + + maxHighPriorityQueueLen = getEnvInt("GOSSIPSUB_MAX_HIGH_PRIORITY_QUEUE_LEN", 256) + maxMediumPriorityQueueLen = getEnvInt("GOSSIPSUB_MAX_MEDIUM_PRIORITY_QUEUE_LEN", 512) + maxLowPriorityQueueLen = getEnvInt("GOSSIPSUB_MAX_LOW_PRIORITY_QUEUE_LEN", 1024) + + slowPeerPenaltyWeight = getEnvFloat("GOSSIPSUB_SLOW_PEER_PENALTY_WEIGHT", 0.0) + slowPeerPenaltyThreshold = getEnvFloat("GOSSIPSUB_SLOW_PEER_PENALTY_THRESHOLD", 2.0) + slowPeerPenaltyDecay = getEnvFloat("GOSSIPSUB_SLOW_PEER_PENALTY_DECAY", 0.2) + + decayIntervalMs = getEnvInt("GOSSIPSUB_DECAY_INTERVAL_MS", 1000) + decayToZero = getEnvFloat("GOSSIPSUB_DECAY_TO_ZERO", 0.01) + + #gossipThreshold = getEnvFloat("GOSSIPSUB_GOSSIP_THRESHOLD", -100.0) + #publishThreshold = getEnvFloat("GOSSIPSUB_PUBLISH_THRESHOLD", -1000.0) + #graylistThreshold = getEnvFloat("GOSSIPSUB_GRAYLIST_THRESHOLD", -10000.0) + + gossipSub.parameters.floodPublish = getEnvBool("GOSSIPSUB_FLOOD_PUBLISH", true) + gossipSub.parameters.opportunisticGraftThreshold = getEnvFloat("GOSSIPSUB_OPPORTUNISTIC_GRAFT_THRESHOLD", -10000) + + gossipSub.parameters.heartbeatInterval = heartbeatMs.milliseconds + gossipSub.parameters.pruneBackoff = pruneBackoffSec.seconds + gossipSub.parameters.gossipFactor = getEnvFloat("GOSSIPSUB_GOSSIP_FACTOR", 0.25) + + gossipSub.parameters.d = d + gossipSub.parameters.dLow = dLow + gossipSub.parameters.dHigh = dHigh + gossipSub.parameters.dScore = dScore + gossipSub.parameters.dOut = dOut + gossipSub.parameters.dLazy = dLazy + + gossipSub.parameters.maxHighPriorityQueueLen = maxHighPriorityQueueLen + gossipSub.parameters.maxMediumPriorityQueueLen = maxMediumPriorityQueueLen + gossipSub.parameters.maxLowPriorityQueueLen = maxLowPriorityQueueLen + + gossipSub.parameters.slowPeerPenaltyWeight = slowPeerPenaltyWeight + gossipSub.parameters.slowPeerPenaltyThreshold = slowPeerPenaltyThreshold + gossipSub.parameters.slowPeerPenaltyDecay = slowPeerPenaltyDecay + + gossipSub.parameters.decayInterval = decayIntervalMs.milliseconds + gossipSub.parameters.decayToZero = decayToZero + + #gossipSub.parameters.gossipThreshold = gossipThreshold + #gossipSub.parameters.publishThreshold = publishThreshold + #gossipSub.parameters.graylistThreshold = graylistThreshold + + info "Configured GossipSub mesh params", + floodPublish = gossipSub.parameters.floodPublish, + opportunisticGraftThreshold = gossipSub.parameters.opportunisticGraftThreshold, + heartbeatMs = heartbeatMs, + pruneBackoffSec = pruneBackoffSec, + gossipFactor = gossipSub.parameters.gossipFactor, + d = d, + dLow = dLow, + dHigh = dHigh, + dScore = dScore, + dOut = dOut, + dLazy = dLazy + + info "Configured GossipSub queue and scoring params", + maxHighPriorityQueueLen = maxHighPriorityQueueLen, + maxMediumPriorityQueueLen = maxMediumPriorityQueueLen, + maxLowPriorityQueueLen = maxLowPriorityQueueLen, + slowPeerPenaltyWeight = slowPeerPenaltyWeight, + slowPeerPenaltyThreshold = slowPeerPenaltyThreshold, + slowPeerPenaltyDecay = slowPeerPenaltyDecay, + decayIntervalMs = decayIntervalMs, + decayToZero = decayToZero + #gossipThreshold = gossipThreshold, + #publishThreshold = publishThreshold, + #graylistThreshold = graylistThreshold + +proc subscribGossipsubTopic(gossipSub: GossipSub, topic: string) = + gossipSub.topicParams[topic] = TopicParams( + topicWeight: 1, + firstMessageDeliveriesWeight: 1, + firstMessageDeliveriesCap: 30, + firstMessageDeliveriesDecay: 0.9 + ) + + gossipSub.subscribe(topic, createMessageHandler()) + gossipSub.addValidator([topic], messageValidator) + + +proc resolveAddress(muxer: string, tAddress: string): Future[Result[seq[MultiAddress], string]] {.async.} = + while true: + try: + let resolvedAddrs = + if muxer.toLowerAscii() == "quic": + let quicV1 = MultiAddress.init("/quic-v1").tryGet() + resolveTAddress(tAddress).mapIt( + MultiAddress.init(it, IPPROTO_UDP).tryGet() + .concat(quicV1).tryGet() + ) + else: + resolveTAddress(tAddress).mapIt(MultiAddress.init(it).tryGet()) + info "Address resolved", tAddress = tAddress, resolvedAddrs = resolvedAddrs + return ok(resolvedAddrs) + except CatchableError as exc: + if inShadow: + return err(exc.msg) + #keep trying for service mode + warn "Failed to resolve address", address = tAddress, error = exc.msg + await sleepAsync(15.seconds) + +proc connectGossipsubPeers( + switch: Switch, muxer: string, networkSize: int, myId: int, connectTo: int +): Future[Result[int, string]] {.async.} = + let rng = libp2p.newRng() + var + addrs: seq[MultiAddress] = @[] + tAddresses: seq[string] + connected = 0 + + if inShadow: + var peers = toSeq(0..= connectTo: break + try: + discard await switch.connect(peer, allowUnknownPeerId=true).wait(5.seconds) + connected.inc() + info "Connected!: current connections ", connected = $connected, target = connectTo + except CatchableError as exc: + warn "Failed to dial ", theirAddress = peer, message = exc.msg + await sleepAsync(15.seconds) + + if connected == 0: + return err("Failed to connect any peer") + elif connected < connectTo: + warn "Connected to fewer peers than target", connected = connected, target = connectTo + return ok(connected) + + +proc main {.async.} = + randomize() + let + rng = libp2p.newRng() + (myId, networkSize, connectTo, muxer, filePath, address) = getPeerDetails().valueOr: + error "Error reading peer settings ", err = error + return + var + gossipSub: GossipSub + # Mix protocol not available in this libp2p version + # mixPublicKey: SkPublicKey + # mixPrivKey: SkPrivateKey + builder = SwitchBuilder + .new() + .withNoise() + .withAddress(MultiAddress.init(address).tryGet()) + .withMaxConnections(parseInt(getEnv("MAXCONNECTIONS", "250"))) + + # Mix protocol not available in this libp2p version + # if mountsMix or usesMix: + # let initResult = initializeMix(myId).valueOr: + # error "Failed to initialize mix", err = error + # return + # let multiAddr = initResult[0] + # mixPublicKey = initResult[1] + # mixPrivKey = initResult[2] + # + # #mix protocol uses same address as yamux + # builder = builder.withRng(crypto.newRng()) + # .withPrivateKey(PrivateKey(scheme: Secp256k1, skkey: mixPrivKey)) + # else: + builder = builder.withRng(rng) + + case muxer.toLowerAscii() + of "quic": + builder = builder.withQuicTransport() + of "yamux": + builder = builder.withTcpTransport(flags = {ServerFlags.TcpNoDelay}) + .withYamux() + of "mplex": + builder = builder.withTcpTransport(flags = {ServerFlags.TcpNoDelay}) + .withMplex() + + let switch = builder.build() + + # Mix protocol not available in this libp2p version + # if mountsMix or usesMix: + # writeMixInfoFiles(switch, myId, mixPublicKey, filePath) + # await sleepAsync(10.seconds) + + # if mountsMix: + # error "Mix not implemented" + # return + # else: + gossipSub = initializeGossipsub(switch, true) + + configureGossipsubParams(gossipSub) + subscribGossipsubTopic(gossipSub, "test") + switch.mount(gossipSub) + await switch.start() + + # Metrics + info "Starting metrics server" + let metricsServer = startMetricsServer(parseIpAddress("0.0.0.0"), prometheusPort) + if metricsServer.isErr: + error "Failed to initialize metrics server", err = metricsServer.error + elif inShadow: + asyncSpawn storeMetrics(myId) + + info "Listening on ", address = switch.peerInfo.addrs + info "Peer details ", peer = myId, peerId = switch.peerInfo.peerId + #Wait for node building + info "GossipSub codecs registered", codecs = gossipSub.codecs + await sleepAsync(60.seconds) + + #connect with peers + discard (await connectGossipsubPeers(switch, muxer, networkSize, myId, connectTo)).valueOr: + error "Failed to establish any connections", error = error + return + + await sleepAsync(15.seconds) # Allow multiple heartbeats to build mesh + let meshSize = gossipSub.mesh.getOrDefault("test").len + let peersConnected = gossipSub.gossipsub.getOrDefault("test").len + dst_testnode_mesh_size.set(meshSize.int64) + dst_testnode_topic_peers.set(peersConnected.int64) + + info "Mesh details ", + meshSize = meshSize, + peersConnected = peersConnected + info "Starting listening endpoint for publish controller" + discard gossipSub.startHttpServer(myId) + + await sleepAsync(2.days) + +waitFor(main()) \ No newline at end of file diff --git a/nim-test-node/gossipsub-queues/test_node.nimble b/nim-test-node/gossipsub-queues/test_node.nimble new file mode 100644 index 0000000..acbce1d --- /dev/null +++ b/nim-test-node/gossipsub-queues/test_node.nimble @@ -0,0 +1,15 @@ +mode = ScriptMode.Verbose + +bin = @["main"] + +packageName = "test_node" +version = "0.1.0" +author = "Status Research & Development GmbH" +description = "A test node for gossipsub" +license = "MIT" +skipDirs = @[] + +requires "nim >= 2.2.4", + "nimcrypto >= 0.6.0", + "https://github.com/vacp2p/nim-libp2p#9067f2a5b004fc54a70f53ab02f13f59befa8460" # fix(gossip): make slow peer penalty opt-in by default (#2429) + #"ggplotnim" \ No newline at end of file diff --git a/nim-test-node/nimble.paths b/nim-test-node/nimble.paths new file mode 100644 index 0000000..6f288d5 --- /dev/null +++ b/nim-test-node/nimble.paths @@ -0,0 +1,54 @@ +--noNimblePath +--path:"/home/mamoutou/.nimble/pkgs2/webview-0.1.1-3d4ae7bc728cff7e8f8bb971a8bb5d0df1ae798c" +--path:"/home/mamoutou/.nimble/pkgs2/ggplotnim-0.7.6-a5b58073ccbc53414494457cb76146561b0cc8a7" +--path:"/home/mamoutou/.nimble/pkgs2/ginger-0.6.2-4e5f7c788276c63e887757826fcb2c496aa72150" +--path:"/home/mamoutou/.nimble/pkgs2/nimlapack-0.3.1-fcb25795c6fb43f9251b7f34c3ad84c39e645afd" +--path:"/home/mamoutou/.nimble/pkgs2/chronos-4.2.2-3a4c9477df8cef20a04e4f1b54a2d74fdfc2a3d0" +--path:"/home/mamoutou/.nimble/pkgs2/zip-0.3.1-747aab3c43ecb7b50671cdd0ec3b2edc2c83494c" +--path:"/home/mamoutou/.nimble/pkgs2/json_serialization-0.4.4-8b3115354104858a0ac9019356fb29720529c2bd" +--path:"/home/mamoutou/.nimble/pkgs2/threading-0.2.1-594ec27a467847166ef78767ff60e1c87d0f743f" +--path:"/home/mamoutou/.nimble/pkgs2/unittest2-0.2.5-02bb3751ba9ddc3c17bfd89f2e41cb6bfb8fc0c9" +--path:"/home/mamoutou/.nimble/pkgs2/cairo-1.1.1-f2c13e59ce9658b2ae6fa0c89173f6012a23fafc" +--path:"/home/mamoutou/.nimble/pkgs2/stb_image-2.5-abf5fd03e72ee4c316c50a0538b973e355dcb175" +--path:"/home/mamoutou/.nimble/pkgs2/secp256k1-0.6.0.3.2-6618ef9de17121846a8c1d0317026b0ce8584e10" +--path:"/home/mamoutou/.nimble/pkgs2/pixie-6.0.0-a17700a73fcb77dfac2aa61dd19c005f67088035" +--path:"/home/mamoutou/.nimble/pkgs2/results-0.5.1-a9c011f74bc9ed5c91103917b9f382b12e82a9e7" +--path:"/home/mamoutou/.nimble/pkgs2/datamancer-0.5.1-c54db7077924522db8af5fe74f0bb850285332c6" +--path:"/home/mamoutou/.nimble/pkgs2/nimblas-0.3.1-e1ecdea4bb8176f12d66efd4aa0c7b3bea970027" +--path:"/home/mamoutou/.nimble/pkgs2/zlib-0.1.0-bbde4f5a97a84b450fef7d107461e5f35cf2b47f" +--path:"/home/mamoutou/.nimble/pkgs2/httputils-0.4.1-016774ab31c3afff9a423f7d80584905ee59c570" +--path:"/home/mamoutou/.nimble/pkgs2/dnsclient-0.3.4-65262c7e533ff49d6aca5539da4bc6c6ce132f40" +--path:"/home/mamoutou/.nimble/pkgs2/nimcrypto-0.6.4-f9ab24fa940ed03d0fb09729a7303feb50b7eaec" +--path:"/home/mamoutou/.nimble/pkgs2/flatty-0.3.4-5775e6ea6ca339efc5bd37b082b8294342d49dc5" +--path:"/home/mamoutou/.nimble/pkgs2/fontim-0.2.0-3aa3f5f541ecd9d24aaf4d56620b3c00fa668884" +--path:"/home/mamoutou/.nimble/pkgs2/vmath-3.0.0-d88141c3844bbb2bb2d6146ebb96e570ca863067" +--path:"/home/mamoutou/.nimble/pkgs2/chroma-1.0.0-76a12834f1b7e211e4232c1e13960accba6bd477" +--path:"/home/mamoutou/.nimble/pkgs2/serialization-0.5.2-fa35c1bb76a0a02a2379fe86eaae0957c7527cb8" +--path:"/home/mamoutou/.nimble/pkgs2/latexdsl-0.2.2-938592e63aa305f184d5bf6d7c9816ec9cf5aa06" +--path:"/home/mamoutou/.nimble/pkgs2/arraymancer-0.7.33-c37fb5d98fce8661ade439d89d0a6a3c9d65c8fc" +--path:"/home/mamoutou/.nimble/pkgs2/websock-0.3.0-1294a66520fa4541e261dec8a6a84f774fb8c0ac" +--path:"/home/mamoutou/.nimble/pkgs2/nimpy-0.2.1-0e88588fb093806b3e2d9f2a4b7e507abf863958" +--path:"/home/mamoutou/.nimble/pkgs2/untar-0.1.0-ceb12634783156ddd511410242dc7855ae2f4a14" +--path:"/home/mamoutou/.nimble/pkgs2/bearssl_pkey_decoder-0.1.0-21b42e2e6ddca6c875d3fc50f36a5115abf51714" +--path:"/home/mamoutou/.nimble/pkgs2/scinim-0.2.5-123625cbd61116b14d229ace3cd62269e4b63f7e" +--path:"/home/mamoutou/.nimble/pkgs2/bumpy-1.1.3-8a55667a7585612b7cefecd9a25ac3e761f03097" +--path:"/home/mamoutou/.nimble/pkgs2/stew-0.5.0-db22942939773ab7d5a0f2b2668c237240c67dd6" +--path:"/home/mamoutou/.nimble/pkgs2/bearssl-0.2.7-a85aab15b1b9a8b2438e9a128ac2eba41227da79" +--path:"/home/mamoutou/.nimble/pkgs2/nimcuda-0.2.2-630aaaad24b7f817f6c576bca246141919870cb8" +--path:"/home/mamoutou/.nimble/pkgs2/parsetoml-0.7.2-2a9fb57ef1f6460fd61b1cfab2d83af44f788a25" +--path:"/home/mamoutou/.nimble/pkgs2/clblast-0.0.2-0b4514b483f83492c49bb0629e5b45ffdf8e7aa3" +--path:"/home/mamoutou/.nimble/pkgs2/opencl-1.0.1-89c70044ee474834a4f7dcd6bf05a9082b0886bb" +--path:"/home/mamoutou/.nimble/pkgs2/libp2p-1.16.0-4d3c4a9a461069ae236e5cba463324d2d585945d" +--path:"/home/mamoutou/.nimble/pkgs2/zippy-0.10.19-706d8c10778f32d28b060dbcc84143ca4bfb9bcd" +--path:"/home/mamoutou/.nimble/pkgs2/metrics-0.2.1-84bb09873d7677c06046f391c7b473cd2fcff8a2" +--path:"/home/mamoutou/.nimble/pkgs2/nimcl-0.1.3-fb304e3b75503fb359ffe649d019a2fac7e2b3dc" +--path:"/home/mamoutou/.nimble/pkgs2/polynumeric-0.2.1-25d14a718148352a3ab9db4dc8b53bf73dc669ab" +--path:"/home/mamoutou/.nimble/pkgs2/chronicles-0.12.2-02febb20d088120b2836d3306cfa21f434f88f65" +--path:"/home/mamoutou/.nimble/pkgs2/testutils-0.8.1-96a11cf8b84fa9bd12d4a553afa1cc4b7f9df4e3" +--path:"/home/mamoutou/.nimble/pkgs2/shell-0.6.0-ce4bd02e0fbe71a56753821b4162ecad488cec66" +--path:"/home/mamoutou/.nimble/pkgs2/faststreams-0.5.0-ee61e507b805ae1df7ec936f03f2d101b0d72383" +--path:"/home/mamoutou/.nimble/pkgs2/nimsimd-1.3.2-5202ce48d46eaf593da54e884774cdb2a884e717" +--path:"/home/mamoutou/.nimble/pkgs2/crunchy-0.1.11-d34f03fc0f6876dbf53f99601096728ec31f47a3" +--path:"/home/mamoutou/.nimble/pkgs2/jwt-0.2-bcfd6fc9c5e10a52b87117219b7ab5c98136bc8e" +--path:"/home/mamoutou/.nimble/pkgs2/lsquic-0.0.1-b11b0a74191cb5bd643d97a0f917dc9668256849" +--path:"/home/mamoutou/.nimble/pkgs2/seqmath-0.2.2-734340a5e463353b96c988e776df8b8760c099ec" From b5aa3c1d44edf106f09635a9ea0d6190501cbf5f Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Sat, 16 May 2026 17:03:49 +0000 Subject: [PATCH 2/8] add muxer and peer_id labels to dst_* metrics --- nim-test-node/gossipsub-queues/main.nim | 45 ++++++++++++++++++------- 1 file changed, 32 insertions(+), 13 deletions(-) diff --git a/nim-test-node/gossipsub-queues/main.nim b/nim-test-node/gossipsub-queues/main.nim index cdd854b..759f090 100644 --- a/nim-test-node/gossipsub-queues/main.nim +++ b/nim-test-node/gossipsub-queues/main.nim @@ -19,39 +19,51 @@ template toUnixNanoseconds(t: times.Time): int64 = template fromUnixNanoseconds(ns: int64): times.Time = initTime(ns div 1_000_000_000, ns mod 1_000_000_000) +# Global variables for metric labels (set in main) +var + gMuxer*: string = "" + gPeerId*: string = "" + declareCounter( dst_testnode_publish_requests_total, - "number of /publish requests accepted by the test node" + "number of /publish requests accepted by the test node", + labels = ["muxer", "peer_id"] ) declareCounter( dst_testnode_publish_failures_total, - "number of failed local publish attempts" + "number of failed local publish attempts", + labels = ["muxer", "peer_id"] ) declareCounter( dst_testnode_received_chunks_total, - "number of application-level message chunks received" + "number of application-level message chunks received", + labels = ["muxer", "peer_id"] ) declareCounter( dst_testnode_completed_messages_total, - "number of application-level messages fully received" + "number of application-level messages fully received", + labels = ["muxer", "peer_id"] ) declareGauge( dst_testnode_last_message_delay_ms, - "last observed application-level end-to-end message delay in milliseconds" + "last observed application-level end-to-end message delay in milliseconds", + labels = ["muxer", "peer_id"] ) declareGauge( dst_testnode_mesh_size, - "current GossipSub mesh size for the test topic" + "current GossipSub mesh size for the test topic", + labels = ["muxer", "peer_id"] ) declareGauge( dst_testnode_topic_peers, - "current number of GossipSub peers for the test topic" + "current number of GossipSub peers for the test topic", + labels = ["muxer", "peer_id"] ) proc getEnvInt(name: string, defaultValue: int): int = let value = getEnv(name, "") @@ -125,15 +137,15 @@ proc createMessageHandler(): proc(topic: string, data: seq[byte]) {.async, gcsaf if messagesChunks[msgId] < chunks: return echo msgId, " milliseconds: ", delay.inMilliseconds() - dst_testnode_completed_messages_total.inc() - dst_testnode_last_message_delay_ms.set(delay.inMilliseconds().int64) + dst_testnode_completed_messages_total.inc(labelValues = [gMuxer, gPeerId]) + dst_testnode_last_message_delay_ms.set(delay.inMilliseconds().int64, labelValues = [gMuxer, gPeerId]) proc messageValidator(topic: string, msg: Message): Future[ValidationResult] {.async.} = return ValidationResult.Accept proc publishNewMessage(gossipSub: GossipSub, msgSize: int, topic: string): Future[(Time, int)] {.async.} = - dst_testnode_publish_requests_total.inc() + dst_testnode_publish_requests_total.inc(labelValues = [gMuxer, gPeerId]) let now = getTime() nowInt = now.toUnixFloat() * 1_000_000_000.0 # seconds + nanoseconds as float @@ -166,7 +178,7 @@ proc publishNewMessage(gossipSub: GossipSub, msgSize: int, topic: string): Futur topicInTopics = (topic in gossipSub.topics), floodPublish = gossipSub.parameters.floodPublish if res <= 0: - dst_testnode_publish_failures_total.inc() + dst_testnode_publish_failures_total.inc(labelValues = [gMuxer, gPeerId]) return (now, res) #http endpoint for detached controller @@ -407,6 +419,10 @@ proc main {.async.} = (myId, networkSize, connectTo, muxer, filePath, address) = getPeerDetails().valueOr: error "Error reading peer settings ", err = error return + + # Set global metric labels + gMuxer = muxer + var gossipSub: GossipSub # Mix protocol not available in this libp2p version @@ -444,6 +460,9 @@ proc main {.async.} = .withMplex() let switch = builder.build() + + # Set peerId for metric labels + gPeerId = $switch.peerInfo.peerId # Mix protocol not available in this libp2p version # if mountsMix or usesMix: @@ -483,8 +502,8 @@ proc main {.async.} = await sleepAsync(15.seconds) # Allow multiple heartbeats to build mesh let meshSize = gossipSub.mesh.getOrDefault("test").len let peersConnected = gossipSub.gossipsub.getOrDefault("test").len - dst_testnode_mesh_size.set(meshSize.int64) - dst_testnode_topic_peers.set(peersConnected.int64) + dst_testnode_mesh_size.set(meshSize.int64, labelValues = [gMuxer, gPeerId]) + dst_testnode_topic_peers.set(peersConnected.int64, labelValues = [gMuxer, gPeerId]) info "Mesh details ", meshSize = meshSize, From 0d41340741806890c18b5044246319d27b40c003 Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Sat, 16 May 2026 17:52:46 +0000 Subject: [PATCH 3/8] fix dst_* labels issues --- nim-test-node/gossipsub-queues/Dockerfile_amd64 | 1 + nim-test-node/gossipsub-queues/main.nim | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/nim-test-node/gossipsub-queues/Dockerfile_amd64 b/nim-test-node/gossipsub-queues/Dockerfile_amd64 index ab9752d..c4f28ae 100644 --- a/nim-test-node/gossipsub-queues/Dockerfile_amd64 +++ b/nim-test-node/gossipsub-queues/Dockerfile_amd64 @@ -6,6 +6,7 @@ COPY . . RUN apt-get update && apt-get install -y \ curl git build-essential ca-certificates \ + libssl-dev \ && rm -rf /var/lib/apt/lists/* RUN apt-get update && apt-get install -y \ diff --git a/nim-test-node/gossipsub-queues/main.nim b/nim-test-node/gossipsub-queues/main.nim index 759f090..7cb2716 100644 --- a/nim-test-node/gossipsub-queues/main.nim +++ b/nim-test-node/gossipsub-queues/main.nim @@ -19,10 +19,10 @@ template toUnixNanoseconds(t: times.Time): int64 = template fromUnixNanoseconds(ns: int64): times.Time = initTime(ns div 1_000_000_000, ns mod 1_000_000_000) -# Global variables for metric labels (set in main) +# Global variables for metric labels (set in main) - thread local for GC safety var - gMuxer*: string = "" - gPeerId*: string = "" + gMuxer* {.threadvar.}: string + gPeerId* {.threadvar.}: string declareCounter( dst_testnode_publish_requests_total, From ec7b5d06232dac7b08009f42bf4eacc7d698ca41 Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Mon, 18 May 2026 00:41:40 +0000 Subject: [PATCH 4/8] add new publisher for gossip queues overflow tests --- nim-test-node/gossipsub-queues/main.nim | 17 +- .../gossipsub-queues/publisher/Dockerfile | 5 + .../gossipsub-queues/publisher/traffic.py | 345 ++++++++++++++++++ 3 files changed, 366 insertions(+), 1 deletion(-) create mode 100644 nim-test-node/gossipsub-queues/publisher/Dockerfile create mode 100644 nim-test-node/gossipsub-queues/publisher/traffic.py diff --git a/nim-test-node/gossipsub-queues/main.nim b/nim-test-node/gossipsub-queues/main.nim index 7cb2716..50dc933 100644 --- a/nim-test-node/gossipsub-queues/main.nim +++ b/nim-test-node/gossipsub-queues/main.nim @@ -48,9 +48,22 @@ declareCounter( labels = ["muxer", "peer_id"] ) +declareCounter( + dst_testnode_message_delay_ms_sum, + "sum of message delays in milliseconds (use with rate)", + labels = ["muxer", "peer_id"] +) + +declareHistogram( + dst_testnode_message_delay_ms, + "message delay histogram for percentile analysis", + labels = ["muxer", "peer_id"], + buckets = [1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0, 10000.0] +) + declareGauge( dst_testnode_last_message_delay_ms, - "last observed application-level end-to-end message delay in milliseconds", + "last observed message delay in milliseconds (real-time)", labels = ["muxer", "peer_id"] ) @@ -138,6 +151,8 @@ proc createMessageHandler(): proc(topic: string, data: seq[byte]) {.async, gcsaf echo msgId, " milliseconds: ", delay.inMilliseconds() dst_testnode_completed_messages_total.inc(labelValues = [gMuxer, gPeerId]) + dst_testnode_message_delay_ms_sum.inc(delay.inMilliseconds().int64, labelValues = [gMuxer, gPeerId]) + dst_testnode_message_delay_ms.observe(delay.inMilliseconds().float64, labelValues = [gMuxer, gPeerId]) dst_testnode_last_message_delay_ms.set(delay.inMilliseconds().int64, labelValues = [gMuxer, gPeerId]) proc messageValidator(topic: string, msg: Message): Future[ValidationResult] {.async.} = diff --git a/nim-test-node/gossipsub-queues/publisher/Dockerfile b/nim-test-node/gossipsub-queues/publisher/Dockerfile new file mode 100644 index 0000000..2193cbe --- /dev/null +++ b/nim-test-node/gossipsub-queues/publisher/Dockerfile @@ -0,0 +1,5 @@ +FROM python:3.10.16-alpine3.21 + +ADD ./traffic.py /app/traffic.py + +RUN pip install requests argparse aiohttp \ No newline at end of file diff --git a/nim-test-node/gossipsub-queues/publisher/traffic.py b/nim-test-node/gossipsub-queues/publisher/traffic.py new file mode 100644 index 0000000..120c1b1 --- /dev/null +++ b/nim-test-node/gossipsub-queues/publisher/traffic.py @@ -0,0 +1,345 @@ +import argparse +import asyncio +import logging +import random +import socket +import time +from dataclasses import dataclass +from typing import Dict, Optional + +import aiohttp + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) + + +@dataclass +class Target: + host: str + ip: str + url: str + + +async def resolve_host(host: str) -> str: + loop = asyncio.get_running_loop() + start = time.time() + + try: + ip = await loop.run_in_executor(None, socket.gethostbyname, host) + elapsed_ms = (time.time() - start) * 1000 + + logging.debug( + "DNS host=%s ip=%s elapsed_ms=%.2f", + host, + ip, + elapsed_ms, + ) + + return ip + + except (socket.gaierror, socket.herror, OSError) as exc: + raise RuntimeError(f"DNS lookup failed for host={host}: {exc}") from exc + + +def build_pod_hostname(args: argparse.Namespace, node_id: int) -> str: + base = f"{args.pod_prefix}-{node_id}" + + if args.pod_domain: + return f"{base}.{args.pod_domain}" + + return base + + +async def resolve_target(args: argparse.Namespace, message_index: int) -> Target: + if args.peer_selection == "service": + host = args.service_host + + elif args.peer_selection == "fixed": + host = build_pod_hostname(args, args.fixed_id) + + elif args.peer_selection == "round-robin": + node_count = args.end_id - args.start_id + 1 + node_id = args.start_id + (message_index % node_count) + host = build_pod_hostname(args, node_id) + + elif args.peer_selection == "random-range": + node_id = random.randint(args.start_id, args.end_id) + host = build_pod_hostname(args, node_id) + + else: + raise ValueError(f"Unsupported peer selection: {args.peer_selection}") + + ip = await resolve_host(host) + url = f"http://{ip}:{args.port}/publish" + + return Target(host=host, ip=ip, url=url) + + +async def send_libp2p_msg( + session: aiohttp.ClientSession, + args: argparse.Namespace, + stats: Dict[str, int], + message_index: int, +): + target = await resolve_target(args, message_index) + + headers = {"Content-Type": "application/json"} + body = { + "topic": args.pubsub_topic, + "msgSize": args.msg_size_bytes, + "version": 1, + } + + logging.info( + "message=%d target_host=%s target_ip=%s url=%s", + message_index, + target.host, + target.ip, + target.url, + ) + + start = time.time() + + try: + async with session.post( + target.url, + json=body, + headers=headers, + timeout=args.request_timeout, + ) as response: + elapsed_ms = (time.time() - start) * 1000 + response_text = await response.text() + + stats["total"] += 1 + + if response.status == 200: + stats["success"] += 1 + else: + stats["failure"] += 1 + + success_rate = ( + (stats["success"] / stats["total"]) * 100 + if stats["total"] > 0 + else 0 + ) + + logging.info( + "message=%d target=%s status=%d elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f response=%s", + message_index, + target.host, + response.status, + elapsed_ms, + stats["success"], + stats["failure"], + stats["total"], + success_rate, + response_text[:300], + ) + + except Exception as exc: + elapsed_ms = (time.time() - start) * 1000 + + stats["total"] += 1 + stats["failure"] += 1 + + success_rate = ( + (stats["success"] / stats["total"]) * 100 + if stats["total"] > 0 + else 0 + ) + + logging.warning( + "message=%d target=%s exception=%s elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f", + message_index, + target.host, + repr(exc), + elapsed_ms, + stats["success"], + stats["failure"], + stats["total"], + success_rate, + ) + + +async def main(args: argparse.Namespace): + stats = { + "success": 0, + "failure": 0, + "total": 0, + } + + background_tasks = set() + start_time = time.time() + message_index = 0 + + timeout = aiohttp.ClientTimeout(total=args.request_timeout) + + async with aiohttp.ClientSession(timeout=timeout) as session: + while True: + if args.messages is not None and message_index >= args.messages: + break + + if ( + args.duration_seconds is not None + and time.time() - start_time >= args.duration_seconds + ): + break + + task = asyncio.create_task( + send_libp2p_msg(session, args, stats, message_index) + ) + + background_tasks.add(task) + task.add_done_callback(background_tasks.discard) + + message_index += 1 + await asyncio.sleep(args.delay_seconds) + + if background_tasks: + await asyncio.gather(*background_tasks) + + elapsed_s = time.time() - start_time + success_rate = ( + (stats["success"] / stats["total"]) * 100 + if stats["total"] > 0 + else 0 + ) + + logging.info( + "finished elapsed_s=%.2f success=%d failure=%d total=%d success_rate=%.2f", + elapsed_s, + stats["success"], + stats["failure"], + stats["total"], + success_rate, + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="nim-libp2p message injector") + + parser.add_argument( + "-t", + "--pubsub-topic", + type=str, + default="test", + help="PubSub topic", + ) + + parser.add_argument( + "-s", + "--msg-size-bytes", + type=int, + default=1000, + help="Message size in bytes", + ) + + parser.add_argument( + "-d", + "--delay-seconds", + type=float, + default=1.0, + help="Delay between publish requests", + ) + + parser.add_argument( + "-m", + "--messages", + type=int, + default=None, + help="Number of messages to inject", + ) + + parser.add_argument( + "--duration-seconds", + type=float, + default=None, + help="Duration of the injection phase. Use either this or --messages.", + ) + + parser.add_argument( + "--peer-selection", + type=str, + choices=["service", "fixed", "round-robin", "random-range"], + default="service", + help="How to select the test node receiving /publish requests", + ) + + parser.add_argument( + "--service-host", + type=str, + default="nimp2p-service", + help="Kubernetes service hostname for service-based peer selection", + ) + + parser.add_argument( + "--pod-prefix", + type=str, + default="nim-quic-normal", + help="StatefulSet pod prefix for fixed/range selection", + ) + + parser.add_argument( + "--pod-domain", + type=str, + default="nimp2p-service", + help="Headless service DNS domain. Example: nimp2p-service", + ) + + parser.add_argument( + "--fixed-id", + type=int, + default=0, + help="Pod ordinal used with --peer-selection fixed", + ) + + parser.add_argument( + "--start-id", + type=int, + default=0, + help="Start ordinal for round-robin/random-range selection", + ) + + parser.add_argument( + "--end-id", + type=int, + default=99, + help="End ordinal for round-robin/random-range selection", + ) + + parser.add_argument( + "-p", + "--port", + type=int, + default=8645, + help="test node HTTP publish port", + ) + + parser.add_argument( + "--request-timeout", + type=float, + default=10.0, + help="HTTP request timeout in seconds", + ) + + args = parser.parse_args() + + if args.messages is None and args.duration_seconds is None: + parser.error("Set either --messages or --duration-seconds") + + if args.messages is not None and args.duration_seconds is not None: + parser.error("Use either --messages or --duration-seconds, not both") + + if args.start_id > args.end_id: + parser.error("--start-id must be <= --end-id") + + return args + + +if __name__ == "__main__": + parsed_args = parse_args() + logging.info("args=%s", parsed_args) + asyncio.run(main(parsed_args)) \ No newline at end of file From 4f355e449f0c1eeb2d105527d15efeddccefc377 Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Mon, 18 May 2026 03:28:54 +0000 Subject: [PATCH 5/8] remove nimble.paths --- nim-test-node/nimble.paths | 54 -------------------------------------- 1 file changed, 54 deletions(-) delete mode 100644 nim-test-node/nimble.paths diff --git a/nim-test-node/nimble.paths b/nim-test-node/nimble.paths deleted file mode 100644 index 6f288d5..0000000 --- a/nim-test-node/nimble.paths +++ /dev/null @@ -1,54 +0,0 @@ ---noNimblePath ---path:"/home/mamoutou/.nimble/pkgs2/webview-0.1.1-3d4ae7bc728cff7e8f8bb971a8bb5d0df1ae798c" ---path:"/home/mamoutou/.nimble/pkgs2/ggplotnim-0.7.6-a5b58073ccbc53414494457cb76146561b0cc8a7" ---path:"/home/mamoutou/.nimble/pkgs2/ginger-0.6.2-4e5f7c788276c63e887757826fcb2c496aa72150" ---path:"/home/mamoutou/.nimble/pkgs2/nimlapack-0.3.1-fcb25795c6fb43f9251b7f34c3ad84c39e645afd" ---path:"/home/mamoutou/.nimble/pkgs2/chronos-4.2.2-3a4c9477df8cef20a04e4f1b54a2d74fdfc2a3d0" ---path:"/home/mamoutou/.nimble/pkgs2/zip-0.3.1-747aab3c43ecb7b50671cdd0ec3b2edc2c83494c" ---path:"/home/mamoutou/.nimble/pkgs2/json_serialization-0.4.4-8b3115354104858a0ac9019356fb29720529c2bd" ---path:"/home/mamoutou/.nimble/pkgs2/threading-0.2.1-594ec27a467847166ef78767ff60e1c87d0f743f" ---path:"/home/mamoutou/.nimble/pkgs2/unittest2-0.2.5-02bb3751ba9ddc3c17bfd89f2e41cb6bfb8fc0c9" ---path:"/home/mamoutou/.nimble/pkgs2/cairo-1.1.1-f2c13e59ce9658b2ae6fa0c89173f6012a23fafc" ---path:"/home/mamoutou/.nimble/pkgs2/stb_image-2.5-abf5fd03e72ee4c316c50a0538b973e355dcb175" ---path:"/home/mamoutou/.nimble/pkgs2/secp256k1-0.6.0.3.2-6618ef9de17121846a8c1d0317026b0ce8584e10" ---path:"/home/mamoutou/.nimble/pkgs2/pixie-6.0.0-a17700a73fcb77dfac2aa61dd19c005f67088035" ---path:"/home/mamoutou/.nimble/pkgs2/results-0.5.1-a9c011f74bc9ed5c91103917b9f382b12e82a9e7" ---path:"/home/mamoutou/.nimble/pkgs2/datamancer-0.5.1-c54db7077924522db8af5fe74f0bb850285332c6" ---path:"/home/mamoutou/.nimble/pkgs2/nimblas-0.3.1-e1ecdea4bb8176f12d66efd4aa0c7b3bea970027" ---path:"/home/mamoutou/.nimble/pkgs2/zlib-0.1.0-bbde4f5a97a84b450fef7d107461e5f35cf2b47f" ---path:"/home/mamoutou/.nimble/pkgs2/httputils-0.4.1-016774ab31c3afff9a423f7d80584905ee59c570" ---path:"/home/mamoutou/.nimble/pkgs2/dnsclient-0.3.4-65262c7e533ff49d6aca5539da4bc6c6ce132f40" ---path:"/home/mamoutou/.nimble/pkgs2/nimcrypto-0.6.4-f9ab24fa940ed03d0fb09729a7303feb50b7eaec" ---path:"/home/mamoutou/.nimble/pkgs2/flatty-0.3.4-5775e6ea6ca339efc5bd37b082b8294342d49dc5" ---path:"/home/mamoutou/.nimble/pkgs2/fontim-0.2.0-3aa3f5f541ecd9d24aaf4d56620b3c00fa668884" ---path:"/home/mamoutou/.nimble/pkgs2/vmath-3.0.0-d88141c3844bbb2bb2d6146ebb96e570ca863067" ---path:"/home/mamoutou/.nimble/pkgs2/chroma-1.0.0-76a12834f1b7e211e4232c1e13960accba6bd477" ---path:"/home/mamoutou/.nimble/pkgs2/serialization-0.5.2-fa35c1bb76a0a02a2379fe86eaae0957c7527cb8" ---path:"/home/mamoutou/.nimble/pkgs2/latexdsl-0.2.2-938592e63aa305f184d5bf6d7c9816ec9cf5aa06" ---path:"/home/mamoutou/.nimble/pkgs2/arraymancer-0.7.33-c37fb5d98fce8661ade439d89d0a6a3c9d65c8fc" ---path:"/home/mamoutou/.nimble/pkgs2/websock-0.3.0-1294a66520fa4541e261dec8a6a84f774fb8c0ac" ---path:"/home/mamoutou/.nimble/pkgs2/nimpy-0.2.1-0e88588fb093806b3e2d9f2a4b7e507abf863958" ---path:"/home/mamoutou/.nimble/pkgs2/untar-0.1.0-ceb12634783156ddd511410242dc7855ae2f4a14" ---path:"/home/mamoutou/.nimble/pkgs2/bearssl_pkey_decoder-0.1.0-21b42e2e6ddca6c875d3fc50f36a5115abf51714" ---path:"/home/mamoutou/.nimble/pkgs2/scinim-0.2.5-123625cbd61116b14d229ace3cd62269e4b63f7e" ---path:"/home/mamoutou/.nimble/pkgs2/bumpy-1.1.3-8a55667a7585612b7cefecd9a25ac3e761f03097" ---path:"/home/mamoutou/.nimble/pkgs2/stew-0.5.0-db22942939773ab7d5a0f2b2668c237240c67dd6" ---path:"/home/mamoutou/.nimble/pkgs2/bearssl-0.2.7-a85aab15b1b9a8b2438e9a128ac2eba41227da79" ---path:"/home/mamoutou/.nimble/pkgs2/nimcuda-0.2.2-630aaaad24b7f817f6c576bca246141919870cb8" ---path:"/home/mamoutou/.nimble/pkgs2/parsetoml-0.7.2-2a9fb57ef1f6460fd61b1cfab2d83af44f788a25" ---path:"/home/mamoutou/.nimble/pkgs2/clblast-0.0.2-0b4514b483f83492c49bb0629e5b45ffdf8e7aa3" ---path:"/home/mamoutou/.nimble/pkgs2/opencl-1.0.1-89c70044ee474834a4f7dcd6bf05a9082b0886bb" ---path:"/home/mamoutou/.nimble/pkgs2/libp2p-1.16.0-4d3c4a9a461069ae236e5cba463324d2d585945d" ---path:"/home/mamoutou/.nimble/pkgs2/zippy-0.10.19-706d8c10778f32d28b060dbcc84143ca4bfb9bcd" ---path:"/home/mamoutou/.nimble/pkgs2/metrics-0.2.1-84bb09873d7677c06046f391c7b473cd2fcff8a2" ---path:"/home/mamoutou/.nimble/pkgs2/nimcl-0.1.3-fb304e3b75503fb359ffe649d019a2fac7e2b3dc" ---path:"/home/mamoutou/.nimble/pkgs2/polynumeric-0.2.1-25d14a718148352a3ab9db4dc8b53bf73dc669ab" ---path:"/home/mamoutou/.nimble/pkgs2/chronicles-0.12.2-02febb20d088120b2836d3306cfa21f434f88f65" ---path:"/home/mamoutou/.nimble/pkgs2/testutils-0.8.1-96a11cf8b84fa9bd12d4a553afa1cc4b7f9df4e3" ---path:"/home/mamoutou/.nimble/pkgs2/shell-0.6.0-ce4bd02e0fbe71a56753821b4162ecad488cec66" ---path:"/home/mamoutou/.nimble/pkgs2/faststreams-0.5.0-ee61e507b805ae1df7ec936f03f2d101b0d72383" ---path:"/home/mamoutou/.nimble/pkgs2/nimsimd-1.3.2-5202ce48d46eaf593da54e884774cdb2a884e717" ---path:"/home/mamoutou/.nimble/pkgs2/crunchy-0.1.11-d34f03fc0f6876dbf53f99601096728ec31f47a3" ---path:"/home/mamoutou/.nimble/pkgs2/jwt-0.2-bcfd6fc9c5e10a52b87117219b7ab5c98136bc8e" ---path:"/home/mamoutou/.nimble/pkgs2/lsquic-0.0.1-b11b0a74191cb5bd643d97a0f917dc9668256849" ---path:"/home/mamoutou/.nimble/pkgs2/seqmath-0.2.2-734340a5e463353b96c988e776df8b8760c099ec" From 95d9e525c835b2aa95a9e8b8cac526614b97477b Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Mon, 18 May 2026 03:43:47 +0000 Subject: [PATCH 6/8] remove nimble.paths --- nim-test-node/gossipsub-queues/.gitignore | 3 --- nim-test-node/gossipsub-queues/libp2p_version | 1 - 2 files changed, 4 deletions(-) delete mode 100644 nim-test-node/gossipsub-queues/.gitignore delete mode 100644 nim-test-node/gossipsub-queues/libp2p_version diff --git a/nim-test-node/gossipsub-queues/.gitignore b/nim-test-node/gossipsub-queues/.gitignore deleted file mode 100644 index f3685e2..0000000 --- a/nim-test-node/gossipsub-queues/.gitignore +++ /dev/null @@ -1,3 +0,0 @@ -nimble.develop -nimble.paths -nimbledeps diff --git a/nim-test-node/gossipsub-queues/libp2p_version b/nim-test-node/gossipsub-queues/libp2p_version deleted file mode 100644 index 3eefcb9..0000000 --- a/nim-test-node/gossipsub-queues/libp2p_version +++ /dev/null @@ -1 +0,0 @@ -1.0.0 From 8c174e78e7734acb50e2efaf81bafaaed49e2edf Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Mon, 18 May 2026 04:14:42 +0000 Subject: [PATCH 7/8] add per node load mode --- .../gossipsub-queues/publisher/traffic.py | 598 +++++++++++++----- 1 file changed, 431 insertions(+), 167 deletions(-) diff --git a/nim-test-node/gossipsub-queues/publisher/traffic.py b/nim-test-node/gossipsub-queues/publisher/traffic.py index 120c1b1..e4b3e72 100644 --- a/nim-test-node/gossipsub-queues/publisher/traffic.py +++ b/nim-test-node/gossipsub-queues/publisher/traffic.py @@ -5,7 +5,7 @@ import random import socket import time from dataclasses import dataclass -from typing import Dict, Optional +from typing import Dict, List, Optional import aiohttp @@ -23,13 +23,39 @@ class Target: url: str +class Stats: + def __init__(self) -> None: + self.success = 0 + self.failure = 0 + self.total = 0 + self.lock = asyncio.Lock() + + async def record(self, ok: bool) -> None: + async with self.lock: + self.total += 1 + if ok: + self.success += 1 + else: + self.failure += 1 + + async def snapshot(self) -> Dict[str, float]: + async with self.lock: + success_rate = (self.success / self.total * 100.0) if self.total else 0.0 + return { + "success": self.success, + "failure": self.failure, + "total": self.total, + "success_rate": success_rate, + } + + async def resolve_host(host: str) -> str: loop = asyncio.get_running_loop() start = time.time() try: ip = await loop.run_in_executor(None, socket.gethostbyname, host) - elapsed_ms = (time.time() - start) * 1000 + elapsed_ms = (time.time() - start) * 1000.0 logging.debug( "DNS host=%s ip=%s elapsed_ms=%.2f", @@ -53,7 +79,50 @@ def build_pod_hostname(args: argparse.Namespace, node_id: int) -> str: return base -async def resolve_target(args: argparse.Namespace, message_index: int) -> Target: +def parse_target_ids(value: str) -> List[int]: + """ + Parse target IDs. + + Supported formats: + 0 + 0,1,2 + 0-9 + 0-9,20,25-30 + """ + result: List[int] = [] + + for part in value.split(","): + part = part.strip() + + if not part: + continue + + if "-" in part: + start_s, end_s = part.split("-", 1) + start = int(start_s) + end = int(end_s) + + if end < start: + raise ValueError(f"Invalid target ID range: {part}") + + result.extend(range(start, end + 1)) + else: + result.append(int(part)) + + return sorted(set(result)) + + +def parse_target_hosts(value: str) -> List[str]: + return [item.strip() for item in value.split(",") if item.strip()] + + +async def make_target(host: str, port: int) -> Target: + ip = await resolve_host(host) + url = f"http://{ip}:{port}/publish" + return Target(host=host, ip=ip, url=url) + + +async def resolve_target_global(args: argparse.Namespace, message_index: int) -> Target: if args.peer_selection == "service": host = args.service_host @@ -62,45 +131,70 @@ async def resolve_target(args: argparse.Namespace, message_index: int) -> Target elif args.peer_selection == "round-robin": node_count = args.end_id - args.start_id + 1 + + if node_count <= 0: + raise ValueError("--end-id must be >= --start-id for round-robin mode") + node_id = args.start_id + (message_index % node_count) host = build_pod_hostname(args, node_id) elif args.peer_selection == "random-range": + if args.end_id < args.start_id: + raise ValueError("--end-id must be >= --start-id for random-range mode") + node_id = random.randint(args.start_id, args.end_id) host = build_pod_hostname(args, node_id) else: raise ValueError(f"Unsupported peer selection: {args.peer_selection}") - ip = await resolve_host(host) - url = f"http://{ip}:{args.port}/publish" + return await make_target(host, args.port) - return Target(host=host, ip=ip, url=url) + +async def get_per_node_targets(args: argparse.Namespace) -> List[Target]: + if args.target_hosts: + hosts = parse_target_hosts(args.target_hosts) + + elif args.target_ids: + ids = parse_target_ids(args.target_ids) + hosts = [build_pod_hostname(args, node_id) for node_id in ids] + + else: + if args.end_id < args.start_id: + raise ValueError("--end-id must be >= --start-id") + + ids = list(range(args.start_id, args.end_id + 1)) + hosts = [build_pod_hostname(args, node_id) for node_id in ids] + + targets = await asyncio.gather( + *[make_target(host, args.port) for host in hosts] + ) + + logging.info( + "resolved per-node targets count=%d hosts=%s", + len(targets), + ",".join(target.host for target in targets), + ) + + return list(targets) async def send_libp2p_msg( session: aiohttp.ClientSession, args: argparse.Namespace, - stats: Dict[str, int], + stats: Stats, + target: Target, message_index: int, -): - target = await resolve_target(args, message_index) - + worker_id: Optional[int] = None, +) -> None: headers = {"Content-Type": "application/json"} + body = { "topic": args.pubsub_topic, "msgSize": args.msg_size_bytes, "version": 1, } - logging.info( - "message=%d target_host=%s target_ip=%s url=%s", - message_index, - target.host, - target.ip, - target.url, - ) - start = time.time() try: @@ -110,111 +204,216 @@ async def send_libp2p_msg( headers=headers, timeout=args.request_timeout, ) as response: - elapsed_ms = (time.time() - start) * 1000 + elapsed_ms = (time.time() - start) * 1000.0 response_text = await response.text() - stats["total"] += 1 - - if response.status == 200: - stats["success"] += 1 - else: - stats["failure"] += 1 - - success_rate = ( - (stats["success"] / stats["total"]) * 100 - if stats["total"] > 0 - else 0 - ) + ok = response.status == 200 + await stats.record(ok) + snap = await stats.snapshot() logging.info( - "message=%d target=%s status=%d elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f response=%s", + "worker=%s message=%d target_host=%s target_ip=%s status=%d " + "elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f " + "response=%s", + worker_id, message_index, target.host, + target.ip, response.status, elapsed_ms, - stats["success"], - stats["failure"], - stats["total"], - success_rate, + snap["success"], + snap["failure"], + snap["total"], + snap["success_rate"], response_text[:300], ) except Exception as exc: - elapsed_ms = (time.time() - start) * 1000 + elapsed_ms = (time.time() - start) * 1000.0 - stats["total"] += 1 - stats["failure"] += 1 - - success_rate = ( - (stats["success"] / stats["total"]) * 100 - if stats["total"] > 0 - else 0 - ) + await stats.record(False) + snap = await stats.snapshot() logging.warning( - "message=%d target=%s exception=%s elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f", + "worker=%s message=%d target_host=%s target_ip=%s exception=%s " + "elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f", + worker_id, message_index, target.host, + target.ip, repr(exc), elapsed_ms, - stats["success"], - stats["failure"], - stats["total"], - success_rate, + snap["success"], + snap["failure"], + snap["total"], + snap["success_rate"], ) -async def main(args: argparse.Namespace): - stats = { - "success": 0, - "failure": 0, - "total": 0, - } - +async def run_global_mode( + args: argparse.Namespace, + session: aiohttp.ClientSession, + stats: Stats, +) -> None: background_tasks = set() start_time = time.time() message_index = 0 - timeout = aiohttp.ClientTimeout(total=args.request_timeout) + while True: + if args.messages is not None and message_index >= args.messages: + break - async with aiohttp.ClientSession(timeout=timeout) as session: - while True: - if args.messages is not None and message_index >= args.messages: - break + if ( + args.duration_seconds is not None + and time.time() - start_time >= args.duration_seconds + ): + break - if ( - args.duration_seconds is not None - and time.time() - start_time >= args.duration_seconds - ): - break + target = await resolve_target_global(args, message_index) - task = asyncio.create_task( - send_libp2p_msg(session, args, stats, message_index) + task = asyncio.create_task( + send_libp2p_msg( + session=session, + args=args, + stats=stats, + target=target, + message_index=message_index, + worker_id=None, ) + ) - background_tasks.add(task) - task.add_done_callback(background_tasks.discard) + background_tasks.add(task) + task.add_done_callback(background_tasks.discard) - message_index += 1 - await asyncio.sleep(args.delay_seconds) + message_index += 1 - if background_tasks: - await asyncio.gather(*background_tasks) + await asyncio.sleep(args.delay_seconds) - elapsed_s = time.time() - start_time - success_rate = ( - (stats["success"] / stats["total"]) * 100 - if stats["total"] > 0 - else 0 - ) + if background_tasks: + await asyncio.gather(*background_tasks) + + +async def per_node_worker( + worker_id: int, + target: Target, + args: argparse.Namespace, + session: aiohttp.ClientSession, + stats: Stats, +) -> None: + """ + Send messages to one selected node. + + Example: + messages_per_node = 1000 + rate_per_node = 5 + + This worker sends 1000 messages to its target node at 5 msg/s. + + All per-node workers run in parallel. + """ + + if args.rate_per_node is not None: + if args.rate_per_node <= 0: + raise ValueError("--rate-per-node must be > 0") + + delay_seconds = 1.0 / args.rate_per_node + else: + delay_seconds = args.delay_seconds + + start_time = time.time() + message_index = 0 + + while True: + if ( + args.messages_per_node is not None + and message_index >= args.messages_per_node + ): + break + + if ( + args.duration_seconds is not None + and time.time() - start_time >= args.duration_seconds + ): + break + + await send_libp2p_msg( + session=session, + args=args, + stats=stats, + target=target, + message_index=message_index, + worker_id=worker_id, + ) + + message_index += 1 + + await asyncio.sleep(delay_seconds) logging.info( - "finished elapsed_s=%.2f success=%d failure=%d total=%d success_rate=%.2f", + "worker=%d target=%s finished local_messages=%d", + worker_id, + target.host, + message_index, + ) + + +async def run_per_node_mode( + args: argparse.Namespace, + session: aiohttp.ClientSession, + stats: Stats, +) -> None: + targets = await get_per_node_targets(args) + + if not targets: + raise RuntimeError("No targets selected for per-node mode") + + workers = [ + asyncio.create_task( + per_node_worker( + worker_id=i, + target=target, + args=args, + session=session, + stats=stats, + ) + ) + for i, target in enumerate(targets) + ] + + await asyncio.gather(*workers) + + +async def main(args: argparse.Namespace) -> None: + stats = Stats() + start_time = time.time() + + timeout = aiohttp.ClientTimeout(total=args.request_timeout) + + connector = aiohttp.TCPConnector( + ttl_dns_cache=300, + ) + + async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: + if args.load_mode == "global": + await run_global_mode(args, session, stats) + + elif args.load_mode == "per-node": + await run_per_node_mode(args, session, stats) + + else: + raise ValueError(f"Unsupported load mode: {args.load_mode}") + + elapsed_s = time.time() - start_time + snap = await stats.snapshot() + + logging.info( + "finished load_mode=%s elapsed_s=%.2f success=%d failure=%d total=%d success_rate=%.2f", + args.load_mode, elapsed_s, - stats["success"], - stats["failure"], - stats["total"], - success_rate, + snap["success"], + snap["failure"], + snap["total"], + snap["success_rate"], ) @@ -237,79 +436,6 @@ def parse_args() -> argparse.Namespace: help="Message size in bytes", ) - parser.add_argument( - "-d", - "--delay-seconds", - type=float, - default=1.0, - help="Delay between publish requests", - ) - - parser.add_argument( - "-m", - "--messages", - type=int, - default=None, - help="Number of messages to inject", - ) - - parser.add_argument( - "--duration-seconds", - type=float, - default=None, - help="Duration of the injection phase. Use either this or --messages.", - ) - - parser.add_argument( - "--peer-selection", - type=str, - choices=["service", "fixed", "round-robin", "random-range"], - default="service", - help="How to select the test node receiving /publish requests", - ) - - parser.add_argument( - "--service-host", - type=str, - default="nimp2p-service", - help="Kubernetes service hostname for service-based peer selection", - ) - - parser.add_argument( - "--pod-prefix", - type=str, - default="nim-quic-normal", - help="StatefulSet pod prefix for fixed/range selection", - ) - - parser.add_argument( - "--pod-domain", - type=str, - default="nimp2p-service", - help="Headless service DNS domain. Example: nimp2p-service", - ) - - parser.add_argument( - "--fixed-id", - type=int, - default=0, - help="Pod ordinal used with --peer-selection fixed", - ) - - parser.add_argument( - "--start-id", - type=int, - default=0, - help="Start ordinal for round-robin/random-range selection", - ) - - parser.add_argument( - "--end-id", - type=int, - default=99, - help="End ordinal for round-robin/random-range selection", - ) - parser.add_argument( "-p", "--port", @@ -318,28 +444,166 @@ def parse_args() -> argparse.Namespace: help="test node HTTP publish port", ) + parser.add_argument( + "--load-mode", + choices=["global", "per-node"], + default="global", + help=( + "global = one global stream of messages; " + "per-node = one stream per selected target node, all in parallel" + ), + ) + + # Global mode target selection. + parser.add_argument( + "--peer-selection", + choices=["service", "fixed", "round-robin", "random-range"], + default="service", + help="Target selection mode for global mode", + ) + + parser.add_argument( + "--service-host", + type=str, + default="nimp2p-service", + help="Service hostname used by peer-selection=service", + ) + + parser.add_argument( + "--fixed-id", + type=int, + default=0, + help="Target node ID used by peer-selection=fixed", + ) + + parser.add_argument( + "--start-id", + type=int, + default=0, + help="Start node ID for round-robin, random-range, or per-node default target range", + ) + + parser.add_argument( + "--end-id", + type=int, + default=0, + help="End node ID for round-robin, random-range, or per-node default target range", + ) + + # Hostname construction. + parser.add_argument( + "--pod-prefix", + type=str, + default="nim-libp2p", + help="StatefulSet pod prefix, e.g. nim-libp2p", + ) + + parser.add_argument( + "--pod-domain", + type=str, + default="", + help=( + "Optional pod DNS suffix. Example: " + "nimp2p-service" + ), + ) + + # Load amount / rate. + parser.add_argument( + "-m", + "--messages", + type=int, + default=None, + help="Total messages for global mode", + ) + + parser.add_argument( + "--messages-per-node", + type=int, + default=None, + help="Messages sent to each selected node in per-node mode", + ) + + parser.add_argument( + "--duration-seconds", + type=float, + default=None, + help="Run duration. Can be used in global or per-node mode.", + ) + + parser.add_argument( + "-d", + "--delay-seconds", + type=float, + default=1.0, + help=( + "Delay between messages. In per-node mode this is per node " + "unless --rate-per-node is set." + ), + ) + + parser.add_argument( + "--rate-per-node", + type=float, + default=None, + help=( + "Per-node send rate in messages/second. " + "Only applies to per-node mode and overrides --delay-seconds." + ), + ) + + # Per-node target selection. + parser.add_argument( + "--target-ids", + type=str, + default="", + help=( + "Per-node mode target IDs. Examples: " + "0, 0-9, 0-9,20,25-30" + ), + ) + + parser.add_argument( + "--target-hosts", + type=str, + default="", + help=( + "Per-node mode explicit target hostnames, comma-separated. " + "Example: nim-libp2p-slow-0,nim-libp2p-1,nim-libp2p-2" + ), + ) + + # HTTP settings. parser.add_argument( "--request-timeout", type=float, - default=10.0, + default=30.0, help="HTTP request timeout in seconds", ) args = parser.parse_args() - if args.messages is None and args.duration_seconds is None: - parser.error("Set either --messages or --duration-seconds") + if args.load_mode == "global": + if args.messages is None and args.duration_seconds is None: + raise ValueError( + "global mode requires --messages or --duration-seconds" + ) - if args.messages is not None and args.duration_seconds is not None: - parser.error("Use either --messages or --duration-seconds, not both") + if args.load_mode == "per-node": + if args.messages_per_node is None and args.duration_seconds is None: + raise ValueError( + "per-node mode requires --messages-per-node or --duration-seconds" + ) - if args.start_id > args.end_id: - parser.error("--start-id must be <= --end-id") + if args.target_hosts and args.target_ids: + raise ValueError( + "Use either --target-hosts or --target-ids, not both" + ) return args if __name__ == "__main__": parsed_args = parse_args() - logging.info("args=%s", parsed_args) + logging.info("%s", parsed_args) asyncio.run(main(parsed_args)) \ No newline at end of file From e4d6066c3ca83b689adeb25791beb397d5b35ece Mon Sep 17 00:00:00 2001 From: mamoutou-diarra Date: Mon, 25 May 2026 03:32:18 +0000 Subject: [PATCH 8/8] remove mix and publisher --- nim-test-node/gossipsub-queues/.gitignore | 3 + nim-test-node/gossipsub-queues/env.nim | 6 +- nim-test-node/gossipsub-queues/main.nim | 44 -- .../gossipsub-queues/publisher/Dockerfile | 5 - .../gossipsub-queues/publisher/traffic.py | 609 ------------------ 5 files changed, 4 insertions(+), 663 deletions(-) create mode 100644 nim-test-node/gossipsub-queues/.gitignore delete mode 100644 nim-test-node/gossipsub-queues/publisher/Dockerfile delete mode 100644 nim-test-node/gossipsub-queues/publisher/traffic.py diff --git a/nim-test-node/gossipsub-queues/.gitignore b/nim-test-node/gossipsub-queues/.gitignore new file mode 100644 index 0000000..f3685e2 --- /dev/null +++ b/nim-test-node/gossipsub-queues/.gitignore @@ -0,0 +1,3 @@ +nimble.develop +nimble.paths +nimbledeps diff --git a/nim-test-node/gossipsub-queues/env.nim b/nim-test-node/gossipsub-queues/env.nim index a0a653e..9ccdc99 100644 --- a/nim-test-node/gossipsub-queues/env.nim +++ b/nim-test-node/gossipsub-queues/env.nim @@ -3,15 +3,11 @@ import chronos, metrics/chronos_httpserver, chronicles from nativesockets import getHostname let - mountsMix* = existsEnv("MOUNTSMIX") #Full mix-net peer - usesMix* = existsEnv("USESMIX") #Supports sending mix messages - mixCount* = parseInt(getEnv("NUMMIX", "0")) #Number of mix peers (mountsMix + usesMix) inShadow* = getEnv("SHADOWENV").cmpIgnoreCase("true") == 0 #If Running for shadow simulator httpPublishPort* = Port(8645) prometheusPort* = Port(8008) myPort* = Port(5000) chunks* = parseInt(getEnv("FRAGMENTS", "1")) #No. of fragments for each message - mix_D* = parseInt(getEnv("MIXD", "4")) #No. of mix tunnels proc getPeerDetails*(): Result[(int, int, int, string, string, string), string] = @@ -35,7 +31,7 @@ proc getPeerDetails*(): Result[(int, int, int, string, string, string), string] if connectTo >= networkSize: return err("Not enough peers to make target connections. Network size : " & $networkSize) - info "Host info ", hostname = hostname, peer = myId, muxer = muxer, mountsMix = mountsMix, usesMix = usesMix, mixCount = mixCount, inShadow = inShadow, address = address + info "Host info ", hostname = hostname, peer = myId, muxer = muxer, inShadow = inShadow, address = address return ok((myId, networkSize, connectTo, muxer, filePath, address)) diff --git a/nim-test-node/gossipsub-queues/main.nim b/nim-test-node/gossipsub-queues/main.nim index 50dc933..bf35253 100644 --- a/nim-test-node/gossipsub-queues/main.nim +++ b/nim-test-node/gossipsub-queues/main.nim @@ -4,8 +4,6 @@ import env import std/[strformat, random, hashes] import libp2p, libp2p/[muxers/mplex/lpchannel, stream/connection, crypto/secp, multiaddress] import libp2p/protocols/[pubsub/pubsubpeer, pubsub/rpc/messages, ping] -# Mix protocol not available in this libp2p version -# import libp2p/protocols/[mix, mix/mix_protocol] import sequtils, math, metrics, metrics/chronos_httpserver from times import getTime, Time, toUnix, fromUnix, `-`, initTime, `$`, inMilliseconds @@ -178,12 +176,6 @@ proc publishNewMessage(gossipSub: GossipSub, msgSize: int, topic: string): Futur #To support message fragmentation, we add fragment #. Each fragment (chunk) differs by one byte for chunk in 0.. None: - self.success = 0 - self.failure = 0 - self.total = 0 - self.lock = asyncio.Lock() - - async def record(self, ok: bool) -> None: - async with self.lock: - self.total += 1 - if ok: - self.success += 1 - else: - self.failure += 1 - - async def snapshot(self) -> Dict[str, float]: - async with self.lock: - success_rate = (self.success / self.total * 100.0) if self.total else 0.0 - return { - "success": self.success, - "failure": self.failure, - "total": self.total, - "success_rate": success_rate, - } - - -async def resolve_host(host: str) -> str: - loop = asyncio.get_running_loop() - start = time.time() - - try: - ip = await loop.run_in_executor(None, socket.gethostbyname, host) - elapsed_ms = (time.time() - start) * 1000.0 - - logging.debug( - "DNS host=%s ip=%s elapsed_ms=%.2f", - host, - ip, - elapsed_ms, - ) - - return ip - - except (socket.gaierror, socket.herror, OSError) as exc: - raise RuntimeError(f"DNS lookup failed for host={host}: {exc}") from exc - - -def build_pod_hostname(args: argparse.Namespace, node_id: int) -> str: - base = f"{args.pod_prefix}-{node_id}" - - if args.pod_domain: - return f"{base}.{args.pod_domain}" - - return base - - -def parse_target_ids(value: str) -> List[int]: - """ - Parse target IDs. - - Supported formats: - 0 - 0,1,2 - 0-9 - 0-9,20,25-30 - """ - result: List[int] = [] - - for part in value.split(","): - part = part.strip() - - if not part: - continue - - if "-" in part: - start_s, end_s = part.split("-", 1) - start = int(start_s) - end = int(end_s) - - if end < start: - raise ValueError(f"Invalid target ID range: {part}") - - result.extend(range(start, end + 1)) - else: - result.append(int(part)) - - return sorted(set(result)) - - -def parse_target_hosts(value: str) -> List[str]: - return [item.strip() for item in value.split(",") if item.strip()] - - -async def make_target(host: str, port: int) -> Target: - ip = await resolve_host(host) - url = f"http://{ip}:{port}/publish" - return Target(host=host, ip=ip, url=url) - - -async def resolve_target_global(args: argparse.Namespace, message_index: int) -> Target: - if args.peer_selection == "service": - host = args.service_host - - elif args.peer_selection == "fixed": - host = build_pod_hostname(args, args.fixed_id) - - elif args.peer_selection == "round-robin": - node_count = args.end_id - args.start_id + 1 - - if node_count <= 0: - raise ValueError("--end-id must be >= --start-id for round-robin mode") - - node_id = args.start_id + (message_index % node_count) - host = build_pod_hostname(args, node_id) - - elif args.peer_selection == "random-range": - if args.end_id < args.start_id: - raise ValueError("--end-id must be >= --start-id for random-range mode") - - node_id = random.randint(args.start_id, args.end_id) - host = build_pod_hostname(args, node_id) - - else: - raise ValueError(f"Unsupported peer selection: {args.peer_selection}") - - return await make_target(host, args.port) - - -async def get_per_node_targets(args: argparse.Namespace) -> List[Target]: - if args.target_hosts: - hosts = parse_target_hosts(args.target_hosts) - - elif args.target_ids: - ids = parse_target_ids(args.target_ids) - hosts = [build_pod_hostname(args, node_id) for node_id in ids] - - else: - if args.end_id < args.start_id: - raise ValueError("--end-id must be >= --start-id") - - ids = list(range(args.start_id, args.end_id + 1)) - hosts = [build_pod_hostname(args, node_id) for node_id in ids] - - targets = await asyncio.gather( - *[make_target(host, args.port) for host in hosts] - ) - - logging.info( - "resolved per-node targets count=%d hosts=%s", - len(targets), - ",".join(target.host for target in targets), - ) - - return list(targets) - - -async def send_libp2p_msg( - session: aiohttp.ClientSession, - args: argparse.Namespace, - stats: Stats, - target: Target, - message_index: int, - worker_id: Optional[int] = None, -) -> None: - headers = {"Content-Type": "application/json"} - - body = { - "topic": args.pubsub_topic, - "msgSize": args.msg_size_bytes, - "version": 1, - } - - start = time.time() - - try: - async with session.post( - target.url, - json=body, - headers=headers, - timeout=args.request_timeout, - ) as response: - elapsed_ms = (time.time() - start) * 1000.0 - response_text = await response.text() - - ok = response.status == 200 - await stats.record(ok) - snap = await stats.snapshot() - - logging.info( - "worker=%s message=%d target_host=%s target_ip=%s status=%d " - "elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f " - "response=%s", - worker_id, - message_index, - target.host, - target.ip, - response.status, - elapsed_ms, - snap["success"], - snap["failure"], - snap["total"], - snap["success_rate"], - response_text[:300], - ) - - except Exception as exc: - elapsed_ms = (time.time() - start) * 1000.0 - - await stats.record(False) - snap = await stats.snapshot() - - logging.warning( - "worker=%s message=%d target_host=%s target_ip=%s exception=%s " - "elapsed_ms=%.2f success=%d failure=%d total=%d success_rate=%.2f", - worker_id, - message_index, - target.host, - target.ip, - repr(exc), - elapsed_ms, - snap["success"], - snap["failure"], - snap["total"], - snap["success_rate"], - ) - - -async def run_global_mode( - args: argparse.Namespace, - session: aiohttp.ClientSession, - stats: Stats, -) -> None: - background_tasks = set() - start_time = time.time() - message_index = 0 - - while True: - if args.messages is not None and message_index >= args.messages: - break - - if ( - args.duration_seconds is not None - and time.time() - start_time >= args.duration_seconds - ): - break - - target = await resolve_target_global(args, message_index) - - task = asyncio.create_task( - send_libp2p_msg( - session=session, - args=args, - stats=stats, - target=target, - message_index=message_index, - worker_id=None, - ) - ) - - background_tasks.add(task) - task.add_done_callback(background_tasks.discard) - - message_index += 1 - - await asyncio.sleep(args.delay_seconds) - - if background_tasks: - await asyncio.gather(*background_tasks) - - -async def per_node_worker( - worker_id: int, - target: Target, - args: argparse.Namespace, - session: aiohttp.ClientSession, - stats: Stats, -) -> None: - """ - Send messages to one selected node. - - Example: - messages_per_node = 1000 - rate_per_node = 5 - - This worker sends 1000 messages to its target node at 5 msg/s. - - All per-node workers run in parallel. - """ - - if args.rate_per_node is not None: - if args.rate_per_node <= 0: - raise ValueError("--rate-per-node must be > 0") - - delay_seconds = 1.0 / args.rate_per_node - else: - delay_seconds = args.delay_seconds - - start_time = time.time() - message_index = 0 - - while True: - if ( - args.messages_per_node is not None - and message_index >= args.messages_per_node - ): - break - - if ( - args.duration_seconds is not None - and time.time() - start_time >= args.duration_seconds - ): - break - - await send_libp2p_msg( - session=session, - args=args, - stats=stats, - target=target, - message_index=message_index, - worker_id=worker_id, - ) - - message_index += 1 - - await asyncio.sleep(delay_seconds) - - logging.info( - "worker=%d target=%s finished local_messages=%d", - worker_id, - target.host, - message_index, - ) - - -async def run_per_node_mode( - args: argparse.Namespace, - session: aiohttp.ClientSession, - stats: Stats, -) -> None: - targets = await get_per_node_targets(args) - - if not targets: - raise RuntimeError("No targets selected for per-node mode") - - workers = [ - asyncio.create_task( - per_node_worker( - worker_id=i, - target=target, - args=args, - session=session, - stats=stats, - ) - ) - for i, target in enumerate(targets) - ] - - await asyncio.gather(*workers) - - -async def main(args: argparse.Namespace) -> None: - stats = Stats() - start_time = time.time() - - timeout = aiohttp.ClientTimeout(total=args.request_timeout) - - connector = aiohttp.TCPConnector( - ttl_dns_cache=300, - ) - - async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: - if args.load_mode == "global": - await run_global_mode(args, session, stats) - - elif args.load_mode == "per-node": - await run_per_node_mode(args, session, stats) - - else: - raise ValueError(f"Unsupported load mode: {args.load_mode}") - - elapsed_s = time.time() - start_time - snap = await stats.snapshot() - - logging.info( - "finished load_mode=%s elapsed_s=%.2f success=%d failure=%d total=%d success_rate=%.2f", - args.load_mode, - elapsed_s, - snap["success"], - snap["failure"], - snap["total"], - snap["success_rate"], - ) - - -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="nim-libp2p message injector") - - parser.add_argument( - "-t", - "--pubsub-topic", - type=str, - default="test", - help="PubSub topic", - ) - - parser.add_argument( - "-s", - "--msg-size-bytes", - type=int, - default=1000, - help="Message size in bytes", - ) - - parser.add_argument( - "-p", - "--port", - type=int, - default=8645, - help="test node HTTP publish port", - ) - - parser.add_argument( - "--load-mode", - choices=["global", "per-node"], - default="global", - help=( - "global = one global stream of messages; " - "per-node = one stream per selected target node, all in parallel" - ), - ) - - # Global mode target selection. - parser.add_argument( - "--peer-selection", - choices=["service", "fixed", "round-robin", "random-range"], - default="service", - help="Target selection mode for global mode", - ) - - parser.add_argument( - "--service-host", - type=str, - default="nimp2p-service", - help="Service hostname used by peer-selection=service", - ) - - parser.add_argument( - "--fixed-id", - type=int, - default=0, - help="Target node ID used by peer-selection=fixed", - ) - - parser.add_argument( - "--start-id", - type=int, - default=0, - help="Start node ID for round-robin, random-range, or per-node default target range", - ) - - parser.add_argument( - "--end-id", - type=int, - default=0, - help="End node ID for round-robin, random-range, or per-node default target range", - ) - - # Hostname construction. - parser.add_argument( - "--pod-prefix", - type=str, - default="nim-libp2p", - help="StatefulSet pod prefix, e.g. nim-libp2p", - ) - - parser.add_argument( - "--pod-domain", - type=str, - default="", - help=( - "Optional pod DNS suffix. Example: " - "nimp2p-service" - ), - ) - - # Load amount / rate. - parser.add_argument( - "-m", - "--messages", - type=int, - default=None, - help="Total messages for global mode", - ) - - parser.add_argument( - "--messages-per-node", - type=int, - default=None, - help="Messages sent to each selected node in per-node mode", - ) - - parser.add_argument( - "--duration-seconds", - type=float, - default=None, - help="Run duration. Can be used in global or per-node mode.", - ) - - parser.add_argument( - "-d", - "--delay-seconds", - type=float, - default=1.0, - help=( - "Delay between messages. In per-node mode this is per node " - "unless --rate-per-node is set." - ), - ) - - parser.add_argument( - "--rate-per-node", - type=float, - default=None, - help=( - "Per-node send rate in messages/second. " - "Only applies to per-node mode and overrides --delay-seconds." - ), - ) - - # Per-node target selection. - parser.add_argument( - "--target-ids", - type=str, - default="", - help=( - "Per-node mode target IDs. Examples: " - "0, 0-9, 0-9,20,25-30" - ), - ) - - parser.add_argument( - "--target-hosts", - type=str, - default="", - help=( - "Per-node mode explicit target hostnames, comma-separated. " - "Example: nim-libp2p-slow-0,nim-libp2p-1,nim-libp2p-2" - ), - ) - - # HTTP settings. - parser.add_argument( - "--request-timeout", - type=float, - default=30.0, - help="HTTP request timeout in seconds", - ) - - args = parser.parse_args() - - if args.load_mode == "global": - if args.messages is None and args.duration_seconds is None: - raise ValueError( - "global mode requires --messages or --duration-seconds" - ) - - if args.load_mode == "per-node": - if args.messages_per_node is None and args.duration_seconds is None: - raise ValueError( - "per-node mode requires --messages-per-node or --duration-seconds" - ) - - if args.target_hosts and args.target_ids: - raise ValueError( - "Use either --target-hosts or --target-ids, not both" - ) - - return args - - -if __name__ == "__main__": - parsed_args = parse_args() - logging.info("%s", parsed_args) - asyncio.run(main(parsed_args)) \ No newline at end of file