⚡️ Speed up function retry_with_backoff by -81%

[codeflash-ai-dev]

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

⏱️ Runtime : 5.66 milliseconds → 29.5 milliseconds (best of 233 runs)

📝 Explanation and details

The optimization replaces the blocking time.sleep() call with the non-blocking await asyncio.sleep(), which is a critical fix for async functions. However, the performance results show a counterintuitive pattern that reveals important nuances about async optimization.

Key Change:

• Replaced time.sleep(0.0001 * attempt) with await asyncio.sleep(0.0001 * attempt)
• Added import asyncio to support the async sleep functionality

Why This Matters:
The original code uses time.sleep(), which blocks the entire event loop during backoff delays. This prevents other coroutines from running concurrently and violates async best practices. The optimized version uses await asyncio.sleep(), which yields control back to the event loop, allowing other async operations to proceed during the delay.

Performance Analysis:
The line profiler shows the sleep operation went from 62.6% of total time (5.638ms) in the original to 25.1% (1.145ms) in the optimized version - a 79% reduction in sleep overhead. However, individual test runtime appears higher in the optimized version because:

1. Async sleep has more overhead for very short delays (0.0001 seconds)
2. Event loop scheduling adds minimal latency per operation
3. The real benefit is concurrency - other tasks can run during sleep periods

Throughput Benefits:
The 6.4% throughput improvement (153,738 → 163,566 ops/sec) demonstrates the optimization's value under concurrent load. The annotated tests show this particularly benefits scenarios with:

• Many concurrent executions (tests with 50-200 concurrent tasks)
• Mixed success/failure patterns where some operations sleep while others continue
• High-volume workloads where event loop efficiency matters

Impact:
This optimization is essential for any async application where retry_with_backoff might be called concurrently. The blocking sleep would create a bottleneck preventing proper async behavior, while the optimized version maintains event loop responsiveness and enables true concurrency.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 700 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
# src/asynchrony/various.py
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_success_first_try():
    # Test that a function that succeeds on the first try returns its result
    async def succeed():
        return "ok"
    result = await retry_with_backoff(succeed)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_second_try():
    # Test that a function that fails once then succeeds returns its result
    state = {"calls": 0}
    async def sometimes_fail():
        if state["calls"] == 0:
            state["calls"] += 1
            raise ValueError("fail first")
        return 42
    result = await retry_with_backoff(sometimes_fail)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_third_try():
    # Test that a function that fails twice then succeeds returns its result
    state = {"calls": 0}
    async def fail_then_succeed():
        if state["calls"] < 2:
            state["calls"] += 1
            raise RuntimeError("fail")
        return "done"
    result = await retry_with_backoff(fail_then_succeed, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_raises_after_max_retries():
    # Test that a function that always fails raises after max_retries
    async def always_fail():
        raise KeyError("nope")
    with pytest.raises(KeyError) as exc_info:
        await retry_with_backoff(always_fail, max_retries=4)

@pytest.mark.asyncio
async def test_retry_with_backoff_default_max_retries():
    # Test that default max_retries is 3
    state = {"calls": 0}
    async def fail_then_succeed():
        if state["calls"] < 2:
            state["calls"] += 1
            raise Exception("fail")
        return "ok"
    # Should succeed on 3rd try (default 3 retries)
    result = await retry_with_backoff(fail_then_succeed)

# 2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries():
    # Test that max_retries < 1 raises ValueError
    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)

@pytest.mark.asyncio

async def test_retry_with_backoff_preserves_exception_type():
    # Test that the original exception type is preserved after retries
    async def fail():
        raise ZeroDivisionError("bad math")
    with pytest.raises(ZeroDivisionError):
        await retry_with_backoff(fail, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_async_func_returns_none():
    # Test that a function returning None is handled correctly
    async def returns_none():
        return None
    result = await retry_with_backoff(returns_none)

@pytest.mark.asyncio

async def test_retry_with_backoff_exception_on_last_try():
    # Test that exception is raised only after all retries are exhausted
    state = {"calls": 0}
    async def fail_n_times():
        state["calls"] += 1
        raise ValueError("fail")
    with pytest.raises(ValueError):
        await retry_with_backoff(fail_n_times, max_retries=2)

# 3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_successes():
    # Test many concurrent successful executions
    async def succeed(i):
        return i * 2
    tasks = [retry_with_backoff(lambda i=i: succeed(i)) for i in range(50)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_failures():
    # Test many concurrent failures, all should raise
    async def fail():
        raise Exception("fail")
    tasks = [retry_with_backoff(fail, max_retries=2) for _ in range(10)]
    results = await asyncio.gather(*tasks, return_exceptions=True)

@pytest.mark.asyncio
async def test_retry_with_backoff_mixed_concurrent():
    # Test a mix of success and failure in concurrent calls
    async def sometimes_fail(i):
        if i % 3 == 0:
            return i
        raise ValueError("fail")
    tasks = [retry_with_backoff(lambda i=i: sometimes_fail(i), max_retries=2) for i in range(20)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if i % 3 == 0:
            pass
        else:
            pass

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_small_load():
    # Throughput test: small number of concurrent successful calls
    async def succeed():
        return "ok"
    tasks = [retry_with_backoff(succeed) for _ in range(10)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load():
    # Throughput test: medium concurrent calls with some failures
    async def sometimes_fail(i):
        if i % 4 == 0:
            return i
        raise RuntimeError("fail")
    tasks = [retry_with_backoff(lambda i=i: sometimes_fail(i), max_retries=3) for i in range(40)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if i % 4 == 0:
            pass
        else:
            pass

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

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_failure_ratio():
    # Throughput test: high volume, most fail, some succeed
    async def sometimes_fail(i):
        if i % 10 == 0:
            return i
        raise Exception("fail")
    tasks = [retry_with_backoff(lambda i=i: sometimes_fail(i), max_retries=2) for i in range(50)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if i % 10 == 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
# --- 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
# -------------------------------

# 1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_value_on_first_try():
    """Test that the function returns the correct value when no retries are needed."""
    async def always_succeed():
        return "success"
    result = await retry_with_backoff(always_succeed)

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_value_after_one_retry():
    """Test that the function retries once and returns the correct value."""
    calls = {"count": 0}
    async def succeed_on_second_try():
        calls["count"] += 1
        if calls["count"] == 1:
            raise ValueError("fail first")
        return "ok"
    result = await retry_with_backoff(succeed_on_second_try)

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_value_after_multiple_retries():
    """Test that the function retries up to max_retries and returns the correct value."""
    calls = {"count": 0}
    async def succeed_on_third_try():
        calls["count"] += 1
        if calls["count"] < 3:
            raise RuntimeError("fail")
        return 42
    result = await retry_with_backoff(succeed_on_third_try, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_raises_on_all_failures():
    """Test that the function raises the last exception if all retries fail."""
    async def always_fail():
        raise KeyError("never succeeds")
    with pytest.raises(KeyError, match="never succeeds"):
        await retry_with_backoff(always_fail, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries():
    """Test that the function raises ValueError when max_retries < 1."""
    async def dummy():
        return 1
    with pytest.raises(ValueError, match="max_retries must be at least 1"):
        await retry_with_backoff(dummy, max_retries=0)

# 2. EDGE TEST CASES

@pytest.mark.asyncio

async def test_retry_with_backoff_preserves_exception_type_and_message():
    """Test that the last exception's type and message are preserved."""
    class CustomError(Exception):
        pass
    async def fail_custom():
        raise CustomError("custom fail")
    with pytest.raises(CustomError, match="custom fail"):
        await retry_with_backoff(fail_custom, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_exception_on_first_and_last_attempt():
    """Test that the last exception is raised if all attempts fail with different errors."""
    errors = [ValueError("first"), RuntimeError("second"), KeyError("third")]
    calls = {"count": 0}
    async def fail_differently():
        idx = calls["count"]
        calls["count"] += 1
        raise errors[idx]
    with pytest.raises(KeyError, match="third"):
        await retry_with_backoff(fail_differently, max_retries=3)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_returns_none():
    """Test that the function works when the async func returns None."""
    async def return_none():
        return None
    result = await retry_with_backoff(return_none)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_is_coroutine_function():
    """Test that the function works with different coroutine functions."""
    async def add(a, b):
        return a + b
    result = await retry_with_backoff(lambda: add(2, 3))

# 3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_successes():
    """Test many concurrent successful executions."""
    async def always_succeed(i):
        return i * i
    tasks = [retry_with_backoff(lambda i=i: always_succeed(i)) for i in range(100)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_failures():
    """Test many concurrent failures and check all raise."""
    async def always_fail(i):
        raise RuntimeError(f"fail {i}")
    tasks = [retry_with_backoff(lambda i=i: always_fail(i), max_retries=2) for i in range(20)]
    for task in tasks:
        with pytest.raises(RuntimeError):
            await task

@pytest.mark.asyncio
async def test_retry_with_backoff_mixed_success_and_failure():
    """Test a mix of successful and failing concurrent executions."""
    async def maybe_fail(i):
        if i % 3 == 0:
            return i
        else:
            raise ValueError(f"fail {i}")
    tasks = [retry_with_backoff(lambda i=i: maybe_fail(i), max_retries=2) for i in range(15)]
    results = []
    for i, task in enumerate(tasks):
        if i % 3 == 0:
            results.append(await task)
        else:
            with pytest.raises(ValueError):
                await task

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_small_load():
    """Test throughput under a small concurrent load."""
    async def fast_func(i):
        return i + 1
    tasks = [retry_with_backoff(lambda i=i: fast_func(i)) for i in range(10)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load():
    """Test throughput under a medium concurrent load."""
    async def sometimes_fail(i):
        # Fail once for even i, succeed immediately for odd i
        state = {"count": 0}
        async def inner():
            state["count"] += 1
            if i % 2 == 0 and state["count"] == 1:
                raise Exception("fail once")
            return i
        return await retry_with_backoff(inner, max_retries=2)
    tasks = [sometimes_fail(i) for i in range(50)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume():
    """Test throughput under high concurrent load (but < 1000)."""
    async def fast_func(i):
        return i * 2
    tasks = [retry_with_backoff(lambda i=i: fast_func(i)) for i in range(200)]
    results = await asyncio.gather(*tasks)
# 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-mhq3zgyd and push.