From c0910a67b5617122412483c98da9d5ae3c4719e1 Mon Sep 17 00:00:00 2001 From: Will McGugan Date: Sun, 8 Jun 2025 15:07:00 +0100 Subject: [PATCH 1/2] sync primitives --- src/textual/app.py | 17 ++++++++++++++++- src/textual/message_pump.py | 14 +++++++++++--- src/textual/timer.py | 1 + 3 files changed, 28 insertions(+), 4 deletions(-) diff --git a/src/textual/app.py b/src/textual/app.py index 478593124e..87370c5152 100644 --- a/src/textual/app.py +++ b/src/textual/app.py @@ -865,6 +865,17 @@ def __init_subclass__(cls, *args, **kwargs) -> None: return super().__init_subclass__(*args, **kwargs) + def _thread_init(self): + """Initialize threading primitives for the current thread. + + https://github.com/Textualize/textual/issues/5845 + + """ + self._message_queue + self._mounted_event + self._exception_event + self._thread_id = threading.get_ident() + def _get_dom_base(self) -> DOMNode: """When querying from the app, we want to query the default screen.""" return self.default_screen @@ -2094,8 +2105,9 @@ async def run_auto_pilot( run_auto_pilot(auto_pilot, pilot), name=repr(pilot) ) + self._thread_init() + app._loop = asyncio.get_running_loop() - app._thread_id = threading.get_ident() with app._context(): try: await app._process_messages( @@ -3120,6 +3132,9 @@ async def _process_messages( terminal_size: tuple[int, int] | None = None, message_hook: Callable[[Message], None] | None = None, ) -> None: + + self._thread_init() + async def app_prelude() -> bool: """Work required before running the app. diff --git a/src/textual/message_pump.py b/src/textual/message_pump.py index 7a1e3bde7d..19e37e4a19 100644 --- a/src/textual/message_pump.py +++ b/src/textual/message_pump.py @@ -11,7 +11,6 @@ from __future__ import annotations import asyncio -import sys import threading from asyncio import CancelledError, QueueEmpty, Task, create_task from contextlib import contextmanager @@ -23,12 +22,10 @@ Awaitable, Callable, Generator, - Generic, Iterable, Type, TypeVar, cast, - overload, ) from weakref import WeakSet @@ -163,6 +160,15 @@ def _prevent_message_types_stack(self) -> list[set[type[Message]]]: prevent_message_types_stack.set(stack) return stack + def _thread_init(self): + """Initialize threading primitives for the current thread. + + Require for Python3.8 https://github.com/Textualize/textual/issues/5845 + + """ + self._message_queue + self._mounted_event + def _get_prevented_messages(self) -> set[type[Message]]: """A set of all the prevented message types.""" return self._prevent_message_types_stack[-1] @@ -506,6 +512,8 @@ async def _close_messages(self, wait: bool = True) -> None: def _start_messages(self) -> None: """Start messages task.""" + self._thread_init() + if self.app._running: self._task = create_task( self._process_messages(), name=f"message pump {self}" diff --git a/src/textual/timer.py b/src/textual/timer.py index f4b7b1ac1c..af657b3b69 100644 --- a/src/textual/timer.py +++ b/src/textual/timer.py @@ -152,6 +152,7 @@ async def _run(self) -> None: count = 0 _repeat = self._repeat _interval = self._interval + self._active # Force instantiation in same thread await self._active.wait() start = _time.get_time() From c6c8b8ea8bd489318e5164ce542d46a17250b4b2 Mon Sep 17 00:00:00 2001 From: Will McGugan Date: Sun, 8 Jun 2025 15:13:02 +0100 Subject: [PATCH 2/2] test --- tests/test_message_pump.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/test_message_pump.py b/tests/test_message_pump.py index 58fb5d325a..e90b576a52 100644 --- a/tests/test_message_pump.py +++ b/tests/test_message_pump.py @@ -1,3 +1,5 @@ +import threading + import pytest from textual._dispatch_key import dispatch_key @@ -169,3 +171,22 @@ def on_button_pressed(self, event: Button.Pressed) -> None: async with app.run_test() as pilot: await pilot.click(MyButton) assert app_button_pressed + + +async def test_thread_safe_post_message(): + class TextMessage(Message): + pass + + class TestApp(App): + + def on_mount(self) -> None: + msg = TextMessage() + threading.Thread(target=self.post_message, args=(msg,)).start() + + def on_text_message(self, message): + self.exit() + + app = TestApp() + + async with app.run_test() as pilot: + await pilot.pause()