A high-throughput and memory-efficient inference and serving engine for LLMs
Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git
Synced Nov 1, 23:45 UTC
[Refactor] Remove dead code in tests and parallel_state (#41471)
Signed-off-by: yewentao256 <zhyanwentao@126.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Wentao Ye committed
Jun 3, 2026 at 22:32 UTC
e6018c644adcdebfb1c59aadd3fbd6cdde1c2608
4 files changed
+7
-63
tests/v1/kv_connector/unit/test_bidirectional_kv_transfer.py
-1
@@ -98,7 +98,6 @@ def _make_connector_with_fake_worker(
98
)
99
worker = connector.connector_worker
100
assert isinstance(worker.nixl_wrapper, FakeNixlWrapper)
101
- worker.nixl_wrapper.set_cycles_before_xfer_done(cycles_before_done)
101
worker.kv_cache_layout = "HND"
102
if do_handshake:
103
remote_agents = worker._nixl_handshake(
tests/v1/kv_connector/unit/test_nixl_connector.py
+1
-15
@@ -197,13 +197,6 @@ class FakeNixlWrapper:
197
def get_xfer_telemetry(self, handle: int) -> dict:
198
return get_default_xfer_telemetry()
199
200
- ############################################################
201
- # Follow are for changing the behavior during testing.
202
- ############################################################
203
-
204
- def set_cycles_before_xfer_done(self, cycles: int):
205
- """Set the number of cycles before a transfer is considered done."""
206
-
200
201
@contextlib.contextmanager
202
def _make_fake_nixl_pkg():
@@ -578,10 +571,7 @@ class TestNixlHandshake:
571
"""Test case where multiple xfers are initiated to the same engine.
572
573
This test triggers the connector to load remote KV for the same
581
- `request_id`. The transfer is not done immediately due to
582
- `set_cycles_before_xfer_done`, so there is a state where there are
583
- multiple transfer states for the same `request_id`, and `get_finished`
584
- should handle it correctly (wait for all transfers to be done).
574
+ `request_id`.
575
"""
576
vllm_config = create_vllm_config()
577
@@ -598,7 +588,6 @@ class TestNixlHandshake:
588
)
589
assert isinstance(connector.connector_worker.nixl_wrapper, FakeNixlWrapper)
590
worker = connector.connector_worker
601
- worker.nixl_wrapper.set_cycles_before_xfer_done(3)
591
# simulate handshake
592
worker.dst_xfer_side_handles = {
593
FakeNixlConnectorWorker.REMOTE_ENGINE_ID: {0: 1}
@@ -1304,7 +1293,6 @@ def test_scheduler_kv_connector_stats_aggregation():
1293
# Worker stats with transfer metrics
1294
worker_stats = NixlKVConnectorStats()
1295
worker_stats.record_transfer(get_default_xfer_telemetry())
1307
- worker_stats.data["remote_tokens"] = []
1296
1297
# Scheduler stats with custom metric (needs dummy transfer to avoid being skipped)
1298
scheduler_stats = NixlKVConnectorStats()
@@ -1314,7 +1302,6 @@ def test_scheduler_kv_connector_stats_aggregation():
1302
"post_duration": [0],
1303
"bytes_transferred": [0],
1304
"num_descriptors": [0],
1317
- "remote_tokens": [128],
1305
}
1306
)
1307
@@ -1355,7 +1342,6 @@ def test_scheduler_kv_connector_stats_aggregation():
1342
).scheduler_stats.kv_connector_stats
1343
nixl_stats = final_stats["NixlConnector"]
1344
assert nixl_stats.num_successful_transfers == 2
1358
- assert nixl_stats.data["remote_tokens"] == [128]
1345
1346
1347
@pytest.mark.parametrize("distributed_executor_backend", ["ray", None])
vllm/distributed/kv_transfer/kv_connector/utils.py
+6
-9
@@ -120,15 +120,12 @@ class KVOutputAggregator:
120
# Use the first worker's kv_connector_stats as accumulator.
121
aggregated_kv_connector_stats = kv_output.kv_connector_stats
122
elif kv_connector_stats := kv_output.kv_connector_stats:
123
- if aggregated_kv_connector_stats is None:
124
- aggregated_kv_connector_stats = kv_connector_stats
125
- else:
126
- assert isinstance(
127
- aggregated_kv_connector_stats, type(kv_connector_stats)
128
- )
129
- aggregated_kv_connector_stats = (
130
- aggregated_kv_connector_stats.aggregate(kv_connector_stats)
131
- )
123
+ assert isinstance(
124
+ aggregated_kv_connector_stats, type(kv_connector_stats)
125
+ )
126
+ aggregated_kv_connector_stats = aggregated_kv_connector_stats.aggregate(
127
+ kv_connector_stats
128
+ )
129
130
# Aggregate kv_connector_worker_meta from all workers.
131
if aggregated_kv_connector_worker_meta is None:
vllm/distributed/parallel_state.py
-38
@@ -1270,9 +1270,6 @@ def get_dcp_group() -> GroupCoordinator:
1270
return _DCP
1271
1272
1273
-# kept for backward compatibility
1274
-get_context_model_parallel_group = get_dcp_group
1275
-
1273
_PP: GroupCoordinator | None = None
1274
1275
@@ -1840,31 +1837,6 @@ def model_parallel_is_initialized():
1837
_TP_STATE_PATCHED = False
1838
1839
1843
-@contextmanager
1844
-def patch_tensor_parallel_group(tp_group: GroupCoordinator):
1845
- """Patch the tp group temporarily until this function ends.
1846
-
1847
- This method is for draft workers of speculative decoding to run draft model
1848
- with different tp degree from that of target model workers.
1849
-
1850
- Args:
1851
- tp_group (GroupCoordinator): the tp group coordinator
1852
- """
1853
- global _TP_STATE_PATCHED
1854
- assert not _TP_STATE_PATCHED, "Should not call when it's already patched"
1855
-
1856
- _TP_STATE_PATCHED = True
1857
- old_tp_group = get_tp_group()
1858
- global _TP
1859
- _TP = tp_group
1860
- try:
1861
- yield
1862
- finally:
1863
- # restore the original state
1864
- _TP_STATE_PATCHED = False
1865
- _TP = old_tp_group
1866
-
1867
-
1840
def get_tensor_model_parallel_world_size() -> int:
1841
"""Return world size for the tensor model parallel group."""
1842
return get_tp_group().world_size
@@ -1875,16 +1847,6 @@ def get_tensor_model_parallel_rank() -> int:
1847
return get_tp_group().rank_in_group
1848
1849
1878
-def get_decode_context_model_parallel_world_size() -> int:
1879
- """Return world size for the decode context model parallel group."""
1880
- return get_dcp_group().world_size
1881
-
1882
-
1883
-def get_decode_context_model_parallel_rank() -> int:
1884
- """Return my rank for the decode context model parallel group."""
1885
- return get_dcp_group().rank_in_group
1886
-
1887
-
1850
def get_node_count() -> int:
1851
"""Return the total number of nodes in the distributed environment."""
1852
assert _NODE_COUNT is not None, "distributed environment is not initialized"