⚡️ Speed up function fix_asyncio_event_loop_policy by 284%

[codeflash-ai]

📄 284% (2.84x) speedup for fix_asyncio_event_loop_policy in modules/initialize_util.py

⏱️ Runtime : 629 microseconds → 164 microseconds (best of 10 runs)

📝 Explanation and details

The optimization achieves a 284% speedup by eliminating redundant work on repeated function calls through intelligent caching:

Key optimizations:

1. Cached base policy resolution: Instead of checking sys.platform and resolving the base policy class on every call, it's cached as a function attribute (_BasePolicy) after the first evaluation. The profiler shows this reduces the platform check from 37 hits to just 1.

2. Singleton class definition: The most expensive operation - creating the AnyThreadEventLoopPolicy class - is now done only once and cached as _AnyThreadEventLoopPolicy. The profiler shows class creation time dropped from 565,686 ns (64.3% of runtime) to just 22,700 ns (6.4%), executing only once instead of 37 times.

3. Attribute-based caching: Uses hasattr() checks on the function itself to determine if cached values exist, avoiding global variables while maintaining thread safety.

Why this works:

• Class definition overhead: Python class creation involves significant bytecode compilation and namespace setup. Doing this 37 times vs. once explains the dramatic speedup.
• Conditional evaluation: Platform checks and attribute lookups are expensive when repeated unnecessarily.
• Memory efficiency: Reusing the same class object reduces memory allocation overhead.

Test case performance:
The optimization is particularly effective for scenarios involving multiple calls (like the "multiple calls" test showing 448% speedup on the second call), making it ideal for applications that frequently reconfigure asyncio policies or run in multi-threaded environments where this function might be called repeatedly.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio
import sys
import threading
import time

# imports
import pytest  # used for our unit tests
from modules.initialize_util import fix_asyncio_event_loop_policy

# unit tests

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

def test_main_thread_event_loop_policy_basic():
    """
    Basic: After applying fix_asyncio_event_loop_policy, the main thread should still be able to get an event loop.
    """
    fix_asyncio_event_loop_policy() # 24.4μs -> 7.00μs (248% faster)
    loop = asyncio.get_event_loop()
    # Should be able to run a simple coroutine
    result = loop.run_until_complete(asyncio.sleep(0, result=42))

def test_main_thread_event_loop_policy_new_loop():
    """
    Basic: Should be able to create a new event loop and set it as current.
    """
    fix_asyncio_event_loop_policy() # 20.8μs -> 4.87μs (327% faster)
    old_loop = asyncio.get_event_loop()
    new_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(new_loop)
    try:
        result = new_loop.run_until_complete(asyncio.sleep(0, result=99))
    finally:
        # Restore old loop to avoid side effects
        asyncio.set_event_loop(old_loop)

# ---------------------- Edge Test Cases ----------------------

def test_thread_event_loop_creation():
    """
    Edge: In a new thread, asyncio.get_event_loop should not raise, and should create a new loop.
    """
    fix_asyncio_event_loop_policy() # 21.0μs -> 5.03μs (318% faster)
    results = {}
    def thread_target():
        try:
            loop = asyncio.get_event_loop()
            results['loop_type'] = type(loop)
            # Should be able to run a coroutine in this thread's loop
            result = loop.run_until_complete(asyncio.sleep(0, result=123))
            results['value'] = result
        except Exception as e:
            results['error'] = str(e)
    t = threading.Thread(target=thread_target)
    t.start()
    t.join()

def test_thread_multiple_event_loops():
    """
    Edge: Multiple threads should each get their own event loop, and not interfere.
    """
    fix_asyncio_event_loop_policy() # 20.8μs -> 5.34μs (290% faster)
    loops = []
    values = []
    errors = []
    def thread_target(idx):
        try:
            loop = asyncio.get_event_loop()
            loops.append(loop)
            val = loop.run_until_complete(asyncio.sleep(0, result=idx))
            values.append(val)
        except Exception as e:
            errors.append(str(e))
    threads = [threading.Thread(target=thread_target, args=(i,)) for i in range(3)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_thread_event_loop_reuse():
    """
    Edge: If set_event_loop is called in a thread, get_event_loop should reuse it.
    """
    fix_asyncio_event_loop_policy() # 23.1μs -> 6.25μs (269% faster)
    results = {}
    def thread_target():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        got_loop = asyncio.get_event_loop()
        results['same'] = (loop is got_loop)
    t = threading.Thread(target=thread_target)
    t.start()
    t.join()

def test_thread_event_loop_policy_restoration():
    """
    Edge: After fix_asyncio_event_loop_policy, restoring the default policy disables auto-creation in threads.
    """
    fix_asyncio_event_loop_policy() # 22.1μs -> 5.58μs (296% faster)
    # Save the current policy
    current_policy = asyncio.get_event_loop_policy()
    # Restore default policy
    if sys.platform == "win32" and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"):
        default_policy = asyncio.WindowsSelectorEventLoopPolicy()
    else:
        default_policy = asyncio.DefaultEventLoopPolicy()
    asyncio.set_event_loop_policy(default_policy)
    error = []
    def thread_target():
        try:
            asyncio.get_event_loop()
        except Exception as e:
            error.append(type(e))
    t = threading.Thread(target=thread_target)
    t.start()
    t.join()

def test_event_loop_policy_multiple_calls():
    """
    Edge: Calling fix_asyncio_event_loop_policy multiple times should not error, and policy should remain correct.
    """
    fix_asyncio_event_loop_policy() # 20.1μs -> 4.35μs (361% faster)
    fix_asyncio_event_loop_policy() # 10.2μs -> 1.86μs (448% faster)
    # Should still work in thread
    result = {}
    def thread_target():
        loop = asyncio.get_event_loop()
        result['type'] = type(loop)
    t = threading.Thread(target=thread_target)
    t.start()
    t.join()

def test_thread_event_loop_policy_with_exception():
    """
    Edge: Simulate an AssertionError in get_event_loop and ensure fallback works.
    """
    fix_asyncio_event_loop_policy() # 17.7μs -> 5.22μs (239% faster)
    # Patch the policy to forcibly raise AssertionError
    class DummyPolicy(asyncio.DefaultEventLoopPolicy):
        def get_event_loop(self):
            raise AssertionError("Dummy error")
    asyncio.set_event_loop_policy(DummyPolicy())
    # Now call fix_asyncio_event_loop_policy again to restore correct behavior
    fix_asyncio_event_loop_policy() # 8.10μs -> 1.79μs (353% faster)
    result = {}
    def thread_target():
        loop = asyncio.get_event_loop()
        result['loop'] = loop
    t = threading.Thread(target=thread_target)
    t.start()
    t.join()

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

def test_many_threads_event_loops():
    """
    Large Scale: Create up to 50 threads, each should get its own event loop and run a coroutine.
    """
    fix_asyncio_event_loop_policy() # 17.4μs -> 4.58μs (279% faster)
    N = 50
    loops = []
    values = []
    errors = []
    def thread_target(idx):
        try:
            loop = asyncio.get_event_loop()
            loops.append(loop)
            val = loop.run_until_complete(asyncio.sleep(0, result=idx))
            values.append(val)
        except Exception as e:
            errors.append(str(e))
    threads = [threading.Thread(target=thread_target, args=(i,)) for i in range(N)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_large_scale_event_loop_creation_and_teardown():
    """
    Large Scale: Create and close many event loops in threads to check for resource leaks or errors.
    """
    fix_asyncio_event_loop_policy() # 23.5μs -> 6.59μs (257% faster)
    N = 30
    loops = []
    errors = []
    def thread_target():
        try:
            loop = asyncio.get_event_loop()
            loops.append(loop)
            loop.run_until_complete(asyncio.sleep(0))
            loop.close()
        except Exception as e:
            errors.append(str(e))
    threads = [threading.Thread(target=thread_target) for _ in range(N)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    for loop in loops:
        pass

def test_thread_event_loop_heavy_workload():
    """
    Large Scale: Each thread runs several coroutines to test event loop robustness.
    """
    fix_asyncio_event_loop_policy() # 23.5μs -> 6.41μs (267% faster)
    N = 20
    results = []
    errors = []
    def thread_target(idx):
        try:
            loop = asyncio.get_event_loop()
            async def workload():
                # Run several sleeps
                total = 0
                for i in range(10):
                    await asyncio.sleep(0)
                    total += i
                return total + idx
            val = loop.run_until_complete(workload())
            results.append(val)
        except Exception as e:
            errors.append(str(e))
    threads = [threading.Thread(target=thread_target, args=(i,)) for i in range(N)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_event_loop_policy_thread_safety():
    """
    Large Scale: Rapidly create and destroy threads to check policy thread safety.
    """
    fix_asyncio_event_loop_policy() # 23.4μs -> 6.43μs (264% faster)
    N = 100
    errors = []
    def thread_target():
        try:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(asyncio.sleep(0))
            loop.close()
        except Exception as e:
            errors.append(str(e))
    threads = []
    for _ in range(N):
        t = threading.Thread(target=thread_target)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import asyncio
import sys
import threading
import time

# imports
import pytest  # used for our unit tests
from modules.initialize_util import fix_asyncio_event_loop_policy

# --- Basic Test Cases ---

def test_main_thread_event_loop_policy_basic():
    """
    Basic: After calling fix_asyncio_event_loop_policy,
    the main thread should still be able to get an event loop.
    """
    fix_asyncio_event_loop_policy() # 20.6μs -> 4.91μs (320% faster)
    loop = asyncio.get_event_loop()

def test_thread_event_loop_policy_basic():
    """
    Basic: After calling fix_asyncio_event_loop_policy,
    a new thread should be able to get an event loop automatically.
    """
    fix_asyncio_event_loop_policy() # 19.8μs -> 4.87μs (307% faster)
    result = {}

    def thread_func():
        loop = asyncio.get_event_loop()
        result['loop'] = loop

    t = threading.Thread(target=thread_func)
    t.start()
    t.join()

def test_multiple_threads_get_distinct_event_loops():
    """
    Basic: Multiple threads should get distinct event loops.
    """
    fix_asyncio_event_loop_policy() # 20.6μs -> 5.16μs (299% faster)
    loops = []

    def thread_func():
        loop = asyncio.get_event_loop()
        loops.append(loop)

    threads = [threading.Thread(target=thread_func) for _ in range(3)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_event_loop_creation_and_close():
    """
    Basic: Event loops created in threads can be closed and new ones created.
    """
    fix_asyncio_event_loop_policy() # 22.5μs -> 6.23μs (260% faster)
    results = {}

    def thread_func():
        loop1 = asyncio.get_event_loop()
        loop1.close()
        # After closing, a new loop should be created automatically
        loop2 = asyncio.get_event_loop()
        results['loop1'] = loop1
        results['loop2'] = loop2

    t = threading.Thread(target=thread_func)
    t.start()
    t.join()

# --- Edge Test Cases ---

def test_get_event_loop_before_and_after_policy():
    """
    Edge: Before calling fix_asyncio_event_loop_policy, getting event loop in a thread should fail.
    After calling, it should succeed.
    """
    import asyncio

    # Reset to default policy
    asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
    errors = []

    def thread_func_fail():
        try:
            asyncio.get_event_loop()
        except (RuntimeError, AssertionError) as e:
            errors.append(str(e))

    t1 = threading.Thread(target=thread_func_fail)
    t1.start()
    t1.join()

    # Now apply the fix
    fix_asyncio_event_loop_policy() # 20.5μs -> 3.99μs (414% faster)
    result = {}

    def thread_func_succeed():
        loop = asyncio.get_event_loop()
        result['loop'] = loop

    t2 = threading.Thread(target=thread_func_succeed)
    t2.start()
    t2.join()


def test_event_loop_policy_reentrancy():
    """
    Edge: Calling fix_asyncio_event_loop_policy multiple times should not break behavior.
    """
    fix_asyncio_event_loop_policy() # 23.7μs -> 7.84μs (202% faster)
    fix_asyncio_event_loop_policy() # 10.5μs -> 1.87μs (462% faster)
    result = {}

    def thread_func():
        loop = asyncio.get_event_loop()
        result['loop'] = loop

    t = threading.Thread(target=thread_func)
    t.start()
    t.join()

def test_event_loop_policy_with_many_threads():
    """
    Edge: Many threads should each get their own event loop without error.
    """
    fix_asyncio_event_loop_policy() # 18.6μs -> 5.07μs (267% faster)
    loops = []
    errors = []

    def thread_func():
        try:
            loop = asyncio.get_event_loop()
            loops.append(loop)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=thread_func) for _ in range(20)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_event_loop_policy_thread_cleanup():
    """
    Edge: After thread exits, its event loop is not accessible from another thread.
    """
    fix_asyncio_event_loop_policy() # 23.5μs -> 6.88μs (242% faster)
    loop_ids = []

    def thread_func():
        loop = asyncio.get_event_loop()
        loop_ids.append(id(loop))

    t = threading.Thread(target=thread_func)
    t.start()
    t.join()
    # Try to get the same loop from main thread (should be different)
    loop_main = asyncio.get_event_loop()

# --- Large Scale Test Cases ---

def test_large_scale_many_threads_event_loops():
    """
    Large Scale: Start 100 threads, each gets its own event loop.
    """
    fix_asyncio_event_loop_policy() # 22.0μs -> 5.41μs (307% faster)
    loops = []
    errors = []

    def thread_func():
        try:
            loop = asyncio.get_event_loop()
            loops.append(loop)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=thread_func) for _ in range(100)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_large_scale_event_loop_creation_and_closure():
    """
    Large Scale: Create and close event loops in 50 threads, then create new ones.
    """
    fix_asyncio_event_loop_policy() # 24.0μs -> 7.05μs (240% faster)
    loops_before = []
    loops_after = []

    def thread_func():
        loop1 = asyncio.get_event_loop()
        loops_before.append(loop1)
        loop1.close()
        loop2 = asyncio.get_event_loop()
        loops_after.append(loop2)

    threads = [threading.Thread(target=thread_func) for _ in range(50)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_large_scale_event_loop_usage():
    """
    Large Scale: In 30 threads, schedule a coroutine and ensure it runs.
    """
    fix_asyncio_event_loop_policy() # 23.6μs -> 6.94μs (241% faster)
    results = []

    async def simple_coro(val):
        await asyncio.sleep(0.01)
        return val * 2

    def thread_func(i):
        loop = asyncio.get_event_loop()
        fut = asyncio.run_coroutine_threadsafe(simple_coro(i), loop)
        results.append(fut.result(timeout=1))

    threads = [threading.Thread(target=thread_func, args=(i,)) for i in range(30)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_large_scale_policy_replacement():
    """
    Large Scale: Replace the policy repeatedly and ensure threads still get event loops.
    """
    for _ in range(10):
        fix_asyncio_event_loop_policy() # 78.9μs -> 19.2μs (311% faster)
    loops = []

    def thread_func():
        loop = asyncio.get_event_loop()
        loops.append(loop)

    threads = [threading.Thread(target=thread_func) for _ in range(30)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from modules.initialize_util import fix_asyncio_event_loop_policy

def test_fix_asyncio_event_loop_policy():
    fix_asyncio_event_loop_policy()

🔎 Concolic Coverage Tests and Runtime

Test File::Test Function	Original ⏱️	Optimized ⏱️	Speedup
codeflash_concolic_vlkwtasl/tmpr7tb07wn/test_concolic_coverage.py::test_fix_asyncio_event_loop_policy	24.6μs	7.07μs	248%✅

To edit these changes git checkout codeflash/optimize-fix_asyncio_event_loop_policy-mha4lgld and push.