New signal API: trio.open_signal_receiver

docs/source/reference-io.rst

@@ -638,5 +638,5 @@ Signals
 
 .. currentmodule:: trio
 
-.. autofunction:: catch_signals
-   :with: batched_signal_aiter
+.. autofunction:: open_signal_receiver
+   :with: signal_aiter

newsfragments/354.feature.rst

@@ -0,0 +1 @@
+New and improved signal catching API: :func:`open_signal_receiver`.

newsfragments/354.removal.rst

@@ -0,0 +1,8 @@
+``trio.catch_signals`` has been deprecated in favor of
+:func:`open_signal_receiver`. The main differences are:
+
+- it takes \*-args now to specify the list of signals (so
+  ``open_signal_receiver(SIGINT)`` instead of
+  ``catch_signals({SIGINT})``)
+- the async iterator now yields individual signals, instead of
+  "batches"

trio/_core/_entry_queue.py

@@ -133,7 +133,7 @@ class TrioToken:
     1. It lets you re-enter the Trio run loop from external threads or signal
        handlers. This is the low-level primitive that
        :func:`trio.run_sync_in_worker_thread` uses to receive results from
-       worker threads, that :func:`trio.catch_signals` uses to receive
+       worker threads, that :func:`trio.open_signal_receiver` uses to receive
        notifications about signals, and so forth.
 
     2. Each call to :func:`trio.run` has exactly one associated

trio/_core/_run.py

@@ -1213,7 +1213,7 @@ def run(
           This setting has no effect if your program has registered a custom
           SIGINT handler, or if :func:`run` is called from anywhere but the
           main thread (this is a Python limitation), or if you use
-          :func:`catch_signals` to catch SIGINT.
+          :func:`open_signal_receiver` to catch SIGINT.
 
     Returns:
       Whatever ``async_fn`` returns.

trio/_signals.py

@@ -1,11 +1,15 @@
 import signal
 from contextlib import contextmanager
+from collections import OrderedDict
 
 from . import _core
-from ._sync import Semaphore
-from ._util import signal_raise, aiter_compat, is_main_thread
+from ._sync import Event
+from ._util import (
+    signal_raise, aiter_compat, is_main_thread, ConflictDetector
+)
+from ._deprecate import deprecated
 
-__all__ = ["catch_signals"]
+__all__ = ["open_signal_receiver", "catch_signals"]
 
 # Discussion of signal handling strategies:
 #
@@ -46,71 +50,80 @@
 @contextmanager
 def _signal_handler(signals, handler):
     original_handlers = {}
-    for signum in signals:
-        original_handlers[signum] = signal.signal(signum, handler)
     try:
+        for signum in set(signals):
+            original_handlers[signum] = signal.signal(signum, handler)
         yield
     finally:
         for signum, original_handler in original_handlers.items():
             signal.signal(signum, original_handler)
 
 
-class SignalQueue:
+class SignalReceiver:
     def __init__(self):
-        self._semaphore = Semaphore(0, max_value=1)
-        self._pending = set()
+        # {signal num: None}
+        self._pending = OrderedDict()
+        self._have_pending = Event()
+        self._conflict_detector = ConflictDetector(
+            "only one task can iterate on a signal receiver at a time"
+        )
         self._closed = False
 
     def _add(self, signum):
         if self._closed:
             signal_raise(signum)
         else:
-            if not self._pending:
-                self._semaphore.release()
-            self._pending.add(signum)
+            self._pending[signum] = None
+            self._have_pending.set()
 
     def _redeliver_remaining(self):
         # First make sure that any signals still in the delivery pipeline will
         # get redelivered
         self._closed = True
 
-        # And then redeliver any that are sitting in pending. This is doen
+        # And then redeliver any that are sitting in pending. This is done
         # using a weird recursive construct to make sure we process everything
         # even if some of the handlers raise exceptions.
         def deliver_next():
             if self._pending:
-                signum = self._pending.pop()
+                signum, _ = self._pending.popitem(last=False)
                 try:
                     signal_raise(signum)
                 finally:
                     deliver_next()
 
         deliver_next()
 
+    # Helper for tests, not public or otherwise used
+    def _pending_signal_count(self):
+        return len(self._pending)
+
     @aiter_compat
     def __aiter__(self):
         return self
 
     async def __anext__(self):
         if self._closed:
-            raise RuntimeError("catch_signals block exited")
-        await self._semaphore.acquire()
-        assert self._pending
-        pending = set(self._pending)
-        self._pending.clear()
-        return pending
+            raise RuntimeError("open_signal_receiver block already exited")
+        # In principle it would be possible to support multiple concurrent
+        # calls to __anext__, but doing it without race conditions is quite
+        # tricky, and there doesn't seem to be any point in trying.
+        with self._conflict_detector.sync:
+            await self._have_pending.wait()
+            signum, _ = self._pending.popitem(last=False)
+            if not self._pending:
+                self._have_pending.clear()
+            return signum
 
 
 @contextmanager
-def catch_signals(signals):
+def open_signal_receiver(*signals):
     """A context manager for catching signals.
 
     Entering this context manager starts listening for the given signals and
     returns an async iterator; exiting the context manager stops listening.
 
-    The async iterator blocks until at least one signal has arrived, and then
-    yields a :class:`set` containing all of the signals that were received
-    since the last iteration.
+    The async iterator blocks until a signal arrives, and then yields it.
 
     Note that if you leave the ``with`` block while the iterator has
     unextracted signals still pending inside it, then they will be
@@ -119,7 +132,7 @@ def catch_signals(signals):
     block.
 
     Args:
-      signals: a set of signals to listen for.
+      signals: the signals to listen for.
 
     Raises:
       RuntimeError: if you try to use this anywhere except Python's main
@@ -129,25 +142,21 @@ def catch_signals(signals):
 
       A common convention for Unix daemons is that they should reload their
       configuration when they receive a ``SIGHUP``. Here's a sketch of what
-      that might look like using :func:`catch_signals`::
+      that might look like using :func:`open_signal_receiver`::
 
-         with trio.catch_signals({signal.SIGHUP}) as batched_signal_aiter:
-             async for batch in batched_signal_aiter:
-                 # We're only listening for one signal, so the batch is always
-                 # {signal.SIGHUP}, but if we were listening to more signals
-                 # then it could vary.
-                 for signum in batch:
-                     assert signum == signal.SIGHUP
-                     reload_configuration()
+         with trio.open_signal_receiver(signal.SIGHUP) as signal_aiter:
+             async for signum in signal_aiter:
+                 assert signum == signal.SIGHUP
+                 reload_configuration()
 
     """
     if not is_main_thread():
         raise RuntimeError(
-            "Sorry, catch_signals is only possible when running in the "
+            "Sorry, open_signal_receiver is only possible when running in "
             "Python interpreter's main thread"
         )
     token = _core.current_trio_token()
-    queue = SignalQueue()
+    queue = SignalReceiver()
 
     def handler(signum, _):
         token.run_sync_soon(queue._add, signum, idempotent=True)
@@ -157,3 +166,22 @@ def handler(signum, _):
             yield queue
     finally:
         queue._redeliver_remaining()
+
+
+class CompatSignalQueue:
+    def __init__(self, signal_queue):
+        self._signal_queue = signal_queue
+
+    @aiter_compat
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        return { await self._signal_queue.__anext__()}
+
+
+@deprecated("0.7.0", issue=354, instead=open_signal_receiver)
+@contextmanager
+def catch_signals(signals):
+    with open_signal_receiver(*signals) as signal_queue:
+        yield CompatSignalQueue(signal_queue)