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
6 changes: 3 additions & 3 deletions docs/async-speculative.md
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ output reservations separately from committed tokens.

A simpler alternative is to copy each result asynchronously but wait before
submitting the next round for that request. That can overlap independent requests,
but retains a CPU dependency between consecutive rounds. The two-bank design
but retains a CPU dependency between consecutive rounds. The two-buffer design
removes that dependency at the cost of device state, reservation slack and up to
one discarded round after EOS. The existing synchronous speculative path remains
available by disabling `async_scheduling`; ordinary decoding remains the default
Expand Down Expand Up @@ -180,7 +180,7 @@ reduction is claimed; reserved and allocated peaks are retained per observation.
### Active-region preparation and transfer

A separate warmed microexperiment used 1000 CPU preparations and 100 CUDA-event
H2D observations per variant, with prompt/output 64/32. Both banks are reused,
H2D observations per variant, with prompt/output 64/32. Both buffers are reused,
and copies complete before the CPU mutates pinned storage. Times below are medians;
they are not complete-round or E2E timings.

Expand Down Expand Up @@ -310,7 +310,7 @@ These ranges describe the observed runs, not confidence intervals.

The largest observed allocated-memory peaks were 18.342 GiB for sync speculation
and 18.657 GiB for async speculation (longer case, concurrency 32, K=2). This
implementation retains device state and pinned/device banks; it does not claim a
implementation retains device state and pinned/device buffers; it does not claim a
memory reduction. JSON also contains reserved-memory peaks for every observation.

## Historical profiling analysis
Expand Down
24 changes: 12 additions & 12 deletions tests/test_async_speculative.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ def test_async_matches_sync_with_ragged_limits_and_prefix_reuse(k, budget, dtype
assert all(o.finished for o in output.values())
if asynchronous:
assert not e.speculative_runner.states
assert all(not bank.leased for bank in e.speculative_runner.banks)
assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
assert all(r == results[0] for r in results)


Expand Down Expand Up @@ -233,7 +233,7 @@ def test_prefill_arrivals_interleave_with_inflight_rounds():
}
assert all(o.finished for o in actual.values())
assert not e.speculative_runner.states
assert all(not bank.leased for bank in e.speculative_runner.banks)
assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
e.close()
reference.close()

Expand All @@ -250,7 +250,7 @@ def test_close_drains_both_submitted_rounds(multi_stream):
assert not e.has_unfinished_requests()
assert e.cache_manager.num_used_blocks == 0
assert not e.speculative_runner.states
assert all(not bank.leased for bank in e.speculative_runner.banks)
assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
e.close()


Expand Down Expand Up @@ -295,7 +295,7 @@ def test_two_round_kv_matches_serial_replay_at_every_depth(dtype):
assert not e.has_unfinished_requests()
assert e.cache_manager.num_used_blocks == 0
assert not e.speculative_runner.states
assert all(not bank.leased for bank in e.speculative_runner.banks)
assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
e.close()


Expand All @@ -310,12 +310,12 @@ def test_active_page_tables_are_contiguous_and_do_not_touch_inactive_storage():
assert cache.allocate("short", 4) and cache.allocate("long", 10)
runner = object.__new__(AsyncSpeculativeRunner)
runner.cache = cache
bank = SimpleNamespace(
buffer = SimpleNamespace(
leased=False,
host_tables=torch.full((128,), -91, dtype=torch.int32),
tables=torch.full((128,), -92, dtype=torch.int32),
)
runner.banks = [bank]
runner.buffers = [buffer]
requests = [Request(rid, [2], SamplingParams()) for rid in ("short", "long")]
batch = SchedulerOutput(
Stage.SPECULATIVE,
Expand All @@ -327,25 +327,25 @@ def test_active_page_tables_are_contiguous_and_do_not_touch_inactive_storage():
_, _, host, device = runner._prepare_round(batch)
assert host.shape == device.shape == (2, 4, 5)
assert host.is_contiguous() and device.is_contiguous()
assert host.data_ptr() == bank.host_tables.data_ptr()
assert host.data_ptr() == buffer.host_tables.data_ptr()
for i, rid in enumerate(("short", "long")):
pages = 2 if i == 0 else 5
for depth in range(4):
expected = cache._get_allocation(rid).block_tables[depth][:pages]
assert host[i, depth, :pages].tolist() == list(expected)
assert not host[0, :, 2:].any()
assert (bank.host_tables[40:] == -91).all()
assert (bank.tables == -92).all() # CPU preparation does not upload.
assert (buffer.host_tables[40:] == -91).all()
assert (buffer.tables == -92).all() # CPU preparation does not upload.
with pytest.raises(RuntimeError, match="exhausted"):
runner._prepare_round(batch)
# Simulate retirement, then shrink both dimensions using the same storage.
bank.leased = False
before = bank.host_tables.clone()
buffer.leased = False
before = buffer.host_tables.clone()
smaller = SchedulerOutput(Stage.SPECULATIVE, [ScheduledItem(requests[0], 0, 1)])
_, _, host, device = runner._prepare_round(smaller)
assert host.shape == device.shape == (1, 4, 1)
assert host.is_contiguous() and device.is_contiguous()
assert torch.equal(bank.host_tables[4:], before[4:])
assert torch.equal(buffer.host_tables[4:], before[4:])


def test_device_metadata_rejects_foreign_and_reused_allocations_without_host_reads(monkeypatch):
Expand Down
Loading