⚡️ Speed up method Lock.statistics by 24%

[codeflash-ai]

📄 24% (0.24x) speedup for Lock.statistics in src/anyio/_backends/_asyncio.py

⏱️ Runtime : 32.5 microseconds → 26.2 microseconds (best of 6 runs)

📝 Explanation and details

The optimization achieves a 24% speedup by eliminating redundant attribute lookups in the statistics() method.

Key Changes:

1. Cached attribute access: Added owner_task = self._owner_task to store the attribute value locally
2. Eliminated method call overhead: Replaced self.locked() call with direct owner_task is not None comparison
3. Reduced attribute lookups: Used the cached owner_task variable instead of accessing self._owner_task multiple times

Why This Works:

• In Python, attribute access (self._owner_task) involves dictionary lookups and method dispatch overhead
• The original code accessed self._owner_task twice: once in the conditional and again when calling self.locked()
• The self.locked() method call added function call overhead for a simple is not None check
• By caching the attribute value and doing the null check inline, we eliminate both the redundant attribute access and the method call

Performance Impact:
The line profiler shows the return statement time dropped from 3.51ms to 1.48ms (58% reduction), while the overall method time improved from 3.86ms to 2.03ms. Test cases consistently show 15-32% speedups across different scenarios, with the optimization being most effective for frequently called code paths where every microsecond matters in async applications.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

from __future__ import annotations

import asyncio
from collections import deque
from typing import TYPE_CHECKING

# imports
import pytest  # used for our unit tests
from anyio._backends._asyncio import Lock


class AsyncIOTaskInfo:
    def __init__(self, task):
        self.task = task

# unit tests

# --- Basic Test Cases ---

def test_statistics_unlocked_no_waiters():
    """
    Test statistics on a new, unlocked Lock with no waiters.
    """
    lock = Lock()
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.40μs -> 1.88μs (27.7% faster)


def test_statistics_unlocked_with_waiters():
    """
    Test statistics on an unlocked Lock with waiters present.
    """
    lock = Lock()
    loop = asyncio.new_event_loop()
    try:
        # Add two waiters
        task1 = loop.create_task(asyncio.sleep(0))
        fut1 = loop.create_future()
        task2 = loop.create_task(asyncio.sleep(0))
        fut2 = loop.create_future()
        lock._waiters.append((task1, fut1))
        lock._waiters.append((task2, fut2))
        codeflash_output = lock.statistics(); stats = codeflash_output
    finally:
        loop.close()


def test_statistics_owner_task_none_waiters_empty():
    """
    Edge case: owner task is None and waiters is empty.
    """
    lock = Lock()
    lock._owner_task = None
    lock._waiters.clear()
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.36μs -> 1.97μs (19.5% faster)

def test_statistics_owner_task_none_waiters_nonempty():
    """
    Edge case: owner task is None but waiters are present.
    """
    lock = Lock()
    loop = asyncio.new_event_loop()
    try:
        t = loop.create_task(asyncio.sleep(0))
        f = loop.create_future()
        lock._waiters.append((t, f))
        lock._owner_task = None
        codeflash_output = lock.statistics(); stats = codeflash_output
    finally:
        loop.close()



def test_statistics_invalid_owner_task_type():
    """
    Edge case: _owner_task is set to an invalid type (not asyncio.Task).
    """
    lock = Lock()
    lock._owner_task = "not_a_task"
    codeflash_output = lock.statistics(); stats = codeflash_output

def test_statistics_waiters_with_invalid_types():
    """
    Edge case: waiters contain invalid types.
    """
    lock = Lock()
    lock._waiters.append(("not_a_task", "not_a_future"))
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.63μs -> 2.05μs (28.1% faster)

# --- Large Scale Test Cases ---

def test_statistics_many_waiters():
    """
    Large scale: Lock with 1000 waiters.
    """
    lock = Lock()
    loop = asyncio.new_event_loop()
    try:
        for _ in range(1000):
            t = loop.create_task(asyncio.sleep(0))
            f = loop.create_future()
            lock._waiters.append((t, f))
        codeflash_output = lock.statistics(); stats = codeflash_output
    finally:
        loop.close()


def test_statistics_performance_many_calls():
    """
    Large scale: Call statistics() 1000 times to check for performance and determinism.
    """
    lock = Lock()
    loop = asyncio.new_event_loop()
    try:
        # Add 10 waiters
        for _ in range(10):
            t = loop.create_task(asyncio.sleep(0))
            f = loop.create_future()
            lock._waiters.append((t, f))
        # Call statistics 1000 times
        for _ in range(1000):
            codeflash_output = lock.statistics(); stats = codeflash_output
    finally:
        loop.close()
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from collections import deque

# imports
import pytest  # used for our unit tests
from anyio._backends._asyncio import Lock


class AsyncIOTaskInfo:
    def __init__(self, task):
        self.task = task

class DummyTask:
    pass

# unit tests

# Basic Test Cases

def test_statistics_unlocked_no_waiters():
    """
    Test statistics when lock is not held and there are no waiters.
    """
    lock = Lock()
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.94μs -> 2.27μs (29.2% faster)


def test_statistics_unlocked_with_waiters():
    """
    Test statistics when lock is not held but there are waiters.
    """
    lock = Lock()
    lock._waiters.append((DummyTask(), object()))
    lock._waiters.append((DummyTask(), object()))
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.53μs -> 2.15μs (18.1% faster)


def test_statistics_owner_task_is_none():
    """
    Test statistics when owner task is explicitly set to None.
    """
    lock = Lock()
    lock._owner_task = None
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.52μs -> 2.17μs (15.9% faster)

def test_statistics_empty_waiters_deque():
    """
    Test statistics when waiters deque is empty.
    """
    lock = Lock()
    lock._waiters = deque()
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.28μs -> 1.87μs (22.2% faster)

def test_statistics_multiple_waiters_varied_types():
    """
    Test statistics with waiters containing varied types (simulate possible real-world misuse).
    """
    lock = Lock()
    lock._waiters.append(('not_a_task', 'not_a_future'))
    lock._waiters.append((DummyTask(), object()))
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.29μs -> 1.73μs (32.3% faster)

def test_statistics_waiters_maximum():
    """
    Test statistics when waiters deque is at maximum allowed size (999).
    """
    lock = Lock()
    for _ in range(999):
        lock._waiters.append((DummyTask(), object()))
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.43μs -> 1.99μs (21.8% faster)

def test_statistics_owner_task_is_falsy_object():
    """
    Test statistics when owner_task is a falsy object (e.g., empty string).
    """
    lock = Lock()
    lock._owner_task = ""  # Falsy but not None
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.44μs -> 1.95μs (25.0% faster)

# Large Scale Test Cases

def test_statistics_large_number_of_waiters():
    """
    Test statistics with a large number of waiters (1000).
    """
    lock = Lock()
    for _ in range(1000):
        lock._waiters.append((DummyTask(), object()))
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.49μs -> 1.97μs (26.1% faster)

def test_statistics_large_number_of_waiters_and_locked():
    """
    Test statistics with a large number of waiters and the lock held.
    """
    lock = Lock()
    lock._owner_task = DummyTask()
    for _ in range(1000):
        lock._waiters.append((DummyTask(), object()))
    codeflash_output = lock.statistics(); stats = codeflash_output

def test_statistics_performance_large_scale():
    """
    Performance test: statistics should execute quickly for large waiters deque.
    """
    import time
    lock = Lock()
    for _ in range(1000):
        lock._waiters.append((DummyTask(), object()))
    start = time.time()
    codeflash_output = lock.statistics(); stats = codeflash_output # 2.79μs -> 2.28μs (22.5% faster)
    duration = time.time() - start
# 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 Lock

def test_Lock_statistics():
    Lock.statistics(Lock())

🔎 Concolic Coverage Tests and Runtime

Test File::Test Function	Original ⏱️	Optimized ⏱️	Speedup
codeflash_concolic_ce_a9iww/tmptsx7ijih/test_concolic_coverage.py::test_Lock_statistics	2.45μs	1.92μs	27.6%✅

To edit these changes git checkout codeflash/optimize-Lock.statistics-mhl6z6pt and push.