Win wait (#3151)

* input waiter

* waiter objects

* try signal handler for windows

* selectors

* fix win wait

* log meta

* log

* meta loop

* loop

* correct wait

* Waiter tweak

* timeout change

* restore loop

* change constant

* quit

* tweak

* loops

* debug

* debug

* exit on no data

* change wait

* loop tweak

* log

* change wait

* experiement

* wrap with handle

* experiment

* Debug

* handle

* DWORD

* another attempt

* test

* log

* reading

* stream

* tweak

* Restore

* input reader

* reader

* Remove debug

* input reader

* shutdown devtools after waiter

* flush

* fileno

* exit meta

* windows reader

* remove logging

* formatting

* docstring

Author: willmcgugan

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "textual"
-version = "0.34.0"
+version = "0.35.0a1"
 homepage = "https://github.com/Textualize/textual"
 description = "Modern Text User Interface framework"
 authors = ["Will McGugan <[email protected]>"]

src/textual/__init__.py

@@ -64,12 +64,6 @@ def __rich_repr__(self) -> rich.repr.Result:
         yield self._verbosity, LogVerbosity.NORMAL
 
     def __call__(self, *args: object, **kwargs) -> None:
-        try:
-            app = active_app.get()
-        except LookupError:
-            print_args = (*args, *[f"{key}={value!r}" for key, value in kwargs.items()])
-            print(*print_args)
-            return
         if constants.LOG_FILE:
             output = " ".join(str(arg) for arg in args)
             if kwargs:
@@ -80,6 +74,12 @@ def __call__(self, *args: object, **kwargs) -> None:
 
             with open(constants.LOG_FILE, "a") as log_file:
                 print(output, file=log_file)
+        try:
+            app = active_app.get()
+        except LookupError:
+            print_args = (*args, *[f"{key}={value!r}" for key, value in kwargs.items()])
+            print(*print_args)
+            return
         if app.devtools is None or not app.devtools.is_connected:
             return
 

src/textual/app.py

@@ -1983,6 +1983,9 @@ async def _process_messages(
         self.log.system(driver=self.driver_class)
         self.log.system(loop=asyncio.get_running_loop())
         self.log.system(features=self.features)
+        if constants.LOG_FILE is not None:
+            _log_path = os.path.abspath(constants.LOG_FILE)
+            self.log.system(f"Writing logs to {_log_path!r}")
 
         try:
             if self.css_path:
@@ -2282,12 +2285,12 @@ async def _shutdown(self) -> None:
 
         await self._dispatch_message(events.Unmount())
 
-        if self.devtools is not None and self.devtools.is_connected:
-            await self._disconnect_devtools()
-
         if self._driver is not None:
             self._driver.close()
 
+        if self.devtools is not None and self.devtools.is_connected:
+            await self._disconnect_devtools()
+
         self._print_error_renderables()
 
         if constants.SHOW_RETURN:

src/textual/drivers/_byte_stream.py

@@ -151,7 +151,7 @@ def parse(
         read = self.read
         from_bytes = int.from_bytes
         while not self.is_eof:
-            packet_type = (yield read1()).decode("utf-8")
+            packet_type = (yield read1()).decode("utf-8", "ignore")
             size = from_bytes((yield read(4)), "big")
             payload = (yield read(size)) if size else b""
             on_token(BytePacket(packet_type, payload))

src/textual/drivers/_input_reader.py

@@ -0,0 +1,10 @@
+import platform
+
+__all__ = ["InputReader"]
+
+WINDOWS = platform.system() == "Windows"
+
+if WINDOWS:
+    from ._input_reader_windows import InputReader
+else:
+    from ._input_reader_linux import InputReader

src/textual/drivers/_input_reader_linux.py

@@ -0,0 +1,49 @@
+import os
+import selectors
+import sys
+from threading import Event
+from typing import Iterator
+
+from textual import log
+
+
+class InputReader:
+    """Read input from stdin."""
+
+    def __init__(self, timeout: float = 0.1) -> None:
+        """
+
+        Args:
+            timeout: Seconds to block for input.
+        """
+        self._fileno = sys.__stdin__.fileno()
+        self.timeout = timeout
+        self._selector = selectors.DefaultSelector()
+        self._selector.register(self._fileno, selectors.EVENT_READ)
+        self._exit_event = Event()
+
+    def more_data(self) -> bool:
+        """Check if there is data pending."""
+        EVENT_READ = selectors.EVENT_READ
+        for _key, events in self._selector.select(0.01):
+            if events & EVENT_READ:
+                return True
+        return False
+
+    def close(self) -> None:
+        """Close the reader (will exit the iterator)."""
+        self._exit_event.set()
+
+    def __iter__(self) -> Iterator[bytes]:
+        """Read input, yield bytes."""
+        fileno = self._fileno
+        read = os.read
+        exit_set = self._exit_event.is_set
+        EVENT_READ = selectors.EVENT_READ
+        while not exit_set():
+            for _key, events in self._selector.select(self.timeout):
+                if events & EVENT_READ:
+                    data = read(fileno, 1024)
+                    if not data:
+                        return
+                    yield data

src/textual/drivers/_input_reader_windows.py

@@ -0,0 +1,37 @@
+import os
+import sys
+from threading import Event
+from typing import Iterator
+
+
+class InputReader:
+    """Read input from stdin."""
+
+    def __init__(self, timeout: float = 0.1) -> None:
+        """
+
+        Args:
+            timeout: Seconds to block for input.
+        """
+        self._fileno = sys.__stdin__.fileno()
+        self.timeout = timeout
+        self._exit_event = Event()
+
+    def more_data(self) -> bool:
+        """Check if there is data pending."""
+        return True
+
+    def close(self) -> None:
+        """Close the reader (will exit the iterator)."""
+        self._exit_event.set()
+
+    def __iter__(self) -> Iterator[bytes]:
+        """Read input, yield bytes."""
+        while not self._exit_event.is_set():
+            try:
+                data = os.read(self._fileno, 1024) or None
+            except Exception:
+                break
+            if not data:
+                break
+            yield data

src/textual/drivers/linux_driver.py

@@ -242,8 +242,9 @@ def run_input_thread(self) -> None:
 
         def more_data() -> bool:
             """Check if there is more data to parse."""
+            EVENT_READ = selectors.EVENT_READ
             for key, events in selector.select(0.01):
-                if events:
+                if events & EVENT_READ:
                     return True
             return False
 
@@ -259,7 +260,7 @@ def more_data() -> bool:
             while not self.exit_event.is_set():
                 selector_events = selector.select(0.1)
                 for _selector_key, mask in selector_events:
-                    if mask | EVENT_READ:
+                    if mask & EVENT_READ:
                         unicode_data = decode(read(fileno, 1024))
                         for event in feed(unicode_data):
                             self.process_event(event)

src/textual/drivers/web_driver.py

@@ -15,7 +15,6 @@
 import json
 import os
 import platform
-import selectors
 import signal
 import sys
 from codecs import getincrementaldecoder
@@ -28,10 +27,15 @@
 from ..driver import Driver
 from ..geometry import Size
 from ._byte_stream import ByteStream
+from ._input_reader import InputReader
 
 WINDOWS = platform.system() == "Windows"
 
 
+class _ExitInput(Exception):
+    """Internal exception to force exit of input loop."""
+
+
 class WebDriver(Driver):
     """A headless driver that may be run remotely."""
 
@@ -41,10 +45,10 @@ def __init__(
         super().__init__(app, debug=debug, size=size)
         self.stdout = sys.__stdout__
         self.fileno = sys.__stdout__.fileno()
-        self.in_fileno = sys.__stdin__.fileno()
         self._write = partial(os.write, self.fileno)
         self.exit_event = Event()
         self._key_thread: Thread = Thread(target=self.run_input_thread)
+        self._input_reader = InputReader()
 
     def write(self, data: str) -> None:
         """Write data to the output device.
@@ -56,6 +60,15 @@ def write(self, data: str) -> None:
         data_bytes = data.encode("utf-8")
         self._write(b"D%s%s" % (len(data_bytes).to_bytes(4, "big"), data_bytes))
 
+    def write_meta(self, data: dict[str, object]) -> None:
+        """Write meta to the controlling process (i.e. textual-web)
+
+        Args:
+            data: Meta dict.
+        """
+        meta_bytes = json.dumps(data).encode("utf-8", errors="ignore")
+        self._write(b"M%s%s" % (len(meta_bytes).to_bytes(4, "big"), meta_bytes))
+
     def flush(self) -> None:
         pass
 
@@ -128,68 +141,61 @@ def disable_input(self) -> None:
     def stop_application_mode(self) -> None:
         """Stop application mode, restore state."""
         self.exit_event.set()
-        self._key_thread.join()
+        self._input_reader.close()
+        self.write_meta({"type": "exit"})
 
     def run_input_thread(self) -> None:
         """Wait for input and dispatch events."""
-        selector = selectors.DefaultSelector()
-        fileno = self.in_fileno
-        selector.register(fileno, selectors.EVENT_READ)
-
-        def more_data() -> bool:
-            """Check if there is more data to parse."""
-            for key, events in selector.select(0.01):
-                if events:
-                    return True
-            return False
-
-        parser = XTermParser(more_data, debug=self._debug)
-        feed = parser.feed
-
+        input_reader = self._input_reader
+        parser = XTermParser(input_reader.more_data, debug=self._debug)
         utf8_decoder = getincrementaldecoder("utf-8")().decode
         decode = utf8_decoder
-        read = os.read
-        EVENT_READ = selectors.EVENT_READ
-
         # The server sends us a stream of bytes, which contains the equivalent of stdin, plus
         # in band data packets.
         byte_stream = ByteStream()
         try:
-            while not self.exit_event.is_set():
-                selector_events = selector.select(0.1)
-                for _selector_key, mask in selector_events:
-                    if mask | EVENT_READ:
-                        data = read(fileno, 1024)  # raw data
-
-                        for packet_type, payload in byte_stream.feed(data):
-                            if packet_type == "D":
-                                # Treat as stdin
-                                for event in feed(decode(payload)):
-                                    self.process_event(event)
-                            else:
-                                # Process meta information separately
-                                self._on_meta(packet_type, payload)
-        except Exception as error:
-            log(error)
+            for data in input_reader:
+                for packet_type, payload in byte_stream.feed(data):
+                    if packet_type == "D":
+                        # Treat as stdin
+                        for event in parser.feed(decode(payload)):
+                            self.process_event(event)
+                    else:
+                        # Process meta information separately
+                        self._on_meta(packet_type, payload)
+        except _ExitInput:
+            pass
+        except Exception:
+            from traceback import format_exc
+
+            log(format_exc())
         finally:
-            selector.close()
+            input_reader.close()
 
     def _on_meta(self, packet_type: str, payload: bytes) -> None:
+        """Private method to dispatch meta.
+
+        Args:
+            packet_type: Packet type (currently always "M")
+            payload: Meta payload (JSON encoded as bytes).
+        """
         payload_map = json.loads(payload)
         _type = payload_map.get("type")
         if isinstance(payload_map, dict):
             self.on_meta(_type, payload_map)
 
     def on_meta(self, packet_type: str, payload: dict) -> None:
+        """Process meta information.
+
+        Args:
+            packet_type: The type of the packet.
+            payload: meta dict.
+        """
         if packet_type == "resize":
             self._size = (payload["width"], payload["height"])
             size = Size(*self._size)
-            event = events.Resize(size, size)
-            asyncio.run_coroutine_threadsafe(
-                self._app._post_message(event),
-                loop=self._loop,
-            )
+            self._app.post_message(events.Resize(size, size))
         elif packet_type == "quit":
-            asyncio.run_coroutine_threadsafe(
-                self._app._post_message(messages.ExitApp()), loop=self._loop
-            )
+            self._app.post_message(messages.ExitApp())
+        elif packet_type == "exit":
+            raise _ExitInput()