From 23a9ac2b885ad743edbedec6777a325f4eb96df5 Mon Sep 17 00:00:00 2001 From: Renaud Cepre <32103211+renaudcepre@users.noreply.github.com> Date: Thu, 25 Jun 2026 06:51:46 +0200 Subject: [PATCH 1/4] perf(events): run sync handlers inline by default The event bus offloaded every sync handler to the thread pool via run_in_threadpool, costing a socket + event-loop wakeup per call. Profiling a suite of trivial async tests showed ~288 such offloads per test, dominating a ~9ms/test overhead. Run sync handlers inline on the loop by default; offload only handlers that opt in with on(..., blocking=True). HANDLER_START/END meta-events and emit_and_collect also run inline. Per-test overhead drops ~9ms -> ~0.4ms; on httpx's real async suite apte goes from ~1.14x slower to ~1.05-1.11x faster than pytest. Full apte test suite green (1264 passed), ruff + mypy clean. --- apte/events/bus.py | 53 ++++++++++++++++++++++++++++++++-------------- 1 file changed, 37 insertions(+), 16 deletions(-) diff --git a/apte/events/bus.py b/apte/events/bus.py index dd52fbb..f011121 100644 --- a/apte/events/bus.py +++ b/apte/events/bus.py @@ -26,12 +26,19 @@ class _RegisteredHandler: func: Callable[..., Any] name: str + blocking: bool = False class EventBus: """Decoupled event dispatch with async support. - Sync handlers are executed in a thread pool (blocking the event emission). + Sync handlers run inline on the event loop by default: they are expected to + be fast and non-blocking (reporters buffering, result collection). Inline + dispatch avoids a thread-pool round-trip (socket + loop wakeup) per call, + which otherwise dominates the per-test cost. Handlers that genuinely block + (heavy disk/network work) opt in with `on(..., blocking=True)` to be + offloaded to the thread pool instead. + Async handlers run fire-and-forget and are tracked for cleanup. Use wait_pending() before session end to ensure all async handlers complete. @@ -84,10 +91,19 @@ def _run_sync_with_lock( else: handler() - def on(self, event: Event, handler: Callable[..., Any]) -> None: - """Register a handler for an event.""" + def on( + self, event: Event, handler: Callable[..., Any], *, blocking: bool = False + ) -> None: + """Register a handler for an event. + + Sync handlers run inline by default. Pass `blocking=True` for sync + handlers that perform heavy blocking work and must not stall the loop; + those are offloaded to the thread pool. + """ name = get_callable_name(handler) - self._handlers[event].append(_RegisteredHandler(func=handler, name=name)) + self._handlers[event].append( + _RegisteredHandler(func=handler, name=name, blocking=blocking) + ) def off(self, event: Event, handler: Callable[..., Any]) -> None: """Unregister a handler for an event.""" @@ -119,16 +135,21 @@ async def emit(self, event: Event, data: Any = None) -> None: ) self._pending_tasks.add(task) task.add_done_callback(self._pending_tasks.discard) - else: + elif handler_entry.blocking: + # Opt-in: offload genuinely blocking handlers to a thread, + # serialized per owner to protect shared instance state. lock = self._get_owner_lock(handler) - if data is not None: - await run_in_threadpool( - self._run_sync_with_lock, lock, handler, data - ) - else: - await run_in_threadpool( - self._run_sync_with_lock, lock, handler, None - ) + await run_in_threadpool( + self._run_sync_with_lock, lock, handler, data + ) + duration = time.perf_counter() - start_time + await self._emit_handler_end( + handler_name, event, False, duration, None + ) + else: + # Default: run inline. No thread, so no cross-thread race and + # no lock needed; the call runs to completion on the loop. + self._run_sync_with_lock(None, handler, data) duration = time.perf_counter() - start_time await self._emit_handler_end( handler_name, event, False, duration, None @@ -176,7 +197,7 @@ async def _emit_handler_start( if asyncio.iscoroutinefunction(handler_entry.func): await handler_entry.func(info) else: - await run_in_threadpool(handler_entry.func, info) + handler_entry.func(info) except Exception: logger.exception("HANDLER_START listener %s failed", handler_entry.name) @@ -197,7 +218,7 @@ async def _emit_handler_end( if asyncio.iscoroutinefunction(handler_entry.func): await handler_entry.func(info) else: - await run_in_threadpool(handler_entry.func, info) + handler_entry.func(info) except Exception: logger.exception("HANDLER_END listener %s failed", handler_entry.name) @@ -219,7 +240,7 @@ async def emit_and_collect(self, event: Event, data: Any) -> Any: if asyncio.iscoroutinefunction(handler): result = await handler(data) else: - result = await run_in_threadpool(handler, data) + result = handler(data) if result is not None: data = result except Exception: From e0925c8088bed09cae85d91d9fcb56a559aaf968 Mon Sep 17 00:00:00 2001 From: Renaud Cepre <32103211+renaudcepre@users.noreply.github.com> Date: Thu, 25 Jun 2026 06:55:37 +0200 Subject: [PATCH 2/4] perf(events): cache handler async-ness, skip empty meta-events iscoroutinefunction was recomputed on every dispatch (~289 calls/test in a synthetic profile, each unwrapping partials and inspecting code flags). A handler's async-ness is fixed, so compute it once at registration and store it on the registered handler. Also skip HANDLER_START/END meta-event emission (HandlerInfo construction + loop) when no listeners are registered, the common case. cProfile of 200 trivial async tests: 0.353s -> 0.179s (on top of the inline fix; ~21x vs the original 3.85s). Full apte suite green (1264), ruff + mypy clean. --- apte/events/bus.py | 28 +++++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/apte/events/bus.py b/apte/events/bus.py index f011121..93b814e 100644 --- a/apte/events/bus.py +++ b/apte/events/bus.py @@ -27,6 +27,9 @@ class _RegisteredHandler: func: Callable[..., Any] name: str blocking: bool = False + # Computed once at registration: a handler's async-ness never changes, and + # asyncio.iscoroutinefunction is costly (unwraps partials, inspects flags). + is_async: bool = False class EventBus: @@ -102,7 +105,12 @@ def on( """ name = get_callable_name(handler) self._handlers[event].append( - _RegisteredHandler(func=handler, name=name, blocking=blocking) + _RegisteredHandler( + func=handler, + name=name, + blocking=blocking, + is_async=asyncio.iscoroutinefunction(handler), + ) ) def off(self, event: Event, handler: Callable[..., Any]) -> None: @@ -121,7 +129,7 @@ async def emit(self, event: Event, data: Any = None) -> None: for handler_entry in self._handlers[event]: handler = handler_entry.func handler_name = handler_entry.name - is_async = asyncio.iscoroutinefunction(handler) + is_async = handler_entry.is_async await self._emit_handler_start(handler_name, event, is_async) start_time = time.perf_counter() @@ -191,10 +199,13 @@ async def _emit_handler_start( self, name: str, event: Event, is_async: bool ) -> None: """Emit HANDLER_START without triggering handler events (avoid recursion).""" + listeners = self._handlers[Event.HANDLER_START] + if not listeners: + return info = HandlerInfo(name=name, event=event, is_async=is_async) - for handler_entry in self._handlers[Event.HANDLER_START]: + for handler_entry in listeners: try: - if asyncio.iscoroutinefunction(handler_entry.func): + if handler_entry.is_async: await handler_entry.func(info) else: handler_entry.func(info) @@ -210,12 +221,15 @@ async def _emit_handler_end( error: Exception | None, ) -> None: """Emit HANDLER_END without triggering handler events (avoid recursion).""" + listeners = self._handlers[Event.HANDLER_END] + if not listeners: + return info = HandlerInfo( name=name, event=event, is_async=is_async, duration=duration, error=error ) - for handler_entry in self._handlers[Event.HANDLER_END]: + for handler_entry in listeners: try: - if asyncio.iscoroutinefunction(handler_entry.func): + if handler_entry.is_async: await handler_entry.func(info) else: handler_entry.func(info) @@ -237,7 +251,7 @@ async def emit_and_collect(self, event: Event, data: Any) -> Any: for handler_entry in self._handlers[event]: handler = handler_entry.func try: - if asyncio.iscoroutinefunction(handler): + if handler_entry.is_async: result = await handler(data) else: result = handler(data) From 5e6a6e0cccc0ff516658bfad13ac58a35d482df9 Mon Sep 17 00:00:00 2001 From: Renaud Cepre <32103211+renaudcepre@users.noreply.github.com> Date: Thu, 25 Jun 2026 10:38:43 +0200 Subject: [PATCH 3/4] fix(interrupt): gate scheduler on sync interrupt state, not soft_stop_event The soft-stop check between tests read only soft_stop_event, an asyncio.Event armed via call_soon_threadsafe (one loop iteration late). With sync handlers now running inline, emit() no longer suspends, so the scheduler reaches the next _should_stop check without yielding to the loop and the event is still unset while a stop is already pending. Read the synchronous should_stop_new_tests state instead, which _handle_signal sets the instant the signal is delivered. Fixes the flaky test_soft_stop_prevents_pending_tests_from_starting under Python 3.14 (stable at 3.12, ~30% failures at 3.14). --- apte/core/execution/parallel.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/apte/core/execution/parallel.py b/apte/core/execution/parallel.py index d9c5fa3..ab475f9 100644 --- a/apte/core/execution/parallel.py +++ b/apte/core/execution/parallel.py @@ -175,8 +175,16 @@ async def _process_queue(self, ctx: _ParallelExecutionState) -> None: ctx.queue.task_done() def _should_stop(self, exitfirst_flag: asyncio.Event | None) -> bool: - """Check if execution should stop (interrupt or exitfirst).""" - return self._interrupt_handler.soft_stop_event.is_set() or ( + """Check if execution should stop (interrupt or exitfirst). + + Reads the synchronous interrupt state, not soft_stop_event: the event is + armed via call_soon_threadsafe (one loop iteration late), but _state is + set the instant the signal handler runs. Between two tests the scheduler + can reach this check without yielding to the loop (sync handlers run + inline, so emit() never suspends), so the event may still be unset while + a stop is already pending. + """ + return self._interrupt_handler.should_stop_new_tests or ( exitfirst_flag is not None and exitfirst_flag.is_set() ) From 930a691c2ab9b8e325059a737753efc4071b7c99 Mon Sep 17 00:00:00 2001 From: Renaud Cepre <32103211+renaudcepre@users.noreply.github.com> Date: Thu, 25 Jun 2026 11:03:05 +0200 Subject: [PATCH 4/4] chore: gitignore JOURNAL.md (private local journal) --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index caeae90..d853d51 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,6 @@ web/node_modules/ /htmlcov/ _sandbox/ site/ + +# Private work journal, kept local only +JOURNAL.md