Repository navigation
Expand file tree
/
Copy pathtandem.patch
More file actions
7948 lines (7605 loc) · 386 KB
/
Copy pathtandem.patch
File metadata and controls
7948 lines (7605 loc) · 386 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
diff --git a/common/arg.cpp b/common/arg.cpp
index e6d6450f4..62d1fdf22 100644
--- a/common/arg.cpp
+++ b/common/arg.cpp
@@ -2786,15 +2786,33 @@ common_params_context common_params_parser_init(common_params & params, llama_ex
}
).set_env("LLAMA_ARG_N_CPU_MOE"));
add_opt(common_arg(
- {"--moe-cache-mib"}, "N",
- "GPU cache size in MiB for the MoE experts kept in the CPU. with multiple GPUs, it is split among them like the layers (--tensor-split) (default: 0, disabled)",
+ {"--moe-cache-mib"}, "N|auto",
+ "GPU cache size in MiB for the MoE experts kept in the CPU. with multiple GPUs, it is split among them like the layers (--tensor-split); "
+ "'auto': the device memory left after the model, the KV cache and the compute buffers, minus --moe-cache-margin-mib "
+ "and the draft model (default: 0, disabled)",
+ [](common_params & params, const std::string & value) {
+ if (value == "auto") {
+ params.moe_cache_size = SIZE_MAX;
+ return;
+ }
+ const int v = std::stoi(value);
+ if (v < 0) {
+ throw std::invalid_argument("invalid value");
+ }
+ params.moe_cache_size = (size_t) v*1024*1024;
+ }
+ ).set_env("LLAMA_ARG_MOE_CACHE_MIB"));
+ add_opt(common_arg(
+ {"--moe-cache-margin-mib"}, "N",
+ string_format("with --moe-cache-mib auto: device memory in MiB left free (default: %d; a draft model's file size is added)",
+ (int) (params.moe_cache_margin/(1024*1024))),
[](common_params & params, int value) {
if (value < 0) {
throw std::invalid_argument("invalid value");
}
- params.moe_cache_size = (size_t) value*1024*1024;
+ params.moe_cache_margin = (size_t) value*1024*1024;
}
- ).set_env("LLAMA_ARG_MOE_CACHE_MIB"));
+ ).set_env("LLAMA_ARG_MOE_CACHE_MARGIN_MIB"));
add_opt(common_arg(
{"-ncffn", "--n-cpu-ffn"}, "N",
"keep the dense FFN weights of the first N layers in the CPU\n"
diff --git a/common/common.cpp b/common/common.cpp
index 221f0897c..4d4d22d0b 100644
--- a/common/common.cpp
+++ b/common/common.cpp
@@ -4,6 +4,7 @@
#include "build-info.h"
#include "common.h"
+#include "../src/llama-prof.h"
#include "../src/llama-ext.h"
@@ -386,7 +387,53 @@ bool parse_cpu_mask(const std::string & mask, bool (&boolmask)[GGML_MAX_N_THREAD
return true;
}
+// tandem.cpp release defaults: the tuned paths that are exact (or differ only in rounding, as any kernel choice) and that
+// the benchmarks use. Applied to the environment when a variable is unset, before anything reads it; VAR=0 turns one off.
+// Model-specific paths (Qwen3.8 GDN/conv/PLE/hc, MTP) only act on their architecture. Not set here: LLAMA_MTP_DRAFT_VOCAB
+// (a draft vocabulary subset whose size depends on the model).
+static const char * const lc_release_defaults[][2] = {
+ { "GGML_CUDA_GRAPH_KEY_SHAPE", "1" }, // one CUDA graph per batch shape (speculative windows do not evict each other)
+ { "GGML_SCHED_BATCH_INPUTS", "3" }, // the graph's host inputs in one packed upload
+ { "GGML_CUDA_MMVF_MAX", "8" }, // vector kernel for float weights up to 8 columns
+ { "GGML_CUDA_Q8_CACHE", "1" }, // reuse the q8_1 activations of a matvec input across matvecs
+ { "LLAMA_GRAPH_POOL", "6" }, // keep 6 more allocated graphs for alternating small batch shapes
+ { "LLAMA_SRV_USER_BREAK", "0" }, // server: no prompt checkpoint at every user message
+ { "LLAMA_MOE_CPU", "auto" }, // MoE cache: the CPU computes a measured share of the missed experts
+ { "LLAMA_MOE_PREFILL_CPU", "auto" }, // ... and of the prompt batches' missed experts
+ { "LLAMA_MOE_CPU_RB", "64" }, // CPU expert job: 64 gate/up rows per work item
+ { "LLAMA_MOE_CPU_DOWN_PE", "64" }, // CPU expert job: down rows per expert, 64 per item
+ { "LLAMA_HC_FUSE", "1" }, // Qwen3.8 / DS4 hc: scale folded into the weights
+ { "GGML_CUDA_HC_MIX_FUSE", "1" }, // hc mix block in 2 kernels (different rounding)
+ { "LLAMA_GDN_DIRECT", "1" }, // Qwen3.8 GDN: state read from the cache by row
+ { "LLAMA_GDN_GATES", "1" }, // Qwen3.8 GDN: beta and alpha in one matvec
+ { "LLAMA_CONV_CONT", "1" }, // Qwen3.8 conv: contiguous token input
+ { "LLAMA_CONV_TAIL", "1" }, // Qwen3.8 conv: rollback slots from the strided tail
+ { "LLAMA_PLE_GPU", "1" }, // per-layer embeddings read by the GPU from pinned host memory
+ { "LLAMA_MTP_DEFER", "1" }, // MTP: catch-up decode deferred to the next draft
+};
+
+static void lc_apply_release_defaults() {
+ if (const char * off = getenv("LLAMA_MOE_NO_DEFAULTS"); off && atoi(off) != 0) {
+ return;
+ }
+ std::string set;
+ for (const auto & d : lc_release_defaults) {
+ if (getenv(d[0]) == nullptr) {
+#if defined(_WIN32)
+ _putenv_s(d[0], d[1]);
+#else
+ setenv(d[0], d[1], 0);
+#endif
+ set += string_format(" %s=%s", d[0], d[1]);
+ }
+ }
+ if (!set.empty()) {
+ LOG_INF("tandem.cpp defaults (unset variables; LLAMA_MOE_NO_DEFAULTS=1 for none):%s\n", set.c_str());
+ }
+}
+
void common_init() {
+ lc_apply_release_defaults();
#if defined(_WIN32)
SetConsoleOutputCP(CP_UTF8);
SetConsoleCP(CP_UTF8);
@@ -1772,7 +1819,38 @@ struct llama_context_params common_context_params_to_llama(const common_params &
cparams.type_k = params.cache_type_k;
cparams.type_v = params.cache_type_v;
- cparams.moe_cache_size = params.moe_cache_size;
+ cparams.moe_cache_size = params.moe_cache_size;
+ cparams.moe_cache_margin = params.moe_cache_margin;
+ if (params.moe_cache_size == SIZE_MAX && params.speculative.has_dft()) {
+ // the draft model is loaded after this context: keep its weights (plus its KV cache and graphs, ~512 MiB) free too
+ const auto & dft = params.speculative.draft;
+ std::error_code ec;
+ size_t sz = std::filesystem::file_size(dft.mparams.path, ec);
+ if (ec) {
+ sz = 0;
+ }
+ // its token_embd stays in host memory (the input layer) unless a draft override (-otd) places it
+ bool embd_override = false;
+ for (const auto & o : dft.tensor_buft_overrides) {
+ if (o.pattern != nullptr && std::regex_search("token_embd.weight", std::regex(o.pattern))) {
+ embd_override = true;
+ }
+ }
+ if (sz > 0 && !embd_override) {
+ gguf_init_params gp = { /* .no_alloc = */ true, /* .ctx = */ nullptr };
+ ggml_context * meta = nullptr;
+ gp.ctx = &meta;
+ gguf_context_ptr g(gguf_init_from_file(dft.mparams.path.c_str(), gp));
+ const ggml_tensor * te = meta ? ggml_get_tensor(meta, "token_embd.weight") : nullptr;
+ if (te != nullptr && ggml_nbytes(te) < sz) {
+ sz -= ggml_nbytes(te);
+ }
+ if (meta) {
+ ggml_free(meta);
+ }
+ }
+ cparams.moe_cache_margin += sz + (size_t) 512*1024*1024;
+ }
return cparams;
}
@@ -2410,6 +2488,38 @@ void common_prompt_checkpoint::update_pos(
this->pos_max = pos_max;
}
+static std::mutex g_ckpt_pool_mu;
+static std::vector<std::vector<uint8_t>> g_ckpt_pool;
+
+void common_prompt_checkpoint_recycle(common_prompt_checkpoint & cur) {
+ static const bool on = !getenv("LLAMA_CKPT_POOL") || atoi(getenv("LLAMA_CKPT_POOL")) != 0;
+ if (!on) {
+ return;
+ }
+ std::lock_guard<std::mutex> lock(g_ckpt_pool_mu);
+ for (auto * v : { &cur.data_tgt, &cur.data_dft }) {
+ if (v->size() >= (1u << 20) && g_ckpt_pool.size() < 8) {
+ g_ckpt_pool.push_back(std::move(*v));
+ v->clear();
+ }
+ }
+}
+
+// a pooled buffer of exactly n bytes (its pages are already in memory), else dst is left as it is
+static void common_ckpt_take(std::vector<uint8_t> & dst, size_t n) {
+ if (dst.size() == n) {
+ return;
+ }
+ std::lock_guard<std::mutex> lock(g_ckpt_pool_mu);
+ for (auto it = g_ckpt_pool.begin(); it != g_ckpt_pool.end(); ++it) {
+ if (it->size() == n) {
+ dst.swap(*it);
+ g_ckpt_pool.erase(it);
+ return;
+ }
+ }
+}
+
void common_prompt_checkpoint::update_tgt(
llama_context * ctx,
llama_seq_id seq_id,
@@ -2420,8 +2530,13 @@ void common_prompt_checkpoint::update_tgt(
const size_t ckpt_size = llama_state_seq_get_size_ext(ctx, seq_id, flags);
+ {
+ llama_prof_scope ps("ckpt.tgt.resize");
+ common_ckpt_take(data_tgt, ckpt_size);
data_tgt.resize(ckpt_size);
+ }
+ llama_prof_scope ps("ckpt.tgt.get");
const size_t n = llama_state_seq_get_data_ext(ctx, data_tgt.data(), ckpt_size, seq_id, flags);
if (n != ckpt_size) {
GGML_ABORT("checkpoint size mismatch: expected %zu, got %zu\n", ckpt_size, n);
@@ -2438,8 +2553,13 @@ void common_prompt_checkpoint::update_dft(
const size_t ckpt_size = llama_state_seq_get_size_ext(ctx, seq_id, flags);
+ {
+ llama_prof_scope ps("ckpt.dft.resize");
+ common_ckpt_take(data_dft, ckpt_size);
data_dft.resize(ckpt_size);
+ }
+ llama_prof_scope ps("ckpt.dft.get");
const size_t n = llama_state_seq_get_data_ext(ctx, data_dft.data(), ckpt_size, seq_id, flags);
if (n != ckpt_size) {
GGML_ABORT("checkpoint size mismatch: expected %zu, got %zu\n", ckpt_size, n);
diff --git a/common/common.h b/common/common.h
index cd89a27e9..5f25a34db 100644
--- a/common/common.h
+++ b/common/common.h
@@ -597,7 +597,8 @@ struct common_params {
ggml_type cache_type_k = GGML_TYPE_F16; // KV cache data type for the K
ggml_type cache_type_v = GGML_TYPE_F16; // KV cache data type for the V
- size_t moe_cache_size = 0; // GPU cache size in bytes for the MoE experts kept in the CPU, split among the GPUs like the layers
+ size_t moe_cache_size = 0; // GPU cache size in bytes for the MoE experts kept in the CPU, split among the GPUs like the layers; SIZE_MAX: auto
+ size_t moe_cache_margin = (size_t) 1024*1024*1024; // auto size: device memory left free (+ the draft model's file size)
common_conversation_mode conversation_mode = COMMON_CONVERSATION_MODE_AUTO;
@@ -1340,3 +1341,7 @@ struct common_prompt_checkpoint {
void clear_tgt();
void clear_dft();
};
+
+// lc-: hands the buffers of a checkpoint about to be erased to a small pool; update_tgt/update_dft take a pooled buffer of the
+// same size instead of a fresh one, which costs ~44 ms per 112 MiB to fault in and zero (asrok)
+void common_prompt_checkpoint_recycle(common_prompt_checkpoint & cur);
diff --git a/common/speculative.cpp b/common/speculative.cpp
index 4305ac6ac..e13b31fa7 100644
--- a/common/speculative.cpp
+++ b/common/speculative.cpp
@@ -1,4 +1,5 @@
#include "speculative.h"
+#include "../src/llama-prof.h"
#include "common.h"
#include "ggml.h"
@@ -350,6 +351,7 @@ struct common_speculative_impl_draft_simple : public common_speculative_impl {
}
void draft(common_speculative_draft_params_vec & dparams) override {
+ llama_prof_lap("lap.d_entry");
auto & ctx_dft = params.ctx_dft;
batch.clear();
@@ -392,6 +394,7 @@ struct common_speculative_impl_draft_simple : public common_speculative_impl {
}
int i = 0;
+ llama_prof_lap("lap.d_setup");
while (n_drafting > 0) {
int i_batch = 0;
@@ -1524,6 +1527,52 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
std::vector<int> i_last;
std::vector<std::vector<float>> chain_h;
+ // LLAMA_MTP_DEFER=1: the catch-up decode of a processed batch waits for the next draft() (or process()), so after
+ // a verify it covers only the accepted tokens and shares one decode with the first draft step
+ struct mtp_stash {
+ std::vector<llama_token> tok;
+ std::vector<llama_pos> pos;
+ std::vector<float> h; // [n][n_embd], the target row each token pairs with
+ };
+ bool defer = false;
+ std::vector<mtp_stash> stash; // per seq
+
+ // decode whatever is stashed (no outputs), e.g. a prompt ubatch followed by another one
+ bool flush_stash() {
+ bool any = false;
+ for (const auto & st : stash) { any = any || !st.tok.empty(); }
+ if (!any) {
+ return true;
+ }
+ std::vector<mtp_stash> hold;
+ hold.swap(stash);
+ stash.assign(n_seq, {});
+ batch.clear();
+ for (llama_seq_id seq_id = 0; seq_id < (llama_seq_id) n_seq; ++seq_id) {
+ const auto & st = hold[seq_id];
+ // a sequence restarted while its catch-up was deferred (the server cleared both memories, then a stale stash was
+ // flushed into the draft's): the newest stash wins, the draft memory from its first position on is dropped
+ if (!st.tok.empty()) {
+ auto * mem = llama_get_memory(params.ctx_dft);
+ if (st.pos[0] <= llama_memory_seq_pos_max(mem, seq_id)) {
+ SPC_WRN("seq %d: deferred batch starts at %d, draft memory ends at %d: dropping the stale part\n",
+ (int) seq_id, (int) st.pos[0], (int) llama_memory_seq_pos_max(mem, seq_id));
+ llama_memory_seq_rm(mem, seq_id, st.pos[0], -1);
+ }
+ }
+ for (size_t k = 0; k < st.tok.size(); ++k) {
+ const int32_t idx = batch.add(st.tok[k], st.pos[k], seq_id, false);
+ batch.set_embd_state(idx, { st.h.data() + k*n_embd, 1, (size_t) n_embd });
+ }
+ }
+ const int32_t rc = llama_process(params.ctx_dft, LLAMA_PROCESS_TYPE_DECODE, batch.get());
+ if (rc != 0) {
+ SPC_ERR("llama_process(ctx_dft) for the deferred batch failed rc=%d\n", (int) rc);
+ }
+ batch.clear();
+ return rc == 0;
+ }
+
common_speculative_impl_draft_mtp(const common_params_speculative & params, uint32_t n_seq)
: common_speculative_impl(COMMON_SPECULATIVE_TYPE_DRAFT_MTP, n_seq, params.draft.n_max)
, params(params.draft)
@@ -1598,6 +1647,12 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
verify_h.assign(n_seq, {});
verify_h_rows.assign(n_seq, 0);
+
+ defer = getenv("LLAMA_MTP_DEFER") != nullptr && atoi(getenv("LLAMA_MTP_DEFER")) != 0 && !is_mem_shared && !chain_heads;
+ stash.assign(n_seq, {});
+ if (defer) {
+ SPC_TRC("%s", "- deferred catch-up decode (LLAMA_MTP_DEFER)\n");
+ }
}
~common_speculative_impl_draft_mtp() override {
@@ -1617,6 +1672,7 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
void begin(llama_seq_id seq_id, const llama_tokens & prompt) override {
// reset here rather than per round, or two identical requests differ
common_sampler_reset(smpls[seq_id].get());
+ flush_stash();
const int32_t N = (int32_t) prompt.size();
if (N <= 0) {
@@ -1666,8 +1722,34 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
const size_t row_bytes = (size_t) n_embd * sizeof(float);
+ const bool use_defer = defer && !batch_in.has_embd();
+ if (use_defer) {
+ // an earlier stash (no draft since) is decoded first, in order
+ {
+ LC_PROF("mtp.proc.flush");
+ if (!flush_stash()) {
+ return false;
+ }
+ }
+ {
+ LC_PROF("mtp.proc.sync_tgt");
+ llama_synchronize(ctx_tgt);
+ }
+ llama_prof_scope lc_prof_p("mtp.proc.h_tgt");
+ const float * h_tgt = llama_get_embeddings_nextn(ctx_tgt);
+ lc_prof_p.~llama_prof_scope(); new (&lc_prof_p) llama_prof_scope("mtp.proc.stash");
+ for (int k = 0; k < n_tokens; ++k) {
+ const auto & t = batch_in.tokens[k];
+ auto & st = stash[t.seq_id];
+ const float * h_row = k == i_batch_beg[t.seq_id] ? pending_h[t.seq_id].data() : h_tgt + (size_t) (k - 1) * n_embd;
+ st.tok.push_back(t.id);
+ st.pos.push_back(t.pos[0]);
+ st.h.insert(st.h.end(), h_row, h_row + n_embd);
+ }
+ }
+
// if kv is shared with target (e.g Gemma4), then we can skip this catch-up decode
- if (!is_mem_shared) {
+ if (!is_mem_shared && !use_defer) {
batch.clear();
// pair each token with the tgt embedding shifted right by one position, and
@@ -1727,6 +1809,7 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
}
}
+ LC_PROF("mtp.proc.verify_h");
for (llama_seq_id seq_id = 0; seq_id < (llama_seq_id) n_seq; ++seq_id) {
if (i_batch_end[seq_id] < 0) {
continue;
@@ -1751,6 +1834,22 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
void draft(common_speculative_draft_params_vec & dparams) override {
auto & ctx_dft = params.ctx_dft;
+ // a stash that does not end right before the draft position, or would not fit one batch together with the
+ // draft tokens, is decoded on its own first
+ size_t n_stash = 0;
+ for (const auto & st : stash) { n_stash += st.tok.size(); }
+ bool flush = n_stash + n_seq > (size_t) llama_n_batch(ctx_dft);
+ for (llama_seq_id seq_id = 0; seq_id < (llama_seq_id) n_seq && !flush; ++seq_id) {
+ const auto & st = stash[seq_id];
+ flush = !st.tok.empty() && (!dparams[seq_id].drafting || st.pos.back() + 1 != dparams[seq_id].pos0);
+ }
+ if (flush) {
+ flush_stash();
+ }
+ std::vector<mtp_stash> hold; // keeps the stashed rows alive through the first decode
+ hold.swap(stash);
+ stash.assign(n_seq, {});
+
batch.clear();
// keep track of which sequences are still drafting
@@ -1781,6 +1880,14 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
common_sampler_reset(smpls[seq_id].get());
}
+ if (!hold.empty()) {
+ const auto & st = hold[seq_id];
+ for (size_t k = 0; k < st.tok.size(); ++k) {
+ const int32_t j = batch.add(st.tok[k], st.pos[k], seq_id, false);
+ batch.set_embd_state(j, { st.h.data() + k*n_embd, 1, (size_t) n_embd });
+ }
+ }
+
const int32_t idx = batch.add(dp.id_last, dp.pos0, seq_id, true);
batch.set_embd_state(idx, { pending_h[seq_id].data(), 1, (size_t) n_embd });
@@ -1791,6 +1898,44 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
}
}
+ // LLAMA_MTP_CHAIN: one greedy sequence, no confidence cut: every step runs on the device back to back, the
+ // first one (catch-up plus the last token) included; the host launches them and reads the tokens once
+ if (n_drafting == 1 && params.p_min <= 0.0f && !chain_heads && !is_mem_shared && llama_mtp_chain_enabled(ctx_dft)) {
+ llama_seq_id sid = 0;
+ while (!drafting[sid]) { sid++; }
+ auto & dp = dparams[sid];
+ if (!dp.result_q) {
+ const int n_steps = std::min<int>(std::min(params.n_max, dp.n_max > 0 ? dp.n_max : params.n_max), 64);
+ llama_mtp_chain_begin(ctx_dft, dp.id_last, pending_h[sid].data());
+ int n_done = 0;
+ for (int k = 0; k < n_steps; ++k) {
+ if (k > 0) {
+ batch.clear();
+ const int32_t idx = batch.add(dp.id_last, dp.pos0 + k, sid, true); // token and h: placeholders
+ batch.set_embd_state(idx, { pending_h[sid].data(), 1, (size_t) n_embd });
+ }
+ LC_PROF("mtp.chain_decode");
+ if (llama_process(ctx_dft, LLAMA_PROCESS_TYPE_DECODE, batch.get()) != 0) {
+ SPC_ERR("chained llama_process[%d] failed\n", k);
+ break;
+ }
+ llama_mtp_chain_record(ctx_dft, k);
+ n_done++;
+ }
+ std::vector<llama_token> toks(n_done);
+ {
+ LC_PROF("mtp.chain_end");
+ llama_mtp_chain_end(ctx_dft, toks.data(), n_done);
+ }
+ dp.result->insert(dp.result->end(), toks.begin(), toks.end());
+ batch.clear();
+ if (dp.result->size() < (size_t) params.n_min) {
+ dp.result->clear();
+ }
+ return;
+ }
+ }
+
int i = 0;
while (n_drafting > 0) {
@@ -1810,7 +1955,12 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
llama_set_nextn_layer_offset(ctx_dft, i);
}
- int ret = llama_process(ctx_dft, LLAMA_PROCESS_TYPE_DECODE, batch.get());
+ int ret;
+ {
+ llama_prof_lap("lap.d_seqrm");
+ LC_PROF("mtp.draft_decode");
+ ret = llama_process(ctx_dft, LLAMA_PROCESS_TYPE_DECODE, batch.get());
+ }
if (ret != 0) {
SPC_ERR("llama_process[%d] returned %d\n", i, ret);
break;
@@ -1828,8 +1978,15 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
auto * smpl = smpls[seq_id].get();
+ {
+ LC_PROF("mtp.sync");
+ llama_synchronize(ctx_dft);
+ }
+ llama_prof_scope lc_prof_s("mtp.sample");
const llama_token id_sampled = common_sampler_sample(smpl, ctx_dft, i_last[seq_id], true);
+ lc_prof_s.~llama_prof_scope(); new (&lc_prof_s) llama_prof_scope("mtp.h_row");
const float * h_row = llama_get_embeddings_nextn_ith(ctx_dft, i_last[seq_id]);
+ lc_prof_s.~llama_prof_scope(); new (&lc_prof_s) llama_prof_scope("mtp.candidates_rest");
const auto * cur_p = common_sampler_get_candidates(smpl, true);
@@ -1843,7 +2000,10 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
const llama_token id = dparams.at(seq_id).result_q ? id_sampled : cur_p->data[0].id;
// only collect very high-confidence draft tokens
- if (cur_p->data[0].p < params.p_min) {
+ // LLAMA_MTP_PMIN_FLOOR=k (hydev): the cut applies only once k drafts exist (Strata always verifies at least
+ // one draft); a round without drafts is a plain 1-token decode
+ static const size_t pmin_floor = getenv("LLAMA_MTP_PMIN_FLOOR") ? (size_t) atoi(getenv("LLAMA_MTP_PMIN_FLOOR")) : 0;
+ if (cur_p->data[0].p < params.p_min && dparams.at(seq_id).result->size() >= pmin_floor) {
drafting[seq_id] = false;
n_drafting--;
@@ -1861,7 +2021,9 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
dp.result_q->emplace_back(cur_p->data, cur_p->data + cur_p->size);
}
- if (params.n_max <= (int) result.size()) {
+ // the request's limit (server: context room, LLAMA_SPEC_AUTO's count) stops the drafting here instead of a
+ // truncation after the full draft
+ if (std::min(params.n_max, dp.n_max > 0 ? dp.n_max : params.n_max) <= (int) result.size()) {
drafting[seq_id] = false;
n_drafting--;
continue;
@@ -1884,6 +2046,34 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
const int32_t idx = batch.add(id, dp.pos0, seq_id, true);
batch.set_embd_state(idx, { h_row, 1, (size_t) n_embd });
i_last[seq_id] = idx;
+ } else if (n_drafting == 1 && !dp.result_q && params.p_min <= 0.0f && llama_mtp_chain_enabled(ctx_dft)) {
+ // LLAMA_MTP_CHAIN: the remaining greedy steps run back to back on the device, each reading the
+ // token and h the previous one wrote; the host only launches them and reads the tokens once
+ const int n_rest = std::min<int>(params.n_max - (int) result.size(), 64);
+ std::vector<llama_token> toks(n_rest);
+ llama_mtp_chain_begin(ctx_dft, id, h_row);
+ int n_done = 0;
+ for (int k = 0; k < n_rest; ++k) {
+ batch.clear();
+ const int32_t idx = batch.add(id, dp.pos0 + i + 1 + k, seq_id, true); // token and h: placeholders
+ batch.set_embd_state(idx, { pending_h[seq_id].data(), 1, (size_t) n_embd });
+ LC_PROF("mtp.chain_decode");
+ if (llama_process(ctx_dft, LLAMA_PROCESS_TYPE_DECODE, batch.get()) != 0) {
+ SPC_ERR("chained llama_process[%d] failed\n", k);
+ break;
+ }
+ llama_mtp_chain_record(ctx_dft, k);
+ n_done++;
+ }
+ {
+ LC_PROF("mtp.chain_end");
+ llama_mtp_chain_end(ctx_dft, toks.data(), n_done);
+ }
+ result.insert(result.end(), toks.begin(), toks.begin() + n_done);
+ batch.clear();
+ drafting[seq_id] = false;
+ n_drafting--;
+ continue;
} else {
const int32_t idx = batch.add(id, dp.pos0 + i + 1, seq_id, true);
batch.set_embd_state(idx, { h_row, 1, (size_t) n_embd });
@@ -1927,6 +2117,15 @@ struct common_speculative_impl_draft_mtp : public common_speculative_impl {
const int32_t i_h = std::min<int32_t>(n_accepted, n_rows - 1);
const size_t row_bytes = (size_t) n_embd * sizeof(float);
std::memcpy(pending_h[seq_id].data(), verify_h[seq_id].data() + (size_t) i_h * n_embd, row_bytes);
+
+ // the deferred catch-up only needs the sampled token and the accepted drafts
+ auto & st = stash[seq_id];
+ if (defer && !st.tok.empty()) {
+ const size_t keep = std::min(st.tok.size(), (size_t) n_accepted + 1);
+ st.tok.resize(keep);
+ st.pos.resize(keep);
+ st.h.resize(keep*n_embd);
+ }
}
};
diff --git a/ggml/include/ggml-backend.h b/ggml/include/ggml-backend.h
index 7a65d8121..f9afcc438 100644
--- a/ggml/include/ggml-backend.h
+++ b/ggml/include/ggml-backend.h
@@ -363,6 +363,8 @@ extern "C" {
GGML_API enum ggml_status ggml_backend_sched_graph_compute(ggml_backend_sched_t sched, struct ggml_cgraph * graph);
GGML_API enum ggml_status ggml_backend_sched_graph_compute_async(ggml_backend_sched_t sched, struct ggml_cgraph * graph);
GGML_API void ggml_backend_sched_synchronize(ggml_backend_sched_t sched);
+ // GGML_SCHED_BATCH_INPUTS=2: the user input copies are not waited for; call this before overwriting the inputs
+ GGML_API void ggml_backend_sched_wait_input_copies(ggml_backend_sched_t sched);
// Reset all assignments and allocators - must be called before changing the node backends or allocating a new graph.
// This in effect deallocates all tensors that were previously allocated and leaves them with dangling pointers.
diff --git a/ggml/include/ggml.h b/ggml/include/ggml.h
index d62a63104..a0f75accd 100644
--- a/ggml/include/ggml.h
+++ b/ggml/include/ggml.h
@@ -602,6 +602,10 @@ extern "C" {
GGML_OP_GLU,
+ GGML_OP_HOST_ADD,
+ GGML_OP_MOE_DOORBELL,
+ GGML_OP_GET_ROWS_HOST,
+
GGML_OP_COUNT,
};
@@ -2678,6 +2682,24 @@ extern "C" {
struct ggml_tensor * state,
int64_t K);
+ // same, with two optional extensions:
+ // ids (I32 [n_seqs]): the initial state of sequence i is states[ids[i]] (states: [S_v, S_v, H_v, n_rows]),
+ // read straight from the recurrent cache instead of a gathered copy
+ // dt, A (F32 [H_v]): g and beta are raw projections (scalar gate only, strided views allowed, same strides);
+ // the op uses g = softplus(g + dt) * A and beta = sigmoid(beta)
+ GGML_API struct ggml_tensor * ggml_gated_delta_net_ext(
+ struct ggml_context * ctx,
+ struct ggml_tensor * q,
+ struct ggml_tensor * k,
+ struct ggml_tensor * v,
+ struct ggml_tensor * g,
+ struct ggml_tensor * beta,
+ struct ggml_tensor * states,
+ struct ggml_tensor * ids,
+ struct ggml_tensor * dt,
+ struct ggml_tensor * A,
+ int64_t K);
+
// DSA lightning indexer
//
// q: [n_embd_idx, n_head_idx, n_batch, ne3 ]
@@ -2731,6 +2753,18 @@ extern "C" {
struct ggml_tensor * gate,
float scale);
+ // hc_pre with the weights as gate logits (DeepSeek V4): weights[h, t] = sigmoid(mix[h, t]*scale[0] + base[h]) + eps,
+ // the same operations as mul -> add -> sigmoid -> scale_bias(1, eps), without contraction into an fma
+ // mix [hc, n_tokens] (may be a view), scale [>= 1], base [hc]
+ //
+ GGML_API struct ggml_tensor * ggml_dsv4_hc_pre_affine(
+ struct ggml_context * ctx,
+ struct ggml_tensor * x,
+ struct ggml_tensor * mix,
+ struct ggml_tensor * scale,
+ struct ggml_tensor * base,
+ float eps);
+
// hc_post: x [n_embd, n_tokens], residual [n_embd, hc, n_tokens],
// post [hc, n_tokens], comb [dst_hc, src_hc, n_tokens]
// -> [n_embd, hc, n_tokens]
@@ -2765,6 +2799,31 @@ extern "C" {
struct ggml_tensor * blocks,
struct ggml_tensor * extra);
+ // as ggml_dsv4_hc_post, with post given as gate logits (DeepSeek V4): the op uses
+ // scale_out * sigmoid(post[h, t]*scale[0] + base[h]), the same operations as mul -> add -> sigmoid -> scale
+ GGML_API struct ggml_tensor * ggml_dsv4_hc_post_affine(
+ struct ggml_context * ctx,
+ struct ggml_tensor * x,
+ struct ggml_tensor * residual,
+ struct ggml_tensor * post,
+ struct ggml_tensor * comb,
+ struct ggml_tensor * scale,
+ struct ggml_tensor * base,
+ float scale_out);
+
+ // as ggml_dsv4_hc_post, with post given as logits: the op uses scale_out * sigmoid(scale_in * post).
+ // add/add_gate (optional, [n_embd, n_tokens] and [1, n_tokens]): x is replaced by x + add * sigmoid(add_gate)
+ GGML_API struct ggml_tensor * ggml_dsv4_hc_post_sig(
+ struct ggml_context * ctx,
+ struct ggml_tensor * x,
+ struct ggml_tensor * residual,
+ struct ggml_tensor * post,
+ struct ggml_tensor * comb,
+ float scale_in,
+ float scale_out,
+ struct ggml_tensor * add,
+ struct ggml_tensor * add_gate);
+
// custom operators
typedef void (*ggml_custom1_op_t)(struct ggml_tensor * dst , const struct ggml_tensor * a, int ith, int nth, void * userdata);
@@ -2870,6 +2929,48 @@ extern "C" {
struct ggml_tensor * v,
struct ggml_tensor * adamw_params); // parameters such as the learning rate
+ // a + data, where data is F32 in pinned host memory with the shape of a (contiguous), written by the host while the graph runs:
+ // the backend waits until flag[0] != flag[1] (the host increments flag[0] after writing data), then sets flag[1] = flag[0] and
+ // flag[2] = how long it waited (ns); flag[3] counts waits that timed out. data and flag must stay valid while the graph is used
+ GGML_API struct ggml_tensor * ggml_host_add(
+ struct ggml_context * ctx,
+ struct ggml_tensor * a,
+ const float * data,
+ uint32_t * flag);
+
+ // MoE cache doorbell: hands the selected experts of a layer to a host thread while the graph runs, and returns their slots
+ // db is a control block in pinned host memory, see struct ggml_moe_db: the backend writes ids (and x, w if not NULL) into it,
+ // sets req = ack + 1 and waits until the host sets ack = req, after it wrote the slots of ids (at ans_off) and the uploads
+ // (at ans_off + slots_cap*4: n_fill, pad, then n_fill ggml_moe_db_fill); the backend copies both to the device, then copies
+ // each upload from host to device memory itself (zero-copy reads)
+ // result: I32 [n_ids + GGML_MOE_DB_SCRATCH(max_fill)], the slots of ids first (the rest is scratch)
+ struct ggml_moe_db {
+ uint32_t req, ack; // written by the backend / by the host
+ uint32_t n_tok, n_used, hyb;
+ uint32_t n_fill; // unused
+ uint32_t wait_ns; // how long the backend waited for ack, last call
+ uint32_t timeouts; // waits that gave up after 2 s
+ uint32_t ids_off, x_off, w_off, ans_off, slots_cap, max_fill; // byte offsets from the block, capacities
+ uint32_t pad[2];
+ };
+ struct ggml_moe_db_fill { uint64_t src, dst, n; }; // host source, device destination, bytes
+ // scratch: from word (n_ids + 1) & ~1: n_fill, pad, then max_fill ggml_moe_db_fill
+ #define GGML_MOE_DB_SCRATCH(max_fill) (3 + 6*(max_fill))
+ GGML_API struct ggml_tensor * ggml_moe_doorbell(
+ struct ggml_context * ctx,
+ struct ggml_tensor * ids, // I32 [n_tok*n_used], contiguous
+ struct ggml_tensor * x, // F32 [n_embd, n_tok] or NULL
+ struct ggml_tensor * w, // F32 [n_used*n_tok] or NULL
+ struct ggml_moe_db * db,
+ int n_used);
+
+ // rows ids of w, read by the backend from host memory it can address (pinned and mapped, e.g. registered), as F32
+ // [w->ne[0], n_ids]; w is only described (type, row size, data pointer), it is not a source, so the op runs where ids is
+ GGML_API struct ggml_tensor * ggml_get_rows_host(
+ struct ggml_context * ctx,
+ const struct ggml_tensor * w,
+ struct ggml_tensor * ids);
+
// stochastic gradient descent step (with weight decay)
GGML_API struct ggml_tensor * ggml_opt_step_sgd(
struct ggml_context * ctx,
diff --git a/ggml/src/ggml-backend.cpp b/ggml/src/ggml-backend.cpp
index 22e5fd9c7..8d335b1b9 100644
--- a/ggml/src/ggml-backend.cpp
+++ b/ggml/src/ggml-backend.cpp
@@ -924,6 +924,9 @@ struct ggml_backend_sched_split {
};
struct ggml_backend_sched {
+ // GGML_SCHED_BATCH_INPUTS=2: per backend, an event recorded after the last user input copies
+ ggml_backend_event_t input_copy_events[GGML_SCHED_MAX_BACKENDS] = {};
+ bool input_copy_pending[GGML_SCHED_MAX_BACKENDS] = {};
bool is_reset; // true if the scheduler has been reset since the last graph split
bool is_alloc;
@@ -1212,6 +1215,19 @@ static void ggml_backend_sched_set_if_supported(ggml_backend_sched_t sched, stru
// assigns backends to ops and splits the graph into subgraphs that can be computed on the same backend
void ggml_backend_sched_split_graph(ggml_backend_sched_t sched, struct ggml_cgraph * graph) {
+ // tandem fix: the context buffer allocation at creation time is unchecked,
+ // so under host RAM pressure it can be NULL, which used to GGML_ABORT here
+ // (crashing the whole server on the first decode after a long prompt).
+ // Retry the allocation now and fail the graph build gracefully instead;
+ // callers report an alloc error and the server stays alive.
+ if (sched->context_buffer == NULL) {
+ sched->context_buffer = (char *) malloc(sched->context_buffer_size);
+ if (sched->context_buffer == NULL) {
+ GGML_LOG_ERROR("%s: failed to allocate %zu bytes of scheduler context (host RAM exhausted)\n", __func__, sched->context_buffer_size);
+ sched->ctx = NULL;
+ return;
+ }
+ }
// reset splits
sched->n_splits = 0;
sched->n_graph_inputs = 0;
@@ -1227,7 +1243,8 @@ void ggml_backend_sched_split_graph(ggml_backend_sched_t sched, struct ggml_cgra
sched->ctx = ggml_init(params);
if (sched->ctx == NULL) {
- GGML_ABORT("%s: failed to initialize context\n", __func__);
+ GGML_LOG_ERROR("%s: failed to initialize context (host RAM exhausted)\n", __func__);
+ return;
}
graph->uid = ggml_graph_next_uid();
@@ -1837,14 +1854,24 @@ static void ggml_backend_sched_copy_input(ggml_backend_sched_t sched, struct ggm
return;
}
+ // without events, a copy callback that takes the input orders its copies on the split backend itself: the host does not wait
+ // for the split backend here, so it can launch the graph of the split while the copies run (GGML_SCHED_CALLBACK_SYNC=1: wait)
+ static const bool callback_sync = getenv("GGML_SCHED_CALLBACK_SYNC") != nullptr;
+ const bool callback = sched->callback_copy != NULL && ggml_backend_sched_is_host_weight(input);
+ const bool no_events = sched->events[split_backend_id][sched->cur_copy] == NULL;
+ if (callback && no_events && !callback_sync &&
+ sched->callback_copy(split_backend, input, input_cpy, &split->graph, sched->callback_copy_user_data)) {
+ return;
+ }
+
// wait for the split backend to finish using the input before overwriting it
- if (sched->events[split_backend_id][sched->cur_copy] != NULL) {
+ if (!no_events) {
ggml_backend_event_wait(split_backend, sched->events[split_backend_id][sched->cur_copy]);
} else {
ggml_backend_synchronize(split_backend);
}
- if (sched->callback_copy != NULL && ggml_backend_sched_is_host_weight(input) &&
+ if (callback && (!no_events || callback_sync) &&
sched->callback_copy(split_backend, input, input_cpy, &split->graph, sched->callback_copy_user_data)) {
return;
}
@@ -1862,6 +1889,7 @@ static void ggml_backend_sched_copy_input(ggml_backend_sched_t sched, struct ggm
}
}
+#include "../../src/llama-prof.h"
static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t sched) {
GGML_ASSERT(sched);
struct ggml_backend_sched_split * splits = sched->splits;
@@ -1872,6 +1900,9 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s
struct ggml_backend_sched_split * split = &splits[split_id];
int split_backend_id = split->backend_id;
ggml_backend_t split_backend = sched->backends[split_backend_id];
+ const bool lc_cpu = split_backend_id == sched->n_backends - 1;
+ const bool lc_big = sched->graph.n_nodes > 2000;
+ llama_prof_scope lc_prof_sp(lc_cpu ? (lc_big ? "sched.inputs.cpu.tgt" : "sched.inputs.cpu.dft") : (lc_big ? "sched.inputs.gpu.tgt" : "sched.inputs.gpu.dft"));
// ensure the previous split's async work has completed before we start
// this split, the allocator may have reused buffer regions across splits
@@ -1883,11 +1914,94 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s
}
}
+ // GGML_SCHED_BATCH_INPUTS=1: the user inputs (graph inputs in host memory) are copied with one wait for the split
+ // backend before and one after, instead of a wait and a blocking copy each; same bytes, fewer host round trips
+ static const bool batch_inputs = getenv("GGML_SCHED_BATCH_INPUTS") != nullptr && atoi(getenv("GGML_SCHED_BATCH_INPUTS")) != 0;
+ static const bool no_wait = batch_inputs && atoi(getenv("GGML_SCHED_BATCH_INPUTS")) >= 2;
+ // mode 3: as mode 2, but the backend packs the copies into one upload when it can (CUDA: one H2D copy and one
+ // scatter kernel instead of a cudaMemcpyAsync per input); the host data is staged at once, so no input wait
+ static const bool packed = batch_inputs && atoi(getenv("GGML_SCHED_BATCH_INPUTS")) == 3;
+ // mode 1: the flagged user inputs; mode 2: every input in host memory (also views of inputs and the results of a
+ // CPU split), which the host writes only after ggml_backend_sched_wait_input_copies
+ auto batched = [&](const struct ggml_tensor * input) {
+ return input->buffer && ggml_backend_buffer_is_host(input->buffer) && !ggml_backend_sched_is_host_weight(input) &&
+ (no_wait || (input->flags & GGML_TENSOR_FLAG_INPUT));
+ };
+ bool user_done = false;
+ if (batch_inputs && split_backend->iface.set_tensor_async != NULL) {
+ int n_user = 0;
+ for (int input_id = 0; input_id < split->n_inputs; input_id++) {
+ struct ggml_tensor * input = split->inputs[input_id];
+ n_user += batched(input);
+ }
+ if (no_wait && sched->input_copy_events[split_backend_id] == NULL) {
+ sched->input_copy_events[split_backend_id] = ggml_backend_event_new(ggml_backend_get_device(split_backend));
+ }
+ typedef bool (*set_packed_t)(ggml_backend_t, int, ggml_tensor * const *, const void * const *, const size_t *);
+ set_packed_t set_packed = nullptr;
+ if (packed && n_user > 1) {
+ ggml_backend_dev_t dev = ggml_backend_get_device(split_backend);
+ ggml_backend_reg_t reg = dev ? ggml_backend_dev_backend_reg(dev) : nullptr;
+ set_packed = reg ? (set_packed_t) ggml_backend_reg_get_proc_address(reg, "ggml_backend_set_tensors_packed") : nullptr;
+ }
+ if (set_packed) {
+ static thread_local std::vector<ggml_tensor *> dst;
+ static thread_local std::vector<const void *> src;
+ static thread_local std::vector<size_t> nb;
+ dst.clear(); src.clear(); nb.clear();
+ for (int input_id = 0; input_id < split->n_inputs; input_id++) {
+ struct ggml_tensor * input = split->inputs[input_id];
+ if (batched(input)) {
+ dst.push_back(tensor_copy(input, split_backend_id, sched->cur_copy));
+ src.push_back(input->data);
+ nb.push_back(ggml_nbytes(input));
+ }
+ }
+ user_done = set_packed(split_backend, (int) dst.size(), dst.data(), src.data(), nb.data());
+ }
+ if (user_done) {
+ // staged by the host already: nothing to wait for before the inputs are written again
+ } else if (n_user > 1 && no_wait && sched->input_copy_events[split_backend_id] != NULL) {
+ // the copies queue on the split backend after its previous work, so they cannot overwrite inputs a
+ // running graph reads; the host waits for the event before it writes the inputs again
+ for (int input_id = 0; input_id < split->n_inputs; input_id++) {
+ struct ggml_tensor * input = split->inputs[input_id];
+ if (batched(input)) {
+ struct ggml_tensor * input_cpy = tensor_copy(input, split_backend_id, sched->cur_copy);
+ ggml_backend_tensor_set_async(split_backend, input_cpy, input->data, 0, ggml_nbytes(input));
+ }
+ }
+ ggml_backend_event_record(sched->input_copy_events[split_backend_id], split_backend);
+ sched->input_copy_pending[split_backend_id] = true;
+ user_done = true;
+ } else if (n_user > 1) {
+ if (sched->events[split_backend_id][sched->cur_copy] != NULL) {
+ ggml_backend_event_synchronize(sched->events[split_backend_id][sched->cur_copy]);
+ } else {
+ ggml_backend_synchronize(split_backend);
+ }
+ for (int input_id = 0; input_id < split->n_inputs; input_id++) {
+ struct ggml_tensor * input = split->inputs[input_id];
+ if (batched(input)) {
+ struct ggml_tensor * input_cpy = tensor_copy(input, split_backend_id, sched->cur_copy);
+ ggml_backend_tensor_set_async(split_backend, input_cpy, input->data, 0, ggml_nbytes(input));
+ }
+ }
+ // the host may overwrite the inputs once this returns
+ ggml_backend_synchronize(split_backend);
+ user_done = true;
+ }
+ }
+
// copy the input tensors to the split backend
// the weights in host memory are copied last, so that the copy callback can read the other inputs of the split
for (int input_id = 0; input_id < split->n_inputs; input_id++) {
- if (!ggml_backend_sched_is_host_weight(split->inputs[input_id])) {
- ggml_backend_sched_copy_input(sched, split, split->inputs[input_id]);
+ struct ggml_tensor * input = split->inputs[input_id];
+ if (user_done && batched(input)) {
+ continue;
+ }
+ if (!ggml_backend_sched_is_host_weight(input)) {
+ ggml_backend_sched_copy_input(sched, split, input);
}
}
for (int input_id = 0; input_id < split->n_inputs; input_id++) {
@@ -1896,6 +2010,8 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s
}
}
+ llama_prof_lap("lap.d_sched_inputs");
+ lc_prof_sp.~llama_prof_scope(); new (&lc_prof_sp) llama_prof_scope(lc_cpu ? (lc_big ? "sched.compute.cpu.tgt" : "sched.compute.cpu.dft") : (lc_big ? "sched.compute.gpu.tgt" : "sched.compute.gpu.dft"));
if (!sched->callback_eval) {
enum ggml_status ec = ggml_backend_graph_compute_async(split_backend, &split->graph);
if (ec != GGML_STATUS_SUCCESS) {
@@ -1978,7 +2094,11 @@ ggml_backend_sched_t ggml_backend_sched_new(
sched->hv_tensor_backend_ids = (int *) malloc(sched->hash_set.size * sizeof(sched->hv_tensor_backend_ids[0]));
sched->hv_tensor_copies = (ggml_tensor **) malloc(sched->hash_set.size * sched->n_backends * sched->n_copies * sizeof(struct ggml_tensor *));
- const size_t ggml_sched_max_splits = graph_size; // at most there is one split for each node in the graph
+ // Tandem (Windows low-RAM fix): the worst-case every-node-is-a-split reservation costs
+// max_splits*GGML_SCHED_MAX_SPLIT_INPUTS*2*sizeof(ggml_tensor) (~20 KB per node!), which is
+// ~487 MB of host RAM for a 24k-node graph (Qwen3.5-MoE). Real splits are bounded by backend
+// boundaries (measured 121 for a 35B MoE with the expert cache), so cap the reservation.
+const size_t ggml_sched_max_splits = graph_size < 4096 ? graph_size : (graph_size/8 > 4096 ? graph_size/8 : 4096);
const size_t nodes_size = graph_size + ggml_sched_max_splits*GGML_SCHED_MAX_SPLIT_INPUTS*2;
sched->node_backend_ids = (int *) calloc(nodes_size, sizeof(sched->node_backend_ids[0]));
sched->leaf_backend_ids = (int *) calloc(nodes_size, sizeof(sched->leaf_backend_ids[0]));
@@ -2081,6 +2201,10 @@ bool ggml_backend_sched_reserve(ggml_backend_sched_t sched, struct ggml_cgraph *
ggml_backend_sched_split_graph(sched, measure_graph);
+ if (sched->ctx == NULL) {
+ return false;
+ }
+
if (!ggml_gallocr_reserve_n(sched->galloc, &sched->graph, sched->node_backend_ids, sched->leaf_backend_ids)) {
return false;
}
@@ -2100,6 +2224,12 @@ bool ggml_backend_sched_alloc_graph(ggml_backend_sched_t sched, struct ggml_cgra
ggml_backend_sched_split_graph(sched, graph);
+ // tandem fix: a failed split leaves sched->ctx NULL; report an alloc
+ // error to the caller instead of running on a stale graph
+ if (sched->ctx == NULL) {
+ return false;
+ }
+
if (!ggml_backend_sched_alloc_splits(sched)) {
return false;
}
@@ -2130,6 +2260,16 @@ enum ggml_status ggml_backend_sched_graph_compute_async(ggml_backend_sched_t sch
return ggml_backend_sched_compute_splits(sched);
}
+void ggml_backend_sched_wait_input_copies(ggml_backend_sched_t sched) {
+ GGML_ASSERT(sched);
+ for (int i = 0; i < sched->n_backends; i++) {
+ if (sched->input_copy_pending[i]) {
+ ggml_backend_event_synchronize(sched->input_copy_events[i]);
+ sched->input_copy_pending[i] = false;
+ }
+ }
+}
+
void ggml_backend_sched_synchronize(ggml_backend_sched_t sched) {
GGML_ASSERT(sched);
for (int i = 0; i < sched->n_backends; i++) {
diff --git a/ggml/src/ggml-cpu/ggml-cpu.c b/ggml/src/ggml-cpu/ggml-cpu.c
index 53d39d249..852fe3c9f 100644
--- a/ggml/src/ggml-cpu/ggml-cpu.c
+++ b/ggml/src/ggml-cpu/ggml-cpu.c
@@ -1742,6 +1742,38 @@ static void ggml_compute_forward_mul_mat_id(
/////////////////////////////////
+// see ggml_host_add
+static void ggml_compute_forward_host_add(struct ggml_tensor * dst) {
+ uint64_t p[2];
+ memcpy(p, dst->op_params, sizeof(p));
+ const float * data = (const float *) (uintptr_t) p[0];
+ volatile uint32_t * flag = (volatile uint32_t *) (uintptr_t) p[1];
+
+ const uint32_t seen = flag[1];
+ const int64_t t0 = ggml_time_us();
+ uint32_t v;
+ while ((v = flag[0]) == seen) {
+ if (ggml_time_us() - t0 > 2000000) {
+ flag[3] = flag[3] + 1;
+ break;
+ }
+ }
+#if defined(_MSC_VER)
+ _ReadBarrier();
+#else
+ __atomic_thread_fence(__ATOMIC_ACQUIRE);
+#endif
+ flag[1] = v;