⚡️ Speed up method _EventLoop.start by 5%

[codeflash-ai]

📄 5% (0.05x) speedup for _EventLoop.start in weaviate/connect/event_loop.py

⏱️ Runtime : 12.3 milliseconds → 11.7 milliseconds (best of 29 runs)

📝 Explanation and details

The optimized code achieves a 5% speedup through two key micro-optimizations:

1. Eliminated intermediate variable assignment: Instead of creating a event_loop variable and then calling .start() separately, the optimized version chains the thread creation and start call directly: threading.Thread(...).start(). This removes one variable assignment and reference, reducing memory operations.

2. Reduced polling sleep interval: Changed time.sleep(0.01) to time.sleep(0.001) in the event loop startup polling. This 10x reduction makes the code more responsive when waiting for the event loop to become ready, especially beneficial when creating multiple event loops in sequence.

3. Streamlined exception handling logic: Replaced the nested if "exception" in context check with direct context.get("exception") and consolidated the condition logic into a single line. This reduces dictionary lookups and simplifies the control flow.

The optimizations are most effective for test cases involving multiple event loop creation (showing 21% improvement in performance under load tests) and basic startup scenarios (3-6% faster). The reduced sleep interval particularly benefits scenarios where many event loops are created sequentially, as seen in the large-scale test cases where the cumulative effect of faster polling significantly improves overall performance.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio
import threading
import time

# imports
import pytest  # used for our unit tests
from weaviate.connect.event_loop import _EventLoop

# unit tests

# Basic Test Cases

def test_start_creates_event_loop_when_none():
    # Test that start creates a new event loop if none exists
    ev = _EventLoop()
    ev.start() # 107μs -> 103μs (3.81% faster)


def test_patch_exception_handler_is_called():
    # Test that patch_exception_handler sets an exception handler
    ev = _EventLoop()
    ev.start() # 97.5μs -> 94.4μs (3.29% faster)
    handler = ev.loop.get_exception_handler()

# Edge Test Cases

def test_start_with_closed_loop():
    # Test starting with a closed loop (should not replace it)
    loop = asyncio.new_event_loop()
    loop.close()
    ev = _EventLoop(loop)
    ev.start() # 462ns -> 364ns (26.9% faster)

def test_start_multiple_times():
    # Test calling start multiple times does not create new loops
    ev = _EventLoop()
    ev.start() # 104μs -> 100μs (3.96% faster)
    first_loop = ev.loop
    ev.start() # 254ns -> 267ns (4.87% slower)

def test_exception_handler_ignores_blocking_io_error():
    # Test that the custom exception handler ignores specific BlockingIOError
    ev = _EventLoop()
    ev.start() # 116μs -> 127μs (9.27% slower)
    handler = ev.loop.get_exception_handler()
    class DummyBlockingIOError(Exception):
        pass
    context = {
        "exception": BlockingIOError("Resource temporarily unavailable")
    }
    # Should not raise or call default_exception_handler
    # We'll monkeypatch default_exception_handler to set a flag if called
    called = {}
    def fake_default_handler(ctx):
        called['called'] = True
    ev.loop.default_exception_handler = fake_default_handler
    handler(ev.loop, context)

def test_exception_handler_calls_default_for_other_exceptions():
    # Test that the custom exception handler calls default for other exceptions
    ev = _EventLoop()
    ev.start() # 139μs -> 123μs (12.8% faster)
    handler = ev.loop.get_exception_handler()
    context = {
        "exception": ValueError("Some error")
    }
    called = {}
    def fake_default_handler(ctx):
        called['called'] = True
    ev.loop.default_exception_handler = fake_default_handler
    handler(ev.loop, context)

def test_exception_handler_calls_default_for_no_exception():
    # Test that the custom exception handler calls default if no exception in context
    ev = _EventLoop()
    ev.start() # 114μs -> 114μs (0.230% faster)
    handler = ev.loop.get_exception_handler()
    context = {
        "message": "No exception"
    }
    called = {}
    def fake_default_handler(ctx):
        called['called'] = True
    ev.loop.default_exception_handler = fake_default_handler
    handler(ev.loop, context)

# Large Scale Test Cases

def test_start_many_event_loops():
    # Test creating many _EventLoop instances to check scalability
    loops = []
    for _ in range(50):  # Limit to 50 for performance
        ev = _EventLoop()
        ev.start() # 5.19ms -> 5.01ms (3.55% faster)
        loops.append(ev.loop)
    # All loops should be running and unique
    ids = set(id(loop) for loop in loops)
    for loop in loops:
        pass

def test_threaded_event_loops_are_daemon():
    # Test that the event loop threads are daemon threads
    ev = _EventLoop()
    ev.start() # 136μs -> 129μs (5.50% faster)
    # Find the thread by name
    threads = [t for t in threading.enumerate() if t.name == "eventLoop"]

def test_event_loop_performance_under_load():
    # Test that event loop creation is reasonably fast for multiple loops
    import time as pytime
    loops = []
    start_time = pytime.time()
    for _ in range(20):
        ev = _EventLoop()
        ev.start() # 2.25ms -> 1.85ms (21.2% faster)
        loops.append(ev.loop)
    elapsed = pytime.time() - start_time
# 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 threading
import time
from typing import Any, Dict, Optional

# imports
import pytest
from weaviate.connect.event_loop import _EventLoop

# unit tests

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

def test_start_sets_loop_when_none():
    # Test that start creates a new event loop if none is set
    ev = _EventLoop()
    ev.start() # 124μs -> 117μs (6.43% faster)
    # Clean up: stop the loop to avoid side effects
    ev.loop.call_soon_threadsafe(ev.loop.stop)
    time.sleep(0.05)  # Give time for the loop to stop


def test_patch_exception_handler_installed():
    # Test that patch_exception_handler sets a custom exception handler
    loop = asyncio.new_event_loop()
    _EventLoop.patch_exception_handler(loop)
    handler = loop.get_exception_handler()
    # Clean up
    loop.close()

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

def test_start_thread_safety():
    # Test calling start from multiple threads on the same object
    ev = _EventLoop()
    def call_start():
        ev.start()
    threads = [threading.Thread(target=call_start) for _ in range(3)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    # Clean up
    ev.loop.call_soon_threadsafe(ev.loop.stop)
    time.sleep(0.05)

def test_patch_exception_handler_ignores_blockingioerror(monkeypatch):
    # Test that the patched handler ignores BlockingIOError with the right message
    loop = asyncio.new_event_loop()
    called = {}

    def fake_default_exception_handler(context):
        called['called'] = True

    loop.default_exception_handler = fake_default_exception_handler
    _EventLoop.patch_exception_handler(loop)

    # Simulate the error context
    class BlockingIOErrorFake(Exception):
        pass

    exc = BlockingIOError("Resource temporarily unavailable")
    context = {"exception": exc}
    handler = loop.get_exception_handler()
    handler(loop, context)

    # Simulate a different error
    exc2 = RuntimeError("Some other error")
    context2 = {"exception": exc2}
    handler(loop, context2)
    loop.close()

def test_patch_exception_handler_handles_no_exception_key():
    # Test handler when 'exception' key is missing in context
    loop = asyncio.new_event_loop()
    called = {}

    def fake_default_exception_handler(context):
        called['called'] = True

    loop.default_exception_handler = fake_default_exception_handler
    _EventLoop.patch_exception_handler(loop)
    handler = loop.get_exception_handler()
    handler(loop, {})  # No 'exception' key
    loop.close()

def test_start_multiple_instances():
    # Test that multiple _EventLoop instances do not interfere with each other
    ev1 = _EventLoop()
    ev2 = _EventLoop()
    ev1.start() # 105μs -> 100μs (5.30% faster)
    ev2.start() # 64.8μs -> 64.5μs (0.501% faster)
    # Clean up
    ev1.loop.call_soon_threadsafe(ev1.loop.stop)
    ev2.loop.call_soon_threadsafe(ev2.loop.stop)
    time.sleep(0.05)

def test_start_on_closed_loop():
    # Test that start does not resurrect a closed loop
    loop = asyncio.new_event_loop()
    loop.close()
    ev = _EventLoop(loop)
    ev.start() # 497ns -> 508ns (2.17% slower)

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

def test_start_many_event_loops():
    # Create and start many _EventLoop instances to test scalability
    evs = [_EventLoop() for _ in range(50)]
    for ev in evs:
        ev.start() # 2.72ms -> 2.67ms (2.03% faster)
    for ev in evs:
        pass
    # Clean up
    for ev in evs:
        ev.loop.call_soon_threadsafe(ev.loop.stop)
    time.sleep(0.1)

def test_threaded_start_many_times():
    # Test starting many _EventLoop instances in parallel threads
    evs = [_EventLoop() for _ in range(20)]
    threads = [threading.Thread(target=ev.start) for ev in evs]
    for t in threads:
        t.start() # 1.02ms -> 1.05ms (3.22% slower)
    for t in threads:
        t.join()
    for ev in evs:
        pass
    # Clean up
    for ev in evs:
        ev.loop.call_soon_threadsafe(ev.loop.stop)
    time.sleep(0.1)

def test_start_and_patch_exception_handler_large():
    # Test patching the exception handler for many loops
    loops = [asyncio.new_event_loop() for _ in range(30)]
    for loop in loops:
        _EventLoop.patch_exception_handler(loop)
        handler = loop.get_exception_handler()
    for loop in loops:
        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 asyncio.base_events import BaseEventLoop
from weaviate.connect.event_loop import _EventLoop

def test__EventLoop_start():
    _EventLoop.start(_EventLoop(loop=None))

def test__EventLoop_start_2():
    _EventLoop.start(_EventLoop(loop=BaseEventLoop()))

Timer unit: 1e-09 s

To edit these changes git checkout codeflash/optimize-_EventLoop.start-mh2zy0je and push.