remove mix and publisher

nim-test-node/gossipsub-queues/.gitignore

@@ -0,0 +1,3 @@
+nimble.develop
+nimble.paths
+nimbledeps

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))
 

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..<chunks:
     nowBytes[16] = byte(chunk)
-    # Mix protocol not available in this libp2p version
-    # res = if mountsMix:
-    #   await gossipSub.publish(topic, nowBytes,
-    #     publishParams = Opt.some(PublishParams(skipMCache: true, useCustomConn: true)),
-    #   )
-    # else:
     res = await gossipSub.publish(topic, nowBytes)
   info "publish result", 
     res = res,
@@ -247,8 +239,6 @@ proc startHttpServer(gossipSub: GossipSub, myId: int): Future[HttpServerRef] {.a
   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,
@@ -257,14 +247,6 @@ proc initializeGossipsub(switch: Switch, anonymize: bool): GossipSub =
       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) =
@@ -440,28 +422,12 @@ proc main {.async.} =
   
   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()
@@ -478,16 +444,6 @@ proc main {.async.} =
   
   # Set peerId for metric labels
   gPeerId = $switch.peerInfo.peerId
-
-  # 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)

nim-test-node/gossipsub-queues/publisher/Dockerfile

@@ -1,5 +0,0 @@
-FROM python:3.10.16-alpine3.21
-
-ADD ./traffic.py /app/traffic.py
-
-RUN pip install requests argparse aiohttp

nim-test-node/gossipsub-queues/publisher/traffic.py

@@ -1,609 +0,0 @@
-import argparse
-import asyncio
-import logging
-import random
-import socket
-import time
-from dataclasses import dataclass
-from typing import Dict, List, 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
-
-
-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.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))