Central Controller: ECMP Collision-Free Routing

Mode A — TraceInjector (recommended for CC experiments)

Bypasses AstraSim entirely. Inject flows from a CSV trace file.

# 1. Generate trace (server 0 GPUs → server 1 GPUs, 10 MB each)
python3 gen_kv_trace.py \
    --arrival server_pair --src-server 0 --dst-server 1 \
    --gpus-per-server 8 --kv-mb 10 --rounds 1 \
    --out traces/my_trace.csv

# 2a. Run WITH CC
CC_ENABLE=1 CC_LOG_FILE=results/cc/paths.csv \
  TRACE_INJECT_FILE=traces/my_trace.csv \
  AS_SEND_LAT=3 ./bin/SimAI_simulator -t 1 \
  -n <topology_dir> -c results/cc/SimAI_local.conf

# 2b. Run WITHOUT CC (A/B comparison)
CC_ENABLE=0 \
  TRACE_INJECT_FILE=traces/my_trace.csv \
  AS_SEND_LAT=3 ./bin/SimAI_simulator -t 1 \
  -n <topology_dir> -c results/nocc/SimAI_local.conf

Mode B — AstraSim Workload (full collective communication)

Run real collective workloads (AllReduce, AllToAll, etc.) through AstraSim → SimCCL → NS-3.

CC_ENABLE=1 CC_LOG_FILE=results/cc/paths.csv \
  AS_SEND_LAT=3 AS_NVLS_ENABLE=1 ./bin/SimAI_simulator -t 1 \
  -w workloads/test_cc_compare.txt \
  -n <topology_dir> \
  -c results/cc/SimAI_local.conf

AstraSimNetwork.cc — Main entry point + TraceInjector

Contains main() and the ASTRASimNetwork class that bridges AstraSim ↔ NS-3.

Execution Order

main(argc, argv)
  ① user_param_prase()         // parse -t thread -w workload -n topo -c conf
  ② main1(topo, conf)           // → common.h: ReadConf → SetConfig → SetupNetwork
                                  //   (builds NS-3 topology, initializes CentralController)
  ③ if env TRACE_INJECT_FILE:   // ← NEW: TraceInjector bypass mode
      for line in csv:
        Simulator::Schedule(NanoSeconds(t), &SendFlowTrace, src, dst, size)
      Simulator::Run()           // run NS-3 with injected flows only
      return
  ④ // Normal mode: create ASTRASimNetwork + AstraSim::Sys per GPU
     for j in nodes_num:
       networks[j] = new ASTRASimNetwork(j, 0)
       systems[j]  = new AstraSim::Sys(networks[j], ...)
     for i in nodes_num:
       systems[i]->workload->fire()  // kick off collective workloads
     Simulator::Run()

ASTRASimNetwork class — AstraSim ↔ NS-3 bridge

entry.h — SendFlow + CentralController integration

Defines the flow injection functions and all flow completion callbacks. This is where the CentralController's AssignSport() is called for every flow.

SendFlow() — normal mode (called by AstraSim)

void SendFlow(src, dst, maxPacketCount, msg_handler, fun_arg, tag, request) {
  port = portNumber[src][dst]++;

  // ← NEW: CentralController picks a conflict-free sport
  port = CentralController::Instance().AssignSport(
      serverAddress[src], serverAddress[dst],
      100 /*dport*/, src, dst, port);

  sender_src_port_map[{port, {src, dst}}] = request->flowTag;

  RdmaClientHelper clientHelper(pg, srcIP, dstIP, port, dport, size, ...);
  appCon = clientHelper.Install(n.Get(src));
  appCon.Start(Time(send_lat));  // inject RDMA QP into NS-3
}

SendFlowTrace() — TraceInjector mode (no AstraSim context)

void SendFlowTrace(src, dst, size_bytes) {
  port = portNumber[src][dst]++;
  port = CentralController::Instance().AssignSport(...);

  AstraSim::ncclFlowTag sentinel;  // current_flow_id = -1 (sentinel)
  sender_src_port_map[{port, {src, dst}}] = sentinel;

  // Same RDMA QP injection, but no AstraSim callbacks
  RdmaClientHelper clientHelper(..., nullptr, nullptr, ...);
  appCon.Start(Time(send_lat));
}

Completion Callbacks

common.h — Network setup + CC initialization

The infrastructure glue file. Defines global state, reads config, builds the NS-3 topology, and initializes the CentralController.

Three main functions, called in order by main1()

main1(topo, conf)
  ① ReadConf(topo, conf)    // Parse config: CC_MODE, DATA_RATE, LINK_DELAY, PFC, DCQCN params...
  ② SetConfig()             // Apply NS-3 defaults (PauseTime, QcnEnabled, IntHeader mode)
                             //   cc_mode=3 → DCQCN, cc_mode=7 → TIMELY, cc_mode=10 → PINT
  ③ SetupNetwork(qp_finish, send_finish)
       // a. Read topology file: node_num, gpus_per_server, nvswitch_num, switch_num
       // b. Create NS-3 nodes (hosts, SwitchNode, NVSwitchNode)
       // c. Install point-to-point links with QBB (lossless Ethernet)
       // d. Configure RDMA per host (RdmaHw with CC mode, rates, PFC)
       // e. CalculateRoutes() → SetRoutingEntries()
       // f. NEW: Initialize CentralController ↓

CentralController initialization (added in SetupNetwork)

// After SetRoutingEntries():
CentralController::Instance().Initialize(n, nodeIdToRdmaHw, ipToNodeId, gpus_per_server);

// Env var control:
if (env "CC_LOG_FILE")   CC.SetLogFile(path);   // CSV: flow_id,sip,dip,sport,hops
if (env "CC_ENABLE" != "0") CC.Enable();  // default: enabled
else                       CC.Disable(); // for A/B comparison

central-controller.h — CC class declaration

Singleton class with the full API and internal data structures for conflict-free path assignment.

Public API

Internal Data Structures

central-controller.cc — CC core logic

AssignSport() — the hot path (called per flow)

AssignSport(sip, dip, dport, srcNodeId, dstNodeId, fallbackSport) {
  if (!m_enabled || !m_initialized) return fallbackSport;
  if (srcNodeId/gpusPerServer == dstNodeId/gpusPerServer) return fallbackSport;
      // ↑ intra-server → NVSwitch, no ECMP, CC can't help

  // Quick check: if probe path has ≤1 hop, CC is useless
  Path probe = ComputePath(sip, dip, fallbackSport, dport, ...);
  if (probe.size() <= 1) return fallbackSport;

  // Exhaustive search: try sport 1..65534
  for (sport = 1; sport < 65535; sport++) {
    Path path = ComputePath(sip, dip, sport, dport, ...);
    if (!HasConflict(path)) {
      ReservePath(path, {sip, dip, sport, dport});
      return sport;  // ← conflict-free sport found
    }
  }
  return fallbackSport;  // no conflict-free sport exists
}

ComputePath() — deterministic ECMP replay

ComputePath(sip, dip, sport, dport, srcNodeId, dstNodeId) {
  buf = {sip, dip, (sport | dport<<16)};  // 12-byte hash input

  // Step 1: NIC selection at host (same hash as rdma-hw.cc)
  h = Murmur3(buf, 12, 0x8BADF00D);
  nicIdx = hostNics[h % hostNics.size()];
  currentNode = nicToSwitch[nicIdx];

  // Step 2: switch-by-switch ECMP traversal
  while (currentNode is switch) {
    h = Murmur3(buf, 12, switch.ecmpSeed);  // ← per-switch seed
    portIdx = switch.rtTable[dip][h % numPorts];
    path.push_back({currentNode, portIdx});
    currentNode = adjacency[currentNode][portIdx];
  }
  return path;
}

HasConflict() + ReservePath() / ReleasePath()

HasConflict(path) {
  // Check all hops EXCEPT the last one (leaf→GPU has no diversity)
  for (i = 0; i + 1 < path.size(); i++)
    if (m_hopToFlows[{path[i].node, path[i].port}].size() > 0)
      return true;
  return false;
}

ReservePath(path, flowKey) {
  m_flowPaths[flowKey] = path;
  for (i = 0; i + 1 < path.size(); i++)
    m_hopToFlows[{path[i].node, path[i].port}].insert(flowKey);
}

ReleasePath(sip, dip, sport, dport) {  // called from rdma-hw.cc QpComplete()
  flowKey = {sip, dip, sport, dport};
  path = m_flowPaths[flowKey];
  for hop in path:
    m_hopToFlows[hop].erase(flowKey);
  m_flowPaths.erase(flowKey);
}