Send task state monitoring record as part of general state update (#4034)

Previous PR #4032 should make this PR a straightforward refactoring.

After this PR, _send_task_info invocations and monitoring TASK_INFO
message sends now happen in only one place.

# Changed Behaviour

none

## Type of change

- Code maintenance/cleanup

Author: benclifford

parsl/dataflow/dflow.py

@@ -359,14 +359,12 @@ def handle_exec_update(self, task_record: TaskRecord, future: Future) -> None:
 
                 # record the final state for this try before we mutate for retries
                 self._update_task_state(task_record, States.fail_retryable)
-                self._send_task_info(task_record)
 
                 task_record['try_id'] += 1
                 task_record['try_time_launched'] = None
                 task_record['try_time_returned'] = None
                 task_record['fail_history'] = []
                 self._update_task_state(task_record, States.pending)
-                self._send_task_info(task_record)
 
                 logger.info("Task {} marked for retry".format(task_id))
 
@@ -395,19 +393,16 @@ def handle_exec_update(self, task_record: TaskRecord, future: Future) -> None:
                     task_record['joins'] = joinable
                     task_record['join_lock'] = threading.Lock()
                     self._update_task_state(task_record, States.joining)
-                    self._send_task_info(task_record)
                     joinable.add_done_callback(partial(self.handle_join_update, task_record))
                 elif joinable == []:  # got a list, but it had no entries, and specifically, no Futures.
                     task_record['joins'] = joinable
                     task_record['join_lock'] = threading.Lock()
                     self._update_task_state(task_record, States.joining)
-                    self._send_task_info(task_record)
                     self.handle_join_update(task_record, None)
                 elif isinstance(joinable, list) and [j for j in joinable if not isinstance(j, Future)] == []:
                     task_record['joins'] = joinable
                     task_record['join_lock'] = threading.Lock()
                     self._update_task_state(task_record, States.joining)
-                    self._send_task_info(task_record)
                     for inner_future in joinable:
                         inner_future.add_done_callback(partial(self.handle_join_update, task_record))
                 else:
@@ -515,7 +510,6 @@ def _complete_task_result(self, task_record: TaskRecord, new_state: States, resu
         self.memoizer.update_memo_result(task_record, result)
 
         self._update_task_state(task_record, new_state)
-        self._send_task_info(task_record)
 
         self.wipe_task(task_record['id'])
 
@@ -535,16 +529,17 @@ def _complete_task_exception(self, task_record: TaskRecord, new_state: States, e
         self.memoizer.update_memo_exception(task_record, exception)
 
         self._update_task_state(task_record, new_state)
-        self._send_task_info(task_record)
 
         self.wipe_task(task_record['id'])
 
         with task_record['app_fu']._update_lock:
             task_record['app_fu'].set_exception(exception)
 
     def _update_task_state(self, task_record: TaskRecord, new_state: States) -> None:
-        """Updates a task record state, and recording an appropriate change
-        to task state counters.
+        """Updates a task record state including accompanying consistency-keeping:
+
+        * record change in task state counters
+        * record change in monitoring
         """
 
         with self.task_state_counts_lock:
@@ -553,6 +548,8 @@ def _update_task_state(self, task_record: TaskRecord, new_state: States) -> None
             self.task_state_counts[new_state] += 1
             task_record['status'] = new_state
 
+        self._send_task_info(task_record)
+
     @staticmethod
     def _unwrap_remote_exception_wrapper(future: Future) -> Any:
         result = future.result()
@@ -705,7 +702,6 @@ def launch_task(self, task_record: TaskRecord) -> Future:
             exec_fu = executor.submit(function, task_record['resource_specification'], *args, **kwargs)
 
         self._update_task_state(task_record, States.launched)
-        self._send_task_info(task_record)
 
         if hasattr(exec_fu, "parsl_executor_task_id"):
             logger.info(
@@ -1025,9 +1021,7 @@ def submit(self,
                                                               waiting_message))
 
         logger.debug("Task {} set to pending state with AppFuture: {}".format(task_id, task_record['app_fu']))
-
         self._update_task_state(task_record, States.pending)
-        self._send_task_info(task_record)
 
         assert task_id not in self.tasks
         self.tasks[task_id] = task_record