Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions python/sglang/srt/disaggregation/decode.py
Original file line number Diff line number Diff line change
Expand Up @@ -3775,6 +3775,8 @@ def _handle_kv_checksum_mismatch(self, req: Req, msg: str) -> None:
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
)
self.output_streamer.stream_output([req], req.return_logprob)
if self.enable_hisparse:
self.hisparse_coordinator.request_finished(req)
release_kv_cache(req, self.tree_cache, checkpoint=False)
if self.metrics_reporter.enable_metrics:
self.metrics_collector.increment_transfer_failed_reqs()
Expand Down
2 changes: 2 additions & 0 deletions python/sglang/srt/managers/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -4038,6 +4038,8 @@ def process_pending_chunked_abort(self) -> None:
self.release_aborted_request(req)
# No member rows exist before the leader's prefill ends.
self.beam_coordinator.retire_group(req)
if self.enable_hisparse:
self.hisparse_coordinator.request_finished(req)
release_kv_cache(req, self.tree_cache, checkpoint=False)

self.chunked_req = None
Expand Down
25 changes: 25 additions & 0 deletions test/registered/unit/disaggregation/test_kv_checksum.py
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,7 @@ def __init__(self, computer, req_to_token):
)
self.req_to_token_pool = SimpleNamespace(req_to_token=req_to_token)
self.tree_cache = None
self.enable_hisparse = False
self.output_streamer = SimpleNamespace(stream_output=self.stream_output)
self.metrics_reporter = SimpleNamespace(enable_metrics=True)
self.metrics_collector = Mock()
Expand Down Expand Up @@ -579,6 +580,30 @@ def test_backend_can_skip_incomparable_checksum(self):
mock_abort.assert_called_once()


class TestKvChecksumMismatchHiSparse(unittest.TestCase):
def test_mismatch_releases_hisparse_before_kv(self):
# Transfer already admitted the request into HiSparse; the abort must
# free its host pages before the KV itself.
sched = _FakeScheduler(None, None)
sched.enable_hisparse = True
calls = []
sched.hisparse_coordinator = Mock()
sched.hisparse_coordinator.request_finished.side_effect = lambda req: (
calls.append(("hisparse", req))
)
req = _make_req(0xDEADBEEF, 4)
with (
envs.SGLANG_IS_IN_CI.override(False),
patch("sglang.srt.disaggregation.decode.prepare_abort"),
patch(
"sglang.srt.disaggregation.decode.release_kv_cache",
side_effect=lambda req, *a, **kw: calls.append(("kv", req)),
),
):
sched._handle_kv_checksum_mismatch(req, "KV checksum mismatch")
self.assertEqual(calls, [("hisparse", req), ("kv", req)])


_SENTINEL = object()


Expand Down
23 changes: 23 additions & 0 deletions test/registered/unit/managers/test_scheduler_chunked_abort_race.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ def _make_scheduler(pending_req, *, chunked_req, running_reqs) -> Scheduler:
sched.last_batch = None
sched.beam_coordinator = Mock()
sched.output_streamer = Mock()
sched.enable_hisparse = False
return sched


Expand Down Expand Up @@ -105,6 +106,28 @@ def test_aborted_chunked_beam_leader_retires_its_group(self):

sched.beam_coordinator.retire_group.assert_called_once_with(req)

def test_aborted_chunked_req_releases_hisparse_before_kv(self):
# Each chunk's extend allocates HiSparse spec scratch for the row.
req = _FakeReq("hisparse_chunked")
sched = _make_scheduler(req, chunked_req=req, running_reqs=[])
sched.release_aborted_request = Mock()
sched.tree_cache = None
sched.ipc_channels = Mock()
sched.enable_hisparse = True
calls = []
sched.hisparse_coordinator = Mock()
sched.hisparse_coordinator.request_finished.side_effect = lambda req: (
calls.append(("hisparse", req))
)

with patch(
"sglang.srt.managers.scheduler.release_kv_cache",
side_effect=lambda req, *a, **kw: calls.append(("kv", req)),
):
sched.process_pending_chunked_abort()

self.assertEqual(calls, [("hisparse", req), ("kv", req)])


if __name__ == "__main__":
unittest.main(verbosity=2)
Loading