From 609e123c2624a73bc480d6334bfb6647763f54dc Mon Sep 17 00:00:00 2001 From: mrciffa <49000955+davide221@users.noreply.github.com> Date: Fri, 10 Jul 2026 18:59:55 +0200 Subject: [PATCH] feat(ipc): add DeepSeek4 DSpark shared worker --- server/CMakeLists.txt | 1 + server/docs/DS4.md | 32 +- server/src/common/backend_ipc.cpp | 5 + server/src/common/backend_ipc.h | 1 + server/src/common/dflash_draft_ipc.cpp | 142 +++++-- server/src/common/dflash_draft_ipc.h | 19 +- server/src/deepseek4/deepseek4_backend.cpp | 41 +- server/src/deepseek4/deepseek4_backend.h | 2 + server/src/deepseek4/deepseek4_dspark.h | 14 +- .../deepseek4_dspark_draft_ipc_daemon.cpp | 354 ++++++++++++++++++ .../src/deepseek4/deepseek4_dspark_spec.cpp | 40 +- server/src/deepseek4/deepseek4_graph.cpp | 4 + server/src/ipc/backend_ipc_main.cpp | 9 + server/test/test_server_unit.cpp | 5 + 14 files changed, 634 insertions(+), 35 deletions(-) create mode 100644 server/src/deepseek4/deepseek4_dspark_draft_ipc_daemon.cpp diff --git a/server/CMakeLists.txt b/server/CMakeLists.txt index 687938351..68cdd238d 100644 --- a/server/CMakeLists.txt +++ b/server/CMakeLists.txt @@ -260,6 +260,7 @@ add_library(dflash_common STATIC src/deepseek4/deepseek4_target_shard_ipc_daemon.cpp src/deepseek4/deepseek4_dspark.cpp src/deepseek4/deepseek4_dspark_spec.cpp + src/deepseek4/deepseek4_dspark_draft_ipc_daemon.cpp src/flashprefill_q8.cpp src/kv_cache.cpp src/kv_quant.cpp diff --git a/server/docs/DS4.md b/server/docs/DS4.md index b4c259d8e..57b227e69 100644 --- a/server/docs/DS4.md +++ b/server/docs/DS4.md @@ -1,6 +1,6 @@ # DeepSeek V4 Flash — DFlash Integration -This document describes the current DeepSeek V4 Flash implementation in DFlash. Today DeepSeek4 runs through the layer-split path: a local CUDA prefix shard plus either a local follow-on shard or a remote Halo/HIP target shard. +This document describes the current DeepSeek V4 Flash implementation in DFlash. DeepSeek4 supports the layer-split target path and an opt-in DSpark speculative path. The DSpark proposal blocks can run in a separate HIP process while the complete target remains local. ## Model Architecture @@ -28,6 +28,7 @@ DeepSeek V4 Flash is a 43-layer MoE model with: | Model weights and metadata | `src/deepseek4/deepseek4_internal.h`, `src/deepseek4/deepseek4_loader.cpp` | | HC pre/post CUDA kernel | `src/deepseek4/deepseek4_hc_cuda.cu`, `.h` | | Remote target-shard daemon | `src/deepseek4/deepseek4_target_shard_ipc_daemon.cpp` | +| DSpark runtime and remote draft daemon | `src/deepseek4/deepseek4_dspark*.{h,cpp}` | | Shared target-shard IPC infrastructure | `src/common/target_shard_ipc.*`, `src/placement/remote_target_shard_config.h` | | Backend IPC CLI entry | `src/ipc/backend_ipc_main.cpp` | @@ -75,6 +76,28 @@ For heterogeneous setups, the CUDA-built server can keep the prefix layers on th This path uses `TargetShardIpcSession`, `deepseek4_target_shard_ipc_daemon.cpp`, and `BackendIpcMode::DeepSeek4TargetShard` rather than the old expert-worker protocol. +### Local target + remote HIP DSpark draft + +The speculative IPC mode keeps the complete DeepSeek4 target, KV cache, tied +embedding, DSpark head, and sampler in the parent. A separate +`BackendIpcMode::DeepSeek4DSparkDraft` process executes the DSpark proposal +graph on the selected HIP GPU. The parent currently retains its locally loaded +draft weights for metadata and fallback; removing that duplicate residency is +a separate optimization. + +Each speculative step transfers: + +- target feature captures in token-major + `[n_tokens, n_target_layers, n_embd]` layout; +- the embedded seed/MASK proposal block; +- the proposal hidden states returned by the HIP worker. + +All target-layer captures are uploaded as one feature block and one +acknowledgement. On POSIX hosts this mode defaults to the existing memfd-backed +shared payload transport; proposal input and output reuse the same mapping. +`stream` remains available as an A/B and compatibility fallback. This is shared +host IPC, not direct `hipIpcMemHandle` peer memory. + ## Shard Boundary State The shard boundary transfers the **full HC state tensor**, not a separate expert-routing payload: @@ -102,6 +125,13 @@ The runtime logs the chosen split with a `[deepseek4-split] auto-split:` banner. |----------|---------| | `DFLASH_DS4_CUDA_LAYERS` | Override the auto-split heuristic and pin the first `N` DeepSeek4 layers to CUDA. The remaining `43 - N` layers run on the Halo shard. | | `DFLASH_DS4_TIMING` | Enable DS4 timing logs for the layer-split parent and target-shard daemon. Useful for profiling prefill/decode breakdowns; leave unset for normal runs. | +| `DFLASH_DS4_SPEC` | Enable the DeepSeek4 DSpark speculative runtime. | +| `DFLASH_DS4_DRAFT` | DSpark draft GGUF loaded locally for metadata/head use and by the remote worker. | +| `DFLASH_DS4_DRAFT_IPC_BIN` | Backend IPC daemon executable used for the remote DSpark worker. | +| `DFLASH_DS4_DRAFT_IPC_GPU` | HIP device index for the remote DSpark proposal blocks. | +| `DFLASH_DS4_DRAFT_IPC_WORK_DIR` | Scratch directory for the child process. | +| `DFLASH_DS4_DRAFT_IPC_REQUIRED` | Fail initialization instead of falling back to a local draft when remote startup fails. | +| `DFLASH_DRAFT_IPC_TRANSPORT` | Payload transport: `auto`, `shared`, or `stream`. DeepSeek4 DSpark defaults to `auto`; other draft modes retain their existing default. | `DFLASH_DS4_TIMING` enables the existing timing banners: diff --git a/server/src/common/backend_ipc.cpp b/server/src/common/backend_ipc.cpp index 4c99daae5..c52dd591d 100644 --- a/server/src/common/backend_ipc.cpp +++ b/server/src/common/backend_ipc.cpp @@ -34,6 +34,7 @@ const char * backend_ipc_mode_name(BackendIpcMode mode) { case BackendIpcMode::LagunaTargetShard: return "laguna-target-shard"; case BackendIpcMode::MoeExpertCompute: return "moe-expert-compute"; case BackendIpcMode::DeepSeek4TargetShard: return "deepseek4-target-shard"; + case BackendIpcMode::DeepSeek4DSparkDraft: return "deepseek4-dspark-draft"; } return "unknown"; } @@ -67,6 +68,10 @@ bool parse_backend_ipc_mode(const std::string & value, BackendIpcMode & out) { out = BackendIpcMode::DeepSeek4TargetShard; return true; } + if (value == "deepseek4-dspark-draft") { + out = BackendIpcMode::DeepSeek4DSparkDraft; + return true; + } return false; } diff --git a/server/src/common/backend_ipc.h b/server/src/common/backend_ipc.h index e9ce21b3c..47a8a5c08 100644 --- a/server/src/common/backend_ipc.h +++ b/server/src/common/backend_ipc.h @@ -29,6 +29,7 @@ enum class BackendIpcMode { LagunaTargetShard, MoeExpertCompute, DeepSeek4TargetShard, + DeepSeek4DSparkDraft, }; const char * backend_ipc_mode_name(BackendIpcMode mode); diff --git a/server/src/common/dflash_draft_ipc.cpp b/server/src/common/dflash_draft_ipc.cpp index 387be58b1..ff6cf4f66 100644 --- a/server/src/common/dflash_draft_ipc.cpp +++ b/server/src/common/dflash_draft_ipc.cpp @@ -25,14 +25,15 @@ namespace dflash::common { namespace { -BackendIpcPayloadTransport draft_ipc_transport_from_env() { +BackendIpcPayloadTransport draft_ipc_transport_from_env( + BackendIpcPayloadTransport fallback) { const char * raw = std::getenv("DFLASH_DRAFT_IPC_TRANSPORT"); if (!raw || !*raw) { - return BackendIpcPayloadTransport::Stream; + return fallback; } - BackendIpcPayloadTransport transport = BackendIpcPayloadTransport::Stream; + BackendIpcPayloadTransport transport = fallback; if (!parse_backend_ipc_payload_transport(raw, transport)) { - return BackendIpcPayloadTransport::Stream; + return fallback; } return transport; } @@ -47,16 +48,23 @@ bool checked_mul_size(size_t a, size_t b, size_t & out) { size_t dflash_draft_ipc_required_shared_bytes(int hidden_size, int block_size, - int ring_cap) { - if (hidden_size <= 0 || block_size <= 0 || ring_cap <= 0) { + int ring_cap, + int n_target_layers) { + if (hidden_size <= 0 || block_size <= 0 || ring_cap <= 0 || + n_target_layers <= 0) { return 0; } - const size_t max_tokens = - (size_t)std::max(block_size, ring_cap); - size_t elements = 0; + size_t noise_elements = 0; + size_t feature_elements = 0; size_t bytes = 0; - if (!checked_mul_size(max_tokens, (size_t)hidden_size, elements) || - !checked_mul_size(elements, sizeof(float), bytes)) { + if (!checked_mul_size((size_t)block_size, (size_t)hidden_size, + noise_elements) || + !checked_mul_size((size_t)ring_cap, (size_t)n_target_layers, + feature_elements) || + !checked_mul_size(feature_elements, (size_t)hidden_size, + feature_elements) || + !checked_mul_size(std::max(noise_elements, feature_elements), + sizeof(float), bytes)) { return 0; } return bytes; @@ -100,22 +108,32 @@ bool DFlashDraftIpcClient::start( const std::string & draft_path, int draft_gpu, int ring_cap, - const std::string & work_dir) { + const std::string & work_dir, + BackendIpcMode mode) { #if defined(_WIN32) - (void)bin; (void)draft_path; (void)draft_gpu; (void)ring_cap; (void)work_dir; + (void)bin; (void)draft_path; (void)draft_gpu; (void)ring_cap; (void)work_dir; (void)mode; std::fprintf(stderr, "DFlash draft IPC is only implemented on POSIX hosts\n"); return false; #else close(); - if (bin.empty() || draft_path.empty() || ring_cap <= 0) return false; + if (bin.empty() || draft_path.empty() || ring_cap <= 0 || + (mode != BackendIpcMode::DFlashDraft && + mode != BackendIpcMode::DeepSeek4DSparkDraft)) return false; BackendIpcLaunchConfig launch; launch.bin = bin; - launch.mode = BackendIpcMode::DFlashDraft; + launch.mode = mode; launch.payload_path = draft_path; launch.work_dir = work_dir; - launch.payload_transport = draft_ipc_transport_from_env(); + const BackendIpcPayloadTransport default_transport = + mode == BackendIpcMode::DeepSeek4DSparkDraft + ? BackendIpcPayloadTransport::Auto + : BackendIpcPayloadTransport::Stream; + launch.payload_transport = draft_ipc_transport_from_env(default_transport); + const int shared_feature_layers = + mode == BackendIpcMode::DeepSeek4DSparkDraft ? n_target_layers_ : 1; launch.shared_payload_bytes = draft_ipc_shared_bytes_from_env( - dflash_draft_ipc_required_shared_bytes(hidden_size_, block_size_, ring_cap)); + dflash_draft_ipc_required_shared_bytes( + hidden_size_, block_size_, ring_cap, shared_feature_layers)); launch.args.push_back("--ring-cap=" + std::to_string(ring_cap)); launch.args.push_back("--draft-gpu=" + std::to_string(std::max(0, draft_gpu))); if (!process_.start(launch)) { @@ -123,13 +141,77 @@ bool DFlashDraftIpcClient::start( return false; } ring_cap_ = ring_cap; + mode_ = mode; active_ = true; - std::printf("[draft-ipc] ready bin=%s gpu=%d ring_cap=%d work_dir=%s\n", - bin.c_str(), draft_gpu, ring_cap, process_.work_dir().c_str()); + std::printf("[draft-ipc] ready mode=%s bin=%s gpu=%d ring_cap=%d work_dir=%s\n", + backend_ipc_mode_name(mode), bin.c_str(), draft_gpu, ring_cap, + process_.work_dir().c_str()); return true; #endif } +bool DFlashDraftIpcClient::send_feature_block( + int start_pos, + int n_tokens, + const float * features, + size_t feature_count) { +#if defined(_WIN32) + (void)start_pos; (void)n_tokens; (void)features; (void)feature_count; + return false; +#else + FILE * cmd = process_.command_stream(); + const int stream_fd = process_.stream_fd(); + const int payload_fd = process_.payload_fd(); + if (!active_ || mode_ != BackendIpcMode::DeepSeek4DSparkDraft || + !cmd || stream_fd < 0 || start_pos < 0 || n_tokens <= 0 || + n_tokens > ring_cap_) { + return false; + } + const size_t expected = (size_t)n_tokens * (size_t)n_target_layers_ * + (size_t)hidden_size_; + if (!features || feature_count != expected) return false; + const size_t bytes = feature_count * sizeof(float); + + if (process_.resolved_payload_transport() == BackendIpcPayloadTransport::Shared) { + uint64_t seq = 0; + if (!process_.write_shared_payload(features, bytes, seq)) { + std::fprintf(stderr, + "draft-ipc feature_block shared payload too large bytes=%zu capacity=%zu\n", + bytes, process_.shared_payload_capacity()); + return false; + } + std::fprintf(cmd, "feature_block_shared %d %d %zu %" PRIu64 "\n", + start_pos, n_tokens, bytes, seq); + std::fflush(cmd); + int32_t status = -1; + const bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) && + status == 0; + if (!ok) { + std::fprintf(stderr, + "draft-ipc feature_block_shared failed status=%d\n", + status); + } + return ok; + } + + if (payload_fd < 0) return false; + std::fprintf(cmd, "feature_block_pipe %d %d %zu\n", + start_pos, n_tokens, bytes); + std::fflush(cmd); + if (!write_exact_fd(payload_fd, features, bytes)) { + std::fprintf(stderr, "draft-ipc feature_block payload write failed\n"); + return false; + } + int32_t status = -1; + const bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) && + status == 0; + if (!ok) { + std::fprintf(stderr, "draft-ipc feature_block failed status=%d\n", status); + } + return ok; +#endif +} + bool DFlashDraftIpcClient::send_feature_slice( int capture_idx, int start_pos, @@ -216,11 +298,19 @@ bool DFlashDraftIpcClient::propose( const int payload_fd = process_.payload_fd(); if (!active_ || !cmd || stream_fd < 0 || committed < 0 || ctx_len <= 0 || ctx_len > ring_cap_) { + std::fprintf(stderr, + "draft-ipc propose rejected active=%d cmd=%p stream_fd=%d committed=%d ctx_len=%d ring_cap=%d\n", + (int)active_, (void *)cmd, stream_fd, committed, ctx_len, ring_cap_); return false; } const size_t noise_expected = (size_t)hidden_size_ * block_size_; - if (noise_embed.size() != noise_expected) return false; + if (noise_embed.size() != noise_expected) { + std::fprintf(stderr, + "draft-ipc propose noise size mismatch got=%zu expected=%zu hidden=%d block=%d\n", + noise_embed.size(), noise_expected, hidden_size_, block_size_); + return false; + } const size_t bytes = noise_embed.size() * sizeof(float); if (process_.resolved_payload_transport() == BackendIpcPayloadTransport::Shared) { uint64_t seq = 0; @@ -230,15 +320,20 @@ bool DFlashDraftIpcClient::propose( bytes, process_.shared_payload_capacity()); return false; } - std::fprintf(cmd, "propose_shared %d %d %zu %" PRIu64 "\n", + const bool bidirectional = + mode_ == BackendIpcMode::DeepSeek4DSparkDraft; + std::fprintf(cmd, bidirectional + ? "propose_shared_bidir %d %d %zu %" PRIu64 "\n" + : "propose_shared %d %d %zu %" PRIu64 "\n", committed, ctx_len, bytes, seq); std::fflush(cmd); int32_t status = -1; bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) && status == 0; if (ok) { hidden_out.assign(noise_expected, 0.0f); - ok = read_exact_fd(stream_fd, hidden_out.data(), - hidden_out.size() * sizeof(float)); + ok = bidirectional + ? process_.read_shared_payload(hidden_out.data(), bytes, seq) + : read_exact_fd(stream_fd, hidden_out.data(), bytes); } if (!ok) { std::fprintf(stderr, "draft-ipc propose_shared failed status=%d\n", status); @@ -348,6 +443,7 @@ void DFlashDraftIpcClient::close() { process_.close(); active_ = false; ring_cap_ = 0; + mode_ = BackendIpcMode::Invalid; } // ── Remote draft feature copy helper ──────────────────────────────── diff --git a/server/src/common/dflash_draft_ipc.h b/server/src/common/dflash_draft_ipc.h index 4830732b4..94e215a44 100644 --- a/server/src/common/dflash_draft_ipc.h +++ b/server/src/common/dflash_draft_ipc.h @@ -17,6 +17,7 @@ #include "ggml.h" #include "ggml-backend.h" +#include #include #include #include @@ -41,13 +42,28 @@ class DFlashDraftIpcClient { const std::string & draft_path, int draft_gpu, int ring_cap, - const std::string & work_dir); + const std::string & work_dir, + BackendIpcMode mode = BackendIpcMode::DFlashDraft); bool send_feature_slice(int capture_idx, int start_pos, int n_tokens, const std::vector & slice); + // Token-major [n_tokens, n_target_layers, hidden_size] upload. This is the + // preferred DSpark path because all capture layers cross IPC in one + // command and one acknowledgement. + bool send_feature_block(int start_pos, + int n_tokens, + const float * features, + size_t feature_count); + bool send_feature_block(int start_pos, + int n_tokens, + const std::vector & features) { + return send_feature_block( + start_pos, n_tokens, features.data(), features.size()); + } + bool propose(int committed, int ctx_len, const std::vector & noise_embed, @@ -71,6 +87,7 @@ class DFlashDraftIpcClient { int hidden_size_ = DFLASH27B_TARGET_HIDDEN; int block_size_ = DFLASH27B_DRAFT_BLOCK_SIZE; int n_target_layers_ = DFLASH27B_DRAFT_N_TARGET_LAYERS; + BackendIpcMode mode_ = BackendIpcMode::Invalid; }; // ── Remote draft feature copy helper ──────────────────────────────── diff --git a/server/src/deepseek4/deepseek4_backend.cpp b/server/src/deepseek4/deepseek4_backend.cpp index 4afe7731c..824f7033d 100644 --- a/server/src/deepseek4/deepseek4_backend.cpp +++ b/server/src/deepseek4/deepseek4_backend.cpp @@ -364,6 +364,32 @@ bool DeepSeek4Backend::init() { if (load_deepseek4_dspark_drafter(dp, backend_, *spec_drafter_)) { spec_enabled_ = true; std::fprintf(stderr, "[deepseek4] DSpark spec-decode ENABLED (drafter=%s)\n", dp); + const char * ipc_bin = std::getenv("DFLASH_DS4_DRAFT_IPC_BIN"); + if (ipc_bin && *ipc_bin) { + const char * ipc_work = std::getenv("DFLASH_DS4_DRAFT_IPC_WORK_DIR"); + int ipc_gpu = 0; + if (const char * gpu = std::getenv("DFLASH_DS4_DRAFT_IPC_GPU")) { + ipc_gpu = std::max(0, std::atoi(gpu)); + } + spec_remote_drafter_ = std::make_unique( + w_.n_embd, spec_drafter_->block_size, + spec_drafter_->n_target_layers); + if (!spec_remote_drafter_->start( + ipc_bin, dp, ipc_gpu, w_.n_swa, + ipc_work && *ipc_work ? ipc_work : "", + BackendIpcMode::DeepSeek4DSparkDraft)) { + std::fprintf(stderr, + "[deepseek4] DSpark remote draft IPC failed; using local draft\n"); + spec_remote_drafter_.reset(); + if (env_flag_enabled("DFLASH_DS4_DRAFT_IPC_REQUIRED")) { + return false; + } + } else { + std::fprintf(stderr, + "[deepseek4] DSpark remote draft IPC ENABLED gpu=%d\n", + ipc_gpu); + } + } } else { std::fprintf(stderr, "[deepseek4] DSpark drafter load FAILED: %s\n", deepseek4_dspark_last_error()); @@ -771,10 +797,14 @@ GenerateResult DeepSeek4Backend::generate_impl(const GenerateRequest & req, const int feat_row = spec_drafter_->n_target_layers * w_.n_embd; const int win_len = feat_row > 0 ? (int) (spec_feat_window_.size() / feat_row) : 0; std::vector spec_toks; - run_deepseek4_dspark_spec_decode(backend_, w_, cache_, *spec_drafter_, - committed, seed, req.n_gen - 1, - win_len > 0 ? spec_feat_window_.data() : nullptr, win_len, - spec_toks, &accept_rate); + if (!run_deepseek4_dspark_spec_decode( + backend_, w_, cache_, *spec_drafter_, + committed, seed, req.n_gen - 1, + win_len > 0 ? spec_feat_window_.data() : nullptr, win_len, + spec_toks, &accept_rate, spec_remote_drafter_.get())) { + result.error = "spec decode"; + return result; + } for (int t : spec_toks) io.emit(t); gen.insert(gen.end(), spec_toks.begin(), spec_toks.end()); } @@ -843,6 +873,8 @@ bool DeepSeek4Backend::handle_compress(const std::string & line, } void DeepSeek4Backend::free_drafter() { + spec_remote_drafter_.reset(); + if (spec_drafter_) free_deepseek4_dspark_drafter(*spec_drafter_); spec_drafter_.reset(); spec_enabled_ = false; spec_feat_window_.clear(); @@ -859,6 +891,7 @@ void DeepSeek4Backend::maybe_save_routing_stats() { void DeepSeek4Backend::shutdown() { maybe_save_routing_stats(); + free_drafter(); for (int i = 0; i < PREFIX_SLOTS; i++) { free_deepseek4_snapshot(snapshots_[i]); } diff --git a/server/src/deepseek4/deepseek4_backend.h b/server/src/deepseek4/deepseek4_backend.h index 0641d9231..430b718a5 100644 --- a/server/src/deepseek4/deepseek4_backend.h +++ b/server/src/deepseek4/deepseek4_backend.h @@ -14,6 +14,7 @@ #include "../common/moe_hybrid_stream.h" #include "deepseek4_internal.h" #include "deepseek4_dspark.h" +#include "common/dflash_draft_ipc.h" #include "ggml.h" #include "ggml-backend.h" @@ -80,6 +81,7 @@ class DeepSeek4Backend : public ModelBackend { // DSpark speculative decode (opt-in: DFLASH_DS4_SPEC=1 + DFLASH_DS4_DRAFT=). bool spec_enabled_ = false; std::unique_ptr spec_drafter_; + std::unique_ptr spec_remote_drafter_; std::vector spec_feat_window_; // Prefill prompt tokens in chunks, return absolute committed position. diff --git a/server/src/deepseek4/deepseek4_dspark.h b/server/src/deepseek4/deepseek4_dspark.h index 9fe20da75..b108252ee 100644 --- a/server/src/deepseek4/deepseek4_dspark.h +++ b/server/src/deepseek4/deepseek4_dspark.h @@ -29,11 +29,14 @@ #include "ggml.h" #include "ggml-backend.h" +#include #include #include namespace dflash::common { +class DFlashDraftIpcClient; + // The drafter weights. `core` reuses DeepSeek4Weights for the n_layer decoder // blocks + per-layer tensors + metadata + out_norm + output_hc_* tail; its // tok_embd/output stay null (tied to target). The DSpark-specific tensors below @@ -136,6 +139,15 @@ bool run_deepseek4_dspark_spec_decode( const float * prompt_feature_window, // [n_target_layers*n_embd * win_len] captured during prefill int win_len, std::vector & out_tokens, - float * accept_rate_out); + float * accept_rate_out, + DFlashDraftIpcClient * remote_draft = nullptr); + +int run_deepseek4_dspark_draft_ipc_daemon(const char * draft_path, + int ring_cap, + int draft_gpu, + int stream_fd, + int payload_fd, + int shared_payload_fd = -1, + size_t shared_payload_bytes = 0); } // namespace dflash::common diff --git a/server/src/deepseek4/deepseek4_dspark_draft_ipc_daemon.cpp b/server/src/deepseek4/deepseek4_dspark_draft_ipc_daemon.cpp new file mode 100644 index 000000000..24cdcbef3 --- /dev/null +++ b/server/src/deepseek4/deepseek4_dspark_draft_ipc_daemon.cpp @@ -0,0 +1,354 @@ +// Out-of-process DeepSeek4 DSpark proposal worker. + +#include "deepseek4_dspark.h" + +#include "common/dflash_draft_ipc.h" +#include "common/io_utils.h" + +#include "ggml-backend.h" +#include "ggml-cuda.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +# include +# include +#endif + +namespace dflash::common { + +int run_deepseek4_dspark_draft_ipc_daemon( + const char * draft_path, + int ring_cap, + int draft_gpu, + int stream_fd, + int payload_fd, + int shared_payload_fd, + size_t shared_payload_bytes) { +#if defined(_WIN32) + (void) draft_path; + (void) ring_cap; + (void) draft_gpu; + (void) stream_fd; + (void) payload_fd; + (void) shared_payload_fd; + (void) shared_payload_bytes; + return 2; +#else + const bool shared_requested = + shared_payload_fd >= 0 || shared_payload_bytes > 0; + if (!draft_path || !*draft_path || ring_cap <= 0 || + stream_fd < 0 || (payload_fd < 0 && !shared_requested) || + (shared_requested && + (shared_payload_fd < 0 || shared_payload_bytes == 0))) { + std::fprintf(stderr, "[ds4-dspark-ipc] bad daemon configuration\n"); + if (stream_fd >= 0) stream_status(stream_fd, -1); + return 2; + } + + void * shared_payload = nullptr; + void * shared_payload_data = nullptr; + size_t shared_payload_map_bytes = 0; + if (shared_requested) { + if (!backend_ipc_shared_payload_map_bytes( + shared_payload_bytes, shared_payload_map_bytes)) { + std::fprintf(stderr, "[ds4-dspark-ipc] invalid shared payload size\n"); + stream_status(stream_fd, -1); + return 1; + } + shared_payload = ::mmap(nullptr, shared_payload_map_bytes, + PROT_READ | PROT_WRITE, MAP_SHARED, + shared_payload_fd, 0); + if (shared_payload == MAP_FAILED) { + std::fprintf(stderr, "[ds4-dspark-ipc] shared payload mmap failed\n"); + stream_status(stream_fd, -1); + return 1; + } + shared_payload_data = static_cast(shared_payload) + + backend_ipc_shared_payload_header_bytes(); + } + auto unmap_shared = [&]() { + if (shared_payload && shared_payload != MAP_FAILED) { + ::munmap(shared_payload, shared_payload_map_bytes); + shared_payload = nullptr; + shared_payload_data = nullptr; + } + }; + auto shared_request_valid = [&](size_t bytes, uint64_t sequence) { + const auto * header = + static_cast(shared_payload); + return sequence != 0 && shared_payload && shared_payload_data && + backend_ipc_payload_in_bounds( + 0, bytes, shared_payload_bytes) && + header->sequence == sequence && header->bytes == bytes; + }; + + ggml_backend_t backend = ggml_backend_cuda_init(std::max(0, draft_gpu)); + if (!backend) { + std::fprintf(stderr, "[ds4-dspark-ipc] backend init failed gpu=%d\n", + draft_gpu); + stream_status(stream_fd, -1); + unmap_shared(); + return 1; + } + + DSparkDrafter drafter; + if (!load_deepseek4_dspark_drafter(draft_path, backend, drafter)) { + std::fprintf(stderr, "[ds4-dspark-ipc] draft load failed: %s\n", + deepseek4_dspark_last_error()); + stream_status(stream_fd, -1); + ggml_backend_free(backend); + unmap_shared(); + return 1; + } + + const int hidden = drafter.core.n_embd; + const int block = drafter.block_size; + const int n_target_layers = drafter.n_target_layers; + const int feature_row = hidden * n_target_layers; + if (hidden <= 0 || block <= 0 || n_target_layers <= 0 || feature_row <= 0) { + std::fprintf(stderr, "[ds4-dspark-ipc] invalid draft dimensions\n"); + stream_status(stream_fd, -1); + free_deepseek4_dspark_drafter(drafter); + ggml_backend_free(backend); + unmap_shared(); + return 1; + } + + std::vector feature_ring((size_t) ring_cap * feature_row, 0.0f); + std::vector noise_embed((size_t) hidden * block); + std::vector context; + std::vector hidden_out; + + auto ring_slot = [ring_cap](int position) { + int slot = position % ring_cap; + return slot < 0 ? slot + ring_cap : slot; + }; + + auto store_feature_slice = [&](int capture_idx, int start_pos, int n_tokens, + const std::vector & slice) { + if (capture_idx < 0 || capture_idx >= n_target_layers || + start_pos < 0 || n_tokens <= 0 || + slice.size() != (size_t) n_tokens * hidden) { + return false; + } + for (int i = 0; i < n_tokens; ++i) { + float * dst = feature_ring.data() + + (size_t) ring_slot(start_pos + i) * feature_row + + (size_t) capture_idx * hidden; + std::memcpy(dst, slice.data() + (size_t) i * hidden, + sizeof(float) * (size_t) hidden); + } + return true; + }; + + auto store_feature_block = [&](int start_pos, int n_tokens, + const float * features, + size_t feature_count) { + const size_t expected = (size_t) n_tokens * (size_t) feature_row; + if (start_pos < 0 || n_tokens <= 0 || n_tokens > ring_cap || + !features || feature_count != expected) { + return false; + } + for (int i = 0; i < n_tokens; ++i) { + float * dst = feature_ring.data() + + (size_t) ring_slot(start_pos + i) * feature_row; + std::memcpy(dst, features + (size_t) i * feature_row, + sizeof(float) * (size_t) feature_row); + } + return true; + }; + + auto run_proposal = [&](int committed, int ctx_len, + const float * noise) { + if (committed < 0 || ctx_len <= 0 || ctx_len > ring_cap || !noise) { + return false; + } + context.resize((size_t) ctx_len * feature_row); + const int start_pos = committed - ctx_len; + for (int i = 0; i < ctx_len; ++i) { + const float * src = feature_ring.data() + + (size_t) ring_slot(start_pos + i) * feature_row; + std::memcpy(context.data() + (size_t) i * feature_row, src, + sizeof(float) * (size_t) feature_row); + } + return deepseek4_dspark_draft_forward( + backend, drafter, noise, context.data(), ctx_len, + committed, hidden_out); + }; + + std::fprintf(stderr, + "[ds4-dspark-ipc] ready gpu=%d ring_cap=%d hidden=%d block=%d captures=%d transport=%s\n", + draft_gpu, ring_cap, hidden, block, n_target_layers, + shared_payload ? "shared" : "stream"); + if (!stream_status(stream_fd, 0)) { + free_deepseek4_dspark_drafter(drafter); + ggml_backend_free(backend); + unmap_shared(); + return 1; + } + + std::string line; + while (std::getline(std::cin, line)) { + std::istringstream iss(line); + std::string cmd; + iss >> cmd; + if (cmd == "quit" || cmd == "exit") break; + + if (cmd == "feature_slice_pipe") { + int capture_idx = -1; + int start_pos = -1; + int n_tokens = 0; + size_t bytes = 0; + iss >> capture_idx >> start_pos >> n_tokens >> bytes; + const size_t expected = (size_t) std::max(0, n_tokens) * hidden * + sizeof(float); + bool ok = iss && bytes == expected && n_tokens > 0; + std::vector slice(ok ? bytes / sizeof(float) : 0); + if (ok) ok = read_exact_fd(payload_fd, slice.data(), bytes); + if (ok) ok = store_feature_slice(capture_idx, start_pos, n_tokens, slice); + if (!stream_status(stream_fd, ok ? 0 : -1)) break; + continue; + } + + if (cmd == "feature_slice_shared") { + int capture_idx = -1; + int start_pos = -1; + int n_tokens = 0; + size_t bytes = 0; + uint64_t sequence = 0; + iss >> capture_idx >> start_pos >> n_tokens >> bytes >> sequence; + const size_t expected = (size_t) std::max(0, n_tokens) * hidden * + sizeof(float); + bool ok = iss && bytes == expected && n_tokens > 0 && + capture_idx >= 0 && capture_idx < n_target_layers && + shared_request_valid(bytes, sequence); + std::vector slice(ok ? bytes / sizeof(float) : 0); + if (ok) std::memcpy(slice.data(), shared_payload_data, bytes); + if (ok) { + ok = store_feature_slice( + capture_idx, start_pos, n_tokens, slice); + } + if (!stream_status(stream_fd, ok ? 0 : -1)) break; + continue; + } + + if (cmd == "feature_block_pipe") { + int start_pos = -1; + int n_tokens = 0; + size_t bytes = 0; + iss >> start_pos >> n_tokens >> bytes; + const size_t expected = (size_t) std::max(0, n_tokens) * + (size_t) feature_row * sizeof(float); + bool ok = iss && payload_fd >= 0 && bytes == expected && + n_tokens > 0 && n_tokens <= ring_cap; + std::vector features(ok ? bytes / sizeof(float) : 0); + if (ok) ok = read_exact_fd(payload_fd, features.data(), bytes); + if (ok) { + ok = store_feature_block( + start_pos, n_tokens, features.data(), features.size()); + } + if (!stream_status(stream_fd, ok ? 0 : -1)) break; + continue; + } + + if (cmd == "feature_block_shared") { + int start_pos = -1; + int n_tokens = 0; + size_t bytes = 0; + uint64_t sequence = 0; + iss >> start_pos >> n_tokens >> bytes >> sequence; + const size_t expected = (size_t) std::max(0, n_tokens) * + (size_t) feature_row * sizeof(float); + bool ok = iss && bytes == expected && n_tokens > 0 && + n_tokens <= ring_cap && + shared_request_valid(bytes, sequence); + if (ok) { + ok = store_feature_block( + start_pos, n_tokens, + static_cast(shared_payload_data), + bytes / sizeof(float)); + } + if (!stream_status(stream_fd, ok ? 0 : -1)) break; + continue; + } + + if (cmd == "propose_pipe") { + int committed = -1; + int ctx_len = 0; + size_t bytes = 0; + iss >> committed >> ctx_len >> bytes; + const size_t expected = noise_embed.size() * sizeof(float); + bool ok = iss && bytes == expected; + if (!ok) { + std::fprintf(stderr, + "[ds4-dspark-ipc] bad propose header committed=%d ctx=%d bytes=%zu expected=%zu\n", + committed, ctx_len, bytes, expected); + } + if (ok && !read_exact_fd(payload_fd, noise_embed.data(), bytes)) { + std::fprintf(stderr, "[ds4-dspark-ipc] propose payload read failed\n"); + ok = false; + } + if (ok && !run_proposal( + committed, ctx_len, noise_embed.data())) { + std::fprintf(stderr, + "[ds4-dspark-ipc] proposal graph failed committed=%d ctx=%d\n", + committed, ctx_len); + ok = false; + } + if (!stream_status(stream_fd, ok ? 0 : -1)) break; + if (ok && !write_exact_fd(stream_fd, hidden_out.data(), + hidden_out.size() * sizeof(float))) { + break; + } + continue; + } + + if (cmd == "propose_shared_bidir") { + int committed = -1; + int ctx_len = 0; + size_t bytes = 0; + uint64_t sequence = 0; + iss >> committed >> ctx_len >> bytes >> sequence; + const size_t expected = noise_embed.size() * sizeof(float); + bool ok = iss && bytes == expected && + shared_request_valid(bytes, sequence); + if (ok) { + ok = run_proposal( + committed, ctx_len, + static_cast(shared_payload_data)); + } + if (ok && hidden_out.size() * sizeof(float) == bytes) { + std::memcpy(shared_payload_data, hidden_out.data(), bytes); + auto * header = + static_cast(shared_payload); + header->bytes = bytes; + header->sequence = sequence; + } else { + ok = false; + } + if (!stream_status(stream_fd, ok ? 0 : -1)) break; + continue; + } + + std::fprintf(stderr, "[ds4-dspark-ipc] unknown command: %s\n", + line.c_str()); + if (!stream_status(stream_fd, -1)) break; + } + + free_deepseek4_dspark_drafter(drafter); + ggml_backend_free(backend); + unmap_shared(); + return 0; +#endif +} + +} // namespace dflash::common diff --git a/server/src/deepseek4/deepseek4_dspark_spec.cpp b/server/src/deepseek4/deepseek4_dspark_spec.cpp index 48527e391..1e0c77fea 100644 --- a/server/src/deepseek4/deepseek4_dspark_spec.cpp +++ b/server/src/deepseek4/deepseek4_dspark_spec.cpp @@ -24,6 +24,7 @@ #include "deepseek4_internal.h" #include "internal.h" #include "common/dspark_head.h" +#include "common/dflash_draft_ipc.h" #include "ggml.h" #include "ggml-backend.h" @@ -370,7 +371,8 @@ bool run_deepseek4_dspark_spec_decode( const float * prompt_feature_window, int win_len, std::vector & out_tokens, - float * accept_rate_out) { + float * accept_rate_out, + DFlashDraftIpcClient * remote_draft) { const int n_embd = target_w.n_embd; const int n_tgt = drafter.n_target_layers; const int block = drafter.block_size; @@ -442,6 +444,18 @@ bool run_deepseek4_dspark_spec_decode( } int feat_count = win_have; // number of valid feature columns ending at committed-1 + if (remote_draft && remote_draft->active() && win_have > 0) { + const int start_pos = committed - win_have; + if (!remote_draft->send_feature_block( + start_pos, win_have, feat_win.data(), + (size_t) win_have * feat_row)) { + std::fprintf(stderr, + "[ds4-spec] remote draft prompt feature upload failed\n"); + ggml_backend_free(snap_backend); + return false; + } + } + auto push_feature = [&](const float * col) { // Shift-append one feature column (keep last n_swa). if (feat_count >= n_swa) { @@ -482,11 +496,16 @@ bool run_deepseek4_dspark_spec_decode( if (!target.embed_tokens(noise_ids.data(), block, noise_embed.data())) break; // Drafter forward -> block normed hidden states. - if (!deepseek4_dspark_draft_forward(backend, drafter, noise_embed.data(), - ctx_len > 0 ? feat_win.data() : nullptr, - ctx_len, pos, local_hidden)) { + const bool draft_ok = remote_draft && remote_draft->active() + ? remote_draft->propose(pos, ctx_len, noise_embed, local_hidden) + : deepseek4_dspark_draft_forward( + backend, drafter, noise_embed.data(), + ctx_len > 0 ? feat_win.data() : nullptr, + ctx_len, pos, local_hidden); + if (!draft_ok) { std::fprintf(stderr, "[ds4-spec] drafter forward failed\n"); - break; + ggml_backend_free(snap_backend); + return false; } } tm_draft += spec_ms_since(t0); @@ -653,6 +672,17 @@ bool run_deepseek4_dspark_spec_decode( t0 = SpecClock::now(); const std::vector & feats = target.last_features(); const int fN = full_snap ? target.last_verify_n() : accept; + if (remote_draft && remote_draft->active() && fN > 0) { + const size_t feature_count = (size_t) fN * feat_row; + if (feats.size() < feature_count || + !remote_draft->send_feature_block( + pos, fN, feats.data(), feature_count)) { + std::fprintf(stderr, + "[ds4-spec] remote draft feature update failed\n"); + ggml_backend_free(snap_backend); + return false; + } + } for (int i = 0; i < fN; i++) push_feature(feats.data() + (size_t) i * feat_row); tm_feat += spec_ms_since(t0); diff --git a/server/src/deepseek4/deepseek4_graph.cpp b/server/src/deepseek4/deepseek4_graph.cpp index 0eeea9a0d..4619792f3 100644 --- a/server/src/deepseek4/deepseek4_graph.cpp +++ b/server/src/deepseek4/deepseek4_graph.cpp @@ -4593,6 +4593,8 @@ bool deepseek4_step_layer_range( } std::vector fused_debug_logits; if (n_tokens == 1 && allow_decode_graph_reuse && layer_begin == 0 && is_last_shard && + !(verify_hooks && verify_hooks->capture_layer_ids && + verify_hooks->capture_out) && out_logits && ds4_backend_is_gpu(backend) && ds4_fused_decode_enabled()) { const int rc = ds4_try_fused_decode_step( fused_decode_graph_cache, backend, w, cache, hc_layer_weights_range, @@ -4646,6 +4648,8 @@ bool deepseek4_step_layer_range( reuse_decode_graphs && ds4_backend_is_gpu(backend) && backend_decode_hc_supported && + !(verify_hooks && verify_hooks->capture_layer_ids && + verify_hooks->capture_out) && !ds4_env_flag("DFLASH_DS4_HC_CPU"); const bool use_backend_decode_hc_direct = use_backend_decode_hc && ds4_backend_is_hip(backend); const bool use_backend_decode_hc_graph = diff --git a/server/src/ipc/backend_ipc_main.cpp b/server/src/ipc/backend_ipc_main.cpp index ca5ba1565..e06983540 100644 --- a/server/src/ipc/backend_ipc_main.cpp +++ b/server/src/ipc/backend_ipc_main.cpp @@ -4,6 +4,7 @@ #include "moe_expert_compute.h" #include "dflash_draft_ipc.h" +#include "deepseek4/deepseek4_dspark.h" #include "deepseek4/deepseek4_layer_split_adapter.h" #include "gemma4/gemma4_layer_split_adapter.h" #include "laguna/laguna_layer_split_adapter.h" @@ -130,6 +131,8 @@ int main(int argc, char ** argv) { " or: %s --backend-ipc-mode=deepseek4-target-shard " "--stream-fd=FD --target-gpus=N[,N...] --layer-begins=N[,N...] " "--layer-ends=N[,N...] --max-ctx=N\n" + " or: %s --backend-ipc-mode=deepseek4-dspark-draft " + "--ring-cap=N --stream-fd=FD --payload-fd=FD [--draft-gpu=N]\n" " or: %s --backend-ipc-mode=moe-expert-compute " "--stream-fd=FD --target-gpu=N --placement=PATH\n", argv[0], @@ -137,6 +140,8 @@ int main(int argc, char ** argv) { argv[0], argv[0], argv[0], + argv[0], + argv[0], argv[0]); return 2; } @@ -351,6 +356,10 @@ int main(int argc, char ** argv) { return run_deepseek4_target_shard_ipc_daemon( payload_path, target_gpus, layer_begins, layer_ends, max_ctx, stream_fd, payload_fd); + case BackendIpcMode::DeepSeek4DSparkDraft: + return run_deepseek4_dspark_draft_ipc_daemon( + payload_path, ring_cap, draft_gpu, stream_fd, payload_fd, + shared_payload_fd, shared_payload_bytes); } std::fprintf(stderr, "[backend-ipc-daemon] unsupported mode\n"); return 2; diff --git a/server/test/test_server_unit.cpp b/server/test/test_server_unit.cpp index 9b1dd749b..065da2883 100644 --- a/server/test/test_server_unit.cpp +++ b/server/test/test_server_unit.cpp @@ -3041,6 +3041,11 @@ static void test_backend_ipc_payload_transport_parse() { TEST_ASSERT(mode == BackendIpcMode::PFlashCompress); TEST_ASSERT(parse_backend_ipc_mode("qwen35-target-shard", mode)); TEST_ASSERT(mode == BackendIpcMode::Qwen35TargetShard); + TEST_ASSERT(parse_backend_ipc_mode("deepseek4-dspark-draft", mode)); + TEST_ASSERT(mode == BackendIpcMode::DeepSeek4DSparkDraft); + TEST_ASSERT(std::strcmp( + backend_ipc_mode_name(BackendIpcMode::DeepSeek4DSparkDraft), + "deepseek4-dspark-draft") == 0); TEST_ASSERT(parse_backend_ipc_mode("moe-expert-compute", mode)); TEST_ASSERT(mode == BackendIpcMode::MoeExpertCompute); TEST_ASSERT(!parse_backend_ipc_mode("moe-ffn", mode));