Pull more completion code into _complete_task and rename (#4001)

This PR adds _send_task_log_info as a mandatory part of task completion,
which is already manually enforced. In one place, around line 500, one
_send_task_log_info is split into two: one which then disappears into
complete_task_result, and the other which remains in place for the
non-factored error handling path.

This PR renames _complete_task to _complete_task_result to reflect that
it is for the result path not the exception path (taking naming from the
Future concepts of the same name), and to open up the way for
factorisation of the exception path into _complete_task_exception.

# Changed Behaviour

none

## Type of change

- Code maintenance/cleanup

Author: benclifford

parsl/dataflow/dflow.py

@@ -386,12 +386,10 @@ def handle_exec_update(self, task_record: TaskRecord, future: Future) -> None:
 
         else:
             if task_record['from_memo']:
-                self._complete_task(task_record, States.memo_done, res)
-                self._send_task_log_info(task_record)
+                self._complete_task_result(task_record, States.memo_done, res)
             else:
                 if not task_record['join']:
-                    self._complete_task(task_record, States.exec_done, res)
-                    self._send_task_log_info(task_record)
+                    self._complete_task_result(task_record, States.exec_done, res)
                 else:
                     # This is a join task, and the original task's function code has
                     # completed. That means that the future returned by that code
@@ -505,6 +503,7 @@ def handle_join_update(self, task_record: TaskRecord, inner_app_future: Optional
                 self.memoizer.update_memo(task_record)
                 with task_record['app_fu']._update_lock:
                     task_record['app_fu'].set_exception(e)
+                self._send_task_log_info(task_record)
 
             else:
                 # all the joinables succeeded, so construct a result:
@@ -517,12 +516,10 @@ def handle_join_update(self, task_record: TaskRecord, inner_app_future: Optional
                         res.append(future.result())
                 else:
                     raise TypeError(f"Unknown joinable type {type(joinable)}")
-                self._complete_task(task_record, States.exec_done, res)
+                self._complete_task_result(task_record, States.exec_done, res)
 
             self._log_std_streams(task_record)
 
-            self._send_task_log_info(task_record)
-
     def handle_app_update(self, task_record: TaskRecord, future: AppFuture) -> None:
         """This function is called as a callback when an AppFuture
         is in its final state.
@@ -548,7 +545,7 @@ def handle_app_update(self, task_record: TaskRecord, future: AppFuture) -> None:
         self.wipe_task(task_id)
         return
 
-    def _complete_task(self, task_record: TaskRecord, new_state: States, result: Any) -> None:
+    def _complete_task_result(self, task_record: TaskRecord, new_state: States, result: Any) -> None:
         """Set a task into a completed state
         """
         assert new_state in FINAL_STATES
@@ -561,6 +558,9 @@ def _complete_task(self, task_record: TaskRecord, new_state: States, result: Any
         task_record['time_returned'] = datetime.datetime.now()
 
         self.memoizer.update_memo(task_record)
+
+        self._send_task_log_info(task_record)
+
         with task_record['app_fu']._update_lock:
             task_record['app_fu'].set_result(result)