Repository navigation
fix(distributed): keep remote pipeline stages on the dense kernel path - #2671
Conversation
Implementation Review SummaryIntent
FindingsNo CRITICAL or HIGH findings; no fix commit.
Remaining Items
Verification
|
Security and performance reviewFixed (HIGH), 25e93de: Performance: no regressions. The stage prefill used to build a dense causal global mask and a sliding mask on every prefill, including above 4096 tokens. It now uses the dense path's helpers: global layers take maskless causal SDPA, the Gemma 4 path keeps the Remaining (pre-existing, outside this diff, authenticated peers only):
|
Two independent defects made 2-stage remote pipeline logits differ from the single-process model on GB10 CUDA. The stage-boundary wire decoder rebuilt float16 activations through `from_bytes_f16`, a checkpoint loader that widens float16 to float32 off Apple. Checkpoints that load as float16 on CUDA (llama-3.2-1b, gemma-3-1b) therefore ran their downstream stage and LM head in float32. `deserialize_wire_tensor` now rebuilds every dtype as sent through `from_bytes`. Measured on llama-3.2-1b-4bit: stage 0 output was byte-identical to the dense hidden state at the split, and only the wire round trip moved the logits (max abs diff 1.5e-2). The Gemma 3 and Gemma 4 stage models built explicit causal and sliding masks for every prefill, while the dense forwards leave both `None` inside the window and take the maskless causal SDPA. Both now share one prefill-mask helper with their dense forward (`text_prefill_sliding_mask`, also used by the Gemma 3 backbone, and `text_prefill_masks`). In-process stage logits differed by 9.4e-2 (gemma-3-1b) and 3.4e-1 (gemma-4-e2b) before, and are byte-identical now. New lib tests fail before the fix on CUDA: a dtype-preserving wire round trip, and two-stage synthetic Gemma 3 / Gemma 4 splits that must match dense logits bit for bit. The PP CI lib step now also runs the two split suites. Closes #2654
deserialize_wire_tensor handed a peer-supplied shape and payload straight to mlxcel_core::from_bytes, which copies product(shape) elements from a raw pointer without looking at the payload length. A remote stage or coordinator peer that sent a large shape with a short payload made the receiver read past the heap buffer. Neither the wire header decode nor deserialize_tensor compared data_len with the shape, and the old from_bytes_f16 branch had no check either; this PR made from_bytes the path for every dtype, so the guard belongs here. The function now builds a RawTensorData and goes through kv_cache_serde::reconstruct_mlx_array, the same guarded constructor the KV-cache path uses: non-negative dims, checked element product with an empty shape counted as one scalar element, and an exact byte-length match before the FFI call. Dtype-preserving behaviour from #2654 is unchanged. New wire_tensor tests reject a short f16 payload, a short f32 payload, and an empty payload for a scalar shape, and round-trip a hand-built bf16 buffer. wire_tensor:: (8) and kv_cache_serde:: (41) pass on GB10; clippy -D warnings and fmt are clean. Refs #2654
The Gemma3 and Gemma4 two-stage parity suites only compared single-shot prefill and decode with the dense model. Add a continuation case to each: a second multi-token prefill on a non-empty cache, once inside the sliding window and once crossing it, followed by a decode step, all compared bitwise to dense. Both pass. The pipeline-parallel CI paths filter did not list the Gemma model files or their stage suites, so a change touching only those never ran the suites. Add src/models/gemma3.rs, gemma3_backbone.rs, gemma4.rs and the two stage test files. Also fix the Gemma3 stage test header (the fixture is f16, not bf16) and narrow sliding_prefill_plans_native in gemma3.rs to file-private since nothing else uses it. Validation: cargo fmt --check, clippy (test-fast, cuda, lib and tests, -D warnings) clean; wire_tensor, gemma3, gemma4 and gemma3_backbone lib modules 108 passed, 0 failed under gpu-lock with MLXCEL_SDPA_DETERMINISTIC=1. Refs #2654
1205a31 to
196d171
Compare
… (#2708) Closes #2694. The seven residuals of #2671, #2680 and #2684 in one PR, each with its own test. Reconciled against current main: the handshake `PROTOCOL_VERSION` stays 4 (#2705), the `FrameRejected` call sites keep their `handshake_warn` budget unchanged, and `ProtectedStream::peer_closed` (#2700) is untouched. ## Per item 1. `read_frame` interprets the kind byte after `FrameOpener::open` on a protected stream: a tampered invalid kind is `FrameRejected` through the existing budgeted call sites (no call-site change); an authenticated unknown kind is `unknown message kind {n} from an authenticated peer`. Plain streams and the header-only oversize check are unchanged. 2. `encode_message_frame` reserves the header and `MAX_TAG_LEN` (32, const-asserted against `FrameSuite::ALL`); `finish_frame` is the one sealing path, used by `write_prepared_frame` (`TcpTransport::send`) and by `write_frame(kind, &[u8])`. A buffer shorter than the header is an error, not a panic, and `write_prepared_frame` is `pub(crate)`. `serialize_message` is test-only. 3. `ActivationMessage::reply_nonce` (`serde(default)`) carries the `RunEntry` nonce from the entry stage through middle stages; the final stage sends `RunResult` to the coordinator's authenticated identity (the source of its most recent command, so a coordinator configured on `0.0.0.0` still gets the reply across hosts). A reverse activation, or a nonce-less forward activation on the final stage, is dropped with a warn. `PendingEntryReply`, its map and the `pending_entry_replies` fields are gone. `ExchangeMatch::result_peer` is `stage_peers.last()` for `RunEntry` and the command's peer otherwise; `Ack`, `State`, `Error` keep `peer`. Frames per decoded token: 3 (`RunEntry`, forward activation, `RunResult`) instead of 4. 4. `render_peer_assignment_toml` (shared `render_toml_body`) is what a `--pp-peer` writes; the coordinator renderer and its callers are unchanged. 5. `Activate` / `Abort { detail }` after `Committed`, `ASSIGNMENT_PROTOCOL_VERSION` 2. The peer delivers only on `Activate`; `Abort`, another frame, close or the commit timeout reset `committed` and return it to listening. The coordinator activates all peers after every commit (a failed `Activate` write is an `AssignmentPushFailed` entry whose message says the other peers are running) or aborts every committed peer (detail cut to 1 KiB) when any commit failed. The peer logs a received detail bounded to 1 KiB and escaped to one line. The residual window is in the module docs. 6. `text_prefill_masks` returns `(None, None)` when `sliding_attn` is `None`. 7. `two_stage_split_matches_dense_after_front_trim_bitwise` (12-token prefill, `trim_front(4)` on every `Cache::Standard`, 3-token continuation, one decode step, bitwise against dense). Docs: `docs/distributed.md` (assignment protocol, peer TOML header, reply authorization, 3-frame per-token flow); no Korean counterpart exists for these sections. ## Failing-first evidence Run against the unchanged behaviour before each fix, green after: - Item 1 `protected_stream_rejects_tampered_invalid_kind_as_frame_rejected`: `a tampered invalid kind must surface as FrameRejected, got: unknown message kind: 255`; `protected_stream_reports_authenticated_unknown_kind_distinctly`: `unexpected error: unknown message kind: 255`. - Item 3 `final_stage_sends_run_result_to_coordinator_not_upstream`: `stage must answer within 5s (it may have died): Elapsed(())`; `run_result_from_last_stage_completes_forward`: `remote stage 10.1.1.1:1 timed out after 5s`; `run_result_from_entry_stage_is_dropped_in_multi_stage_pipeline`: `a RunResult from a stage other than the last must not complete the exchange`; `final_stage_sends_run_result_to_authenticated_coordinator_identity` (coordinator configured as `[::]:port`): `Elapsed(())` when the reply went to the configured address. - Item 5 `peer_discards_committed_assignment_on_abort_and_activates_on_the_next`: `the peer delivered an assignment that was never activated`; `coordinator_aborts_committed_peers_when_another_commit_fails`: `the real peer delivered an assignment the coordinator aborted`. - Item 6 `text_prefill_masks_without_sliding_layer_builds_no_mask`: `a stage without a sliding layer must not allocate a sliding mask`. ## Item 7 sensitivity A temporary mutation (never committed, file restored byte for byte, sha256 `de5f8b65…c457036` before and after) gave the stage path an offset-sized global prefill mask (`Some(create_causal_mask(seq_len, first_cache_offset(caches, "full_attention")))` instead of `None`). The trim test then fails with `3-token continuation prefill after trim_front(4): stage logits differ from the dense logits (max abs diff 5.46875e-2)`. It is the only gemma4 stage test whose scenario has `offset != live_len` (12 vs 8). ## Decode throughput `pipeline_remote_decode_bench` (qwen3-0.6b-4bit, 2 stages, loopback, 128 tokens, 7 timed repeats after a warm-up per run), main 433560c (A) and this branch (B) interleaved A B over 5 rounds on a quiet GB10, each run under `gpu-lock`. Both binaries were built with the bench's fixed port moved to 19990 (local only, not committed). Greedy tokens are identical in all 80 repeats. | | per-round medians (tok/s) | median of medians | range | |---|---|---|---| | main | 120.67, 140.70, 131.38, 140.29, 136.73 | 136.73 | 120.67 to 140.70 | | branch | 149.33, 142.75, 159.04, 156.94, 162.01 | 156.94 | 142.75 to 162.01 | The branch is about 15% faster, and its slowest round beats main's fastest. ## Verification (GB10 spark-102, `--profile test-fast --features cuda,test-utils`, test binaries built outside the lock and run under `gpu-lock`, `--test-threads=1`, `MLXCEL_REQUIRE_MODELS=1 MLXCEL_SDPA_DETERMINISTIC=1`, at 7c8d340) - `cargo fmt --check` and `cargo clippy ... --lib --tests -- -D warnings`: clean. - `--lib distributed::`: 1494 passed, 4 ignored. `--lib models::gemma4`: 127 passed, 15 ignored. - `zero_config_cluster_init --include-ignored`: 7/7 (the `zero_config_cli_*` rehearsals drive the real binaries through Activate). - `pipeline_remote_real_models --include-ignored --skip glm_moe_dsa`: 10/10. `pipeline_server_remote_real_models --include-ignored --skip glm_moe_dsa`: 7/7. The skipped `glm_moe_dsa` tests need the 418.6 GB `glm5-4bit` checkpoint, which cannot load on a 128 GB GB10. - `pipeline_ci_multi_stage_real_models --include-ignored`: 4/4, including `pipeline_multi_stage_three_host_real_model_parity` (logits now skip the middle stage) and the two-host smoke. - `disaggregated_serving_e2e` 2/2, `disaggregated_router_e2e` 7/7, `disaggregated_e2e` 51/51 (all `--include-ignored`). ## Compatibility The stage RPC messages change shape (`pending_entry_replies` is gone from `Ack` and `State`, `RunResult` comes from the final stage) while the handshake `PROTOCOL_VERSION` stays 4. A node built from main between #2705 and this PR therefore completes the handshake with a node built from this PR and then fails with exchange timeouts instead of a version error. No tagged release is affected (the latest is v0.8.0-beta.1, before the v3 and v4 handshakes); upgrade all nodes together.
Root cause
Two independent defects, found by comparing hidden states at the split boundary on GB10 before changing anything:
deserialize_wire_tensorrebuilt float16 activations withfrom_bytes_f16, a checkpoint loader that converts float16 to float32 everywhere except Apple. Checkpoints that load as float16 on CUDA (llama-3.2-1b, gemma-3-1b) then ran the downstream stage and LM head in float32. On llama-3.2-1b-4bit, stage 0 output was byte-identical to the dense hidden state at the split, in-process stage logits were byte-identical to dense, and only the wire round trip moved the logits (max abs diff 1.5e-2, float32 instead of float16). qwen3 passed because it stays bfloat16, which that loader keeps native.Gemma3StageModelandGemma4StageModelbuilt explicit causal and sliding masks for every prefill, while the dense forwards leave bothNoneinside the window and take the maskless causal SDPA. In-process stage logits (no wire) differed from dense by 9.4e-2 (gemma-3-1b) and 3.4e-1 (gemma-4-e2b, split 13).Fix
kv_cache_serde::reconstruct_mlx_array, whose existingvalidate_raw_tensorcheck (from feat: harden paged KV handoff deserialization boundary #186) rejects a payload whose length does not match the shape before the FFI copy. The wire path previously bypassed that check, so negative dims, overflowing element counts, and a short payload were unchecked there (a heap over-read from an authenticated peer; this predates the PR).text_prefill_sliding_mask) and Gemma 4 (text_prefill_masks) mask decisions are shared helpers used by the dense model, the stage model, andGemma3Backbone. Dense behavior is unchanged in every branch, and stages now also keep the 4096-token no-dense-mask cutoff from fix: gemma-4 long-prompt prefill OOMs on CUDA because attention materializes O(heads*L^2) scores instead of using flash SDPA #672.All three families now give byte-identical logits to dense, in-process and over the wire. No tolerance was changed.
Regression coverage
Non-ignored lib tests:
wire_tensor::tests(f16 round trip, plus short f16/f32 and empty-scalar payload rejection) andgemma3_stage_tests/gemma4_stage_tests, which split a synthetic model (stage 1 opening on a global layer) and require dense logits bit for bit for 2- and 8-token prefill, continuation prefill on a non-empty cache inside and across the sliding window, and decode. The f16 round trip and both 2-token split tests fail against main's sources (diffs 8.8e-3 and 2.3e-2).GB10 CI coverage decision
The real-model remote parity tests are not added to CI: the PP workflow's two-host job runs on
ubuntu-latest(CPU) with only the qwen3 fixture, and no GB10 job runs the PP suites or holds llama/gemma checkpoints. Insteadrun-pp-two-host-logical.shruns the two weight-free split suites (itsdistributed::pipeline::filter already covers the wire tests), and the workflow's path filter now includes the Gemma model and test files so a dense-mask change triggers them.Validation
On GB10 (spark-102) with
MLXCEL_REQUIRE_MODELS=1 MLXCEL_SDPA_DETERMINISTIC=1 --test-threads=1at the final head:pipeline_remote_real_models6/6 (llama, gemma3, gemma4, qwen3, qwen3.5 parity, llama drain),pipeline_server_remote_real_models5/5, andpipeline_multi_stage_two_host_logical_smoke;pipeline_stage_executor_real_models10/10 at the first commit (its gemma variants and llama worker loop also failed on main). Libdistributed::pipeline::,models::gemma3,models::gemma4,models::gemma3_backbone,kv_cache_serde, fmt and clippy pass. The new lib tests have not yet run on a CPU build; PP CI on this PR is the first. Metal not checked; on Apple the float16 wire path was already native, so only the mask change applies there.Closes #2654