Collectives

Scope: Days AGO and Legacy Days. Both engines model collective traffic, not application payload arithmetic, and they accept different algorithms.

Days AGO collectives

Days AGO lowers each collective into stage flows in the SimulationImage. Every message of every rank is an ordinary TCP flow or RoCE queue pair that carries one chunk, together with the list of stages it waits for. A [[compute]] group adds one delay-only stage per host. Groups wait for one another through after lists, matched by host: rank r of a group waits, at its host, for each listed group that runs on that host. A stage with several predecessors of one kind is a counted join; it starts when every predecessor is complete.

On the Spectrum-X rail fabric, a message between two GPUs of one server does not enter the network: it lowers to a stage notify, a delay-only NVLink message. See Spectrum-X. An AICB training trace lowers through the same stage groups from a [workload.aicb] table; see Workloads.

This scenario runs a forward compute stage on three hosts, then a TCP ring all-reduce, then a backward compute stage:

seed = 26
edges = [[0, 3], [1, 3], [2, 3]]
hosts = [0, 1, 2]
duration = 0.05

[switch]
port_rate = 8000000000
capacity = 100
discipline = "FIFO"
drop = "TailDrop"

[[compute]]
name = "forward"
hosts = [0, 1, 2]
duration_ns = 5000

[[collective]]
name = "grad"
after = "forward"
collective_type = "RingAllReduce"
flow_type = "TCP"
flow_count = 3
sources = [0, 1, 2]
sinks = [1, 2, 0]

[collective.traffic]
initial_delay = 0.0
size = 9001
arr_dist = { type = "Uniform", low = 0.000001, high = 0.000001 }
pkt_size_dist = { type = "DiscreteUniform", low = 500, high = 500 }

[collective.traffic.tcp]
cc_algorithm = "TCPReno"

[[compute]]
name = "backward"
hosts = [0, 1, 2]
duration_ns = 7000
after = "grad"

Save it as chain.toml and run it from the repository root:

cargo run --release --bin days -- chain.toml --engine scalar
cargo run --release --bin days -- chain.toml --engine cpu --workers 2

Both runs end with result_bytes=69710 result_fnv1a64=bf4bb10a72685679. --engine metal and --engine cuda run it too (see Backends).

[[collective]] fields

Lowering rejects first_flow_id, graph, routing, and paths: flow identity comes from the collective stage key, and routes come from the topology.

[[collective_set]] accepts the five algorithms with per-member sources and sinks lists (an all-to-all or send/receive member takes sources only); each member lowers as the same [[collective]] would. A set has no name, after, stream, channels, chunk, or alltoall key. Like every Days AGO table, it refuses an unknown key, naming the key and the table it sits in, so a stage-group key written on a set, or a misspelled key anywhere, fails to lower rather than being ignored. The only keys Days AGO accepts without reading are the legacy engine's twelve root keys (listed on Overview). A traffic table's arr_dist and pkt_size_dist must be one-line inline tables with no comment and no trailing comma.

[[compute]] fields

Lowering rejects an unknown or duplicate name and a cycle of after references.

Host-matched after

after is resolved host by host. For each host h where the dependent group starts, and for each listed group G that runs on h, the dependent's stage at h waits for G's completion at h. A collective starts on all its ranks; a SendRecv starts at its sender only. When a listed group runs on exactly the dependent's hosts in the same order, rank r waits for rank r; otherwise the match is by host, and the two groups may have different sizes and orders.

Three rules bound the lists:

• every host where the dependent starts must run at least one listed group, or lowering fails with "{kind} x rank r (host h) starts after no operation its after lists";
• every listed group must run on at least one of those hosts ("collective x names stage group y, which runs on none of the hosts it starts at"; "compute x names stage group y, which runs on none of its hosts");
• a collective's list must contain at least one compute group ("unsupported collective x dependency on collective y alone; a collective follows at least one compute group"). A collective with no after starts at its traffic's initial_delay and needs no compute group.

This scenario runs two expert compute groups on disjoint host pairs, then a data-parallel all-gather over all four hosts whose ranks interleave the pairs, then an optimizer stage:

seed = 26
edges = [[0, 4], [1, 4], [2, 4], [3, 4]]
hosts = [0, 1, 2, 3]
duration = 0.05

[switch]
port_rate = 8000000000
capacity = 200
discipline = "FIFO"
drop = "TailDrop"

[[compute]]
name = "ep0"
hosts = [0, 1]
duration_ns = 2000

[[compute]]
name = "ep1"
hosts = [2, 3]
duration_ns = 3000

[[collective]]
name = "dp"
after = ["ep0", "ep1"]
collective_type = "AllGather"
flow_type = "TCP"
flow_count = 4
sources = [0, 2, 1, 3]
sinks = [2, 1, 3, 0]

[collective.traffic]
initial_delay = 0.0
size = 8000
arr_dist = { type = "Uniform", low = 1, high = 1 }
pkt_size_dist = { type = "DiscreteUniform", low = 500, high = 500 }

[collective.traffic.tcp]
cc_algorithm = "TCPReno"

[[compute]]
name = "opt"
hosts = [0, 1, 2, 3]
duration_ns = 1000
after = "dp"

Hosts 0 and 1 start their all-gather roots when ep0 ends at 2,000 ns; hosts 2 and 3 when ep1 ends at 3,000 ns. Scalar and CPU end this scenario with result_bytes=76218 result_fnv1a64=4b6275b54d5c3678.

A data-parallel chain uses the same lists: a collective can name the compute group that releases it and the previous collective of its stream, so the second collective starts at each host only once the first has completed there:

[[collective]]
name = "rs2"
after = ["wg2", "rs1"]
stream = 1
collective_type = "ReduceScatter"
flow_type = "TCP"
flow_count = 4
sources = [0, 1, 2, 3]
sinks = [1, 2, 3, 0]

Endpoints

For a ring algorithm with explicit endpoints, the ring must be in ring-next order: sinks[i] == sources[(i + 1) % n]. Every participant must be a configured host, and the sources must be distinct. If sources and sinks are omitted, the ring uses every configured host in ascending host-ID order, and each host sends to the next one, wrapping around at the end. flow_count must then equal the number of configured hosts.

With channels, the collective runs several rings over the same ranks at once, as SimAI's NCCL ring channels do. Each channel is a ring order of all the sources, and sinks is omitted:

[[collective]]
name = "ag"
after = "fwd"
collective_type = "AllGather"
flow_type = "TCP"
flow_count = 4
sources = [0, 1, 2, 3]
channels = [[0, 1, 2, 3], [0, 2, 1, 3]]
chunk = "UniformFloor"

An all-to-all's sources are its ranks; omitted, they are every configured host in ascending order. A SendRecv's sources are [sender, receiver].

Stages and messages

Each algorithm lowers to a fixed set of stages. S is the declared byte size, n the rank count, and c the channel count (1 without channels):

The two chunk policies differ in how S is divided:

• EqualRemainderLast (rings, the default): S is split into n chunks of S / n bytes, and the last chunk also carries the remainder, so the ring moves exactly S bytes per phase.
• UniformFloor (rings with several channels, and every all-to-all): every message is floor(floor(S / n) / c) bytes for a ring and floor(S / n) for an all-to-all, SimAI's rule; the remainder is not sent.

A stage flow is a TCP flow or a RoCE queue pair from the stage's source to its target that carries its message, using the collective's transport settings. Because a message must carry at least one byte, S must be at least n for every algorithm, and a UniformFloor ring needs floor(floor(S / n) / c) >= 1; otherwise lowering fails with "AllGather of 7 bytes over 4 ranks and 2 channels sends empty messages; every message must carry a byte".

Seeded all-to-all

[collective.alltoall] gives an all-to-all per-pair sizes from a seeded token-routing model, the imbalanced-routing arm of an MoE workload:

[[collective]]
name = "dispatch"
after = "fwd"
collective_type = "AllToAll"
flow_type = "RoCE"
flow_count = 4
sources = [0, 1, 2, 3]

[collective.alltoall]
seed = 7
matrix = 0
transpose = false
experts = 8
topk = 2
tokens = 16
bytes_per_copy = 100
skew = "Zipf1"

The matrix is integer arithmetic throughout: a SplitMix64 stream keyed by (seed, matrix, group, source), a seeded permutation of the experts, popularity weights floor(2^40 / (p + 1)) for the expert at popularity rank p under Zipf1 (equal weights under Uniform), and tokens * topk draws mapped to experts through the integer cumulative weights. A pair's bytes are its copies times bytes_per_copy; copies to the source itself are not sent, and a pair with zero copies has no stage. The matrix is computed at lowering, so every backend runs the same sizes. The example above routes 3,200 bytes per source rank (16 * 2 * 100); a combine that answers it sets transpose = true with the same seed, matrix and group.

When a stage starts

Each stage lists its predecessors, in two kinds:

• a local predecessor is a stage at the same host. It is complete at the first acknowledgement that covers all of its bytes (TCP and RoCE), or when its timer fires (a compute stage, or the lead of a stage notify);
• an inbound predecessor is a stage delivered to this host. It is complete when the host's in-order receive frontier for that stage reaches the stage's whole message. Retransmitted or out-of-order bytes count only once the frontier covers them, and a partial delivery changes no stage state.

A stage keeps a count of its completed predecessors of each kind and starts at the event that brings both counts to their requirement. A TCP stage sends its first congestion window from sequence 0 at that instant; a RoCE stage starts as described in RoCE stages.

For a ring step, the local predecessor is the same rank's previous step on the same channel (for the first all-gather step of RingAllReduce, the rank's last reduce-scatter step), and the inbound predecessor is the previous rank's previous step on the same channel. The first step of each channel has no predecessor in the collective: with after, it waits for the listed groups' stages at its host; without, it starts at the traffic's initial_delay.

Every stage of an all-to-all is a root: a rank's sends start together, at its release by the listed groups at its host. A SendRecv's one stage starts at its sender's release.

Completion

A group that lists a collective in its after waits, at each host, for the collective's completion there:

• Rings. At rank r's host: the acknowledgement that completes r's final stage on each channel (local), and the delivery of the final stage that targets the host on each channel (inbound). With c channels that is c + c predecessors.
• All-to-all. The acknowledgement of every message the rank sends (local) and the delivery of every message that targets the host (inbound). A seeded rank that sends nothing completes at the all-to-all's release at its host (the groups the all-to-all's own after names there) and the delivery of its arrivals; a group listed both by the all-to-all and by the dependent counts once.
• Send/receive. At the sender, the acknowledgement of the message (local); at the receiver, its delivery (inbound). The receiver's next group lists the SendRecv as any other predecessor, and the coverage rules of Host-matched after apply at the sender only.

Two limits of the zero-send case are recorded rather than fixed: in an all-to-all with no after, a rank that neither sends nor receives starts at time 0 rather than at initial_delay; and the after lists are not part of the LeanGuard certificate, so a certificate that consistently drops such a rank's gate is accepted (see LeanGuard).

RoCE stages

With flow_type = "RoCE", each stage is a RoCE queue pair: Mellanox-form DCQCN rate control, fed by the ECN echo on its ACKs and NACKs, with Go-back-N recovery, as described for RoCE flows. The collective's tables replace the flow's:

[[collective]]
collective_type = "RingAllReduce"
flow_type = "RoCE"
priority = 3
flow_count = 4
sources = [0, 2, 1, 3]
sinks = [2, 1, 3, 0]

[collective.traffic]
initial_delay = 0.0
size = 400000
arr_dist = { type = "Uniform", low = 1, high = 1 }
pkt_size_dist = { type = "DiscreteUniform", low = 1000, high = 1000 }

[collective.traffic.dcqcn]
rate_gbps = 1.0
min_rate_gbps = 0.01
max_rate_gbps = 1.0
g = 0.00390625
ai_rate_gbps = 0.005
hai_rate_gbps = 0.05
rp_timer_ns = 50000
pacing_interval_ns = 1000

[collective.traffic.roce]
retransmit_timeout_ns = 1000000
feedback_priority = 0

configs/p15/roce_compute_dag.toml puts this collective between two compute groups on a four-switch line with host-link PFC and an ECN step. Scalar and CPU at 2 workers both end it with result_bytes=208285 result_fnv1a64=b9462a2b3cbe0b26.

A RoCE stage differs from a TCP stage in four points:

• Start. A stage that starts at time t behaves exactly like a queue pair whose initial_delay is t. Its first pacing tick is at t and its pacing grid is anchored at t. Its DCQCN controller is pristine until the stage's first ECN echo, and its retransmission timeout is armed by its first packet. Before it starts, a stage sends nothing and has no pending event.
• Inbound progress. The receiver discards out-of-order packets (Go-back-N), so only the packet whose PSN equals the receiver's expected PSN advances the inbound predecessor.
• Completion. The local predecessor is complete at the first ACK that acknowledges all of its bytes. A NACK never completes a stage.
• Pause. If the stage's data class is paused at its host when it starts (host-link PFC), its first tick sends nothing and parks the queue pair, and the RESUME restarts it on the grid anchored at t.

With the timeout off (retransmit_timeout_ns = 0), a loss is repaired only by a NACK, which a later packet of the same stage triggers. If the last data packet or the last ACK of a stage is lost, the stage stalls: it never completes, so its successors never start, and the run continues to its stop time without an error. Two outputs show the stall:

• In the final state, the stalled stage's generator is not Finished (its snd_una is below its total), and its successors' stage records still read activated: false.
• In the collective trace, no local-completion row names the stalled stage as its cause.

With the timeout off, a stage is safe from a stall only if neither its data nor its feedback can be dropped: no loss on its data class and none on its feedback_priority class. PFC on the data class alone is not enough. A switch queue's capacity counts packets of every class, so unprotected packets that share the queue, such as traffic on a class without PFC or a burst of 64-byte ACKs, can fill it and tail-drop protected data too. The example above protects only its data class: its ACKs and NACKs ride class 0, which has no PFC, so other unprotected traffic on the same switches can make it stall.

To check a run, read dropped_packets in the run summary (days prints it in its result line). A run that dropped no packet has no stall of this kind. If it dropped packets, the two signals above show whether a stage stalled.

Compute stages

A compute stage is a timer on its host; it sends no packets and has no route. It finishes duration_ns after it starts:

• A compute group without after starts at time 0.
• With after, the stage at host h starts when every listed group has completed at h: a compute group's stage there when its timer fires, and a collective as described in Completion.

In the first example, every forward stage finishes at 5,000 ns, and each backward stage finishes 7,000 ns after it starts. A stage whose finish time falls after the scenario's stop time does not finish.

Traces

With Full observation, Scalar and CPU record one row per collective progress event; collective_transitions_csv in the executor library serializes them. stage_kind is tcp, roce, compute, or notify, and duration_ns is the compute duration, or a notify's delay (0 on TCP and RoCE rows). packet_size_bytes is the MSS on TCP rows and the MTU on RoCE rows; interval_ns is 0 on TCP rows and the pacing interval on RoCE rows. A compute row whose inbound predecessor is a RoCE stage carries that queue pair's MTU and pacing interval in the same two columns, so a checker can replay its inbound packets as the Go-back-N frontier even when the predecessor itself is not logged; every other compute row writes 0 in both. Compute rows leave algorithm and collective_phase empty.

The dependency columns name the stage's predecessors and its progress through them: local_predecessors and inbound_predecessors (flow ids, ;-separated), local_required, before_local_completed and after_local_completed (the local count), the inbound flags and byte columns, and cause_total_bytes, the whole size of the predecessor a row's cause belongs to. channel, chunk_policy, channel_policy and group_stages describe the collective's shape, and seeded_matrix names a seeded all-to-all's matrix, which the checker re-derives. Five further columns certify completion signals:

• segment_sequence, segment_bytes: on inbound rows, the data segment that arrived (its TCP sequence number or RoCE PSN, and its size). arrival_bytes is how far it advanced the receiver's in-order frontier, which is 0 for a duplicate or out-of-order packet;
• ack_number, cause_origin_ns, cause_delay_ns: on a local completion caused by an acknowledgement, the acknowledgement number, the send time it echoes, and the unloaded round trip;
• cause_origin_ns, cause_delay_ns: on a local completion caused by a compute timer, the time the timer was armed and its duration.

Four columns bind each row to the flows it names: cause_kind (tcp, roce, compute, or notify: what carries the cause), cause_collective_id (the collective of a notify cause), and the two delivery hosts target_node (the host the row's own stage delivers to; a compute stage's own host) and cause_target_node (the host the cause is delivered to). A column that does not apply to a row is 0. The LeanGuard checker p10c_collective_check checks these traces; see LeanGuard.

Backends

Metal and CUDA run collectives, compute stages, counted joins and stage notifies over TCP and RoCE byte-identically to Scalar: complete state and the full-observation measurement planes. They do not return the collective transition records, which are diagnostics; generate those on Scalar or CPU.

Errors

The compiler and the validator name the rejected field or backend:

unsupported collective algorithm `...`; Days lowers RingAllReduce, AllGather, ReduceScatter, AllToAll and SendRecv
unsupported collective flow type `...`; collectives require a reliable transport, flow_type = "TCP" or "RoCE" (a RoCE queue pair is DCQCN with Go-back-N)
SendRecv flow_count must be 2: sources = [sender, receiver]
AllToAll takes `sources` only: no `sinks`
TCP collective byte size 2 must be at least flow_count 3
a collective of several ring channels requires chunk = "UniformFloor"
every ring channel must order exactly the collective's sources
`[collective.alltoall]` sizes apply to an AllToAll only
a seeded all-to-all's size 3000 must equal tokens x topk x bytes_per_copy
collective flow_count 3 must match the 4 configured hosts when sources/sinks are omitted
a collective on a stream other than 0 needs a `name`
compute `...` rank 2 (host 2) starts after no operation its `after` lists
collective `...` names stage group `...`, which runs on none of the hosts it starts at
unsupported collective `...` dependency on collective `...` alone; a collective follows at least one compute group
compute `...` host 7 must be a configured host attachment

Legacy [[collective]]

Legacy Days expands collectives into ordinary flow actors before routing.

[[collective]]
collective_type = "Broadcast"
flow_type = "PacketDistribution"
flow_count = 2
sources = [0, 0]
sinks = [1, 2]

[collective.traffic]
initial_delay = 1.0
duration = 10.0
arr_dist = { type = "Uniform", low = 1.0, high = 1.0 }
pkt_size_dist = { type = "DiscreteUniform", low = 1000, high = 1500 }

If you provide more than one endpoint representation, graph edges, path endpoints, and explicit sources/sinks must agree.

Legacy endpoint rules are algorithm-specific:

• Broadcast requires one repeated source.
• Gather requires one repeated sink.
• AllReduce is a fan-in: it uses Gather's endpoint pattern, many sources and one sink.
• RingAllReduce requires sources[i] == sinks[i - 1] around the ring.
• With no explicit endpoints, Broadcast and Gather/AllReduce sample endpoints from hosts. RingAllReduce uses every configured host and requires flow_count == hosts.len().

Legacy RingAllReduce accepts byte termination only and requires size >= flow_count. It expands reduce-scatter and all-gather phases into 2 * flow_count * (flow_count - 1) concrete flows.

Legacy [[collective_set]]

[[collective_set]]
collective_type = "Gather"
collective_count = 2
flow_type = "PacketDistribution"
flow_count = 2
sources = [[0, 1], [1, 2]]
sinks = [[2, 2], [3, 3]]

[collective_set.traffic]
initial_delay = 1.0
duration = 10.0
arr_dist = { type = "Uniform", low = 1.0, high = 1.0 }
pkt_size_dist = { type = "DiscreteUniform", low = 512, high = 512 }

collective_count gives the number of members in the set. Each nested source/sink list must contain flow_count endpoints. Legacy reserves a contiguous concrete-flow range for each member; RingAllReduce uses its expanded flow count for that reservation. Days AGO also accepts [[collective_set]] for all five of its algorithms, with the per-member rules of [[collective]] and without its stage-group keys (see [[collective]] fields).

Legacy TCP application buffers

Legacy TCP Broadcast shares one collective-wide byte buffer across receivers. TCP RingAllReduce creates one buffer per directed ring link and gives each hop a byte-range handle. The actors bound and pace traffic; they do not mutate or reduce payload data.

Two consequences of this design affect TCP RingAllReduce timing. A stage does not send when it is released: its sender pulls bytes from the buffer actor, which answers only on polling ticks, so each stage start is delayed to the next tick. A stage whose chunk exceeds the initial congestion window of two segments must pull again mid-flow, and that run stops with a Nexosim simulation deadlock; only chunks of at most two segments run.

Configure polling under [app_source] as described in Legacy runtime and CSV configuration.