Add _wait argument for local runner. This fixes broken pipe error.

severb · Mar 15, 2015 · bf201f8 · bf201f8
1 parent 68379b4
commit bf201f8
Showing 1 changed file with 5 additions and 4 deletions.
diff --git a/flowy/backend/local.py b/flowy/backend/local.py
@@ -47,11 +47,11 @@ def child_runner(self, workflow, w_id, args, kwargs):
     def set_decision(self, decision):
         self.decision = decision
 
-    def run(self):
+    def run(self, wait=False):
         self.reschedule_decision()
         self.stop.wait()
-        self.activity_executor.shutdown(wait=False)
-        self.workflow_executor.shutdown(wait=False)
+        self.activity_executor.shutdown(wait=wait)
+        self.workflow_executor.shutdown(wait=wait)
         if hasattr(self, 'result'):
             return self.result
         if hasattr(self, 'exception'):
@@ -365,7 +365,8 @@ def __call__(self, state, *args, **kwargs):
         return d
 
     def run(self, *args, **kwargs):
+        wait = kwargs.pop('_wait', False)
         a_executor = self.executor(max_workers=self.activity_workers)
         w_executor = self.executor(max_workers=self.workflow_workers)
         wr = WorkflowRunner(self, w_executor, a_executor, args, kwargs)
-        return wr.run()
+        return wr.run(wait=wait)