feat: support Apple RDMA in layer parallel distributed mode

Makefile

@@ -24,8 +24,12 @@ DS4_DSPARK_SUPPORT ?= gguf/DeepSeek-V4-Flash-DSpark-support-0731.gguf
 
 ifeq ($(UNAME_S),Darwin)
 METAL_LDLIBS := $(LDLIBS) -framework Foundation -framework Metal
-CORE_OBJS = ds4.o ds4_distributed.o ds4_tp.o ds4_ssd.o ds4_metal.o ds4_layer_pack.o
-CPU_CORE_OBJS = ds4_cpu.o ds4_distributed.o ds4_tp.o ds4_ssd.o ds4_layer_pack.o
+# RDMA-over-Thunderbolt (ds4_tp.o and ds4_distributed_rdma.o) resolves librdma at
+# runtime via dlopen (ds4_rdma_verbs.o), so there is no -lrdma link step and no
+# build flag to set: the code compiles wherever <infiniband/verbs.h> exists and
+# falls back to TCP where the provider is absent. See Apple TN3205.
+CORE_OBJS = ds4.o ds4_distributed.o ds4_distributed_rdma.o ds4_rdma_verbs.o ds4_tp.o ds4_ssd.o ds4_metal.o ds4_layer_pack.o
+CPU_CORE_OBJS = ds4_cpu.o ds4_distributed_cpu.o ds4_distributed_rdma.o ds4_rdma_verbs.o ds4_tp.o ds4_ssd.o ds4_layer_pack.o
 else
 CFLAGS += -D_GNU_SOURCE -fno-finite-math-only
 CUDA_HOME ?= $(shell if [ -x /usr/local/cuda/bin/nvcc ]; then \
@@ -50,8 +54,10 @@ NVCCFLAGS ?= -O3 -g -lineinfo --use_fast_math $(NVCC_ARCH_FLAGS) -Xcompiler $(NA
 # Vendored llama.cpp mmq prefill tier (cuda/mmq/, see cuda/mmq/VENDOR.md).
 MMQ_INCLUDES := -Icuda/mmq
 MMQ_OBJS := cuda/mmq/ds4_ggml_stubs.o cuda/mmq/ds4_mmq.o cuda/mmq/ds4_mmq_d2r.o cuda/mmq/quantize.o cuda/mmq/mmid.o cuda/mmq/mmvq.o cuda/mmq/ds4_repack.o
-CORE_OBJS = ds4.o ds4_distributed.o ds4_tp.o ds4_ssd.o ds4_cuda.o ds4_layer_pack.o $(MMQ_OBJS)
-CPU_CORE_OBJS = ds4_cpu.o ds4_distributed.o ds4_tp.o ds4_ssd.o ds4_layer_pack.o
+# ds4_distributed_rdma.o / ds4_rdma_verbs.o compile to TCP-only stubs off Apple
+# (no <infiniband/verbs.h>), so the distributed layer links the same everywhere.
+CORE_OBJS = ds4.o ds4_distributed.o ds4_distributed_rdma.o ds4_rdma_verbs.o ds4_tp.o ds4_ssd.o ds4_cuda.o ds4_layer_pack.o $(MMQ_OBJS)
+CPU_CORE_OBJS = ds4_cpu.o ds4_distributed_cpu.o ds4_distributed_rdma.o ds4_rdma_verbs.o ds4_tp.o ds4_ssd.o ds4_layer_pack.o
 CUDA_LDLIBS ?= -lm -Xcompiler -pthread -L$(CUDA_HOME)/targets/sbsa-linux/lib -L$(CUDA_HOME)/lib64 -lcudart -lcublas
 HIPCC ?= $(shell command -v hipcc 2>/dev/null || echo /opt/rocm/bin/hipcc)
 ROCM_ARCH ?= gfx1151
@@ -178,7 +184,7 @@ cuda:
 
 strix-halo:
 	$(MAKE) -B ds4 ds4-server ds4-bench ds4-eval ds4-agent \
-		CORE_OBJS="ds4.o ds4_distributed.o ds4_tp.o ds4_ssd.o ds4_rocm.o ds4_rocm_compat.o ds4_rocm_unavailable.o ds4_layer_pack.o" \
+		CORE_OBJS="ds4.o ds4_distributed.o ds4_distributed_rdma.o ds4_rdma_verbs.o ds4_tp.o ds4_ssd.o ds4_rocm.o ds4_rocm_compat.o ds4_rocm_unavailable.o ds4_layer_pack.o" \
 		CFLAGS="$(CFLAGS) -DDS4_ROCM_BUILD" \
 		DS4_LINK="$(HIPCC) $(ROCM_CFLAGS)" \
 		DS4_LINK_LIBS="$(ROCM_LDLIBS)"
@@ -229,15 +235,24 @@ ds4.o: ds4.c ds4.h ds4_ssd.h ds4_distributed.h ds4_gpu.h
 ds4_ssd.o: ds4_ssd.c ds4_ssd.h
 	$(CC) $(CFLAGS) -c -o $@ ds4_ssd.c
 
-ds4_cli.o: ds4_cli.c ds4.h ds4_ssd.h ds4_distributed.h ds4_help.h linenoise.h
+ds4_cli.o: ds4_cli.c ds4.h ds4_ssd.h ds4_distributed.h ds4_distributed_rdma.h ds4_help.h linenoise.h
 	$(CC) $(CFLAGS) -c -o $@ ds4_cli.c
 
-ds4_distributed.o: ds4_distributed.c ds4_distributed.h ds4.h ds4_ssd.h
+ds4_distributed.o: ds4_distributed.c ds4_distributed.h ds4_distributed_rdma.h ds4_dist_dchan.h ds4.h ds4_ssd.h
 	$(CC) $(CFLAGS) -c -o $@ ds4_distributed.c
 
-ds4_tp.o: ds4_tp.c ds4_tp.h ds4.h ds4_ssd.h
+ds4_distributed_cpu.o: ds4_distributed.c ds4_distributed.h ds4_distributed_rdma.h ds4_dist_dchan.h ds4.h ds4_ssd.h
+	$(CC) $(CFLAGS) -DDS4_NO_GPU -c -o $@ ds4_distributed.c
+
+ds4_tp.o: ds4_tp.c ds4_tp.h ds4.h ds4_ssd.h ds4_rdma_verbs.h
 	$(CC) $(CFLAGS) -c -o $@ ds4_tp.c
 
+ds4_distributed_rdma.o: ds4_distributed_rdma.c ds4_distributed_rdma.h ds4_dist_dchan.h ds4_rdma_verbs.h
+	$(CC) $(CFLAGS) -c -o $@ ds4_distributed_rdma.c
+
+ds4_rdma_verbs.o: ds4_rdma_verbs.c ds4_rdma_verbs.h
+	$(CC) $(CFLAGS) -c -o $@ ds4_rdma_verbs.c
+
 ds4_help.o: ds4_help.c ds4_help.h
 	$(CC) $(CFLAGS) -c -o $@ ds4_help.c
 
@@ -358,7 +373,7 @@ ds4_cpu_test_hooks.o: ds4.c ds4.h ds4_gpu.h ds4_gpu_mgpu.h ds4_layer_pack.h
 tests/test_engine_mgpu_placement.o: tests/test_engine_mgpu_placement.c ds4.h ds4_gpu_mgpu.h ds4_layer_pack.h
 	$(CC) $(CFLAGS) -I. -c -o $@ $<
 
-tests/test_engine_mgpu_placement: tests/test_engine_mgpu_placement.o ds4_cpu_test_hooks.o ds4_distributed.o ds4_tp.o ds4_ssd.o ds4_layer_pack.o
+tests/test_engine_mgpu_placement: tests/test_engine_mgpu_placement.o ds4_cpu_test_hooks.o ds4_distributed_cpu.o ds4_distributed_rdma.o ds4_rdma_verbs.o ds4_tp.o ds4_ssd.o ds4_layer_pack.o
 	$(CC) $(CFLAGS) -o $@ $^ $(LDLIBS)
 
 ifneq ($(UNAME_S),Darwin)

README.md

@@ -576,6 +576,15 @@ the output for your use case. However experimentally reduction activation
 size didn't provide a significant improvement, so this option may be removed
 in the future.
 
+On Apple Metal, setting `DS4_METAL_FAST_SYNC=1` on every node also enables the
+experimental pipeline-parallel activation fence for one-token, 32-bit decode
+handoffs. Workers advertise support during route formation; unsupported paths
+(prefill, reduced-width activations, GLM, or disabled worker prefetch) retain
+the synchronous copy path. The worker pre-arms a system-coherent Metal wait,
+receives into a reusable shared staging slot, and releases the GPU after the
+activation is complete. `DS4_PP_FENCE_MAX_ITERS` bounds the GPU wait and is a
+diagnostic/rollback knob rather than a normal tuning parameter.
+
 **If a worker disconnects, the coordinator removes that worker from the active
 route**. The request already in flight can fail, and later calls report an
 incomplete route until a compatible worker reconnects and sends a new

ds4.c

@@ -428,13 +428,31 @@ static bool ds4_backend_supports_glm_streaming_full_layers(ds4_backend backend)
     return false;
 }
 
+/* getenv() is a locked linear scan of the environment. This is called from the
+ * per-layer graph path, so it showed up as 0.15 ms/token in __findenv_locked in
+ * a symbolicated profile. The environment does not change under us, so memoize
+ * on the name pointer - callers pass literals. */
 static bool glm_graph_env_present(const char *rocm_name, const char *metal_name) {
+    enum { GLM_ENV_MEMO = 64 };
+    static struct { const char *r, *m; bool present; } memo[GLM_ENV_MEMO];
+    static uint32_t memo_len;
+    for (uint32_t i = 0; i < memo_len; i++) {
+        if (memo[i].r == rocm_name && memo[i].m == metal_name) return memo[i].present;
+    }
+    bool present = false;
 #ifdef DS4_ROCM_BUILD
-    if (rocm_name && getenv(rocm_name) != NULL) return true;
+    if (rocm_name && getenv(rocm_name) != NULL) present = true;
 #else
     (void)rocm_name;
 #endif
-    return metal_name && getenv(metal_name) != NULL;
+    if (!present) present = metal_name && getenv(metal_name) != NULL;
+    if (memo_len < GLM_ENV_MEMO) {
+        memo[memo_len].r = rocm_name;
+        memo[memo_len].m = metal_name;
+        memo[memo_len].present = present;
+        memo_len++;
+    }
+    return present;
 }
 
 static const char *glm_graph_env_value(const char *rocm_name,
@@ -51123,6 +51141,12 @@ static void session_greedy_splitkv_reset(ds4_session *s) {
 }
 #endif
 
+int ds4_session_kv_snapshot_stable(const ds4_session *s) {
+    if (!s) return 0;
+    if (s->distributed) return ds4_dist_session_kv_snapshot_stable(s->distributed);
+    return 1;
+}
+
 uint64_t ds4_session_payload_bytes(ds4_session *s) {
     if (!s || !s->checkpoint_valid) return 0;
     if (s->distributed) return 0;
@@ -59217,18 +59241,19 @@ static DS4_MAYBE_UNUSED void ds4_session_slice_commit_timeline(ds4_session *s, c
     ds4_session_dspark_capture_note_checkpoint(s);
 }
 
-int ds4_session_eval_layer_slice(ds4_session *s,
-                                 const int *tokens,
-                                 uint32_t n_tokens,
-                                 uint32_t pos0,
-                                 uint32_t layer_start,
-                                 uint32_t layer_end,
-                                 const float *input_hc,
-                                 float *output_hc,
-                                 bool output_logits,
-                                 float *logits,
-                                 char *err,
-                                 size_t errlen) {
+static int ds4_session_eval_layer_slice_impl(ds4_session *s,
+                                             const int *tokens,
+                                             uint32_t n_tokens,
+                                             uint32_t pos0,
+                                             uint32_t layer_start,
+                                             uint32_t layer_end,
+                                             const float *input_hc,
+                                             const ds4_pp_staged_input *staged_input,
+                                             float *output_hc,
+                                             bool output_logits,
+                                             float *logits,
+                                             char *err,
+                                             size_t errlen) {
     if (!s || !s->engine) {
         if (errlen) snprintf(err, errlen, "missing layer-slice session");
         return 1;
@@ -59241,11 +59266,15 @@ int ds4_session_eval_layer_slice(ds4_session *s,
                              layer_start, layer_end);
         return 1;
     }
-    if (layer_start != 0 && !input_hc) {
+    if (layer_start != 0 && !input_hc && !staged_input) {
         if (errlen) snprintf(err, errlen, "layer-slice layer %u requires input hidden-state",
                              layer_start);
         return 1;
     }
+    if (input_hc && staged_input) {
+        if (errlen) snprintf(err, errlen, "layer-slice input hidden-state is ambiguous");
+        return 1;
+    }
     if (output_logits && layer_end + 1u != executable_layers) {
         if (errlen) snprintf(err, errlen, "layer-slice logits require final transformer layer");
         return 1;
@@ -59259,7 +59288,7 @@ int ds4_session_eval_layer_slice(ds4_session *s,
                              layer_start, layer_end);
         return 1;
     }
-    if (!input_hc && !s->engine->weights.token_embd) {
+    if (!input_hc && !staged_input && !s->engine->weights.token_embd) {
         if (errlen) snprintf(err, errlen, "token embedding is not loaded");
         return 1;
     }
@@ -59284,6 +59313,11 @@ int ds4_session_eval_layer_slice(ds4_session *s,
     return 1;
 #else
     if (ds4_session_is_glm(s)) {
+        if (staged_input) {
+            if (errlen) snprintf(err, errlen,
+                                 "GLM layer slices do not support staged PP input");
+            return 1;
+        }
         ds4_engine *e = s->engine;
         ds4_glm_gpu_graph *g = &s->glm_graph;
         if (!s->glm_graph_ready) {
@@ -59497,6 +59531,13 @@ int ds4_session_eval_layer_slice(ds4_session *s,
 
     const uint64_t hc_dim = (uint64_t)DS4_N_HC * DS4_N_EMBD;
     const uint64_t hc_bytes = (uint64_t)n_tokens * hc_dim * sizeof(float);
+    if (staged_input &&
+        (!staged_input->tensor || staged_input->ready_value == 0 ||
+         n_tokens != 1 || pos0 == 0 || layer_start == 0 ||
+         staged_input->bytes != hc_dim * sizeof(float))) {
+        if (errlen) snprintf(err, errlen, "invalid staged PP layer-slice input");
+        return 1;
+    }
     if (n_tokens == 1 && pos0 > 0) {
         if (g->raw_cap == 0) {
             if (errlen) snprintf(err, errlen, "%s layer-slice decode has no raw KV cache",
@@ -59506,15 +59547,31 @@ int ds4_session_eval_layer_slice(ds4_session *s,
         }
 
         bool ok = true;
-        if (g->ssd_streaming && !input_hc) {
+        const bool has_remote_input = input_hc != NULL || staged_input != NULL;
+        if (g->ssd_streaming && !has_remote_input) {
             g->streaming_static_decode_map_current = false;
             ok = metal_graph_stream_map_token(&e->model, &e->weights);
         }
         if (input_hc) {
             ok = ds4_gpu_tensor_write(metal_graph_cur_hc(g), 0, input_hc, hc_dim * sizeof(float)) != 0;
         }
         if (ok) ok = ds4_gpu_begin_commands() != 0;
-        if (ok && !input_hc) {
+        if (ok && staged_input) {
+#if defined(__APPLE__)
+            ok = ds4_gpu_pp_fence_wait_copy(metal_graph_cur_hc(g),
+                                             staged_input->tensor,
+                                             staged_input->sync_tensor,
+                                             staged_input->bytes,
+                                             staged_input->ready_offset,
+                                             staged_input->timeout_offset,
+                                             staged_input->ready_value) != 0;
+            if (ok) ok = ds4_gpu_commit_commands_async() != 0;
+            if (ok) ok = ds4_gpu_begin_commands() != 0;
+#else
+            ok = false;
+#endif
+        }
+        if (ok && !has_remote_input) {
             ok = ds4_gpu_embed_token_hc_tensor(metal_graph_cur_hc(g),
                                                e->model.map,
                                                e->model.size,
@@ -59614,6 +59671,12 @@ int ds4_session_eval_layer_slice(ds4_session *s,
         .cap = (int)n_tokens,
     };
 
+    if (staged_input) {
+        if (errlen) snprintf(err, errlen,
+                             "staged PP input is decode-only");
+        return 1;
+    }
+
     bool ok = true;
     if (g->ssd_streaming && !input_hc) {
         g->streaming_static_decode_map_current = false;
@@ -59715,6 +59778,60 @@ int ds4_session_eval_layer_slice(ds4_session *s,
 #endif
 }
 
+int ds4_session_eval_layer_slice(ds4_session *s,
+                                 const int *tokens,
+                                 uint32_t n_tokens,
+                                 uint32_t pos0,
+                                 uint32_t layer_start,
+                                 uint32_t layer_end,
+                                 const float *input_hc,
+                                 float *output_hc,
+                                 bool output_logits,
+                                 float *logits,
+                                 char *err,
+                                 size_t errlen) {
+    return ds4_session_eval_layer_slice_impl(s,
+                                             tokens,
+                                             n_tokens,
+                                             pos0,
+                                             layer_start,
+                                             layer_end,
+                                             input_hc,
+                                             NULL,
+                                             output_hc,
+                                             output_logits,
+                                             logits,
+                                             err,
+                                             errlen);
+}
+
+int ds4_session_eval_layer_slice_staged(ds4_session *s,
+                                        const int *tokens,
+                                        uint32_t n_tokens,
+                                        uint32_t pos0,
+                                        uint32_t layer_start,
+                                        uint32_t layer_end,
+                                        const ds4_pp_staged_input *staged_input,
+                                        float *output_hc,
+                                        bool output_logits,
+                                        float *logits,
+                                        char *err,
+                                        size_t errlen) {
+    return ds4_session_eval_layer_slice_impl(s,
+                                             tokens,
+                                             n_tokens,
+                                             pos0,
+                                             layer_start,
+                                             layer_end,
+                                             NULL,
+                                             staged_input,
+                                             output_hc,
+                                             output_logits,
+                                             logits,
+                                             err,
+                                             errlen);
+}
+
 #ifndef DS4_NO_GPU
 typedef struct {
     ds4_session *session;