⚡️ Speed up function is_anyio_cancellation by 25%

[codeflash-ai]

📄 25% (0.25x) speedup for is_anyio_cancellation in src/anyio/_backends/_asyncio.py

⏱️ Runtime : 553 microseconds → 443 microseconds (best of 74 runs)

📝 Explanation and details

The optimized code achieves a 24% speedup by eliminating redundant attribute lookups within the while loop.

Key optimizations:

1. Reduced attribute access: The original code repeatedly accessed exc.args, exc.args[0], and exc.__context__ multiple times per iteration. The optimized version stores these in local variables (args, context) to avoid repeated property lookups.

2. Single variable tracking: Instead of reassigning the exc parameter directly, the optimized code uses a separate current variable to track the exception being examined, making the control flow clearer.

3. Streamlined condition logic: The nested if-statement structure is flattened into more direct checks, reducing the overhead of multiple condition evaluations.

Why this matters for Python performance:

• Attribute access (obj.attr) involves dictionary lookups in Python, which are relatively expensive when done repeatedly
• Local variable access is significantly faster than attribute access
• The optimizations are most effective for cases with long exception chains, as shown in the test results where large chain tests show 25-30% speedups

Performance characteristics:

• Best case: Long exception chains (500+ elements) see the largest gains (25-30% faster)
• Moderate case: Short chains (2-3 elements) show modest improvements (1-10% faster)
• Neutral/slight regression: Very simple cases with no chains may be 1-8% slower due to the overhead of extra variable assignments, but this is offset by the dramatic gains in complex cases

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

from asyncio import CancelledError

# imports
import pytest
from anyio._backends._asyncio import is_anyio_cancellation

# unit tests

# --- Basic Test Cases ---

def test_basic_true_direct_message():
    # Direct CancelledError with correct message
    exc = CancelledError("Cancelled via cancel scope 123")
    codeflash_output = is_anyio_cancellation(exc) # 1.24μs -> 1.21μs (1.89% faster)

def test_basic_false_different_message():
    # Direct CancelledError with incorrect message
    exc = CancelledError("Some other cancellation reason")
    codeflash_output = is_anyio_cancellation(exc) # 1.44μs -> 1.49μs (3.36% slower)

def test_basic_false_no_args():
    # CancelledError with no args
    exc = CancelledError()
    codeflash_output = is_anyio_cancellation(exc) # 978ns -> 1.06μs (7.91% slower)

def test_basic_false_non_string_arg():
    # CancelledError with non-string arg
    exc = CancelledError(12345)
    codeflash_output = is_anyio_cancellation(exc) # 1.41μs -> 1.45μs (2.35% slower)

def test_basic_true_context_message():
    # CancelledError with correct message in __context__
    context_exc = CancelledError("Cancelled via cancel scope 456")
    exc = CancelledError("Other message")
    exc.__context__ = context_exc
    codeflash_output = is_anyio_cancellation(exc) # 1.82μs -> 1.85μs (1.62% slower)

def test_basic_false_context_wrong_message():
    # CancelledError with incorrect message in __context__
    context_exc = CancelledError("Not the right message")
    exc = CancelledError("Other message")
    exc.__context__ = context_exc
    codeflash_output = is_anyio_cancellation(exc) # 1.88μs -> 1.92μs (2.08% slower)

# --- Edge Test Cases ---

def test_edge_context_chain_true_deep():
    # Deep chain where only the last CancelledError has the correct message
    exc1 = CancelledError("Wrong message")
    exc2 = CancelledError("Another wrong message")
    exc3 = CancelledError("Cancelled via cancel scope 999")
    exc2.__context__ = exc3
    exc1.__context__ = exc2
    codeflash_output = is_anyio_cancellation(exc1) # 2.08μs -> 2.01μs (3.84% faster)

def test_edge_context_chain_false_all_wrong():
    # Deep chain where none have the correct message
    exc1 = CancelledError("Wrong message")
    exc2 = CancelledError("Another wrong message")
    exc3 = CancelledError("Still wrong")
    exc2.__context__ = exc3
    exc1.__context__ = exc2
    codeflash_output = is_anyio_cancellation(exc1) # 2.08μs -> 2.14μs (2.85% slower)

def test_edge_context_non_cancelled_error():
    # __context__ is not a CancelledError
    class DummyException(Exception): pass
    dummy = DummyException("irrelevant")
    exc = CancelledError("Some message")
    exc.__context__ = dummy
    codeflash_output = is_anyio_cancellation(exc) # 1.70μs -> 1.71μs (0.528% slower)

def test_edge_context_none():
    # __context__ is None
    exc = CancelledError("Cancelled via cancel scope 42")
    exc.__context__ = None
    codeflash_output = is_anyio_cancellation(exc) # 1.12μs -> 1.18μs (4.58% slower)

def test_edge_args_empty_tuple():
    # args is an empty tuple
    exc = CancelledError()
    exc.args = ()
    codeflash_output = is_anyio_cancellation(exc) # 1.04μs -> 1.08μs (3.88% slower)

def test_edge_args_non_tuple():
    # args is not a tuple (shouldn't happen, but test robustness)
    exc = CancelledError("Cancelled via cancel scope 123")
    exc.args = "Cancelled via cancel scope 123"
    # This will fail the isinstance(exc.args[0], str) check, as exc.args[0] is 'a'
    codeflash_output = is_anyio_cancellation(exc) # 1.44μs -> 1.41μs (2.63% faster)

def test_edge_args_tuple_non_str_first_element():
    # args is a tuple but first element is not a string
    exc = CancelledError(123, "Cancelled via cancel scope 123")
    codeflash_output = is_anyio_cancellation(exc) # 1.44μs -> 1.43μs (0.980% faster)

def test_edge_args_tuple_str_second_element():
    # args is a tuple, first element is not string, second element is correct string
    exc = CancelledError(123, "Cancelled via cancel scope 123")
    codeflash_output = is_anyio_cancellation(exc) # 1.45μs -> 1.41μs (2.62% faster)

def test_edge_cancelled_error_subclass():
    # Subclass of CancelledError with correct message
    class MyCancelledError(CancelledError): pass
    exc = MyCancelledError("Cancelled via cancel scope 888")
    codeflash_output = is_anyio_cancellation(exc) # 1.41μs -> 1.31μs (7.94% faster)

def test_edge_cancelled_error_subclass_context_chain():
    # Subclass chain, correct message deep in context
    class MyCancelledError(CancelledError): pass
    exc1 = MyCancelledError("Wrong message")
    exc2 = MyCancelledError("Cancelled via cancel scope 777")
    exc1.__context__ = exc2
    codeflash_output = is_anyio_cancellation(exc1) # 1.96μs -> 2.02μs (2.97% slower)

# --- Large Scale Test Cases ---

def test_large_chain_first_has_message():
    # Large chain, first CancelledError has correct message
    chain_length = 500
    exc_list = [CancelledError("Cancelled via cancel scope 0")]
    for i in range(1, chain_length):
        exc = CancelledError(f"Wrong message {i}")
        exc.__context__ = exc_list[-1]
        exc_list.append(exc)
    # exc_list[-1] is the head of the chain
    codeflash_output = is_anyio_cancellation(exc_list[-1]) # 81.0μs -> 62.9μs (28.9% faster)

def test_large_chain_last_has_message():
    # Large chain, last CancelledError has correct message
    chain_length = 500
    exc_list = [CancelledError("Cancelled via cancel scope 999")]
    for i in range(1, chain_length):
        exc = CancelledError(f"Wrong message {i}")
        exc.__context__ = exc_list[-1]
        exc_list.append(exc)
    # exc_list[-1] is the head of the chain
    codeflash_output = is_anyio_cancellation(exc_list[-1]) # 80.8μs -> 62.8μs (28.6% faster)

def test_large_chain_none_have_message():
    # Large chain, none have correct message
    chain_length = 500
    exc_list = [CancelledError("Wrong message 0")]
    for i in range(1, chain_length):
        exc = CancelledError(f"Wrong message {i}")
        exc.__context__ = exc_list[-1]
        exc_list.append(exc)
    # exc_list[-1] is the head of the chain
    codeflash_output = is_anyio_cancellation(exc_list[-1]) # 80.8μs -> 62.7μs (29.0% faster)

def test_large_chain_mixed_types():
    # Large chain with some non-CancelledError contexts
    chain_length = 250
    exc_list = [CancelledError("Cancelled via cancel scope 42")]
    # Alternate between CancelledError and Exception
    for i in range(1, chain_length):
        if i % 2 == 0:
            exc = CancelledError(f"Wrong message {i}")
        else:
            exc = Exception(f"Not a CancelledError {i}")
        exc.__context__ = exc_list[-1]
        exc_list.append(exc)
    # exc_list[-1] is the head of the chain, but it's an Exception, not CancelledError
    # So we need to start with a CancelledError at the top
    top_exc = CancelledError("Wrong message at top")
    top_exc.__context__ = exc_list[-1]
    codeflash_output = is_anyio_cancellation(top_exc) # 1.45μs -> 1.52μs (4.35% slower)

def test_large_chain_all_subclasses():
    # Large chain of CancelledError subclasses
    class MyCancelledError(CancelledError): pass
    chain_length = 300
    exc_list = [MyCancelledError("Cancelled via cancel scope subclass")]
    for i in range(1, chain_length):
        exc = MyCancelledError(f"Wrong message {i}")
        exc.__context__ = exc_list[-1]
        exc_list.append(exc)
    codeflash_output = is_anyio_cancellation(exc_list[-1]) # 50.7μs -> 39.7μs (27.8% faster)

def test_large_chain_context_none_at_end():
    # Large chain ending with __context__ = None
    chain_length = 400
    exc_list = [CancelledError("Cancelled via cancel scope at end")]
    for i in range(1, chain_length):
        exc = CancelledError(f"Wrong message {i}")
        exc.__context__ = exc_list[-1]
        exc_list.append(exc)
    # Set __context__ of last one to None
    exc_list[-1].__context__ = None
    codeflash_output = is_anyio_cancellation(exc_list[-1]) # 1.37μs -> 1.39μs (1.15% slower)

# --- Determinism and Robustness ---

def test_determinism_multiple_calls():
    # Ensure multiple calls with same input produce same output
    exc = CancelledError("Cancelled via cancel scope 123")
    for _ in range(10):
        codeflash_output = is_anyio_cancellation(exc) # 4.09μs -> 3.89μs (5.27% faster)



def test_robustness_cancelled_error_with_empty_string_arg():
    # args is a tuple with empty string
    exc = CancelledError("")
    codeflash_output = is_anyio_cancellation(exc) # 1.65μs -> 1.58μs (4.25% faster)

def test_robustness_cancelled_error_with_partial_message():
    # args is a tuple with a string that partially matches
    exc = CancelledError("Cancelled via cancel scope")
    # Should match, as it startswith "Cancelled via cancel scope "
    codeflash_output = is_anyio_cancellation(exc) # 1.45μs -> 1.38μs (4.77% faster)

def test_robustness_cancelled_error_with_whitespace_message():
    # args is a tuple with a string that has whitespace
    exc = CancelledError("Cancelled via cancel scope  ")
    codeflash_output = is_anyio_cancellation(exc) # 1.19μs -> 1.14μs (4.21% faster)

def test_robustness_cancelled_error_with_long_message():
    # args is a tuple with a long correct message
    exc = CancelledError("Cancelled via cancel scope " + "x" * 500)
    codeflash_output = is_anyio_cancellation(exc) # 1.20μs -> 1.17μs (2.21% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from asyncio import CancelledError

# imports
import pytest
from anyio._backends._asyncio import is_anyio_cancellation

# unit tests

# 1. Basic Test Cases

def test_basic_true_direct_message():
    # Direct CancelledError with matching message
    exc = CancelledError("Cancelled via cancel scope 123")
    codeflash_output = is_anyio_cancellation(exc) # 1.22μs -> 1.11μs (9.33% faster)

def test_basic_false_different_message():
    # Direct CancelledError with non-matching message
    exc = CancelledError("Some other cancellation reason")
    codeflash_output = is_anyio_cancellation(exc) # 1.51μs -> 1.36μs (11.2% faster)

def test_basic_false_no_message():
    # Direct CancelledError with no message
    exc = CancelledError()
    codeflash_output = is_anyio_cancellation(exc) # 1.04μs -> 1.10μs (5.37% slower)

def test_basic_false_non_string_arg():
    # Direct CancelledError with non-string first arg
    exc = CancelledError(123)
    codeflash_output = is_anyio_cancellation(exc) # 1.40μs -> 1.44μs (2.71% slower)

def test_basic_true_context_message():
    # CancelledError with context containing matching message
    context_exc = CancelledError("Cancelled via cancel scope 456")
    exc = CancelledError("Outer error")
    exc.__context__ = context_exc
    codeflash_output = is_anyio_cancellation(exc) # 1.83μs -> 1.85μs (1.13% slower)

def test_basic_false_context_non_matching_message():
    # CancelledError with context containing non-matching message
    context_exc = CancelledError("Some other reason")
    exc = CancelledError("Outer error")
    exc.__context__ = context_exc
    codeflash_output = is_anyio_cancellation(exc) # 1.82μs -> 1.95μs (6.55% slower)

# 2. Edge Test Cases

def test_edge_multiple_context_chain_true():
    # Chain of CancelledErrors, only one matches
    exc1 = CancelledError("Non-matching")
    exc2 = CancelledError("Also non-matching")
    exc3 = CancelledError("Cancelled via cancel scope 789")
    exc2.__context__ = exc3
    exc1.__context__ = exc2
    codeflash_output = is_anyio_cancellation(exc1) # 1.90μs -> 1.95μs (2.61% slower)

def test_edge_multiple_context_chain_false():
    # Chain of CancelledErrors, none matches
    exc1 = CancelledError("First")
    exc2 = CancelledError("Second")
    exc3 = CancelledError("Third")
    exc2.__context__ = exc3
    exc1.__context__ = exc2
    codeflash_output = is_anyio_cancellation(exc1) # 1.91μs -> 2.08μs (8.00% slower)

def test_edge_context_not_cancelled_error():
    # __context__ is not a CancelledError
    class Dummy(Exception):
        pass
    exc = CancelledError("Non-matching")
    exc.__context__ = Dummy("Not a CancelledError")
    codeflash_output = is_anyio_cancellation(exc) # 1.63μs -> 1.70μs (4.12% slower)

def test_edge_args_tuple_with_matching_first():
    # args is a tuple, first element matches
    exc = CancelledError("Cancelled via cancel scope 999", "extra")
    codeflash_output = is_anyio_cancellation(exc) # 1.17μs -> 1.20μs (2.75% slower)

def test_edge_args_tuple_with_non_string_first():
    # args is a tuple, first element is not string
    exc = CancelledError(42, "Cancelled via cancel scope 1000")
    codeflash_output = is_anyio_cancellation(exc) # 1.43μs -> 1.45μs (1.31% slower)

def test_edge_empty_args():
    # args is empty tuple
    exc = CancelledError()
    exc.args = ()
    codeflash_output = is_anyio_cancellation(exc) # 1.01μs -> 1.11μs (8.31% slower)


def test_edge_context_cycle():
    # __context__ chain forms a cycle, should not infinite loop
    exc1 = CancelledError("Non-matching")
    exc2 = CancelledError("Cancelled via cancel scope 202")
    exc1.__context__ = exc2
    exc2.__context__ = exc1  # cycle
    # Should still find the matching message and not infinite loop
    codeflash_output = is_anyio_cancellation(exc1) # 1.91μs -> 1.95μs (2.10% slower)

def test_edge_context_chain_with_non_cancelled_error_middle():
    # __context__ chain with a non-CancelledError in the middle
    class Dummy(Exception): pass
    exc1 = CancelledError("Non-matching")
    exc2 = Dummy("dummy")
    exc3 = CancelledError("Cancelled via cancel scope 303")
    exc1.__context__ = exc2
    exc2.__context__ = exc3
    # Should not traverse past exc2 since it's not CancelledError
    codeflash_output = is_anyio_cancellation(exc1) # 1.58μs -> 1.64μs (3.78% slower)

# 3. Large Scale Test Cases

def test_large_chain_with_match_at_end():
    # Large chain of CancelledErrors, match at the last one
    chain_length = 500
    excs = [CancelledError(f"Non-matching {i}") for i in range(chain_length)]
    excs[-1] = CancelledError("Cancelled via cancel scope 404")
    # Link the chain
    for i in range(chain_length - 1):
        excs[i].__context__ = excs[i + 1]
    # Should find the match at the end
    codeflash_output = is_anyio_cancellation(excs[0]) # 81.4μs -> 62.6μs (30.1% faster)

def test_large_chain_no_match():
    # Large chain of CancelledErrors, none matches
    chain_length = 500
    excs = [CancelledError(f"Non-matching {i}") for i in range(chain_length)]
    for i in range(chain_length - 1):
        excs[i].__context__ = excs[i + 1]
    codeflash_output = is_anyio_cancellation(excs[0]) # 81.6μs -> 62.9μs (29.8% faster)

def test_large_chain_with_intermediate_non_cancelled_error():
    # Large chain, with a non-CancelledError in the middle
    chain_length = 250
    excs = [CancelledError(f"Non-matching {i}") for i in range(chain_length)]
    excs[100].__context__ = Exception("dummy")
    for i in range(100):
        excs[i].__context__ = excs[i + 1]
    # Should not traverse past excs[100]
    codeflash_output = is_anyio_cancellation(excs[0]) # 17.9μs -> 14.2μs (25.5% faster)

def test_large_chain_with_cycle():
    # Large chain with a cycle, match at the cycle point
    chain_length = 100
    excs = [CancelledError(f"Non-matching {i}") for i in range(chain_length)]
    excs[-1] = CancelledError("Cancelled via cancel scope 505")
    for i in range(chain_length - 1):
        excs[i].__context__ = excs[i + 1]
    # Create a cycle
    excs[-1].__context__ = excs[0]
    # Should still find the match and not infinite loop
    codeflash_output = is_anyio_cancellation(excs[0]) # 17.7μs -> 14.1μs (25.0% faster)

def test_large_chain_with_multiple_matches():
    # Large chain with multiple matching CancelledErrors
    chain_length = 300
    excs = [CancelledError("Cancelled via cancel scope {}".format(i) if i % 100 == 0 else "Non-matching {}".format(i)) for i in range(chain_length)]
    for i in range(chain_length - 1):
        excs[i].__context__ = excs[i + 1]
    # Should find the first match in the chain
    codeflash_output = is_anyio_cancellation(excs[0]) # 1.25μs -> 1.13μs (10.4% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from anyio._backends._asyncio import is_anyio_cancellation
from asyncio.exceptions import CancelledError

def test_is_anyio_cancellation():
    is_anyio_cancellation(CancelledError(''))

🔎 Concolic Coverage Tests and Runtime

Test File::Test Function	Original ⏱️	Optimized ⏱️	Speedup
codeflash_concolic_tgs86_zb/tmpvh8ytz_g/test_concolic_coverage.py::test_is_anyio_cancellation	1.47μs	1.49μs	-1.47%⚠️

To edit these changes git checkout codeflash/optimize-is_anyio_cancellation-mhfnn0fs and push.