Remove unnecessary intermediate collection (#3834)

# Description

The results handling currently moves the messages around twice for no
benefit. Just remove the tasks from the manager's bookkeeping data
structure, and prepare the final list to send. No need for the second
iteration nor the `got_result` boolean.

# Changed Behaviour

None.

## Type of change

- Code maintenance/cleanup

Author: khk-globus

parsl/executors/high_throughput/interchange.py

@@ -505,13 +505,24 @@ def process_results_incoming(self, interesting_managers: Set[bytes], monitoring_
             else:
                 logger.debug("Got %s result items in batch from manager %r", len(all_messages), manager_id)
 
-                b_messages = []
+                m = self._ready_managers[manager_id]
+                b_messages_to_send = []
 
                 for p_message in all_messages:
                     r = pickle.loads(p_message)
                     if r['type'] == 'result':
                         # process this for task ID and forward to executor
-                        b_messages.append((p_message, r))
+                        logger.debug("Removing task %s from manager record %r", r["task_id"], manager_id)
+                        try:
+                            m['tasks'].remove(r['task_id'])
+                            b_messages_to_send.append(p_message)
+                        except Exception:
+                            logger.exception(
+                                "Ignoring exception removing task_id %s for manager %r with task list %s",
+                                r['task_id'],
+                                manager_id,
+                                m["tasks"]
+                            )
                     elif r['type'] == 'monitoring':
                         # the monitoring code makes the assumption that no
                         # monitoring messages will be received if monitoring
@@ -525,43 +536,19 @@ def process_results_incoming(self, interesting_managers: Set[bytes], monitoring_
                     else:
                         logger.error("Interchange discarding result_queue message of unknown type: %s", r["type"])
 
-                got_result = False
-                m = self._ready_managers[manager_id]
-                for (_, r) in b_messages:
-                    assert 'type' in r, f"Message is missing type entry: {r}"
-                    if r['type'] == 'result':
-                        got_result = True
-                        try:
-                            logger.debug("Removing task %s from manager record %r", r["task_id"], manager_id)
-                            m['tasks'].remove(r['task_id'])
-                        except Exception:
-                            # If we reach here, there's something very wrong.
-                            logger.exception(
-                                "Ignoring exception removing task_id %s for manager %r with task list %s",
-                                r['task_id'],
-                                manager_id,
-                                m["tasks"]
-                            )
-
-                b_messages_to_send = []
-                for (b_message, _) in b_messages:
-                    b_messages_to_send.append(b_message)
-
                 if b_messages_to_send:
                     logger.debug("Sending messages on results_outgoing")
                     self.results_outgoing.send_multipart(b_messages_to_send)
                     logger.debug("Sent messages on results_outgoing")
 
-                logger.debug("Current tasks on manager %r: %s", manager_id, m["tasks"])
-                if len(m['tasks']) == 0 and m['idle_since'] is None:
-                    m['idle_since'] = time.time()
-
-                # A manager is only made interesting here if a result was
-                # received, which means there should be capacity for a new
-                # task now. Heartbeats and monitoring messages do not make a
-                # manager become interesting.
-                if got_result:
+                    # At least one result received, so manager now has idle capacity
                     interesting_managers.add(manager_id)
+
+                    if len(m['tasks']) == 0 and m['idle_since'] is None:
+                        m['idle_since'] = time.time()
+
+                logger.debug("Current tasks on manager %r: %s", manager_id, m["tasks"])
+
             logger.debug("leaving results_incoming section")
 
     def expire_bad_managers(self, interesting_managers: Set[bytes], monitoring_radio: Optional[MonitoringRadioSender]) -> None: