Merge pull request #4 from cherrypy/transition-graph

Give each wait()ing thread its own state transition pipe

Author: jaraco

.gitattributes

@@ -1,3 +1,8 @@
+v4.1.0
+======
+
+* #4: Give each waiting thread its own state transition pipe.
+
 v4.0.0
 ======
 

CHANGES.rst

@@ -169,13 +169,7 @@ def __init__(self, transitions=None, errors=None,
             id = hex(random.randint(0, sys.maxint))[-8:]
         self.id = id
         self._priorities = {}
-
-        if select:
-            (self._state_transition_pipe_read,
-             self._state_transition_pipe_write) = os.pipe()
-        else:
-            (self._state_transition_pipe_read,
-             self._state_transition_pipe_write) = (None, None)
+        self._state_transition_pipes = set()
 
     @property
     def states(self):
@@ -204,8 +198,11 @@ def _transition(self, newstate, *args, **kwargs):
         """
         try:
             self.state = newstate
-            if self._state_transition_pipe_write is not None:
-                os.write(self._state_transition_pipe_write, '1')
+
+            # Write to any pipes created by threads calling self.wait().
+            # Use list() to avoid "Set changed size during iteration" errors.
+            for read_fd, write_fd in list(self._state_transition_pipes):
+                os.write(write_fd, "1")
 
             # Note: logging here means 1) the initial transition
             # will not be logged if loggers are set up in the initial
@@ -294,28 +291,52 @@ def publish(self, channel, *args, **kwargs):
             raise exc
         return output
 
-    def wait(self, state, interval=0.1, channel=None):
-        """Poll for the given state(s) at intervals; publish to channel."""
+    def wait(self, state, interval=0.1, channel=None, sleep=False):
+        """Poll for the given state(s) at intervals; publish to channel.
+
+        If sleep is True, the calling thread loops, sleeping for the given
+        interval each time, then returning only when the bus state is
+        one of the given states to wait for.
+
+        If sleep is False (the default) and the operating system supports
+        I/O multiplexing via the 'select' module, then an anonymous pipe
+        will be used to signal the waiting thread to wake up whenever
+        the state transitions. This allows the waiting thread to return
+        when the bus shuts down, for example, rather than waiting for
+        the sleep interval to elapse first. Each thread that calls wait()
+        creates a new pipe, so if file descriptors are in short supply
+        on your system you might need to use sleep instead.
+        """
         if isinstance(state, (tuple, list)):
             _states_to_wait_for = state
         else:
             _states_to_wait_for = [state]
 
+        if select:
+            pipe = os.pipe()
+            read_fd, write_fd = pipe
+            self._state_transition_pipes.add(pipe)
+
         def _wait():
-            while self.state not in _states_to_wait_for:
-                if self._state_transition_pipe_read is not None:
-                    try:
-                        r, w, x = select.select([self._state_transition_pipe_read], [], [], interval)
-                        if r:
-                            os.read(self._state_transition_pipe_read, 1)
-                    except (select.error, OSError):
-                        # Interrupted due to a signal (being handled by some
-                        # other thread). No need to panic, here, just check
-                        # the new state and proceed/return.
-                        pass
-                else:
-                    time.sleep(interval)
-                self.publish(channel)
+            try:
+                while self.state not in _states_to_wait_for:
+                    if select:
+                        try:
+                            r, w, x = select.select([read_fd], [], [], interval)
+                            if r:
+                                os.read(read_fd, 1)
+                        except (select.error, OSError):
+                            # Interrupted due to a signal (being handled by some
+                            # other thread). No need to panic, here, just check
+                            # the new state and proceed/return.
+                            pass
+                    else:
+                        time.sleep(interval)
+                    self.publish(channel)
+            finally:
+                self._state_transition_pipes.discard(pipe)
+                os.close(read_fd)
+                os.close(write_fd)
 
         # From http://psyco.sourceforge.net/psycoguide/bugs.html:
         # "The compiled machine code does not include the regular polling

magicbus/base.py

@@ -132,7 +132,7 @@ def graceful(self):
         self.transition('IDLE')
         self.transition('RUN')
 
-    def block(self, interval=0.1):
+    def block(self, interval=0.1, sleep=False):
         """Wait for the EXITED state, KeyboardInterrupt or SystemExit.
 
         This function is intended to be called only by the main thread.
@@ -142,7 +142,7 @@ def block(self, interval=0.1):
         the actual execv call (required on some platforms).
         """
         try:
-            self.wait('EXITED', interval=interval, channel='main')
+            self.wait('EXITED', interval=interval, channel='main', sleep=sleep)
         except (KeyboardInterrupt, IOError):
             # The time.sleep call might raise
             # "IOError: [Errno 4] Interrupted function call" on KBInt.

magicbus/process.py

@@ -192,6 +192,7 @@ def test_idle_to_exit(self):
 
     def test_wait(self):
         b = ProcessBus()
+        self.log(b)
 
         def f(desired_state):
             time.sleep(0.2)
@@ -221,6 +222,11 @@ def f():
         def g():
             time.sleep(0.4)
 
+        def main_listener():
+            main_calls.append(1)
+        main_calls = []
+        b.subscribe("main", main_listener)
+
         f_thread = threading.Thread(target=f, name='f')
         f_thread.start()
         threading.Thread(target=g, name='g').start()
@@ -250,6 +256,10 @@ def g():
             ]
         )
 
+        # While the bus was blocked, it should have published periodically
+        # to the "main" channel.
+        self.assertGreater(len(main_calls), 0)
+
     @unittest.skip("Fails intermittently; https://tinyurl.com/ybwwu4gz")
     def test_start_with_callback(self):
         b = ProcessBus()