Repository navigation
cluster: a v2 request sends the batch rows once, each expert naming its rows by index - #1970
Merged
Merged
Conversation
This was referenced Oct 9, 2026
Avicennasis
force-pushed
the
p/cluster-rows-once
branch
from
October 9, 2026 21:26
7d36d9a to
d2fda03
Compare
Contributor
Author
|
Restacked on #1964's new head: this branch is now The rebase had one conflict, in Ran here on the same Linux box, against
The push should restart the cancelled matrix. |
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.
Avicennasis
force-pushed
the
p/cluster-rows-once
branch
from
October 16, 2026 16:58
d2fda03 to
e1ff5cd
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 #1964 (which is stacked on #1963): the change to review is the last commit,
7d36d9ab.A v1 request carries a copy of the input row inside every item it routes to, so at decode with top-k the same row crosses the wire k times (GLM-5.2 decode, top-8, D=6144: eight copies of a 24 KB row per layer). In a v2 request the coordinator now sends
Sand the S batch rows once (inact, so q8 under q8), and each item iseid nrplusnrrow indices; the worker gathers (nr ≤ S,idx < S,S ≤ 65536). Replies are unchanged. v1 keeps row copies, so default bytes are untouched. The shared-expert item is the index list0..S-1.Tests:
test_v2_sends_the_rows_oncepins the exact 104 bytes of a top-2 request where row 0 routes to experts 42 and 43 and row 1 to 42 only, shared item riding; the worker gathers and the rows accumulate exactly.tests/test_cluster_sharding.pynow prints bytes per position: sent 4102 → 2080 B/position on the top-2 fixtures (1.97×, the remainder is indices, headers and the hello; GLM-5.2 decode would be ~8×), asserted assent_v1 ≥ 0.9·K·sent_v2with K fromconfig.json; f32 parity and the byte-exact logits of the shared-expert run unchanged; the q8 drift gate goes from 2317 to 1758 B/position (4.67×) with mismatches still 0, since each row is q8-encoded once by the same encoder.make colibri tests/test_cluster_protocol0 warnings, 9 protocol cases, 4 gate tests passed.