Make executor.outstanding a regular method (#3915)

Prior to this PR it was a dynamically generated attributed, which has
made understanding of hangs hard in the past - for example, by
disguising that these innocent looking lines might hang despite "only"
being attribute references:

logger.debug(f"Marking all {self.outstanding} outstanding tasks as
failed")

or

  active_tasks = executor.outstanding

After this PR, it should be clearer that those references to outstanding
are function invocations that might have complex behaviour.

# Changed Behaviour

external users of the parsl executor abstractions (eg Globus Compute)
might need to change their API use

## Type of change

- New feature

Author: benclifford

parsl/executors/high_throughput/executor.py

@@ -621,7 +621,6 @@ def _hold_manager(self, manager_id: str) -> None:
         self.command_client.run("HOLD_WORKER;{}".format(manager_id))
         logger.debug("Sent hold request to manager: {}".format(manager_id))
 
-    @property
     def outstanding(self) -> int:
         """Returns the count of tasks outstanding across the interchange
         and managers"""

parsl/executors/status_handling.py

@@ -122,7 +122,7 @@ def status_polling_interval(self):
         else:
             return self._provider.status_polling_interval
 
-    @abstractproperty
+    @abstractmethod
     def outstanding(self) -> int:
         """This should return the number of tasks that the executor has been given to run (waiting to run, and running now)"""
 

parsl/executors/taskvine/executor.py

@@ -573,7 +573,6 @@ def initialize_scaling(self):
         self._worker_command = self._construct_worker_command()
         self._patch_providers()
 
-    @property
     def outstanding(self) -> int:
         """Count the number of outstanding tasks."""
         logger.debug(f"Counted {self._outstanding_tasks} outstanding tasks")
@@ -659,7 +658,7 @@ def _collect_taskvine_results(self):
                 with self._outstanding_tasks_lock:
                     self._outstanding_tasks -= 1
         finally:
-            logger.debug(f"Marking all {self.outstanding} outstanding tasks as failed")
+            logger.debug(f"Marking all {self.outstanding()} outstanding tasks as failed")
             logger.debug("Acquiring tasks_lock")
             with self._tasks_lock:
                 logger.debug("Acquired tasks_lock")

parsl/executors/workqueue/executor.py

@@ -674,7 +674,6 @@ def initialize_scaling(self):
         self.worker_command = self._construct_worker_command()
         self._patch_providers()
 
-    @property
     def outstanding(self) -> int:
         """Count the number of outstanding slots required. This is inefficiently
         implemented and probably could be replaced with a counter.

parsl/jobs/strategy.py

@@ -193,7 +193,7 @@ def _general_strategy(self, executors: List[BlockProviderExecutor], *, strategy_
                 self.executors[label]['first'] = False
 
             # Tasks that are either pending completion
-            active_tasks = executor.outstanding
+            active_tasks = executor.outstanding()
 
             status = executor.status_facade
 

parsl/tests/site_tests/test_site.py

@@ -36,7 +36,7 @@ def test_platform(n=2, sleep_dur=10):
 
     print("Executor : ", dfk.executors[name])
     print("Connected   : ", dfk.executors[name].connected_workers)
-    print("Outstanding : ", dfk.executors[name].outstanding)
+    print("Outstanding : ", dfk.executors[name].outstanding())
 
     d = []
     for i in range(0, n):

parsl/tests/test_scaling/test_regression_3696_oscillation.py

@@ -48,7 +48,7 @@ def test_htex_strategy_does_not_oscillate(ns):
     statuses = {}
 
     executor.provider = provider
-    executor.outstanding = n_tasks
+    executor.outstanding = lambda: n_tasks
     executor.status_facade = statuses
     executor.workers_per_node = n_workers
 
@@ -97,7 +97,7 @@ def scale_in(n, max_idletime=None):
     # this assert fails due to issue #3696
 
     # Now check scale in happens with 0 load
-    executor.outstanding = 0
+    executor.outstanding = lambda: 0
     s.strategize([executor])
     executor.scale_in_facade.assert_called()
     assert len([k for k in statuses if statuses[k].state == JobState.PENDING]) == 0

parsl/tests/test_scaling/test_scale_down.py

@@ -66,7 +66,7 @@ def test_scale_out(tmpd_cwd, try_assert):
     num_managers = len(dfk.executors['htex_local'].connected_managers())
 
     assert num_managers == 0, "Expected 0 managers at start"
-    assert dfk.executors['htex_local'].outstanding == 0, "Expected 0 tasks at start"
+    assert dfk.executors['htex_local'].outstanding() == 0, "Expected 0 tasks at start"
 
     ntasks = 10
     ready_path = tmpd_cwd / "workers_ready"
@@ -85,7 +85,7 @@ def test_scale_out(tmpd_cwd, try_assert):
     finish_path.touch()  # Approximation of Event, via files
     [x.result() for x in futs]
 
-    assert dfk.executors['htex_local'].outstanding == 0
+    assert dfk.executors['htex_local'].outstanding() == 0
 
     def assert_kernel():
         return len(dfk.executors['htex_local'].connected_managers()) == _min_blocks

parsl/tests/test_scaling/test_scale_down_htex_auto_scale.py

@@ -63,7 +63,7 @@ def test_scale_out(tmpd_cwd, try_assert):
     num_managers = len(dfk.executors['htex_local'].connected_managers())
 
     assert num_managers == 0, "Expected 0 managers at start"
-    assert dfk.executors['htex_local'].outstanding == 0, "Expected 0 tasks at start"
+    assert dfk.executors['htex_local'].outstanding() == 0, "Expected 0 tasks at start"
 
     ntasks = _max_blocks * 2
     ready_path = tmpd_cwd / "workers_ready"
@@ -84,7 +84,7 @@ def test_scale_out(tmpd_cwd, try_assert):
     finish_path.touch()  # Approximation of Event, via files
     [x.result() for x in futs]
 
-    assert dfk.executors['htex_local'].outstanding == 0
+    assert dfk.executors['htex_local'].outstanding() == 0
 
     # now we can launch one "long" task -
     # and what should happen is that the connected_managers count "eventually" (?) converges to 1 and stays there.