⚡️ Speed up function retry_with_backoff by -63%

[codeflash-ai]

📄 -63% (-0.63x) speedup for retry_with_backoff in src/asynchrony/various.py

⏱️ Runtime : 17.7 milliseconds → 47.8 milliseconds (best of 224 runs)

📝 Explanation and details

The optimization replaces time.sleep() with await asyncio.sleep(), which is a critical fix for asynchronous code. While the individual function runtime appears slower (-62%), this is misleading because the original code was blocking the entire event loop during sleep operations.

Key optimization:

• Blocking → Non-blocking sleep: time.sleep(0.0001 * attempt) blocks the entire event loop, while await asyncio.sleep(0.0001 * attempt) yields control back to the event loop, allowing other coroutines to run concurrently.

Why this leads to better performance:

• The line profiler shows the sleep operation taking 17.5ms (75% of total time) in the original vs 3.2ms (35.8% of total time) in the optimized version
• Throughput improvement of 10.3% (184,324 → 203,392 ops/sec) demonstrates the real benefit: the event loop can process more operations concurrently
• In async environments, blocking calls create bottlenecks that prevent other coroutines from executing

Impact on workloads:

• Concurrent execution: When multiple retry operations run simultaneously (as shown in the test cases with asyncio.gather()), the optimized version allows all coroutines to progress during sleep periods
• Event loop efficiency: Non-blocking sleep prevents the entire application from freezing during backoff periods
• Scalability: The throughput improvement becomes more significant as the number of concurrent retry operations increases

The individual runtime being slower is expected because asyncio.sleep() has more overhead than time.sleep(), but the concurrent processing capability more than compensates for this, resulting in better overall system throughput.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 908 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# function to test
import time

import pytest  # used for our unit tests
from src.asynchrony.various import retry_with_backoff

# ---------------------- UNIT TESTS ----------------------

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_expected_value():
    # Test that the function returns the expected value when func succeeds
    async def success_func():
        return "success"
    result = await retry_with_backoff(success_func)

@pytest.mark.asyncio
async def test_retry_with_backoff_basic_async_await_behavior():
    # Test that retry_with_backoff can await an async function
    called = []
    async def func():
        called.append(True)
        return 42
    result = await retry_with_backoff(func)

@pytest.mark.asyncio
async def test_retry_with_backoff_default_max_retries():
    # Test that default max_retries is 3
    attempts = []
    async def flaky_func():
        attempts.append(True)
        if len(attempts) < 3:
            raise ValueError("fail")
        return "ok"
    result = await retry_with_backoff(flaky_func)

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_retry_with_backoff_max_retries_less_than_one():
    # Should raise ValueError if max_retries < 1
    async def dummy():
        return "x"
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=0)
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=-5)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_always_fails():
    # Should raise the last exception if func always fails
    async def always_fail():
        raise RuntimeError("fail always")
    with pytest.raises(RuntimeError) as excinfo:
        await retry_with_backoff(always_fail, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_succeeds_on_last_try():
    # Should succeed if func passes on last retry
    attempts = []
    async def eventually_success():
        attempts.append(True)
        if len(attempts) < 4:
            raise Exception("not yet")
        return "finally"
    result = await retry_with_backoff(eventually_success, max_retries=4)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_raises_different_exceptions():
    # Should raise the last exception if all attempts fail with different errors
    errors = [ValueError("A"), KeyError("B"), RuntimeError("C")]
    async def multi_error():
        raise errors.pop(0)
    with pytest.raises(RuntimeError) as excinfo:
        await retry_with_backoff(multi_error, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_concurrent_execution():
    # Test concurrent calls to retry_with_backoff
    async def sometimes_fail(i):
        if i % 2 == 0:
            return i
        else:
            raise ValueError("fail")
    async def wrapper(i):
        async def func():
            return await sometimes_fail(i)
        try:
            return await retry_with_backoff(func, max_retries=2)
        except Exception:
            return "error"
    results = await asyncio.gather(*(wrapper(i) for i in range(6)))

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_successes():
    # Run many concurrent successful calls
    async def always_success(i):
        return f"ok-{i}"
    coros = [retry_with_backoff(lambda i=i: always_success(i)) for i in range(50)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_failures():
    # Run many concurrent failing calls
    async def always_fail(i):
        raise Exception(f"fail-{i}")
    coros = [retry_with_backoff(lambda i=i: always_fail(i), max_retries=2) for i in range(30)]
    results = []
    for c in coros:
        try:
            await c
        except Exception as e:
            results.append(str(e))

@pytest.mark.asyncio
async def test_retry_with_backoff_large_scale_mixed():
    # Test a mix of success and failure at scale
    async def mixed(i):
        if i % 5 == 0:
            return i
        else:
            raise Exception(f"bad-{i}")
    async def wrapper(i):
        async def func():
            return await mixed(i)
        try:
            return await retry_with_backoff(func, max_retries=3)
        except Exception:
            return "fail"
    results = await asyncio.gather(*(wrapper(i) for i in range(50)))
    # Only multiples of 5 succeed
    for idx, res in enumerate(results):
        if idx % 5 == 0:
            pass
        else:
            pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_small_load():
    # Throughput test: small load, all succeed
    async def fast_success(i):
        return i
    coros = [retry_with_backoff(lambda i=i: fast_success(i)) for i in range(10)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load():
    # Throughput test: medium load, some fail
    async def sometimes_fail(i):
        if i % 3 == 0:
            return i
        else:
            raise ValueError("fail")
    async def wrapper(i):
        async def func():
            return await sometimes_fail(i)
        try:
            return await retry_with_backoff(func, max_retries=2)
        except Exception:
            return "error"
    coros = [wrapper(i) for i in range(40)]
    results = await asyncio.gather(*coros)
    for idx, res in enumerate(results):
        if idx % 3 == 0:
            pass
        else:
            pass

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume():
    # Throughput test: high volume, all succeed
    async def always_success(i):
        return i * 2
    coros = [retry_with_backoff(lambda i=i: always_success(i)) for i in range(100)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_sustained_execution():
    # Throughput test: sustained execution pattern, alternating success/failure
    async def alt_func(i):
        if i % 2 == 0:
            return f"even-{i}"
        else:
            raise Exception(f"odd-{i}")
    async def wrapper(i):
        async def func():
            return await alt_func(i)
        try:
            return await retry_with_backoff(func, max_retries=3)
        except Exception:
            return "fail"
    coros = [wrapper(i) for i in range(60)]
    results = await asyncio.gather(*coros)
    for idx, res in enumerate(results):
        if idx % 2 == 0:
            pass
        else:
            pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import asyncio  # used to run async functions
# function to test
import time

import pytest  # used for our unit tests
from src.asynchrony.various import retry_with_backoff

# -----------------------
# Basic Test Cases
# -----------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_basic_success():
    # Test that function returns the expected value on first try
    async def always_succeeds():
        return "success"
    result = await retry_with_backoff(always_succeeds)

@pytest.mark.asyncio
async def test_retry_with_backoff_basic_failure_then_success():
    # Test that function retries and succeeds on second attempt
    state = {"calls": 0}
    async def fail_once_then_succeed():
        if state["calls"] == 0:
            state["calls"] += 1
            raise ValueError("fail first")
        return "ok"
    result = await retry_with_backoff(fail_once_then_succeed, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_basic_all_failures():
    # Test that function raises the last exception after all retries fail
    async def always_fails():
        raise RuntimeError("always fails")
    with pytest.raises(RuntimeError) as excinfo:
        await retry_with_backoff(always_fails, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_basic_custom_max_retries():
    # Test with custom max_retries parameter
    state = {"calls": 0}
    async def fail_n_minus_one_then_succeed():
        if state["calls"] < 4:
            state["calls"] += 1
            raise Exception("fail")
        return "final"
    result = await retry_with_backoff(fail_n_minus_one_then_succeed, max_retries=5)

# -----------------------
# Edge Test Cases
# -----------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_edge_max_retries_one():
    # Should only try once, and raise if fails
    async def always_fails():
        raise ValueError("fail once")
    with pytest.raises(ValueError):
        await retry_with_backoff(always_fails, max_retries=1)

@pytest.mark.asyncio
async def test_retry_with_backoff_edge_max_retries_zero():
    # Should raise ValueError for invalid max_retries
    async def dummy():
        return 42
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=0)

@pytest.mark.asyncio
async def test_retry_with_backoff_edge_non_exception_return():
    # Test that function returns None if the coroutine returns None
    async def returns_none():
        return None
    result = await retry_with_backoff(returns_none)

@pytest.mark.asyncio
async def test_retry_with_backoff_edge_concurrent_success():
    # Test concurrent execution where all succeed
    async def always_succeeds():
        return "ok"
    coros = [retry_with_backoff(always_succeeds) for _ in range(10)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_edge_concurrent_mixed():
    # Test concurrent execution where some fail and some succeed
    async def sometimes_fails(i):
        if i % 2 == 0:
            return f"even-{i}"
        else:
            raise ValueError(f"odd-{i}")
    coros = [retry_with_backoff(lambda i=i: sometimes_fails(i), max_retries=1) for i in range(6)]
    results = []
    for i, coro in enumerate(coros):
        if i % 2 == 0:
            results.append(await coro)
        else:
            with pytest.raises(ValueError) as excinfo:
                await coro

@pytest.mark.asyncio
async def test_retry_with_backoff_edge_exception_type_preserved():
    # Test that the last exception type is preserved
    state = {"calls": 0}
    async def fail_then_fail_with_different_type():
        if state["calls"] == 0:
            state["calls"] += 1
            raise KeyError("first fail")
        raise IndexError("second fail")
    with pytest.raises(IndexError) as excinfo:
        await retry_with_backoff(fail_then_fail_with_different_type, max_retries=2)

# -----------------------
# Large Scale Test Cases
# -----------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_large_scale_many_concurrent_success():
    # Test many concurrent calls that all succeed
    async def always_succeeds():
        return 123
    coros = [retry_with_backoff(always_succeeds) for _ in range(100)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_large_scale_many_concurrent_failures():
    # Test many concurrent calls that all fail
    async def always_fails():
        raise RuntimeError("fail")
    coros = [retry_with_backoff(always_fails, max_retries=3) for _ in range(50)]
    for coro in coros:
        with pytest.raises(RuntimeError):
            await coro

@pytest.mark.asyncio
async def test_retry_with_backoff_large_scale_partial_success():
    # Test large scale where some succeed and some fail
    async def succeed_or_fail(i):
        if i % 10 == 0:
            return f"success-{i}"
        else:
            raise Exception(f"fail-{i}")
    coros = [retry_with_backoff(lambda i=i: succeed_or_fail(i), max_retries=2) for i in range(40)]
    for i, coro in enumerate(coros):
        if i % 10 == 0:
            result = await coro
        else:
            with pytest.raises(Exception) as excinfo:
                await coro

# -----------------------
# Throughput Test Cases
# -----------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_small_load():
    # Test throughput with small load
    async def simple():
        return "done"
    coros = [retry_with_backoff(simple) for _ in range(10)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load():
    # Test throughput with medium load and some failures
    async def sometimes_fails(i):
        if i % 3 == 0:
            return f"ok-{i}"
        else:
            raise Exception(f"fail-{i}")
    coros = [retry_with_backoff(lambda i=i: sometimes_fails(i), max_retries=2) for i in range(30)]
    for i, coro in enumerate(coros):
        if i % 3 == 0:
            result = await coro
        else:
            with pytest.raises(Exception) as excinfo:
                await coro

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume():
    # Test throughput with high volume, all succeed
    async def always_succeeds():
        return "high"
    coros = [retry_with_backoff(always_succeeds) for _ in range(200)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume_failures():
    # Test throughput with high volume, all fail
    async def always_fails():
        raise RuntimeError("fail-high")
    coros = [retry_with_backoff(always_fails, max_retries=2) for _ in range(100)]
    for coro in coros:
        with pytest.raises(RuntimeError) as excinfo:
            await coro
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-retry_with_backoff-mhpyv85f and push.