⚡️ Speed up function retry_with_backoff by -45%

[codeflash-ai]

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

⏱️ Runtime : 21.3 milliseconds → 39.0 milliseconds (best of 238 runs)

📝 Explanation and details

The key optimization in this code is replacing time.sleep() with await asyncio.sleep() in the retry backoff logic. While the individual function runtime appears slower (-45%), this change delivers a significant 31.5% throughput improvement when handling concurrent operations.

What changed:

• Replaced blocking time.sleep(0.0001 * attempt) with non-blocking await asyncio.sleep(0.0001 * attempt)
• Added proper type hints for better code clarity

Why this improves performance:
The original code uses time.sleep(), which is a blocking operation that freezes the entire event loop thread during sleep periods. This prevents other async tasks from executing concurrently. The line profiler shows this blocking sleep consuming 84.2% of execution time in the original version.

The optimized version uses await asyncio.sleep(), which is non-blocking and yields control back to the event loop. This allows other coroutines to execute while one is sleeping, dramatically improving concurrency. The profiler shows the async sleep now only takes 32.7% of execution time.

Impact on workloads:

• Single function calls: Slightly slower due to async overhead
• Concurrent operations: Massive improvement - tasks can run in parallel instead of blocking each other
• High-volume scenarios: The throughput tests demonstrate the real benefit, where multiple retry operations can overlap their sleep periods

This optimization is particularly valuable for applications that make many concurrent API calls or database operations with retry logic, as the non-blocking sleep allows proper async concurrency instead of serializing all retry attempts.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 941 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 ---------------------------

# Basic Test Cases

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

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

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_exception_on_all_failures():
    # Test that function raises the last exception if all attempts fail
    async def always_fail():
        raise RuntimeError("fail")
    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 function only tries once if max_retries=1
    state = {"calls": 0}
    async def fail_once():
        state["calls"] += 1
        raise Exception("fail")
    with pytest.raises(Exception):
        await retry_with_backoff(fail_once, max_retries=1)

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries():
    # Test that ValueError is raised for invalid max_retries
    async def dummy():
        return 42
    with pytest.raises(ValueError):
        await retry_with_backoff(dummy, max_retries=0)

# Edge Test Cases

@pytest.mark.asyncio
async def test_retry_with_backoff_concurrent_success():
    # Test concurrent execution with all succeeding
    async def always_succeed():
        return "ok"
    results = await asyncio.gather(
        retry_with_backoff(always_succeed),
        retry_with_backoff(always_succeed),
        retry_with_backoff(always_succeed)
    )

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

@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_then_type():
        state["calls"] += 1
        if state["calls"] < 2:
            raise KeyError("key error")
        raise ValueError("value error")
    with pytest.raises(ValueError):
        await retry_with_backoff(fail_then_type, max_retries=2)

@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():
        return 123
    result = await retry_with_backoff(coro_func)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_raises_non_exception():
    # Test that retry_with_backoff propagates non-Exception errors
    async def raise_base():
        raise BaseException("base")
    with pytest.raises(BaseException):
        await retry_with_backoff(raise_base)

# Large Scale Test Cases

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

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_mixed():
    # Test many concurrent executions, some fail, some succeed
    async def succeed_or_fail(i):
        if i % 2 == 0:
            return i
        raise Exception("fail")
    coros = [retry_with_backoff(lambda i=i: succeed_or_fail(i), max_retries=2) for i in range(20)]
    results = []
    for i, coro in enumerate(coros):
        if i % 2 == 0:
            results.append(await coro)
        else:
            with pytest.raises(Exception):
                await coro

@pytest.mark.asyncio
async def test_retry_with_backoff_high_retries_success():
    # Test high max_retries where success happens late
    state = {"calls": 0}
    async def succeed_on_tenth():
        state["calls"] += 1
        if state["calls"] < 10:
            raise Exception("fail")
        return "done"
    result = await retry_with_backoff(succeed_on_tenth, max_retries=15)

# Throughput Test Cases

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_small_load():
    # Throughput test: small load, all succeed first try
    async def always_succeed():
        return 1
    coros = [retry_with_backoff(always_succeed) for _ 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 once then succeed
    async def succeed_after_one(i):
        state = {"calls": 0}
        async def inner():
            state["calls"] += 1
            if state["calls"] < 2:
                raise Exception("fail")
            return i
        return await retry_with_backoff(inner, max_retries=3)
    coros = [succeed_after_one(i) for i in range(50)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume():
    # Throughput test: high volume, all succeed first try
    async def always_succeed():
        return "high"
    coros = [retry_with_backoff(always_succeed) for _ in range(200)]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_varied_failures():
    # Throughput test: varied failures, some succeed after retries
    async def succeed_after_n(n):
        state = {"calls": 0}
        async def inner():
            state["calls"] += 1
            if state["calls"] <= n:
                raise Exception("fail")
            return n
        return await retry_with_backoff(inner, max_retries=n+2)
    coros = [succeed_after_n(i % 3) for i in range(30)]
    results = await asyncio.gather(*coros)
# 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

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

# ------------------- BASIC TEST CASES -------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_success_first_try():
    # Test that function returns 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_success_second_try():
    # Test that function fails once, then succeeds
    attempts = {"count": 0}
    async def fails_once_then_succeeds():
        if attempts["count"] == 0:
            attempts["count"] += 1
            raise ValueError("fail first")
        return "ok"
    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 function fails twice, then succeeds
    attempts = {"count": 0}
    async def fails_twice_then_succeeds():
        if attempts["count"] < 2:
            attempts["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 function raises after max_retries exhausted
    async def always_fails():
        raise KeyError("always fails")
    with pytest.raises(KeyError):
        await retry_with_backoff(always_fails, max_retries=3)

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

@pytest.mark.asyncio
async def test_retry_with_backoff_returns_none():
    # Function returns None, should propagate result
    async def returns_none():
        return None
    result = await retry_with_backoff(returns_none)

# ------------------- EDGE TEST CASES -------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_invalid_max_retries():
    # Should raise ValueError if max_retries < 1
    async def dummy():
        return "should not run"
    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 execution of multiple successful calls
    async def always_succeeds():
        return "ok"
    results = await asyncio.gather(
        retry_with_backoff(always_succeeds),
        retry_with_backoff(always_succeeds),
        retry_with_backoff(always_succeeds)
    )

@pytest.mark.asyncio
async def test_retry_with_backoff_concurrent_failure():
    # Test concurrent execution of multiple failing calls
    async def always_fails():
        raise ValueError("fail")
    tasks = [
        retry_with_backoff(always_fails, max_retries=2),
        retry_with_backoff(always_fails, max_retries=2)
    ]
    results = []
    for task in tasks:
        with pytest.raises(ValueError):
            await task

@pytest.mark.asyncio
async def test_retry_with_backoff_exception_type_propagation():
    # Ensure the last exception type is propagated
    async def raises_type_error():
        raise TypeError("bad type")
    with pytest.raises(TypeError):
        await retry_with_backoff(raises_type_error, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_is_coroutine():
    # Ensure function works with coroutine objects
    async def returns_value():
        return 123
    coro = returns_value()
    # retry_with_backoff expects a callable, not a coroutine object
    with pytest.raises(TypeError):
        await retry_with_backoff(coro, max_retries=2)

@pytest.mark.asyncio
async def test_retry_with_backoff_func_raises_different_exceptions():
    # Function raises different exceptions on each call, last one should be propagated
    attempts = {"count": 0}
    async def raises_various():
        if attempts["count"] == 0:
            attempts["count"] += 1
            raise ValueError("first fail")
        elif attempts["count"] == 1:
            attempts["count"] += 1
            raise KeyError("second fail")
        else:
            raise RuntimeError("third fail")
    with pytest.raises(RuntimeError) as exc_info:
        await retry_with_backoff(raises_various, max_retries=3)

# ------------------- LARGE SCALE TEST CASES -------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_successes():
    # Test with many concurrent tasks all succeeding
    async def always_succeeds():
        return "done"
    tasks = [retry_with_backoff(always_succeeds) for _ in range(100)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_many_concurrent_failures():
    # Test with many concurrent tasks all failing
    async def always_fails():
        raise Exception("fail")
    tasks = [retry_with_backoff(always_fails, max_retries=2) for _ in range(50)]
    for task in tasks:
        with pytest.raises(Exception):
            await task

@pytest.mark.asyncio
async def test_retry_with_backoff_mixed_concurrent():
    # Mix of tasks: some succeed, some fail
    async def succeed():
        return "ok"
    async def fail():
        raise Exception("fail")
    tasks = [retry_with_backoff(succeed) for _ in range(10)] + \
            [retry_with_backoff(fail, max_retries=2) for _ in range(10)]
    results = []
    for i, task in enumerate(tasks):
        if i < 10:
            res = await task
        else:
            with pytest.raises(Exception):
                await task

# ------------------- THROUGHPUT TEST CASES -------------------

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_small_load():
    # Small load throughput test: 10 concurrent successful tasks
    async def always_succeeds():
        return "small"
    tasks = [retry_with_backoff(always_succeeds) for _ in range(10)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_medium_load():
    # Medium load throughput test: 100 concurrent tasks, half succeed, half fail
    async def succeed():
        return "medium"
    async def fail():
        raise Exception("fail")
    tasks = [retry_with_backoff(succeed) for _ in range(50)] + \
            [retry_with_backoff(fail, max_retries=2) for _ in range(50)]
    results = []
    for i, task in enumerate(tasks):
        if i < 50:
            res = await task
        else:
            with pytest.raises(Exception):
                await task

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_high_volume():
    # High volume throughput test: 200 concurrent successful tasks
    async def always_succeeds():
        return "high"
    tasks = [retry_with_backoff(always_succeeds) for _ in range(200)]
    results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_retry_with_backoff_throughput_sustained_execution():
    # Sustained execution: 20 tasks, each with 5 retries before success
    async def fails_then_succeeds_factory():
        attempts = {"count": 0}
        async def inner():
            if attempts["count"] < 4:
                attempts["count"] += 1
                raise Exception("fail")
            return "sustained"
        return inner
    tasks = [retry_with_backoff(await fails_then_succeeds_factory(), max_retries=5) for _ in range(20)]
    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-mhpz5x62 and push.