chore: update mix

Author: richard-ramos

Dockerfile

@@ -39,17 +39,18 @@ RUN apt-get update && apt install -y git build-essential bash ca-certificates li
 RUN git config --global http.sslVerify false
 
 # Install latest nimble version
-RUN nimble install nimble@#head
+RUN nimble install nimble@0.20.1
 
 # Copy only files needed to install Nimble deps (optimizes layer caching)
 COPY test_node.nimble .
+
 RUN nimble install -y --depsOnly.
 
 # Copy full source AFTER deps are cached
 COPY . .
 
 # Compile the Nim application
-RUN nimble c -d:chronicles_colors=None --threads:on -d:metrics -d:libp2p_network_protocols_metrics  -d:release main
+RUN nimble c -d:chronicles_colors=None --threads:on -d:metrics -d:libp2p_network_protocols_metrics -d:enable_mix_benchmarks  -d:release main
 
 # =============================================================================
 # Run the app
@@ -65,7 +66,7 @@ WORKDIR /node
 # Copy the compiled binary from the build stage
 COPY --from=build_app /node/main /node/main
 
-COPY cron_runner.sh .
+COPY ./cron_runner.sh .
 
 RUN chmod +x cron_runner.sh
 RUN chmod +x main

main.nim

@@ -1,9 +1,13 @@
-import chronos, chronicles, hashes, math, sequtils, strutils, tables, os
+import chronos, chronicles, results
 import metrics, metrics/chronos_httpserver
 import stew/[byteutils, endians2]
-import std/[options, strformat, random, posix]
-import node
+import
+  std/[
+    strformat, random, posix, hashes, math, sequtils, strutils, tables, os,
+    nativesockets,
+  ]
 import mix
+import ./node
 import
   libp2p,
   libp2p/[
@@ -13,10 +17,7 @@ import
     protocols/pubsub/gossipsub,
     protocols/pubsub/pubsubpeer,
     protocols/pubsub/rpc/messages,
-    transports/tcptransport,
   ]
-from times import getTime, toUnix, fromUnix, `-`, initTime, `$`, inMilliseconds
-from nativesockets import getHostname
 
 const D* = 4 # No. of peers to forward to
 
@@ -44,8 +45,12 @@ proc createSwitch(id, port: int, isMix: bool, filePath: string): Switch =
       libp2pPubKey: SkPublicKey
       libp2pPrivKey: SkPrivateKey
 
+    var mixNodes: MixNodes = @[]
+
     if isMix:
-      discard initializeMixNodes(1, port)
+      mixNodes = initializeMixNodes(1, port).valueOr:
+        error "Could not generate mix nodes"
+        return
 
       let mixNodeInfo = getMixNodeInfo(mixNodes[0])
       multiAddrStr = mixNodeInfo[0]
@@ -86,14 +91,17 @@ proc createSwitch(id, port: int, isMix: bool, filePath: string): Switch =
       externalMultiAddr = fmt"/ip4/{externalAddr}/tcp/{port}/p2p/{peerId}"
 
     if isMix:
-      discard initMixMultiAddrByIndex(0, externalMultiAddr)
+      let initRes = mixNodes.initMixMultiAddrByIndex(0, externalMultiAddr)
+      if initRes.isErr:
+        error "Failed to initialize mix node", id = 0, err = initRes.error
+        return
       let writeNodeRes =
         writeMixNodeInfoToFile(mixNodes[0], id, filePath / fmt"nodeInfo")
       if writeNodeRes.isErr:
         error "Failed to write mix info to file", nodeId = id, err = writeNodeRes.error
         return
 
-      let nodePubInfo = getMixPubInfoByIndex(0).valueOr:
+      let nodePubInfo = mixNodes.getMixPubInfoByIndex(0).valueOr:
         error "Get mix pub info by index error", err = error
         return
 
@@ -132,6 +140,19 @@ proc startMetricsServer(
 
   ok(server)
 
+proc makeMixConnCb(mixProto: MixProtocol): CustomConnCreationProc =
+  return proc(
+      destAddr: Option[MultiAddress], destPeerId: PeerId, codec: string
+  ): Connection {.gcsafe, raises: [].} =
+    try:
+      let dest = destAddr.valueOr:
+        debug "No destination address available"
+        return
+      return mixProto.toConnection(MixDestination.init(destPeerId, dest), codec).get()
+    except CatchableError as e:
+      error "Error during execution of MixEntryConnection callback: ", err = e.msg
+      return nil
+
 proc main() {.async.} =
   randomize()
 
@@ -171,14 +192,7 @@ proc main() {.async.} =
         "could not instantiate mix"
       )
 
-    let mixConn = proc(
-        destAddr: Option[MultiAddress], destPeerId: PeerId, codec: string
-    ): Connection {.gcsafe, raises: [].} =
-      try:
-        return mixProto.toConnection(destPeerId, Opt.none(MultiAddress), codec)
-      except CatchableError as e:
-        error "Error during execution of MixEntryConnection callback", err = e.msg
-        return nil
+    let mixConn = makeMixConnCb(mixProto)
 
     let mixPeerSelect = proc(
         allPeers: HashSet[PubSubPeer],
@@ -189,7 +203,7 @@ proc main() {.async.} =
       try:
         return mixPeerSelection(allPeers, directPeers, meshPeers, fanoutPeers)
       except CatchableError as e:
-        error "Error during execution of MixPeerSelection callback", err = e.msg
+        error "Error during execution of MixPeerSelection callback: ", err = e.msg
         return initHashSet[PubSubPeer]()
 
     gossipSub = GossipSub.init(
@@ -245,14 +259,12 @@ proc main() {.async.} =
     let
       timestampNs = uint64.fromBytesLE(data[0 ..< 8])
       msgId = uint64.fromBytesLE(data[8 ..< 16])
-      sentMoment = nanoseconds(int64(timestampNs))
-      sentNanosecs = nanoseconds(sentMoment - seconds(sentMoment.seconds))
-      sentDate = initTime(sentMoment.seconds, sentNanosecs)
-      recvTime = getTime()
-      delay = recvTime - sentDate
+      sentTime = Moment.init(int64(timestampNs), Nanosecond)
+      recvTime = Moment.now()
+      delay = recvTime - sentTime
 
     info "Received message",
-      msgId = msgId, sentAt = timestampNs, delayMs = delay.inMilliseconds()
+      msgId = msgId, sentAt = timestampNs, delayMs = delay.milliseconds()
 
   proc messageValidator(
       topic: string, msg: Message
@@ -312,12 +324,11 @@ proc main() {.async.} =
   for msg in 0 ..< messages: #client.param(int, "message_count"):
     await sleepAsync(msg_rate.milliseconds)
     if msg mod publisherCount == myId:
-      let now = getTime()
-      let timestampNs = now.toUnix().int64 * 1_000_000_000 + times.nanosecond(now).int64
+      let timestampNs = Moment.now().epochNanoSeconds()
       let msgId = uint64(msg)
 
       var payload: seq[byte]
-      payload.add(toBytesLE(uint64(timestampNs)))
+      payload.add(toBytesLE(timestampNs.uint64))
       payload.add(toBytesLE(msgId))
       payload.add(newSeq[byte](msg_size - 16)) # Fill the rest with padding
 

mixrunner.sh

@@ -1,3 +1,4 @@
+
 #!/bin/bash
 
 N="$1"
@@ -15,7 +16,7 @@ if [[ "$(ls -A "$DATADIR")" ]]; then
 fi
 
 for i in $(seq 0 $((N-1))); do
-  docker run --rm \
+  docker run \
     -d \
     --name node-$i \
     --hostname node-$i \
@@ -28,4 +29,4 @@ for i in $(seq 0 $((N-1))); do
     -e CONNECTTO=4 \
     --entrypoint /node/main \
     mixrunner
-done
+done

node.nim

@@ -119,7 +119,7 @@ proc readPubInfoFromFile*(
     let data = file.readAll()
     if data.len != PubInfoSize:
       return err(
-        "Invalid data size for NodeInfo: expected " & $NodeInfoSize & " bytes, but got " &
+        "Invalid data size for PubInfo: expected " & $PubInfoSize & " bytes, but got " &
           $(data.len) & " bytes."
       )
     let dPubInfo = deserializePubInfo(cast[seq[byte]](data)).valueOr:
@@ -152,11 +152,9 @@ proc generateNodes(count: int, basePort: int = 4242): Result[seq[NodeInfo], stri
       pubKeyProto = PublicKey(scheme: Secp256k1, skkey: libp2pPubKey)
       peerId = PeerId.init(pubKeyProto).get()
       multiAddr = fmt"/ip4/0.0.0.0/tcp/{basePort + i}/p2p/{peerId}"
-
     nodes[i] = NodeInfo(
       multiAddr: multiAddr, libp2pPubKey: libp2pPubKey, libp2pPrivKey: libp2pPrivKey
     )
-
   ok(nodes)
 
 proc initializeNodes*(count: int, basePort: int = 4242): Result[void, string] =

test_node.nimble

@@ -1,14 +1,12 @@
 mode = ScriptMode.Verbose
 
-packageName   = "test_node"
-version       = "0.1.0"
-author        = "Status Research & Development GmbH"
-description   = "A test node for gossipsub"
-license       = "MIT"
-skipDirs      = @[]
+packageName = "test_node"
+version = "0.1.0"
+author = "Status Research & Development GmbH"
+description = "A test node for gossipsub"
+license = "MIT"
+skipDirs = @[]
 
 requires "nim >= 1.6.0",
-         "https://github.com/vacp2p/nim-libp2p#master",
-         "https://github.com/vacp2p/mix#tmp/benchmark-logging",
-         "chronos",
-         "ggplotnim"
+  "https://github.com/vacp2p/nim-libp2p#master", "https://github.com/vacp2p/mix#main",
+  "chronos", "ggplotnim"

utils.nim

@@ -0,0 +1,21 @@
+import std/[sequtils, sets]
+import libp2p/[protocols/pubsub/pubsubpeer, switch]
+
+const D* = 4 # No. of peers to forward to
+
+proc mixPeerSelection*(
+    allPeers: HashSet[PubSubPeer],
+    directPeers: HashSet[PubSubPeer],
+    meshPeers: HashSet[PubSubPeer],
+    fanoutPeers: HashSet[PubSubPeer],
+): HashSet[PubSubPeer] {.gcsafe, raises: [].} =
+  var
+    peers: HashSet[PubSubPeer]
+    allPeersSeq = allPeers.toSeq()
+  let rng = newRng()
+  rng.shuffle(allPeersSeq)
+  for p in allPeersSeq:
+    peers.incl(p)
+    if peers.len >= D:
+      break
+  return peers