File size: 3,156 Bytes
e1acc15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Per-slot window bookkeeping of CONTRACT.md 7.2 (v1.3), the worker's copy.
The rule (same as lane O3's remote_dspark/state.py WindowRange, which the Spark proxy uses for its rank-0 ring): a
request's rows form one contiguous position range [lo, hi], at most W long, stored at pos mod W. Applying rows
[p0, p0+n): empty or gap (p0 > hi+1) -> lo = p0; else roll back rows >= p0 and lo = min(lo, p0); then hi = p0+n-1,
lo = max(lo, hi-W+1). n == 0 is a pure rollback to anchor_pos-1 (requires hi+1 >= anchor_pos).
Because [lo, hi] never exceeds W consecutive positions, its pos-mod-W slots are distinct, and every position in it was
written by the current generation (lane O3's argument, CONTRACT 7.2); stale ring entries lie outside [lo, hi].
"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class WindowRange:
window: int
lo: int | None = None
hi: int | None = None
@property
def empty(self) -> bool:
return self.lo is None
def clear(self) -> None:
self.lo = self.hi = None
def apply(self, first_pos: int, n_rows: int, anchor_pos: int) -> None:
"""Raises ValueError (-> BAD_ENTRY) when the entry is inconsistent (CONTRACT 7.2 steps 2-4)."""
if not 0 <= n_rows <= self.window:
raise ValueError(f"n_rows {n_rows} outside 0..{self.window}")
if anchor_pos < 0:
raise ValueError(f"anchor_pos {anchor_pos} < 0")
if n_rows > 0:
if first_pos < 0:
raise ValueError(f"first_pos {first_pos} < 0")
if anchor_pos != first_pos + n_rows:
raise ValueError(f"anchor_pos {anchor_pos} != first_pos {first_pos} + n_rows {n_rows}")
p0 = first_pos
if self.empty or p0 > self.hi + 1:
self.lo = p0
else:
self.lo = min(self.lo, p0)
self.hi = p0 + n_rows - 1
else:
if self.empty or self.hi + 1 < anchor_pos:
raise ValueError(f"no rows sent and the window ends at {self.hi}, anchor_pos {anchor_pos}")
self.hi = anchor_pos - 1
if self.hi < self.lo:
self.clear()
return
self.lo = max(self.lo, self.hi - self.window + 1)
if self.hi != anchor_pos - 1:
raise ValueError(f"window ends at {self.hi}, anchor_pos {anchor_pos}")
def read_lo(self, anchor_pos: int) -> int:
"""Lowest position an anchor at anchor_pos reads: max(lo, anchor_pos - W, 0)."""
return max(self.lo if self.lo is not None else anchor_pos, anchor_pos - self.window, 0)
def rows_for_anchor(self, anchor_pos: int) -> int:
if self.empty or self.hi != anchor_pos - 1:
return 0
return max(0, self.hi - self.read_lo(anchor_pos) + 1)
def expected_rows(self, anchor_pos: int) -> int:
return min(self.window, max(anchor_pos, 0))
def positions(self, anchor_pos: int | None = None) -> range:
if self.empty:
return range(0)
start = self.lo if anchor_pos is None else self.read_lo(anchor_pos)
return range(start, self.hi + 1)
|