Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions server/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -741,6 +741,12 @@ if(DFLASH27B_TESTS)
target_include_directories(test_derived_scalars PRIVATE ${DFLASH27B_SRC_INCLUDE_DIRS})
add_test(NAME derived_scalars COMMAND test_derived_scalars)
endif()
if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/test/test_chain_rollback_policy.cpp")
add_executable(test_chain_rollback_policy test/test_chain_rollback_policy.cpp)
target_include_directories(test_chain_rollback_policy PRIVATE
${CMAKE_CURRENT_SOURCE_DIR}/src/common)
add_test(NAME chain_rollback_policy COMMAND test_chain_rollback_policy)
endif()
if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/test/test_kvflash_placement.cpp")
add_executable(test_kvflash_placement test/test_kvflash_placement.cpp)
target_include_directories(test_kvflash_placement PRIVATE ${DFLASH27B_SRC_INCLUDE_DIRS})
Expand Down
38 changes: 38 additions & 0 deletions server/src/common/chain_rollback_policy.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
#pragma once

#include <cstdlib>
#include <cstring>

namespace dflash::common {

struct ChainRollbackPolicy {
bool checkpoint_f32 = false;
int fast_rollback_threshold = 5;
bool diagnostics = false;
};

inline bool env_flag_enabled(const char * name) {
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
const char * value = std::getenv(name);
return value != nullptr && value[0] != '\0' && std::strcmp(value, "0") != 0;
}

inline ChainRollbackPolicy resolve_chain_rollback_policy() {
ChainRollbackPolicy policy;
policy.checkpoint_f32 = env_flag_enabled("DFLASH_SINGLE_CHAIN_CHECKPOINT_F32");
policy.diagnostics = env_flag_enabled("DFLASH_SINGLE_CHAIN_ROLLBACK_DIAG");

// Lower thresholds are valid only with exact F32 checkpoints. This keeps
// the established F16 behavior unchanged when no opt-in flags are set.
if (policy.checkpoint_f32) {
const char * value = std::getenv("DFLASH_FAST_ROLLBACK_THRESHOLD");
if (value != nullptr) {
const int requested = std::atoi(value);
if (requested >= 1 && requested <= 5) {
policy.fast_rollback_threshold = requested;
}
}
}
return policy;
}

} // namespace dflash::common
31 changes: 29 additions & 2 deletions server/src/common/dflash_spec_decode.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
#include <cstdio>
#include <vector>

#include "chain_rollback_policy.h"

namespace dflash::common {

namespace {
Expand Down Expand Up @@ -84,6 +86,13 @@ bool run_dflash_spec_decode(
int n_accept_sum = 0;
int n_hint_proposed = 0;
int n_hint_accepted = 0;
const ChainRollbackPolicy rollback_policy = resolve_chain_rollback_policy();
int rollback_accept_hist[17] = {};
int rollback_fast_low = 0;
int rollback_fast_high = 0;
int rollback_legacy_replay = 0;
int rollback_failed_fallback = 0;


auto t_dec0 = std::chrono::steady_clock::now();
while (n_generated < n_gen) {
Expand Down Expand Up @@ -213,9 +222,10 @@ bool run_dflash_spec_decode(

// ── Commit accepted tokens to KV state ──────────────────────────
// Adaptive: use fast-rollback when acceptance is high enough to benefit.
constexpr int kFastRollbackThreshold = 5;
rollback_accept_hist[std::min(accept_n, 16)]++;
const bool use_fast_rollback =
target.supports_fast_rollback() && (accept_n >= kFastRollbackThreshold);
target.supports_fast_rollback() &&
(accept_n >= rollback_policy.fast_rollback_threshold);

std::vector<int32_t> replay_tok((size_t)commit_n);
for (int i = 0; i < commit_n; i++) {
Expand All @@ -236,15 +246,19 @@ bool run_dflash_spec_decode(
if (target.rollback_to(committed, commit_n)) {
last_tok = target_tok[commit_n - 1];
fast_rolled_back = true;
if (accept_n < 5) rollback_fast_low++;
else rollback_fast_high++;
} else {
// Rollback failed (e.g. CUDA error / unsupported state type).
// The pre-verify snapshot is still valid, so degrade to the
// legacy restore+replay path below instead of aborting.
std::fprintf(stderr, "dflash-spec rollback_to failed; "
"falling back to restore+replay\n");
rollback_failed_fallback++;
}
}
if (!fast_rolled_back) {
rollback_legacy_replay++;
// Legacy path: restore SSM snapshot and replay accepted + bonus tokens.
// (When falling back from fast-rollback, bonus_tok is already -1 and
// replay_tok/commit_n reflect the budget-clamped accepted set.)
Expand Down Expand Up @@ -296,6 +310,19 @@ bool run_dflash_spec_decode(
std::printf("[target-split-dflash] %d draft steps, accepted=%d/%d (%.1f%%), avg commit/step=%.2f\n",
n_draft_steps, n_accept_sum, total_draft_pos, accept_pct,
n_draft_steps > 0 ? (double)n_generated / (double)n_draft_steps : 0.0);
if (rollback_policy.diagnostics) {
std::printf("[chain-rollback-policy] checkpoint=%s threshold=%d fast_low=%d fast_high=%d legacy_replay=%d failed_fallback=%d accept_hist=1:%d,2:%d,3:%d,4:%d,5:%d,6:%d,7:%d,8:%d,9:%d,10:%d,11:%d,12:%d,13:%d,14:%d,15:%d,16+:%d\n",
rollback_policy.checkpoint_f32 ? "F32" : "default",
rollback_policy.fast_rollback_threshold,
rollback_fast_low, rollback_fast_high, rollback_legacy_replay,
rollback_failed_fallback,
rollback_accept_hist[1], rollback_accept_hist[2], rollback_accept_hist[3],
rollback_accept_hist[4], rollback_accept_hist[5], rollback_accept_hist[6],
rollback_accept_hist[7], rollback_accept_hist[8], rollback_accept_hist[9],
rollback_accept_hist[10], rollback_accept_hist[11], rollback_accept_hist[12],
rollback_accept_hist[13], rollback_accept_hist[14], rollback_accept_hist[15],
rollback_accept_hist[16]);
}
if (n_hint_proposed > 0) {
std::printf("[target-split-dflash] hint tokens: %d/%d accepted (%.1f%%)\n",
n_hint_accepted, n_hint_proposed,
Expand Down
6 changes: 4 additions & 2 deletions server/src/internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -654,7 +654,8 @@ ggml_tensor * build_qwen35_layer(
int fa_window = 0,
ggml_tensor * q_tail_capture = nullptr,
int q_tail_start = 0,
ggml_tensor * kv_write_rows = nullptr);
ggml_tensor * kv_write_rows = nullptr,
ggml_tensor * parent_ids = nullptr);

// Overload that also exposes the MoE router selection tensor (if MoE layer).
ggml_tensor * build_qwen35_layer(
Expand All @@ -673,7 +674,8 @@ ggml_tensor * build_qwen35_layer(
ggml_tensor * q_tail_capture,
int q_tail_start,
ggml_tensor ** moe_selected_out,
ggml_tensor * kv_write_rows = nullptr);
ggml_tensor * kv_write_rows = nullptr,
ggml_tensor * parent_ids = nullptr);

QwenLayerPrefnOutputs build_qwen35_layer_prefn(
ggml_context * ctx,
Expand Down
11 changes: 9 additions & 2 deletions server/src/qwen35/graph_builders.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@ bool build_layer_step(
bool capture,
int fa_window,
int kq_stride_pad,
bool kvflash) {
bool kvflash,
bool tree_mode) {
if (kvflash) with_mask = true;
step_graph_free(sg);

Expand Down Expand Up @@ -74,14 +75,20 @@ bool build_layer_step(
}
}

if (tree_mode && !is_attn) {
sg.parent_ids = ggml_new_tensor_1d(sg.ctx, GGML_TYPE_I32, n_tokens);
ggml_set_name(sg.parent_ids, "parent_ids");
ggml_set_input(sg.parent_ids);
}

sg.gf = ggml_new_graph_custom(sg.ctx, 16384, false);

ggml_tensor * layer_out = build_qwen35_layer(
sg.ctx, sg.gf, w, cache, layer_idx,
sg.inp_embed, sg.positions, sg.attn_mask,
kv_start, n_tokens, capture, fa_window,
/*q_tail_capture=*/nullptr, /*q_tail_start=*/0,
sg.kv_write_rows);
sg.kv_write_rows, sg.parent_ids);
if (!layer_out) return false;

ggml_tensor * out_view = ggml_view_2d(sg.ctx, act_out,
Expand Down
3 changes: 2 additions & 1 deletion server/src/qwen35/graph_builders.h
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ bool build_layer_step(
bool capture,
int fa_window = 0,
int kq_stride_pad = KQ_MASK_PAD,
bool kvflash = false);
bool kvflash = false,
bool tree_mode = false);

// `kvflash`: pooled mode — KV rows go through a set_rows input
// (sg.kv_write_rows, [n_tokens, n_head_kv] ne0-major slots) and the mask
Expand Down
35 changes: 31 additions & 4 deletions server/src/qwen35/qwen35_backend.cpp
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
#include "qwen35_backend.h"
#include "common/chain_rollback_policy.h"
#include "placement/skip_park_guard.h"
#include "qwen35_dflash_target.h"
#include "graph_builders.h"
Expand Down Expand Up @@ -1927,12 +1928,33 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen,
int n_hint_proposed = 0;
int n_hint_accepted = 0;
int target_forwards = 0;
const ChainRollbackPolicy rollback_policy = resolve_chain_rollback_policy();
int rollback_accept_hist[17] = {};

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: The rollback diagnostic counters and print logic are duplicated across both decode loops (qwen35_backend.cpp and dflash_spec_decode.cpp): five counter variables, the accept_n histogram indexing, three counter-update sites, and the diagnostics print block. Approximately 26 lines are verbatim copies. This makes the two paths prone to drift when someone updates one side but forgets the other. Consider extracting the counters and the print block into a small RAII accumulator struct (e.g. RollbackDiag) shared from chain_rollback_policy.h, then use it from both decode loops. The struct could hold the counters, expose record_fast_rollback(accept_n), record_legacy_replay(), record_fallback(), and a print() method — keeping the two decode loops focused on decode logic rather than diagnostic bookkeeping.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At server/src/qwen35/qwen35_backend.cpp, line 1932:

<comment>The rollback diagnostic counters and print logic are duplicated across both decode loops (`qwen35_backend.cpp` and `dflash_spec_decode.cpp`): five counter variables, the accept_n histogram indexing, three counter-update sites, and the diagnostics print block. Approximately 26 lines are verbatim copies. This makes the two paths prone to drift when someone updates one side but forgets the other. Consider extracting the counters and the print block into a small RAII accumulator struct (e.g. `RollbackDiag`) shared from `chain_rollback_policy.h`, then use it from both decode loops. The struct could hold the counters, expose `record_fast_rollback(accept_n)`, `record_legacy_replay()`, `record_fallback()`, and a `print()` method — keeping the two decode loops focused on decode logic rather than diagnostic bookkeeping.</comment>

<file context>
@@ -1927,12 +1928,33 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen,
     int n_hint_accepted = 0;
     int target_forwards = 0;
+    const ChainRollbackPolicy rollback_policy = resolve_chain_rollback_policy();
+    int rollback_accept_hist[17] = {};
+    int rollback_fast_low = 0;
+    int rollback_fast_high = 0;
</file context>

int rollback_fast_low = 0;
int rollback_fast_high = 0;
int rollback_legacy_replay = 0;
int rollback_failed_fallback = 0;


auto log_target_forward_stats = [&]() {
std::fprintf(stderr, "[spec-decode] target_forwards=%d forwards_per_token=%.6f forwards_per_step=%.3f\n",
target_forwards,
n_generated > 0 ? (double)target_forwards / n_generated : 0.0,
n_draft_steps > 0 ? (double)target_forwards / n_draft_steps : 0.0);
if (rollback_policy.diagnostics) {
std::fprintf(stderr,
"[chain-rollback-policy] checkpoint=%s threshold=%d fast_low=%d fast_high=%d legacy_replay=%d failed_fallback=%d accept_hist=1:%d,2:%d,3:%d,4:%d,5:%d,6:%d,7:%d,8:%d,9:%d,10:%d,11:%d,12:%d,13:%d,14:%d,15:%d,16+:%d\n",
rollback_policy.checkpoint_f32 ? "F32" : "default",
rollback_policy.fast_rollback_threshold,
rollback_fast_low, rollback_fast_high, rollback_legacy_replay,
rollback_failed_fallback,
rollback_accept_hist[1], rollback_accept_hist[2], rollback_accept_hist[3],
rollback_accept_hist[4], rollback_accept_hist[5], rollback_accept_hist[6],
rollback_accept_hist[7], rollback_accept_hist[8], rollback_accept_hist[9],
rollback_accept_hist[10], rollback_accept_hist[11], rollback_accept_hist[12],
rollback_accept_hist[13], rollback_accept_hist[14], rollback_accept_hist[15],
rollback_accept_hist[16]);
}
};

// kvflash: an in-pool prompt prefills contiguously without registering
Expand Down Expand Up @@ -2488,11 +2510,12 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen,
// 6. Fix state: adaptive fast-rollback vs legacy replay.
// Fast-rollback (implicit bonus, skip replay) is profitable when
// accept_n is large enough that skipping the replay saves more compute
// than the cost of deferring the bonus to the next step. Breakeven
// is around accept_n ≈ 5. Below that, legacy replay is cheaper.
constexpr int kFastRollbackThreshold = 5;
// than the cost of deferring the bonus to the next step. The default
// threshold is 5; exact F32 checkpoints may opt in to a lower value.
rollback_accept_hist[std::min(accept_n, 16)]++;
const bool use_fast_rollback =
target->supports_fast_rollback() && (accept_n >= kFastRollbackThreshold);
target->supports_fast_rollback() &&
(accept_n >= rollback_policy.fast_rollback_threshold);

int replay_last_tok = -1;
bool fast_rolled_back = false;
Expand All @@ -2508,15 +2531,19 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen,
if (target->rollback_to(committed, commit_n)) {
replay_last_tok = target_tok[commit_n - 1];
fast_rolled_back = true;
if (accept_n < 5) rollback_fast_low++;
else rollback_fast_high++;
} else {
// Rollback failed (CUDA error / unsupported state type). The
// pre-verify snapshot is still valid, so degrade to the legacy
// restore+replay path below instead of aborting the request.
std::fprintf(stderr, "spec-decode: rollback_to failed; "
"falling back to restore+replay\n");
rollback_failed_fallback++;
}
}
if (!fast_rolled_back) {
rollback_legacy_replay++;
// Legacy replay: restore SSM snapshot, replay accepted + bonus tokens.
// (When falling back from fast-rollback, bonus_tok is -1 and commit_n
// is the budget-clamped accepted count.)
Expand Down
54 changes: 39 additions & 15 deletions server/src/qwen35/qwen35_dflash_target.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -376,13 +376,24 @@ bool Qwen35DFlashTarget::rollback_to_tree(
(size_t)rollback_dfs * cap.ssm_intermediate_states->nb[3];
const void * ssm_src =
(const char *)cap.ssm_intermediate_states->data + ssm_src_offset;
const auto to_fp32 = ggml_get_to_fp32_cuda(cap.ssm_intermediate_states->type);
if (!to_fp32) {
std::fprintf(stderr, "rollback_to_tree: no fp32 converter for type %d (layer %d)\n",
(int)cap.ssm_intermediate_states->type, il);
return false;
if (cap.ssm_intermediate_states->type == GGML_TYPE_F32) {
const cudaError_t ce = cudaMemcpyAsync(cache_.ssm_state[il]->data, ssm_src,
ssm_elems * sizeof(float),
cudaMemcpyDeviceToDevice, stream);
if (ce != cudaSuccess) {
std::fprintf(stderr, "rollback_to_tree: F32 SSM copy failed at layer %d: %s\n",
il, cudaGetErrorString(ce));
return false;
}
} else {
const auto to_fp32 = ggml_get_to_fp32_cuda(cap.ssm_intermediate_states->type);
if (!to_fp32) {
std::fprintf(stderr, "rollback_to_tree: no fp32 converter for type %d (layer %d)\n",
(int)cap.ssm_intermediate_states->type, il);
return false;
}
to_fp32(ssm_src, (float *)cache_.ssm_state[il]->data, (int64_t)ssm_elems, stream);
}
to_fp32(ssm_src, (float *)cache_.ssm_state[il]->data, (int64_t)ssm_elems, stream);

// Conv state ← the K-1 most recent inputs along rollback_dfs's ancestry.
const int K_conv = 4;
Expand Down Expand Up @@ -551,7 +562,6 @@ bool Qwen35DFlashTarget::rollback_to(int base_pos, int commit_n) {
cache_.cur_pos = base_pos + commit_n;
return true;
}

const int rollback_idx = commit_n - 1; // index into per-step intermediates
cudaStream_t stream = nullptr;

Expand Down Expand Up @@ -589,16 +599,30 @@ bool Qwen35DFlashTarget::rollback_to(int base_pos, int commit_n) {
(size_t)rollback_idx * cap.ssm_intermediate_states->nb[3];
const void * ssm_src =
(const char *)cap.ssm_intermediate_states->data + ssm_src_offset;
const auto to_fp32 = ggml_get_to_fp32_cuda(cap.ssm_intermediate_states->type);
if (!to_fp32) {
if (kFastRollbackDiag) {
std::fprintf(stderr, "rollback_to: no fp32 converter type=%d layer=%d\n",
(int)cap.ssm_intermediate_states->type, il);
if (cap.ssm_intermediate_states->type == GGML_TYPE_F32) {
const size_t ssm_bytes = ssm_elems * sizeof(float);
const cudaError_t ce = cudaMemcpyAsync(cache_.ssm_state[il]->data, ssm_src,
ssm_bytes,
cudaMemcpyDeviceToDevice, stream);
if (ce != cudaSuccess) {
if (kFastRollbackDiag) {
std::fprintf(stderr, "rollback_to: F32 SSM copy failed layer=%d: %s\n",
il, cudaGetErrorString(ce));
}
return false;
}
return false;
} else {
const auto to_fp32 = ggml_get_to_fp32_cuda(cap.ssm_intermediate_states->type);
if (!to_fp32) {
if (kFastRollbackDiag) {
std::fprintf(stderr, "rollback_to: no fp32 converter type=%d layer=%d\n",
(int)cap.ssm_intermediate_states->type, il);
}
return false;
}
to_fp32(ssm_src, (float *)cache_.ssm_state[il]->data,
(int64_t)ssm_elems, stream);
}
to_fp32(ssm_src, (float *)cache_.ssm_state[il]->data,
(int64_t)ssm_elems, stream);

// Conv rollback: copy conv_input[commit_n..commit_n+K-2, :, :]
// into cache.conv_state[il].
Expand Down
Loading