feat(v1/sched): SchedulerOutput + NewRequestData/CachedRequestData

mudler · claude · mudler · commit c65e650e45bc · 2026-07-03T20:59:15.000Z
Ported from vllm/v1/core/sched/output.py @ e24d1b24 (M1.4 Task 2).
Value types the scheduler produces + the model runner (M1.5) consumes:
NewRequestData (full new-req payload + from_request), CachedRequestData
(diff-only running-req payload + make_empty/is_context_phase/num_reqs),
SchedulerOutput (T0 field set + make_empty). Behavioral only, all CI.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -68,6 +68,7 @@ add_library(vllm STATIC
   src/vllm/v1/core/kv_cache_coordinator.cpp
   src/vllm/v1/core/kv_cache_manager.cpp
   src/vllm/v1/core/sched/request_queue.cpp
+  src/vllm/v1/core/sched/output.cpp
   src/vllm/v1/engine/types.cpp
   src/vllm/v1/engine/detokenizer.cpp
   src/vt/dtype.cpp
diff --git a/include/vllm/v1/core/sched/output.h b/include/vllm/v1/core/sched/output.h
@@ -0,0 +1,152 @@
+// Ported from: vllm/v1/core/sched/output.py @ e24d1b24
+//
+// Scope (M1.4 Task 2): the value types the V1 Scheduler PRODUCES each step and
+// the model runner (M1.5) CONSUMES — the new-request full payload, the
+// running-request diff payload, and the SchedulerOutput envelope that carries
+// them across the (upstream: process) boundary. Behavioral only: plain value
+// carriers, no CUDA / model. Field names + the new-vs-cached diff semantics are
+// mirrored 1:1 with upstream.
+//
+// THE new-vs-cached DIFF PROTOCOL (mirrored 1:1):
+//   * scheduled_new_reqs: requests scheduled for the FIRST time carry their
+//     FULL data (NewRequestData) — prompt, sampling params, the complete
+//     per-group block_ids, num_computed_tokens. The worker caches this so it is
+//     never re-sent.
+//   * scheduled_cached_reqs: requests scheduled BEFORE carry only the DIFF
+//     (CachedRequestData) — parallel arrays over req_ids: the NEWLY allocated
+//     block_ids per request per group (appended to the cached block table,
+//     unless the req_id is in resumed_req_ids in which case they REPLACE it),
+//     the updated num_computed_tokens, num_output_tokens, and (PP-only)
+//     new_token_ids. The heavy per-request state already lives in the worker.
+//
+// DEFERRED upstream state (marked; T0 never populates these):
+//   NewRequestData: mm_features (multimodal), pooling_params, lora_request,
+//     prompt_embeds, prompt_is_token_ids, prefill_token_ids (v2 model runner).
+//   SchedulerOutput trailing optionals, OMITTED here (a later unit slots them
+//     back in without reshaping the struct): preempted_req_ids /
+//     new_block_ids_to_zero (v2 model runner), has_structured_output_requests /
+//     pending_structured_output_tokens (grammar), num_invalid_spec_tokens /
+//     num_spec_tokens_to_schedule (spec decode), kv_connector_metadata /
+//     ec_connector_metadata (KV/EC transfer). GrammarOutput (structured output)
+//     is likewise omitted.
+//
+// DEVIATIONS, recorded:
+//   - Upstream NewRequestData.prompt_token_ids is `list[int] | None` and
+//     .sampling_params is `SamplingParams | None`; represented here as
+//     std::optional<...>. from_request always populates both in the T0
+//     pure-token / generation path (pooling is deferred), but the optional
+//     preserves the upstream nullability.
+//   - block_ids is `tuple[list[int], ...]` upstream (one list per KV cache
+//     group); here std::vector<std::vector<int>>, exactly what
+//     KVCacheBlocks::get_block_ids() / KVCacheManager::get_block_ids() return.
+//   - CachedRequestData.new_block_ids entries are
+//     `tuple[list[int], ...] | None` upstream (None => no new blocks this step,
+//     from get_block_ids(allow_none=True)); here
+//     std::optional<std::vector<std::vector<int>>>.
+//   - Upstream caches _req_id_to_num_output_tokens (a cached_property) for O(1)
+//     is_context_phase lookups; here is_context_phase does a linear scan over
+//     req_ids (T0 batches are small; the value types are rebuilt fresh each
+//     step so no cache-invalidation concern). Semantics are identical.
+//   - all_token_ids is the MRV1-only connector-propagation map (dict[str,
+//     list[int]] upstream); carried for shape fidelity, left empty in T0.
+//   - The scheduler assembles CachedRequestData field-by-field via
+//     scheduler.py::_make_cached_request_data (lands in Task 3); output.py
+//     itself exposes ONLY make_empty (no from_request/append constructor). The
+//     public aggregate here matches that — the diff-building loop is Task 3.
+#pragma once
+
+#include <map>
+#include <optional>
+#include <set>
+#include <string>
+#include <vector>
+
+#include "vllm/sampling_params.h"
+
+namespace vllm::v1 {
+
+struct Request;  // vllm/v1/request.h (from_request source; cpp includes it).
+
+// NewRequestData: the FULL payload for a request scheduled for the first time.
+// (Upstream NewRequestData dataclass.)
+struct NewRequestData {
+  std::string req_id;
+  // Upstream list[int] | None — always populated in the T0 token path.
+  std::optional<std::vector<int32_t>> prompt_token_ids;
+  // Upstream SamplingParams | None — always populated for generation requests.
+  std::optional<SamplingParams> sampling_params;
+  // Per-KV-cache-group block ids (upstream tuple[list[int], ...]).
+  std::vector<std::vector<int>> block_ids;
+  int num_computed_tokens = 0;
+
+  // from_request: build the full payload from the Request + its allocated
+  // per-group block ids. Copies req_id, prompt_token_ids, sampling_params,
+  // block_ids, num_computed_tokens. (Upstream NewRequestData.from_request;
+  // mm/pooling/lora/prompt_embeds/prefill_token_ids DEFERRED.)
+  static NewRequestData from_request(const Request& request,
+                                     std::vector<std::vector<int>> block_ids);
+};
+
+// CachedRequestData: the DIFF-only payload for requests scheduled before.
+// Parallel arrays indexed by position over req_ids. (Upstream CachedRequestData
+// dataclass.)
+struct CachedRequestData {
+  std::vector<std::string> req_ids;
+  // req_ids in this set have their block table REPLACED by new_block_ids
+  // (resumed from preemption); those not in it have new_block_ids APPENDED.
+  std::set<std::string> resumed_req_ids;
+  // PP-only: the sampled token ids to forward. Empty when PP is not used.
+  std::vector<std::vector<int32_t>> new_token_ids;
+  // MRV1-only connector propagation (req_id -> full token ids). Empty in T0.
+  std::map<std::string, std::vector<int32_t>> all_token_ids;
+  // Per request: the NEWLY allocated per-group block ids, or nullopt when no
+  // new blocks were allocated this step (upstream get_block_ids(allow_none)).
+  std::vector<std::optional<std::vector<std::vector<int>>>> new_block_ids;
+  std::vector<int> num_computed_tokens;
+  std::vector<int> num_output_tokens;
+
+  // num_reqs (property): len(req_ids).
+  int num_reqs() const { return static_cast<int>(req_ids.size()); }
+
+  // is_context_phase: true iff the request is present and still in its prefill
+  // (context) phase, i.e. num_output_tokens == 0 for that req_id. (Upstream
+  // is_context_phase via the _req_id_to_num_output_tokens cache; linear scan
+  // here — see the header DEVIATIONS note.)
+  bool is_context_phase(const std::string& req_id) const;
+
+  // make_empty: the no-cached-requests diff.
+  static CachedRequestData make_empty();
+};
+
+// SchedulerOutput: the per-step envelope the model runner consumes. (Upstream
+// SchedulerOutput dataclass — T0 field subset; trailing optionals OMITTED, see
+// the header DEFERRED note.)
+struct SchedulerOutput {
+  // Requests scheduled for the first time (full data, worker-cached).
+  std::vector<NewRequestData> scheduled_new_reqs;
+  // Requests scheduled before (diff only).
+  CachedRequestData scheduled_cached_reqs;
+
+  // req_id -> number of tokens scheduled for it this step.
+  std::map<std::string, int> num_scheduled_tokens;
+  // sum(num_scheduled_tokens.values()).
+  int total_num_scheduled_tokens = 0;
+
+  // req_id -> spec decode token ids. DEFERRED semantics: always empty in T0.
+  std::map<std::string, std::vector<int32_t>> scheduled_spec_decode_tokens;
+  // req_id -> encoder input indices to process. DEFERRED semantics: empty in T0.
+  std::map<std::string, std::vector<int>> scheduled_encoder_inputs;
+  // Number of common prefix blocks per KV cache group (for cascade attention).
+  std::vector<int> num_common_prefix_blocks;
+
+  // Request ids finished between the previous and current step (free cached
+  // worker state for these).
+  std::set<std::string> finished_req_ids;
+  // mm_hash strings whose encoder outputs can be freed. DEFERRED: empty in T0.
+  std::vector<std::string> free_encoder_mm_hashes;
+
+  // make_empty: an empty step output.
+  static SchedulerOutput make_empty();
+};
+
+}  // namespace vllm::v1
diff --git a/src/vllm/v1/core/sched/output.cpp b/src/vllm/v1/core/sched/output.cpp
@@ -0,0 +1,42 @@
+// Ported from: vllm/v1/core/sched/output.py @ e24d1b24
+// See include/vllm/v1/core/sched/output.h for the new-vs-cached diff protocol,
+// the deferred-field list, and the recorded deviations.
+#include "vllm/v1/core/sched/output.h"
+
+#include <utility>
+
+#include "vllm/v1/request.h"
+
+namespace vllm::v1 {
+
+NewRequestData NewRequestData::from_request(
+    const Request& request, std::vector<std::vector<int>> block_ids) {
+  NewRequestData data;
+  data.req_id = request.request_id;
+  data.prompt_token_ids = request.prompt_token_ids;
+  data.sampling_params = request.sampling_params;
+  data.block_ids = std::move(block_ids);
+  data.num_computed_tokens = request.num_computed_tokens;
+  return data;
+}
+
+bool CachedRequestData::is_context_phase(const std::string& req_id) const {
+  for (std::size_t i = 0; i < req_ids.size(); ++i) {
+    if (req_ids[i] == req_id) {
+      return i < num_output_tokens.size() && num_output_tokens[i] == 0;
+    }
+  }
+  return false;
+}
+
+CachedRequestData CachedRequestData::make_empty() {
+  return CachedRequestData{};
+}
+
+SchedulerOutput SchedulerOutput::make_empty() {
+  SchedulerOutput output;
+  output.scheduled_cached_reqs = CachedRequestData::make_empty();
+  return output;
+}
+
+}  // namespace vllm::v1
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
@@ -28,6 +28,7 @@ vllm_cpp_add_test(test_kv_cache_coordinator vllm/v1/test_kv_cache_coordinator.cp
 vllm_cpp_add_test(test_kv_cache_manager vllm/v1/test_kv_cache_manager.cpp)
 vllm_cpp_add_test(test_scheduler_config vllm/test_scheduler_config.cpp)
 vllm_cpp_add_test(test_request_queue vllm/v1/test_request_queue.cpp)
+vllm_cpp_add_test(test_sched_output vllm/v1/test_sched_output.cpp)
 vllm_cpp_add_test(test_engine_types vllm/test_engine_types.cpp)
 vllm_cpp_add_test(test_outputs vllm/test_outputs.cpp)
 vllm_cpp_add_test(test_cuda_backend vt/test_cuda_backend.cpp)
diff --git a/tests/vllm/v1/test_sched_output.cpp b/tests/vllm/v1/test_sched_output.cpp
@@ -0,0 +1,182 @@
+// Tests for the SchedulerOutput / NewRequestData / CachedRequestData port
+// (vllm/v1/core/sched/output.py @ e24d1b24).
+//
+// Upstream output.py has no dedicated unit test; these types are exercised via
+// tests/v1/core/test_scheduler.py (the new-vs-cached diff shape the model runner
+// consumes). Ported here as direct construct-and-check value-carrier tests: the
+// diff protocol (full NewRequestData vs diff-only CachedRequestData), the
+// from_request field copy, make_empty, and the SchedulerOutput envelope.
+#include <doctest/doctest.h>
+
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "vllm/sampling_params.h"
+#include "vllm/v1/core/sched/output.h"
+#include "vllm/v1/request.h"
+
+using vllm::SamplingParams;
+using vllm::v1::CachedRequestData;
+using vllm::v1::NewRequestData;
+using vllm::v1::Request;
+using vllm::v1::SchedulerOutput;
+
+namespace {
+
+Request MakeRequest(const std::string& id,
+                    std::vector<int32_t> prompt = {1, 2, 3, 4}) {
+  SamplingParams params;
+  params.max_tokens = 16;
+  Request req(id, std::move(prompt), params, /*arrival_time=*/0.0);
+  req.num_computed_tokens = 2;
+  return req;
+}
+
+}  // namespace
+
+TEST_CASE("NewRequestData::from_request copies the T0 fields + block_ids") {
+  Request req = MakeRequest("req-0", {10, 11, 12});
+  // Per-group block ids (one group here).
+  std::vector<std::vector<int>> block_ids = {{7, 8, 9}};
+
+  NewRequestData data = NewRequestData::from_request(req, block_ids);
+
+  CHECK(data.req_id == "req-0");
+  REQUIRE(data.prompt_token_ids.has_value());
+  CHECK(data.prompt_token_ids.value() == std::vector<int32_t>{10, 11, 12});
+  REQUIRE(data.sampling_params.has_value());
+  CHECK(data.sampling_params->max_tokens == 16);
+  CHECK(data.num_computed_tokens == 2);
+  REQUIRE(data.block_ids.size() == 1);
+  CHECK(data.block_ids[0] == std::vector<int>{7, 8, 9});
+}
+
+TEST_CASE("NewRequestData carries multi-group block_ids by group") {
+  Request req = MakeRequest("req-mg");
+  std::vector<std::vector<int>> block_ids = {{1, 2}, {3, 4, 5}};
+
+  NewRequestData data = NewRequestData::from_request(req, block_ids);
+
+  REQUIRE(data.block_ids.size() == 2);
+  CHECK(data.block_ids[0] == std::vector<int>{1, 2});
+  CHECK(data.block_ids[1] == std::vector<int>{3, 4, 5});
+}
+
+TEST_CASE("CachedRequestData::make_empty is the empty diff") {
+  CachedRequestData cached = CachedRequestData::make_empty();
+
+  CHECK(cached.num_reqs() == 0);
+  CHECK(cached.req_ids.empty());
+  CHECK(cached.resumed_req_ids.empty());
+  CHECK(cached.new_token_ids.empty());
+  CHECK(cached.all_token_ids.empty());
+  CHECK(cached.new_block_ids.empty());
+  CHECK(cached.num_computed_tokens.empty());
+  CHECK(cached.num_output_tokens.empty());
+}
+
+TEST_CASE("CachedRequestData diff shape: parallel arrays over req_ids") {
+  CachedRequestData cached;
+  cached.req_ids = {"a", "b"};
+  // "b" is resumed from preemption -> its block table is REPLACED.
+  cached.resumed_req_ids = {"b"};
+  // Per request: newly allocated per-group block ids; nullopt = none this step.
+  cached.new_block_ids = {
+      std::optional<std::vector<std::vector<int>>>{{{20, 21}}},  // a: append
+      std::optional<std::vector<std::vector<int>>>{{{30}}},      // b: replace
+  };
+  cached.num_computed_tokens = {5, 8};
+  cached.num_output_tokens = {1, 0};
+
+  CHECK(cached.num_reqs() == 2);
+  // "a" is appended (not resumed); "b" is resumed (replace).
+  CHECK(cached.resumed_req_ids.count("a") == 0);
+  CHECK(cached.resumed_req_ids.count("b") == 1);
+
+  REQUIRE(cached.new_block_ids.size() == 2);
+  REQUIRE(cached.new_block_ids[0].has_value());
+  CHECK(cached.new_block_ids[0].value()[0] == std::vector<int>{20, 21});
+  REQUIRE(cached.new_block_ids[1].has_value());
+  CHECK(cached.new_block_ids[1].value()[0] == std::vector<int>{30});
+
+  CHECK(cached.num_computed_tokens == std::vector<int>{5, 8});
+  CHECK(cached.num_output_tokens == std::vector<int>{1, 0});
+}
+
+TEST_CASE("CachedRequestData new_block_ids nullopt = no new blocks this step") {
+  CachedRequestData cached;
+  cached.req_ids = {"only"};
+  cached.new_block_ids = {std::nullopt};
+  cached.num_computed_tokens = {3};
+  cached.num_output_tokens = {2};
+
+  REQUIRE(cached.new_block_ids.size() == 1);
+  CHECK_FALSE(cached.new_block_ids[0].has_value());
+}
+
+TEST_CASE("CachedRequestData::is_context_phase reflects num_output_tokens") {
+  CachedRequestData cached;
+  cached.req_ids = {"prefill", "decode"};
+  cached.num_output_tokens = {0, 4};
+
+  // prefill: still 0 output tokens -> context (prefill) phase.
+  CHECK(cached.is_context_phase("prefill"));
+  // decode: has output tokens -> not context phase.
+  CHECK_FALSE(cached.is_context_phase("decode"));
+  // unknown req_id -> false.
+  CHECK_FALSE(cached.is_context_phase("missing"));
+}
+
+TEST_CASE("SchedulerOutput::make_empty is an empty step") {
+  SchedulerOutput out = SchedulerOutput::make_empty();
+
+  CHECK(out.scheduled_new_reqs.empty());
+  CHECK(out.scheduled_cached_reqs.num_reqs() == 0);
+  CHECK(out.num_scheduled_tokens.empty());
+  CHECK(out.total_num_scheduled_tokens == 0);
+  CHECK(out.scheduled_spec_decode_tokens.empty());
+  CHECK(out.scheduled_encoder_inputs.empty());
+  CHECK(out.num_common_prefix_blocks.empty());
+  CHECK(out.finished_req_ids.empty());
+  CHECK(out.free_encoder_mm_hashes.empty());
+}
+
+TEST_CASE("SchedulerOutput carries new + cached reqs, token map, finished ids") {
+  Request new_req = MakeRequest("new-1", {100, 101});
+
+  SchedulerOutput out;
+  out.scheduled_new_reqs.push_back(
+      NewRequestData::from_request(new_req, {{42}}));
+
+  CachedRequestData cached;
+  cached.req_ids = {"cached-1"};
+  cached.new_block_ids = {std::optional<std::vector<std::vector<int>>>{{{43}}}};
+  cached.num_computed_tokens = {6};
+  cached.num_output_tokens = {3};
+  out.scheduled_cached_reqs = cached;
+
+  out.num_scheduled_tokens = {{"new-1", 2}, {"cached-1", 1}};
+  out.total_num_scheduled_tokens = 3;
+  out.num_common_prefix_blocks = {0};
+  out.finished_req_ids = {"done-1"};
+
+  // New reqs carry FULL data.
+  REQUIRE(out.scheduled_new_reqs.size() == 1);
+  CHECK(out.scheduled_new_reqs[0].req_id == "new-1");
+  CHECK(out.scheduled_new_reqs[0].block_ids[0] == std::vector<int>{42});
+
+  // Cached reqs carry only the DIFF.
+  CHECK(out.scheduled_cached_reqs.num_reqs() == 1);
+  CHECK(out.scheduled_cached_reqs.req_ids[0] == "cached-1");
+
+  // Token accounting: total == sum of the per-request map.
+  int sum = 0;
+  for (const auto& [id, n] : out.num_scheduled_tokens) sum += n;
+  CHECK(sum == out.total_num_scheduled_tokens);
+  CHECK(out.num_scheduled_tokens.at("new-1") == 2);
+  CHECK(out.num_scheduled_tokens.at("cached-1") == 1);
+
+  CHECK(out.finished_req_ids.count("done-1") == 1);
+}
