Repository navigation
cluster: a layer's requests to every worker before any reply is read - #1967
Merged
Merged
Conversation
Contributor
Author
|
CI on this PR has no verdict yet: of its 114 checks, 75 are CANCELLED and 39 SUCCESS (read from the check rollup on 2026-10-09); no job failed. The branch still merges cleanly against For context: #1968 is closed as superseded by #1774 ( |
This was referenced Oct 16, 2026
Owner
…tocol v2 The expert-worker protocol moves every routed row as raw f32 both ways, so on a LAN the transfer, not the expert FFN, bounds a delegated layer. distributed-llama syncs its activations as blocks of 32 int8 behind one scale for a ~4x cut. Same idea here, behind an env switch and off by default: the cluster parity gate is token-exact and int8 rows are not. - COLI_CLUSTER_ACT=q8: the coordinator speaks a v2 header (magic, version 2, act, layer, D, I, n) whose rows are ceil(D/32) f32 block scales followed by D int8 -- qrow_i8 per block, the engine's own activation rounding, so no new quantizer -- and the worker answers in the same format. f32, the default, sends the v1 bytes unchanged, so a cluster of mixed binaries keeps working. - The worker accepts v1 and v2, names the versions it speaks when it refuses one, and refuses an unknown act word by name. A coordinator whose v2 request meets a closed socket names the worker and the COLI_CLUSTER_ACT=f32 escape hatch instead of just "failed". The worker's response-header writes free the batch when the peer is gone mid-header, as a bad request already did; the act word had widened that leak. - The coordinator reports bytes sent and received at exit, so the cut is measurable: on the tiny GLM oracle, 8203 -> 2316 B/position (3.54x) with the same argmax at every position. - test_cluster_protocol: the q8 codec byte for byte against qrow_i8 (partial last block, zero row, decode within half a block scale), a v2 q8 round trip over a socketpair, and a v2 header with act=f32 still carrying raw rows.
test_cluster_sharding's f32 gate is unchanged: the same assertions, now on shared _expert_worker/_coordinator helpers. ClusterQ8DriftTest runs the delegated oracle with COLI_CLUSTER_ACT=q8 on both fixtures and asserts that the f32 delegation stays exact, that the q8 run scores the same positions and mismatches at most Q8_MAX_MISMATCHES of them (measured 0 on fmt6 and fmt4; the bound carries no slack, raise it with a measurement), and that q8 moves less than a third of f32's bytes; it prints bytes per position for the record.
…ert on the worker
On top of the q8 protocol v2 (COLI_CLUSTER_ACT=q8, header magic version act
layer D I n): the default still speaks v1 bytes, so a cluster of mixed engine
builds keeps working. The coordinator speaks v2 when it needs q8 rows or the
shared expert, and v2 gains two words v1 has none for:
- hello: an empty v2 request (n==0, act as the coordinator will speak it) at
connect time, sent only when v2 will be spoken at all. A worker built from
an older engine closes on version 2, and cluster_init refuses it by name
("expert worker HOST:PORT closed on the COLIEX01 v2 hello: it runs an older
engine (v1); rebuild the worker, or run this coordinator without
COLI_CLUSTER_ACT=q8 and CLUSTER_SHARED=1") before any token is generated. A
worker answers the hello before its model checks, with an empty response in
the same act; its refusals of an unknown version or act are unchanged.
- item eid 0xFFFFFFFF: the layer's shared expert. With CLUSTER_SHARED=1 the
coordinator adds one item of every row to the layer's first routed request,
on the worker layer%n picks, in the request's act; the worker loads the
three shared matrices once (qt_load + qt_planarize, the coordinator's own
resident branch) and runs FASE E's CPU sequence (matmul_qt, siluf,
matmul_qt); the coordinator adds the rows where FASE E adds its own, so
with f32 rows the f32 order of the sum does not move. Off by default.
GLM-5.2: 2.83 B params a token off the coordinator. A v2 request may carry
65 items (64 routed + the shared one); v1 stays at 64.
DUMP=<path> (raw f32 logits rows) was compiled only under COLI_VULKAN; it is
now on every build so the cluster gate can compare runs byte for byte.
Tests:
- tests/test_cluster_protocol.c: the hello against a v1 worker (refused, by
name and version, read back from stderr) and a v2 worker (accepted); the
real cluster_moe_batch over a socketpair with the shared item (a v2 request)
and without it (v1 bytes, the default).
- tests/test_cluster_sharding.py: the f32 parity gate gains a third run with
CLUSTER_SHARED=1 that must match the local run's teacher-forcing signature,
show zero oracle mismatches, and match the local run's logits bytes; the
worker's stderr must name every sparse layer it loaded the shared expert
for. A 0.1 % scale on the worker's shared output flips no token on the tiny
fixtures but fails the byte comparison; zeroing it fails the oracle on
27/32 and 30/32 positions. The q8 drift gate is unchanged; its transfer
summary now includes the hello's 56 bytes per worker per run.
…nnect The coordinator's exchange with a worker -- the request header and items, then the reply's header and items -- is one function now, cluster_exchange, and a transport or protocol failure in it (a short read, EOF mid-reply, a magic, version, act, status, item id or row count that is not the one asked for) marks that worker's link failed. Every later exchange on a failed link returns -1 before touching the socket: nothing is sent, nothing is read, so the bytes a worker still sends after a failed exchange -- the rest of a reply the coordinator stopped reading, or a well-formed reply behind it -- are never read as the answer to a later request. A link comes back only by reconnecting. The hello's refusals mark the link the same way. cluster_moe_batch is unchanged in what it does: it builds the items, calls the exchange, and on -1 prints the worker by name and exits, as before. The rule is ds4's after a one-sided transport failure (antirez/ds4 f0962e3): refuse everything on the link, never consume drained bytes as a reply. tests/test_cluster_protocol.c pins it over a socketpair in two shapes. A worker whose reply header claims two items for a one-item request, with a well-formed 52-byte reply queued right behind it: the first exchange returns -1, the second returns -1 at once and the 52 bytes are still on the link, unread, and the worker saw no second request. A worker whose reply stops after the first of two rows and shuts its write side: the same, with nothing pending. Without the link guard the second exchange reads the queued reply as its own (the test fails on rc == -1); with it, 8 of 8 runs pass. 0 warnings.
…ts rows by index A v1 request copies a row into every item that routes it: at decode with top-K the same row crosses the wire K times (GLM-5.2, top-8: eight copies of a 24 KB row per layer). A v2 request now carries S and the batch's S rows once, in `act`, then per item eid, nr and nr row indices (u32, network order); the worker gathers each item's rows from the batch. Replies are per item as before. v1 requests keep their copies, so the default bytes are unchanged; the shared-expert item names every row by index instead of carrying S more copies, and the coordinator no longer copies rows per item for v2 either (cluster_item's copy_inputs). Measured on the tiny GLM oracle (32 positions, top-2, D=256), bytes the coordinator sends per position: v1 4102, v2 2080 (1.97x; the rest is indices and headers). q8 in total: 2317 -> 1758 B/position (f32 8203, 4.67x), the same 0 oracle mismatches. tests/test_cluster_protocol: the exact 104 bytes of a two-expert, top-2 v2 request where row 0 routes to both experts and row 1 to one, the shared item riding (header, S, the two rows once, three index lists); the worker gathers and the rows accumulate exactly. The q8 round trip and the shared-expert test move to the new layout. tests/test_cluster_sharding: f32 parity and q8 drift unchanged; the parity gate prints the bytes sent per position for v1 and v2 and requires the v1/v2 ratio to be at least 0.9 K.
cluster_moe_batch served the workers in turn: build worker 0's batch, send it, wait for its reply, then worker 1's. A layer therefore took the SUM of the workers' times and, with N workers, paid N round trips end to end. Now every worker's request goes on the wire first, then every reply is read, in worker order -- the order the routed rows were summed in before, so the f32 sum does not move and the parity gate stays token-exact. The bytes per worker are unchanged (protocol v1, no wire change); a layer waits for its slowest worker instead of for each of them. tests/test_cluster_protocol: two socketpair workers that will not answer until both have received their request (a 2 s barrier), each then holding its reply 200 ms. Served in turn, the barrier times out and the test fails (checked against the previous cluster_moe_batch: `bar.late == 0` fails); with fan-out the batch takes 200 ms, not 400. The accumulated rows are checked exactly. test_cluster_sharding's f32 parity passes unchanged on both fixtures.
Avicennasis
force-pushed
the
p/cluster-fanout
branch
from
October 16, 2026 16:58
d6f0ddd to
50b1f8d
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #1970 (which is stacked on #1964, itself on #1963): the change to review is the last commit,
50b1f8df; the diff shows the whole stack until those merge.cluster_moe_batchserved the workers in turn: build worker 0's request, send it, wait for its reply, then worker 1. With N workers a layer costs the sum of the round trips instead of the slowest one. This splits it into two phases, send every worker its request, then read the replies in worker order, so the f32 accumulation order is exactly the serial code's. No wire change: the bytes each worker sees are the v1 bytes it saw before.Test: two socketpair workers that refuse to reply until both have received their request (a 2 s barrier), each then holding 200 ms. The serial code deadlocks on the barrier (negative control:
Assertion 'bar.late == 0' failed); the fan-out batch takes 200 ms, not 400, with the accumulation checked exactly (expert 0 ×2 on worker 0, expert 1 ×3 on worker 1, both rows).make colibri tests/test_cluster_protocolwarning-free; the token-exact gatetests/test_cluster_sharding.pyunchanged, 2 passed on fmt6 and fmt4.The timing assertion is
elapsed < 1.5 × hold; the barrier assertion is the deterministic discriminator if a loaded CI box ever stretches the clock.