From 6a9cbe5b1d133ad4c8952582f2ffe61e98122d34 Mon Sep 17 00:00:00 2001 From: weicj Date: Fri, 22 May 2026 00:56:20 +0800 Subject: [PATCH 1/3] feat(server): add native mixed-backend draft placement --- dflash/CMakeLists.txt | 21 ++ dflash/src/common/backend_factory.cpp | 5 + dflash/src/common/backend_factory.h | 4 + dflash/src/common/model_backend.h | 6 + dflash/src/ipc/dflash_draft_ipc_main.cpp | 45 +++++ dflash/src/placement/remote_draft_config.h | 18 ++ dflash/src/qwen35/qwen35_backend.cpp | 221 +++++++++++++++------ dflash/src/qwen35/qwen35_backend.h | 7 + dflash/src/server/server_main.cpp | 84 +++++++- 9 files changed, 341 insertions(+), 70 deletions(-) create mode 100644 dflash/src/ipc/dflash_draft_ipc_main.cpp create mode 100644 dflash/src/placement/remote_draft_config.h diff --git a/dflash/CMakeLists.txt b/dflash/CMakeLists.txt index 71d81a255..c84a5f56f 100644 --- a/dflash/CMakeLists.txt +++ b/dflash/CMakeLists.txt @@ -671,6 +671,27 @@ if(DFLASH27B_TESTS) endif() endif() + if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/src/ipc/dflash_draft_ipc_main.cpp") + add_executable(dflash_draft_ipc_daemon + src/ipc/dflash_draft_ipc_main.cpp + ) + target_include_directories(dflash_draft_ipc_daemon PRIVATE ${DFLASH27B_SRC_INCLUDE_DIRS}) + if(DFLASH27B_GPU_BACKEND STREQUAL "hip") + target_compile_definitions(dflash_draft_ipc_daemon PRIVATE DFLASH27B_BACKEND_HIP=1 GGML_USE_HIP) + else() + target_compile_definitions(dflash_draft_ipc_daemon PRIVATE + DFLASH27B_BACKEND_CUDA=1 + DFLASH27B_CUDA_MIN_SM=${_dflash_cuda_min_sm}) + endif() + target_link_libraries(dflash_draft_ipc_daemon PRIVATE dflash_common ggml ${DFLASH27B_GGML_BACKEND_TARGET} pthread) + if(DFLASH27B_GPU_BACKEND STREQUAL "cuda") + find_package(CUDAToolkit REQUIRED) + target_link_libraries(dflash_draft_ipc_daemon PRIVATE CUDA::cudart) + else() + target_link_libraries(dflash_draft_ipc_daemon PRIVATE hip::host) + endif() + endif() + # Tokenizer test harness (no GPU needed — links static lib for tokenizer + GGUF reader) if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/test/test_tokenizer_harness.cpp") add_executable(test_tokenizer_harness test/test_tokenizer_harness.cpp) diff --git a/dflash/src/common/backend_factory.cpp b/dflash/src/common/backend_factory.cpp index 236989548..e5136d4e5 100644 --- a/dflash/src/common/backend_factory.cpp +++ b/dflash/src/common/backend_factory.cpp @@ -17,6 +17,10 @@ std::string detect_arch(const char * model_path) { return info.arch; } +bool arch_supports_remote_draft(const std::string & arch) { + return arch == "qwen35"; +} + std::unique_ptr create_backend(const BackendArgs & args) { if (!args.model_path) { std::fprintf(stderr, "[backend_factory] model_path is null\n"); @@ -38,6 +42,7 @@ std::unique_ptr create_backend(const BackendArgs & args) { cfg.draft_path = args.draft_path; cfg.device = args.device; cfg.draft_gpu = args.draft_device.gpu; + cfg.remote_draft = args.remote_draft; cfg.stream_fd = args.stream_fd; cfg.fa_window = args.fa_window; cfg.kq_stride_pad = args.kq_stride_pad; diff --git a/dflash/src/common/backend_factory.h b/dflash/src/common/backend_factory.h index 1e959948b..64a2ac973 100644 --- a/dflash/src/common/backend_factory.h +++ b/dflash/src/common/backend_factory.h @@ -12,6 +12,7 @@ #include "model_backend.h" #include "placement/placement_config.h" +#include "placement/remote_draft_config.h" #include #include @@ -31,6 +32,7 @@ struct BackendArgs { // Device placement DevicePlacement device; DevicePlacement draft_device; + RemoteDraftConfig remote_draft; // I/O — only used when running under daemon_loop (legacy). The new // server passes -1 and uses on_token callbacks instead. @@ -62,4 +64,6 @@ std::unique_ptr create_backend(const BackendArgs & args); // Useful for early dispatch (e.g. printing which backend will be used). std::string detect_arch(const char * model_path); +bool arch_supports_remote_draft(const std::string & arch); + } // namespace dflash::common diff --git a/dflash/src/common/model_backend.h b/dflash/src/common/model_backend.h index fc1682ee6..64441bd91 100644 --- a/dflash/src/common/model_backend.h +++ b/dflash/src/common/model_backend.h @@ -183,6 +183,12 @@ struct ModelBackend { // growth over time. Default is a no-op. virtual void release_scratch() {} + // Return true when the backend can route draft execution through the + // common remote-draft IPC transport. Model families that do not implement + // the DFlash feature boundary keep the default false and are rejected by + // the server before startup. + virtual bool supports_remote_draft() const { return false; } + // ── Cleanup ────────────────────────────────────────────────────── // Release all resources (weights, cache, snapshots, drafter). // Called by run_daemon() before returning. diff --git a/dflash/src/ipc/dflash_draft_ipc_main.cpp b/dflash/src/ipc/dflash_draft_ipc_main.cpp new file mode 100644 index 000000000..97034008a --- /dev/null +++ b/dflash/src/ipc/dflash_draft_ipc_main.cpp @@ -0,0 +1,45 @@ +// Standalone DFlash draft IPC daemon entry point. + +#include "dflash_draft_ipc.h" + +#include +#include +#include +#include + +using namespace dflash::common; + +int main(int argc, char ** argv) { + if (argc < 3 || std::strcmp(argv[1], "--draft-ipc-daemon") != 0) { + std::fprintf(stderr, + "usage: %s --draft-ipc-daemon " + "--ring-cap=N --stream-fd=FD [--draft-gpu=N]\n", + argv[0]); + return 2; + } + + const char * draft_path = argv[2]; + int ring_cap = 4096; + int draft_gpu = 0; + int stream_fd = -1; + for (int i = 3; i < argc; i++) { + if (std::strncmp(argv[i], "--ring-cap=", 11) == 0) { + ring_cap = std::atoi(argv[i] + 11); + } else if (std::strcmp(argv[i], "--ring-cap") == 0) { + if (i + 1 < argc) ring_cap = std::atoi(argv[++i]); + } else if (std::strncmp(argv[i], "--draft-gpu=", 12) == 0) { + draft_gpu = std::max(0, std::atoi(argv[i] + 12)); + } else if (std::strcmp(argv[i], "--draft-gpu") == 0) { + if (i + 1 < argc) draft_gpu = std::max(0, std::atoi(argv[++i])); + } else if (std::strncmp(argv[i], "--stream-fd=", 12) == 0) { + stream_fd = std::atoi(argv[i] + 12); + } else if (std::strcmp(argv[i], "--stream-fd") == 0) { + if (i + 1 < argc) stream_fd = std::atoi(argv[++i]); + } else { + std::fprintf(stderr, "[draft-ipc-daemon] unknown option: %s\n", argv[i]); + return 2; + } + } + + return run_dflash_draft_ipc_daemon(draft_path, ring_cap, draft_gpu, stream_fd); +} diff --git a/dflash/src/placement/remote_draft_config.h b/dflash/src/placement/remote_draft_config.h new file mode 100644 index 000000000..a4f998f58 --- /dev/null +++ b/dflash/src/placement/remote_draft_config.h @@ -0,0 +1,18 @@ +// Remote draft execution configuration for mixed-backend target/draft placement. + +#pragma once + +#include + +namespace dflash::common { + +struct RemoteDraftConfig { + std::string ipc_bin; + std::string work_dir; + int ring_cap = 0; + + bool enabled() const { return !ipc_bin.empty(); } + bool has_aux_options() const { return !work_dir.empty() || ring_cap > 0; } +}; + +} // namespace dflash::common diff --git a/dflash/src/qwen35/qwen35_backend.cpp b/dflash/src/qwen35/qwen35_backend.cpp index f2ea5cecb..1d636c531 100644 --- a/dflash/src/qwen35/qwen35_backend.cpp +++ b/dflash/src/qwen35/qwen35_backend.cpp @@ -20,10 +20,22 @@ #include #include #include +#include #include namespace dflash::common { +namespace { +static float bf16_bits_to_f32(uint16_t bits) { + union { + uint32_t u; + float f; + } v; + v.u = (uint32_t)bits << 16; + return v.f; +} +} // namespace + #define IS_EOS_TOK(tok, w) \ ( ((w).eos_chat_id >= 0 && (tok) == (w).eos_chat_id) \ || ((w).eos_id >= 0 && (tok) == (w).eos_id ) ) @@ -37,14 +49,15 @@ Qwen35Backend::~Qwen35Backend() { shutdown(); } // ── init() ────────────────────────────────────────────────────────────── bool Qwen35Backend::init() { - split_gpus_ = (cfg_.device.gpu != cfg_.draft_gpu); + const bool use_remote_draft = cfg_.remote_draft.enabled(); + split_gpus_ = !use_remote_draft && (cfg_.device.gpu != cfg_.draft_gpu); target_backend_ = ggml_backend_cuda_init(cfg_.device.gpu); if (!target_backend_) { std::fprintf(stderr, "target cuda init failed\n"); return false; } - draft_backend_ = target_backend_; + draft_backend_ = use_remote_draft ? nullptr : target_backend_; if (split_gpus_) { draft_backend_ = ggml_backend_cuda_init(cfg_.draft_gpu); if (!draft_backend_) { @@ -72,7 +85,22 @@ bool Qwen35Backend::init() { std::printf("[target] %s\n", dflash27b_last_error()); // Load draft - if (cfg_.draft_path) { + if (cfg_.draft_path && use_remote_draft) { + const int cap = cfg_.remote_draft.ring_cap > 0 + ? std::min(cfg_.remote_draft.ring_cap, cfg_.device.max_ctx) + : std::min(cfg_.device.max_ctx, cfg_.draft_ctx_max); + if (!remote_draft_.start(cfg_.remote_draft.ipc_bin, cfg_.draft_path, + cfg_.draft_gpu, cap, + cfg_.remote_draft.work_dir)) { + std::fprintf(stderr, "remote draft start failed\n"); + return false; + } + dw_.n_embd = DFLASH27B_TARGET_HIDDEN; + dw_.block_size = DFLASH27B_DRAFT_BLOCK_SIZE; + dw_.n_target_layers = DFLASH27B_DRAFT_N_TARGET_LAYERS; + std::printf("[draft] remote ipc ready gpu=%d cap=%d\n", + cfg_.draft_gpu, cap); + } else if (cfg_.draft_path) { std::string dp(cfg_.draft_path); bool draft_ok = (dp.size() >= 5 && dp.substr(dp.size() - 5) == ".gguf") ? load_draft_gguf(cfg_.draft_path, draft_backend_, dw_, &w_) @@ -104,7 +132,7 @@ bool Qwen35Backend::init() { // Init feature mirror when draft model is available (needed for spec decode). // On single-GPU, this is an F32 conversion buffer; on split-GPU, a cross-device mirror. - if (cfg_.draft_path) { + if (cfg_.draft_path && !use_remote_draft) { const int mirror_cap = std::min({cfg_.draft_ctx_max, cfg_.device.max_ctx, cache_.target_feat_cap > 0 ? cache_.target_feat_cap : cfg_.device.max_ctx}); if (!draft_feature_mirror_init(feature_mirror_, draft_backend_, @@ -130,10 +158,15 @@ void Qwen35Backend::print_ready_banner() const { bool Qwen35Backend::park(const std::string & what) { bool want_draft = (what.empty() || what == "all" || what == "draft"); bool want_target = (what.empty() || what == "all" || what == "target"); + const bool use_remote_draft = cfg_.remote_draft.enabled(); if (want_draft && !draft_parked_) { - step_graph_destroy(draft_sg_); - free_draft_weights(dw_); + if (use_remote_draft) { + remote_draft_.close(); + } else { + step_graph_destroy(draft_sg_); + free_draft_weights(dw_); + } draft_parked_ = true; std::printf("[park] draft released\n"); std::fflush(stdout); } @@ -149,6 +182,7 @@ bool Qwen35Backend::park(const std::string & what) { bool Qwen35Backend::unpark(const std::string & what) { bool want_target = (what.empty() || what == "all" || what == "target"); bool want_draft = (what.empty() || what == "all" || what == "draft"); + const bool use_remote_draft = cfg_.remote_draft.enabled(); if (want_target && target_parked_) { if (!load_target_gguf(cfg_.target_path, target_backend_, w_)) { @@ -159,18 +193,30 @@ bool Qwen35Backend::unpark(const std::string & what) { std::printf("[unpark] target restored\n"); std::fflush(stdout); } if (want_draft && draft_parked_ && cfg_.draft_path) { - std::string dp(cfg_.draft_path); - bool draft_ok = (dp.size() >= 5 && dp.substr(dp.size() - 5) == ".gguf") - ? load_draft_gguf(cfg_.draft_path, draft_backend_, dw_, &w_) - : load_draft_safetensors(cfg_.draft_path, draft_backend_, dw_, &w_); - if (!draft_ok) { - std::fprintf(stderr, "[unpark] draft: %s\n", dflash27b_last_error()); - return false; - } - if (cfg_.draft_swa_window > 0) { - dw_.swa_window = cfg_.draft_swa_window; - for (int il = 0; il < dw_.n_layer - 1; il++) - dw_.layers[il].is_swa = true; + if (use_remote_draft) { + const int cap = cfg_.remote_draft.ring_cap > 0 + ? std::min(cfg_.remote_draft.ring_cap, cfg_.device.max_ctx) + : std::min(cfg_.device.max_ctx, cfg_.draft_ctx_max); + if (!remote_draft_.start(cfg_.remote_draft.ipc_bin, cfg_.draft_path, + cfg_.draft_gpu, cap, + cfg_.remote_draft.work_dir)) { + std::fprintf(stderr, "[unpark] remote draft failed\n"); + return false; + } + } else { + std::string dp(cfg_.draft_path); + bool draft_ok = (dp.size() >= 5 && dp.substr(dp.size() - 5) == ".gguf") + ? load_draft_gguf(cfg_.draft_path, draft_backend_, dw_, &w_) + : load_draft_safetensors(cfg_.draft_path, draft_backend_, dw_, &w_); + if (!draft_ok) { + std::fprintf(stderr, "[unpark] draft: %s\n", dflash27b_last_error()); + return false; + } + if (cfg_.draft_swa_window > 0) { + dw_.swa_window = cfg_.draft_swa_window; + for (int il = 0; il < dw_.n_layer - 1; il++) + dw_.layers[il].is_swa = true; + } } draft_parked_ = false; std::printf("[unpark] draft restored\n"); std::fflush(stdout); @@ -421,16 +467,18 @@ DFlashTarget * Qwen35Backend::dflash_target() { // ── Shutdown ──────────────────────────────────────────────────────────── void Qwen35Backend::shutdown() { + const bool use_remote_draft = cfg_.remote_draft.enabled(); free_drafter(); step_graph_destroy(sg_); step_graph_destroy(draft_sg_); step_graph_destroy(proj_sg_); + remote_draft_.close(); draft_feature_mirror_free(feature_mirror_); for (int i = 0; i < PREFIX_SLOTS; i++) { free_prefix_snapshot(prefix_snapshots_[i]); } if (!target_parked_) free_target_weights(w_); - if (!draft_parked_) free_draft_weights(dw_); + if (!use_remote_draft && !draft_parked_) free_draft_weights(dw_); free_target_cache(cache_); if (split_gpus_ && draft_backend_) { ggml_backend_free(draft_backend_); @@ -702,8 +750,10 @@ int Qwen35Backend::do_prefill(const std::vector & tokens, snap_slot = -1; } - // Sync feature mirror if active - if (feature_mirror_.target_feat && !draft_parked_) { + // Sync draft-side features if active. + if (remote_draft_.active() && !draft_parked_) { + if (!sync_remote_draft_features(kv_pos, n_tokens)) return -1; + } else if (feature_mirror_.target_feat && !draft_parked_) { draft_feature_mirror_sync_range(cache_.target_feat, cache_.target_feat_cap, feature_mirror_, kv_pos, n_tokens); } @@ -793,6 +843,39 @@ bool Qwen35Backend::do_ar_decode(int committed, int n_gen, return true; } +bool Qwen35Backend::sync_remote_draft_features(int start_pos, int n_tokens) { + if (!remote_draft_.active() || !cache_.target_feat || n_tokens <= 0) return true; + if (cache_.target_feat_cap <= 0) return false; + + const int n_capture = w_.n_capture_layers; + const int feat_hidden = w_.n_embd; + const size_t src_stride = cache_.target_feat->nb[1]; + std::vector slice((size_t)n_tokens * (size_t)feat_hidden); + std::vector bf16(feat_hidden); + ggml_backend_synchronize(target_backend_); + for (int cap_idx = 0; cap_idx < n_capture; ++cap_idx) { + for (int t = 0; t < n_tokens; ++t) { + const int slot = (start_pos + t) % cache_.target_feat_cap; + const size_t src_offset = (size_t)slot * src_stride + + (size_t)cap_idx * (size_t)feat_hidden * sizeof(uint16_t); + ggml_backend_tensor_get(cache_.target_feat, bf16.data(), + src_offset, + sizeof(uint16_t) * (size_t)feat_hidden); + float * dst = slice.data() + (size_t)t * feat_hidden; + for (int h = 0; h < feat_hidden; ++h) { + dst[h] = bf16_bits_to_f32(bf16[h]); + } + } + if (!remote_draft_.send_feature_slice(cap_idx, start_pos, n_tokens, slice)) { + std::fprintf(stderr, + "spec-decode: remote feature sync failed capture=%d\n", + cap_idx); + return false; + } + } + return true; +} + // ── DFlash speculative decode loop ───────────────────────────────────── bool Qwen35Backend::do_spec_decode(int committed, int n_gen, @@ -815,7 +898,9 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen, // - greedy decoding (no logit processing) — spec decode uses argmax verification const bool can_spec = cfg_.draft_path && !draft_parked_ - && feature_mirror_.target_feat + && (cfg_.remote_draft.enabled() + ? remote_draft_.active() + : feature_mirror_.target_feat != nullptr) && !sampler_.needs_logit_processing(); if (!can_spec) { @@ -829,7 +914,8 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen, // ── DFlash spec-decode: draft → verify → accept → replay ────────── DFlashTarget * target = dflash_target(); - const int q_len = dw_.block_size; + const bool use_remote_draft = cfg_.remote_draft.enabled() && remote_draft_.active(); + const int q_len = dw_.block_size > 0 ? dw_.block_size : DFLASH27B_DRAFT_BLOCK_SIZE; StepGraph draft_sg; @@ -864,50 +950,60 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen, // 2. Draft compute constexpr int DRAFT_CTX_MAX_DEFAULT = 2048; - const int ring_cap = feature_mirror_.cap; + const int ring_cap = use_remote_draft ? remote_draft_.ring_cap() : feature_mirror_.cap; const int draft_ctx = std::min(committed, std::min(ring_cap, std::max(DRAFT_CTX_MAX_DEFAULT, cfg_.draft_ctx_max))); const int draft_start = committed - draft_ctx; int mirror_slot0 = 0; const bool use_mirror_view = + !use_remote_draft && draft_feature_mirror_can_view(feature_mirror_, committed, draft_ctx, mirror_slot0); - if (!build_draft_step(draft_sg, dw_, /*lm_head=*/nullptr, draft_backend_, - draft_ctx, use_mirror_view ? &feature_mirror_ : nullptr, - committed, - /*ctx_len_max=*/std::min(ring_cap, std::max(DRAFT_CTX_MAX_DEFAULT, cfg_.draft_ctx_max)))) { - std::fprintf(stderr, "spec-decode: draft build failed\n"); - step_graph_destroy(draft_sg); - return false; - } - if (!use_mirror_view && - !copy_feature_ring_range_to_tensor(feature_mirror_, draft_sg.target_hidden_cat, - draft_start, draft_ctx)) { - std::fprintf(stderr, "spec-decode: feature copy failed\n"); - step_graph_destroy(draft_sg); - return false; - } - ggml_backend_tensor_set(draft_sg.inp_embed, noise_embed.data(), 0, - sizeof(float) * noise_embed.size()); - pos_k.resize((size_t)draft_ctx + q_len); - for (int i = 0; i < q_len; i++) pos_q[i] = draft_ctx + i; - for (int i = 0; i < draft_ctx + q_len; i++) pos_k[i] = i; - ggml_backend_tensor_set(draft_sg.positions, pos_q.data(), 0, - sizeof(int32_t) * pos_q.size()); - ggml_backend_tensor_set(draft_sg.positions_k, pos_k.data(), 0, - sizeof(int32_t) * pos_k.size()); - - auto st = ggml_backend_graph_compute(draft_backend_, draft_sg.gf); - if (st != GGML_STATUS_SUCCESS) { - std::fprintf(stderr, "spec-decode: draft compute failed\n"); - step_graph_destroy(draft_sg); - return false; - } + if (use_remote_draft) { + local_hidden.clear(); + if (!remote_draft_.propose(committed, draft_ctx, noise_embed, local_hidden)) { + std::fprintf(stderr, "spec-decode: remote draft propose failed\n"); + step_graph_destroy(draft_sg); + return false; + } + } else { + if (!build_draft_step(draft_sg, dw_, /*lm_head=*/nullptr, draft_backend_, + draft_ctx, use_mirror_view ? &feature_mirror_ : nullptr, + committed, + /*ctx_len_max=*/std::min(ring_cap, std::max(DRAFT_CTX_MAX_DEFAULT, cfg_.draft_ctx_max)))) { + std::fprintf(stderr, "spec-decode: draft build failed\n"); + step_graph_destroy(draft_sg); + return false; + } + if (!use_mirror_view && + !copy_feature_ring_range_to_tensor(feature_mirror_, draft_sg.target_hidden_cat, + draft_start, draft_ctx)) { + std::fprintf(stderr, "spec-decode: feature copy failed\n"); + step_graph_destroy(draft_sg); + return false; + } + ggml_backend_tensor_set(draft_sg.inp_embed, noise_embed.data(), 0, + sizeof(float) * noise_embed.size()); + pos_k.resize((size_t)draft_ctx + q_len); + for (int i = 0; i < q_len; i++) pos_q[i] = draft_ctx + i; + for (int i = 0; i < draft_ctx + q_len; i++) pos_k[i] = i; + ggml_backend_tensor_set(draft_sg.positions, pos_q.data(), 0, + sizeof(int32_t) * pos_q.size()); + ggml_backend_tensor_set(draft_sg.positions_k, pos_k.data(), 0, + sizeof(int32_t) * pos_k.size()); + + auto st = ggml_backend_graph_compute(draft_backend_, draft_sg.gf); + if (st != GGML_STATUS_SUCCESS) { + std::fprintf(stderr, "spec-decode: draft compute failed\n"); + step_graph_destroy(draft_sg); + return false; + } - // Read draft hidden states to host for LM-head projection. - local_hidden.resize((size_t)hidden * q_len); - ggml_backend_tensor_get(draft_sg.hidden_states, local_hidden.data(), 0, - sizeof(float) * local_hidden.size()); + // Read draft hidden states to host for LM-head projection. + local_hidden.resize((size_t)hidden * q_len); + ggml_backend_tensor_get(draft_sg.hidden_states, local_hidden.data(), 0, + sizeof(float) * local_hidden.size()); + } // 3. Project draft hidden → token IDs via target LM head if (!target->project_hidden_to_tokens(local_hidden.data(), q_len, draft_tok)) { @@ -979,7 +1075,12 @@ bool Qwen35Backend::do_spec_decode(int committed, int n_gen, last_tok = replay_last_tok; // 7. Sync features for replayed range to mirror (needed for next draft step) - if (feature_mirror_.target_feat && cache_.target_feat) { + if (use_remote_draft && cache_.target_feat) { + if (!sync_remote_draft_features(committed, commit_n)) { + step_graph_destroy(draft_sg); + return false; + } + } else if (feature_mirror_.target_feat && cache_.target_feat) { draft_feature_mirror_sync_range(cache_.target_feat, cache_.target_feat_cap, feature_mirror_, committed, commit_n); } diff --git a/dflash/src/qwen35/qwen35_backend.h b/dflash/src/qwen35/qwen35_backend.h index 506e30da4..277a8d3d5 100644 --- a/dflash/src/qwen35/qwen35_backend.h +++ b/dflash/src/qwen35/qwen35_backend.h @@ -13,7 +13,9 @@ #include "common/model_backend.h" #include "common/dflash_target.h" +#include "common/dflash_draft_ipc.h" #include "placement/placement_config.h" +#include "placement/remote_draft_config.h" #include "step_graph.h" #include "ddtree.h" #include "dflash_feature_ring.h" @@ -37,6 +39,7 @@ struct Qwen35Config { const char * draft_path = nullptr; DevicePlacement device; // target GPU placement int draft_gpu = 0; + RemoteDraftConfig remote_draft; int stream_fd = -1; // FA/KV @@ -107,6 +110,7 @@ class Qwen35Backend : public ModelBackend { bool supports_dflash_spec_decode() const override { return true; } DFlashTarget * dflash_target() override; + bool supports_remote_draft() const override { return true; } void shutdown() override; @@ -136,6 +140,7 @@ class Qwen35Backend : public ModelBackend { // ── Draft feature mirror (cross-GPU feature transfer) ──────────── DraftFeatureMirror feature_mirror_; + DFlashDraftIpcClient remote_draft_; // ── Prefix cache (snapshots) ───────────────────────────────────── static constexpr int PREFIX_SLOTS = 64; @@ -181,6 +186,8 @@ class Qwen35Backend : public ModelBackend { std::vector & out_tokens, const DaemonIO & io); + bool sync_remote_draft_features(int start_pos, int n_tokens); + // Chain-mode verify (single batch of q_len tokens). int verify_chain(int committed, const int32_t * draft_tok, int q_len); diff --git a/dflash/src/server/server_main.cpp b/dflash/src/server/server_main.cpp index dba9401de..166b41df2 100644 --- a/dflash/src/server/server_main.cpp +++ b/dflash/src/server/server_main.cpp @@ -65,24 +65,42 @@ static bool validate_server_placement(const BackendArgs & bargs) { placement_backend_name(compiled)); return false; } - if (!placement_backend_supported(bargs.draft_device.backend)) { - std::fprintf(stderr, - "[server] --draft-device=%s is unsupported in this binary " - "(compiled backend: %s)\n", - placement_device_name(bargs.draft_device).c_str(), - placement_backend_name(compiled)); - return false; - } const PlacementBackend target = bargs.device.backend == PlacementBackend::Auto ? compiled : bargs.device.backend; const PlacementBackend draft = bargs.draft_device.backend == PlacementBackend::Auto ? target : bargs.draft_device.backend; + if (!bargs.remote_draft.enabled() && bargs.remote_draft.has_aux_options()) { + std::fprintf(stderr, + "[server] --draft-ipc-work-dir and --draft-ipc-ring-cap require " + "--draft-ipc-bin\n"); + return false; + } if (target != draft) { + if (!bargs.remote_draft.enabled()) { + std::fprintf(stderr, + "[server] mixed target/draft backends require --draft-ipc-bin " + "(target=%s draft=%s)\n", + placement_backend_name(target), placement_backend_name(draft)); + return false; + } + if (!bargs.draft_path) { + std::fprintf(stderr, + "[server] mixed target/draft backends require --draft \n"); + return false; + } + } else if (bargs.remote_draft.enabled()) { std::fprintf(stderr, - "[server] mixed target/draft backends are not implemented in the " - "native server yet (target=%s draft=%s)\n", + "[server] --draft-ipc-bin is only needed for mixed target/draft " + "backends (target=%s draft=%s)\n", placement_backend_name(target), placement_backend_name(draft)); return false; + } else if (!placement_backend_supported(bargs.draft_device.backend)) { + std::fprintf(stderr, + "[server] --draft-device=%s is unsupported in this binary " + "(compiled backend: %s)\n", + placement_device_name(bargs.draft_device).c_str(), + placement_backend_name(compiled)); + return false; } if (!bargs.device.layer_split_gpus.empty()) { std::fprintf(stderr, @@ -111,6 +129,9 @@ static void print_usage(const char * prog) { " --max-tokens Default max output tokens (default: 4096)\n" " --target-device Target device (default: auto:0)\n" " --draft-device Draft device (default: auto:0)\n" + " --draft-ipc-bin Remote draft IPC daemon for mixed backends\n" + " --draft-ipc-work-dir Remote draft IPC scratch directory\n" + " --draft-ipc-ring-cap Remote draft feature ring capacity\n" " --target-devices Reserved layer-split devices, e.g. cuda:0,cuda:1\n" " --target-layer-split Reserved layer-split weights\n" " --peer-access Enable peer access for multi-GPU placement\n" @@ -197,6 +218,16 @@ int main(int argc, char ** argv) { std::fprintf(stderr, "[server] bad --draft-device value (expected backend:gpu)\n"); return 2; } + } else if (std::strcmp(argv[i], "--draft-ipc-bin") == 0 && i + 1 < argc) { + bargs.remote_draft.ipc_bin = argv[++i]; + } else if (std::strcmp(argv[i], "--draft-ipc-work-dir") == 0 && i + 1 < argc) { + bargs.remote_draft.work_dir = argv[++i]; + } else if (std::strcmp(argv[i], "--draft-ipc-ring-cap") == 0 && i + 1 < argc) { + bargs.remote_draft.ring_cap = std::atoi(argv[++i]); + if (bargs.remote_draft.ring_cap <= 0) { + std::fprintf(stderr, "[server] bad --draft-ipc-ring-cap value\n"); + return 2; + } } else if (std::strcmp(argv[i], "--target-devices") == 0 && i + 1 < argc) { if (target_device_seen) { std::fprintf(stderr, "[server] --target-devices conflicts with --target-device\n"); @@ -301,6 +332,21 @@ int main(int argc, char ** argv) { if (!validate_server_placement(bargs)) return 2; + if (bargs.remote_draft.enabled()) { + const std::string arch = detect_arch(bargs.model_path); + if (arch.empty()) { + std::fprintf(stderr, + "[server] failed to detect model architecture for remote draft validation\n"); + return 1; + } + if (!arch_supports_remote_draft(arch)) { + std::fprintf(stderr, + "[server] model architecture '%s' does not support remote draft execution\n", + arch.c_str()); + return 2; + } + } + // Sync max_ctx: if --max-ctx was not provided, use the backend's default. // This prevents the HTTP server from accepting prompts larger than the // KV cache the backend actually allocates. @@ -376,6 +422,12 @@ int main(int argc, char ** argv) { std::fprintf(stderr, "[server] backend creation failed\n"); return 1; } + if (bargs.remote_draft.enabled() && !backend->supports_remote_draft()) { + std::fprintf(stderr, + "[server] detected model backend does not support remote draft execution\n"); + backend->shutdown(); + return 2; + } // Start HTTP server. std::fprintf(stderr, "\n"); @@ -399,6 +451,18 @@ int main(int argc, char ** argv) { } std::fprintf(stderr, "[server] │ draft_device = %s\n", placement_device_name(bargs.draft_device).c_str()); + std::fprintf(stderr, "[server] │ draft_exec = %s\n", + bargs.remote_draft.enabled() ? "remote-ipc" : "local"); + if (bargs.remote_draft.enabled()) { + std::fprintf(stderr, "[server] │ draft_ipc_bin = %s\n", + bargs.remote_draft.ipc_bin.c_str()); + if (!bargs.remote_draft.work_dir.empty()) { + std::fprintf(stderr, "[server] │ draft_ipc_dir = %s\n", + bargs.remote_draft.work_dir.c_str()); + } + std::fprintf(stderr, "[server] │ draft_ipc_cap = %d\n", + bargs.remote_draft.ring_cap); + } std::fprintf(stderr, "[server] │ peer_access = %s\n", bargs.device.peer_access ? "ON" : "off"); std::fprintf(stderr, "[server] │ chunk = %d\n", bargs.chunk); From ad8662b03527855464526b2423c675f203c831fa Mon Sep 17 00:00:00 2001 From: weicj Date: Sat, 23 May 2026 05:04:46 +0800 Subject: [PATCH 2/3] refactor(server): add backend-neutral IPC process layer --- dflash/CMakeLists.txt | 19 +- dflash/docs/ARCHITECTURE.md | 16 +- dflash/docs/MIXED_BACKEND.md | 9 +- dflash/src/common/backend_ipc.cpp | 179 ++++++++++++++++++ dflash/src/common/backend_ipc.h | 65 +++++++ dflash/src/common/dflash_draft_ipc.cpp | 143 +++----------- dflash/src/common/dflash_draft_ipc.h | 23 +-- dflash/src/common/dflash_draft_ipc_daemon.cpp | 4 +- dflash/src/ipc/backend_ipc_main.cpp | 70 +++++++ dflash/src/ipc/dflash_draft_ipc_main.cpp | 45 ----- dflash/src/server/server_main.cpp | 2 +- dflash/test/test_server_unit.cpp | 25 +++ 12 files changed, 396 insertions(+), 204 deletions(-) create mode 100644 dflash/src/common/backend_ipc.cpp create mode 100644 dflash/src/common/backend_ipc.h create mode 100644 dflash/src/ipc/backend_ipc_main.cpp delete mode 100644 dflash/src/ipc/dflash_draft_ipc_main.cpp diff --git a/dflash/CMakeLists.txt b/dflash/CMakeLists.txt index c84a5f56f..d7e95981b 100644 --- a/dflash/CMakeLists.txt +++ b/dflash/CMakeLists.txt @@ -235,6 +235,7 @@ add_library(dflash_common STATIC src/laguna/laguna_target_graph.cpp src/laguna/laguna_daemon.cpp src/laguna/laguna_backend.cpp + src/common/backend_ipc.cpp src/common/dflash_feature_ring.cpp src/common/dflash_capture.cpp src/common/dflash_draft_ipc.cpp @@ -671,24 +672,24 @@ if(DFLASH27B_TESTS) endif() endif() - if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/src/ipc/dflash_draft_ipc_main.cpp") - add_executable(dflash_draft_ipc_daemon - src/ipc/dflash_draft_ipc_main.cpp + if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/src/ipc/backend_ipc_main.cpp") + add_executable(backend_ipc_daemon + src/ipc/backend_ipc_main.cpp ) - target_include_directories(dflash_draft_ipc_daemon PRIVATE ${DFLASH27B_SRC_INCLUDE_DIRS}) + target_include_directories(backend_ipc_daemon PRIVATE ${DFLASH27B_SRC_INCLUDE_DIRS}) if(DFLASH27B_GPU_BACKEND STREQUAL "hip") - target_compile_definitions(dflash_draft_ipc_daemon PRIVATE DFLASH27B_BACKEND_HIP=1 GGML_USE_HIP) + target_compile_definitions(backend_ipc_daemon PRIVATE DFLASH27B_BACKEND_HIP=1 GGML_USE_HIP) else() - target_compile_definitions(dflash_draft_ipc_daemon PRIVATE + target_compile_definitions(backend_ipc_daemon PRIVATE DFLASH27B_BACKEND_CUDA=1 DFLASH27B_CUDA_MIN_SM=${_dflash_cuda_min_sm}) endif() - target_link_libraries(dflash_draft_ipc_daemon PRIVATE dflash_common ggml ${DFLASH27B_GGML_BACKEND_TARGET} pthread) + target_link_libraries(backend_ipc_daemon PRIVATE dflash_common ggml ${DFLASH27B_GGML_BACKEND_TARGET} pthread) if(DFLASH27B_GPU_BACKEND STREQUAL "cuda") find_package(CUDAToolkit REQUIRED) - target_link_libraries(dflash_draft_ipc_daemon PRIVATE CUDA::cudart) + target_link_libraries(backend_ipc_daemon PRIVATE CUDA::cudart) else() - target_link_libraries(dflash_draft_ipc_daemon PRIVATE hip::host) + target_link_libraries(backend_ipc_daemon PRIVATE hip::host) endif() endif() diff --git a/dflash/docs/ARCHITECTURE.md b/dflash/docs/ARCHITECTURE.md index 5dbbda0dd..1583aed71 100644 --- a/dflash/docs/ARCHITECTURE.md +++ b/dflash/docs/ARCHITECTURE.md @@ -42,6 +42,7 @@ dflash/src/ │ ├── snapshot_backend.h # Platform-aware snapshot backend selection │ ├── dflash_target.h # DFlashTarget interface (spec decode) │ ├── daemon_loop.{h,cpp} # Generic stdin/stdout daemon loop +│ ├── backend_ipc.{h,cpp} # Generic backend IPC process lifecycle │ ├── device_placement.h # Multi-GPU placement config │ ├── gguf_inspect.{h,cpp}# Read arch + layer count from GGUF │ ├── layer_split_utils.{h,cpp} # compute_layer_ranges() @@ -49,7 +50,7 @@ dflash/src/ │ ├── dflash_feature_ring.{h,cpp} # DraftFeatureMirror + ring copy helpers │ ├── dflash_capture.{h,cpp} # target_capture_index() helper │ ├── dflash_draft_ipc.{h,cpp} # DFlash draft IPC client + remote copy -│ ├── dflash_draft_ipc_daemon.cpp # Generic DFlash draft IPC daemon body +│ ├── dflash_draft_ipc_daemon.cpp # DFlash draft mode for backend_ipc_daemon │ ├── dflash_draft_graph.{h,cpp} # Universal build_draft_step (DFlash draft graph) │ ├── dflash_spec_decode.{h,cpp} # Generic spec-decode loop over DFlashTarget │ ├── ddtree.{h,cpp} # Dynamic Draft Tree algorithm @@ -240,15 +241,14 @@ Key components: Bridge qwen35 internals (`TargetWeights`, `TargetCache`, `TargetLayerSplitShard`) to the generic `DFlashTarget` interface so the shared spec-decode loop can drive verification. -- **Feature transfer + draft daemon** (`common/dflash_feature_ring.{h,cpp}`, - `common/dflash_capture.{h,cpp}`, `common/dflash_draft_ipc.{h,cpp}`, - `common/dflash_draft_ipc_daemon.cpp`): +- **Feature transfer + backend IPC daemon** (`common/backend_ipc.{h,cpp}`, + `common/dflash_feature_ring.{h,cpp}`, `common/dflash_capture.{h,cpp}`, + `common/dflash_draft_ipc.{h,cpp}`, `common/dflash_draft_ipc_daemon.cpp`): Move captured target activations into the draft-side ring buffer (`DraftFeatureMirror`) and ship them across processes/GPUs. The IPC - client, parent-side feature-slice helper, and the daemon body itself - all live in `common/` and are reusable by any DFlash target architecture - (the DFlash draft model is a single universal Qwen3-style network shared - across every target). + process lifecycle is shared through `backend_ipc`; the DFlash draft client, + parent-side feature-slice helper, and daemon mode stay on top of that common + process layer and remain reusable by any DFlash target architecture. ### Qwen3Backend, Gemma4Backend, LagunaBackend diff --git a/dflash/docs/MIXED_BACKEND.md b/dflash/docs/MIXED_BACKEND.md index 93e456dde..8daeb60fc 100644 --- a/dflash/docs/MIXED_BACKEND.md +++ b/dflash/docs/MIXED_BACKEND.md @@ -16,12 +16,12 @@ kept at host-data or process boundaries: ```bash cmake -S . -B build-cuda -DCMAKE_BUILD_TYPE=Release \ -DDFLASH27B_GPU_BACKEND=cuda -cmake --build build-cuda --target pflash_daemon test_dflash -j +cmake --build build-cuda --target pflash_daemon test_dflash backend_ipc_daemon -j cmake -S . -B build-hip -DCMAKE_BUILD_TYPE=Release \ -DDFLASH27B_GPU_BACKEND=hip \ -DDFLASH27B_HIP_ARCHITECTURES= -cmake --build build-hip --target pflash_daemon test_dflash -j +cmake --build build-hip --target pflash_daemon test_dflash backend_ipc_daemon -j ``` ## PFlash phase split @@ -69,13 +69,14 @@ python scripts/phase_split_dual_gpu.py run-prompt \ ## DFlash draft split -For DFlash, the target process can launch a separate draft IPC daemon from a +For DFlash, the target process can launch a separate backend IPC daemon from a different backend build. The target process keeps target execution and any target layer split inside its own backend binary. Use these `test_dflash` options for the target process: -- `--draft-ipc-bin ` points to the other backend's `test_dflash` binary. +- `--draft-ipc-bin ` points to the other backend's `backend_ipc_daemon` + binary. - `--draft-ipc-gpu ` selects the draft daemon device in that backend's visible-device namespace. - `--draft-ipc-work-dir ` selects where temporary IPC payload files are diff --git a/dflash/src/common/backend_ipc.cpp b/dflash/src/common/backend_ipc.cpp new file mode 100644 index 000000000..f96820177 --- /dev/null +++ b/dflash/src/common/backend_ipc.cpp @@ -0,0 +1,179 @@ +// backend_ipc.cpp - generic backend IPC process launcher. + +#include "backend_ipc.h" +#include "io_utils.h" + +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +# include +# include +# include +# include +# include +#endif + +namespace dflash::common { + +const char * backend_ipc_mode_name(BackendIpcMode mode) { + switch (mode) { + case BackendIpcMode::DFlashDraft: return "dflash-draft"; + } + return "unknown"; +} + +bool parse_backend_ipc_mode(const std::string & value, BackendIpcMode & out) { + if (value == "dflash-draft") { + out = BackendIpcMode::DFlashDraft; + return true; + } + return false; +} + +bool BackendIpcProcess::start(const BackendIpcLaunchConfig & cfg) { +#if defined(_WIN32) + (void)cfg; + std::fprintf(stderr, "Backend IPC is only implemented on POSIX hosts\n"); + return false; +#else + close(); + if (cfg.bin.empty() || cfg.payload_path.empty()) return false; + if (!init_work_dir(cfg.work_dir)) return false; + + int cmd_pipe[2] = {-1, -1}; + int stream_pipe[2] = {-1, -1}; + if (::pipe(cmd_pipe) != 0 || ::pipe(stream_pipe) != 0) { + std::fprintf(stderr, "backend-ipc pipe failed: %s\n", std::strerror(errno)); + if (cmd_pipe[0] >= 0) ::close(cmd_pipe[0]); + if (cmd_pipe[1] >= 0) ::close(cmd_pipe[1]); + if (stream_pipe[0] >= 0) ::close(stream_pipe[0]); + if (stream_pipe[1] >= 0) ::close(stream_pipe[1]); + return false; + } + + pid_ = ::fork(); + if (pid_ < 0) { + std::fprintf(stderr, "backend-ipc fork failed: %s\n", std::strerror(errno)); + ::close(cmd_pipe[0]); ::close(cmd_pipe[1]); + ::close(stream_pipe[0]); ::close(stream_pipe[1]); + pid_ = -1; + return false; + } + if (pid_ == 0) { + ::dup2(cmd_pipe[0], STDIN_FILENO); + ::close(cmd_pipe[0]); + ::close(cmd_pipe[1]); + ::close(stream_pipe[0]); + + std::vector argv_storage; + argv_storage.reserve(cfg.args.size() + 5); + argv_storage.emplace_back(cfg.bin); + argv_storage.emplace_back( + std::string("--backend-ipc-mode=") + backend_ipc_mode_name(cfg.mode)); + argv_storage.emplace_back(cfg.payload_path); + for (const std::string & arg : cfg.args) argv_storage.emplace_back(arg); + argv_storage.emplace_back("--stream-fd=" + std::to_string(stream_pipe[1])); + + std::vector argv; + argv.reserve(argv_storage.size() + 1); + for (std::string & arg : argv_storage) argv.push_back(arg.data()); + argv.push_back(nullptr); + ::execv(cfg.bin.c_str(), argv.data()); + std::fprintf(stderr, "backend-ipc exec failed: %s: %s\n", + cfg.bin.c_str(), std::strerror(errno)); + _exit(127); + } + + ::close(cmd_pipe[0]); + ::close(stream_pipe[1]); + stream_fd_ = stream_pipe[0]; + cmd_ = ::fdopen(cmd_pipe[1], "w"); + if (!cmd_) { + std::fprintf(stderr, "backend-ipc fdopen failed: %s\n", std::strerror(errno)); + ::close(cmd_pipe[1]); + close(); + return false; + } + int32_t status = -1; + if (!read_exact_fd(stream_fd_, &status, sizeof(status)) || status != 0) { + std::fprintf(stderr, "backend-ipc daemon did not become ready (status=%d)\n", status); + close(); + return false; + } + active_ = true; + std::printf("[backend-ipc] ready mode=%s bin=%s work_dir=%s\n", + backend_ipc_mode_name(cfg.mode), cfg.bin.c_str(), work_dir_.c_str()); + return true; +#endif +} + +void BackendIpcProcess::close() { +#if !defined(_WIN32) + if (cmd_) { + std::fclose(cmd_); + cmd_ = nullptr; + } + if (stream_fd_ >= 0) { + ::close(stream_fd_); + stream_fd_ = -1; + } + if (pid_ > 0) { + int status = 0; + ::waitpid(pid_, &status, 0); + pid_ = -1; + } + if (owns_work_dir_ && !work_dir_.empty()) { + ::rmdir(work_dir_.c_str()); + } +#endif + active_ = false; + owns_work_dir_ = false; + work_dir_.clear(); + seq_ = 0; +} + +std::string BackendIpcProcess::next_path(const char * prefix) { + return work_dir_ + "/" + prefix + "_" + std::to_string(seq_++) + ".bin"; +} + +#if !defined(_WIN32) +bool BackendIpcProcess::init_work_dir(const std::string & requested) { + if (!requested.empty()) { + work_dir_ = requested; + owns_work_dir_ = false; + if (::mkdir(work_dir_.c_str(), 0700) != 0) { + if (errno != EEXIST) { + std::fprintf(stderr, "backend-ipc mkdir failed: %s: %s\n", + work_dir_.c_str(), std::strerror(errno)); + return false; + } + struct stat st; + if (::stat(work_dir_.c_str(), &st) != 0 || !S_ISDIR(st.st_mode)) { + std::fprintf(stderr, "backend-ipc work_dir is not a directory: %s\n", + work_dir_.c_str()); + return false; + } + } + return true; + } + const char * tmp = std::getenv("TMPDIR"); + std::string templ = std::string(tmp && *tmp ? tmp : "/tmp") + + "/backend-ipc-XXXXXX"; + std::vector buf(templ.begin(), templ.end()); + buf.push_back('\0'); + char * dir = ::mkdtemp(buf.data()); + if (!dir) { + std::fprintf(stderr, "backend-ipc mkdtemp failed: %s\n", std::strerror(errno)); + return false; + } + work_dir_ = dir; + owns_work_dir_ = true; + return true; +} +#endif + +} // namespace dflash::common diff --git a/dflash/src/common/backend_ipc.h b/dflash/src/common/backend_ipc.h new file mode 100644 index 000000000..b5089bdf9 --- /dev/null +++ b/dflash/src/common/backend_ipc.h @@ -0,0 +1,65 @@ +// backend_ipc.h - generic backend IPC process launcher. +// +// Owns the out-of-process backend daemon lifecycle: fork/exec, command pipe, +// binary status stream, and scratch work directory. Individual IPC modes keep +// their own payload protocol on top of this process wrapper. + +#pragma once + +#include +#include +#include + +#if !defined(_WIN32) +# include +#endif + +namespace dflash::common { + +enum class BackendIpcMode { + DFlashDraft, +}; + +const char * backend_ipc_mode_name(BackendIpcMode mode); +bool parse_backend_ipc_mode(const std::string & value, BackendIpcMode & out); + +struct BackendIpcLaunchConfig { + std::string bin; + BackendIpcMode mode = BackendIpcMode::DFlashDraft; + std::string payload_path; + std::vector args; + std::string work_dir; +}; + +class BackendIpcProcess { +public: + BackendIpcProcess() = default; + BackendIpcProcess(const BackendIpcProcess &) = delete; + BackendIpcProcess & operator=(const BackendIpcProcess &) = delete; + ~BackendIpcProcess() { close(); } + + bool start(const BackendIpcLaunchConfig & cfg); + void close(); + + bool active() const { return active_; } + FILE * command_stream() const { return cmd_; } + int stream_fd() const { return stream_fd_; } + const std::string & work_dir() const { return work_dir_; } + + std::string next_path(const char * prefix); + +private: +#if !defined(_WIN32) + bool init_work_dir(const std::string & requested); + + pid_t pid_ = -1; +#endif + FILE * cmd_ = nullptr; + int stream_fd_ = -1; + std::string work_dir_; + int seq_ = 0; + bool owns_work_dir_ = false; + bool active_ = false; +}; + +} // namespace dflash::common diff --git a/dflash/src/common/dflash_draft_ipc.cpp b/dflash/src/common/dflash_draft_ipc.cpp index 0cab347b8..0f8542379 100644 --- a/dflash/src/common/dflash_draft_ipc.cpp +++ b/dflash/src/common/dflash_draft_ipc.cpp @@ -1,4 +1,4 @@ -// dflash_draft_ipc.cpp — DFlashDraftIpcClient + remote feature copy. +// dflash_draft_ipc.cpp - DFlashDraftIpcClient + remote feature copy. // // Target-agnostic portion of the DFlash draft IPC: parent-side client that // spawns the daemon, sends commands, and the row-extraction helper that @@ -34,65 +34,21 @@ bool DFlashDraftIpcClient::start( #else close(); if (bin.empty() || draft_path.empty() || ring_cap <= 0) return false; - if (!init_work_dir(work_dir)) return false; - - int cmd_pipe[2] = {-1, -1}; - int stream_pipe[2] = {-1, -1}; - if (::pipe(cmd_pipe) != 0 || ::pipe(stream_pipe) != 0) { - std::fprintf(stderr, "draft-ipc pipe failed: %s\n", std::strerror(errno)); - if (cmd_pipe[0] >= 0) ::close(cmd_pipe[0]); - if (cmd_pipe[1] >= 0) ::close(cmd_pipe[1]); - if (stream_pipe[0] >= 0) ::close(stream_pipe[0]); - if (stream_pipe[1] >= 0) ::close(stream_pipe[1]); - return false; - } - - pid_ = ::fork(); - if (pid_ < 0) { - std::fprintf(stderr, "draft-ipc fork failed: %s\n", std::strerror(errno)); - ::close(cmd_pipe[0]); ::close(cmd_pipe[1]); - ::close(stream_pipe[0]); ::close(stream_pipe[1]); - pid_ = -1; - return false; - } - if (pid_ == 0) { - ::dup2(cmd_pipe[0], STDIN_FILENO); - ::close(cmd_pipe[0]); - ::close(cmd_pipe[1]); - ::close(stream_pipe[0]); - - const std::string cap_arg = "--ring-cap=" + std::to_string(ring_cap); - const std::string gpu_arg = "--draft-gpu=" + std::to_string(std::max(0, draft_gpu)); - const std::string fd_arg = "--stream-fd=" + std::to_string(stream_pipe[1]); - ::execl(bin.c_str(), bin.c_str(), - "--draft-ipc-daemon", draft_path.c_str(), - cap_arg.c_str(), gpu_arg.c_str(), fd_arg.c_str(), - (char *)nullptr); - std::fprintf(stderr, "draft-ipc exec failed: %s: %s\n", - bin.c_str(), std::strerror(errno)); - _exit(127); - } - - ::close(cmd_pipe[0]); - ::close(stream_pipe[1]); - stream_fd_ = stream_pipe[0]; - cmd_ = ::fdopen(cmd_pipe[1], "w"); - if (!cmd_) { - std::fprintf(stderr, "draft-ipc fdopen failed: %s\n", std::strerror(errno)); - ::close(cmd_pipe[1]); - close(); - return false; - } - int32_t status = -1; - if (!read_exact_fd(stream_fd_, &status, sizeof(status)) || status != 0) { - std::fprintf(stderr, "draft-ipc daemon did not become ready (status=%d)\n", status); - close(); + BackendIpcLaunchConfig launch; + launch.bin = bin; + launch.mode = BackendIpcMode::DFlashDraft; + launch.payload_path = draft_path; + launch.work_dir = work_dir; + 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)) { + std::fprintf(stderr, "draft-ipc backend process start failed\n"); return false; } ring_cap_ = ring_cap; 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, work_dir_.c_str()); + bin.c_str(), draft_gpu, ring_cap, process_.work_dir().c_str()); return true; #endif } @@ -106,19 +62,21 @@ bool DFlashDraftIpcClient::send_feature_slice( (void)capture_idx; (void)start_pos; (void)n_tokens; (void)slice; return false; #else - if (!active_ || !cmd_ || n_tokens <= 0) return false; + FILE * cmd = process_.command_stream(); + const int stream_fd = process_.stream_fd(); + if (!active_ || !cmd || stream_fd < 0 || n_tokens <= 0) return false; const size_t expected = (size_t)n_tokens * hidden_size_; if (slice.size() != expected) return false; - const std::string path = next_path("feature"); + const std::string path = process_.next_path("feature"); if (!write_binary_file(path, slice.data(), slice.size() * sizeof(float))) { std::fprintf(stderr, "draft-ipc write feature failed: %s\n", path.c_str()); return false; } - std::fprintf(cmd_, "feature_slice %d %d %d %s\n", + std::fprintf(cmd, "feature_slice %d %d %d %s\n", capture_idx, start_pos, n_tokens, path.c_str()); - std::fflush(cmd_); + std::fflush(cmd); int32_t status = -1; - const bool ok = read_exact_fd(stream_fd_, &status, sizeof(status)) && status == 0; + const bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) && status == 0; std::remove(path.c_str()); if (!ok) { std::fprintf(stderr, "draft-ipc feature_slice failed status=%d\n", status); @@ -136,22 +94,24 @@ bool DFlashDraftIpcClient::propose( (void)committed; (void)ctx_len; (void)noise_embed; (void)hidden_out; return false; #else - if (!active_ || !cmd_ || ctx_len <= 0) return false; + FILE * cmd = process_.command_stream(); + const int stream_fd = process_.stream_fd(); + if (!active_ || !cmd || stream_fd < 0 || ctx_len <= 0) return false; const size_t noise_expected = (size_t)hidden_size_ * block_size_; if (noise_embed.size() != noise_expected) return false; - const std::string path = next_path("noise"); + const std::string path = process_.next_path("noise"); if (!write_binary_file(path, noise_embed.data(), noise_embed.size() * sizeof(float))) { std::fprintf(stderr, "draft-ipc write noise failed: %s\n", path.c_str()); return false; } - std::fprintf(cmd_, "propose %d %d %s\n", committed, ctx_len, path.c_str()); - std::fflush(cmd_); + std::fprintf(cmd, "propose %d %d %s\n", committed, ctx_len, path.c_str()); + std::fflush(cmd); int32_t status = -1; - bool ok = read_exact_fd(stream_fd_, &status, sizeof(status)) && status == 0; + 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(), + ok = read_exact_fd(stream_fd, hidden_out.data(), hidden_out.size() * sizeof(float)); } std::remove(path.c_str()); @@ -163,60 +123,11 @@ bool DFlashDraftIpcClient::propose( } void DFlashDraftIpcClient::close() { -#if !defined(_WIN32) - if (cmd_) { - std::fclose(cmd_); - cmd_ = nullptr; - } - if (stream_fd_ >= 0) { - ::close(stream_fd_); - stream_fd_ = -1; - } - if (pid_ > 0) { - int status = 0; - ::waitpid(pid_, &status, 0); - pid_ = -1; - } - if (owns_work_dir_ && !work_dir_.empty()) { - ::rmdir(work_dir_.c_str()); - } -#endif + process_.close(); active_ = false; ring_cap_ = 0; } -#if !defined(_WIN32) -bool DFlashDraftIpcClient::init_work_dir(const std::string & requested) { - if (!requested.empty()) { - work_dir_ = requested; - owns_work_dir_ = false; - if (::mkdir(work_dir_.c_str(), 0700) != 0 && errno != EEXIST) { - std::fprintf(stderr, "draft-ipc mkdir failed: %s: %s\n", - work_dir_.c_str(), std::strerror(errno)); - return false; - } - return true; - } - const char * tmp = std::getenv("TMPDIR"); - std::string templ = std::string(tmp && *tmp ? tmp : "/tmp") + - "/dflash-draft-ipc-XXXXXX"; - std::vector buf(templ.begin(), templ.end()); - buf.push_back('\0'); - char * dir = ::mkdtemp(buf.data()); - if (!dir) { - std::fprintf(stderr, "draft-ipc mkdtemp failed: %s\n", std::strerror(errno)); - return false; - } - work_dir_ = dir; - owns_work_dir_ = true; - return true; -} - -std::string DFlashDraftIpcClient::next_path(const char * prefix) { - return work_dir_ + "/" + prefix + "_" + std::to_string(seq_++) + ".bin"; -} -#endif - // ── Remote draft feature copy helper ──────────────────────────────── bool copy_capture_slice_to_remote_draft( diff --git a/dflash/src/common/dflash_draft_ipc.h b/dflash/src/common/dflash_draft_ipc.h index f5eb47ac4..f719555ec 100644 --- a/dflash/src/common/dflash_draft_ipc.h +++ b/dflash/src/common/dflash_draft_ipc.h @@ -1,4 +1,4 @@ -// dflash_draft_ipc.h — DFlash draft model IPC client + daemon entry. +// dflash_draft_ipc.h - DFlash draft IPC mode. // // The draft IPC mechanism spawns a child process running the draft model on // a separate GPU. Communication is via stdin commands + a stream pipe for @@ -12,6 +12,7 @@ #pragma once +#include "backend_ipc.h" #include "dflash27b.h" #include "io_utils.h" @@ -23,14 +24,6 @@ #include #include -#if !defined(_WIN32) -# include -# include -# include -# include -# include -#endif - namespace dflash::common { // ── IPC Client (parent process) ───────────────────────────────────── @@ -70,17 +63,7 @@ class DFlashDraftIpcClient { void close(); private: -#if !defined(_WIN32) - bool init_work_dir(const std::string & requested); - std::string next_path(const char * prefix); - - pid_t pid_ = -1; - FILE * cmd_ = nullptr; - int stream_fd_ = -1; - std::string work_dir_; - int seq_ = 0; - bool owns_work_dir_ = false; -#endif + BackendIpcProcess process_; bool active_ = false; int ring_cap_ = 0; int hidden_size_ = DFLASH27B_TARGET_HIDDEN; diff --git a/dflash/src/common/dflash_draft_ipc_daemon.cpp b/dflash/src/common/dflash_draft_ipc_daemon.cpp index 176798f85..9cbc019f7 100644 --- a/dflash/src/common/dflash_draft_ipc_daemon.cpp +++ b/dflash/src/common/dflash_draft_ipc_daemon.cpp @@ -36,7 +36,9 @@ int run_dflash_draft_ipc_daemon(const char * draft_path, return 2; #else if (!draft_path || ring_cap <= 0 || stream_fd < 0) { - std::fprintf(stderr, "usage: test_dflash --draft-ipc-daemon --ring-cap=N --stream-fd=FD [--draft-gpu=N]\n"); + std::fprintf(stderr, + "usage: backend_ipc_daemon --backend-ipc-mode=dflash-draft " + " --ring-cap=N --stream-fd=FD [--draft-gpu=N]\n"); return 2; } diff --git a/dflash/src/ipc/backend_ipc_main.cpp b/dflash/src/ipc/backend_ipc_main.cpp new file mode 100644 index 000000000..2a6c9051d --- /dev/null +++ b/dflash/src/ipc/backend_ipc_main.cpp @@ -0,0 +1,70 @@ +// Standalone backend IPC daemon entry point. + +#include "backend_ipc.h" +#include "dflash_draft_ipc.h" + +#include +#include +#include +#include + +using namespace dflash::common; + +int main(int argc, char ** argv) { + BackendIpcMode mode = BackendIpcMode::DFlashDraft; + const char * payload_path = nullptr; + int arg_begin = 0; + + if (argc >= 3 && std::strncmp(argv[1], "--backend-ipc-mode=", 19) == 0) { + std::string mode_name(argv[1] + 19); + if (!parse_backend_ipc_mode(mode_name, mode)) { + std::fprintf(stderr, "[backend-ipc-daemon] unknown mode: %s\n", + mode_name.c_str()); + return 2; + } + payload_path = argv[2]; + arg_begin = 3; + } else if (argc >= 3 && std::strcmp(argv[1], "--backend-ipc-mode") == 0) { + if (!parse_backend_ipc_mode(argv[2], mode) || argc < 4) { + std::fprintf(stderr, "[backend-ipc-daemon] bad --backend-ipc-mode\n"); + return 2; + } + payload_path = argv[3]; + arg_begin = 4; + } else { + std::fprintf(stderr, + "usage: %s --backend-ipc-mode=dflash-draft " + "--ring-cap=N --stream-fd=FD [--draft-gpu=N]\n", + argv[0]); + return 2; + } + + int ring_cap = 4096; + int draft_gpu = 0; + int stream_fd = -1; + for (int i = arg_begin; i < argc; i++) { + if (std::strncmp(argv[i], "--ring-cap=", 11) == 0) { + ring_cap = std::atoi(argv[i] + 11); + } else if (std::strcmp(argv[i], "--ring-cap") == 0) { + if (i + 1 < argc) ring_cap = std::atoi(argv[++i]); + } else if (std::strncmp(argv[i], "--draft-gpu=", 12) == 0) { + draft_gpu = std::max(0, std::atoi(argv[i] + 12)); + } else if (std::strcmp(argv[i], "--draft-gpu") == 0) { + if (i + 1 < argc) draft_gpu = std::max(0, std::atoi(argv[++i])); + } else if (std::strncmp(argv[i], "--stream-fd=", 12) == 0) { + stream_fd = std::atoi(argv[i] + 12); + } else if (std::strcmp(argv[i], "--stream-fd") == 0) { + if (i + 1 < argc) stream_fd = std::atoi(argv[++i]); + } else { + std::fprintf(stderr, "[backend-ipc-daemon] unknown option: %s\n", argv[i]); + return 2; + } + } + + switch (mode) { + case BackendIpcMode::DFlashDraft: + return run_dflash_draft_ipc_daemon(payload_path, ring_cap, draft_gpu, stream_fd); + } + std::fprintf(stderr, "[backend-ipc-daemon] unsupported mode\n"); + return 2; +} diff --git a/dflash/src/ipc/dflash_draft_ipc_main.cpp b/dflash/src/ipc/dflash_draft_ipc_main.cpp deleted file mode 100644 index 97034008a..000000000 --- a/dflash/src/ipc/dflash_draft_ipc_main.cpp +++ /dev/null @@ -1,45 +0,0 @@ -// Standalone DFlash draft IPC daemon entry point. - -#include "dflash_draft_ipc.h" - -#include -#include -#include -#include - -using namespace dflash::common; - -int main(int argc, char ** argv) { - if (argc < 3 || std::strcmp(argv[1], "--draft-ipc-daemon") != 0) { - std::fprintf(stderr, - "usage: %s --draft-ipc-daemon " - "--ring-cap=N --stream-fd=FD [--draft-gpu=N]\n", - argv[0]); - return 2; - } - - const char * draft_path = argv[2]; - int ring_cap = 4096; - int draft_gpu = 0; - int stream_fd = -1; - for (int i = 3; i < argc; i++) { - if (std::strncmp(argv[i], "--ring-cap=", 11) == 0) { - ring_cap = std::atoi(argv[i] + 11); - } else if (std::strcmp(argv[i], "--ring-cap") == 0) { - if (i + 1 < argc) ring_cap = std::atoi(argv[++i]); - } else if (std::strncmp(argv[i], "--draft-gpu=", 12) == 0) { - draft_gpu = std::max(0, std::atoi(argv[i] + 12)); - } else if (std::strcmp(argv[i], "--draft-gpu") == 0) { - if (i + 1 < argc) draft_gpu = std::max(0, std::atoi(argv[++i])); - } else if (std::strncmp(argv[i], "--stream-fd=", 12) == 0) { - stream_fd = std::atoi(argv[i] + 12); - } else if (std::strcmp(argv[i], "--stream-fd") == 0) { - if (i + 1 < argc) stream_fd = std::atoi(argv[++i]); - } else { - std::fprintf(stderr, "[draft-ipc-daemon] unknown option: %s\n", argv[i]); - return 2; - } - } - - return run_dflash_draft_ipc_daemon(draft_path, ring_cap, draft_gpu, stream_fd); -} diff --git a/dflash/src/server/server_main.cpp b/dflash/src/server/server_main.cpp index 166b41df2..a15f93c47 100644 --- a/dflash/src/server/server_main.cpp +++ b/dflash/src/server/server_main.cpp @@ -129,7 +129,7 @@ static void print_usage(const char * prog) { " --max-tokens Default max output tokens (default: 4096)\n" " --target-device Target device (default: auto:0)\n" " --draft-device Draft device (default: auto:0)\n" - " --draft-ipc-bin Remote draft IPC daemon for mixed backends\n" + " --draft-ipc-bin Remote backend IPC daemon for mixed backends\n" " --draft-ipc-work-dir Remote draft IPC scratch directory\n" " --draft-ipc-ring-cap Remote draft feature ring capacity\n" " --target-devices Reserved layer-split devices, e.g. cuda:0,cuda:1\n" diff --git a/dflash/test/test_server_unit.cpp b/dflash/test/test_server_unit.cpp index 6cda54c5c..42ee9150b 100644 --- a/dflash/test/test_server_unit.cpp +++ b/dflash/test/test_server_unit.cpp @@ -17,6 +17,7 @@ #include "server/http_server.h" #include "server/chat_template.h" #include "common/sampler.h" +#include "common/backend_ipc.h" #include #include @@ -26,6 +27,7 @@ #include #include #include +#include #include #include #include @@ -1238,6 +1240,28 @@ static void test_disk_cache_save_below_min_tokens() { rm_rf(dir); } +static void test_backend_ipc_rejects_file_work_dir() { + const std::string file_path = "/tmp/dflash_test_backend_ipc_work_dir_file"; + unlink(file_path.c_str()); + int fd = open(file_path.c_str(), O_CREAT | O_TRUNC | O_WRONLY, 0600); + TEST_ASSERT(fd >= 0); + if (fd >= 0) { + const char payload[] = "not a dir"; + (void)write(fd, payload, sizeof(payload) - 1); + close(fd); + } + + BackendIpcLaunchConfig cfg; + cfg.bin = "/bin/true"; + cfg.payload_path = "/tmp/dflash_test_backend_ipc_payload"; + cfg.work_dir = file_path; + + BackendIpcProcess proc; + TEST_ASSERT(!proc.start(cfg)); + TEST_ASSERT(!proc.active()); + unlink(file_path.c_str()); +} + // ═══════════════════════════════════════════════════════════════════════ // Sampler tests (model-independent, CPU-only) // ═══════════════════════════════════════════════════════════════════════ @@ -1648,6 +1672,7 @@ int main() { RUN_TEST(test_disk_cache_budget_enforcement_scoring); RUN_TEST(test_disk_cache_lookup_miss_no_layout); RUN_TEST(test_disk_cache_save_below_min_tokens); + RUN_TEST(test_backend_ipc_rejects_file_work_dir); std::fprintf(stderr, "\n── Sampler ──\n"); RUN_TEST(test_sampler_cfg_defaults); From ff7592d9f92921be1bad6ca2e03aa079c029df07 Mon Sep 17 00:00:00 2001 From: weicj Date: Sun, 24 May 2026 01:07:59 +0800 Subject: [PATCH 3/3] fix(server): preserve backend IPC stdin fd --- dflash/src/common/backend_ipc.cpp | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/dflash/src/common/backend_ipc.cpp b/dflash/src/common/backend_ipc.cpp index f96820177..9f908ae6c 100644 --- a/dflash/src/common/backend_ipc.cpp +++ b/dflash/src/common/backend_ipc.cpp @@ -64,8 +64,11 @@ bool BackendIpcProcess::start(const BackendIpcLaunchConfig & cfg) { return false; } if (pid_ == 0) { - ::dup2(cmd_pipe[0], STDIN_FILENO); - ::close(cmd_pipe[0]); + if (cmd_pipe[0] != STDIN_FILENO && ::dup2(cmd_pipe[0], STDIN_FILENO) < 0) { + std::fprintf(stderr, "backend-ipc dup2 failed: %s\n", std::strerror(errno)); + _exit(127); + } + if (cmd_pipe[0] != STDIN_FILENO) ::close(cmd_pipe[0]); ::close(cmd_pipe[1]); ::close(stream_pipe[0]);