⚡️ Speed up function retry_with_backoff by 48%

[codeflash-ai]

📄 48% (0.48x) speedup for retry_with_backoff in src/async_examples/concurrency.py

⏱️ Runtime : 208 milliseconds → 141 milliseconds (best of 151 runs)

📝 Explanation and details

The optimization replaces time.sleep() with await asyncio.sleep() in the retry backoff mechanism. This change is critical for async code correctness and concurrency.

Key Performance Impact:

• 47% faster runtime (208ms → 141ms) due to proper async/await usage
• The line profiler shows the sleep operation time reduced from 3.36ms to 3.08ms (30.2% → 39.2% of total time)

Why This Optimization Works:

• time.sleep() is a blocking synchronous call that freezes the entire event loop, preventing any other async tasks from running during backoff periods
• await asyncio.sleep() is non-blocking and yields control back to the event loop, allowing concurrent execution of other async operations
• This enables proper async concurrency - while one retry operation is sleeping, other operations can proceed

Throughput Trade-off Explained:

• Individual throughput appears lower (-28.8%) because await asyncio.sleep() has slightly more overhead than the blocking time.sleep()
• However, this enables true concurrency - multiple retry operations can run simultaneously instead of blocking each other
• In real concurrent scenarios, the overall system throughput would be dramatically higher because the event loop isn't blocked

Test Case Performance:
The optimization particularly benefits concurrent test cases like test_retry_with_backoff_many_concurrent_* and test_retry_with_backoff_throughput_* where multiple retry operations run simultaneously. Without this fix, all concurrent operations would be serialized by the blocking sleep calls.

This is a mandatory fix for async code - using blocking calls in async functions breaks the async execution model.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 1137 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.async_examples.concurrency import 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 first attempt
    async def always_success():
        return "success"
    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 then succeeds
    state = {"calls": 0}
    async def fail_once_then_success():
        state["calls"] += 1
        if state["calls"] == 1:
            raise ValueError("fail first")
        return "success"
    result = await retry_with_backoff(fail_once_then_success)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_third_try():
    # Test that the function retries twice and then succeeds
    state = {"calls": 0}
    async def fail_twice_then_success():
        state["calls"] += 1
        if state["calls"] < 3:
            raise RuntimeError("fail")
        return "done"
    result = await retry_with_backoff(fail_twice_then_success)

@pytest.mark.asyncio
async def test_retry_with_backoff_raises_after_max_retries():
    # Test that the function raises if all retries fail
    state = {"calls": 0}
    async def always_fail():
        state["calls"] += 1
        raise KeyError("fail always")
    with pytest.raises(KeyError, match="fail always"):
        await retry_with_backoff(always_fail)

@pytest.mark.asyncio
async def test_retry_with_backoff_custom_max_retries():
    # Test that max_retries parameter is respected
    state = {"calls": 0}
    async def always_fail():
        state["calls"] += 1
        raise Exception("fail")
    with pytest.raises(Exception, match="fail"):
        await retry_with_backoff(always_fail, max_retries=5)

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

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries():
    # Test that invalid max_retries raises ValueError
    async def dummy():
        return "never called"
    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_exception_type_preserved():
    # Test that the last exception type is preserved
    state = {"calls": 0}
    async def fail_with_different_exceptions():
        state["calls"] += 1
        if state["calls"] == 1:
            raise ValueError("first")
        elif state["calls"] == 2:
            raise KeyError("second")
        else:
            raise RuntimeError("third")
    with pytest.raises(RuntimeError, match="third"):
        await retry_with_backoff(fail_with_different_exceptions)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_is_coroutine():
    # Test that retry_with_backoff works with coroutine functions
    async def coro_func():
        await asyncio.sleep(0)
        return 123
    result = await retry_with_backoff(coro_func)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_returns_none():
    # Test that retry_with_backoff correctly 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_raises_nonstandard_exception():
    # Test that retry_with_backoff propagates custom exceptions
    class CustomException(Exception):
        pass
    async def fail_custom():
        raise CustomException("custom fail")
    with pytest.raises(CustomException, match="custom fail"):
        await retry_with_backoff(fail_custom)

@pytest.mark.asyncio
async def test_retry_with_backoff_concurrent_execution():
    # Test concurrent execution with asyncio.gather
    state = {"calls": 0}
    async def sometimes_fail():
        state["calls"] += 1
        if state["calls"] % 2 == 0:
            return state["calls"]
        raise Exception("fail")
    coros = [retry_with_backoff(sometimes_fail) for _ in range(4)]
    results = await asyncio.gather(*coros, return_exceptions=True)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_is_lambda():
    # Test that retry_with_backoff works with lambda coroutines
    async def sample():
        return "lambda"
    result = await retry_with_backoff(lambda: sample())

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

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_success():
    # Test many concurrent successful executions
    async def always_success():
        await asyncio.sleep(0)
        return "ok"
    coros = [retry_with_backoff(always_success) for _ in range(50)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_failures():
    # Test many concurrent executions that all fail
    async def always_fail():
        await asyncio.sleep(0)
        raise Exception("fail")
    coros = [retry_with_backoff(always_fail) for _ in range(30)]
    results = await asyncio.gather(*coros, return_exceptions=True)

@pytest.mark.asyncio
async def test_retry_with_backoff_large_max_retries():
    # Test function with large max_retries, but succeed before hitting the max
    state = {"calls": 0}
    async def fail_then_success():
        state["calls"] += 1
        if state["calls"] < 10:
            raise Exception("fail")
        return "done"
    result = await retry_with_backoff(fail_then_success, max_retries=20)

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

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

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load_mixed():
    # Throughput test: medium load, mix of success and failure
    async def sometimes_fail(i):
        if i % 3 == 0:
            raise Exception("fail")
        return i
    coros = [retry_with_backoff(lambda i=i: sometimes_fail(i)) for i in range(40)]
    results = await asyncio.gather(*coros, return_exceptions=True)
    # Every third should fail
    for i, r in enumerate(results):
        if i % 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():
        await asyncio.sleep(0)
        return "bulk"
    coros = [retry_with_backoff(always_success) for _ in range(100)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume_failures():
    # Throughput test: high volume, all fail
    async def always_fail():
        await asyncio.sleep(0)
        raise Exception("bulk fail")
    coros = [retry_with_backoff(always_fail) for _ in range(80)]
    results = await asyncio.gather(*coros, return_exceptions=True)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_varying_max_retries():
    # Throughput test: varying max_retries per coroutine
    async def fail_then_success(i):
        if i % 2 == 0:
            raise Exception("fail")
        return i
    coros = [
        retry_with_backoff(lambda i=i: fail_then_success(i), max_retries=(i % 5) + 1)
        for i in range(25)
    ]
    results = await asyncio.gather(*coros, return_exceptions=True)
    for i, r in enumerate(results):
        if i % 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.async_examples.concurrency import retry_with_backoff

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

@pytest.mark.asyncio
async def test_retry_with_backoff_success_first_try():
    # Test that a successful async function returns its value immediately
    async def successful_func():
        return "success"
    result = await retry_with_backoff(successful_func)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_after_one_failure():
    # Test that a function that fails once then succeeds returns the correct value
    state = {"called": 0}
    async def flaky_func():
        state["called"] += 1
        if state["called"] == 1:
            raise ValueError("fail first time")
        return "ok"
    result = await retry_with_backoff(flaky_func, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_success_after_multiple_failures():
    # Test that a function that fails multiple times then succeeds works
    state = {"called": 0}
    async def flaky_func():
        state["called"] += 1
        if state["called"] < 3:
            raise RuntimeError("fail")
        return "done"
    result = await retry_with_backoff(flaky_func, max_retries=4)

@pytest.mark.asyncio
async def test_retry_with_backoff_raises_after_all_failures():
    # Test that the function raises the last exception after all retries fail
    async def always_fail():
        raise KeyError("always fails")
    with pytest.raises(KeyError) 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 max_retries=1 only calls the function once and raises if it fails
    called = {"count": 0}
    async def fail_once():
        called["count"] += 1
        raise Exception("fail")
    with pytest.raises(Exception):
        await retry_with_backoff(fail_once, max_retries=1)

# ---------------------------
# 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 42
    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_concurrent_success():
    # Test concurrent calls all succeed independently
    async def always_ok():
        return "ok"
    coros = [retry_with_backoff(always_ok) for _ in range(10)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_concurrent_failures():
    # Test concurrent calls all fail independently
    async def always_fail():
        raise Exception("fail")
    coros = [retry_with_backoff(always_fail, max_retries=2) for _ in range(5)]
    for coro in coros:
        with pytest.raises(Exception):
            await coro

@pytest.mark.asyncio
async def test_retry_with_backoff_async_exception_propagation():
    # Test that exceptions from async functions are propagated correctly
    class CustomError(Exception):
        pass
    async def raise_custom():
        raise CustomError("custom error")
    with pytest.raises(CustomError):
        await retry_with_backoff(raise_custom, max_retries=2)

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

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

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_success():
    # Test many concurrent successful executions
    async def always_ok():
        return "ok"
    coros = [retry_with_backoff(always_ok) for _ in range(100)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio

async def test_retry_with_backoff_throughput_small_load():
    # Throughput test with small load
    async def always_ok():
        return 123
    coros = [retry_with_backoff(always_ok) for _ in range(10)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load():
    # Throughput test with medium load
    async def always_ok():
        return "medium"
    coros = [retry_with_backoff(always_ok) for _ in range(100)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume():
    # Throughput test with high volume (but bounded < 1000)
    async def always_ok():
        return "high"
    coros = [retry_with_backoff(always_ok) for _ in range(500)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_mixed_success_failure():
    # Throughput test with mixed success and failure
    async def sometimes_fails(i):
        if i % 5 == 0:
            raise Exception("fail")
        return i
    coros = [retry_with_backoff(lambda i=i: sometimes_fails(i), max_retries=2) for i in range(50)]
    results = []
    for i, coro in enumerate(coros):
        if i % 5 == 0:
            with pytest.raises(Exception):
                await coro
        else:
            val = 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-mfvq8sae and push.