Files
NexQuant/rdagent/utils/workflow/misc.py
T
you-n-g 09be71d586 feat: parallel loop running based on asyncio (#932)
* refactor: split workflow into pkg, add WorkflowTracker & wait_retry

* feat: add async LoopBase with parallel workers and step semaphores

* fix: replace pickle with dill and run blocking tasks via joblib wrapper

* feat: add log format settings, dynamic parallelism & pickle-based snapshot

* fix: default step semaphore to 1 and avoid subprocess when single worker

* merge bowen's changes

* merge tim's changes

* refactor: extract component task mapping, add conditional logger setup

* lint

* refactor: add type hints and safer remain_time metric logging in workflow

* lint

* fix: allow BadRequestError to be pickled via custom copyreg reducer

* fix: stop loop when LoopTerminationError is raised in LoopBase

* lint

* refactor: make log tag context-local using ContextVar for thread safety

* feat: add subproc_step flag and helper to decide subprocess execution

* fix: use ./cache path and normalize relative volume bind paths

* fix: reset loop_idx to 0 on loop restart/resume to ensure correct flow

* fix: avoid chmod on cache and input dirs in Env timeout wrapper

* fix: skip chmod on 'cache' and 'input' dirs using find -prune

* fix: restrict chmod to immediate mount dirs excluding cache/input

* fix: chmod cache and input dirs alongside their contents after entry run

* fix: guard chmod with directory checks for cache and input

* fix: prefix mount_path in chmod command for cache/input dirs

* fix: drop quotes from find exclude patterns to ensure chmod executes

* fix: skip chmod on cache/input directories to avoid warning spam

* feat: support string volume mappings and poll subprocess stdout/stderr

* support remove symbolic link

* test: use dynamic home path and code volume in LocalEnv local_simple

* fix: skip trace and progress update when loop step is withdrawn

* refactor: add clean_workspace util and non-destructive workspace backup

* fix: preserve symlinks when backing up workspace with copytree

* fix: prevent AttributeError when _pbar not yet initialized in LoopBase

* perf: replace shutil.copytree with rsync for faster workspace backup

* fix: cast log directory Path to str in tar command of data science loop

* fix: use portable 'cp -r -P' instead of rsync for workspace backup

* fix: add retry and logging to workspace backup for robustness

* refactor: extract backup_folder helper and reuse in DataScienceRDLoop

* fix: propagate backup errors & default _pbar getattr to avoid error

* fix the division by zero bug

* refactor: execute RD loops via asyncio.run and add necessary imports

* lint

* lint

* lint

---------

Co-authored-by: Xu <v-xuminrui@microsoft.com>
2025-06-12 11:44:14 +08:00

55 lines
1.8 KiB
Python

import time
from collections.abc import Callable
from typing import Any, TypeVar
ASpecificRet = TypeVar("ASpecificRet")
def wait_retry(
retry_n: int = 3, sleep_time: int = 1, transform_args_fn: Callable[[tuple, dict], tuple[tuple, dict]] | None = None
) -> Callable[[Callable[..., ASpecificRet]], Callable[..., ASpecificRet]]:
"""Decorator to wait and retry the function for retry_n times.
Example:
>>> import time
>>> @wait_retry(retry_n=2, sleep_time=1)
... def test_func():
... global counter
... counter += 1
... if counter < 3:
... raise ValueError("Counter is less than 3")
... return counter
>>> counter = 0
>>> try:
... test_func()
... except ValueError as e:
... print(f"Caught an exception: {e}")
Error: Counter is less than 3
Error: Counter is less than 3
Caught an exception: Counter is less than 3
>>> counter
2
"""
assert retry_n > 0, "retry_n should be greater than 0"
def decorator(f: Callable[..., ASpecificRet]) -> Callable[..., ASpecificRet]:
def wrapper(*args: Any, **kwargs: Any) -> ASpecificRet:
for i in range(retry_n + 1):
try:
return f(*args, **kwargs)
except Exception as e:
print(f"Error: {e}")
time.sleep(sleep_time)
if i == retry_n:
raise
# Update args and kwargs using the transform function if provided.
if transform_args_fn is not None:
args, kwargs = transform_args_fn(args, kwargs)
else:
# just for passing mypy CI.
return f(*args, **kwargs)
return wrapper
return decorator