Replace one interchange task counter with task_id (#4013)

Prior to this PR, the interchange had two task counters. The one removed
by this PR was used to:

i) disambiguate tasks of equal priority - in effect using the counter
   as yet another task identifier, distinct from the Parsl or HTEX
   task IDs.

ii) log how many tasks have been received by the interchange

This PR removes that counter:

i) tasks of equal priority are disambiguated by their HTEX-level
task_id. It is already a protocol requirement that different tasks have
different task IDs.

ii) The total tasks received log message is replaced by per-task logging
at the same place which makes it clear *which* tasks have been received,
instead of recording the count received. In debugging task flow, this
has been more useful to me.

# Changed Behaviour

Tasks of equal priority might be differently disambiguated now, but
probably not -- in general tasks will arrive in task_id order. Equal
priority tasks, by definition, have no expected relative execution
order, though.

## Type of change

- Update to human readable text: Documentation/error messages/comments
- Code maintenance/cleanup

Author: benclifford

parsl/executors/high_throughput/interchange.py

@@ -138,9 +138,6 @@ def __init__(self,
 
         self.pending_task_queue: SortedList[Any] = SortedList(key=lambda tup: (tup[0], tup[1]))
 
-        # count of tasks that have been received from the submit side
-        self.task_counter = 0
-
         # count of tasks that have been sent out to worker pools
         self.count = 0
 
@@ -332,15 +329,15 @@ def process_task_incoming(self) -> None:
             msg = self.task_incoming.recv_pyobj()
 
             # Process priority, higher number = lower priority
+            task_id = msg['task_id']
             resource_spec = msg['context'].get('resource_spec', {})
             priority = resource_spec.get('priority', float('inf'))
-            queue_entry = (-priority, -self.task_counter, msg)
+            queue_entry = (-priority, -task_id, msg)
 
-            logger.debug("putting message onto pending_task_queue")
+            logger.debug("Putting task %s onto pending_task_queue", task_id)
 
             self.pending_task_queue.add(queue_entry)
-            self.task_counter += 1
-            logger.debug(f"Fetched {self.task_counter} tasks so far")
+            logger.debug("Put task %s onto pending_task_queue", task_id)
 
     def process_manager_socket_message(
         self,

parsl/tests/test_htex/test_priority_queue.py

@@ -61,7 +61,7 @@ def interchange_logs_task_count():
             with open(htex.worker_logdir + "/interchange.log", "r") as f:
                 lines = f.readlines()
                 for line in lines:
-                    if f"Fetched {n} tasks so far" in line:
+                    if f"Put task {n} onto pending_task_queue" in line:
                         return True
             return False