MCPcopy Index your code
hub / github.com/python/cpython / busy_retry

Function busy_retry

Lib/test/support/__init__.py:2682–2721  ·  view source on GitHub ↗

Run the loop body until "break" stops the loop. After *timeout* seconds, raise an AssertionError if *error* is true, or just stop if *error is false. Example: for _ in support.busy_retry(support.SHORT_TIMEOUT): if check(): break Example of

(timeout, err_msg=None, /, *, error=True)

Source from the content-addressed store, hash-verified

2680
2681
2682def busy_retry(timeout, err_msg=None, /, *, error=True):
2683 """
2684 Run the loop body until "break" stops the loop.
2685
2686 After *timeout* seconds, raise an AssertionError if *error* is true,
2687 or just stop if *error is false.
2688
2689 Example:
2690
2691 for _ in support.busy_retry(support.SHORT_TIMEOUT):
2692 if check():
2693 break
2694
2695 Example of error=False usage:
2696
2697 for _ in support.busy_retry(support.SHORT_TIMEOUT, error=False):
2698 if check():
2699 break
2700 else:
2701 raise RuntimeError('my custom error')
2702
2703 """
2704 if timeout <= 0:
2705 raise ValueError("timeout must be greater than zero")
2706
2707 start_time = time.monotonic()
2708 deadline = start_time + timeout
2709
2710 while True:
2711 yield
2712
2713 if time.monotonic() >= deadline:
2714 break
2715
2716 if error:
2717 dt = time.monotonic() - start_time
2718 msg = f"timeout ({dt:.1f} seconds)"
2719 if err_msg:
2720 msg = f"{msg}: {err_msg}"
2721 raise AssertionError(msg)
2722
2723
2724def sleeping_retry(timeout, err_msg=None, /,

Calls

no outgoing calls

Used in the wild real call sites across dependent graphs

searching dependent graphs…