diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 04ccc527..72aea3a2 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -409,9 +409,6 @@ jobs: ref: ${{ github.event_name == 'schedule' && 'master' || env.ROSS_REF }} path: ross - # No dependency-install step: the codes-ci-ubuntu24-mpich container already ships - # the toolchain, cmake/ninja, flex/bison, lcov, and our source-built mpich. - - name: Configure ROSS run: > cmake -S ross -B ross/build -G Ninja diff --git a/doc/example/CMakeLists.txt b/doc/example/CMakeLists.txt index 1b0c5605..78d15ab8 100644 --- a/doc/example/CMakeLists.txt +++ b/doc/example/CMakeLists.txt @@ -14,6 +14,7 @@ configure_file(tutorial-ping-pong-surrogate.conf.in tutorial-ping-pong-surrogate configure_file(kb.dfdally-72-zeromq-director.conf.in kb.dfdally-72-zeromq-director.template.conf.in @ONLY) configure_file(kb.dfdally-72-event-time-director.conf.in kb.dfdally-72-event-time-director.template.conf.in @ONLY) configure_file(kb.dfdally-72-milc-small.workload.conf.in kb.dfdally-72-milc-small.workload.template.conf.in @ONLY) +configure_file(fluid-flow-wan.conf.in fluid-flow-wan.template.conf.in @ONLY) configure_file(kb.dfdally-72-zeromq-director.conf.in kb.dfdally-72-zeromq-director.template.conf.in @ONLY) set(single_quote "'") @@ -42,6 +43,8 @@ configure_file(tutorial-ping-pong-surrogate.conf.in tutorial-ping-pong-surrogate configure_file(kb.dfdally-72-zeromq-director.conf.in kb.dfdally-72-zeromq-director.conf) configure_file(kb.dfdally-72-event-time-director.conf.in kb.dfdally-72-event-time-director.conf) configure_file(kb.dfdally-72-milc-small.workload.conf.in kb.dfdally-72-milc-small.workload.conf) +configure_file(fluid-flow-wan.conf.in fluid-flow-wan.conf) configure_file(kb.dfdally-72-milc-small.json kb.dfdally-72-milc-small.json COPYONLY) configure_file(kb.dfdally-72-milc-small.alloc.conf kb.dfdally-72-milc-small.alloc.conf COPYONLY) +configure_file(fluid-flow-wan-topology.yaml fluid-flow-wan-topology.yaml COPYONLY) configure_file(kb.dfdally-72-zeromq-director.conf.in kb.dfdally-72-zeromq-director.conf) diff --git a/doc/example/fluid-flow-wan-topology.yaml b/doc/example/fluid-flow-wan-topology.yaml new file mode 100644 index 00000000..0e3040be --- /dev/null +++ b/doc/example/fluid-flow-wan-topology.yaml @@ -0,0 +1,24 @@ +topology: + switches: + A: + terminals: 2 + terminal_bandwidth: "20 Mbps" + switch_buffer: "1000 Mb" + connections: + B: "15 Mbps" + + B: + terminals: 2 + terminal_bandwidth: "25 Mbps" + switch_buffer: "1000 Mb" + connections: + A: "15 Mbps" + C: "10 Mbps" + + C: + terminals: 2 + terminal_bandwidth: "20 Mbps" + switch_buffer: "1000 Mb" + connections: + A: "10 Mbps" + B: "15 Mbps" diff --git a/doc/example/fluid-flow-wan.conf.in b/doc/example/fluid-flow-wan.conf.in new file mode 100644 index 00000000..22b2b256 --- /dev/null +++ b/doc/example/fluid-flow-wan.conf.in @@ -0,0 +1,45 @@ +# Interval-fluid switch/terminal pure-PDES example. +# Run from the build directory, for example: +# mpirun -np 1 ./src/model-net-fluid-flow-wan --synch=1 -- doc/example/fluid-flow-wan.conf + +LPGROUPS +{ + FLUID_FLOW_WAN_GRP + { + repetitions="1"; + fluid-flow-wan-switch-lp="3"; + fluid-flow-wan-terminal-lp="6"; + } +} + +PARAMS +{ + message_size="32768"; + pe_mem_factor="1024"; +} + +FLUID_FLOW_WAN +{ + topology_yaml_file="fluid-flow-wan-topology.yaml"; + + interval_seconds="1"; + num_send_intervals="20"; + num_drain_intervals="20"; + rng_seed="12345"; + + terminal_send_every_n_intervals="1"; + terminal_send_probability="1.0"; + terminal_min_send_mbit="0"; + terminal_max_send_fraction_of_link_capacity="1.0"; + + debug_prints="0"; + # This can be either round_robin or fifo. + switch_scheduler="round_robin"; + round_robin_max_entries_per_egress="8"; + round_robin_quantum_mbit="0"; + + terminal_log_path="../../zmqml_artifacts/fluid-flow-wan/terminal-events.csv"; + switch_log_path="../../zmqml_artifacts/fluid-flow-wan/switch-events.csv"; + flowlet_log_path="../../zmqml_artifacts/fluid-flow-wan/flowlet-events.csv"; + switch_training_log_path="../../zmqml_artifacts/fluid-flow-wan/switch-training.csv"; +} diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 56e0ddd8..0f4cba1b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -206,6 +206,7 @@ add_executable(model-net-synthetic network-workloads/model-net-synthetic.c) add_executable(model-net-synthetic-slimfly network-workloads/model-net-synthetic-slimfly.c) add_executable(model-net-synthetic-fattree network-workloads/model-net-synthetic-fattree.c) add_executable(model-net-synthetic-dragonfly-all network-workloads/model-net-synthetic-dragonfly-all.c) +add_executable(model-net-fluid-flow-wan network-workloads/model-net-fluid-flow-wan.cxx) set(CODES_TARGETS topology-test @@ -214,6 +215,7 @@ set(CODES_TARGETS model-net-synthetic-slimfly model-net-synthetic-fattree model-net-synthetic-dragonfly-all + model-net-fluid-flow-wan ) if(USE_DUMPI) diff --git a/src/network-workloads/generate-fluid-flow-wan-topology.py b/src/network-workloads/generate-fluid-flow-wan-topology.py new file mode 100755 index 00000000..280935f5 --- /dev/null +++ b/src/network-workloads/generate-fluid-flow-wan-topology.py @@ -0,0 +1,310 @@ +#!/usr/bin/env python3 +""" +Generate a WAN-like topology YAML file for model-net-fluid-flow-wan. + +The generated graph is intentionally not a symmetric supercomputer-style +topology. It starts with a bidirectional ring to guarantee strong connectivity, +then adds random directed/asymmetric switch-to-switch links. Terminal access +links are generated with higher bandwidth than switch-to-switch links. +""" + +from __future__ import annotations + +import argparse +import random +import sys +from pathlib import Path + + +def positive_int(value: str) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"expected integer, got {value!r}") from exc + if parsed <= 0: + raise argparse.ArgumentTypeError(f"expected positive integer, got {parsed}") + return parsed + + +def nonnegative_int(value: str) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"expected integer, got {value!r}") from exc + if parsed < 0: + raise argparse.ArgumentTypeError(f"expected nonnegative integer, got {parsed}") + return parsed + + +def positive_float(value: str) -> float: + try: + parsed = float(value) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"expected float, got {value!r}") from exc + if parsed <= 0.0: + raise argparse.ArgumentTypeError(f"expected positive float, got {parsed}") + return parsed + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Generate a random WAN-like topology YAML for model-net-fluid-flow-wan." + ) + + parser.add_argument( + "--switches", + type=positive_int, + default=64, + help="number of switches to generate; default: 64", + ) + parser.add_argument( + "--terminals-per-switch", + type=positive_int, + default=2, + help="number of terminals attached to each switch; default: 2", + ) + parser.add_argument( + "--avg-switch-degree", + type=positive_float, + default=3.0, + help=( + "target average directed switch out-degree, including the ring links; " + "default: 3.0" + ), + ) + parser.add_argument( + "--reverse-link-probability", + type=positive_float, + default=0.35, + help=( + "probability of adding the reverse direction for each random extra link; " + "default: 0.35" + ), + ) + parser.add_argument( + "--switch-link-min-mbps", + type=positive_float, + default=5.0, + help="minimum switch-to-switch bandwidth in Mbps; default: 5", + ) + parser.add_argument( + "--switch-link-max-mbps", + type=positive_float, + default=25.0, + help="maximum switch-to-switch bandwidth in Mbps; default: 25", + ) + parser.add_argument( + "--terminal-link-min-mbps", + type=positive_float, + default=80.0, + help="minimum terminal-to-switch bandwidth in Mbps; default: 80", + ) + parser.add_argument( + "--terminal-link-max-mbps", + type=positive_float, + default=160.0, + help="maximum terminal-to-switch bandwidth in Mbps; default: 160", + ) + parser.add_argument( + "--switch-buffer-min-mb", + type=positive_float, + default=1000.0, + help="minimum shared switch buffer in Mbit; default: 1000", + ) + parser.add_argument( + "--switch-buffer-max-mb", + type=positive_float, + default=4000.0, + help="maximum shared switch buffer in Mbit; default: 4000", + ) + parser.add_argument( + "--seed", + type=nonnegative_int, + default=12345, + help="random seed; default: 12345", + ) + parser.add_argument( + "--output", + type=Path, + default=Path("doc/example/fluid-flow-wan-topology.generated.yaml"), + help=( + "output YAML path, relative to the current working directory unless absolute; " + "default: doc/example/fluid-flow-wan-topology.generated.yaml" + ), + ) + parser.add_argument( + "--name-prefix", + default="S", + help="switch name prefix; default: S", + ) + + args = parser.parse_args() + + if args.switches < 2: + parser.error("--switches must be at least 2 to form a connected switch graph") + + if args.avg_switch_degree < 2.0: + parser.error( + "--avg-switch-degree must be at least 2.0 because the generator " + "uses a bidirectional ring backbone" + ) + + if args.reverse_link_probability < 0.0 or args.reverse_link_probability > 1.0: + parser.error("--reverse-link-probability must be in [0, 1]") + + if args.switch_link_min_mbps > args.switch_link_max_mbps: + parser.error("--switch-link-min-mbps cannot exceed --switch-link-max-mbps") + + if args.terminal_link_min_mbps > args.terminal_link_max_mbps: + parser.error("--terminal-link-min-mbps cannot exceed --terminal-link-max-mbps") + + if args.terminal_link_min_mbps <= args.switch_link_max_mbps: + parser.error( + "terminal access links must be faster than switch-to-switch links: " + "require --terminal-link-min-mbps > --switch-link-max-mbps" + ) + + if args.switch_buffer_min_mb > args.switch_buffer_max_mb: + parser.error("--switch-buffer-min-mb cannot exceed --switch-buffer-max-mb") + + return args + + +def rand_uniform_rounded(rng: random.Random, lo: float, hi: float) -> float: + return round(rng.uniform(lo, hi), 3) + + +def add_link( + links: dict[int, dict[int, float]], + src: int, + dst: int, + bandwidth_mbps: float, +) -> bool: + if src == dst: + return False + if dst in links[src]: + return False + links[src][dst] = bandwidth_mbps + return True + + +def generate_links(args: argparse.Namespace, rng: random.Random) -> dict[int, dict[int, float]]: + n = args.switches + links: dict[int, dict[int, float]] = {i: {} for i in range(n)} + + def new_switch_bw() -> float: + return rand_uniform_rounded( + rng, args.switch_link_min_mbps, args.switch_link_max_mbps + ) + + # Strongly connected but asymmetric backbone. Each direction gets its own + # independently sampled capacity. + for i in range(n): + add_link(links, i, (i + 1) % n, new_switch_bw()) + add_link(links, i, (i - 1) % n, new_switch_bw()) + + target_edges = max(int(round(args.switches * args.avg_switch_degree)), 2 * n) + max_edges = n * (n - 1) + if target_edges > max_edges: + target_edges = max_edges + + current_edges = sum(len(v) for v in links.values()) + attempts = 0 + max_attempts = max(1000, 50 * max_edges) + + while current_edges < target_edges and attempts < max_attempts: + attempts += 1 + src = rng.randrange(n) + dst = rng.randrange(n) + if src == dst: + continue + + if add_link(links, src, dst, new_switch_bw()): + current_edges += 1 + + if current_edges >= target_edges: + break + + if rng.random() < args.reverse_link_probability: + if add_link(links, dst, src, new_switch_bw()): + current_edges += 1 + + if current_edges < target_edges: + raise RuntimeError( + f"could only generate {current_edges} directed switch links; " + f"target was {target_edges}" + ) + + return links + + +def switch_name(index: int, n_switches: int, prefix: str) -> str: + width = max(2, len(str(n_switches - 1))) + return f"{prefix}{index:0{width}d}" + + +def write_topology(args: argparse.Namespace, links: dict[int, dict[int, float]]) -> None: + rng = random.Random(args.seed + 1) + n = args.switches + names = [switch_name(i, n, args.name_prefix) for i in range(n)] + + args.output.parent.mkdir(parents=True, exist_ok=True) + + with args.output.open("w", encoding="utf-8") as f: + f.write("# Generated by src/network-workloads/generate-fluid-flow-wan-topology.py\n") + f.write(f"# switches={args.switches}\n") + f.write(f"# terminals_per_switch={args.terminals_per_switch}\n") + f.write(f"# avg_switch_degree={args.avg_switch_degree}\n") + f.write(f"# seed={args.seed}\n") + f.write("topology:\n") + f.write(" switches:\n") + + for i in range(n): + terminal_bw = rand_uniform_rounded( + rng, args.terminal_link_min_mbps, args.terminal_link_max_mbps + ) + switch_buffer = rand_uniform_rounded( + rng, args.switch_buffer_min_mb, args.switch_buffer_max_mb + ) + + f.write(f" {names[i]}:\n") + f.write(f" terminals: {args.terminals_per_switch}\n") + f.write(f' terminal_bandwidth: "{terminal_bw} Mbps"\n') + f.write(f' switch_buffer: "{switch_buffer} Mb"\n') + f.write(" connections:\n") + + for dst in sorted(links[i]): + f.write(f' {names[dst]}: "{links[i][dst]} Mbps"\n') + + f.write("\n") + + +def summarize(args: argparse.Namespace, links: dict[int, dict[int, float]]) -> None: + edge_count = sum(len(v) for v in links.values()) + degrees = [len(links[i]) for i in range(args.switches)] + bws = [bw for edges in links.values() for bw in edges.values()] + + print(f"wrote {args.output}") + print(f"switches: {args.switches}") + print(f"terminals: {args.switches * args.terminals_per_switch}") + print(f"directed switch links: {edge_count}") + print(f"average directed switch out-degree: {edge_count / args.switches:.3f}") + print(f"min/max directed switch out-degree: {min(degrees)} / {max(degrees)}") + print(f"switch-link Mbps min/max: {min(bws):.3f} / {max(bws):.3f}") + print( + "terminal-link Mbps range: " + f"{args.terminal_link_min_mbps:.3f} / {args.terminal_link_max_mbps:.3f}" + ) + + +def main() -> int: + args = parse_args() + rng = random.Random(args.seed) + links = generate_links(args, rng) + write_topology(args, links) + summarize(args, links) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/network-workloads/model-net-fluid-flow-wan.cxx b/src/network-workloads/model-net-fluid-flow-wan.cxx new file mode 100644 index 00000000..7ecf2c71 --- /dev/null +++ b/src/network-workloads/model-net-fluid-flow-wan.cxx @@ -0,0 +1,2200 @@ + +/* + * Standalone interval-fluid switch/terminal workload model. + * + * Terminal LPs generate stochastic bounded workload flowlets. Switch LPs route + * flowlets, queue them per output link, enforce a shared switch-wide buffer, + * and send per-flowlet fragments subject to interval link capacity. + * + * Supports sequential validation and optimistic execution. Optimistic mode uses + * event-local reverse metadata to undo terminal generation, switch arrivals, + * queue mutations, egress byte deltas, and RNG draws. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "codes/codes.h" +#include "codes/codes_mapping.h" +#include "codes/configuration.h" +#include "codes/lp-type-lookup.h" + +static const char* GROUP_NAME = "FLUID_FLOW_WAN_GRP"; +static const char* TERMINAL_LP_NAME = "fluid-flow-wan-terminal-lp"; +static const char* SWITCH_LP_NAME = "fluid-flow-wan-switch-lp"; + +static constexpr int MAX_PORTS_PER_SWITCH = 128; +/* + * This upper bound is intentionally part of the event-message footprint: every + * fluid_msg carries rc_allocs[MAX_RC_ALLOCATIONS] so optimistic execution can + * reverse a switch-egress allocation without external state. Keep + * PARAMS.message_size >= sizeof(fluid_msg), and reduce this bound only after + * measuring the maximum allocations per switch-egress event for the target + * topology/workload. + */ +static constexpr int MAX_RC_ALLOCATIONS = 256; +static constexpr double EPS = 1e-9; + +static constexpr double PHASE_GENERATE = 0.10; +static constexpr double PHASE_ARRIVAL = 0.20; +static constexpr double PHASE_SWITCH_EGRESS = 0.60; + +struct link_info { + int dst_switch; + double bandwidth_mbps; + double buffer_mbit; +}; + +struct switch_info { + std::string name; + int terminal_count = 0; + int terminal_start = 0; + double terminal_bandwidth_mbps = 10.0; + double switch_buffer_mbit = 1024.0; + std::vector links; +}; + +struct terminal_info { + int switch_id = -1; + int local_id = -1; + std::string name; +}; + +struct sim_config { + double interval_seconds = 1.0; + int num_send_intervals = 20; + int num_drain_intervals = 20; + int rng_seed = 12345; + int terminal_send_every_n_intervals = 1; + double terminal_send_probability = 1.0; + double terminal_min_send_mbit = 0.0; + double terminal_max_send_fraction_of_link_capacity = 1.0; + int debug_prints = 0; + char topology_yaml_file[1024] = ""; + char terminal_log_path[1024] = ""; + char switch_log_path[1024] = ""; + char flowlet_log_path[1024] = ""; + char switch_training_log_path[1024] = ""; + char switch_scheduler[64] = "round_robin"; + int round_robin_max_entries_per_egress = 32; + double round_robin_quantum_mbit = 0.0; +}; + +static sim_config cfg; +static std::vector switches; +static std::vector terminals; +static std::map switch_name_to_id; +static std::vector> next_switch_table; +static int total_switch_lps = 0; +static int total_terminal_lps = 0; + +struct queued_flowlet { + int valid; + unsigned long long flowlet_id; + int source_terminal; + int destination_terminal; + int creation_interval; + int enqueue_interval; + int age_intervals; + double remaining_mbit; +}; + +struct port_desc { + int is_terminal; + int target_index; + double capacity_mbit_per_interval; +}; + +struct terminal_state { + int terminal_id; + int attached_switch; + int local_terminal_id; + unsigned long long next_flowlet_seq; + double generated_mbit; + double sent_to_switch_mbit; + double received_mbit; + int generated_flowlets; + int received_fragments; +}; + +struct switch_state { + int switch_id; + int num_ports; + double shared_buffer_mbit; + port_desc ports[MAX_PORTS_PER_SWITCH]; + std::vector* queues[MAX_PORTS_PER_SWITCH]; + + /* + * Demand-driven egress scheduling. Each port may have at most one + * outstanding SWITCH_EGRESS event. -1 means no egress event is currently + * outstanding for that port. + */ + int scheduled_egress_interval[MAX_PORTS_PER_SWITCH]; + int rr_next_queue_index[MAX_PORTS_PER_SWITCH]; + + double enqueued_mbit; + double sent_mbit; + double delivered_local_mbit; + double dropped_mbit; + int received_fragments; + int sent_fragments; +}; + +enum fluid_event_type { + WORKLOAD_GENERATE = 1, + FLOWLET_ARRIVAL = 2, + SWITCH_EGRESS = 3, +}; + +struct rc_alloc_record { + int valid; + int queue_index; + queued_flowlet before; + double send_mbit; +}; + +struct fluid_msg { + int event_type; + int interval_id; + int source_terminal; + int destination_terminal; + int source_switch; + int destination_switch; + int port_id; + int creation_interval; + unsigned long long flowlet_id; + double mbit; + + /* + * Reverse-computation metadata. These fields are written by the forward + * event handler and consumed by the reverse handler if the event rolls back. + */ + int rc_rng_count; + int rc_generated; + int rc_no_route; + int rc_port_id; + int rc_queue_index; + int rc_coalesced; + double rc_accepted_mbit; + double rc_dropped_mbit; + int rc_alloc_count; + + int rc_prev_scheduled_egress_interval; + int rc_prev_rr_next_queue_index; + + int rc_log_target_is_terminal; + int rc_log_target_index; + double rc_log_capacity_mbit; + double rc_log_port_queued_before_mbit; + double rc_log_port_queued_after_mbit; + double rc_log_shared_queued_before_mbit; + double rc_log_shared_queued_after_mbit; + double rc_log_flowlet_remaining_after_mbit; + double rc_log_sent_total_mbit; + int rc_log_active_before_entries; + int rc_log_active_after_entries; + + rc_alloc_record rc_allocs[MAX_RC_ALLOCATIONS]; +}; + +static void terminal_init(terminal_state* ns, tw_lp* lp); +static void terminal_event(terminal_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp); +static void terminal_rev_event(terminal_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp); +static void terminal_commit_event(terminal_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp); +static void terminal_finalize(terminal_state* ns, tw_lp* lp); + +static void switch_init(switch_state* ns, tw_lp* lp); +static void switch_event(switch_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp); +static void switch_rev_event(switch_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp); +static void switch_commit_event(switch_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp); +static void switch_finalize(switch_state* ns, tw_lp* lp); + +static tw_lptype terminal_lp = { + (init_f)terminal_init, + (pre_run_f)NULL, + (event_f)terminal_event, + (revent_f)terminal_rev_event, + (commit_f)terminal_commit_event, + (final_f)terminal_finalize, + (map_f)codes_mapping, + sizeof(terminal_state), +}; + +static tw_lptype switch_lp = { + (init_f)switch_init, (pre_run_f)NULL, + (event_f)switch_event, (revent_f)switch_rev_event, + (commit_f)switch_commit_event, (final_f)switch_finalize, + (map_f)codes_mapping, sizeof(switch_state), +}; + +static const tw_lptype* terminal_get_lp_type(void) { + return &terminal_lp; +} +static const tw_lptype* switch_get_lp_type(void) { + return &switch_lp; +} + +static void add_lp_types(void) { + lp_type_register(TERMINAL_LP_NAME, terminal_get_lp_type()); + lp_type_register(SWITCH_LP_NAME, switch_get_lp_type()); +} + +static int get_configured_message_size_bytes(void) { + int configured_message_size = 0; + if (configuration_get_value_int(&config, "PARAMS", "message_size", NULL, + &configured_message_size) != 0) { + return 0; + } + return configured_message_size; +} + +static void validate_ross_message_size_or_abort(int rank) { + int configured_message_size = get_configured_message_size_bytes(); + size_t required_message_size = sizeof(fluid_msg); + + if (configured_message_size <= 0 || (size_t)configured_message_size < required_message_size) { + if (rank == 0) { + fprintf(stderr, + "fluid-flow-wan error: PARAMS.message_size=%d is too small for " + "sizeof(fluid_msg)=%zu. Increase PARAMS.message_size in the " + "CODES config before calling codes_mapping_setup(). " + "fluid_msg is large because MAX_RC_ALLOCATIONS=%d reverse " + "allocation records are carried in each event.\n", + configured_message_size, required_message_size, MAX_RC_ALLOCATIONS); + } + MPI_Abort(MPI_COMM_CODES, 1); + } +} + +static double seconds_to_ns(double seconds) { + return seconds * 1000.0 * 1000.0 * 1000.0; +} + +static double event_time_ns(int interval_id, double phase_seconds) { + return seconds_to_ns(((double)interval_id * cfg.interval_seconds) + phase_seconds); +} + +static double delay_until_ns(int target_interval, double phase_seconds, tw_lp* lp) { + double target = event_time_ns(target_interval, phase_seconds); + double delta = target - tw_now(lp); + if (delta <= g_tw_lookahead) { + delta = g_tw_lookahead + 1.0; + } + return delta; +} + +static std::string trim(const std::string& s) { + size_t b = 0; + while (b < s.size() && std::isspace((unsigned char)s[b])) { + ++b; + } + size_t e = s.size(); + while (e > b && std::isspace((unsigned char)s[e - 1])) { + --e; + } + return s.substr(b, e - b); +} + +static int leading_spaces(const std::string& s) { + int n = 0; + while (n < (int)s.size() && s[n] == ' ') { + ++n; + } + return n; +} + +static std::string strip_quotes(std::string s) { + s = trim(s); + if (s.size() >= 2 && + ((s.front() == '"' && s.back() == '"') || (s.front() == '\'' && s.back() == '\''))) { + return s.substr(1, s.size() - 2); + } + return s; +} + +static std::string remove_inline_comment(const std::string& s) { + bool in_quote = false; + char quote_char = 0; + for (size_t i = 0; i < s.size(); ++i) { + if ((s[i] == '"' || s[i] == '\'') && (i == 0 || s[i - 1] != '\\')) { + if (!in_quote) { + in_quote = true; + quote_char = s[i]; + } else if (quote_char == s[i]) { + in_quote = false; + quote_char = 0; + } + } + if (!in_quote && s[i] == '#') { + return s.substr(0, i); + } + } + return s; +} + +static bool split_key_value(const std::string& line, std::string* key, std::string* value) { + size_t pos = line.find(':'); + if (pos == std::string::npos) { + return false; + } + *key = trim(line.substr(0, pos)); + *value = trim(line.substr(pos + 1)); + return !key->empty(); +} + +static double parse_mbps(const std::string& raw) { + std::string s = strip_quotes(raw); + std::stringstream ss(s); + double v = 0.0; + ss >> v; + if (!std::isfinite(v) || v < 0.0) { + return 0.0; + } + return v; +} + +static double parse_mbit(const std::string& raw) { + std::string s = strip_quotes(raw); + std::stringstream ss(s); + double v = 0.0; + ss >> v; + if (!std::isfinite(v) || v < 0.0) { + return 0.0; + } + if (s.find("GB") != std::string::npos || s.find("Gb") != std::string::npos) { + return v * 1000.0; + } + if (s.find("MB") != std::string::npos || s.find("Mb") != std::string::npos) { + return v; + } + return v; +} + +static int get_or_add_switch(const std::string& name) { + std::map::iterator it = switch_name_to_id.find(name); + if (it != switch_name_to_id.end()) { + return it->second; + } + int id = (int)switches.size(); + switch_info sw; + sw.name = name; + switches.push_back(sw); + switch_name_to_id[name] = id; + return id; +} + +static void add_or_update_link(int src, int dst, double bw_mbps, double buffer_mbit) { + for (size_t i = 0; i < switches[src].links.size(); ++i) { + if (switches[src].links[i].dst_switch == dst) { + switches[src].links[i].bandwidth_mbps = bw_mbps; + if (buffer_mbit > 0.0) { + switches[src].links[i].buffer_mbit = buffer_mbit; + } + return; + } + } + link_info li; + li.dst_switch = dst; + li.bandwidth_mbps = bw_mbps; + li.buffer_mbit = buffer_mbit > 0.0 ? buffer_mbit : switches[src].switch_buffer_mbit; + switches[src].links.push_back(li); +} + +static void load_topology_yaml(const char* path) { + switches.clear(); + terminals.clear(); + switch_name_to_id.clear(); + + std::ifstream in(path); + if (!in.good()) { + tw_error(TW_LOC, "could not open topology YAML file: %s", path); + } + + std::string section; + bool in_switches = false; + bool in_connections = false; + int current_switch = -1; + int current_conn_dst = -1; + double current_conn_buffer = 0.0; + + std::string raw; + while (std::getline(in, raw)) { + std::string no_comment = remove_inline_comment(raw); + if (trim(no_comment).empty()) { + continue; + } + int indent = leading_spaces(no_comment); + std::string line = trim(no_comment); + if (line == "topology:") { + section = "topology"; + in_switches = false; + in_connections = false; + current_switch = -1; + continue; + } + if (section == "topology" && line == "switches:") { + in_switches = true; + in_connections = false; + current_switch = -1; + continue; + } + + std::string key; + std::string value; + if (!split_key_value(line, &key, &value)) { + continue; + } + + if (section == "topology" && in_switches) { + if (indent == 4 && value.empty()) { + current_switch = get_or_add_switch(key); + in_connections = false; + current_conn_dst = -1; + continue; + } + if (current_switch < 0) { + continue; + } + if (indent == 6 && key == "connections") { + in_connections = true; + current_conn_dst = -1; + current_conn_buffer = switches[current_switch].switch_buffer_mbit; + continue; + } + if (!in_connections && indent >= 6) { + if (key == "terminals") + switches[current_switch].terminal_count = atoi(strip_quotes(value).c_str()); + else if (key == "terminal_bandwidth") + switches[current_switch].terminal_bandwidth_mbps = parse_mbps(value); + else if (key == "switch_buffer") + switches[current_switch].switch_buffer_mbit = parse_mbit(value); + continue; + } + if (in_connections && indent == 8) { + int dst = get_or_add_switch(key); + current_conn_dst = dst; + current_conn_buffer = switches[current_switch].switch_buffer_mbit; + if (!value.empty()) { + add_or_update_link(current_switch, dst, parse_mbps(value), current_conn_buffer); + } + continue; + } + if (in_connections && indent >= 10 && current_conn_dst >= 0) { + if (key == "bandwidth") { + double bw = parse_mbps(value); + add_or_update_link(current_switch, current_conn_dst, bw, current_conn_buffer); + } else if (key == "buffer") { + current_conn_buffer = parse_mbit(value); + for (size_t i = 0; i < switches[current_switch].links.size(); ++i) { + if (switches[current_switch].links[i].dst_switch == current_conn_dst) { + switches[current_switch].links[i].buffer_mbit = current_conn_buffer; + } + } + } + } + } + } + + int terminal_start = 0; + for (int s = 0; s < (int)switches.size(); ++s) { + switches[s].terminal_start = terminal_start; + for (int t = 0; t < switches[s].terminal_count; ++t) { + terminal_info ti; + ti.switch_id = s; + ti.local_id = t; + std::ostringstream name; + name << switches[s].name << "." << t; + ti.name = name.str(); + terminals.push_back(ti); + ++terminal_start; + } + } + + if (switches.empty()) { + tw_error(TW_LOC, "topology YAML defined no switches"); + } + if (terminals.size() < 2) { + tw_error(TW_LOC, "topology YAML must define at least two terminals"); + } + if (cfg.interval_seconds <= 0.0) + tw_error(TW_LOC, "interval_seconds must be positive"); + if (cfg.num_send_intervals <= 0) + tw_error(TW_LOC, "num_send_intervals must be positive"); + if (cfg.num_drain_intervals < 0) + cfg.num_drain_intervals = 0; + if (cfg.terminal_send_every_n_intervals <= 0) + cfg.terminal_send_every_n_intervals = 1; + if (cfg.terminal_send_probability < 0.0) + cfg.terminal_send_probability = 0.0; + if (cfg.terminal_send_probability > 1.0) + cfg.terminal_send_probability = 1.0; + if (cfg.terminal_max_send_fraction_of_link_capacity < 0.0) + cfg.terminal_max_send_fraction_of_link_capacity = 0.0; + if (cfg.terminal_max_send_fraction_of_link_capacity > 1.0) + cfg.terminal_max_send_fraction_of_link_capacity = 1.0; +} + +static void compute_routes(void) { + int n = (int)switches.size(); + next_switch_table.assign(n, std::vector(n, -1)); + for (int src = 0; src < n; ++src) { + std::vector prev(n, -1); + std::vector seen(n, 0); + std::queue q; + seen[src] = 1; + q.push(src); + while (!q.empty()) { + int u = q.front(); + q.pop(); + for (size_t i = 0; i < switches[u].links.size(); ++i) { + int v = switches[u].links[i].dst_switch; + if (!seen[v]) { + seen[v] = 1; + prev[v] = u; + q.push(v); + } + } + } + for (int dst = 0; dst < n; ++dst) { + if (dst == src) { + next_switch_table[src][dst] = src; + continue; + } + if (!seen[dst]) { + next_switch_table[src][dst] = -1; + continue; + } + int cur = dst; + int parent = prev[cur]; + while (parent >= 0 && parent != src) { + cur = parent; + parent = prev[cur]; + } + next_switch_table[src][dst] = cur; + } + } +} + +static void read_int_param(const char* section, const char* key, int* value) { + int tmp = 0; + if (configuration_get_value_int(&config, section, key, NULL, &tmp) == 0) { + *value = tmp; + } +} + +static void read_double_param(const char* section, const char* key, double* value) { + double tmp = 0.0; + if (configuration_get_value_double(&config, section, key, NULL, &tmp) == 0) { + *value = tmp; + } +} + +static void read_string_param(const char* section, const char* key, char* value, size_t len) { + char tmp[1024]; + memset(tmp, 0, sizeof(tmp)); + if (configuration_get_value(&config, section, key, NULL, tmp, sizeof(tmp)) > 0) { + snprintf(value, len, "%s", tmp); + } +} + +static void read_relpath_param(const char* section, const char* key, char* value, size_t len) { + char tmp[1024]; + memset(tmp, 0, sizeof(tmp)); + if (configuration_get_value_relpath(&config, section, key, NULL, tmp, sizeof(tmp)) > 0) { + snprintf(value, len, "%s", tmp); + } +} + +static void load_config(void) { + read_relpath_param("FLUID_FLOW_WAN", "topology_yaml_file", cfg.topology_yaml_file, + sizeof(cfg.topology_yaml_file)); + read_relpath_param("FLUID_FLOW_WAN", "terminal_log_path", cfg.terminal_log_path, + sizeof(cfg.terminal_log_path)); + read_relpath_param("FLUID_FLOW_WAN", "switch_log_path", cfg.switch_log_path, + sizeof(cfg.switch_log_path)); + read_relpath_param("FLUID_FLOW_WAN", "flowlet_log_path", cfg.flowlet_log_path, + sizeof(cfg.flowlet_log_path)); + read_relpath_param("FLUID_FLOW_WAN", "switch_training_log_path", cfg.switch_training_log_path, + sizeof(cfg.switch_training_log_path)); + read_string_param("FLUID_FLOW_WAN", "switch_scheduler", cfg.switch_scheduler, + sizeof(cfg.switch_scheduler)); + read_int_param("FLUID_FLOW_WAN", "round_robin_max_entries_per_egress", + &cfg.round_robin_max_entries_per_egress); + read_double_param("FLUID_FLOW_WAN", "round_robin_quantum_mbit", &cfg.round_robin_quantum_mbit); + read_double_param("FLUID_FLOW_WAN", "interval_seconds", &cfg.interval_seconds); + read_int_param("FLUID_FLOW_WAN", "num_send_intervals", &cfg.num_send_intervals); + read_int_param("FLUID_FLOW_WAN", "num_drain_intervals", &cfg.num_drain_intervals); + read_int_param("FLUID_FLOW_WAN", "rng_seed", &cfg.rng_seed); + read_int_param("FLUID_FLOW_WAN", "terminal_send_every_n_intervals", + &cfg.terminal_send_every_n_intervals); + read_double_param("FLUID_FLOW_WAN", "terminal_send_probability", + &cfg.terminal_send_probability); + read_double_param("FLUID_FLOW_WAN", "terminal_min_send_mbit", &cfg.terminal_min_send_mbit); + read_double_param("FLUID_FLOW_WAN", "terminal_max_send_fraction_of_link_capacity", + &cfg.terminal_max_send_fraction_of_link_capacity); + read_int_param("FLUID_FLOW_WAN", "debug_prints", &cfg.debug_prints); + + if (strcmp(cfg.switch_scheduler, "fifo") != 0 && + strcmp(cfg.switch_scheduler, "round_robin") != 0) { + tw_error(TW_LOC, + "FLUID_FLOW_WAN.switch_scheduler must be one of: fifo, round_robin; got '%s'", + cfg.switch_scheduler); + } + + if (cfg.round_robin_max_entries_per_egress <= 0) { + tw_error(TW_LOC, + "FLUID_FLOW_WAN.round_robin_max_entries_per_egress must be positive; got %d", + cfg.round_robin_max_entries_per_egress); + } + + if (cfg.round_robin_max_entries_per_egress > MAX_RC_ALLOCATIONS) { + tw_error(TW_LOC, + "FLUID_FLOW_WAN.round_robin_max_entries_per_egress=%d exceeds " + "MAX_RC_ALLOCATIONS=%d", + cfg.round_robin_max_entries_per_egress, MAX_RC_ALLOCATIONS); + } + + if (cfg.round_robin_quantum_mbit < 0.0) { + tw_error(TW_LOC, "FLUID_FLOW_WAN.round_robin_quantum_mbit must be nonnegative; got %.12f", + cfg.round_robin_quantum_mbit); + } + + if (cfg.topology_yaml_file[0] == '\0') { + tw_error(TW_LOC, "FLUID_FLOW_WAN.topology_yaml_file is required"); + } + load_topology_yaml(cfg.topology_yaml_file); + compute_routes(); +} + +static tw_lpid get_switch_gid(int switch_id) { + if (switch_id < 0 || switch_id >= (int)switches.size()) { + tw_error(TW_LOC, "switch id %d out of range [0, %zu)", switch_id, switches.size()); + } + + tw_lpid gid = 0; + + /* + * LPGROUPS has one FLUID_FLOW_WAN_GRP repetition containing all switch LPs. + * The switch index is therefore the LP-type offset, not the group repetition. + */ + codes_mapping_get_lp_id(GROUP_NAME, SWITCH_LP_NAME, NULL, 1, 0, switch_id, &gid); + + return gid; +} + +static tw_lpid get_terminal_gid(int terminal_id) { + if (terminal_id < 0 || terminal_id >= (int)terminals.size()) { + tw_error(TW_LOC, "terminal id %d out of range [0, %zu)", terminal_id, terminals.size()); + } + + tw_lpid gid = 0; + + /* + * LPGROUPS has one FLUID_FLOW_WAN_GRP repetition containing all terminal LPs. + * The terminal index is therefore the LP-type offset, not the group repetition. + */ + codes_mapping_get_lp_id(GROUP_NAME, TERMINAL_LP_NAME, NULL, 1, 0, terminal_id, &gid); + + return gid; +} + +static int find_switch_link_port(const switch_state* ns, int dst_switch) { + for (int p = 0; p < ns->num_ports; ++p) { + if (!ns->ports[p].is_terminal && ns->ports[p].target_index == dst_switch) { + return p; + } + } + return -1; +} + +static int find_terminal_port(const switch_state* ns, int dst_terminal) { + for (int p = 0; p < ns->num_ports; ++p) { + if (ns->ports[p].is_terminal && ns->ports[p].target_index == dst_terminal) { + return p; + } + } + return -1; +} + +static double queued_mbit_on_switch(const switch_state* ns); + +static double enqueue_flowlet(switch_state* ns, int port_id, const fluid_msg* m, + double* port_queued_before_out, double* shared_queued_before_out, + double* shared_queued_after_out, double* dropped_out, + double* flowlet_remaining_after_out, int* coalesced_out, + int* queue_index_out) { + if (port_queued_before_out != NULL) { + *port_queued_before_out = 0.0; + } + if (shared_queued_before_out != NULL) { + *shared_queued_before_out = 0.0; + } + if (shared_queued_after_out != NULL) { + *shared_queued_after_out = 0.0; + } + if (dropped_out != NULL) { + *dropped_out = 0.0; + } + if (flowlet_remaining_after_out != NULL) { + *flowlet_remaining_after_out = 0.0; + } + if (coalesced_out != NULL) { + *coalesced_out = 0; + } + if (queue_index_out != NULL) { + *queue_index_out = -1; + } + + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + if (dropped_out != NULL) { + *dropped_out = m->mbit; + } + return 0.0; + } + + std::vector& qv = *ns->queues[port_id]; + + double port_queued_before = 0.0; + for (size_t i = 0; i < qv.size(); ++i) { + port_queued_before += qv[i].remaining_mbit; + } + + double shared_queued_before = queued_mbit_on_switch(ns); + + if (port_queued_before_out != NULL) { + *port_queued_before_out = port_queued_before; + } + if (shared_queued_before_out != NULL) { + *shared_queued_before_out = shared_queued_before; + } + + /* + * Admission is controlled by the shared switch-wide buffer. Output queues + * remain per port, but they all draw from this one occupancy limit. + */ + double shared_available = std::max(0.0, ns->shared_buffer_mbit - shared_queued_before); + double accepted = std::min(m->mbit, shared_available); + double dropped = std::max(0.0, m->mbit - accepted); + + if (dropped > EPS) { + ns->dropped_mbit += dropped; + } + + if (dropped_out != NULL) { + *dropped_out = dropped; + } + + if (accepted <= EPS) { + if (shared_queued_after_out != NULL) { + *shared_queued_after_out = shared_queued_before; + } + return 0.0; + } + + for (size_t i = 0; i < qv.size(); ++i) { + if (!qv[i].valid) { + continue; + } + + if (qv[i].flowlet_id == m->flowlet_id && qv[i].source_terminal == m->source_terminal && + qv[i].destination_terminal == m->destination_terminal && + qv[i].creation_interval == m->creation_interval) { + qv[i].remaining_mbit += accepted; + qv[i].age_intervals = m->interval_id - m->creation_interval; + ns->enqueued_mbit += accepted; + + if (flowlet_remaining_after_out != NULL) { + *flowlet_remaining_after_out = qv[i].remaining_mbit; + } + if (coalesced_out != NULL) { + *coalesced_out = 1; + } + if (queue_index_out != NULL) { + *queue_index_out = (int)i; + } + if (shared_queued_after_out != NULL) { + *shared_queued_after_out = shared_queued_before + accepted; + } + + return accepted; + } + } + + queued_flowlet q; + memset(&q, 0, sizeof(q)); + q.valid = 1; + q.flowlet_id = m->flowlet_id; + q.source_terminal = m->source_terminal; + q.destination_terminal = m->destination_terminal; + q.creation_interval = m->creation_interval; + q.enqueue_interval = m->interval_id; + q.age_intervals = m->interval_id - m->creation_interval; + q.remaining_mbit = accepted; + + qv.push_back(q); + if (queue_index_out != NULL) { + *queue_index_out = (int)qv.size() - 1; + } + ns->enqueued_mbit += accepted; + + if (flowlet_remaining_after_out != NULL) { + *flowlet_remaining_after_out = accepted; + } + if (shared_queued_after_out != NULL) { + *shared_queued_after_out = shared_queued_before + accepted; + } + + return accepted; +} + + +static double queued_mbit_on_port(const switch_state* ns, int port_id) { + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + return 0.0; + } + double total = 0.0; + const std::vector& qv = *ns->queues[port_id]; + for (size_t i = 0; i < qv.size(); ++i) { + total += qv[i].remaining_mbit; + } + return total; +} + +static double queued_mbit_on_switch(const switch_state* ns) { + double total = 0.0; + for (int p = 0; p < ns->num_ports; ++p) { + total += queued_mbit_on_port(ns, p); + } + return total; +} + + +static int active_flowlet_count_on_port(const switch_state* ns, int port_id) { + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + return 0; + } + int count = 0; + const std::vector& qv = *ns->queues[port_id]; + for (size_t i = 0; i < qv.size(); ++i) { + if (qv[i].valid && qv[i].remaining_mbit > EPS) { + ++count; + } + } + return count; +} + +static int find_queue_index_for_flowlet(const switch_state* ns, int port_id, + const queued_flowlet& needle) { + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + return -1; + } + + const std::vector& qv = *ns->queues[port_id]; + + for (int i = 0; i < (int)qv.size(); ++i) { + if (!qv[i].valid) { + continue; + } + + if (qv[i].flowlet_id == needle.flowlet_id && + qv[i].source_terminal == needle.source_terminal && + qv[i].destination_terminal == needle.destination_terminal && + qv[i].creation_interval == needle.creation_interval) { + return i; + } + } + + return -1; +} + +static int find_queue_index_for_msg(const switch_state* ns, int port_id, const fluid_msg* m) { + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + return -1; + } + + const std::vector& qv = *ns->queues[port_id]; + + for (int i = 0; i < (int)qv.size(); ++i) { + if (!qv[i].valid) { + continue; + } + + if (qv[i].flowlet_id == m->flowlet_id && qv[i].source_terminal == m->source_terminal && + qv[i].destination_terminal == m->destination_terminal && + qv[i].creation_interval == m->creation_interval) { + return i; + } + } + + return -1; +} + +static void compact_port_queue(switch_state* ns, int port_id) { + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + return; + } + std::vector& qv = *ns->queues[port_id]; + qv.erase(std::remove_if(qv.begin(), qv.end(), + [](const queued_flowlet& q) { + return !q.valid || q.remaining_mbit <= EPS; + }), + qv.end()); +} + +static bool fluid_commit_logging_active = false; +static std::ostringstream terminal_log_buffer; +static std::ostringstream switch_log_buffer; +static std::ostringstream flowlet_log_buffer; +static std::ostringstream switch_training_log_buffer; + +static bool fluid_optimistic_mode(void) { + return g_tw_synchronization_protocol == OPTIMISTIC || + g_tw_synchronization_protocol == OPTIMISTIC_DEBUG || + g_tw_synchronization_protocol == OPTIMISTIC_REALTIME; +} + +static bool fluid_csv_forward_logs_enabled(void) { + return !fluid_optimistic_mode(); +} + +static bool fluid_csv_commit_logs_enabled(void) { + return fluid_optimistic_mode(); +} + +static bool fluid_csv_logs_enabled(void) { + return fluid_csv_forward_logs_enabled() || fluid_commit_logging_active; +} + +static void fluid_commit_logging_begin(void) { + fluid_commit_logging_active = true; +} + +static void fluid_commit_logging_end(void) { + fluid_commit_logging_active = false; +} + +static void append_log_row(const char* path, std::ostringstream* log_buffer, + const std::string& row) { + if (path[0] == '\0' || !fluid_csv_logs_enabled()) { + return; + } + + (*log_buffer) << row; +} + +static void append_terminal_log(int interval_id, const char* event_name, int terminal_id, + int peer_id, double mbit) { + std::ostringstream row; + row << interval_id << ',' << event_name << ',' << terminal_id << ',' + << terminals[terminal_id].name << ',' << terminals[terminal_id].switch_id << ',' << peer_id + << ',' << mbit << '\n'; + append_log_row(cfg.terminal_log_path, &terminal_log_buffer, row.str()); +} + +static void append_switch_log(int interval_id, const char* event_name, int switch_id, int port_id, + int target_is_terminal, int target_index, double capacity_mbit, + double queued_before_mbit, double sent_mbit, double queued_after_mbit, + double shared_queued_before_mbit, double shared_queued_after_mbit, + double shared_buffer_mbit, double dropped_mbit, + int active_queue_entries) { + std::ostringstream row; + row << interval_id << ',' << event_name << ',' << switch_id << ',' << switches[switch_id].name + << ',' << port_id << ',' << (target_is_terminal ? "terminal" : "switch") << ',' + << target_index << ',' << capacity_mbit << ',' << queued_before_mbit << ',' << sent_mbit + << ',' << queued_after_mbit << ',' << shared_queued_before_mbit << ',' + << shared_queued_after_mbit << ',' << shared_buffer_mbit << ',' << dropped_mbit << ',' + << active_queue_entries << '\n'; + append_log_row(cfg.switch_log_path, &switch_log_buffer, row.str()); +} + +static void append_flowlet_log(int interval_id, const char* event_name, int switch_id, int port_id, + int target_is_terminal, int target_index, + unsigned long long flowlet_id, int source_terminal, + int destination_terminal, int creation_interval, + double capacity_mbit, double queued_before_mbit, double send_mbit, + double remaining_after_mbit, double dropped_mbit) { + std::ostringstream row; + row << interval_id << ',' << event_name << ',' << switch_id << ',' << switches[switch_id].name + << ',' << port_id << ',' << (target_is_terminal ? "terminal" : "switch") << ',' + << target_index << ',' << flowlet_id << ',' << source_terminal << ',' + << destination_terminal << ',' << creation_interval << ',' + << (interval_id - creation_interval) << ',' << capacity_mbit << ',' << queued_before_mbit + << ',' << send_mbit << ',' << remaining_after_mbit << ',' << dropped_mbit << '\n'; + append_log_row(cfg.flowlet_log_path, &flowlet_log_buffer, row.str()); +} + +static void append_switch_training_log(int interval_id, int switch_id, int port_id, + int target_is_terminal, int target_index, + double capacity_mbit, double queued_before_mbit, + double sent_mbit, double queued_after_mbit, + double dropped_mbit, int active_before, int active_after, + const std::vector& allocations) { + std::ostringstream row; + row << interval_id << ',' << switch_id << ',' << switches[switch_id].name << ',' << port_id + << ',' << (target_is_terminal ? "terminal" : "switch") << ',' << target_index << ',' + << cfg.switch_scheduler << ',' << capacity_mbit << ',' << queued_before_mbit << ',' + << sent_mbit << ',' << queued_after_mbit << ',' << dropped_mbit << ',' << active_before + << ',' << active_after << ','; + for (size_t i = 0; i < allocations.size(); ++i) { + if (i > 0) { + row << ';'; + } + row << allocations[i]; + } + row << '\n'; + append_log_row(cfg.switch_training_log_path, &switch_training_log_buffer, row.str()); +} + +static int next_terminal_generate_interval_after(int interval_id) { + int period = cfg.terminal_send_every_n_intervals; + if (period <= 0) { + period = 1; + } + + if (interval_id < 0) { + return 0; + } + + return interval_id + period; +} + +static int first_terminal_generate_interval(void) { + /* + * The first eligible terminal workload interval is always 0. Later events + * advance by terminal_send_every_n_intervals, so terminals do not carry a + * speculative self-event chain through intervals where no workload can be + * generated. + */ + return 0; +} + +static void schedule_workload_generate(terminal_state* ns, int interval_id, tw_lp* lp) { + if (interval_id >= cfg.num_send_intervals) { + return; + } + + int period = cfg.terminal_send_every_n_intervals; + if (period <= 0) { + period = 1; + } + + /* + * Defensive alignment: callers should already pass eligible send intervals, + * but if a future caller passes an arbitrary interval, round up to the next + * eligible workload-generation interval instead of scheduling a no-op event. + */ + int rem = interval_id % period; + if (rem != 0) { + interval_id += period - rem; + } + + if (interval_id >= cfg.num_send_intervals) { + return; + } + + tw_event* e = tw_event_new(lp->gid, delay_until_ns(interval_id, PHASE_GENERATE, lp), lp); + fluid_msg* m = (fluid_msg*)tw_event_data(e); + memset(m, 0, sizeof(*m)); + m->event_type = WORKLOAD_GENERATE; + m->interval_id = interval_id; + m->source_terminal = ns->terminal_id; + tw_event_send(e); +} + +static void schedule_switch_egress(int interval_id, int port_id, tw_lp* lp) { + if (interval_id >= cfg.num_send_intervals + cfg.num_drain_intervals) { + return; + } + tw_event* e = tw_event_new(lp->gid, delay_until_ns(interval_id, PHASE_SWITCH_EGRESS, lp), lp); + fluid_msg* m = (fluid_msg*)tw_event_data(e); + memset(m, 0, sizeof(*m)); + m->event_type = SWITCH_EGRESS; + m->interval_id = interval_id; + m->port_id = port_id; + tw_event_send(e); +} + +static void request_switch_egress(switch_state* ns, int interval_id, int port_id, tw_lp* lp, + fluid_msg* cause_msg) { + if (port_id < 0 || port_id >= ns->num_ports) { + tw_error(TW_LOC, "invalid request_switch_egress port %d on switch %d", port_id, + ns->switch_id); + } + + int prev = ns->scheduled_egress_interval[port_id]; + + if (cause_msg != NULL) { + cause_msg->rc_prev_scheduled_egress_interval = prev; + } + + if (interval_id >= cfg.num_send_intervals + cfg.num_drain_intervals) { + return; + } + + if (prev == interval_id) { + return; + } + + if (prev >= 0 && prev < interval_id) { + /* + * An earlier egress is already outstanding. Let that earlier event + * decide whether residual bytes require a later egress. + */ + return; + } + + if (prev >= 0 && prev > interval_id) { + tw_error(TW_LOC, + "switch %d port %d has future egress interval %d while trying " + "to schedule earlier interval %d", + ns->switch_id, port_id, prev, interval_id); + } + + schedule_switch_egress(interval_id, port_id, lp); + ns->scheduled_egress_interval[port_id] = interval_id; + + if (cause_msg != NULL) { + } +} + +static void schedule_arrival(int interval_id, tw_lpid dst_gid, const fluid_msg* src_msg, + double mbit, int dst_switch, tw_lp* lp) { + tw_event* e = tw_event_new(dst_gid, delay_until_ns(interval_id, PHASE_ARRIVAL, lp), lp); + fluid_msg* m = (fluid_msg*)tw_event_data(e); + memset(m, 0, sizeof(*m)); + m->event_type = FLOWLET_ARRIVAL; + m->interval_id = interval_id; + m->source_terminal = src_msg->source_terminal; + m->destination_terminal = src_msg->destination_terminal; + m->source_switch = src_msg->source_switch; + m->destination_switch = dst_switch; + m->creation_interval = src_msg->creation_interval; + m->flowlet_id = src_msg->flowlet_id; + m->mbit = mbit; + tw_event_send(e); +} + +static void terminal_init(terminal_state* ns, tw_lp* lp) { + memset(ns, 0, sizeof(*ns)); + ns->terminal_id = codes_mapping_get_lp_relative_id(lp->gid, 0, 0); + if (ns->terminal_id < 0 || ns->terminal_id >= (int)terminals.size()) { + tw_error(TW_LOC, "terminal LP relative id %d out of range", ns->terminal_id); + } + ns->attached_switch = terminals[ns->terminal_id].switch_id; + ns->local_terminal_id = terminals[ns->terminal_id].local_id; + ns->next_flowlet_seq = 0; + schedule_workload_generate(ns, first_terminal_generate_interval(), lp); +} + +static int choose_random_destination(int self, tw_lp* lp) { + int n = (int)terminals.size(); + if (n <= 1) { + return self; + } + int offset = (int)tw_rand_integer(lp->rng, 1, n - 1); + return (self + offset) % n; +} + +static double random_workload_mbit(int terminal_id, tw_lp* lp) { + int sw = terminals[terminal_id].switch_id; + double cap = switches[sw].terminal_bandwidth_mbps * cfg.interval_seconds; + double max_mbit = cap * cfg.terminal_max_send_fraction_of_link_capacity; + double min_mbit = std::min(cfg.terminal_min_send_mbit, max_mbit); + double u = tw_rand_unif(lp->rng); + return min_mbit + u * (max_mbit - min_mbit); +} + +static void build_allocation_strings_from_rc(const fluid_msg* m, + std::vector& allocations) { + allocations.clear(); + + for (int i = 0; i < m->rc_alloc_count; ++i) { + const rc_alloc_record* rc = &m->rc_allocs[i]; + + if (!rc->valid || rc->send_mbit <= EPS) { + continue; + } + + double remaining_after = rc->before.remaining_mbit - rc->send_mbit; + if (remaining_after < 0.0 && remaining_after > -EPS) { + remaining_after = 0.0; + } + + std::ostringstream ss; + ss << rc->before.flowlet_id << ':' << rc->before.source_terminal << ':' + << rc->before.destination_terminal << ':' << rc->before.creation_interval << ':' + << rc->before.remaining_mbit << ':' << rc->send_mbit << ':' << remaining_after; + allocations.push_back(ss.str()); + } +} + +static void log_terminal_generate_event(const terminal_state* ns, const fluid_msg* m) { + if (m->rc_generated) { + append_terminal_log(m->interval_id, "generate_send", ns->terminal_id, + m->destination_terminal, m->mbit); + } +} + +static void log_terminal_receive_event(const terminal_state* ns, const fluid_msg* m) { + append_terminal_log(m->interval_id, "receive", ns->terminal_id, m->source_terminal, m->mbit); +} + +static void log_switch_arrival_event(const switch_state* ns, const fluid_msg* m) { + if (m->rc_no_route) { + append_switch_log(m->interval_id, "drop_no_route", ns->switch_id, -1, 0, -1, 0.0, 0.0, 0.0, + 0.0, m->rc_log_shared_queued_before_mbit, + m->rc_log_shared_queued_after_mbit, ns->shared_buffer_mbit, + m->rc_dropped_mbit, 0); + append_flowlet_log(m->interval_id, "drop_no_route", ns->switch_id, -1, 0, -1, m->flowlet_id, + m->source_terminal, m->destination_terminal, m->creation_interval, 0.0, + 0.0, 0.0, 0.0, m->rc_dropped_mbit); + return; + } + + if (m->rc_accepted_mbit > EPS) { + append_flowlet_log(m->interval_id, m->rc_coalesced ? "enqueue_coalesce" : "enqueue", + ns->switch_id, m->rc_port_id, m->rc_log_target_is_terminal, + m->rc_log_target_index, m->flowlet_id, m->source_terminal, + m->destination_terminal, m->creation_interval, m->rc_log_capacity_mbit, + m->rc_log_port_queued_before_mbit, 0.0, + m->rc_log_flowlet_remaining_after_mbit, 0.0); + } + + if (m->rc_dropped_mbit > EPS) { + append_switch_log(m->interval_id, "drop_shared_buffer_overflow", ns->switch_id, + m->rc_port_id, m->rc_log_target_is_terminal, m->rc_log_target_index, + m->rc_log_capacity_mbit, m->rc_log_port_queued_before_mbit, 0.0, + m->rc_log_port_queued_after_mbit, m->rc_log_shared_queued_before_mbit, + m->rc_log_shared_queued_after_mbit, ns->shared_buffer_mbit, + m->rc_dropped_mbit, m->rc_log_active_after_entries); + + append_flowlet_log(m->interval_id, "drop_shared_buffer_overflow", ns->switch_id, + m->rc_port_id, m->rc_log_target_is_terminal, m->rc_log_target_index, + m->flowlet_id, m->source_terminal, m->destination_terminal, + m->creation_interval, m->rc_log_capacity_mbit, + m->rc_log_port_queued_before_mbit, 0.0, + m->rc_log_flowlet_remaining_after_mbit, m->rc_dropped_mbit); + } +} + +static void log_switch_egress_event(const switch_state* ns, const fluid_msg* m) { + for (int i = 0; i < m->rc_alloc_count; ++i) { + const rc_alloc_record* rc = &m->rc_allocs[i]; + + if (!rc->valid || rc->send_mbit <= EPS) { + continue; + } + + double remaining_after = rc->before.remaining_mbit - rc->send_mbit; + if (remaining_after < 0.0 && remaining_after > -EPS) { + remaining_after = 0.0; + } + + append_flowlet_log(m->interval_id, "allocate_send", ns->switch_id, m->port_id, + m->rc_log_target_is_terminal, m->rc_log_target_index, + rc->before.flowlet_id, rc->before.source_terminal, + rc->before.destination_terminal, rc->before.creation_interval, + m->rc_log_capacity_mbit, m->rc_log_port_queued_before_mbit, + rc->send_mbit, remaining_after, 0.0); + } + + append_switch_log(m->interval_id, "egress", ns->switch_id, m->port_id, + m->rc_log_target_is_terminal, m->rc_log_target_index, m->rc_log_capacity_mbit, + m->rc_log_port_queued_before_mbit, m->rc_log_sent_total_mbit, + m->rc_log_port_queued_after_mbit, m->rc_log_shared_queued_before_mbit, + m->rc_log_shared_queued_after_mbit, ns->shared_buffer_mbit, 0.0, + m->rc_log_active_after_entries); + + std::vector allocations; + build_allocation_strings_from_rc(m, allocations); + append_switch_training_log(m->interval_id, ns->switch_id, m->port_id, + m->rc_log_target_is_terminal, m->rc_log_target_index, + m->rc_log_capacity_mbit, m->rc_log_port_queued_before_mbit, + m->rc_log_sent_total_mbit, m->rc_log_port_queued_after_mbit, 0.0, + m->rc_log_active_before_entries, m->rc_log_active_after_entries, + allocations); +} + +static void handle_workload_generate(terminal_state* ns, fluid_msg* m, tw_lp* lp) { + int interval = m->interval_id; + + m->rc_rng_count = 0; + m->rc_generated = 0; + m->mbit = 0.0; + m->destination_terminal = -1; + m->flowlet_id = 0; + + double p = tw_rand_unif(lp->rng); + m->rc_rng_count++; + + if (p <= cfg.terminal_send_probability) { + int dst = choose_random_destination(ns->terminal_id, lp); + m->rc_rng_count++; + + double mbit = random_workload_mbit(ns->terminal_id, lp); + m->rc_rng_count++; + + if (mbit > EPS) { + fluid_msg out_msg; + memset(&out_msg, 0, sizeof(out_msg)); + out_msg.event_type = FLOWLET_ARRIVAL; + out_msg.interval_id = interval + 1; + out_msg.source_terminal = ns->terminal_id; + out_msg.destination_terminal = dst; + out_msg.source_switch = ns->attached_switch; + out_msg.destination_switch = ns->attached_switch; + out_msg.creation_interval = interval; + out_msg.flowlet_id = ((unsigned long long)ns->terminal_id << 48) | + (unsigned long long)ns->next_flowlet_seq++; + out_msg.mbit = mbit; + + tw_lpid sw_gid = get_switch_gid(ns->attached_switch); + schedule_arrival(interval + 1, sw_gid, &out_msg, mbit, ns->attached_switch, lp); + + ns->generated_mbit += mbit; + ns->sent_to_switch_mbit += mbit; + ns->generated_flowlets++; + + m->rc_generated = 1; + m->destination_terminal = dst; + m->flowlet_id = out_msg.flowlet_id; + m->mbit = mbit; + + log_terminal_generate_event(ns, m); + + if (cfg.debug_prints) { + printf("[fluid terminal] interval=%d terminal=%d dst=%d mbit=%.6f\n", interval, + ns->terminal_id, dst, mbit); + } + } + } + + schedule_workload_generate(ns, next_terminal_generate_interval_after(interval), lp); +} + + +static void handle_terminal_arrival(terminal_state* ns, fluid_msg* m) { + ns->received_mbit += m->mbit; + ns->received_fragments++; + log_terminal_receive_event(ns, m); +} + +static void terminal_event(terminal_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp) { + (void)b; + switch (m->event_type) { + case WORKLOAD_GENERATE: + handle_workload_generate(ns, m, lp); + break; + case FLOWLET_ARRIVAL: + handle_terminal_arrival(ns, m); + break; + default: + tw_error(TW_LOC, "terminal received unknown event type %d", m->event_type); + } +} + +static void terminal_rev_event(terminal_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp) { + (void)b; + + switch (m->event_type) { + case WORKLOAD_GENERATE: + if (m->rc_generated) { + ns->generated_mbit -= m->mbit; + ns->sent_to_switch_mbit -= m->mbit; + ns->generated_flowlets--; + + if (ns->next_flowlet_seq == 0) { + tw_error(TW_LOC, "terminal %d next_flowlet_seq underflow during rollback", + ns->terminal_id); + } + + ns->next_flowlet_seq--; + } + + for (int i = 0; i < m->rc_rng_count; ++i) { + tw_rand_reverse_unif(lp->rng); + } + + break; + + case FLOWLET_ARRIVAL: + ns->received_mbit -= m->mbit; + ns->received_fragments--; + break; + + default: + tw_error(TW_LOC, "terminal reverse received unknown event type %d", m->event_type); + } +} + +static void terminal_commit_event(terminal_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp) { + (void)b; + (void)lp; + + if (!fluid_csv_commit_logs_enabled()) { + return; + } + + fluid_commit_logging_begin(); + + switch (m->event_type) { + case WORKLOAD_GENERATE: + log_terminal_generate_event(ns, m); + break; + + case FLOWLET_ARRIVAL: + log_terminal_receive_event(ns, m); + break; + + default: + break; + } + + fluid_commit_logging_end(); +} + +static void terminal_finalize(terminal_state* ns, tw_lp* lp) { + printf( + "fluid-flow-wan-terminal gid=%llu terminal=%d switch=%d generated_mbit=%.6f sent_mbit=%.6f " + "received_mbit=%.6f generated_flowlets=%d received_fragments=%d\n", + (unsigned long long)lp->gid, ns->terminal_id, ns->attached_switch, ns->generated_mbit, + ns->sent_to_switch_mbit, ns->received_mbit, ns->generated_flowlets, ns->received_fragments); +} + +static void switch_init(switch_state* ns, tw_lp* lp) { + memset(ns, 0, sizeof(*ns)); + ns->switch_id = codes_mapping_get_lp_relative_id(lp->gid, 0, 0); + if (ns->switch_id < 0 || ns->switch_id >= (int)switches.size()) { + tw_error(TW_LOC, "switch LP relative id %d out of range", ns->switch_id); + } + + for (int p = 0; p < MAX_PORTS_PER_SWITCH; ++p) { + ns->scheduled_egress_interval[p] = -1; + ns->rr_next_queue_index[p] = 0; + } + + const switch_info& sw = switches[ns->switch_id]; + + /* + * switch_buffer is modeled as one shared switch-wide pool. Output queues + * remain per port for scheduling, but admission checks total queued bytes + * across all ports on this switch. + */ + ns->shared_buffer_mbit = sw.switch_buffer_mbit; + if (ns->shared_buffer_mbit < 0.0) { + tw_error(TW_LOC, "negative shared switch buffer on switch %d", ns->switch_id); + } + + for (size_t i = 0; i < sw.links.size(); ++i) { + if (ns->num_ports >= MAX_PORTS_PER_SWITCH) { + tw_error(TW_LOC, "too many ports on switch %d", ns->switch_id); + } + port_desc* p = &ns->ports[ns->num_ports++]; + memset(p, 0, sizeof(*p)); + p->is_terminal = 0; + p->target_index = sw.links[i].dst_switch; + p->capacity_mbit_per_interval = sw.links[i].bandwidth_mbps * cfg.interval_seconds; + ns->queues[ns->num_ports - 1] = new std::vector(); + } + for (int t = 0; t < sw.terminal_count; ++t) { + if (ns->num_ports >= MAX_PORTS_PER_SWITCH) { + tw_error(TW_LOC, "too many ports on switch %d", ns->switch_id); + } + port_desc* p = &ns->ports[ns->num_ports++]; + memset(p, 0, sizeof(*p)); + p->is_terminal = 1; + p->target_index = sw.terminal_start + t; + p->capacity_mbit_per_interval = sw.terminal_bandwidth_mbps * cfg.interval_seconds; + ns->queues[ns->num_ports - 1] = new std::vector(); + } + + /* + * Egress is demand-driven. We no longer pre-schedule periodic empty + * egress events for every port. Arrivals schedule the first egress, and + * egress events reschedule themselves only while residual queued bytes + * remain. + */ +} + + +static void handle_switch_arrival(switch_state* ns, fluid_msg* m, tw_lp* lp) { + ns->received_fragments++; + int dst_sw = terminals[m->destination_terminal].switch_id; + int port_id = -1; + if (dst_sw == ns->switch_id) { + port_id = find_terminal_port(ns, m->destination_terminal); + } else { + int next_sw = next_switch_table[ns->switch_id][dst_sw]; + if (next_sw >= 0) { + port_id = find_switch_link_port(ns, next_sw); + } + } + + if (port_id < 0) { + double shared_before = queued_mbit_on_switch(ns); + m->rc_no_route = 1; + m->rc_port_id = -1; + m->rc_accepted_mbit = 0.0; + m->rc_dropped_mbit = m->mbit; + m->rc_prev_scheduled_egress_interval = -1; + m->rc_log_target_is_terminal = 0; + m->rc_log_target_index = -1; + m->rc_log_capacity_mbit = 0.0; + m->rc_log_port_queued_before_mbit = 0.0; + m->rc_log_port_queued_after_mbit = 0.0; + m->rc_log_shared_queued_before_mbit = shared_before; + m->rc_log_shared_queued_after_mbit = shared_before; + m->rc_log_flowlet_remaining_after_mbit = 0.0; + m->rc_log_sent_total_mbit = 0.0; + m->rc_log_active_before_entries = 0; + m->rc_log_active_after_entries = 0; + ns->dropped_mbit += m->mbit; + log_switch_arrival_event(ns, m); + return; + } + + port_desc* p = &ns->ports[port_id]; + double port_queued_before = 0.0; + double shared_queued_before = 0.0; + double shared_queued_after = 0.0; + double dropped = 0.0; + double flowlet_remaining_after = 0.0; + int coalesced = 0; + int queue_index = -1; + + m->rc_no_route = 0; + m->rc_port_id = port_id; + m->rc_queue_index = -1; + m->rc_coalesced = 0; + m->rc_accepted_mbit = 0.0; + m->rc_dropped_mbit = 0.0; + m->rc_prev_scheduled_egress_interval = ns->scheduled_egress_interval[port_id]; + m->rc_log_target_is_terminal = p->is_terminal; + m->rc_log_target_index = p->target_index; + m->rc_log_capacity_mbit = p->capacity_mbit_per_interval; + m->rc_log_port_queued_before_mbit = 0.0; + m->rc_log_port_queued_after_mbit = 0.0; + m->rc_log_shared_queued_before_mbit = 0.0; + m->rc_log_shared_queued_after_mbit = 0.0; + m->rc_log_flowlet_remaining_after_mbit = 0.0; + m->rc_log_sent_total_mbit = 0.0; + m->rc_log_active_before_entries = 0; + m->rc_log_active_after_entries = 0; + + double accepted = enqueue_flowlet(ns, port_id, m, &port_queued_before, &shared_queued_before, + &shared_queued_after, &dropped, &flowlet_remaining_after, + &coalesced, &queue_index); + + m->rc_queue_index = queue_index; + m->rc_coalesced = coalesced; + m->rc_accepted_mbit = accepted; + m->rc_dropped_mbit = dropped; + + double port_queued_after = queued_mbit_on_port(ns, port_id); + int active_after = active_flowlet_count_on_port(ns, port_id); + + m->rc_log_port_queued_before_mbit = port_queued_before; + m->rc_log_port_queued_after_mbit = port_queued_after; + m->rc_log_shared_queued_before_mbit = shared_queued_before; + m->rc_log_shared_queued_after_mbit = shared_queued_after; + m->rc_log_flowlet_remaining_after_mbit = flowlet_remaining_after; + m->rc_log_active_after_entries = active_after; + + log_switch_arrival_event(ns, m); + + if (port_queued_after > EPS) { + request_switch_egress(ns, m->interval_id, port_id, lp, m); + } +} + + +static void send_flowlet_fragment(switch_state* ns, int port_id, const queued_flowlet* q, + double send_mbit, int interval_id, tw_lp* lp) { + if (send_mbit <= EPS) { + return; + } + fluid_msg out_msg; + memset(&out_msg, 0, sizeof(out_msg)); + out_msg.event_type = FLOWLET_ARRIVAL; + out_msg.interval_id = interval_id + 1; + out_msg.source_terminal = q->source_terminal; + out_msg.destination_terminal = q->destination_terminal; + out_msg.source_switch = ns->switch_id; + out_msg.creation_interval = q->creation_interval; + out_msg.flowlet_id = q->flowlet_id; + out_msg.mbit = send_mbit; + + const port_desc* p = &ns->ports[port_id]; + if (p->is_terminal) { + out_msg.destination_switch = ns->switch_id; + tw_lpid term_gid = get_terminal_gid(p->target_index); + schedule_arrival(interval_id + 1, term_gid, &out_msg, send_mbit, ns->switch_id, lp); + ns->delivered_local_mbit += send_mbit; + } else { + out_msg.destination_switch = p->target_index; + tw_lpid sw_gid = get_switch_gid(p->target_index); + schedule_arrival(interval_id + 1, sw_gid, &out_msg, send_mbit, p->target_index, lp); + } + ns->sent_mbit += send_mbit; + ns->sent_fragments++; +} + +static void handle_switch_egress(switch_state* ns, fluid_msg* m, tw_lp* lp) { + int port_id = m->port_id; + if (port_id < 0 || port_id >= ns->num_ports) { + tw_error(TW_LOC, "invalid switch egress port %d", port_id); + } + if (ns->queues[port_id] == NULL) { + m->rc_prev_scheduled_egress_interval = -1; + m->rc_prev_rr_next_queue_index = 0; + return; + } + + port_desc* p = &ns->ports[port_id]; + std::vector& qv = *ns->queues[port_id]; + + m->rc_prev_scheduled_egress_interval = ns->scheduled_egress_interval[port_id]; + m->rc_prev_rr_next_queue_index = ns->rr_next_queue_index[port_id]; + if (ns->scheduled_egress_interval[port_id] == m->interval_id) { + ns->scheduled_egress_interval[port_id] = -1; + } + + double capacity = p->capacity_mbit_per_interval; + double remaining_capacity = capacity; + double sent_total = 0.0; + double queued_before = queued_mbit_on_port(ns, port_id); + double shared_queued_before = queued_mbit_on_switch(ns); + int active_before = active_flowlet_count_on_port(ns, port_id); + + std::vector send_plan(qv.size(), 0.0); + m->rc_alloc_count = 0; + m->rc_log_target_is_terminal = p->is_terminal; + m->rc_log_target_index = p->target_index; + m->rc_log_capacity_mbit = capacity; + m->rc_log_port_queued_before_mbit = queued_before; + m->rc_log_port_queued_after_mbit = queued_before; + m->rc_log_shared_queued_before_mbit = shared_queued_before; + m->rc_log_shared_queued_after_mbit = shared_queued_before; + m->rc_log_flowlet_remaining_after_mbit = 0.0; + m->rc_log_sent_total_mbit = 0.0; + m->rc_log_active_before_entries = active_before; + m->rc_log_active_after_entries = active_before; + + auto add_to_plan = [&](int idx, double requested_send) -> double { + if (idx < 0 || idx >= (int)send_plan.size() || requested_send <= EPS) { + return 0.0; + } + + double still_available_for_flowlet = qv[idx].remaining_mbit - send_plan[idx]; + if (still_available_for_flowlet <= EPS) { + return 0.0; + } + + double actual_send = std::min(requested_send, still_available_for_flowlet); + send_plan[idx] += actual_send; + return actual_send; + }; + + if (strcmp(cfg.switch_scheduler, "fifo") == 0) { + for (int i = 0; i < (int)qv.size() && remaining_capacity > EPS; ++i) { + if (!qv[i].valid || qv[i].remaining_mbit <= EPS) { + continue; + } + + double requested_send = std::min(qv[i].remaining_mbit, remaining_capacity); + double planned_send = add_to_plan(i, requested_send); + remaining_capacity -= planned_send; + } + } else if (strcmp(cfg.switch_scheduler, "round_robin") == 0) { + /* + * Bounded round-robin: + * - keep a persistent per-port cursor, + * - visit queued flowlets circularly, + * - serve at most round_robin_max_entries_per_egress flowlets, + * - use a fixed quantum if configured, otherwise use one full + * output-port interval as the quantum. + * + * This avoids touching hundreds of active flowlets in one egress event, + * which keeps reverse metadata, logs, and optimistic rollback bounded. + */ + int qsize = (int)qv.size(); + int entries_to_serve = std::min(active_before, cfg.round_robin_max_entries_per_egress); + + if (qsize > 0 && entries_to_serve > 0 && remaining_capacity > EPS) { + int idx = ns->rr_next_queue_index[port_id]; + if (idx < 0 || idx >= qsize) { + idx = 0; + } + + double quantum = cfg.round_robin_quantum_mbit; + if (quantum <= EPS) { + /* + * Default round-robin quantum is one full output-port interval. + * This keeps the scheduler fair over time without splitting one + * interval into many tiny fragments. + */ + quantum = capacity; + } + + int visited = 0; + int served = 0; + while (remaining_capacity > EPS && visited < qsize && served < entries_to_serve) { + if (qv[idx].valid && qv[idx].remaining_mbit > EPS) { + double requested_send = std::min(qv[idx].remaining_mbit, quantum); + requested_send = std::min(requested_send, remaining_capacity); + double planned_send = add_to_plan(idx, requested_send); + remaining_capacity -= planned_send; + if (planned_send > EPS) { + served++; + } + } + + idx = (idx + 1) % qsize; + visited++; + } + + ns->rr_next_queue_index[port_id] = idx; + } + } else { + tw_error(TW_LOC, "unknown switch scheduler '%s'", cfg.switch_scheduler); + } + + for (int i = 0; i < (int)qv.size(); ++i) { + double send = send_plan[i]; + + if (!qv[i].valid || send <= EPS) { + continue; + } + + if (send > qv[i].remaining_mbit + EPS) { + tw_error(TW_LOC, + "computed send %.12f exceeds remaining %.12f for flowlet %llu " + "on switch %d port %d", + send, qv[i].remaining_mbit, (unsigned long long)qv[i].flowlet_id, + ns->switch_id, port_id); + } + + queued_flowlet before = qv[i]; + send = std::min(send, qv[i].remaining_mbit); + + if (m->rc_alloc_count >= MAX_RC_ALLOCATIONS) { + tw_error(TW_LOC, + "too many egress allocations for reverse metadata: switch %d port %d " + "interval %d count %d max %d", + ns->switch_id, port_id, m->interval_id, m->rc_alloc_count, MAX_RC_ALLOCATIONS); + } + + rc_alloc_record* rc = &m->rc_allocs[m->rc_alloc_count++]; + memset(rc, 0, sizeof(*rc)); + rc->valid = 1; + rc->queue_index = i; + rc->before = before; + rc->send_mbit = send; + + send_flowlet_fragment(ns, port_id, &before, send, m->interval_id, lp); + qv[i].remaining_mbit -= send; + sent_total += send; + } + + compact_port_queue(ns, port_id); + + /* + * This keeps the round-robin cursor valid after completed flowlets are removed. + * The cursor is restored exactly by rollback through rc_prev_rr_next_queue_index. + */ + if (strcmp(cfg.switch_scheduler, "round_robin") == 0) { + if (qv.empty()) { + ns->rr_next_queue_index[port_id] = 0; + } else { + ns->rr_next_queue_index[port_id] %= (int)qv.size(); + } + } + + double queued_after = queued_mbit_on_port(ns, port_id); + double shared_queued_after = queued_mbit_on_switch(ns); + int active_after = active_flowlet_count_on_port(ns, port_id); + + m->rc_log_port_queued_after_mbit = queued_after; + m->rc_log_shared_queued_after_mbit = shared_queued_after; + m->rc_log_sent_total_mbit = sent_total; + m->rc_log_active_after_entries = active_after; + + log_switch_egress_event(ns, m); + + if (queued_after > EPS) { + /* + * This event's reverse handler restores scheduled_egress_interval from + * rc_prev_scheduled_egress_interval. Pass NULL here so the next-event + * scheduling does not overwrite this event's saved pre-state. + */ + request_switch_egress(ns, m->interval_id + 1, port_id, lp, NULL); + } +} + + +static void switch_event(switch_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp) { + (void)b; + switch (m->event_type) { + case FLOWLET_ARRIVAL: + handle_switch_arrival(ns, m, lp); + break; + case SWITCH_EGRESS: + handle_switch_egress(ns, m, lp); + break; + default: + tw_error(TW_LOC, "switch received unknown event type %d", m->event_type); + } +} + +static void rollback_switch_arrival(switch_state* ns, fluid_msg* m) { + ns->received_fragments--; + + if (!m->rc_no_route && m->rc_port_id >= 0 && m->rc_port_id < ns->num_ports) { + ns->scheduled_egress_interval[m->rc_port_id] = m->rc_prev_scheduled_egress_interval; + } + + if (m->rc_no_route) { + ns->dropped_mbit -= m->rc_dropped_mbit; + return; + } + + int port_id = m->rc_port_id; + + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + tw_error(TW_LOC, "invalid rollback arrival port %d on switch %d", port_id, ns->switch_id); + } + + ns->dropped_mbit -= m->rc_dropped_mbit; + + if (m->rc_accepted_mbit <= EPS) { + return; + } + + std::vector& qv = *ns->queues[port_id]; + + int idx = m->rc_queue_index; + if (idx < 0 || idx >= (int)qv.size() || qv[idx].flowlet_id != m->flowlet_id) { + idx = find_queue_index_for_msg(ns, port_id, m); + } + + if (idx < 0) { + tw_error(TW_LOC, "could not find flowlet %llu on switch %d port %d during arrival rollback", + (unsigned long long)m->flowlet_id, ns->switch_id, port_id); + } + + if (m->rc_coalesced) { + qv[idx].remaining_mbit -= m->rc_accepted_mbit; + + if (qv[idx].remaining_mbit <= EPS) { + tw_error( + TW_LOC, + "coalesced flowlet %llu became empty during arrival rollback on switch %d port %d", + (unsigned long long)m->flowlet_id, ns->switch_id, port_id); + } + } else { + qv.erase(qv.begin() + idx); + } + + ns->enqueued_mbit -= m->rc_accepted_mbit; +} + +static void rollback_switch_egress(switch_state* ns, fluid_msg* m) { + int port_id = m->port_id; + + if (port_id < 0 || port_id >= ns->num_ports || ns->queues[port_id] == NULL) { + tw_error(TW_LOC, "invalid rollback egress port %d on switch %d", port_id, ns->switch_id); + } + + ns->scheduled_egress_interval[port_id] = m->rc_prev_scheduled_egress_interval; + ns->rr_next_queue_index[port_id] = m->rc_prev_rr_next_queue_index; + + std::vector& qv = *ns->queues[port_id]; + const port_desc* p = &ns->ports[port_id]; + + for (int r = m->rc_alloc_count - 1; r >= 0; --r) { + rc_alloc_record* rc = &m->rc_allocs[r]; + + if (!rc->valid || rc->send_mbit <= EPS) { + continue; + } + + int idx = rc->queue_index; + + if (idx < 0 || idx >= (int)qv.size() || qv[idx].flowlet_id != rc->before.flowlet_id) { + idx = find_queue_index_for_flowlet(ns, port_id, rc->before); + } + + if (idx >= 0) { + qv[idx] = rc->before; + } else { + int insert_idx = rc->queue_index; + + if (insert_idx < 0) { + insert_idx = 0; + } + if (insert_idx > (int)qv.size()) { + insert_idx = (int)qv.size(); + } + + qv.insert(qv.begin() + insert_idx, rc->before); + } + + ns->sent_mbit -= rc->send_mbit; + ns->sent_fragments--; + + if (p->is_terminal) { + ns->delivered_local_mbit -= rc->send_mbit; + } + } +} + +static void switch_rev_event(switch_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp) { + (void)b; + (void)lp; + + switch (m->event_type) { + case FLOWLET_ARRIVAL: + rollback_switch_arrival(ns, m); + break; + + case SWITCH_EGRESS: + rollback_switch_egress(ns, m); + break; + + default: + tw_error(TW_LOC, "switch reverse received unknown event type %d", m->event_type); + } +} + +static void switch_commit_event(switch_state* ns, tw_bf* b, fluid_msg* m, tw_lp* lp) { + (void)b; + (void)lp; + + if (!fluid_csv_commit_logs_enabled()) { + return; + } + + fluid_commit_logging_begin(); + + switch (m->event_type) { + case FLOWLET_ARRIVAL: + log_switch_arrival_event(ns, m); + break; + + case SWITCH_EGRESS: + log_switch_egress_event(ns, m); + break; + + default: + break; + } + + fluid_commit_logging_end(); +} + +static void switch_finalize(switch_state* ns, tw_lp* lp) { + double queued = 0.0; + for (int p = 0; p < ns->num_ports; ++p) { + queued += queued_mbit_on_port(ns, p); + } + printf("fluid-flow-wan gid=%llu switch=%d name=%s ports=%d shared_buffer_mbit=%.6f " + "received_fragments=%d sent_fragments=%d enqueued_mbit=%.6f sent_mbit=%.6f " + "local_delivery_mbit=%.6f dropped_mbit=%.6f queued_mbit=%.6f\n", + (unsigned long long)lp->gid, ns->switch_id, switches[ns->switch_id].name.c_str(), + ns->num_ports, ns->shared_buffer_mbit, ns->received_fragments, ns->sent_fragments, + ns->enqueued_mbit, ns->sent_mbit, ns->delivered_local_mbit, ns->dropped_mbit, queued); + + for (int p = 0; p < ns->num_ports; ++p) { + delete ns->queues[p]; + ns->queues[p] = NULL; + } +} + +const tw_optdef app_opt[] = {TWOPT_GROUP("interval-fluid switch/terminal workload"), TWOPT_END()}; + +static const char* find_config_arg(int argc, char** argv) { + for (int i = argc - 1; i >= 1; --i) { + if (argv[i] == NULL) { + continue; + } + const char* arg = argv[i]; + size_t len = strlen(arg); + if (len >= 5 && strcmp(arg + len - 5, ".conf") == 0) { + return arg; + } + } + return NULL; +} + +static void write_log_header_file(const char* path, const char* header) { + if (path[0] == '\0') { + return; + } + + std::remove(path); + + std::ofstream out(path, std::ios::out | std::ios::trunc); + if (!out) { + tw_error(TW_LOC, "could not open CSV log file '%s' for header", path); + } + out << header; +} + +static void append_merged_rows_to_file(const char* path, const char* data, int len) { + if (path[0] == '\0' || len <= 0) { + return; + } + + std::ofstream out(path, std::ios::app); + if (!out) { + tw_error(TW_LOC, "could not open CSV log file '%s' for merged append", path); + } + out.write(data, len); +} + +static void merge_log_buffer(const char* path, const std::ostringstream& buffer, MPI_Comm comm) { + if (path[0] == '\0') { + return; + } + + int rank = 0; + int size = 1; + MPI_Comm_rank(comm, &rank); + MPI_Comm_size(comm, &size); + + std::string local_rows = buffer.str(); + int local_len = (int)local_rows.size(); + + std::vector recv_counts; + std::vector displs; + if (rank == 0) { + recv_counts.resize(size, 0); + displs.resize(size, 0); + } + + MPI_Gather(&local_len, 1, MPI_INT, rank == 0 ? recv_counts.data() : NULL, 1, MPI_INT, 0, comm); + + int total_len = 0; + if (rank == 0) { + for (int i = 0; i < size; ++i) { + displs[i] = total_len; + total_len += recv_counts[i]; + } + } + + std::vector merged; + if (rank == 0 && total_len > 0) { + merged.resize(total_len); + } + + MPI_Gatherv(local_len > 0 ? local_rows.data() : NULL, local_len, MPI_CHAR, + rank == 0 && total_len > 0 ? merged.data() : NULL, + rank == 0 ? recv_counts.data() : NULL, rank == 0 ? displs.data() : NULL, MPI_CHAR, + 0, comm); + + if (rank == 0 && total_len > 0) { + append_merged_rows_to_file(path, merged.data(), total_len); + } +} + +static void merge_all_log_buffers(MPI_Comm comm) { + merge_log_buffer(cfg.terminal_log_path, terminal_log_buffer, comm); + merge_log_buffer(cfg.switch_log_path, switch_log_buffer, comm); + merge_log_buffer(cfg.flowlet_log_path, flowlet_log_buffer, comm); + merge_log_buffer(cfg.switch_training_log_path, switch_training_log_buffer, comm); +} + +static void write_log_headers(int rank) { + if (rank != 0) { + return; + } + + write_log_header_file( + cfg.terminal_log_path, + "interval,event,terminal,terminal_name,attached_switch,peer_terminal,mbit\n"); + + write_log_header_file(cfg.switch_log_path, + "interval,event,switch,switch_name,port,target_type,target_index," + "capacity_mbit,queued_before_mbit,sent_mbit,queued_after_mbit," + "shared_queued_before_mbit,shared_queued_after_mbit," + "shared_buffer_mbit,dropped_mbit,active_queue_entries\n"); + + write_log_header_file(cfg.flowlet_log_path, + "interval,event,switch,switch_name,port,target_type,target_index," + "flowlet_id,source_terminal,destination_terminal,creation_interval," + "age_intervals,capacity_mbit,queued_before_mbit,send_mbit," + "remaining_after_mbit,dropped_mbit\n"); + + write_log_header_file(cfg.switch_training_log_path, + "interval,switch,switch_name,port,target_type,target_index,scheduler," + "capacity_mbit,queued_before_mbit,sent_mbit,queued_after_mbit," + "dropped_mbit,active_before,active_after,allocations\n"); +} + +int main(int argc, char** argv) { + int rank = 0; + + g_tw_ts_end = seconds_to_ns(60.0 * 60.0 * 24.0 * 365.0); + + tw_opt_add(app_opt); + tw_init(&argc, &argv); + + const char* config_file = find_config_arg(argc, argv); + if (config_file == NULL) { + printf("Usage: mpirun -np %s --synch=1 -- \n", argv[0]); + MPI_Finalize(); + return 1; + } + + MPI_Comm_rank(MPI_COMM_CODES, &rank); + configuration_load(config_file, MPI_COMM_CODES, &config); + load_config(); + validate_ross_message_size_or_abort(rank); + + add_lp_types(); + codes_mapping_setup(); + + total_switch_lps = codes_mapping_get_lp_count(GROUP_NAME, 0, SWITCH_LP_NAME, NULL, 1); + total_terminal_lps = codes_mapping_get_lp_count(GROUP_NAME, 0, TERMINAL_LP_NAME, NULL, 1); + if (total_switch_lps != (int)switches.size()) { + tw_error(TW_LOC, "config defines %d switch LPs but topology YAML defines %zu switches", + total_switch_lps, switches.size()); + } + if (total_terminal_lps != (int)terminals.size()) { + tw_error(TW_LOC, "config defines %d terminal LPs but topology YAML defines %zu terminals", + total_terminal_lps, terminals.size()); + } + + write_log_headers(rank); + MPI_Barrier(MPI_COMM_CODES); + + if (rank == 0) { + printf("fluid-flow-wan config: switches=%zu terminals=%zu interval_seconds=%.6f " + "num_send_intervals=%d num_drain_intervals=%d switch_scheduler=%s " + "buffer_mode=shared csv_logs=%s ross_message_size=%d fluid_msg_size=%zu\n", + switches.size(), terminals.size(), cfg.interval_seconds, cfg.num_send_intervals, + cfg.num_drain_intervals, cfg.switch_scheduler, + fluid_csv_forward_logs_enabled() ? "buffered-forward" : "buffered-commit", + get_configured_message_size_bytes(), sizeof(fluid_msg)); + } + + tw_run(); + merge_all_log_buffers(MPI_COMM_CODES); + tw_end(); + return 0; +} diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 294d5c44..26b58954 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -56,7 +56,7 @@ add_test(NAME mpi-multirank-sanity # REPEAT for "same command N times", VARIANTS when only a small fragment differs # (e.g. VARIANTS "--sync=1" "--sync=3" for a seq-vs-optimistic check). function(codes_add_equivalence_test) - cmake_parse_arguments(T "" "NAME;BINARY;NP;CONFIG;MARKER;REPEAT;SETUP;REQUIRE" "ARGS;VARIANTS" ${ARGN}) + cmake_parse_arguments(T "" "NAME;BINARY;NP;CONFIG;MARKER;REPEAT;SETUP" "ARGS;VARIANTS;REQUIRE" ${ARGN}) if(NOT T_NAME OR NOT T_BINARY OR NOT T_CONFIG) message(FATAL_ERROR "codes_add_equivalence_test: NAME, BINARY and CONFIG are required") @@ -101,9 +101,11 @@ function(codes_add_equivalence_test) set(_setup --setup "${CMAKE_CURRENT_SOURCE_DIR}/${T_SETUP}") endif() set(_require "") - if(T_REQUIRE) - set(_require --require "${T_REQUIRE}") - endif() + foreach(_req IN LISTS T_REQUIRE) + if(NOT _req STREQUAL "") + list(APPEND _require --require "${_req}") + endif() + endforeach() add_test(NAME ${T_NAME} COMMAND "${CMAKE_CURRENT_BINARY_DIR}/run-test.sh" @@ -135,7 +137,7 @@ endfunction() # [MARKER ] # ) function(codes_add_run_test) - cmake_parse_arguments(T "" "NAME;BINARY;NP;SETUP;MARKER" "ARGS" ${ARGN}) + cmake_parse_arguments(T "" "NAME;BINARY;NP;SETUP;MARKER" "ARGS;REQUIRE" ${ARGN}) if(NOT T_NAME OR NOT T_BINARY) message(FATAL_ERROR "codes_add_run_test: NAME and BINARY are required") endif() @@ -145,12 +147,17 @@ function(codes_add_run_test) set(_run ${MPIEXEC_EXECUTABLE} ${MPIEXEC_NUMPROC_FLAG} ${T_NP} ${MPIEXEC_PREFLAGS} "$" ${T_ARGS} ${MPIEXEC_POSTFLAGS}) - if(T_SETUP OR T_MARKER) - # SETUP must be sourced and MARKER checked -> go through the staged runner. + if(T_SETUP OR T_MARKER OR T_REQUIRE) + # SETUP must be sourced and MARKER/REQUIRE checked -> go through the staged runner. set(_opts "") if(T_MARKER) list(APPEND _opts --marker "${T_MARKER}") endif() + foreach(_req IN LISTS T_REQUIRE) + if(NOT _req STREQUAL "") + list(APPEND _opts --require "${_req}") + endif() + endforeach() if(T_SETUP) list(APPEND _opts --setup "${CMAKE_CURRENT_SOURCE_DIR}/${T_SETUP}") endif() @@ -311,15 +318,49 @@ if(USE_UNION) set_tests_properties(${union-shell-files} PROPERTIES LABELS "nightly" TIMEOUT 1200) endif() + +# Fluid-flow WAN smoke tests. +# +# These tests start from the CMake-generated doc/example/fluid-flow-wan.conf, +# rewrite only the scheduler and output paths in an isolated test directory, +# and verify that the model runs cleanly and writes CSV logs. +foreach(_ffw_sched fifo round_robin) + foreach(_ffw_mode sequential optimistic) + if(_ffw_mode STREQUAL "sequential") + set(_ffw_synch 1) + set(_ffw_np 1) + else() + set(_ffw_synch 3) + set(_ffw_np 2) + endif() + + set(_ffw_test "fluid-flow-wan-${_ffw_sched}-${_ffw_mode}") + + add_test(NAME ${_ffw_test} + COMMAND "${CMAKE_CURRENT_BINARY_DIR}/run-test.sh" + "${CMAKE_CURRENT_SOURCE_DIR}/fluid-flow-wan-ci.sh" + "${_ffw_sched}" + "${_ffw_synch}" + "${_ffw_np}" + "${_ffw_test}" + "${MPIEXEC_EXECUTABLE}" + "${MPIEXEC_NUMPROC_FLAG}" + WORKING_DIRECTORY "${CODES_BINARY_DIR}") + + set_tests_properties(${_ffw_test} PROPERTIES TIMEOUT 120) + endforeach() +endforeach() + # Equivalence / determinism tests (replacing per-scenario shell scripts). # example-ping-pong-determinism.sh: run the optimistic sim twice, compare # committed-event counts (reproducibility). codes_add_equivalence_test( NAME example-ping-pong-determinism BINARY tutorial-synthetic-ping-pong - NP 3 - ARGS --sync=3 --num_messages=10 --payload_sz=8192 - CONFIG doc/example/tutorial-ping-pong.conf) + NP 3 + ARGS --sync=3 --num_messages=10 --payload_sz=8192 + CONFIG doc/example/tutorial-ping-pong.conf + REQUIRE "START PARALLEL OPTIMISTIC SIMULATION") # Surrogate-mode determinism: same surrogate config run twice (np 3, sync=3), # committed-event counts must match. The two variants differ only in network @@ -331,7 +372,7 @@ codes_add_equivalence_test( ARGS --sync=3 --num_messages=100 --payload_sz=8192 CONFIG tutorial-ping-pong-surrogate.conf SETUP surrogate-determinism-no-freeze-setup.sh - REQUIRE "Network switch completed") + REQUIRE "Network switch completed" "START PARALLEL OPTIMISTIC SIMULATION") codes_add_equivalence_test( NAME example-ping-pong-surrogate-determinism-freeze BINARY tutorial-synthetic-ping-pong @@ -339,7 +380,7 @@ codes_add_equivalence_test( ARGS --sync=3 --num_messages=100 --payload_sz=8192 CONFIG tutorial-ping-pong-surrogate.conf SETUP surrogate-determinism-freeze-setup.sh - REQUIRE "Network switch completed") + REQUIRE "Network switch completed" "START PARALLEL OPTIMISTIC SIMULATION") # Smoke test: the sim runs cleanly with an empty packet-latency trace path. # Single run (SETUP generates the config; MARKER asserts it produced output). @@ -364,7 +405,8 @@ codes_add_run_test(NAME rc-stack-test BINARY rc-stack-test) codes_add_run_test(NAME modelnet-prio-sched-test-seq BINARY modelnet-prio-sched-test NP 1 ARGS --sync=1 -- "${_tconf}/modelnet-prio-sched-test.conf") codes_add_run_test(NAME modelnet-prio-sched-test-opt BINARY modelnet-prio-sched-test - NP 2 ARGS --sync=3 -- "${_tconf}/modelnet-prio-sched-test.conf") + NP 2 ARGS --sync=3 -- "${_tconf}/modelnet-prio-sched-test.conf" + REQUIRE "START PARALLEL OPTIMISTIC SIMULATION") codes_add_run_test(NAME jobmap-test BINARY jobmap-test ARGS "${_tconf}/jobmap-test-list.conf") codes_add_run_test(NAME map-ctx-test BINARY map-ctx-test ARGS "${_tconf}/map-ctx-test.conf") codes_add_run_test(NAME resource-test BINARY resource-test ARGS --sync=1 "--codes-config=${_tconf}/buffer_test.conf") diff --git a/tests/fluid-flow-wan-ci.sh b/tests/fluid-flow-wan-ci.sh new file mode 100755 index 00000000..499eb59c --- /dev/null +++ b/tests/fluid-flow-wan-ci.sh @@ -0,0 +1,112 @@ +#!/bin/bash +set -euo pipefail + +scheduler="${1:?scheduler required: fifo or round_robin}" +synch="${2:?synch mode required: 1 or 3}" +np="${3:?MPI rank count required}" +case_name="${4:-fluid-flow-wan-${scheduler}-synch${synch}}" +mpi_exec="${5:-mpirun}" +mpi_np_flag="${6:--np}" + +if [[ "$scheduler" != "fifo" && "$scheduler" != "round_robin" ]]; then + echo "unsupported scheduler: $scheduler" + exit 1 +fi + +if [[ -z "${bindir:-}" ]]; then + echo "bindir is not set; this script should be run through tests/run-test.sh" + exit 1 +fi + +if [[ -z "${srcdir:-}" ]]; then + echo "srcdir is not set; this script should be run through tests/run-test.sh" + exit 1 +fi + +binary="$bindir/src/model-net-fluid-flow-wan" +base_conf="$bindir/doc/example/fluid-flow-wan.conf" + +if [[ ! -x "$binary" ]]; then + echo "missing executable: $binary" + exit 1 +fi + +if [[ ! -f "$base_conf" ]]; then + echo "missing generated config: $base_conf" + exit 1 +fi + +topology="$bindir/doc/example/fluid-flow-wan-topology.yaml" +if [[ ! -f "$topology" ]]; then + topology="$srcdir/doc/example/fluid-flow-wan-topology.yaml" +fi + +if [[ ! -f "$topology" ]]; then + echo "missing topology file" + exit 1 +fi + +rm -rf "$case_name" +mkdir -p "$case_name/logs" +cp "$topology" "$case_name/fluid-flow-wan-topology.yaml" + +sed \ + -e "s|switch_scheduler=\"[^\"]*\";|switch_scheduler=\"$scheduler\";|" \ + -e 's|topology_yaml_file="[^"]*";|topology_yaml_file="fluid-flow-wan-topology.yaml";|' \ + -e 's|terminal_log_path="[^"]*";|terminal_log_path="logs/terminal-events.csv";|' \ + -e 's|switch_log_path="[^"]*";|switch_log_path="logs/switch-events.csv";|' \ + -e 's|flowlet_log_path="[^"]*";|flowlet_log_path="logs/flowlet-events.csv";|' \ + -e 's|switch_training_log_path="[^"]*";|switch_training_log_path="logs/switch-training.csv";|' \ + "$base_conf" > "$case_name/fluid-flow-wan.conf" + +for expected in \ + "switch_scheduler=\"$scheduler\";" \ + 'topology_yaml_file="fluid-flow-wan-topology.yaml";' \ + 'terminal_log_path="logs/terminal-events.csv";' \ + 'switch_log_path="logs/switch-events.csv";' \ + 'flowlet_log_path="logs/flowlet-events.csv";' \ + 'switch_training_log_path="logs/switch-training.csv";' +do + if ! grep -q "$expected" "$case_name/fluid-flow-wan.conf"; then + echo "could not rewrite generated config; missing: $expected" + cat "$case_name/fluid-flow-wan.conf" || true + exit 1 + fi +done + +if ! ( + cd "$case_name" + "$mpi_exec" "$mpi_np_flag" "$np" "$binary" --sync="$synch" -- fluid-flow-wan.conf \ + > model-output.txt 2> model-output-error.txt +); then + echo "fluid-flow-wan model run failed" + echo "--- stdout ---" + cat "$case_name/model-output.txt" || true + echo "--- stderr ---" + cat "$case_name/model-output-error.txt" || true + exit 1 +fi + +out="$case_name/model-output.txt" + +grep "fluid-flow-wan config:" "$out" +grep "switch_scheduler=$scheduler" "$out" +grep "Net Events Processed" "$out" + +if [[ "$synch" == "1" ]]; then + grep "csv_logs=buffered-forward" "$out" +else + grep "csv_logs=buffered-commit" "$out" +fi + +for csv in \ + "$case_name/logs/terminal-events.csv" \ + "$case_name/logs/switch-events.csv" \ + "$case_name/logs/flowlet-events.csv" \ + "$case_name/logs/switch-training.csv" +do + if [[ ! -s "$csv" ]]; then + echo "missing or empty CSV log: $csv" + exit 1 + fi +done