Skip to content

Commit b69e307

Browse files
authored
fix: preserve both private draft pages in shifted prefix checkpoints (#165)
A draft page table shifted by one physical page can have a complete draft page that is still private at the target prefix endpoint. Snapshotting only the partial page loses that history after restore, including at consumed=257 with 256-token pages. Preserve the preceding boundary page and the current partial page, and validate conflicts across both parts before copying. Expand GLM continuation storage and its planning costs accordingly. Qwen's existing custom checkpoint path already preserves both boundary pages; other shared paged recipes do not declare trailing draft pages. Update regression fixtures to use the native shifted table and shared prefix pages, and repair the existing single-rank maintenance fixture's completion-output wrapper. Validation on RTX PRO 4500 SM120: the original GPU reproducer lost both draft planes without a transfer error; the fix restores exact bytes. Transfer, maintenance and pool tests: 36 passed, 2 tensor-parallel tests skipped. Related CPU tests: 19 passed, 1 GPU test skipped. Full GLM serving and TP2/TP4 were not run. Existing exported packages remain loadable; re-export to obtain the fix. Design-Signed-Off-By: Codex (GPT-6); design-review=6aeb Signed-off-by: Maxrio9 <70799635+Maxrio9@users.noreply.github.com>
1 parent 87cc8c1 commit b69e307

5 files changed

Lines changed: 102 additions & 33 deletions

File tree

‎ignition/components/state_transfer.py‎

Lines changed: 43 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -111,29 +111,39 @@ def _validate_trailing(Sources, Destinations, Operations, Expected, Count, Error
111111
WIDTH: tl.constexpr, MemoryCapacity, PAGE_TOKENS: tl.constexpr,
112112
LIVE: tl.constexpr):
113113
row = tl.program_id(0)
114-
source, destination = tl.load(Sources + row), tl.load(Destinations + row)
114+
source = tl.load(Sources + row)
115115
PAGES = tl.load(MemoryCapacity + 1)
116-
endpoint = tl.maximum(0, tl.load(Expected + row) - 1)
116+
endpoint = tl.load(Expected + row)
117117
slot = tl.load(RequestSlots + row)
118118
state = tl.load(StateRows + slot * 8, (slot >= 0) & (slot < LIVE), other=0)
119119
snapshot = source == state
120120
active = (row < tl.load(Count)) & (tl.load(Errors + row) == 0) & (tl.load(Operations + row) == 0)
121-
tail = active & (endpoint % PAGE_TOKENS != 0)
122-
page = tl.load(Tables + slot * WIDTH + endpoint // PAGE_TOKENS, tail, other=-1)
123121
others = tl.arange(0, LIVE)
124-
other_source, other_dest = tl.load(Sources + others), tl.load(Destinations + others)
122+
other_source = tl.load(Sources + others)
125123
other_slot = tl.load(RequestSlots + others)
126124
other_state = tl.load(StateRows + other_slot * 8, (other_slot >= 0) & (other_slot < LIVE), other=0)
127125
other_snapshot = other_source == other_state
128-
other_endpoint = tl.maximum(0, tl.load(Expected + others) - 1)
126+
other_endpoint = tl.load(Expected + others)
129127
other_active = (others < tl.load(Count)) & (others != row) & (tl.load(Operations + others) == 0)
130-
other_active &= (other_slot >= 0) & (other_slot < LIVE) & (other_endpoint // PAGE_TOKENS < WIDTH)
131-
other_active &= other_endpoint % PAGE_TOKENS != 0
132-
other_page = tl.load(Tables + other_slot * WIDTH + other_endpoint // PAGE_TOKENS, other_active, other=-1)
133-
conflict = tl.sum((other_active & (page == other_page) & (~snapshot | ~other_snapshot)).to(tl.int32), 0) > 0
128+
other_active &= (other_slot >= 0) & (other_slot < LIVE)
134129
error = tl.load(Errors + row)
135-
error = tl.where(tail & ((page < 0) | (page >= PAGES)), 6, error)
136-
error = tl.where(tail & conflict, 4, error)
130+
for part in tl.static_range(2):
131+
logical = endpoint // PAGE_TOKENS + part - 1
132+
tail = active & (endpoint > 1) & (logical >= 0)
133+
if part == 1:
134+
tail &= endpoint % PAGE_TOKENS >= 2
135+
bounded = (logical >= 0) & (logical < WIDTH)
136+
page = tl.load(Tables + slot * WIDTH + logical, tail & bounded, other=-1)
137+
conflict = tl.full((), False, tl.int1)
138+
for other_part in tl.static_range(2):
139+
other_logical = other_endpoint // PAGE_TOKENS + other_part - 1
140+
used = other_active & (other_endpoint > 1) & (other_logical >= 0) & (other_logical < WIDTH)
141+
if other_part == 1:
142+
used &= other_endpoint % PAGE_TOKENS >= 2
143+
other_page = tl.load(Tables + other_slot * WIDTH + other_logical, used, other=-1)
144+
conflict |= tl.sum((used & (page == other_page) & (~snapshot | ~other_snapshot)).to(tl.int32), 0) > 0
145+
error = tl.where(tail & ((page < 0) | (page >= PAGES)), 6, error)
146+
error = tl.where(tail & conflict, 4, error)
137147
tl.store(Errors + row, error)
138148

139149

@@ -156,18 +166,28 @@ def _copy_rows(Backing, Sources, Destinations, Operations, Count, Errors,
156166
def _copy_pages(Pointers, Backing, Tables, Sources, Destinations, Operations, Expected, Count, Errors, RequestSlots, StateRows,
157167
ROW_BYTES: tl.constexpr, PAGE_BYTES: tl.constexpr, WIDTH: tl.constexpr,
158168
PAGE_TOKENS: tl.constexpr, BLOCK: tl.constexpr, LIVE: tl.constexpr = 4,
159-
ENDPOINT_SHIFT: tl.constexpr = 0):
169+
DRAFT_PART: tl.constexpr = -1):
160170
row, region, tile = tl.program_id(0), tl.program_id(1), tl.program_id(2)
161171
source, destination = tl.load(Sources + row), tl.load(Destinations + row)
162-
endpoint = tl.maximum(0, tl.load(Expected + row) + ENDPOINT_SHIFT)
172+
endpoint = tl.load(Expected + row)
163173
slot = tl.load(RequestSlots + row)
164174
state = tl.load(StateRows + slot * 8, (slot >= 0) & (slot < LIVE), other=0)
165175
snapshot = source == state
166176
active = (row < tl.load(Count)) & (tl.load(Errors + row) == 0)
167-
active &= (tl.load(Operations + row) == 0) & (endpoint % PAGE_TOKENS != 0)
168-
page = tl.load(Tables + slot * WIDTH + endpoint // PAGE_TOKENS, active, other=0)
177+
active &= tl.load(Operations + row) == 0
178+
logical = endpoint // PAGE_TOKENS
179+
if DRAFT_PART < 0:
180+
active &= endpoint % PAGE_TOKENS != 0
181+
else:
182+
logical += DRAFT_PART - 1
183+
active &= (endpoint > 1) & (logical >= 0)
184+
if DRAFT_PART == 1:
185+
active &= endpoint % PAGE_TOKENS >= 2
186+
page = tl.load(Tables + slot * WIDTH + logical, active, other=0)
169187
cache = tl.load(Pointers + region * 2).to(tl.pointer_type(tl.uint8))
170188
offset = tl.load(Pointers + region * 2 + 1)
189+
if DRAFT_PART >= 0:
190+
offset += DRAFT_PART * PAGE_BYTES
171191
saved_slot = tl.where(snapshot, destination, source)
172192
index = tile * BLOCK + tl.arange(0, BLOCK)
173193
saved = Backing + saved_slot * ROW_BYTES + offset + index
@@ -185,6 +205,9 @@ class PreparedStateTransfer:
185205
186206
``pages`` contains (state region name, contiguous uint8 [pages, page_bytes]
187207
cache) pairs. Cache owners and pointer tables remain alive until graph drain.
208+
``trailing_pages`` uses a one-page-shifted table with a one-token lag. Each
209+
saved region holds two pages: the preceding boundary page, then the current
210+
one. Both may be private even when the preceding draft page is complete.
188211
These maintenance operations are captured directly. Tensor-parallel callers
189212
must combine validation failures across ranks between validate() and apply().
190213
A nonzero command error suppresses both row and external-page writes.
@@ -233,11 +256,11 @@ def __init__(self, pool: StateStorage, commands: StateCommands, tables, *, pages
233256
_check_tensor(trailing_tables, tables.shape, torch.int32, device, "trailing_page_tables")
234257
for name, cache in trailing_pages:
235258
region = pool.layout.region(name)
236-
if name in names or region.dtype != torch.uint8 or cache.shape != (page_count, region.nbytes):
237-
raise ValueError("trailing partial pages must match the model page capacity and state regions")
259+
if name in names or region.dtype != torch.uint8 or cache.shape != (page_count, region.nbytes // 2) or region.nbytes % 2:
260+
raise ValueError("shifted draft pages require two saved pages per cache plane")
238261
_check_tensor(cache, cache.shape, torch.uint8, device, name)
239262
names.add(name)
240-
trailing_groups[region.nbytes].append((cache.data_ptr(), region.offset))
263+
trailing_groups[cache.shape[1]].append((cache.data_ptr(), region.offset))
241264
self.trailing_pointers = {size: torch.tensor(values, dtype=torch.int64, device=device)
242265
for size, values in trailing_groups.items()}
243266
self.pointers = {size: torch.tensor(values, dtype=torch.int64, device=device) for size, values in groups.items()}
@@ -253,7 +276,7 @@ def __init__(self, pool: StateStorage, commands: StateCommands, tables, *, pages
253276
for size, pointer in self.pointers.items()]
254277
self.page_args.extend((pointer, pool.backing, trailing_tables, c.source_slots, c.destination_slots, c.operation,
255278
c.expected_consumed, c.num_ops, c.errors, c.request_slots, state_rows, pool.layout.row_bytes, size, tables.shape[1], page_tokens,
256-
1024, self.capacity, -1) for size, pointer in self.trailing_pointers.items())
279+
1024, self.capacity, part) for size, pointer in self.trailing_pointers.items() for part in (0, 1))
257280
self.trailing_validation_args = None if not trailing_pages else (
258281
c.source_slots, c.destination_slots, c.operation, c.expected_consumed, c.num_ops, c.errors,
259282
trailing_tables, c.request_slots, state_rows, tables.shape[1], pool.memory_capacity, page_tokens, self.capacity)

‎ignition/recipes/glm53/pool.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,8 @@ def continuation_layout(layer_types, world_size, *, mtp=False):
4040
raise ValueError(f"unsupported canonical GLM attention type: {kind}")
4141
if mtp:
4242
regions.extend((("draft.index_tail", (2, 12, 128), torch.bfloat16),
43-
("draft.latent_page", (4, 64, 528), torch.uint8),
44-
("draft.index_page", (64, 132), torch.uint8)))
43+
("draft.latent_page", (2, 4, 64, 528), torch.uint8),
44+
("draft.index_page", (2, 64, 132), torch.uint8)))
4545
regions.extend((("last_hidden", (4096,), torch.bfloat16), ("consumed_tokens", (1,), torch.int64)))
4646
return StateLayout.from_specs(regions)
4747

‎ignition/recipes/glm53/recipe.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -88,9 +88,9 @@ def usage(prepared):
8888
ranks=MappingProxyType({
8989
# DGX Station can hold GLM on one GPU; TP1 must remain a supported rank count.
9090
# One rank holds every head: twice TP2's recurrent rows less the replicated part.
91-
1: _costs(161_123_584, _GRAPH_BYTES, ranks=1, speculating=True),
92-
2: _costs(81_464_576, _GRAPH_BYTES, ranks=2, speculating=True),
93-
4: _costs(41_635_072, _GRAPH_BYTES, ranks=4, speculating=True),
91+
1: _costs(161_267_200, _GRAPH_BYTES, ranks=1, speculating=True),
92+
2: _costs(81_608_192, _GRAPH_BYTES, ranks=2, speculating=True),
93+
4: _costs(41_778_688, _GRAPH_BYTES, ranks=4, speculating=True),
9494
}),
9595
plain_ranks=MappingProxyType({
9696
1: _costs(149_230_592, _GRAPH_BYTES, ranks=1, speculating=False),

‎tests/recipes/test_glm53_maintenance_cuda.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ def _run(rank, world, rendezvous=None):
6464
pages=pool.partial_pages(), page_tokens=256, state_rows=inputs.request_state_rows, requests=requests)
6565
slots, epochs, count = (inputs.row_slot_ids, inputs.row_generations, inputs.live_rows) if mode == "snapshot" else (
6666
admission.admission_slots, admission.admission_generations, admission.admission_count)
67-
local = LocalMaintenance(mode, requests, transfer, slots, epochs, count, records, frame, sequence,
67+
local = LocalMaintenance(mode, requests, transfer, slots, epochs, count, completion_outputs(records), frame, sequence,
6868
world_size=1, admission_buffers=admission)
6969
local.prepare()
7070
def call(local=local):

‎tests/recipes/test_glm53_state_transfer_cuda.py‎

Lines changed: 53 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -148,12 +148,16 @@ def test_rejected_commands_preserve_every_state_and_cache_byte(transfers, fault)
148148
assert torch.equal(cache, caches[name])
149149

150150

151-
@pytest.mark.parametrize("endpoint", [1, 4, 255, 256, 257, 512])
151+
@pytest.mark.parametrize("endpoint", [1, 2, 4, 255, 256, 257, 258, 511, 512, 513])
152152
def test_mtp_trailing_partial_page_restores_at_target_page_boundaries(endpoint):
153153
pool = ContinuationPool.allocate(("linear_attention", "indexed_attention"), 4, 8, 6, "cuda:0", mtp=True)
154154
commands = StateCommands.allocate("cuda:0")
155155
tables = torch.tensor([[0, 1, 2], [3, 4, 5], [-1, -1, -1], [-1, -1, -1]], device="cuda:0", dtype=torch.int32)
156-
draft_tables = tables.clone()
156+
# Helios shares complete target pages and shifts the draft view by one.
157+
shared = endpoint // 256
158+
tables[1, :shared] = tables[0, :shared]
159+
draft_tables = torch.full_like(tables, -1)
160+
draft_tables[:, :-1] = tables[:, 1:]
157161
requests = DeviceRequestState.allocate("cuda:0", vocab_size=128, consumed_tokens=pool.consumed_tokens[1:5, 0])
158162
state_rows = torch.arange(1, 5, device="cuda:0")[:, None].expand(4, 8).contiguous()
159163
operation = PreparedStateTransfer(pool, commands, tables, pages=pool.partial_pages(), page_tokens=256, state_rows=state_rows, requests=requests,
@@ -163,7 +167,11 @@ def test_mtp_trailing_partial_page_restores_at_target_page_boundaries(endpoint):
163167
pool.consumed_tokens[1] = endpoint
164168
for _, cache in (*pool.partial_pages(), *pool.draft_pages()):
165169
cache.random_(0, 256)
166-
expected = {name: cache[(endpoint - 1) // 256].clone() for name, cache in pool.draft_pages()}
170+
logical = [endpoint // 256 - 1] if endpoint > 1 and endpoint // 256 > 0 else []
171+
if endpoint > 1 and endpoint % 256 >= 2:
172+
logical.append(endpoint // 256)
173+
expected = {(name, page): cache[draft_tables[0, page]].clone()
174+
for name, cache in pool.draft_pages() for page in logical}
167175
graph = torch.cuda.CUDAGraph()
168176
_commands(commands, [(1, 5, 0, endpoint)])
169177
with torch.cuda.graph(graph):
@@ -177,12 +185,50 @@ def test_mtp_trailing_partial_page_restores_at_target_page_boundaries(endpoint):
177185
torch.cuda.synchronize()
178186
assert not commands.errors.count_nonzero()
179187
for name, cache in pool.draft_pages():
180-
if (endpoint - 1) % 256:
181-
torch.testing.assert_close(cache[3 + (endpoint - 1) // 256], expected[name], rtol=0, atol=0)
182-
if (endpoint - 1) % 256:
188+
for page in logical:
189+
torch.testing.assert_close(cache[draft_tables[1, page]], expected[name, page], rtol=0, atol=0)
190+
if logical:
183191
before = pool.backing.clone()
184-
draft_tables[1, (endpoint - 1) // 256] = -1
192+
draft_tables[1, logical[-1]] = -1
185193
graph.replay()
186194
torch.cuda.synchronize()
187195
assert commands.errors[0] == 6
188196
torch.testing.assert_close(pool.backing, before, rtol=0, atol=0)
197+
198+
199+
@pytest.mark.parametrize("snapshot", [False, True])
200+
def test_shifted_draft_boundary_aliases_are_validated_across_both_saved_pages(snapshot):
201+
pool = ContinuationPool.allocate(("indexed_attention",), 4, 8, 6, "cuda:0", mtp=True)
202+
commands = StateCommands.allocate("cuda:0")
203+
tables = torch.tensor([[0, 1, 2], [3, 4, 5], [-1, -1, -1], [-1, -1, -1]],
204+
device="cuda", dtype=torch.int32)
205+
# A preceding full page in lane zero aliases lane one's current partial page.
206+
draft_tables = torch.tensor([[4, 2, -1], [4, 5, -1], [-1, -1, -1], [-1, -1, -1]],
207+
device="cuda", dtype=torch.int32)
208+
requests = DeviceRequestState.allocate("cuda:0", vocab_size=128,
209+
consumed_tokens=pool.consumed_tokens[1:5, 0])
210+
state_rows = torch.arange(1, 5, device="cuda")[:, None].expand(4, 8).contiguous()
211+
operation = PreparedStateTransfer(pool, commands, tables, pages=pool.partial_pages(),
212+
page_tokens=256, state_rows=state_rows, requests=requests,
213+
trailing_pages=pool.draft_pages(), trailing_tables=draft_tables)
214+
operation.prepare()
215+
pool.backing.random_(0, 256)
216+
for _, cache in (*pool.partial_pages(), *pool.draft_pages()):
217+
cache.random_(0, 256)
218+
for live, saved, endpoint in ((1, 5, 257), (2, 6, 2)):
219+
pool.consumed_tokens[live if snapshot else saved] = endpoint
220+
rows = [(1, 5, 0, 257), (2, 6, 0, 2)] if snapshot else [(5, 1, 0, 257), (6, 2, 0, 2)]
221+
_commands(commands, rows)
222+
before = pool.backing.clone()
223+
caches = {name: cache.clone() for name, cache in (*pool.partial_pages(), *pool.draft_pages())}
224+
graph = torch.cuda.CUDAGraph()
225+
with torch.cuda.graph(graph):
226+
operation.validate()
227+
operation.apply()
228+
graph.replay()
229+
torch.cuda.synchronize()
230+
assert commands.errors[:2].tolist() == ([0, 0] if snapshot else [4, 4])
231+
if not snapshot:
232+
torch.testing.assert_close(pool.backing, before, rtol=0, atol=0)
233+
for name, cache in (*pool.partial_pages(), *pool.draft_pages()):
234+
torch.testing.assert_close(cache, caches[name], rtol=0, atol=0)

0 commit comments

Comments
 (0)