From ad7782f094bd76e68c8a611cc0f5103fb2f77ffd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 6 Nov 2026 10:04:16 -0500 Subject: [PATCH] The pilot waits only on pumps that owe work; the queue is the pump's and says so Pilot._wait_for_screen posted a Callback to the app and every widget on the screen after each pause, press and click, then waited for all of them. Waking an idle widget is not free: it dispatches the Callback and then runs a full Idle pass. Across Toad's suite profile that was 757,738 posts from _wait_for_screen and 588,863 _on_idle passes, almost all on widgets with nothing queued. - The message queue carries its activity: Owing after a put, Settled when its pump asks for a message and finds none (everything before is handled, idle work included). A refresh already enqueues Prompt through check_idle, so outstanding work always shows as a put. - MessagePump.call_after_pending asks that activity: Owing schedules the callback behind the work, Settled answers False. _wait_for_screen keeps its shape and calls it. - Queue moves into message_pump.py and _queue.py is deleted: it existed only to run pumps, and the split was what would have needed an import cycle once its state had pump behaviour. Toad's eight heaviest test files (FamilyMeta agent_comms, same Toad): 82.7 s -> 63.6 s wall, same three pre-existing channel-tab failures. Textual queue/pilot/pump/on/screens: 195 passed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XbFMSqJMarXn2ZE8iAL7Sd --- src/textual/_queue.py | 59 ----------------------- src/textual/message_pump.py | 94 ++++++++++++++++++++++++++++++++++++- src/textual/pilot.py | 2 +- tests/test_queue.py | 2 +- 4 files changed, 95 insertions(+), 62 deletions(-) delete mode 100644 src/textual/_queue.py diff --git a/src/textual/_queue.py b/src/textual/_queue.py deleted file mode 100644 index e4daa4b063..0000000000 --- a/src/textual/_queue.py +++ /dev/null @@ -1,59 +0,0 @@ -from __future__ import annotations - -import asyncio -from asyncio import Event -from collections import deque -from typing import Generic, TypeVar - -QueueType = TypeVar("QueueType") - - -class Queue(Generic[QueueType]): - """A cut-down version of asyncio.Queue - - This has just enough functionality to run the message pumps. - - """ - - def __init__(self) -> None: - self.values: deque[QueueType] = deque() - self.ready_event = Event() - - def put_nowait(self, value: QueueType) -> None: - self.values.append(value) - self.ready_event.set() - - def qsize(self) -> int: - return len(self.values) - - def empty(self) -> bool: - return not self.values - - def task_done(self) -> None: - pass - - async def get(self) -> QueueType: - # An Event wakes every waiter; a synchronous consumer may also drain - # the item before an awakened task resumes. Notification is not a - # reservation, so recheck the queue after each wait. - while not self.values: - self.ready_event.clear() - await self.ready_event.wait() - value = self.values.popleft() - if not self.values: - self.ready_event.clear() - return value - - def peek_nowait(self) -> QueueType: - """Read the next value without taking it from a waiting consumer.""" - if not self.values: - raise asyncio.QueueEmpty() - return self.values[0] - - def get_nowait(self) -> QueueType: - if not self.values: - raise asyncio.QueueEmpty() - value = self.values.popleft() - if not self.values: - self.ready_event.clear() - return value diff --git a/src/textual/message_pump.py b/src/textual/message_pump.py index 79341da0dd..8d80d8bab1 100644 --- a/src/textual/message_pump.py +++ b/src/textual/message_pump.py @@ -12,7 +12,9 @@ import asyncio import threading +from abc import ABC, abstractmethod from asyncio import CancelledError, QueueEmpty, Task, create_task +from collections import deque from contextlib import contextmanager from functools import partial from time import perf_counter @@ -23,6 +25,7 @@ Callable, ClassVar, Generator, + Generic, Iterable, Type, TypeVar, @@ -38,7 +41,6 @@ from textual._context import message_hook as message_hook_context_var from textual._context import prevent_message_types_stack from textual._on import DispatchScope, MessageHandler, NamedScope -from textual._queue import Queue from textual._time import time from textual._types import CallbackType from textual.constants import SLOW_THRESHOLD @@ -74,6 +76,86 @@ class MessagePumpClosed(Exception): pass +class QueueActivity(ABC): + """Whether a pump still owes work for the messages posted to it before now.""" + + @abstractmethod + def call_after(self, pump: MessagePump, callback: Callable[[], object]) -> bool: + """Schedule ``callback`` behind that work, or answer False when none is owed.""" + + +class Owing(QueueActivity): + """A message was posted since the pump last asked for one and found none.""" + + def call_after(self, pump: MessagePump, callback: Callable[[], object]) -> bool: + return pump.call_later(callback) + + +class Settled(QueueActivity): + """The pump asked for a message with none left: everything posted before is handled, + the idle work after it included.""" + + def call_after(self, pump: MessagePump, callback: Callable[[], object]) -> bool: + return False + + +OWING = Owing() +SETTLED = Settled() + +QueueType = TypeVar("QueueType") + + +class Queue(Generic[QueueType]): + """A cut-down asyncio.Queue: just enough to run a message pump, and whether it owes work.""" + + def __init__(self) -> None: + self.values: deque[QueueType] = deque() + self.ready_event = asyncio.Event() + self.activity: QueueActivity = SETTLED + + def put_nowait(self, value: QueueType) -> None: + self.values.append(value) + self.activity = OWING + self.ready_event.set() + + def qsize(self) -> int: + return len(self.values) + + def empty(self) -> bool: + return not self.values + + def task_done(self) -> None: + pass + + async def get(self) -> QueueType: + # An Event wakes every waiter; a synchronous consumer may also drain + # the item before an awakened task resumes. Notification is not a + # reservation, so recheck the queue after each wait. + while not self.values: + # Asking with none left: every value put before has been handled. + self.activity = SETTLED + self.ready_event.clear() + await self.ready_event.wait() + value = self.values.popleft() + if not self.values: + self.ready_event.clear() + return value + + def peek_nowait(self) -> QueueType: + """Read the next value without taking it from a waiting consumer.""" + if not self.values: + raise QueueEmpty() + return self.values[0] + + def get_nowait(self) -> QueueType: + if not self.values: + raise QueueEmpty() + value = self.values.popleft() + if not self.values: + self.ready_event.clear() + return value + + _MessagePumpMetaSub = TypeVar("_MessagePumpMetaSub", bound="_MessagePumpMeta") @@ -595,6 +677,16 @@ def call_later(self, callback: Callback, *args: Any, **kwargs: Any) -> bool: message = events.Callback(callback=partial(callback, *args, **kwargs)) return self.post_message(message) + def call_after_pending(self, callback: Callable[[], object]) -> bool: + """Schedule a callback after the messages already posted to this pump, the one in + dispatch included. + + Returns: + `True` if the callback was scheduled; `False` if this pump owes no work (or is + closed), so there is nothing to wait for. + """ + return self._message_queue.activity.call_after(self, callback) + def call_next( self, callback: Callback | events.Callback, *args: Any, **kwargs: Any ) -> None: diff --git a/src/textual/pilot.py b/src/textual/pilot.py index 65c0a4241b..f3b9daf3d4 100644 --- a/src/textual/pilot.py +++ b/src/textual/pilot.py @@ -492,7 +492,7 @@ def decrement_counter() -> None: # Increase the count for every successful call_later for child in children: - if child.call_later(decrement_counter): + if child.call_after_pending(decrement_counter): count += 1 if count: diff --git a/tests/test_queue.py b/tests/test_queue.py index bfbefe47ff..f58b87c74c 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -2,7 +2,7 @@ import pytest -from textual._queue import Queue +from textual.message_pump import Queue async def test_get_rechecks_after_nowait_consumer_drains_wakeup():