From 549b26a3d6b4af8904f90aec2f51ec4982e7bb6c Mon Sep 17 00:00:00 2001 From: omerpaz95 Date: Thu, 23 Jul 2026 14:09:49 +0300 Subject: [PATCH 1/3] Added call from scheduler to EC Connector update_connector_output. Signed-off-by: omerpaz95 --- tests/v1/core/test_scheduler.py | 42 ++++++++++++++++++++++++++++++++- vllm/v1/core/sched/scheduler.py | 5 ++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/tests/v1/core/test_scheduler.py b/tests/v1/core/test_scheduler.py index b782e34b011d..42350ecf7a67 100644 --- a/tests/v1/core/test_scheduler.py +++ b/tests/v1/core/test_scheduler.py @@ -37,7 +37,12 @@ KVCacheGroupSpec, MambaSpec, ) -from vllm.v1.outputs import DraftTokenIds, KVConnectorOutput, ModelRunnerOutput +from vllm.v1.outputs import ( + DraftTokenIds, + ECConnectorOutput, + KVConnectorOutput, + ModelRunnerOutput, +) from vllm.v1.request import Request, RequestStatus from vllm.v1.structured_output import StructuredOutputManager @@ -4697,6 +4702,41 @@ def update_connector_output( assert final_stats == expected_data +def test_ec_connector_update_connector_output_called(): + """Test that worker-side EC connector output is forwarded to the + EC connector's update_connector_output hook.""" + scheduler = create_scheduler( + model="llava-hf/llava-1.5-7b-hf", + use_ec_connector=True, + ec_role="ec_consumer", + ) + scheduler.ec_connector.update_connector_output = Mock() + + scheduler_output = SchedulerOutput( + scheduled_new_reqs=[], + scheduled_cached_reqs=CachedRequestData.make_empty(), + num_scheduled_tokens={}, + total_num_scheduled_tokens=0, + scheduled_encoder_inputs={}, + scheduled_spec_decode_tokens={}, + num_common_prefix_blocks=[], + finished_req_ids=set(), + free_encoder_mm_hashes=[], + ) + ec_connector_output = ECConnectorOutput(finished_sending={"hash_test1"}) + model_runner_output = ModelRunnerOutput( + req_ids=[], + req_id_to_index={}, + ec_connector_output=ec_connector_output, + ) + + scheduler.update_from_output(scheduler_output, model_runner_output) + + scheduler.ec_connector.update_connector_output.assert_called_once_with( + ec_connector_output + ) + + # ============================================================================== # Variable-length encoder cross-attention block allocation tests # ============================================================================== diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index fcd60804a31a..5a0456c150e5 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -1581,6 +1581,7 @@ def update_from_output( pooler_outputs = model_runner_output.pooler_output num_nans_in_logits = model_runner_output.num_nans_in_logits kv_connector_output = model_runner_output.kv_connector_output + ec_connector_output = model_runner_output.ec_connector_output cudagraph_stats = model_runner_output.cudagraph_stats # Every GPU write enqueued by this and earlier steps has completed, so it is @@ -1864,6 +1865,10 @@ def update_from_output( if kv_connector_output: self._update_from_kv_xfer_finished(kv_connector_output) + # EC Connector: update state from worker-side EC connector output. + if ec_connector_output and self.ec_connector is not None: + self.ec_connector.update_connector_output(ec_connector_output) + # Worker-side KV connector stats from the model runner output. kv_connector_stats: KVConnectorStats | None = ( kv_connector_output.kv_connector_stats if kv_connector_output else None From 51f343014258c26caddfe3daa7cafd416c669075 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Lucchesi?= Date: Mon, 27 Jul 2026 14:44:16 +0200 Subject: [PATCH 2/3] Update vllm/v1/core/sched/scheduler.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Nicolò Lucchesi --- vllm/v1/core/sched/scheduler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index 5a0456c150e5..68de38a14a8a 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -1866,7 +1866,7 @@ def update_from_output( self._update_from_kv_xfer_finished(kv_connector_output) # EC Connector: update state from worker-side EC connector output. - if ec_connector_output and self.ec_connector is not None: + if self.ec_connector is not None and ec_connector_output: self.ec_connector.update_connector_output(ec_connector_output) # Worker-side KV connector stats from the model runner output. From f26f409ebd3cf32acd16f89a378001a75baf4541 Mon Sep 17 00:00:00 2001 From: omerpaz95 Date: Tue, 28 Jul 2026 11:34:49 +0300 Subject: [PATCH 3/3] Fixed CI. Signed-off-by: omerpaz95 --- tests/v1/core/test_async_scheduler.py | 1 + tests/v1/core/test_scheduler.py | 1 + 2 files changed, 2 insertions(+) diff --git a/tests/v1/core/test_async_scheduler.py b/tests/v1/core/test_async_scheduler.py index e34a0da54d8d..960d9139deed 100644 --- a/tests/v1/core/test_async_scheduler.py +++ b/tests/v1/core/test_async_scheduler.py @@ -271,6 +271,7 @@ def test_abort_request_when_structured_output_fsm_cannot_advance(): scheduler.perf_metrics = None scheduler.connector = None + scheduler.ec_connector = None scheduler.structured_output_manager = Mock() scheduler.structured_output_manager.should_advance.return_value = True scheduler.structured_output_manager.trim_reasoning_for_advance.side_effect = ( diff --git a/tests/v1/core/test_scheduler.py b/tests/v1/core/test_scheduler.py index fe054d4e1523..b857f96503ce 100644 --- a/tests/v1/core/test_scheduler.py +++ b/tests/v1/core/test_scheduler.py @@ -3222,6 +3222,7 @@ def test_abort_request_when_structured_output_fsm_cannot_advance(): scheduler.perf_metrics = None scheduler.connector = None + scheduler.ec_connector = None scheduler.structured_output_manager = Mock() scheduler.structured_output_manager.should_advance.return_value = True scheduler.structured_output_manager.trim_reasoning_for_advance.side_effect = (