Skip to content

Commit efce4bb

Browse files
lukebaumanncopybara-github
authored andcommitted
Add pause_resume decorator to Manager.
This change introduces a `pause_resume` decorator to the `Manager` class. This decorator wraps a function to automatically retry execution when a `jax.errors.JaxRuntimeError` occurs due to a slice down event. Before each attempt, it waits for all slices to be available and performs necessary cleanup of JAX caches and live arrays upon failure. PiperOrigin-RevId: 796970321
1 parent 5756f63 commit efce4bb

1 file changed

Lines changed: 69 additions & 1 deletion

File tree

pathwaysutils/elastic/manager.py

Lines changed: 69 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import collections
2929
from collections.abc import Callable, Mapping, Sequence
3030
import copy
31+
import functools
3132
import itertools
3233
import logging
3334
import time
@@ -44,7 +45,10 @@
4445

4546

4647
class ElasticRuntimeError(RuntimeError):
47-
"""Error raised when too many elastic down events or reshard retries occur."""
48+
"""Error raised when elasticity cannot continue.
49+
50+
Some causes of this error are due to too many elastic down events or retries.
51+
"""
4852

4953

5054
class Manager:
@@ -718,3 +722,67 @@ def wait_for_slices(
718722
)
719723

720724
return good_slice_indices
725+
726+
def pause_resume(
727+
self,
728+
max_retries: int,
729+
wait_period: float | int = 10,
730+
timeout: float | None = None,
731+
) -> Any:
732+
"""Retries a function with pause/resume fault tolerance.
733+
734+
This decorator wraps a function to automatically retry execution in case of
735+
`jax.errors.JaxRuntimeError` caused by slice down events. It waits for
736+
available slices before each attempt and cleans up JAX caches on failure.
737+
The function will not be attempted (or reattempted) until all of the slices
738+
are available and will negate some of the benefits of late-binding.
739+
740+
Args:
741+
max_retries: The maximum number of times to retry the function.
742+
wait_period: The number of seconds to wait between availability checks.
743+
Defaults to 10 seconds.
744+
timeout: The maximum number of seconds to wait for slices to become
745+
available before each retry attempt. If None, there is no timeout.
746+
747+
Returns:
748+
The result of the wrapped function.
749+
750+
Raises:
751+
ElasticRuntimeError: If all retry attempts fail.
752+
Exception: Any other exception raised by the wrapped function that is not
753+
due to a slice down event.
754+
"""
755+
def decorator(func):
756+
@functools.wraps(func)
757+
def wrapper(*args, **kwargs):
758+
for retry_index in range(max_retries):
759+
try:
760+
_logger.info(
761+
"Elastic attempt %d out of %d", retry_index + 1, max_retries
762+
)
763+
764+
self.wait_for_slices(wait_period=wait_period, timeout=timeout)
765+
766+
return func(*args, **kwargs)
767+
except jax.errors.JaxRuntimeError as error:
768+
if not self.is_error_due_to_slice_down(error):
769+
raise
770+
771+
try:
772+
_logger.info("Cleaning up any ongoing traces")
773+
jax.profiler.stop_trace()
774+
except (RuntimeError, ValueError) as e:
775+
_logger.info("No ongoing traces to clean up")
776+
except Exception:
777+
_logger.exception("Error cleaning up ongoing traces")
778+
raise
779+
780+
jax.clear_caches()
781+
for array in jax.live_arrays():
782+
array.delete()
783+
raise ElasticRuntimeError(
784+
f"Elastic attempt {max_retries} out of {max_retries} failed."
785+
)
786+
787+
return wrapper
788+
return decorator

0 commit comments

Comments
 (0)