work in progress: working on nodes lifecycle.

bonobo/execution/contexts/base.py

@@ -53,8 +53,7 @@ class Lifecycle:
 
     @property
     def should_loop(self):
-        # TODO XXX started/stopped?
-        return not any((self.defunct, self.killed))
+        return self.alive and not any((self.defunct, self.killed))
 
     @property
     def status(self):

bonobo/execution/contexts/graph.py

@@ -1,14 +1,19 @@
+import logging
 from functools import partial
+from queue import Empty
 from time import sleep
 
 from bonobo.config import create_container
 from bonobo.constants import BEGIN, END
+from bonobo.errors import InactiveReadableError
 from bonobo.execution import events
 from bonobo.execution.contexts.base import BaseContext
 from bonobo.execution.contexts.node import NodeExecutionContext
 from bonobo.execution.contexts.plugin import PluginExecutionContext
 from whistle import EventDispatcher
 
+logger = logging.getLogger(__name__)
+
 
 class GraphExecutionContext(BaseContext):
     """
@@ -104,11 +109,16 @@ class GraphExecutionContext(BaseContext):
             sleep(self.TICK_PERIOD)
 
     def loop(self):
-        while self.should_loop:
-            self.tick()
-            for node in self.nodes:
-                if node.should_loop:
+        nodes = set(node for node in self.nodes if node.should_loop)
+        while self.should_loop and len(nodes):
+            self.tick(pause=False)
+            for node in list(nodes):
+                try:
                     node.step()
+                except Empty:
+                    continue
+                except InactiveReadableError:
+                    nodes.discard(node)
 
     def stop(self, stopper=None):
         super(GraphExecutionContext, self).stop()

bonobo/execution/contexts/node.py

@@ -120,20 +120,22 @@ class NodeExecutionContext(BaseContext, WithStatistics):
                 break
             except Empty:
                 sleep(TICK_PERIOD)  # XXX: How do we determine this constant?
-                continue
-            except (
-                    NotImplementedError,
-                    UnrecoverableError,
-            ):
-                self.fatal(sys.exc_info())  # exit loop
-            except Exception:  # pylint: disable=broad-except
-                self.error(sys.exc_info())  # does not exit loop
-            except BaseException:
-                self.fatal(sys.exc_info())  # exit loop
 
         logger.debug('Node loop ends for {!r}.'.format(self))
 
     def step(self):
+        try:
+            self._step()
+        except InactiveReadableError:
+            raise
+        except (NotImplementedError, UnrecoverableError, ):
+            self.fatal(sys.exc_info())  # exit loop
+        except Exception:  # pylint: disable=broad-except
+            self.error(sys.exc_info())  # does not exit loop
+        except BaseException:
+            self.fatal(sys.exc_info())  # exit loop
+
+    def _step(self):
         """
         A single step in the loop.
 
@@ -280,7 +282,7 @@ class NodeExecutionContext(BaseContext, WithStatistics):
         If Queue raises (like Timeout or Empty), stat won't be changed.
 
         """
-        input_bag = self.input.get()
+        input_bag = self.input.get(timeout=0)
 
         # Store or check input type
         if self._input_type is None: