Improve task processing time and query accuracy

Give the task manager a hint when new tasks are added and force a task
harvest prior to a task's status query to improve responsiveness of
dynamic workflows.

pyflow/doc/ChangeLog.txt

@@ -1,4 +1,5 @@
 v1.1.16 20170523
+* Improve task processing time and query accuracy
 * Fix task status query methods to work correctly within subworkflows
 v1.1.15 20170519
 * Add new launchTasksUntil demo

pyflow/src/pyflow.py

@@ -3853,7 +3853,7 @@ def _isWorkflowStopped() :
         return WorkflowRunner._allStop.isSet()
 
     def _addTaskCore(self, namespace, label, payload, dependencies) :
-        # private core taskAdd routine for hijacking
+        # private core addTask routine for hijacking
         # fromWorkflow is the workflow instance used to launch the task
         #
 
@@ -3942,6 +3942,11 @@ def _isTaskCompleteCore(self, namespace, taskLabel) :
 
         if not self._tdag.isTaskPresent(namespace, taskLabel) :
             return (False, False)
+
+        # Run a task harvest just before checking the task status
+        # to help ensure the status is up to date
+        self._tman.harvestTasks()
+
         task = self._tdag.getTask(namespace, taskLabel)
         return ( task.isDone(), task.isError() )
 
@@ -3963,9 +3968,12 @@ def _checkTaskLabel(label) :
 
 
     def _startTaskManager(self) :
-        # start a new task manager if one isn't already running:
+        # Start a new task manager if one isn't already running. If it is running
+        # provide a hint that a new task has just been added to the workflow.
         #
-        if (self._tman is not None) and (self._tman.isAlive()) : return
+        if (self._tman is not None) and (self._tman.isAlive()) :
+            self._tdag.isFinishedEvent.set()
+            return
         if not self._cdata().isTaskManagerException :
             self._tman = TaskManager(self._cdata(), self._tdag)
             self._tman.start()

scratch/test/pyflow_unit_tests.py

@@ -498,9 +498,7 @@ def __init__(self) :
         class SelfWorkflow2(WorkflowRunner) :
             def workflow(self2) :
                 self2.addTask("A2",getSleepCmd()+["0"])
-
-                # TODO Find a more robust way to ensure that A2 'should' be complete by the time a query is made below
-                time.sleep(102)
+                time.sleep(1)
                 self.taskStatus2 = self2.isTaskComplete("A2")
                 (self.taskStatus3, _) = self2.isTaskDone("A2")