Address race condition in htex priority test (#3998)

This race condition is in the test, not in the main Parsl codebase, and
is that the test was not waiting for tasks to all arrive in the
interchange before launching a worker.

This was happening fast enough most of the time, but under some extreme
race condition testing, this was breaking the test - I'm working towards
that testing being automated in CI, so this is not a theoretical
concern.

If the tasks are not all in the interchange when a worker connects, they
won't necessarily be run in priority order: the highest priority task of
the subset of tasks that has already arrived will be run first, even if
it is not the highest priority task of the entire task set.

# Changed Behaviour

this is a test behaviour change, should not be user-exposed

## Type of change

- Bug fix

Author: benclifford

parsl/tests/test_htex/test_priority_queue.py

@@ -18,7 +18,7 @@ def fake_task(parsl_resource_specification=None):
 
 
 @pytest.mark.local
-def test_priority_queue():
+def test_priority_queue(try_assert):
     provider = LocalProvider(
         init_blocks=0,
         max_blocks=0,
@@ -30,6 +30,7 @@ def test_priority_queue():
         max_workers_per_node=1,
         manager_selector=RandomManagerSelector(),
         provider=provider,
+        worker_debug=True,  # needed to instrospect interchange logs
     )
 
     config = Config(
@@ -50,6 +51,22 @@ def test_priority_queue():
             spec = {'priority': priority}
             futures[(priority, i)] = fake_task(parsl_resource_specification=spec)
 
+        # wait for the interchange to have received all tasks
+        # (which happens asynchronously to the main thread, and is otherwise
+        # a race condition which can cause this test to fail)
+
+        n = len(priorities)
+
+        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:
+                        return True
+            return False
+
+        try_assert(interchange_logs_task_count)
+
         provider.max_blocks = 1
         htex.scale_out_facade(1)  # don't wait for the JSP to catch up