feat(ds41): two-box expert-parallel serving

server/CMakeLists.txt

@@ -59,6 +59,12 @@ set_property(CACHE LUCE_GPU_BACKEND PROPERTY STRINGS cuda hip)
 option(LUCE_ENABLE_MIXED_CUDA_HIP
     "Primary GPU runtime plus an isolated in-process CUDA/HIP peer (Linux only)"
     OFF)
+# Expert-parallel cluster (after maikzz32/lucebox-halo-cluster): one
+# luce_server rank per box, routed experts split across the ranks, one RCCL
+# all-reduce per MoE layer over RoCE. The GPU-free cluster sources (config,
+# protocol, control channel, placement, decisions) are always compiled; ON
+# adds the RCCL communicator.
+option(LUCE_CLUSTER "Multi-box RCCL expert-parallel cluster (HIP only)" OFF)
 string(TOLOWER "${LUCE_GPU_BACKEND}" LUCE_GPU_BACKEND)
 if(LUCE_ENABLE_MIXED_CUDA_HIP)
     # Reject unsupported targets before project() tries to discover both GPU
@@ -199,6 +205,9 @@ endif()
 # No BLAS, no Metal, no Vulkan, no examples/tests/tools.
 
 if(LUCE_GPU_BACKEND STREQUAL "cuda")
+    if(LUCE_CLUSTER)
+        message(FATAL_ERROR "LUCE_CLUSTER requires LUCE_GPU_BACKEND=hip (RCCL)")
+    endif()
     set(GGML_CUDA ON CACHE BOOL "" FORCE)
     if(LUCE_ENABLE_MIXED_CUDA_HIP)
         set(GGML_HIP ON CACHE BOOL "" FORCE)
@@ -234,7 +243,11 @@ elseif(LUCE_GPU_BACKEND STREQUAL "hip")
         set(GGML_CUDA_MODULE OFF CACHE BOOL "" FORCE)
     endif()
     set(GGML_HIP  ON  CACHE BOOL "" FORCE)
-    set(GGML_HIP_RCCL OFF CACHE BOOL "" FORCE)
+    if(LUCE_CLUSTER)
+        set(GGML_HIP_RCCL ON CACHE BOOL "" FORCE)
+    else()
+        set(GGML_HIP_RCCL OFF CACHE BOOL "" FORCE)
+    endif()
     set(GGML_HIP_GRAPHS OFF CACHE BOOL "Enable HIP graph replay (held off: first-capture quality variance, see experiment log E30/T12)")
     set(LUCE_GGML_BACKEND_TARGET ggml-hip)
     set(LUCE_HIP_ARCHITECTURES "" CACHE STRING "HIP GPU targets, e.g. gfx906;gfx1100")
@@ -279,6 +292,31 @@ if(LUCE_ROCMFP2_AFFINE)
     add_compile_definitions(ROCMFP2_AFFINE=1)
 endif()
 
+# Ranks exchange this build identity in the control-channel Hello and refuse
+# to form a cluster from different binaries.
+if(LUCE_CLUSTER)
+    add_compile_definitions(LUCE_CLUSTER=1 LUCE_CLUSTER_RCCL=1)
+    find_package(Git QUIET)
+    set(_luce_build_sha "unknown")
+    if(GIT_FOUND)
+        execute_process(
+            COMMAND ${GIT_EXECUTABLE} describe --always --dirty --tags
+            WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
+            OUTPUT_VARIABLE _luce_build_sha_out
+            RESULT_VARIABLE _luce_build_sha_rc
+            OUTPUT_STRIP_TRAILING_WHITESPACE
+            ERROR_QUIET)
+        if(_luce_build_sha_rc EQUAL 0 AND _luce_build_sha_out)
+            set(_luce_build_sha "${_luce_build_sha_out}")
+        endif()
+        unset(_luce_build_sha_out)
+        unset(_luce_build_sha_rc)
+    endif()
+    add_compile_definitions(LUCE_BUILD_SHA="${_luce_build_sha}")
+    message(STATUS "luce cluster: LUCE_CLUSTER=ON (build sha ${_luce_build_sha})")
+    unset(_luce_build_sha)
+endif()
+
 # HIP Phase 2 — opt-in rocWMMA flashprefill kernels (Strix Halo / gfx1151).
 # Default OFF (Phase 1 = ggml q8 fallback). Set ON for the 1.7-2.7× compress
 # speedup at 8K-32K context. Requires rocwmma headers installed under
@@ -582,8 +620,10 @@ add_library(luce_common STATIC
     src/deepseek4/deepseek4_target_shard_ipc_daemon.cpp
     src/deepseek4/deepseek4_dspark.cpp
     src/deepseek4/deepseek4_dspark_spec.cpp
+    src/deepseek4/deepseek4_cluster.cpp
     src/deepseek4/deepseek4_vision.cpp
     src/common/gpu_page_pool.cpp
+    src/common/host_stream_flag.cpp
     src/common/vision/image_decode.cpp
     src/common/vision/image_resize.cpp
     src/common/vision/mmproj_file.cpp
@@ -671,6 +711,18 @@ add_library(luce_common STATIC
     src/common/feature_gate.cpp
     src/placement/placement_config.cpp
     src/placement/device_select.cpp
+    # Expert-parallel cluster runtime. GPU-free parts are always compiled;
+    # cluster_comm.cpp's RCCL half and fast_reduce.cpp's RDMA half are under
+    # LUCE_CLUSTER.
+    src/cluster/cluster_config.cpp
+    src/cluster/cluster_protocol.cpp
+    src/cluster/cluster_control.cpp
+    src/cluster/cluster_comm.cpp
+    src/cluster/fast_reduce.cpp
+    src/cluster/cluster_expert_placement.cpp
+    src/cluster/cluster_decision_hooks.cpp
+    src/cluster/cluster_head_backend.cpp
+    src/cluster/cluster_worker_main.cpp
     src/common/layer_split_utils.cpp
     src/common/ddtree.cpp
     src/common/peer_access.cpp
@@ -748,6 +800,19 @@ elseif(LUCE_GPU_BACKEND STREQUAL "hip")
     # hip_compat shim is needed by ALL luce_common sources (peer_access.cpp,
     # dflash_feature_ring.cpp, flashprefill.cpp), not just the SM80_EQUIV path.
     target_include_directories(luce_common PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/hip_compat)
+    if(LUCE_CLUSTER)
+        # RCCL communicator plus the f32<->bf16 kernels behind
+        # allreduce_sum_bf16_compressed.
+        find_package(rccl REQUIRED)
+        target_sources(luce_common PRIVATE src/cluster/cluster_comm_kernels.cu)
+        set_source_files_properties(src/cluster/cluster_comm_kernels.cu PROPERTIES LANGUAGE HIP)
+        target_link_libraries(luce_common PUBLIC roc::rccl)
+        # Lean decode all-reduce over raw RDMA (stream wait/write flags, no RCCL kernel).
+        find_library(LUCE_IBVERBS_LIB ibverbs REQUIRED)
+        target_sources(luce_common PRIVATE src/cluster/fast_reduce_kernels.cu)
+        set_source_files_properties(src/cluster/fast_reduce_kernels.cu PROPERTIES LANGUAGE HIP)
+        target_link_libraries(luce_common PUBLIC ${LUCE_IBVERBS_LIB})
+    endif()
 endif()
 
 if(LUCE_GPU_BACKEND STREQUAL "cuda")
@@ -1075,6 +1140,7 @@ if(LUCE_TESTS)
         test_ds4_mix_fitter)
     set(_luce_cpu_cppunit_targets
         test_feature_gate
+        test_cluster_unit
         test_platform_compat
         test_paged_kv_pool
         test_blocking_row_pool
@@ -1375,6 +1441,12 @@ if(LUCE_TESTS)
     target_include_directories(test_qwen35_tensor_parallel PRIVATE ${LUCE_SRC_INCLUDE_DIRS})
     target_link_libraries(test_qwen35_tensor_parallel PRIVATE luce_common)
 
+    if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/test/bench/bench_ds41_q8_mmvq.cpp")
+        # Benchmark, not a test: DS4.1 dense Q8_0 projections through the ggml
+        # mul_mat dispatch at 1-4 columns, with the column-identity check.
+        add_executable(bench_ds41_q8_mmvq test/bench/bench_ds41_q8_mmvq.cpp)
+        target_link_libraries(bench_ds41_q8_mmvq PRIVATE ggml ggml-base ${LUCE_GGML_BACKEND_TARGET})
+    endif()
     add_executable(test_rope_tail test/unit/test_rope_tail.cpp)
     target_compile_definitions(test_rope_tail PRIVATE LUCE_ROPE_TEST_GPU=1)
     target_link_libraries(test_rope_tail PRIVATE ggml ggml-base ggml-cpu ${LUCE_GGML_BACKEND_TARGET})
@@ -1683,6 +1755,27 @@ if(LUCE_TESTS)
             target_link_libraries(test_ggml_rmsnorm_batch PRIVATE CUDA::cudart)
         endif()
     endif()
+    # Cluster unit tests (maikzz32/lucebox-halo-cluster): protocol
+    # round-trips, placement invariants, decision hooks over a socketpair,
+    # control-channel handshake. GPU-free at run time.
+    if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/test/unit/test_cluster_unit.cpp")
+        add_executable(test_cluster_unit test/unit/test_cluster_unit.cpp)
+        target_include_directories(test_cluster_unit PRIVATE ${LUCE_SRC_INCLUDE_DIRS})
+        target_link_libraries(test_cluster_unit PRIVATE luce_common ggml ${LUCE_GGML_BACKEND_TARGET} ggml-base)
+        if(NOT WIN32)
+            target_link_libraries(test_cluster_unit PRIVATE pthread)
+        else()
+            target_link_libraries(test_cluster_unit PRIVATE ws2_32)
+        endif()
+        if(LUCE_GPU_BACKEND STREQUAL "hip")
+            target_compile_definitions(test_cluster_unit PRIVATE LUCE_BACKEND_HIP=1 GGML_USE_HIP)
+            target_link_libraries(test_cluster_unit PRIVATE hip::host)
+        else()
+            target_compile_definitions(test_cluster_unit PRIVATE LUCE_BACKEND_CUDA=1)
+            find_package(CUDAToolkit REQUIRED)
+            target_link_libraries(test_cluster_unit PRIVATE CUDA::cudart)
+        endif()
+    endif()
     if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/test/unit/test_ggml_cluster_allreduce.cpp")
         add_executable(test_ggml_cluster_allreduce test/unit/test_ggml_cluster_allreduce.cpp)
         target_include_directories(test_ggml_cluster_allreduce PRIVATE ${LUCE_SRC_INCLUDE_DIRS})
@@ -2586,6 +2679,7 @@ if(LUCE_TESTS)
         test_ds4_dmix_entry_validation
         test_ggml_rmsnorm_batch
         test_ggml_cluster_allreduce
+        test_cluster_unit
         test_qwen35_tensor_parallel
         test_deepseek4_hc_cuda
         test_capture_safe_copies

server/deps/llama.cpp/ggml/include/ggml-backend.h

@@ -390,6 +390,13 @@ extern "C" {
     GGML_API void                 ggml_backend_sched_set_batch_split_copies(
                                             ggml_backend_sched_t sched,
                                             bool enabled);
+    // Split-input copies get storage of their own (never shared with another
+    // tensor of the graph), so filling one cannot overwrite data a split on
+    // its backend still reads: the scheduler then skips the blocking wait for
+    // the destination backend before each split's copies, and the host keeps
+    // queueing while the devices run. Must be set before allocation.
+    GGML_API void                 ggml_backend_sched_set_dedicated_copies(
+            ggml_backend_sched_t sched, bool enabled);
 
     //
     // Meta backend

server/deps/llama.cpp/ggml/include/ggml-cuda.h

@@ -162,6 +162,13 @@ GGML_BACKEND_API int ggml_backend_cuda_set_mmvq_max_ncols_override(int max_ncols
 // Returns the previous setting.
 GGML_BACKEND_API bool ggml_backend_cuda_set_mmvq_batch_invariant(bool enabled);
 
+// A MUL_MAT node whose src0 is a column range of wider Q8_0 rows (a view with
+// the rows' stride, e.g. one rank's slice of a contraction): on the RDNA4 Q8_0
+// matvec every block keeps the lane it takes in the whole rows, so the product
+// is bit-identical to the whole rows times src1 zero-padded outside the range,
+// at the range's reads. Other kernels compute the range's product as usual.
+GGML_BACKEND_API void ggml_backend_cuda_mul_mat_whole_row_lanes(struct ggml_tensor * mul_mat);
+
 // Calling-thread DS4 mixed-expert dispatch ceiling, scoped to a graph compute.
 // Accepts 0 (the default of five) or 1..16; returns the previous ceiling.
 GGML_BACKEND_API int ggml_backend_cuda_set_ds4_mix_mmv_max_tokens_override(int max_tokens);

server/deps/llama.cpp/ggml/include/ggml.h

@@ -1880,6 +1880,17 @@ extern "C" {
             struct ggml_tensor  * sinks,
             float                 scale);
 
+    // ggml_soft_max_ext_sink_col with an additive mask (scale, then the mask,
+    // rounded as separate scale and add launches round them); a is
+    // [n_cols, rows, ...] and mask [n_cols, >= rows, ...] broadcast over
+    // dims 2 and 3.
+    GGML_API struct ggml_tensor * ggml_soft_max_ext_sink_col_mask(
+            struct ggml_context * ctx,
+            struct ggml_tensor  * a,
+            struct ggml_tensor  * mask,
+            struct ggml_tensor  * sinks,
+            float                 scale);
+
     GGML_API void ggml_soft_max_add_sinks(
             struct ggml_tensor * a,
             struct ggml_tensor * sinks);
@@ -2568,6 +2579,21 @@ extern "C" {
             struct ggml_tensor * a,
             struct ggml_tensor * positions);
 
+    // Run several single-token query lanes of one raw ring in one op, as if
+    // each lane wrote its own ring row and then attended before the next lane
+    // wrote: query token t reads lane_kv row j in place of ring row
+    // ring_rows[j] for every j <= t, and the ring's own contents elsewhere.
+    // ring_rows is I32 [n_query] (row indices into the raw segment), lane_kv
+    // has K's type and is contiguous [D, n_query]. Every other row, the mask
+    // and the selection keep their per-token meaning, so each token's result
+    // equals the single-token op over its own ring state bit for bit. Only the
+    // D=512 split-KV decode kernel consumes it (segmented K/V, indexed mask,
+    // n_query <= 8, no fused RoPE: ring_rows takes the RoPE positions slot).
+    GGML_API void ggml_flash_attn_ext_set_ds4_lane_rows(
+            struct ggml_tensor * a,
+            struct ggml_tensor * ring_rows,
+            struct ggml_tensor * lane_kv);
+
     // True when flash_attn_ext carries the DS4 sparse-layout or fused-RoPE
     // contract. Backends must implement that complete contract or reject it.
     GGML_API bool ggml_flash_attn_ext_is_ds4(
@@ -2685,6 +2711,7 @@ extern "C" {
         GGML_MOE_FUSED_HOST_POST          = -8,
         GGML_MOE_FUSED_HOST_WAIT          = -9,
         GGML_MOE_FUSED_PROTECTED_ROUTES   = -10,
+        GGML_MOE_FUSED_COMBINE_MASKED     = -11,
     };
 
     // Word offsets in ggml_tensor::op_params for the host mailbox pair. All
@@ -2746,11 +2773,34 @@ extern "C" {
             ggml_cluster_allreduce_fn fn,
             void                    * user);
 
+    // The same collective in place: the result is a view of `a`, so the sum
+    // overwrites `a` and no device copy runs ahead of the collective. Only
+    // for an `a` that no other node reads.
+    GGML_API struct ggml_tensor * ggml_cluster_allreduce_inplace(
+            struct ggml_context     * ctx,
+            struct ggml_tensor      * a,
+            ggml_cluster_allreduce_fn fn,
+            void                    * user);
+
     GGML_API struct ggml_tensor * ggml_laguna_moe_combine(
             struct ggml_context * ctx,
             struct ggml_tensor  * experts,
             struct ggml_tensor  * expert_weights);
 
+    // One owner's routed-expert sum: out[:, t] = sum_e experts[:, e, t] *
+    // (weights[e, t] * valid[0, ids[e, t], t]), the masked weight, product and
+    // sum rounded as mul(weights, get_rows(valid, ids)), mul and repeat_back
+    // round them (routes with a zero masked weight add nothing).
+    // experts [n_embd, n_used, T], weights [n_used, T] F32, valid [1, n_expert, T]
+    // F32, ids [n_used, T] I32 (global expert ids; a route masked to -1 adds
+    // nothing).
+    GGML_API struct ggml_tensor * ggml_moe_combine_masked(
+            struct ggml_context * ctx,
+            struct ggml_tensor  * experts,
+            struct ggml_tensor  * weights,
+            struct ggml_tensor  * valid,
+            struct ggml_tensor  * ids);
+
     // Coarse DeepSeek-V4 routed-owner op. gate_up contains concatenated gate
     // and up output rows; the backend performs fused gate/up MMVQ + clamped
     // SwiGLU, down MMVQ, route weighting, and local reduction.
@@ -2917,6 +2967,23 @@ extern "C" {
     // ggml_ds4_hc_collapse: dst[d, t] = sum_h hc[d + h*n_embd, t] * pre[h, t],
     // rounded as mul, permute and sum_rows round it.
     // hc: F32 [n_embd*n_hc, T] (nb0 = 4), pre: F32 [n_hc, T] (nb0 = 4) -> F32 [n_embd, T].
+    // ggml_ds4_argmax_pair: the argmax of each column of logits [n_rows, T]
+    // (rows row_offset.. of a vocabulary split across n_slots ranks) as one
+    // slot of an F32 [8 * n_slots, T] exchange row: the value's 32 bits and the
+    // global index as eight byte-valued floats (exact through a bf16 sum),
+    // zeros in the other slots. ggml_ds4_argmax_pick reads the summed rows
+    // back: the largest value, ties to the lowest index (I32 [T]).
+    GGML_API struct ggml_tensor * ggml_ds4_argmax_pair(
+            struct ggml_context * ctx,
+            struct ggml_tensor  * logits,
+            int                   row_offset,
+            int                   slot,
+            int                   n_slots);
+    GGML_API struct ggml_tensor * ggml_ds4_argmax_pick(
+            struct ggml_context * ctx,
+            struct ggml_tensor  * pairs,
+            int                   n_slots);
+
     GGML_API struct ggml_tensor * ggml_ds4_hc_collapse(
             struct ggml_context * ctx,
             struct ggml_tensor  * hc,

server/deps/llama.cpp/ggml/src/ggml-backend.cpp

@@ -883,6 +883,7 @@ struct ggml_backend_sched {
     int deferred_peer_copies_capacity;
     bool split_deferred_peer_copies;
     bool batch_split_copies;
+    bool dedicated_copies;
     ggml_backend_buffer_t batch_staging_buffers[GGML_SCHED_MAX_BACKENDS];
     uint8_t * batch_staging_bases[GGML_SCHED_MAX_BACKENDS];
     size_t batch_staging_sizes[GGML_SCHED_MAX_BACKENDS];
@@ -1522,7 +1523,7 @@ void ggml_backend_sched_split_graph(ggml_backend_sched_t sched, struct ggml_cgra
                         for (int c = 0; c < sched->n_copies; c++) {
                             struct ggml_tensor * tensor_copy = ggml_dup_tensor_layout(sched->ctx, src);
                             ggml_format_name(tensor_copy, "%s#%s#%d", ggml_backend_name(backend), src->name, c);
-                            if (sched->n_copies > 1 || sched->copy_events) {
+                            if (sched->n_copies > 1 || sched->copy_events || sched->dedicated_copies) {
                                 ggml_set_input(tensor_copy);
                                 ggml_set_output(tensor_copy); // prevent ggml-alloc from overwriting the tensor
                             }
@@ -2370,7 +2371,11 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s
                 ggml_backend_tensor_copy(input, input_cpy);
             } else {
                 // wait for the split backend to finish using the input before overwriting it
-                copy_state.wait_for_destination_generation();
+                // (a dedicated copy is not shared with anything the backend still reads,
+                // and the previous evaluation ended with every backend synchronized)
+                if (!sched->dedicated_copies) {
+                    copy_state.wait_for_destination_generation();
+                }
 
                 // when offloading MoE weights, we can reduce the amount of data copied by copying only the experts that are used
                 ggml_tensor * node = split->graph.nodes[0];
@@ -2970,6 +2975,13 @@ void ggml_backend_sched_set_batch_split_copies(
     sched->batch_split_copies = enabled;
 }
 
+void ggml_backend_sched_set_dedicated_copies(
+        ggml_backend_sched_t sched, bool enabled) {
+    GGML_ASSERT(sched);
+    GGML_ASSERT(!sched->is_alloc);
+    sched->dedicated_copies = enabled;
+}
+
 int ggml_backend_sched_get_n_splits(ggml_backend_sched_t sched) {
     GGML_ASSERT(sched);
     return sched->n_splits;

server/deps/llama.cpp/ggml/src/ggml-cpu/ops.cpp

@@ -5362,11 +5362,12 @@ static void ggml_compute_forward_soft_max_f32(
 
     // sinks
     const float * sk = src2 ? (float *)((char *) src2->data) : nullptr;
-    // Lucebox sink-as-virtual-column mode (see ggml_soft_max_ext_sink_col)
+    // Lucebox sink-as-virtual-column mode (see ggml_soft_max_ext_sink_col and
+    // its _mask form: the mask is added below like any other).
     int32_t sink_col = 0;
     memcpy(&sink_col, (int32_t *) dst->op_params + 2, sizeof(int32_t));
     if (sink_col) {
-        GGML_ASSERT(sk && !src1);
+        GGML_ASSERT(sk);
     }
 
     for (int64_t i03 = 0; i03 < ne03; i03++) {

server/deps/llama.cpp/ggml/src/ggml-cuda/common.cuh

@@ -2,6 +2,9 @@
 
 #include "ggml.h"
 #include "ggml-impl.h"
+
+// MUL_MAT op_params[1]: see ggml_backend_cuda_mul_mat_whole_row_lanes.
+#define GGML_CUDA_WHOLE_ROW_LANES 0x574C524B
 #include "ggml-cuda.h"
 
 #include <cerrno>
@@ -1492,6 +1495,13 @@ struct ggml_backend_cuda_context {
     };
     std::vector<luce_q8_memo_entry> luce_q8_memo;
 
+    // DS4 HC boundary launches (ds4-hc.cu) synchronize their blocks on this
+    // device barrier: an arrival count and a generation.
+    unsigned int * ds4_hc_barrier = nullptr;
+    // The split boundary's mix rows ([64][32] floats) and per-token block
+    // counts (64 uints), allocated with the barrier.
+    void * ds4_hc_split_scratch = nullptr;
+
     cudaStream_t streams[GGML_CUDA_MAX_DEVICES][GGML_CUDA_MAX_STREAMS] = { { nullptr } };
     cublasHandle_t cublas_handles[GGML_CUDA_MAX_DEVICES] = {nullptr};