diff --git a/server/deps/llama.cpp/ggml/include/ggml-alloc.h b/server/deps/llama.cpp/ggml/include/ggml-alloc.h index a7926a21a..0a8979139 100644 --- a/server/deps/llama.cpp/ggml/include/ggml-alloc.h +++ b/server/deps/llama.cpp/ggml/include/ggml-alloc.h @@ -46,6 +46,14 @@ GGML_API enum ggml_status ggml_tallocr_alloc(struct ggml_tallocr * talloc, st typedef struct ggml_gallocr * ggml_gallocr_t; GGML_API ggml_gallocr_t ggml_gallocr_new(ggml_backend_buffer_type_t buft); +// Uses max_chunk_size as the preferred backing-allocation limit while +// preserving a single logical graph allocator. Individual tensors are never +// split and may exceed the limit. This is useful on devices without virtual +// memory support, where a large contiguous allocation can fail despite +// sufficient aggregate free memory. +GGML_API ggml_gallocr_t ggml_gallocr_new_with_max_chunk_size( + ggml_backend_buffer_type_t buft, + size_t max_chunk_size); GGML_API ggml_gallocr_t ggml_gallocr_new_n(ggml_backend_buffer_type_t * bufts, int n_bufs); GGML_API void ggml_gallocr_free(ggml_gallocr_t galloc); @@ -72,6 +80,7 @@ GGML_API bool ggml_gallocr_reserve_n( GGML_API bool ggml_gallocr_alloc_graph(ggml_gallocr_t galloc, struct ggml_cgraph * graph); GGML_API size_t ggml_gallocr_get_buffer_size(ggml_gallocr_t galloc, int buffer_id); +GGML_API int ggml_gallocr_get_buffer_n_chunks(ggml_gallocr_t galloc, int buffer_id); // Utils // Create a buffer and allocate all the tensors in a ggml_context diff --git a/server/deps/llama.cpp/ggml/include/ggml.h b/server/deps/llama.cpp/ggml/include/ggml.h index dba76defe..a09f132e6 100644 --- a/server/deps/llama.cpp/ggml/include/ggml.h +++ b/server/deps/llama.cpp/ggml/include/ggml.h @@ -2589,7 +2589,8 @@ extern "C" { // Exact mode-1 variant for heterogeneous MoE verification. It computes // block_out[d] = peer_block[d] + main_block[d] inside the HC-post kernel, - // eliminating the standalone reduction and tokenwise CONT copies. + // eliminating the standalone reduction and tokenwise CONT copies. All + // non-base tensors may carry the same n_tokens second dimension. GGML_API struct ggml_tensor * ggml_ds4_hc_post_split( struct ggml_context * ctx, struct ggml_tensor * residual_hc, diff --git a/server/deps/llama.cpp/ggml/src/ggml-alloc.c b/server/deps/llama.cpp/ggml/src/ggml-alloc.c index a4b01ccf8..ce4290469 100644 --- a/server/deps/llama.cpp/ggml/src/ggml-alloc.c +++ b/server/deps/llama.cpp/ggml/src/ggml-alloc.c @@ -494,7 +494,10 @@ struct ggml_gallocr { int n_leafs; }; -ggml_gallocr_t ggml_gallocr_new_n(ggml_backend_buffer_type_t * bufts, int n_bufs) { +static ggml_gallocr_t ggml_gallocr_new_n_impl( + ggml_backend_buffer_type_t * bufts, + int n_bufs, + size_t max_chunk_size) { ggml_gallocr_t galloc = (ggml_gallocr_t)calloc(1, sizeof(struct ggml_gallocr)); GGML_ASSERT(galloc != NULL); @@ -522,6 +525,9 @@ ggml_gallocr_t ggml_gallocr_new_n(ggml_backend_buffer_type_t * bufts, int n_bufs if (galloc->buf_tallocs[i] == NULL) { size_t alignment = ggml_backend_buft_get_alignment(bufts[i]); size_t max_size = ggml_backend_buft_get_max_size(bufts[i]); + if (max_chunk_size > 0) { + max_size = MIN(max_size, max_chunk_size); + } galloc->buf_tallocs[i] = ggml_dyn_tallocr_new(alignment, max_size); } } @@ -530,10 +536,21 @@ ggml_gallocr_t ggml_gallocr_new_n(ggml_backend_buffer_type_t * bufts, int n_bufs return galloc; } +ggml_gallocr_t ggml_gallocr_new_n(ggml_backend_buffer_type_t * bufts, int n_bufs) { + return ggml_gallocr_new_n_impl(bufts, n_bufs, 0); +} + ggml_gallocr_t ggml_gallocr_new(ggml_backend_buffer_type_t buft) { return ggml_gallocr_new_n(&buft, 1); } +ggml_gallocr_t ggml_gallocr_new_with_max_chunk_size( + ggml_backend_buffer_type_t buft, + size_t max_chunk_size) { + GGML_ASSERT(max_chunk_size > 0); + return ggml_gallocr_new_n_impl(&buft, 1, max_chunk_size); +} + void ggml_gallocr_free(ggml_gallocr_t galloc) { if (galloc == NULL) { return; @@ -1114,6 +1131,21 @@ size_t ggml_gallocr_get_buffer_size(ggml_gallocr_t galloc, int buffer_id) { return ggml_vbuffer_size(galloc->buffers[buffer_id]); } +int ggml_gallocr_get_buffer_n_chunks(ggml_gallocr_t galloc, int buffer_id) { + GGML_ASSERT(buffer_id >= 0 && buffer_id < galloc->n_buffers); + + if (galloc->buffers[buffer_id] == NULL) { + return 0; + } + + int n_chunks = 0; + while (n_chunks < GGML_VBUFFER_MAX_CHUNKS && + galloc->buffers[buffer_id]->chunks[n_chunks] != NULL) { + ++n_chunks; + } + return n_chunks; +} + // utils static void free_buffers(ggml_backend_buffer_t ** buffers, const size_t * n_buffers) { diff --git a/server/deps/llama.cpp/ggml/src/ggml-cuda/ds4-hc.cu b/server/deps/llama.cpp/ggml/src/ggml-cuda/ds4-hc.cu index 2e8ebd61b..661c36320 100644 --- a/server/deps/llama.cpp/ggml/src/ggml-cuda/ds4-hc.cu +++ b/server/deps/llama.cpp/ggml/src/ggml-cuda/ds4-hc.cu @@ -397,7 +397,19 @@ static __global__ void ds4_hc_post_split_kernel( const float * __restrict__ split, float * __restrict__ dst, int n_embd, - int n_hc) { + int n_hc, + size_t residual_stride, + size_t main_block_stride, + size_t peer_block_stride, + size_t split_stride, + size_t dst_stride) { + const int token = (int) blockIdx.y; + residual += (size_t) token * residual_stride; + main_block += (size_t) token * main_block_stride; + peer_block += (size_t) token * peer_block_stride; + split += (size_t) token * split_stride; + dst += (size_t) token * dst_stride; + const int i = blockIdx.x * blockDim.x + threadIdx.x; const int total = n_embd * n_hc; if (i >= total) { @@ -530,10 +542,14 @@ void ggml_cuda_op_ds4_hc(ggml_backend_cuda_context & ctx, ggml_tensor * dst) { GGML_ASSERT(src3 && src3->type == GGML_TYPE_F32); const int total = n_embd * n_hc; const int blocks = (total + 255) / 256; - ds4_hc_post_split_kernel<<>>( + const dim3 grid(blocks, n_tokens, 1); + ds4_hc_post_split_kernel<<>>( (const float *) src0->data, (const float *) src1->data, (const float *) src3->data, (const float *) src2->data, - (float *) dst->data, n_embd, n_hc); + (float *) dst->data, n_embd, n_hc, + src0->nb[1] / sizeof(float), src1->nb[1] / sizeof(float), + src3->nb[1] / sizeof(float), src2->nb[1] / sizeof(float), + dst->nb[1] / sizeof(float)); } break; default: GGML_ABORT("ds4_hc: unknown mode"); diff --git a/server/deps/llama.cpp/ggml/src/ggml.c b/server/deps/llama.cpp/ggml/src/ggml.c index 4e3af6ed9..2db6c2bf7 100644 --- a/server/deps/llama.cpp/ggml/src/ggml.c +++ b/server/deps/llama.cpp/ggml/src/ggml.c @@ -8420,17 +8420,26 @@ struct ggml_tensor * ggml_ds4_hc_post_split( GGML_ASSERT(ggml_is_contiguous(residual_hc)); GGML_ASSERT(ggml_is_contiguous(main_block)); GGML_ASSERT(ggml_is_contiguous(peer_block)); - GGML_ASSERT(ggml_is_contiguous(split)); + GGML_ASSERT(split->nb[0] == sizeof(float)); + GGML_ASSERT(split->nb[1] >= split->ne[0] * split->nb[0]); + GGML_ASSERT(split->ne[2] == 1 && split->ne[3] == 1); GGML_ASSERT(n_hc > 0 && n_hc <= 8); + GGML_ASSERT(residual_hc->ne[2] == 1 && residual_hc->ne[3] == 1); + GGML_ASSERT(main_block->ne[2] == 1 && main_block->ne[3] == 1); + GGML_ASSERT(peer_block->ne[2] == 1 && peer_block->ne[3] == 1); const int64_t mix_dim = 2*(int64_t)n_hc + (int64_t)n_hc*n_hc; - GGML_ASSERT(ggml_nelements(split) == mix_dim); - GGML_ASSERT(ggml_nelements(residual_hc) % n_hc == 0); - const int64_t n_embd = ggml_nelements(residual_hc) / n_hc; - GGML_ASSERT(ggml_nelements(main_block) == n_embd); - GGML_ASSERT(ggml_nelements(peer_block) == n_embd); - - struct ggml_tensor * result = ggml_new_tensor_1d( - ctx, GGML_TYPE_F32, (int64_t) n_embd * n_hc); + const int64_t n_tokens = residual_hc->ne[1]; + GGML_ASSERT(n_tokens > 0); + GGML_ASSERT(split->ne[0] == mix_dim && split->ne[1] == n_tokens); + GGML_ASSERT(residual_hc->ne[0] % n_hc == 0); + const int64_t n_embd = residual_hc->ne[0] / n_hc; + GGML_ASSERT(main_block->ne[0] == n_embd && main_block->ne[1] == n_tokens); + GGML_ASSERT(peer_block->ne[0] == n_embd && peer_block->ne[1] == n_tokens); + + struct ggml_tensor * result = n_tokens == 1 + ? ggml_new_tensor_1d(ctx, GGML_TYPE_F32, (int64_t) n_embd * n_hc) + : ggml_new_tensor_2d( + ctx, GGML_TYPE_F32, (int64_t) n_embd * n_hc, n_tokens); result->op = GGML_OP_DS4_HC; result->src[0] = residual_hc; result->src[1] = main_block; diff --git a/server/docs/DS4.md b/server/docs/DS4.md index 2c77e8a47..d164a9b17 100644 --- a/server/docs/DS4.md +++ b/server/docs/DS4.md @@ -248,6 +248,9 @@ The runtime logs the chosen split with a `[deepseek4-split] auto-split:` banner. | `DFLASH_EXPERT_BUDGET_MB` | Main-GPU memory budget for hot experts. | | `DFLASH_DS4_HOTNESS_CSV` | Optional per-layer routing profile for hot placement. | | `GGML_BATCH_PEER_COPIES` | Batch peer-runtime copies and unlike-runtime pinned-host staging with one source wait per split. The old `GGML_CUDA_BATCH_PEER_COPIES` spelling remains an alias. | +| `DFLASH_DS4_TP_CRITICAL_PATH_PLACEMENT` | Use the routing profile and measured owner-rate ratio to minimize the predicted two-owner MoE critical path instead of maximizing aggregate hot-hit rate. Requires `DFLASH_DS4_HOTNESS_CSV`. | +| `DFLASH_DS4_TP_MAIN_TO_PEER_RATE` | Relative main/peer routed-expert rate used by critical-path placement. It must be finite and greater than zero; the default is `3.4`. | +| `DFLASH_DS4_TP_BALANCE_MIN_HOT` | Minimum hot experts retained on every routed layer by critical-path placement. Defaults to `0`. | | `DFLASH_DS4_Q5_VERIFY` | Opt in to the AMD q=5 fused verifier. This also selects the qualified MMVQ width and verifier-cache defaults when they are not explicitly overridden. | | `DFLASH_CUDA_MMVQ_FP4_Q5_X4_PLUS1` | Select the q=5 ROCmFP4 dense verifier kernel that reuses the existing x4 dot product for columns 0-3 and the exact scalar path for column 4. Defaults to `1` for q=5 on `gfx1201`; set `0` to force the generic five-column kernel. | | `DFLASH_DS4_TP_FUSED_CACHE_SLOTS` | Number of heterogeneous verifier graph slots. Defaults to `2` for q<=4 and `9` for the opt-in q=5 verifier; each slot retains scheduler scratch on both GPUs. | @@ -350,24 +353,50 @@ the shape that crosses two ratio-4 compressor boundaries, preserves five raw SWA rows for rollback, and restores plus replays only the accepted prefix after a partial rejection. q<=4 behavior is unchanged when the flag is absent. -On the qualified R9700 + Strix Halo profile, leaving the related controls +The heterogeneous verifier keeps all five lanes in one +`[n_embd * n_hc, q]` tensor. Attention HC-pre, attention HC-post, FFN HC-pre, +FFN HC-post, drafter-feature capture, and output HC merge are batched across +the verifier width. This removes the former per-lane HC controller paths and +progressive concatenations without changing the verifier result. The split +HC-post kernel also accepts a token dimension and joins the two owner outputs +inside that batched kernel. + +Critical-path placement models each routed layer as two concurrent branches: +the main branch includes its fixed shared-expert work and hot routed work, +while the peer branch executes the remaining routed work. The allocator adds +the next profiled expert only when its marginal reduction in +`max(main / main_to_peer_rate, peer)` is positive. The expert memory budget is +therefore an upper bound; leaving part of it unused is valid when another hot +expert would lengthen the predicted fork. + +On the qualified R9700 + Strix Halo profile, leaving the related q=5 controls unset selects `LUCE_MMVQ_MAX_NCOLS=5`, nine heterogeneous verifier slots, and the ROCmFP4 x4+1 dense kernel on `gfx1201`. The wider MMVQ ceiling avoids the slow small-matrix crossover, while nine slots hold the recurring compressor phases without steady graph rebuilds. The x4+1 kernel decodes shared weights through the existing four-column vector path and retains the original scalar accumulation for the fifth verifier column. -Explicit environment values still take priority. The hot-36 full sweep peaked -at 30.561 GiB on the reported 31.86 GiB R9700 and must be requalified on -smaller devices. +Explicit environment values still take priority. + +Sparse heterogeneous prefill uses a reusable graph allocator with preferred +128 MiB backing chunks. This avoids depending on one large contiguous HIP +allocation on devices without virtual-memory-backed buffers. Individual +tensors remain unsplit and may exceed the preferred chunk size. Prompts ending +above 4K use a 1K-token prefill shape, and that cap remains sticky for later +requests in the process so a post-16K request cannot force a fragmented +1K-to-2K arena replacement. Reproducible decode graph caches are retired before +a necessary prefill-arena growth; persistent HC mirrors remain resident. The exact qualification launch used: ```bash export DFLASH_DS4_Q5_VERIFY=1 export DFLASH_DS4_SPEC_Q=5 -export DFLASH_EXPERT_BUDGET_MB=13200 # 36 hot experts/layer on this profile +export DFLASH_EXPERT_BUDGET_MB=14350 export DFLASH_DS4_HOTNESS_CSV=/path/to/ds4_moe_tp_hotness.csv +export DFLASH_DS4_TP_CRITICAL_PATH_PLACEMENT=1 +export DFLASH_DS4_TP_MAIN_TO_PEER_RATE=4.4 +export DFLASH_DS4_TP_BALANCE_MIN_HOT=0 ``` The checked-in wrapper reproduces the full exact-context protocol and records @@ -378,6 +407,10 @@ trace: TARGET_MODEL=/path/to/target.gguf \ DRAFT_MODEL=/path/to/dspark-draft.gguf \ HOTNESS_CSV=/path/to/ds4_moe_tp_hotness.csv \ +CRITICAL_PATH_PLACEMENT=1 \ +MAIN_TO_PEER_RATE=4.4 \ +BALANCE_MIN_HOT=0 \ +EXPERT_BUDGET_MB=14350 \ server/scripts/qualify_ds4_q5_amd.sh ``` @@ -388,10 +421,44 @@ compatible artifact. At temperature zero, all 25 requests in the 2K -> 4K -> 8K -> 16K -> 2K burn-in produced the same expected response hash. With the automatic q=5 -MMVQ/cache/kernel defaults and the explicit hot-36 hardware profile, measured -medians were 67.957, 65.927, 62.544, 56.335, and 67.458 tok/s. Treat these as -workload-specific burn-in measurements, not as a portable default for unrelated -AMD memory layouts. +MMVQ/cache/kernel defaults and the critical-path profile above, measured client +decode medians were 75.818, 74.530, 69.898, 62.685, and 76.703 tok/s. The final +2K measurements after repeated 16K prefill were 76.685-76.727 tok/s, confirming +that the bounded sticky arena recovers steady decode rather than merely +surviving the request. The placement retained 1,688 profiled hot experts, 23-63 +per layer, and the full sweep peaked at 31.089 GiB on the reported 31.86 GiB +main GPU. Treat these as workload-specific burn-in measurements, not as a +portable default for unrelated memory layouts. + +For an overlap trace, run the same wrapper with the delayed profiler launcher: + +```bash +SERVER_BIN=server/scripts/rocprof_server_wrapper.sh \ +PROFILED_SERVER_BIN=/path/to/dflash_server \ +ROCPROF_OUTPUT_DIR=/path/to/trace-output \ +ROCPROF_START_SECONDS=180 \ +ROCPROF_DURATION_SECONDS=90 \ +server/scripts/qualify_ds4_q5_amd.sh + +server/scripts/analyze_rocprof_overlap.py \ + /path/to/trace-output/trace_kernel_trace.csv +``` + +The analyzer reports per-owner busy time, simultaneous kernel-busy time, +time-binned overlap, and the kernels dominating each owner. Use a steady decode +window rather than model load or prefill when comparing placement changes. + +In the post-batching trace, steady 2K decode windows placed only 16-22% of +either owner's kernel-busy time inside a simultaneously busy interval. The +unprofiled server timing attributed 63.7 ms of each 74.4 ms speculative step to +target verification; draft, head, snapshot, and apply work together accounted +for the remaining 10.7 ms. Changing the placement rate from 4.4 to 3.8 at the +same 14,350 MiB budget moved 116 experts to the peer but changed the measured +2K median by less than 0.1 tok/s. These measurements show that placement is +already near its local balance point. Further large gains require removing +split/copy dispatches or parallelizing work outside the routed-expert fork; +adding the two devices' headline bandwidths is not a valid throughput model +because attention, routing, HC boundaries, and every layer join remain ordered. On HIP `gfx1151`, enabling DSpark defaults `LUCE_MMVQ_MAX_NCOLS` to `4` when the variable is unset. This keeps the four-row verifier on MMVQ. On a 128 GiB diff --git a/server/scripts/analyze_rocprof_overlap.py b/server/scripts/analyze_rocprof_overlap.py new file mode 100755 index 000000000..4d5d9f11a --- /dev/null +++ b/server/scripts/analyze_rocprof_overlap.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +"""Summarize two-GPU overlap from a rocprofv3 kernel trace CSV.""" + +from __future__ import annotations + +import argparse +import csv +from collections import defaultdict +from pathlib import Path + + +def merge_intervals(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]: + if not intervals: + return [] + intervals.sort() + merged = [intervals[0]] + for start, end in intervals[1:]: + prev_start, prev_end = merged[-1] + if start <= prev_end: + if end > prev_end: + merged[-1] = (prev_start, end) + else: + merged.append((start, end)) + return merged + + +def merge_nearby_intervals( + intervals: list[tuple[int, int]], max_gap_ns: int +) -> list[tuple[int, int]]: + """Merge dispatch bursts separated only by launch-sized idle gaps.""" + if not intervals: + return [] + intervals.sort() + merged = [intervals[0]] + for start, end in intervals[1:]: + prev_start, prev_end = merged[-1] + if start <= prev_end + max_gap_ns: + merged[-1] = (prev_start, max(prev_end, end)) + else: + merged.append((start, end)) + return merged + + +def intersect_intervals( + left: list[tuple[int, int]], right: list[tuple[int, int]] +) -> list[tuple[int, int]]: + intersections: list[tuple[int, int]] = [] + i = 0 + j = 0 + while i < len(left) and j < len(right): + start = max(left[i][0], right[j][0]) + end = min(left[i][1], right[j][1]) + if start < end: + intersections.append((start, end)) + if left[i][1] <= right[j][1]: + i += 1 + else: + j += 1 + return intersections + + +def clipped_duration( + intervals: list[tuple[int, int]], start: int, end: int +) -> int: + total = 0 + for interval_start, interval_end in intervals: + if interval_end <= start: + continue + if interval_start >= end: + break + total += max(0, min(interval_end, end) - max(interval_start, start)) + return total + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("kernel_trace", type=Path) + parser.add_argument("--bin-ms", type=float, default=1000.0) + parser.add_argument("--window-start-s", type=float, default=0.0) + parser.add_argument("--window-end-s", type=float) + parser.add_argument("--top", type=int, default=12) + parser.add_argument( + "--timeline-max", type=int, default=0, + help="print at most this many per-agent dispatch bursts in the window", + ) + parser.add_argument( + "--timeline-merge-gap-us", type=float, default=25.0, + help="merge same-agent intervals separated by at most this gap", + ) + args = parser.parse_args() + if args.bin_ms <= 0 or args.window_start_s < 0: + parser.error("bin size must be positive and window start non-negative") + + intervals_by_agent: dict[str, list[tuple[int, int]]] = defaultdict(list) + trace_start: int | None = None + trace_end: int | None = None + with args.kernel_trace.open(newline="") as handle: + for row in csv.DictReader(handle): + agent = row["Agent_Id"] + start = int(row["Start_Timestamp"]) + end = int(row["End_Timestamp"]) + if end <= start: + continue + intervals_by_agent[agent].append((start, end)) + trace_start = start if trace_start is None else min(trace_start, start) + trace_end = end if trace_end is None else max(trace_end, end) + + if trace_start is None or trace_end is None: + raise SystemExit("kernel trace contains no positive-duration dispatches") + agents = sorted(intervals_by_agent) + if len(agents) != 2: + raise SystemExit(f"expected exactly two GPU agents, found {agents}") + merged = {agent: merge_intervals(intervals_by_agent[agent]) for agent in agents} + overlap = intersect_intervals(merged[agents[0]], merged[agents[1]]) + + window_start = trace_start + int(args.window_start_s * 1e9) + requested_end = ( + trace_start + int(args.window_end_s * 1e9) + if args.window_end_s is not None + else trace_end + ) + window_end = min(trace_end, requested_end) + if window_end <= window_start: + raise SystemExit("selected window is empty") + span = window_end - window_start + busy = { + agent: clipped_duration(merged[agent], window_start, window_end) + for agent in agents + } + overlap_ns = clipped_duration(overlap, window_start, window_end) + + duration_by_kernel: dict[str, dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) + count_by_kernel: dict[str, dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) + if args.top > 0: + with args.kernel_trace.open(newline="") as handle: + for row in csv.DictReader(handle): + agent = row["Agent_Id"] + start = max(window_start, int(row["Start_Timestamp"])) + end = min(window_end, int(row["End_Timestamp"])) + if end <= start: + continue + name = row["Kernel_Name"] + duration_by_kernel[agent][name] += end - start + count_by_kernel[agent][name] += 1 + + print( + f"window_s={(window_start-trace_start)/1e9:.3f}:" + f"{(window_end-trace_start)/1e9:.3f} span_s={span/1e9:.3f}" + ) + for agent in agents: + print( + f"{agent} busy_s={busy[agent]/1e9:.3f} " + f"utilization={100.0*busy[agent]/span:.2f}%" + ) + union_ns = busy[agents[0]] + busy[agents[1]] - overlap_ns + print( + f"both_busy_s={overlap_ns/1e9:.3f} " + f"overlap_of_{agents[0]}={100.0*overlap_ns/max(1,busy[agents[0]]):.2f}% " + f"overlap_of_{agents[1]}={100.0*overlap_ns/max(1,busy[agents[1]]):.2f}% " + f"either_busy_s={union_ns/1e9:.3f}" + ) + + bin_ns = max(1, int(args.bin_ms * 1e6)) + print("bin_start_s,agent1_busy_pct,agent2_busy_pct,both_busy_pct") + cursor = window_start + while cursor < window_end: + end = min(cursor + bin_ns, window_end) + width = end - cursor + print( + f"{(cursor-trace_start)/1e9:.3f}," + f"{100.0*clipped_duration(merged[agents[0]], cursor, end)/width:.2f}," + f"{100.0*clipped_duration(merged[agents[1]], cursor, end)/width:.2f}," + f"{100.0*clipped_duration(overlap, cursor, end)/width:.2f}" + ) + cursor = end + + for agent in agents: + print(f"top_kernels_{agent}") + top = sorted( + duration_by_kernel[agent].items(), key=lambda item: item[1], reverse=True + )[: args.top] + for name, duration in top: + print( + f"{duration/1e9:.6f}s count={count_by_kernel[agent][name]} {name}" + ) + + if args.timeline_max > 0: + gap_ns = max(0, int(args.timeline_merge_gap_us * 1e3)) + bursts: list[tuple[int, int, str]] = [] + for agent in agents: + for start, end in merge_nearby_intervals( + intervals_by_agent[agent], gap_ns + ): + start = max(start, window_start) + end = min(end, window_end) + if start < end: + bursts.append((start, end, agent)) + bursts.sort() + print( + "timeline_start_s,duration_us,agent," + f"merge_gap_us={args.timeline_merge_gap_us:g}" + ) + for start, end, agent in bursts[: args.timeline_max]: + print( + f"{(start-trace_start)/1e9:.9f}," + f"{(end-start)/1e3:.3f},{agent}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server/scripts/qualify_ds4_q5_amd.sh b/server/scripts/qualify_ds4_q5_amd.sh index 1a38d63ff..d86d01b1e 100755 --- a/server/scripts/qualify_ds4_q5_amd.sh +++ b/server/scripts/qualify_ds4_q5_amd.sh @@ -26,6 +26,9 @@ BLOCK_RADIX_TOPK="${BLOCK_RADIX_TOPK:-1}" PACK_Q4_INDEXER="${PACK_Q4_INDEXER:-0}" Q5_VERIFY="${Q5_VERIFY:-1}" FP4_Q5_X4_PLUS1="${FP4_Q5_X4_PLUS1:-auto}" +CRITICAL_PATH_PLACEMENT="${CRITICAL_PATH_PLACEMENT:-0}" +MAIN_TO_PEER_RATE="${MAIN_TO_PEER_RATE:-3.4}" +BALANCE_MIN_HOT="${BALANCE_MIN_HOT:-0}" EXPERT_BUDGET_MB="${EXPERT_BUDGET_MB:-13200}" WARMUP="${WARMUP:-2}" RUNS="${RUNS:-3}" @@ -33,7 +36,9 @@ MAX_TOKENS="${MAX_TOKENS:-128}" TARGETS="${TARGETS:-2048 4096 8192 16384 2048}" VRAM_MONITOR_SECONDS="${VRAM_MONITOR_SECONDS:-2}" HASH_MODELS="${HASH_MODELS:-0}" -RUN_ID="${RUN_ID:-ds4-q5-fr${FORCE_GRAPH_REPLAY}-direct${DIRECT_INDEXER_TOPK}-radix${BLOCK_RADIX_TOPK}-x4p1${FP4_Q5_X4_PLUS1}-$(date -u +%Y%m%dT%H%M%SZ)}" +CUDA_GRAPH_STATS_EVERY="${CUDA_GRAPH_STATS_EVERY:-200}" +CUDA_DISABLE_GRAPHS_DEVICES="${CUDA_DISABLE_GRAPHS_DEVICES:-}" +RUN_ID="${RUN_ID:-ds4-q5-fr${FORCE_GRAPH_REPLAY}-direct${DIRECT_INDEXER_TOPK}-radix${BLOCK_RADIX_TOPK}-x4p1${FP4_Q5_X4_PLUS1}-cp${CRITICAL_PATH_PLACEMENT}-r${MAIN_TO_PEER_RATE}-$(date -u +%Y%m%dT%H%M%SZ)}" OUT_ROOT="${OUT_ROOT:-$CHECKOUT/results/ds4_q5_context_qualification}" OUT_DIR="$OUT_ROOT/$RUN_ID" SERVER_LOG="$OUT_DIR/server.log" @@ -70,6 +75,19 @@ case "$FP4_Q5_X4_PLUS1" in auto|0|1) ;; *) echo "FP4_Q5_X4_PLUS1 must be auto, 0, or 1" >&2; exit 2 ;; esac +case "$CRITICAL_PATH_PLACEMENT" in + 0|1) ;; + *) echo "CRITICAL_PATH_PLACEMENT must be 0 or 1" >&2; exit 2 ;; +esac +if [[ ! "$MAIN_TO_PEER_RATE" =~ ^[0-9]+([.][0-9]+)?$ ]] || + ! awk -v value="$MAIN_TO_PEER_RATE" 'BEGIN { exit !(value > 0) }'; then + echo "MAIN_TO_PEER_RATE must be greater than zero" >&2 + exit 2 +fi +if [[ ! "$BALANCE_MIN_HOT" =~ ^[0-9]+$ ]]; then + echo "BALANCE_MIN_HOT must be a non-negative integer" >&2 + exit 2 +fi if [[ "$MMVQ_MAX_NCOLS" != auto && ! "$MMVQ_MAX_NCOLS" =~ ^[1-8]$ ]]; then echo "MMVQ_MAX_NCOLS must be auto or an integer from 1 through 8" >&2 exit 2 @@ -116,6 +134,7 @@ server_env=( "PATH=$PATH" "LD_LIBRARY_PATH=${LD_LIBRARY_PATH:-}" "GGML_CUDA_GRAPH_STATS=1" + "GGML_CUDA_GRAPH_STATS_EVERY=$CUDA_GRAPH_STATS_EVERY" "LUCE_CUDA_I32_REPEAT=1" "DFLASH_DS4_TOPK=4" "DFLASH_DS4_FUSED_VERIFY=1" @@ -166,6 +185,21 @@ server_env=( "DFLASH_MOE_FUSED_COMBINE=0" ) +if [[ -n "$CUDA_DISABLE_GRAPHS_DEVICES" ]]; then + server_env+=( + "GGML_CUDA_DISABLE_GRAPHS_DEVICES=$CUDA_DISABLE_GRAPHS_DEVICES" + ) +fi + +# Preserve only the explicit profiler-wrapper controls across env -i. Ordinary +# qualification runs leave these unset and retain the exact established env. +for profiler_var in PROFILED_SERVER_BIN ROCPROF_OUTPUT_DIR \ + ROCPROF_START_SECONDS ROCPROF_DURATION_SECONDS; do + if [[ -n "${!profiler_var:-}" ]]; then + server_env+=("$profiler_var=${!profiler_var}") + fi +done + if [[ "$MMVQ_MAX_NCOLS" != auto ]]; then server_env+=("LUCE_MMVQ_MAX_NCOLS=$MMVQ_MAX_NCOLS") fi @@ -194,7 +228,13 @@ fi if [[ "$FP4_Q5_X4_PLUS1" != auto ]]; then server_env+=("DFLASH_CUDA_MMVQ_FP4_Q5_X4_PLUS1=$FP4_Q5_X4_PLUS1") fi - +if [[ "$CRITICAL_PATH_PLACEMENT" == 1 ]]; then + server_env+=( + "DFLASH_DS4_TP_CRITICAL_PATH_PLACEMENT=1" + "DFLASH_DS4_TP_MAIN_TO_PEER_RATE=$MAIN_TO_PEER_RATE" + "DFLASH_DS4_TP_BALANCE_MIN_HOT=$BALANCE_MIN_HOT" + ) +fi server_args=( "$SERVER_BIN" "$TARGET_MODEL" --host 127.0.0.1 --port "$PORT" @@ -221,6 +261,9 @@ server_args=( echo "pack_q4_indexer=$PACK_Q4_INDEXER" echo "q5_verify=$Q5_VERIFY" echo "fp4_q5_x4_plus1=$FP4_Q5_X4_PLUS1" + echo "critical_path_placement=$CRITICAL_PATH_PLACEMENT" + echo "main_to_peer_rate=$MAIN_TO_PEER_RATE" + echo "balance_min_hot=$BALANCE_MIN_HOT" echo "cache_slots=$CACHE_SLOTS" echo "mmvq_max_ncols=$MMVQ_MAX_NCOLS" echo "targets=$TARGETS" @@ -228,6 +271,8 @@ server_args=( echo "runs=$RUNS" echo "max_tokens=$MAX_TOKENS" echo "max_ctx=$MAX_CTX" + echo "cuda_graph_stats_every=$CUDA_GRAPH_STATS_EVERY" + echo "cuda_disable_graphs_devices=$CUDA_DISABLE_GRAPHS_DEVICES" sha256sum "$SERVER_BIN" stat -c 'target_model=%n bytes=%s mtime=%y' "$TARGET_MODEL" stat -c 'draft_model=%n bytes=%s mtime=%y' "$DRAFT_MODEL" diff --git a/server/scripts/rocprof_server_wrapper.sh b/server/scripts/rocprof_server_wrapper.sh new file mode 100755 index 000000000..bacaca14e --- /dev/null +++ b/server/scripts/rocprof_server_wrapper.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Launch dflash_server under a delayed rocprofv3 collection window. This is a +# diagnostic wrapper for model-load-heavy runs where ptrace attachment is not +# permitted by the host policy. + +PROFILED_SERVER_BIN="${PROFILED_SERVER_BIN:?set PROFILED_SERVER_BIN}" +ROCPROF_OUTPUT_DIR="${ROCPROF_OUTPUT_DIR:?set ROCPROF_OUTPUT_DIR}" +ROCPROF_START_SECONDS="${ROCPROF_START_SECONDS:-180}" +ROCPROF_DURATION_SECONDS="${ROCPROF_DURATION_SECONDS:-90}" + +if [[ ! "$ROCPROF_START_SECONDS" =~ ^[0-9]+$ ]] || + [[ ! "$ROCPROF_DURATION_SECONDS" =~ ^[1-9][0-9]*$ ]]; then + echo "rocprof collection timing must use non-negative integer seconds" >&2 + exit 2 +fi + +mkdir -p "$ROCPROF_OUTPUT_DIR" +exec rocprofv3 \ + --kernel-trace \ + --memory-copy-trace \ + --group-by-queue true \ + --collection-period \ + "${ROCPROF_START_SECONDS}:${ROCPROF_DURATION_SECONDS}:1" \ + --output-format csv \ + --output-directory "$ROCPROF_OUTPUT_DIR" \ + --output-file trace \ + -- "$PROFILED_SERVER_BIN" "$@" diff --git a/server/src/common/moe_hybrid_placement.cpp b/server/src/common/moe_hybrid_placement.cpp index 49bcd5ea1..f9237f709 100644 --- a/server/src/common/moe_hybrid_placement.cpp +++ b/server/src/common/moe_hybrid_placement.cpp @@ -4,7 +4,9 @@ #include #include +#include #include +#include #include namespace dflash::common { @@ -261,4 +263,140 @@ bool MoeHybridPlacement::build_from_stats_with_layer_bytes( return true; } +bool MoeHybridPlacement::build_critical_path_balanced_from_stats( + const MoeHybridRoutingStats & stats, + const std::vector & layer_expert_bytes, + const std::vector & layer_main_fixed_bytes, + uint64_t total_hot_budget_bytes, + const MoeHybridCriticalPathConfig & config, + MoeHybridPlacement & out, + std::string * err) { + if (stats.empty() || stats.n_layer <= 0 || stats.n_expert <= 0) { + if (err) *err = "stats not initialized"; + return false; + } + if ((int) layer_expert_bytes.size() != stats.n_layer || + (int) layer_main_fixed_bytes.size() != stats.n_layer) { + if (err) *err = "critical-path layer byte vector size mismatch"; + return false; + } + if (total_hot_budget_bytes == 0) { + if (err) *err = "total_hot_budget_bytes must be > 0"; + return false; + } + if (config.active_experts <= 0 || + config.active_experts > stats.n_expert_used) { + if (err) *err = "active_experts must be within the routing profile width"; + return false; + } + if (!std::isfinite(config.main_to_peer_rate) || + config.main_to_peer_rate <= 0.0) { + if (err) *err = "main_to_peer_rate must be finite and > 0"; + return false; + } + + const int floor = std::clamp( + config.min_hot_per_layer, 0, stats.n_expert); + uint64_t used_bytes = 0; + for (int il = 0; il < stats.n_layer; ++il) { + const uint64_t expert_bytes = layer_expert_bytes[(size_t) il]; + if (expert_bytes == 0) continue; + if ((uint64_t) floor > + (std::numeric_limits::max() - used_bytes) / + expert_bytes) { + if (err) *err = "minimum hot placement byte count overflow"; + return false; + } + used_bytes += (uint64_t) floor * expert_bytes; + } + if (used_bytes > total_hot_budget_bytes) { + if (err) *err = "min_hot_per_layer exceeds byte budget"; + return false; + } + + MoeHybridPlacement tmp; + tmp.n_layer = stats.n_layer; + tmp.n_expert = stats.n_expert; + tmp.n_expert_used = stats.n_expert_used; + tmp.hot_counts.assign((size_t) tmp.n_layer, 0); + + std::vector> ranked((size_t) tmp.n_layer); + std::vector> prefix_counts((size_t) tmp.n_layer); + for (int il = 0; il < tmp.n_layer; ++il) { + ranked[(size_t) il] = stats.ranked_experts(il); + auto & prefix = prefix_counts[(size_t) il]; + prefix.assign((size_t) tmp.n_expert + 1, 0); + for (int n = 0; n < tmp.n_expert; ++n) { + prefix[(size_t) n + 1] = prefix[(size_t) n] + + stats.count(il, ranked[(size_t) il][(size_t) n]); + } + if (layer_expert_bytes[(size_t) il] > 0) { + tmp.hot_counts[(size_t) il] = floor; + } + } + + auto layer_cost = [&](int il, int hot_count) { + const uint64_t expert_bytes = layer_expert_bytes[(size_t) il]; + if (expert_bytes == 0) return 0.0; + const auto & prefix = prefix_counts[(size_t) il]; + const uint64_t total = prefix.back(); + const double hot_probability = total > 0 + ? (double) prefix[(size_t) hot_count] / (double) total + : (double) hot_count / (double) tmp.n_expert; + const double routed_bytes = + (double) config.active_experts * (double) expert_bytes; + const double main_work = + (double) layer_main_fixed_bytes[(size_t) il] + + routed_bytes * hot_probability; + const double peer_work = routed_bytes * (1.0 - hot_probability); + return std::max( + main_work / config.main_to_peer_rate, + peer_work); + }; + + uint64_t remaining = total_hot_budget_bytes - used_bytes; + while (true) { + int best_layer = -1; + double best_value = 0.0; + double best_gain = 0.0; + for (int il = 0; il < tmp.n_layer; ++il) { + const int current = tmp.hot_counts[(size_t) il]; + const uint64_t bytes = layer_expert_bytes[(size_t) il]; + if (current >= tmp.n_expert || bytes == 0 || bytes > remaining) { + continue; + } + const double gain = + layer_cost(il, current) - layer_cost(il, current + 1); + if (!(gain > 0.0)) continue; + const double value = gain / (double) bytes; + if (best_layer < 0 || value > best_value || + (value == best_value && gain > best_gain)) { + best_layer = il; + best_value = value; + best_gain = gain; + } + } + if (best_layer < 0) break; + const uint64_t bytes = layer_expert_bytes[(size_t) best_layer]; + tmp.hot_counts[(size_t) best_layer]++; + remaining -= bytes; + } + + tmp.total_hot = + std::accumulate(tmp.hot_counts.begin(), tmp.hot_counts.end(), 0); + tmp.hot_expert_ids.resize((size_t) tmp.n_layer); + for (int il = 0; il < tmp.n_layer; ++il) { + const int hot_count = tmp.hot_counts[(size_t) il]; + auto & hot = tmp.hot_expert_ids[(size_t) il]; + hot.reserve((size_t) hot_count); + for (int n = 0; n < hot_count; ++n) { + hot.push_back( + (int32_t) ranked[(size_t) il][(size_t) n]); + } + } + + out = std::move(tmp); + return true; +} + } // namespace dflash::common diff --git a/server/src/common/moe_hybrid_placement.h b/server/src/common/moe_hybrid_placement.h index a522a4825..d726b10a5 100644 --- a/server/src/common/moe_hybrid_placement.h +++ b/server/src/common/moe_hybrid_placement.h @@ -14,6 +14,15 @@ namespace dflash::common { struct MoeHybridRoutingStats; // forward decl +// Cost model for balancing two concurrently executed MoE owners. Rates are +// relative, so peer_rate is normalized to one and main_to_peer_rate expresses +// how many equivalent expert bytes the main owner consumes in the same time. +struct MoeHybridCriticalPathConfig { + int active_experts = 0; + int min_hot_per_layer = 0; + double main_to_peer_rate = 1.0; +}; + inline uint64_t moe_hybrid_core_bytes_from_memory(const char * log_prefix, size_t gpu_free, size_t gpu_total) { @@ -64,6 +73,21 @@ struct MoeHybridPlacement { int min_hot_per_layer, MoeHybridPlacement & out, std::string * err = nullptr); + + // Distribute main-owner experts to minimize the sum of predicted per-layer + // fork times, max(main, peer), rather than merely maximizing aggregate hit + // rate. layer_main_fixed_bytes accounts for owner-local work such as the + // shared expert that runs on every route. The byte budget is an upper + // bound; allocation stops when another hot expert would lengthen the + // critical path. + static bool build_critical_path_balanced_from_stats( + const MoeHybridRoutingStats & stats, + const std::vector & layer_expert_bytes, + const std::vector & layer_main_fixed_bytes, + uint64_t total_hot_budget_bytes, + const MoeHybridCriticalPathConfig & config, + MoeHybridPlacement & out, + std::string * err = nullptr); }; } // namespace dflash::common diff --git a/server/src/deepseek4/deepseek4_backend.cpp b/server/src/deepseek4/deepseek4_backend.cpp index b28b54c78..3aafad282 100644 --- a/server/src/deepseek4/deepseek4_backend.cpp +++ b/server/src/deepseek4/deepseek4_backend.cpp @@ -42,6 +42,24 @@ static bool env_flag_enabled(const char * name) { return value && value[0] && std::strcmp(value, "0") != 0; } +static bool positive_env_double(const char * name, double fallback, + double & out, std::string * err) { + out = fallback; + const char * raw = std::getenv(name); + if (!raw || !*raw) return true; + char * end = nullptr; + const double parsed = std::strtod(raw, &end); + if (end == raw || *end != '\0' || !std::isfinite(parsed) || + parsed <= 0.0) { + if (err) { + *err = std::string(name) + " must be a finite value greater than zero"; + } + return false; + } + out = parsed; + return true; +} + static void configure_dspark_mmvq_defaults(int gpu) { #if defined(DFLASH27B_BACKEND_HIP) || defined(GGML_USE_HIP) if (!env_flag_enabled("DFLASH_DS4_SPEC")) { @@ -278,6 +296,14 @@ static uint64_t layer_expert_bytes(const DeepSeek4Layer & layer, int n_expert) { return bytes; } +static uint64_t layer_shared_expert_bytes(const DeepSeek4Layer & layer) { + uint64_t bytes = 0; + if (layer.ffn_gate_shexp) bytes += ggml_nbytes(layer.ffn_gate_shexp); + if (layer.ffn_up_shexp) bytes += ggml_nbytes(layer.ffn_up_shexp); + if (layer.ffn_down_shexp) bytes += ggml_nbytes(layer.ffn_down_shexp); + return bytes; +} + struct Ds4ExpertMemoryInfo { std::vector layer_expert_bytes; uint64_t total_expert_bytes = 0; @@ -1020,6 +1046,10 @@ bool DeepSeek4Backend::compute_uniform_hybrid_placement(const DeepSeek4Weights & const bool concentrate_requested = tp.concentrate_secondary; bool concentrated = false; int retained_local = 0; + const char * profile_path = std::getenv("DFLASH_DS4_HOTNESS_CSV"); + const bool critical_path_placement = + !tp.all_on_secondary && !concentrate_requested && + env_flag_enabled("DFLASH_DS4_TP_CRITICAL_PATH_PLACEMENT"); const int requested_cold = w.n_layer * std::max(0, w.n_expert - hot_per_layer); if (concentrate_requested && requested_cold >= w.n_expert) { @@ -1031,7 +1061,79 @@ bool DeepSeek4Backend::compute_uniform_hybrid_placement(const DeepSeek4Weights & "[deepseek4] concentrated secondary placement needs at least " "one complete layer; using uniform placement\n"); fill_prefix_hot_placement(w, hot_per_layer, out); - } else if (const char * profile_path = std::getenv("DFLASH_DS4_HOTNESS_CSV")) { + } else if (critical_path_placement) { + if (!profile_path || !*profile_path) { + if (err) { + *err = "critical-path placement requires DFLASH_DS4_HOTNESS_CSV"; + } + return false; + } + MoeHybridRoutingStats stats; + if (!MoeHybridRoutingStats::load_csv(profile_path, stats, err)) { + return false; + } + if (stats.n_layer != w.n_layer || stats.n_expert != w.n_expert) { + if (err) { + *err = "routing profile shape does not match DeepSeek V4 target"; + } + return false; + } + + int active_experts = cfg_.expert_top_k > 0 + ? cfg_.expert_top_k : w.n_expert_used; + if (const char * raw_top_k = std::getenv("DFLASH_DS4_TOPK")) { + const int env_top_k = std::atoi(raw_top_k); + if (env_top_k > 0) active_experts = env_top_k; + } + if (active_experts <= 0 || active_experts > w.n_expert_used) { + if (err) *err = "critical-path placement active expert count is invalid"; + return false; + } + + double main_to_peer_rate = 3.4; + if (!positive_env_double( + "DFLASH_DS4_TP_MAIN_TO_PEER_RATE", 3.4, + main_to_peer_rate, err)) { + return false; + } + MoeHybridCriticalPathConfig balance_cfg; + balance_cfg.active_experts = active_experts; + balance_cfg.main_to_peer_rate = main_to_peer_rate; + if (const char * raw_floor = + std::getenv("DFLASH_DS4_TP_BALANCE_MIN_HOT")) { + balance_cfg.min_hot_per_layer = std::max(0, std::atoi(raw_floor)); + } + + std::vector main_fixed_bytes((size_t) w.n_layer, 0); + for (int il = 0; il < w.n_layer; ++il) { + main_fixed_bytes[(size_t) il] = + layer_shared_expert_bytes(w.layers[(size_t) il]); + } + if (!MoeHybridPlacement::build_critical_path_balanced_from_stats( + stats, budget.mem.layer_expert_bytes, main_fixed_bytes, + budget.expert_budget, balance_cfg, out, err)) { + return false; + } + + const auto [min_hot, max_hot] = std::minmax_element( + out.hot_counts.begin(), out.hot_counts.end()); + const double mean_hot = out.hot_counts.empty() ? 0.0 + : (double) out.total_hot / (double) out.hot_counts.size(); + std::fprintf(stderr, + "[deepseek4] hybrid critical-path placement: " + "profile=%s active=%d main/peer=%.3f " + "hot/layer=%.1f [%d,%d]\n", + profile_path, active_experts, main_to_peer_rate, + mean_hot, + min_hot != out.hot_counts.end() ? *min_hot : 0, + max_hot != out.hot_counts.end() ? *max_hot : 0); + std::fprintf(stderr, + "[deepseek4] hybrid critical-path hot counts:"); + for (int count : out.hot_counts) { + std::fprintf(stderr, " %d", count); + } + std::fprintf(stderr, "\n"); + } else if (profile_path) { if (*profile_path) { const bool profile_hot_on_secondary = tp.in_process && tp.profile_hot_on_secondary; @@ -1080,8 +1182,10 @@ bool DeepSeek4Backend::compute_uniform_hybrid_placement(const DeepSeek4Weights & retained_local); } + const std::string hot_label = critical_path_placement + ? "balanced" : std::to_string(hot_per_layer); std::fprintf(stderr, - "[deepseek4] hybrid placement: gpu_total=%.2f GiB gpu_free=%.2f GiB core=%.2f GiB kv=%.2f GiB warm=%.2f GiB safety=%.2f GiB expert_budget=%.2f GiB hot/layer=%d\n", + "[deepseek4] hybrid placement: gpu_total=%.2f GiB gpu_free=%.2f GiB core=%.2f GiB kv=%.2f GiB warm=%.2f GiB safety=%.2f GiB expert_budget=%.2f GiB hot/layer=%s\n", gib((uint64_t) budget.gpu_total), gib((uint64_t) budget.gpu_free), gib(budget.core_bytes), @@ -1089,7 +1193,7 @@ bool DeepSeek4Backend::compute_uniform_hybrid_placement(const DeepSeek4Weights & gib(budget.warm_bytes), gib(budget.safety_bytes), gib(budget.expert_budget), - hot_per_layer); + hot_label.c_str()); log_ds4_expert_memory_info("placement", placed_mem, w.n_layer); return true; } @@ -1333,6 +1437,21 @@ bool DeepSeek4Backend::unpark(ParkTarget target) { return true; } +int deepseek4_hybrid_prefill_chunk_tokens( + int requested_chunk, + int context_end, + int current_cap) { + constexpr int long_context_begin = 4096; + constexpr int long_context_chunk = 1024; + int bounded = std::max(1, requested_chunk); + if (current_cap > 0) { + bounded = std::min(bounded, current_cap); + } + return context_end > long_context_begin + ? std::min(bounded, long_context_chunk) + : bounded; +} + int DeepSeek4Backend::do_prefill(const std::vector & tokens, const DaemonIO & io, int kv_offset, @@ -1356,12 +1475,34 @@ int DeepSeek4Backend::do_prefill(const std::vector & tokens, // post-mixing and corrupt the hidden state. const bool hybrid_batch_supported = !moe_hybrid_ || cfg_.prefill_mode == PrefillAttentionMode::Sparse; - const int chunk = + const int base_chunk = !prefill_attention_mode_is_approximate(cfg_.prefill_mode) || !hybrid_batch_supported ? 1 : std::max(1, std::min(requested_chunk, layer_major_cap)); + const bool bound_hybrid_scratch = + moe_hybrid_ && + cfg_.prefill_mode == PrefillAttentionMode::Sparse; + const int safe_chunk = bound_hybrid_scratch + ? deepseek4_hybrid_prefill_chunk_tokens( + base_chunk, kv_offset + n_total, + hybrid_prefill_chunk_cap_) + : base_chunk; + if (safe_chunk < base_chunk) { + hybrid_prefill_chunk_cap_ = hybrid_prefill_chunk_cap_ > 0 + ? std::min(hybrid_prefill_chunk_cap_, safe_chunk) + : safe_chunk; + } + const int chunk = bound_hybrid_scratch && hybrid_prefill_chunk_cap_ > 0 + ? std::min(base_chunk, hybrid_prefill_chunk_cap_) + : base_chunk; + if (chunk < base_chunk) { + std::fprintf(stderr, + "[deepseek4] hybrid prefill scratch bound: " + "chunk %d->%d for context_end=%d (sticky)\n", + base_chunk, chunk, kv_offset + n_total); + } int pos = kv_offset; const bool save_snapshot = snap_slot >= 0 && snap_slot < PREFIX_SLOTS && diff --git a/server/src/deepseek4/deepseek4_backend.h b/server/src/deepseek4/deepseek4_backend.h index 32f7230ce..98e239e42 100644 --- a/server/src/deepseek4/deepseek4_backend.h +++ b/server/src/deepseek4/deepseek4_backend.h @@ -26,6 +26,13 @@ namespace dflash::common { +// Bounds the sparse heterogeneous prefill arena once accumulated attention +// context dominates its memory footprint. Decode batching is unaffected. +int deepseek4_hybrid_prefill_chunk_tokens( + int requested_chunk, + int context_end, + int current_cap = 0); + class DeepSeek4Backend : public ModelBackend { public: explicit DeepSeek4Backend(const DeepSeek4BackendConfig & cfg); @@ -99,6 +106,9 @@ class DeepSeek4Backend : public ModelBackend { ggml_backend_t spec_backend_ = nullptr; std::unique_ptr spec_drafter_; std::vector spec_feat_window_; + // Once a long prompt selects the fragmentation-safe prefill shape, retain + // it for later requests so the HIP arenas never switch back under load. + int hybrid_prefill_chunk_cap_ = 0; bool load_spec_drafter(); void release_spec_drafter(bool mark_parked); diff --git a/server/src/deepseek4/deepseek4_fused_verify.inc b/server/src/deepseek4/deepseek4_fused_verify.inc index fcf228095..8cd1655ef 100644 --- a/server/src/deepseek4/deepseek4_fused_verify.inc +++ b/server/src/deepseek4/deepseek4_fused_verify.inc @@ -424,16 +424,18 @@ static bool ds4_build_fused_verify_graph( ggml_set_input(fg.mask_bundle); int64_t mask_off = 0; - // per-token HC streams - std::vector hc_cur(q); - for (int t = 0; t < q; ++t) { - ggml_tensor * col = ggml_view_2d(ctx, fg.inp_embed, n_embd, 1, - fg.inp_embed->nb[1], (size_t) t * fg.inp_embed->nb[1]); - hc_cur[(size_t) t] = ggml_repeat_4d(ctx, col, n_embd, n_hc, 1, 1); - } + // Keep every speculative lane in one HC tensor. The previous graph built + // five independent HC ops and progressively concatenated their outputs at + // every pre/post boundary. HC kernels already support a token dimension, + // so preserve it throughout the verifier and issue one batched op instead. + ggml_tensor * embed_3d = ggml_reshape_3d( + ctx, fg.inp_embed, n_embd, 1, q); + ggml_tensor * hc_repeated = ggml_repeat_4d( + ctx, embed_3d, n_embd, n_hc, q, 1); + ggml_tensor * hc_cur = ggml_reshape_2d( + ctx, hc_repeated, (int64_t) n_embd * n_hc, q); - std::vector capture_pieces( - capture_ids.size() * (size_t) q, nullptr); // order [ci][t] + std::vector capture_layers(capture_ids.size(), nullptr); for (int il = 0; il < w.n_layer; ++il) { const DeepSeek4Layer & L = w.layers[(size_t) il]; @@ -447,20 +449,16 @@ static bool ds4_build_fused_verify_graph( const int lane_token_pos = lane_kv_start + lane_q - 1; const size_t hybrid_idx = (size_t) il; - // ── HC pre (attention), per token ── - std::vector hc_flat(lane_q), split_attn(lane_q); - ggml_tensor * attn_in = nullptr; - for (int t = 0; t < lane_q; ++t) { - const size_t token_idx = (size_t) lane_start + (size_t) t; - hc_flat[(size_t) t] = ggml_reshape_1d( - ctx, hc_cur[token_idx], (int64_t) n_embd * n_hc); - ggml_tensor * working = ds4_build_fused_hc_pre(ctx, w, hc_flat[(size_t) t], - mc.fn_attn_f16[(size_t) il], L.hc_attn_base, - hlw.attn, &split_attn[(size_t) t]); - if (!working) return false; - ggml_tensor * w2 = ggml_reshape_2d(ctx, working, n_embd, 1); - attn_in = attn_in ? ggml_concat(ctx, attn_in, w2, 1) : w2; - } + // ── Batched HC pre (attention) ── + ggml_tensor * hc_flat = hc_cur; + ggml_tensor * split_attn = nullptr; + ggml_tensor * attn_working = ds4_build_fused_hc_pre( + ctx, w, hc_flat, mc.fn_attn_f16[(size_t) il], L.hc_attn_base, + hlw.attn, &split_attn); + if (!attn_working) return false; + // HC-pre stores working and split values together per token. Materialize + // the strided working view once, replacing q kernels and q-1 concats. + ggml_tensor * attn_in = ggml_cont(ctx, attn_working); // ── Batched attention ── DeepSeek4AttentionGraphInputs ain{}; @@ -525,27 +523,18 @@ static bool ds4_build_fused_verify_graph( return false; } - // ── HC post (attention) + HC pre (FFN), per token ── - ggml_tensor * ffn_in = nullptr; - std::vector split_ffn(lane_q); - for (int t = 0; t < lane_q; ++t) { - const size_t token_idx = (size_t) lane_start + (size_t) t; - ggml_tensor * ao = ggml_view_2d(ctx, attn_out, n_embd, 1, - attn_out->nb[1], (size_t) t * attn_out->nb[1]); - ggml_tensor * ao_flat = ggml_reshape_1d(ctx, ggml_cont(ctx, ao), n_embd); - ggml_tensor * hc_next = ggml_ds4_hc_post(ctx, hc_flat[(size_t) t], ao_flat, - split_attn[(size_t) t], n_hc); - hc_cur[token_idx] = ggml_reshape_2d( - ctx, hc_next, n_embd, n_hc); - hc_flat[(size_t) t] = ggml_reshape_1d( - ctx, hc_cur[token_idx], (int64_t) n_embd * n_hc); - ggml_tensor * fworking = ds4_build_fused_hc_pre(ctx, w, hc_flat[(size_t) t], - mc.fn_ffn_f16[(size_t) il], L.hc_ffn_base, - hlw.ffn, &split_ffn[(size_t) t]); - if (!fworking) return false; - ggml_tensor * f2 = ggml_reshape_2d(ctx, fworking, n_embd, 1); - ffn_in = ffn_in ? ggml_concat(ctx, ffn_in, f2, 1) : f2; - } + // ── Batched HC post (attention) + HC pre (FFN) ── + ggml_tensor * attn_batch = ggml_is_contiguous(attn_out) + ? attn_out : ggml_cont(ctx, attn_out); + hc_cur = ggml_ds4_hc_post( + ctx, hc_flat, attn_batch, split_attn, n_hc); + hc_flat = hc_cur; + ggml_tensor * split_ffn = nullptr; + ggml_tensor * ffn_working = ds4_build_fused_hc_pre( + ctx, w, hc_flat, mc.fn_ffn_f16[(size_t) il], L.hc_ffn_base, + hlw.ffn, &split_ffn); + if (!ffn_working) return false; + ggml_tensor * ffn_in = ggml_cont(ctx, ffn_working); // ── Batched FFN ── ggml_tensor * ffn_normed = build_rms_norm(ctx, ffn_in, L.ffn_norm, w.rms_eps); @@ -739,64 +728,39 @@ static bool ds4_build_fused_verify_graph( } if (!ffn_out) return false; - // ── HC post (FFN), per token; capture at drafter layers ── - for (int t = 0; t < lane_q; ++t) { - const size_t token_idx = (size_t) lane_start + (size_t) t; - ggml_tensor * hc_next = nullptr; - if (fused_hc_join_inputs) { - ggml_tensor * main_part = ggml_view_2d( - ctx, fused_hc_join_inputs->main_output, n_embd, 1, - fused_hc_join_inputs->main_output->nb[1], - (size_t) t * fused_hc_join_inputs->main_output->nb[1]); - ggml_tensor * peer_part = ggml_view_2d( - ctx, fused_hc_join_inputs->peer_output, n_embd, 1, - fused_hc_join_inputs->peer_output->nb[1], - (size_t) t * fused_hc_join_inputs->peer_output->nb[1]); - hc_next = ggml_ds4_hc_post_split( - ctx, hc_flat[(size_t) t], main_part, peer_part, - split_ffn[(size_t) t], n_hc); - } else { - ggml_tensor * fo = ggml_view_2d( - ctx, ffn_out, n_embd, 1, ffn_out->nb[1], - (size_t) t * ffn_out->nb[1]); - ggml_tensor * fo_flat = ggml_reshape_1d( - ctx, ggml_cont(ctx, fo), n_embd); - hc_next = ggml_ds4_hc_post( - ctx, hc_flat[(size_t) t], fo_flat, - split_ffn[(size_t) t], n_hc); - } - hc_cur[token_idx] = ggml_reshape_2d( - ctx, hc_next, n_embd, n_hc); + // ── Batched HC post (FFN); capture at drafter layers ── + if (fused_hc_join_inputs) { + hc_cur = ggml_ds4_hc_post_split( + ctx, hc_flat, fused_hc_join_inputs->main_output, + fused_hc_join_inputs->peer_output, split_ffn, n_hc); + } else { + ggml_tensor * ffn_batch = ggml_is_contiguous(ffn_out) + ? ffn_out : ggml_cont(ctx, ffn_out); + hc_cur = ggml_ds4_hc_post( + ctx, hc_flat, ffn_batch, split_ffn, n_hc); } for (size_t ci = 0; ci < capture_ids.size(); ++ci) { if (capture_ids[ci] != il) continue; - for (int t = 0; t < lane_q; ++t) { - const size_t token_idx = - (size_t) lane_start + (size_t) t; - ggml_tensor * hs = hc_cur[token_idx]; // [n_embd, n_hc] - ggml_tensor * hsT = ggml_cont(ctx, ggml_transpose(ctx, hs)); // [n_hc, n_embd] - ggml_tensor * summed = ggml_sum_rows(ctx, hsT); // [1, n_embd] - ggml_tensor * mean = ggml_scale(ctx, summed, 1.0f / (float) n_hc); - capture_pieces[ci * (size_t) q + (size_t) t] = - ggml_reshape_1d(ctx, ggml_cont(ctx, mean), n_embd); - } + ggml_tensor * hs = ggml_reshape_3d( + ctx, hc_cur, n_embd, n_hc, lane_q); + ggml_tensor * hs_t = ggml_cont( + ctx, ggml_transpose(ctx, hs)); // [n_hc,n_embd,q] + ggml_tensor * summed = ggml_sum_rows(ctx, hs_t); // [1,n_embd,q] + ggml_tensor * mean = ggml_scale( + ctx, summed, 1.0f / (float) n_hc); + capture_layers[ci] = ggml_reshape_2d( + ctx, mean, n_embd, lane_q); // [n_embd,q] } } - // ── Output: per-token HC merge → batched out_norm + lm_head ── - ggml_tensor * final_all = nullptr; - for (int t = 0; t < q; ++t) { - ggml_tensor * hc_flat = ggml_reshape_1d(ctx, hc_cur[(size_t) t], (int64_t) n_embd * n_hc); - ggml_tensor * onorm = ggml_rms_norm(ctx, hc_flat, w.hc_eps); - ggml_tensor * omix = ggml_mul_mat(ctx, mc.fn_out_f16, onorm); - omix = ggml_reshape_1d(ctx, omix, ggml_nelements(omix)); - ggml_tensor * obase = ds4_fused_hc_base_f32(ctx, w.output_hc_base); - if (!obase || hc_out_weights.scale_data.empty()) return false; - ggml_tensor * fe = ggml_ds4_hc_out(ctx, omix, obase, hc_flat, n_hc, - hc_out_weights.scale_data[0]); - ggml_tensor * fe2 = ggml_reshape_2d(ctx, fe, n_embd, 1); - final_all = final_all ? ggml_concat(ctx, final_all, fe2, 1) : fe2; - } + // ── Batched output HC merge → out_norm + lm_head ── + ggml_tensor * onorm = ggml_rms_norm(ctx, hc_cur, w.hc_eps); + ggml_tensor * omix = ggml_mul_mat(ctx, mc.fn_out_f16, onorm); + omix = ggml_reshape_2d(ctx, omix, ggml_nelements(omix) / q, q); + ggml_tensor * obase = ds4_fused_hc_base_f32(ctx, w.output_hc_base); + if (!obase || hc_out_weights.scale_data.empty()) return false; + ggml_tensor * final_all = ggml_ds4_hc_out( + ctx, omix, obase, hc_cur, n_hc, hc_out_weights.scale_data[0]); ggml_tensor * out_normed = build_rms_norm(ctx, final_all, w.out_norm, w.rms_eps); fg.logits = ggml_mul_mat(ctx, w.output, out_normed); // [n_vocab, q] ggml_set_output(fg.logits); @@ -807,16 +771,18 @@ static bool ds4_build_fused_verify_graph( ggml_build_forward_expand(gf, ex.argmax); } - if (!capture_pieces.empty()) { - for (ggml_tensor * piece : capture_pieces) { - if (!piece) { + if (!capture_layers.empty()) { + for (ggml_tensor * layer : capture_layers) { + if (!layer) { std::fprintf(stderr, "[ds4-fused-verify] capture layer id is invalid\n"); return false; } } - ggml_tensor * cap = capture_pieces[0]; - for (size_t i = 1; i < capture_pieces.size(); ++i) { - cap = ggml_concat(ctx, cap, capture_pieces[i], 0); + // Concatenating layer matrices on dim 0 produces the token-major + // [n_capture*n_embd,q] layout consumed by the drafter directly. + ggml_tensor * cap = capture_layers[0]; + for (size_t i = 1; i < capture_layers.size(); ++i) { + cap = ggml_concat(ctx, cap, capture_layers[i], 0); } ex.capture = cap; ggml_set_output(ex.capture); @@ -1359,17 +1325,10 @@ static int ds4_try_fused_verify_step( out_logits.clear(); } if (hooks->capture_out && ex->capture && ncap > 0) { - std::vector flat((size_t) w.n_embd * ncap * q); - ggml_backend_tensor_get(ex->capture, flat.data(), 0, sizeof(float) * flat.size()); - hooks->capture_out->assign((size_t) ncap * w.n_embd * q, 0.0f); - for (int ci = 0; ci < ncap; ++ci) { - for (int t = 0; t < q; ++t) { - const float * src = flat.data() + ((size_t) ci * q + t) * w.n_embd; - float * dst = hooks->capture_out->data() + - (size_t) t * ncap * w.n_embd + (size_t) ci * w.n_embd; - std::memcpy(dst, src, sizeof(float) * (size_t) w.n_embd); - } - } + hooks->capture_out->resize((size_t) ncap * w.n_embd * q); + ggml_backend_tensor_get( + ex->capture, hooks->capture_out->data(), 0, + sizeof(float) * hooks->capture_out->size()); } if (telemetry) { telemetry->full_graph_read_us += ds4_elapsed_us(read_t0, Ds4TimingClock::now()); diff --git a/server/src/deepseek4/deepseek4_graph.cpp b/server/src/deepseek4/deepseek4_graph.cpp index c7db80112..3fa4e4e9a 100644 --- a/server/src/deepseek4/deepseek4_graph.cpp +++ b/server/src/deepseek4/deepseek4_graph.cpp @@ -4677,8 +4677,15 @@ struct DeepSeek4FusedDecodeCache { std::vector fn_ffn_f16; ggml_tensor * fn_out_f16 = nullptr; + void evict_graphs() { + for (auto & slot : slots) { + slot.destroy(backend); + } + counter = 0; + } + void destroy() { - for (auto & s : slots) s.destroy(backend); + evict_graphs(); if (fn_buf) { ggml_backend_buffer_free(fn_buf); fn_buf = nullptr; } if (fn_ctx) { ggml_free(fn_ctx); fn_ctx = nullptr; } fn_attn_f16.clear(); @@ -4712,7 +4719,7 @@ struct Ds4FusedVerifyCache { ggml_tensor * ape128 = nullptr; // i32 [q] ggml_tensor * st4 = nullptr; // i64 [1,q] ggml_tensor * st128 = nullptr; // i64 [1,q] - ggml_tensor * capture = nullptr; // f32 [n_embd*ncap*q], order [ci][t] + ggml_tensor * capture = nullptr; // f32 [n_embd*ncap,q], token-major ggml_tensor * argmax = nullptr; // i32 [q], optional greedy output // Reused host staging for the context-sized additive attention mask. // Keeping it per slot removes one allocation from every verify step. @@ -4844,22 +4851,32 @@ static ggml_tensor * ds4_fused_hc_base_f32(ggml_context * ctx, ggml_tensor * bas static ggml_tensor * ds4_build_fused_hc_pre( ggml_context * ctx, const DeepSeek4Weights & w, - ggml_tensor * hc_flat, // [n_embd*n_hc] contiguous f32 + ggml_tensor * hc_flat, // [n_embd*n_hc,n_tokens] contiguous f32 ggml_tensor * fn, ggml_tensor * base, const HcWeightsCpu & cw, ggml_tensor ** out_split) { if (!fn || !base || !cw.loaded || cw.scale_data.size() < 3) return nullptr; const int mix_dim = 2 * w.n_hc + w.n_hc * w.n_hc; + const int64_t n_tokens = hc_flat->ne[1]; ggml_tensor * normed = ggml_rms_norm(ctx, hc_flat, w.hc_eps); ggml_tensor * mix = ggml_mul_mat(ctx, fn, normed); - mix = ggml_reshape_1d(ctx, mix, mix_dim); + mix = n_tokens == 1 + ? ggml_reshape_1d(ctx, mix, mix_dim) + : ggml_reshape_2d(ctx, mix, mix_dim, n_tokens); ggml_tensor * base_f32 = ds4_fused_hc_base_f32(ctx, base); ggml_tensor * pre = ggml_ds4_hc_pre(ctx, mix, base_f32, hc_flat, w.n_hc, w.n_hc_sinkhorn_iter, cw.scale_data[0], cw.scale_data[1], cw.scale_data[2]); - *out_split = ggml_view_1d(ctx, pre, mix_dim, (size_t) w.n_embd * sizeof(float)); - return ggml_view_1d(ctx, pre, w.n_embd, 0); + if (n_tokens == 1) { + *out_split = ggml_view_1d( + ctx, pre, mix_dim, (size_t) w.n_embd * sizeof(float)); + return ggml_view_1d(ctx, pre, w.n_embd, 0); + } + *out_split = ggml_view_2d( + ctx, pre, mix_dim, n_tokens, pre->nb[1], + (size_t) w.n_embd * sizeof(float)); + return ggml_view_2d(ctx, pre, w.n_embd, n_tokens, pre->nb[1], 0); } static ggml_tensor * ds4_build_hash_routed_ffn( @@ -7401,17 +7418,97 @@ bool deepseek4_step_layer_range( auto & attn_alloc = heterogeneous_sparse_prefill ? shared_prefill_attn_alloc : cached_attn_allocs[(size_t)il]; + constexpr size_t shared_prefill_max_chunk = + 128u * 1024u * 1024u; if (!attn_alloc.valid() || attn_alloc.owner_ctx != w.ctx || attn_alloc.backend != backend) { attn_alloc.free(); - attn_alloc.alloc = ggml_gallocr_new(ggml_backend_get_default_buffer_type(backend)); + attn_alloc.alloc = heterogeneous_sparse_prefill + ? ggml_gallocr_new_with_max_chunk_size( + ggml_backend_get_default_buffer_type(backend), + shared_prefill_max_chunk) + : ggml_gallocr_new( + ggml_backend_get_default_buffer_type(backend)); attn_alloc.owner_ctx = w.ctx; attn_alloc.backend = backend; } - if (!attn_alloc.alloc || !ggml_gallocr_alloc_graph(attn_alloc.alloc, gf)) { + const size_t attn_bytes_before = + heterogeneous_sparse_prefill && attn_alloc.alloc + ? ggml_gallocr_get_buffer_size(attn_alloc.alloc, 0) + : 0; + if (heterogeneous_sparse_prefill && attn_alloc.alloc) { + ggml_gallocr_t sizing = + ggml_gallocr_new_with_max_chunk_size( + ggml_backend_get_default_buffer_type(backend), + shared_prefill_max_chunk); + size_t required_bytes = 0; + ggml_gallocr_reserve_n_size( + sizing, gf, nullptr, nullptr, &required_bytes); + ggml_gallocr_free(sizing); + + if (required_bytes > attn_bytes_before) { + size_t free_bytes = 0; + size_t total_bytes = 0; + ggml_backend_cuda_get_device_memory( + device, &free_bytes, &total_bytes); + (void) total_bytes; + constexpr size_t growth_margin = + 128u * 1024u * 1024u; + const size_t replaceable_bytes = + free_bytes + attn_bytes_before; + if (replaceable_bytes < required_bytes + growth_margin) { + // The shared arena is about to grow after decode + // graphs have occupied the remaining VRAM. Retire + // those reproducible caches before cudaMalloc: a + // failed allocator resize cannot safely be retried + // in place on every HIP runtime. + ggml_backend_synchronize(backend); + if (moe_hybrid && moe_hybrid->cold_backend && + moe_hybrid->cold_backend != backend) { + ggml_backend_synchronize( + moe_hybrid->cold_backend); + } + layer_range_cache.fused_verify_graph_cache.destroy(); + layer_range_cache.fused_capture_graph_cache.destroy(); + // q=1 decode slots are also reproducible, while + // their F16 HC mirrors are required by the active + // prefill and must remain resident. + layer_range_cache.fused_decode_graph_cache.evict_graphs(); + size_t free_after_evict = 0; + ggml_backend_cuda_get_device_memory( + device, &free_after_evict, &total_bytes); + std::fprintf(stderr, + "[deepseek4] evicted fused decode graphs " + "before prefill scratch growth at pos=%d " + "layer=%d required=%.1f MiB current=%.1f MiB " + "free=%.1f->%.1f MiB\n", + kv_start, il, + required_bytes / (1024.0 * 1024.0), + attn_bytes_before / (1024.0 * 1024.0), + free_bytes / (1024.0 * 1024.0), + free_after_evict / (1024.0 * 1024.0)); + } + } + } + const bool attn_allocated = attn_alloc.alloc && + ggml_gallocr_alloc_graph(attn_alloc.alloc, gf); + if (!attn_allocated) { std::fprintf(stderr, "[deepseek4] attn graph alloc failed layer %d\n", il); ggml_free(ctx); return false; } + if (heterogeneous_sparse_prefill) { + const size_t attn_bytes_after = + ggml_gallocr_get_buffer_size(attn_alloc.alloc, 0); + if (attn_bytes_after > attn_bytes_before) { + std::fprintf(stderr, + "[deepseek4] shared prefill scratch grew " + "%.1f->%.1f MiB across %d chunk(s)\n", + attn_bytes_before / (1024.0 * 1024.0), + attn_bytes_after / (1024.0 * 1024.0), + ggml_gallocr_get_buffer_n_chunks( + attn_alloc.alloc, 0)); + } + } if (telemetry) telemetry->attn_build_us += ds4_elapsed_us(attn_build_t0, Ds4TimingClock::now()); if (attn_in_backend) { ggml_backend_tensor_copy(attn_in_backend, inp); diff --git a/server/test/test_qwen35moe_expert_placement.cpp b/server/test/test_qwen35moe_expert_placement.cpp index 459925866..e0ac3f0b7 100644 --- a/server/test/test_qwen35moe_expert_placement.cpp +++ b/server/test/test_qwen35moe_expert_placement.cpp @@ -50,5 +50,40 @@ TEST_CASE(Qwen35MoeExpertPlacementFixture, moe_expert_placement_suite) { REQUIRE(loaded.hot_expert_ids == placement.hot_expert_ids); std::filesystem::remove(tmp); + // Aggregate hit-rate placement can overfeed a highly skewed layer while a + // flat layer remains peer-bound. The critical-path model stops at each + // layer's branch crossover and may deliberately leave spare memory unused. + MoeHybridRoutingStats balance_stats; + balance_stats.n_layer = 2; + balance_stats.n_expert = 4; + balance_stats.n_expert_used = 2; + balance_stats.counts = { + 100, 100, 100, 100, // flat: needs three hot experts + 400, 1, 1, 1, // skewed: one hot expert is sufficient + }; + balance_stats.layer_totals = {400, 403}; + + MoeHybridCriticalPathConfig balance_cfg; + balance_cfg.active_experts = 2; + balance_cfg.main_to_peer_rate = 3.0; + MoeHybridPlacement balanced; + REQUIRE(MoeHybridPlacement::build_critical_path_balanced_from_stats( + balance_stats, + /*layer_expert_bytes=*/{100, 100}, + /*layer_main_fixed_bytes=*/{100, 100}, + /*total_hot_budget_bytes=*/600, + balance_cfg, balanced, &err)); + REQUIRE(balanced.hot_counts == std::vector({3, 1})); + REQUIRE(balanced.total_hot == 4); + REQUIRE(balanced.is_hot(0, 0)); + REQUIRE(balanced.is_hot(0, 1)); + REQUIRE(balanced.is_hot(0, 2)); + REQUIRE(balanced.is_hot(1, 0)); + + balance_cfg.main_to_peer_rate = 0.0; + REQUIRE(!MoeHybridPlacement::build_critical_path_balanced_from_stats( + balance_stats, {100, 100}, {100, 100}, 600, + balance_cfg, balanced, &err)); + std::printf("OK\n"); } diff --git a/server/tests/test_deepseek4_unit.cpp b/server/tests/test_deepseek4_unit.cpp index 2bc8f49ec..179773775 100644 --- a/server/tests/test_deepseek4_unit.cpp +++ b/server/tests/test_deepseek4_unit.cpp @@ -321,6 +321,40 @@ static ggml_context * make_test_context(size_t mem_size = 1u << 20) { return ggml_init(params); } +static void test_chunked_graph_allocator(ggml_backend_t backend) { + std::fprintf(stderr, " test_chunked_graph_allocator ..."); + ggml_context * ctx = make_test_context(); + TEST_ASSERT_MSG(ctx != nullptr, "ggml_init failed"); + if (!ctx) { + std::fprintf(stderr, " FAIL\n"); + return; + } + + // The input and output must coexist. Each fits under the cap, while their + // combined live range does not, so the allocator must create two chunks. + constexpr int64_t n_elements = 96 * 1024; // 384 KiB of F32 + ggml_tensor * input = ggml_new_tensor_1d(ctx, GGML_TYPE_F32, n_elements); + ggml_set_input(input); + ggml_tensor * output = ggml_dup(ctx, input); + ggml_set_output(output); + ggml_cgraph * graph = ggml_new_graph_custom(ctx, 16, false); + ggml_build_forward_expand(graph, output); + + constexpr size_t max_chunk_size = 512u * 1024u; + ggml_gallocr_t alloc = ggml_gallocr_new_with_max_chunk_size( + ggml_backend_get_default_buffer_type(backend), max_chunk_size); + TEST_ASSERT_MSG(alloc != nullptr, "chunked graph allocator creation failed"); + if (alloc) { + TEST_ASSERT_MSG(ggml_gallocr_alloc_graph(alloc, graph), + "chunked graph allocation failed"); + TEST_ASSERT_MSG(ggml_gallocr_get_buffer_n_chunks(alloc, 0) == 2, + "graph allocator did not split the backing buffer"); + ggml_gallocr_free(alloc); + } + ggml_free(ctx); + std::fprintf(stderr, g_failures ? " done\n" : " ok\n"); +} + static void test_dspark_confidence_uses_separate_hidden(ggml_backend_t backend) { std::fprintf(stderr, " test_dspark_confidence_uses_separate_hidden ..."); @@ -1627,6 +1661,19 @@ static void test_safe_compressor_batch_tokens() { std::fprintf(stderr, g_failures ? " done\n" : " ok\n"); } +static void test_hybrid_prefill_chunk_tokens() { + std::fprintf(stderr, " test_hybrid_prefill_chunk_tokens ..."); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens(2048, 0) == 2048); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens(2048, 4096) == 2048); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens(2048, 4097) == 1024); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens(1024, 8192) == 1024); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens(512, 8192) == 512); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens(0, 8192) == 1); + TEST_ASSERT(deepseek4_hybrid_prefill_chunk_tokens( + 2048, 2048, 1024) == 1024); + std::fprintf(stderr, g_failures ? " done\n" : " ok\n"); +} + static void test_dspark_park_all_releases_drafter() { std::fprintf(stderr, " test_dspark_park_all_releases_drafter ..."); @@ -3120,12 +3167,16 @@ static void test_hc_post_strided_split_gpu() { std::vector residual((size_t) hc_dim * n_tokens); std::vector block_out((size_t) n_embd * n_tokens); + std::vector main_block((size_t) n_embd * n_tokens); + std::vector peer_block((size_t) n_embd * n_tokens); std::vector split_storage((size_t) split_stride * n_tokens, -99.0f); for (size_t i = 0; i < residual.size(); ++i) { residual[i] = ((int) (i % 17) - 8) * 0.03125f; } for (size_t i = 0; i < block_out.size(); ++i) { block_out[i] = ((int) (i % 11) - 5) * 0.0625f; + main_block[i] = block_out[i] * 0.25f; + peer_block[i] = block_out[i] - main_block[i]; } for (int token = 0; token < n_tokens; ++token) { float * split = split_storage.data() + (size_t) token * split_stride; @@ -3166,6 +3217,10 @@ static void test_hc_post_strided_split_gpu() { ctx, GGML_TYPE_F32, hc_dim, n_tokens); ggml_tensor * block_t = ggml_new_tensor_2d( ctx, GGML_TYPE_F32, n_embd, n_tokens); + ggml_tensor * main_t = ggml_new_tensor_2d( + ctx, GGML_TYPE_F32, n_embd, n_tokens); + ggml_tensor * peer_t = ggml_new_tensor_2d( + ctx, GGML_TYPE_F32, n_embd, n_tokens); ggml_tensor * split_storage_t = ggml_new_tensor_2d( ctx, GGML_TYPE_F32, split_stride, n_tokens); ggml_tensor * split_t = ggml_view_2d( @@ -3174,10 +3229,14 @@ static void test_hc_post_strided_split_gpu() { TEST_ASSERT(!ggml_is_contiguous(split_t)); ggml_tensor * output_t = ggml_ds4_hc_post( ctx, residual_t, block_t, split_t, n_hc); + ggml_tensor * split_output_t = ggml_ds4_hc_post_split( + ctx, residual_t, main_t, peer_t, split_t, n_hc); ggml_set_output(output_t); + ggml_set_output(split_output_t); ggml_cgraph * graph = ggml_new_graph_custom(ctx, 64, false); ggml_build_forward_expand(graph, output_t); + ggml_build_forward_expand(graph, split_output_t); ggml_gallocr_t alloc = ggml_gallocr_new( ggml_backend_get_default_buffer_type(backend)); const bool allocated = ggml_gallocr_alloc_graph(alloc, graph); @@ -3187,6 +3246,10 @@ static void test_hc_post_strided_split_gpu() { residual.size() * sizeof(float)); ggml_backend_tensor_set(block_t, block_out.data(), 0, block_out.size() * sizeof(float)); + ggml_backend_tensor_set(main_t, main_block.data(), 0, + main_block.size() * sizeof(float)); + ggml_backend_tensor_set(peer_t, peer_block.data(), 0, + peer_block.size() * sizeof(float)); ggml_backend_tensor_set(split_storage_t, split_storage.data(), 0, split_storage.size() * sizeof(float)); const bool computed = @@ -3194,12 +3257,18 @@ static void test_hc_post_strided_split_gpu() { TEST_ASSERT_MSG(computed, "strided HC-post graph compute failed"); if (computed) { std::vector actual(expected.size()); + std::vector split_actual(expected.size()); ggml_backend_tensor_get(output_t, actual.data(), 0, actual.size() * sizeof(float)); + ggml_backend_tensor_get(split_output_t, split_actual.data(), 0, + split_actual.size() * sizeof(float)); for (size_t i = 0; i < actual.size(); ++i) { TEST_ASSERT_MSG(nearly_equal(actual[i], expected[i], 1.0e-6f, 1.0e-6f), "strided HC-post output mismatch"); + TEST_ASSERT_MSG(nearly_equal(split_actual[i], expected[i], + 1.0e-6f, 1.0e-6f), + "batched split HC-post output mismatch"); } } } @@ -3951,6 +4020,7 @@ int main() { } test_compressor_pooling_correctness(backend); + test_chunked_graph_allocator(backend); test_swiglu_ds4_cpu_correctness(backend); test_moe_routing_correctness(backend); test_rmsnorm_correctness(backend); @@ -3978,6 +4048,7 @@ int main() { test_dspark_loader_contract_and_bounds(backend); test_dspark_confidence_uses_separate_hidden(backend); test_safe_compressor_batch_tokens(); + test_hybrid_prefill_chunk_tokens(); test_dspark_park_all_releases_drafter(); test_dspark_raw_ring_rollback_after_wrap(backend); test_snapshot_save_restore();