Skip to content

Share budgeted CUDA cache recovery with gradient handoff #888

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
bradhilton wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Share budgeted CUDA cache recovery with gradient handoff #888

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
53053bf
Share measured recovery budget with post-forward cache handoff
bradhilton Sep 17, 2026
f557bc4
Fix CPU allocator fixtures and route Megatron handoff tests
bradhilton Sep 17, 2026
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
4 changes: 4 additions & 0 deletions .github/workflows/prek.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,8 @@ jobs:
tests/unit/test_prefix_tree_attention_builder.py \
tests/unit/test_prefix_tree_grad_parity.py \
tests/unit/test_prefix_tree_packing.py \
tests/unit/test_trainer_rank_handoff_budget.py \
tests/unit/test_trainer_rank_physical_reserve.py \
tests/unit/test_trainer_rank_validation.py \
tests/unit/test_trainer_rank_weird_shapes.py \
tests/unit/test_trainer_rank_split.py \
Expand Down Expand Up @@ -252,5 +254,7 @@ jobs:
--ignore=tests/unit/test_prefix_tree_attention_builder.py \
--ignore=tests/unit/test_prefix_tree_grad_parity.py \
--ignore=tests/unit/test_prefix_tree_packing.py \
--ignore=tests/unit/test_trainer_rank_handoff_budget.py \
--ignore=tests/unit/test_trainer_rank_physical_reserve.py \
--ignore=tests/unit/test_trainer_rank_validation.py \
--ignore=tests/unit/test_trainer_rank_weird_shapes.py
155 changes: 121 additions & 34 deletions src/art/trainer_rank/_impl.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2363,24 +2363,37 @@ def _forward_micro_batches(
items, start, checkpoint=checkpoint
)
self._snapshot_planning_telemetry(candidate.plan, candidate.check)
if isinstance(candidate.plan, _FlatForwardPlan):
tracked_outputs, memory_baseline = (
self._run_flat_plan_with_memory_tracking(
candidate.plan,
check=candidate.check,
context="forward_micro_batches",
tracked_outputs: list[AnyForwardOutput] = []
outputs: list[Any] = []
flat_outputs = iter(tracked_outputs)
error: BaseException | None = None
try:
if isinstance(candidate.plan, _FlatForwardPlan):
tracked_outputs, memory_baseline = (
self._run_flat_plan_with_memory_tracking(
candidate.plan,
check=candidate.check,
context="forward_micro_batches",
)
)
)
else:
tracked_outputs, memory_baseline, forward_peak = (
self._execute_split_plan_with_memory_tracking(
candidate.plan,
check=candidate.check,
context="forward_micro_batches",
else:
tracked_outputs, memory_baseline, forward_peak = (
self._execute_split_plan_with_memory_tracking(
candidate.plan,
check=candidate.check,
context="forward_micro_batches",
)
)
)
flat_outputs = iter(tracked_outputs)
outputs = [_unflatten(item, flat_outputs) for item in candidate.inputs]
flat_outputs = iter(tracked_outputs)
outputs = [_unflatten(item, flat_outputs) for item in candidate.inputs]
except BaseException as exc:
error = exc
try:
self._release_cached_memory_for_backward(candidate.plan, error=error)
except BaseException:
# Do not retain our completed graph through a new handoff traceback.
del tracked_outputs, flat_outputs, outputs
raise
stop = start + candidate.stats_global_count
if stop < len(items):
self._last_global_micro_batch_size = max(
Expand Down Expand Up @@ -2434,6 +2447,40 @@ def _forward_micro_batches(
del tracked_outputs, flat_outputs, outputs
start = stop

def _release_cached_memory_for_backward(
self, plan: _AnyForwardPlan, *, error: BaseException | None = None
) -> None:
# Every WORLD wave reaches this before the public iterator skips empty
# outputs. Forward has already executed: never replan or retry here.
with self._cache_recovery_episode() as (owner, started):
exchange_error: BaseException | None = None
try:
failed, gradients = self._recovery_reduce(
[
float(error is not None),
float(any(group.grad_enabled for group in plan.groups)),
],
op="MAX",
sync_across_dp=True,
)
except BaseException as exc:
if error is None:
raise
exchange_error = exc
if error is not None:
raise self._memory_error_with_reduction_note(error, exchange_error)
if failed:
raise RuntimeError("Forward failed on another rank before handoff")
if not gradients:
return
self._try_cache_recovery(
None,
sync_across_dp=True,
owner=owner,
started=started,
handoff_grad=any(group.grad_enabled for group in plan.groups),
)

@overload
def dp_rank_forward(
self,
Expand Down Expand Up @@ -5187,14 +5234,7 @@ def finish(value: Any) -> Any:
return result
assert refused is not None
original = refused.error(context)
state = self._recovery_state()
started = self._recovery_clock()
owner = object()
with state.lock:
if state.owner is None:
state.owner = owner
primary: BaseException | None = None
try:
with self._cache_recovery_episode() as (owner, started):
if not isinstance(value, _ForwardRefusal):
# A formerly fitting width is not proof that the minimum cannot fit.
value = search()
Expand All @@ -5221,6 +5261,18 @@ def finish(value: Any) -> Any:
self._snapshot_planning_telemetry(refused.plan, refused.check)
latest = refused.error(context)
raise latest from original

@contextmanager
def _cache_recovery_episode(self) -> Iterator[tuple[object, float | None]]:
state = self._recovery_state()
started = self._recovery_clock()
owner = object()
with state.lock:
if state.owner is None:
state.owner = owner
primary: BaseException | None = None
try:
yield owner, started
except BaseException as exc:
primary = exc
raise
Expand Down Expand Up @@ -5335,11 +5387,12 @@ def _memory_error_with_reduction_note(

def _try_cache_recovery(
self,
check: _MemoryCheck,
check: _MemoryCheck | None,
*,
sync_across_dp: bool,
owner: object,
started: float | None,
handoff_grad: bool = False,
) -> bool:
state = self._recovery_state()
now = self._recovery_clock()
Expand Down Expand Up @@ -5371,7 +5424,7 @@ def _try_cache_recovery(
invalid |= any(not math.isfinite(value) for value in (*costs, sum(costs)))
values = self._recovery_reduce(
[
float(check.estimated_required_bytes),
float(check.estimated_required_bytes) if check is not None else 0.0,
0.0 if invalid else state.work,
float(state.first_consumed),
float(invalid),
Expand All @@ -5388,16 +5441,20 @@ def _try_cache_recovery(
needed = False
cap_blocks = False
try:
available = self._available_memory_bytes()
available = self._available_memory_bytes() if check is not None else 0
if (
available < required
(available < required if check is not None else handoff_grad)
and self.device.type == "cuda"
and torch.cuda.is_available()
and torch.cuda.get_allocator_backend() == "native"
):
free, total = torch.cuda.mem_get_info(self.device)
needed = int(free) < required + int(total * _MEMORY_RESERVE_FRACTION)
if os.environ.get(_TEST_HOOKS_ENV) == "1":
if check is None:
needed &= int(torch.cuda.memory_reserved(self.device)) > int(
torch.cuda.memory_allocated(self.device)
)
elif os.environ.get(_TEST_HOOKS_ENV) == "1":
limit = os.environ.get(_TEST_MEMORY_LIMIT_ENV)
if limit:
cap_blocks = required > max(
Expand Down Expand Up @@ -5427,7 +5484,7 @@ def _try_cache_recovery(
if sampled[0] < 0:
raise RuntimeError("Memory recovery sampling failed on another rank")
state.invalid |= not bool(sampled[3])
if required <= sampled[0]:
if check is not None and required <= sampled[0]:
return True
if sampled[1] == 0 or sampled[2] == 0 or state.invalid:
return False
Expand All @@ -5445,9 +5502,37 @@ def _try_cache_recovery(
# physical condition again immediately before the sole call.
free, total = torch.cuda.mem_get_info(self.device)
if int(free) < required + int(total * _MEMORY_RESERVE_FRACTION):
attempted = True
torch.cuda.empty_cache()
available = self._available_memory_bytes()
if check is not None:
attempted = True
torch.cuda.empty_cache()
else:
allocated = int(torch.cuda.memory_allocated(self.device))
reserved = int(torch.cuda.memory_reserved(self.device))
if reserved > allocated:
# A soft trigger, not calibrated library demand. The
# native release affects unused caches process-wide.
evidence = dict(
device=str(self.device),
reserve_trigger_bytes=int(
total * _MEMORY_RESERVE_FRACTION
),
physical_free_before_bytes=int(free),
allocated_bytes=allocated,
reserved_before_bytes=reserved,
)
with _telemetry_phase(
"gradient_handoff_cache_release", evidence
):
attempted = True
with torch.cuda.device(self.device):
torch.cuda.empty_cache()
evidence["physical_free_after_bytes"] = int(
torch.cuda.mem_get_info(self.device)[0]
)
evidence["reserved_after_bytes"] = int(
torch.cuda.memory_reserved(self.device)
)
available = self._available_memory_bytes() if check is not None else 0
except BaseException as exc:
error, available = exc, -1
exchange_error: BaseException | None = None
Expand All @@ -5467,7 +5552,9 @@ def _try_cache_recovery(
raise self._memory_error_with_reduction_note(error, exchange_error)
if sampled[0] < 0:
raise RuntimeError("Memory recovery failed on another rank")
# Rebuild pure search caches even when this fresh sample decreased.
# Admission rebuilds pure search caches even when the sample decreased.
# Handoff ignores this return: denied/insufficient recovery still yields
# the completed outputs, with no claim that backward will fit.
return True

def _memory_check_required(
Expand Down
Loading
Loading