Add monitoring info for tasks given to a manager (#3833)

# Description

Update any interested listeners to the state of the manager when new
tasks are assigned. Based on `.start()`, this now covers most (all?)
relevant changes to a manager's info block that will be shared.

# Changed Behaviour

Adds a monitoring message per manager in the interchange loop, for any
outstanding tasks that move from the interchange to a worker manager. I
don't think any existing workflows will be impacted, but there will now
be an extra NODEINFO record in the sqlite database per task.

## Type of change

- New feature

Author: khk-globus

parsl/executors/high_throughput/interchange.py

@@ -328,7 +328,7 @@ def start(self) -> None:
             self.process_results_incoming(interesting_managers, monitoring_radio)
             self.expire_bad_managers(interesting_managers, monitoring_radio)
             self.expire_drained_managers(interesting_managers, monitoring_radio)
-            self.process_tasks_to_send(interesting_managers)
+            self.process_tasks_to_send(interesting_managers, monitoring_radio)
 
         self.zmq_context.destroy()
         delta = time.time() - start
@@ -452,7 +452,7 @@ def expire_drained_managers(self, interesting_managers: Set[bytes], monitoring_r
                 m['active'] = False
                 self._send_monitoring_info(monitoring_radio, m)
 
-    def process_tasks_to_send(self, interesting_managers: Set[bytes]) -> None:
+    def process_tasks_to_send(self, interesting_managers: Set[bytes], monitoring_radio: Optional[MonitoringRadioSender]) -> None:
         # Check if there are tasks that could be sent to managers
 
         logger.debug(
@@ -488,6 +488,7 @@ def process_tasks_to_send(self, interesting_managers: Set[bytes]) -> None:
                         else:
                             logger.debug("Manager %r is now saturated", manager_id)
                             interesting_managers.remove(manager_id)
+                    self._send_monitoring_info(monitoring_radio, m)
                 else:
                     interesting_managers.remove(manager_id)
                     # logger.debug("Nothing to send to manager {}".format(manager_id))

parsl/tests/test_monitoring/test_basic.py

@@ -120,7 +120,7 @@ def test_row_counts(tmpd_cwd, fresh_config):
             # Two entries: one showing manager active, one inactive
             result = connection.execute(text("SELECT COUNT(*) FROM node"))
             (c, ) = result.first()
-            assert c == 2
+            assert c == 3
 
         # There should be one block polling status
         # local provider has a status_polling_interval of 5s