⚡️ Speed up function retry_with_backoff by 5%

[codeflash-ai-dev]

📄 5% (0.05x) speedup for retry_with_backoff in src/asynchrony/various.py

⏱️ Runtime : 142 milliseconds → 136 milliseconds (best of 156 runs)

📝 Explanation and details

The optimization replaces blocking time.sleep() with non-blocking await asyncio.sleep(), delivering a 200% throughput improvement (from 21,216 to 63,648 operations/second) and 4% runtime speedup.

Key Change:

• Line 12: time.sleep(0.0001 * attempt) → await asyncio.sleep(0.0001 * attempt)
• Import: import time → import asyncio

Why This Works:
The original code used blocking time.sleep(), which blocks the entire event loop during backoff delays. This prevents other async operations from running concurrently, creating a bottleneck. The line profiler shows the blocking sleep consumed 98% of execution time (144.7ms out of 147.7ms total).

The optimized version uses await asyncio.sleep(), which yields control back to the event loop during delays. This allows other coroutines to execute concurrently while retries are backing off, dramatically improving overall system throughput.

Performance Impact:

• Throughput: 3x improvement in concurrent scenarios where multiple retry operations can overlap
• Runtime: 4% faster due to reduced event loop blocking overhead
• Concurrency: Enables proper async behavior - other operations can run during backoff periods

Test Case Benefits:
The optimization particularly excels in concurrent test scenarios (like test_retry_with_backoff_many_concurrent_calls and throughput tests) where multiple retry operations can now execute simultaneously instead of blocking each other. Single-operation tests see modest improvements, but the real gains come from preserving async concurrency semantics.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 408 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
# --- DO NOT MODIFY BELOW ---
import time

import pytest  # used for our unit tests
from src.asynchrony.various import \
    retry_with_backoff  # --- DO NOT MODIFY ABOVE ---

# -------------------------------
# Unit Tests for retry_with_backoff
# -------------------------------

# BASIC TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_success_first_try():
    """Test that the function returns the correct value on the first try."""
    async def always_success():
        return 42
    result = await retry_with_backoff(always_success)

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

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_exception_after_retries():
    """Test that the function raises the last exception after all retries fail."""
    async def always_fail():
        raise RuntimeError("persistent failure")
    with pytest.raises(RuntimeError) as excinfo:
        await retry_with_backoff(always_fail, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_max_retries_one():
    """Test that the function only tries once if max_retries=1."""
    calls = {"count": 0}
    async def fail_once():
        calls["count"] += 1
        raise KeyError("fail")
    with pytest.raises(KeyError):
        await retry_with_backoff(fail_once, max_retries=1)

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries():
    """Test that the function raises ValueError if max_retries < 1."""
    async def dummy():
        return 1
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=0)
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=-5)

# EDGE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_concurrent_success():
    """Test concurrent execution where all coroutines succeed on first try."""
    async def always_success():
        return "done"
    coros = [retry_with_backoff(always_success) for _ in range(10)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio

async def test_retry_with_backoff_exception_propagation():
    """Test that the last exception is the one raised after all retries."""
    class CustomError(Exception):
        pass
    errors = [ValueError("first"), KeyError("second"), CustomError("third")]
    state = {"idx": 0}
    async def fail_different_each_time():
        idx = state["idx"]
        state["idx"] += 1
        raise errors[idx]
    with pytest.raises(CustomError) as excinfo:
        await retry_with_backoff(fail_different_each_time, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_async_func_returns_coroutine():
    """Test that the function works if the provided func returns a coroutine."""
    async def inner():
        return "coroutine result"
    async def wrapper():
        return await inner()
    result = await retry_with_backoff(wrapper)

# LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_success():
    """Test that many concurrent calls succeed and return correct results."""
    async def echo(x):
        return x
    tasks = [retry_with_backoff(lambda x=x: echo(x)) for x in range(50)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_failures():
    """Test that many concurrent calls all fail and raise the correct exception."""
    async def always_fail():
        raise RuntimeError("fail!")
    tasks = [retry_with_backoff(always_fail, max_retries=2) for _ in range(20)]
    for task in tasks:
        with pytest.raises(RuntimeError):
            await task

# THROUGHPUT TEST CASES

@pytest.mark.asyncio

async def test_retry_with_backoff_throughput_medium_load():
    """Throughput: Test performance and correctness under a medium load."""
    async def quick_success(x):
        return x + 1
    tasks = [retry_with_backoff(lambda x=x: quick_success(x)) for x in range(100)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio

import asyncio  # used to run async functions
# function to test
# src/asynchrony/various.py
import time

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

# --------------------------
# UNIT TESTS FOR THE FUNCTION
# --------------------------

# 1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_success_first_try():
    # Test that the function returns the correct value when no exception is raised
    async def always_succeeds():
        return "success"
    result = await retry_with_backoff(always_succeeds)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_second_try():
    # Test that the function retries once after an exception and then succeeds
    calls = {"count": 0}
    async def fails_once_then_succeeds():
        if calls["count"] == 0:
            calls["count"] += 1
            raise ValueError("fail first time")
        return "recovered"
    result = await retry_with_backoff(fails_once_then_succeeds, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_third_try():
    # Test that the function retries twice and succeeds on the third attempt
    calls = {"count": 0}
    async def fails_twice_then_succeeds():
        if calls["count"] < 2:
            calls["count"] += 1
            raise RuntimeError("fail")
        return 42
    result = await retry_with_backoff(fails_twice_then_succeeds, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_raises_after_max_retries():
    # Test that the function raises the last exception after all retries fail
    calls = {"count": 0}
    async def always_fails():
        calls["count"] += 1
        raise KeyError("fail always")
    with pytest.raises(KeyError) as excinfo:
        await retry_with_backoff(always_fails, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_none():
    # Test that the function can return None as a valid value
    async def returns_none():
        return None
    result = await retry_with_backoff(returns_none)

# 2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_max_retries_one():
    # Test that the function only tries once if max_retries=1
    calls = {"count": 0}
    async def fails_once():
        calls["count"] += 1
        raise ValueError("fail once")
    with pytest.raises(ValueError):
        await retry_with_backoff(fails_once, max_retries=1)

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries_zero():
    # Test that the function raises ValueError for invalid max_retries < 1
    async def dummy():
        return 1
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=0)

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries_negative():
    # Test that the function raises ValueError for negative max_retries
    async def dummy():
        return 1
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=-5)

@pytest.mark.asyncio
async def test_retry_with_backoff_exception_type_preserved():
    # Test that the last exception is the one raised
    class CustomError1(Exception): pass
    class CustomError2(Exception): pass
    calls = {"count": 0}
    async def fails_with_different_errors():
        if calls["count"] == 0:
            calls["count"] += 1
            raise CustomError1("first")
        else:
            raise CustomError2("second")
    with pytest.raises(CustomError2) as excinfo:
        await retry_with_backoff(fails_with_different_errors, max_retries=2)

@pytest.mark.asyncio

async def test_retry_with_backoff_async_func_with_side_effects():
    # Test that side effects (like state changes) happen the expected number of times
    state = {"calls": 0}
    async def func():
        state["calls"] += 1
        if state["calls"] < 3:
            raise Exception("fail")
        return "done"
    result = await retry_with_backoff(func, max_retries=5)

# 3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_calls():
    # Test many concurrent calls to ensure isolation and correctness
    async def succeed_after_n(n):
        state = {"count": 0}
        async def inner():
            if state["count"] < n:
                state["count"] += 1
                raise Exception("fail")
            return n
        return await retry_with_backoff(inner, max_retries=n+1)
    # Launch 20 concurrent tasks with increasing retry counts
    results = await asyncio.gather(*(succeed_after_n(i) for i in range(20)))

@pytest.mark.asyncio
async def test_retry_with_backoff_large_number_of_successes():
    # Test a large number of successful calls to check for resource leaks
    async def always_succeeds():
        return "ok"
    results = await asyncio.gather(*(retry_with_backoff(always_succeeds) for _ in range(100)))

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio

async def test_retry_with_backoff_throughput_medium_load():
    # Medium load throughput: 50 concurrent calls, all succeed
    async def always_succeeds(i):
        return i * 2
    results = await asyncio.gather(*[retry_with_backoff(lambda i=i: always_succeeds(i)) for i in range(50)])

@pytest.mark.asyncio

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