TODO: factor a complete_exception method, like complete_task. and make all this stuff happen inside those?

TODO: should these run in the app_fu update lock? there's a bunch of other completion stuff that is not
locked, so probably no? in general this lock isn't protecting against two completion actions happening
at once.

split update_memo into two notify-exception and notify-result forms, that run before the future is set and which
means checkpointing can run in that place rather than racing with the user workflow.

this is needed to address an existing race condition - that task_exit tasks aren't checkpointed by the time
the workflow observes they have finished. it's debatable whether that is actually the intended functionality
or not, but i think it should be. otherwise what's the behaviour? a little bit later?

this manifests as awkwardness in implementation in later checkpoint methods too.

and moving more checkpoint code into update_memo makes the checkpoint/memo API unification cleaner.

See PR #NNNN which talks about the app future being used internally when it should only be user-facing.

Author: benclifford

parsl/dataflow/dflow.py

@@ -539,7 +539,7 @@ def _complete_task_result(self, task_record: TaskRecord, new_state: States, resu
         logger.info(f"Task {task_record['id']} completed ({old_state.name} -> {new_state.name})")
         task_record['time_returned'] = datetime.datetime.now()
 
-        self.memoizer.update_memo(task_record)
+        self.memoizer.update_memo_result(task_record, result)
 
         self._send_task_log_info(task_record)
 
@@ -558,7 +558,7 @@ def _complete_task_exception(self, task_record: TaskRecord, new_state: States, e
         logger.info(f"Task {task_record['id']} failed ({old_state.name} -> {new_state.name})")
         task_record['time_returned'] = datetime.datetime.now()
 
-        self.memoizer.update_memo(task_record)
+        self.memoizer.update_memo_exception(task_record, exception)
 
         self._send_task_log_info(task_record)
 

parsl/dataflow/memoization.py

@@ -283,7 +283,13 @@ def check_memo(self, task: TaskRecord) -> Optional[Future[Any]]:
         assert isinstance(result, Future) or result is None
         return result
 
-    def update_memo(self, task: TaskRecord) -> None:
+    def update_memo_result(self, task: TaskRecord, r: Any) -> None:
+        self._update_memo(task)
+
+    def update_memo_exception(self, task: TaskRecord, e: BaseException) -> None:
+        self._update_memo(task)
+
+    def _update_memo(self, task: TaskRecord) -> None:
         """Updates the memoization lookup table with the result from a task.
         This doesn't move any values around but associates the memoization
         hashsum with the completed (by success or failure) AppFuture.