re-adding len(to_keep)

Author: HansVRP

openeo/extra/job_management/_manager.py

@@ -403,7 +403,7 @@ def stop_job_thread(self, timeout_seconds: Optional[float] = _UNSET):
 
         .. versionadded:: 0.32.0
         """
-        if self._worker_pool is not None or self._worker_pool.number_pending_tasks() > 0:
+        if self._worker_pool is not None:
             self._worker_pool.shutdown()
             self._worker_pool = None
 
@@ -519,7 +519,7 @@ def run_jobs(
                     statuses=["not_started", "created", "queued_for_start", "queued", "running"]
                 ).values()) > 0
 
-            or (self._worker_pool.number_pending_tasks() > 0)
+            or (self._worker_pool.number_pending_tasks() > 0) #avoid stopping the manager too early if there are still tasks being processed
 
         ):
             self._job_update_loop(job_db=job_db, start_job=start_job, stats=stats)
@@ -704,7 +704,7 @@ def _process_threadworker_updates(
         :param stats:       Dictionary accumulating statistic counters
         """
         # Retrieve completed task results immediately
-        results = worker_pool.process_futures(timeout=0)
+        results, _ = worker_pool.process_futures(timeout=0)
 
         # Collect update dicts
         updates: List[Dict[str, Any]] = []

openeo/extra/job_management/_thread_worker.py

@@ -194,10 +194,10 @@ class _TaskThreadPool:
 
     :param max_workers:
         Maximum number of concurrent threads to use for execution.
-        Defaults to 1.
+        Defaults to 2.
     """
 
-    def __init__(self, max_workers: int = 1, name: str = 'default'):
+    def __init__(self, max_workers: int = 2, name: str = 'default'):
         self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
         self._future_task_pairs: List[Tuple[concurrent.futures.Future, Task]] = []
         self._name = name
@@ -251,10 +251,10 @@ def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskRe
         _log.info("process_futures: %d tasks done, %d tasks remaining", len(results), len(to_keep))
 
         self._future_task_pairs = to_keep
-        return results
+        return results, len(to_keep)
 
     def number_pending_tasks(self) -> int:
-        """Return the number of tasks that are still pending (not completed)."""
+        """Approximation of the number of tasks is used to avoid stopping the job manager too early."""
         return len(self._future_task_pairs)
 
     def shutdown(self) -> None:
@@ -305,13 +305,15 @@ def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskRe
         Returns: (all_results, dict of remaining tasks per pool)
         """
         all_results = []
+        all_remaining = {}
 
         for pool_name, pool in self._pools.items():
-            results = pool.process_futures(timeout)
+            results, remaining = pool.process_futures(timeout)
             all_results.extend(results)
-            
-        return all_results
-    
+            all_remaining[pool_name] = remaining
+
+        return all_results, all_remaining
+
     def number_pending_tasks(self, pool_name: Optional[str] = None) -> int:
         if pool_name:
             pool = self._pools.get(pool_name)

tests/extra/job_management/test_thread_worker.py

@@ -221,12 +221,12 @@ def worker_pool(self) -> Iterator[_TaskThreadPool]:
         pool.shutdown()
 
     def test_no_tasks(self, worker_pool):
-        results = worker_pool.process_futures(timeout=10)
+        results, _ = worker_pool.process_futures(timeout=10)
         assert results == []
 
     def test_submit_and_process(self, worker_pool):
         worker_pool.submit_task(DummyTask(job_id="j-123", df_idx=0))
-        results = worker_pool.process_futures(timeout=10)
+        results, _ = worker_pool.process_futures(timeout=10)
         assert results == [
             _TaskResult(job_id="j-123", df_idx=0, db_update={"status": "dummified"}, stats_update={"dummy": 1}),
         ]
@@ -235,14 +235,14 @@ def test_submit_and_process_zero_timeout(self, worker_pool):
         worker_pool.submit_task(DummyTask(job_id="j-123", df_idx=0))
         # Trigger context switch
         time.sleep(0.1)
-        results = worker_pool.process_futures(timeout=0)
+        results, _ = worker_pool.process_futures(timeout=0)
         assert results == [
             _TaskResult(job_id="j-123", df_idx=0, db_update={"status": "dummified"}, stats_update={"dummy": 1}),
         ]
 
     def test_submit_and_process_with_error(self, worker_pool):
         worker_pool.submit_task(DummyTask(job_id="j-666", df_idx=0))
-        results = worker_pool.process_futures(timeout=10)
+        results, _ = worker_pool.process_futures(timeout=10)
         assert results == [
             _TaskResult(
                 job_id="j-666",
@@ -255,13 +255,13 @@ def test_submit_and_process_with_error(self, worker_pool):
 
     def test_submit_and_process_iterative(self, worker_pool):
         worker_pool.submit_task(NopTask(job_id="j-1", df_idx=1))
-        results = worker_pool.process_futures(timeout=1)
+        results, _ = worker_pool.process_futures(timeout=1)
         assert results == [_TaskResult(job_id="j-1", df_idx=1)]
 
         # Add some more
         worker_pool.submit_task(NopTask(job_id="j-22", df_idx=22))
         worker_pool.submit_task(NopTask(job_id="j-222", df_idx=222))
-        results = worker_pool.process_futures(timeout=1)
+        results, _ = worker_pool.process_futures(timeout=1)
         assert results == [_TaskResult(job_id="j-22", df_idx=22), _TaskResult(job_id="j-222", df_idx=222)]
 
     def test_submit_multiple_simple(self, worker_pool):
@@ -270,7 +270,7 @@ def test_submit_multiple_simple(self, worker_pool):
             worker_pool.submit_task(NopTask(job_id=f"j-{j}", df_idx=j))
 
         # Process all of them (non-zero timeout, which should be plenty of time for all of them to finish)
-        results = worker_pool.process_futures(timeout=1)
+        results, _ = worker_pool.process_futures(timeout=1)
         expected = [_TaskResult(job_id=f"j-{j}", df_idx=j) for j in range(5)]
         assert sorted(results, key=lambda r: r.job_id) == expected
 
@@ -291,24 +291,24 @@ def test_submit_multiple_blocking_and_failing(self, worker_pool):
             )
 
         # Initial state: nothing happened yet
-        results = worker_pool.process_futures(timeout=0)
+        results, _ = worker_pool.process_futures(timeout=0)
         assert results == []
 
         # No changes even after timeout
-        results = worker_pool.process_futures(timeout=0.1)
+        results, _ = worker_pool.process_futures(timeout=0.1)
         assert results == []
 
         # Set one event and wait for corresponding result
         events[0].set()
-        results = worker_pool.process_futures(timeout=0.1)
+        results, _ = worker_pool.process_futures(timeout=0.1)
         assert results == [
             _TaskResult(job_id="j-0", df_idx=0, db_update={"status": "all fine"}),
         ]
 
         # Release all but one event
         for j in range(n - 1):
             events[j].set()
-        results = worker_pool.process_futures(timeout=0.1)
+        results, _ = worker_pool.process_futures(timeout=0.1)
         assert results == [
             _TaskResult(job_id="j-1", df_idx=1, db_update={"status": "all fine"}),
             _TaskResult(job_id="j-2", df_idx=2, db_update={"status": "all fine"}),
@@ -323,15 +323,15 @@ def test_submit_multiple_blocking_and_failing(self, worker_pool):
         # Release all events
         for j in range(n):
             events[j].set()
-        results = worker_pool.process_futures(timeout=0.1)
+        results, _ = worker_pool.process_futures(timeout=0.1)
         assert results == [
             _TaskResult(job_id="j-4", df_idx=4, db_update={"status": "all fine"}),
         ]
 
     def test_shutdown(self, worker_pool):
         # Before shutdown
         worker_pool.submit_task(NopTask(job_id="j-123", df_idx=0))
-        results = worker_pool.process_futures(timeout=0.1)
+        results, _ = worker_pool.process_futures(timeout=0.1)
         assert results == [_TaskResult(job_id="j-123", df_idx=0)]
 
         worker_pool.shutdown()
@@ -346,7 +346,7 @@ def test_job_start_task(self, worker_pool, dummy_backend, caplog):
         task = _JobStartTask(job_id=job.job_id, df_idx=0, root_url=dummy_backend.connection.root_url, bearer_token=None)
         worker_pool.submit_task(task)
 
-        results = worker_pool.process_futures(timeout=1)
+        results, _ = worker_pool.process_futures(timeout=1)
         assert results == [
             _TaskResult(
                 job_id="job-000",
@@ -365,7 +365,7 @@ def test_job_start_task_failure(self, worker_pool, dummy_backend, caplog):
         task = _JobStartTask(job_id=job.job_id, df_idx=0, root_url=dummy_backend.connection.root_url, bearer_token=None)
         worker_pool.submit_task(task)
 
-        results = worker_pool.process_futures(timeout=1)
+        results, _ = worker_pool.process_futures(timeout=1)
         assert results == [
             _TaskResult(
                 job_id="job-000", df_idx=0, db_update={"status": "start_failed"}, stats_update={"start_job error": 1}
@@ -433,7 +433,7 @@ def test_submit_task_creates_pool(self, thread_pool):
         assert "default" in thread_pool._pools
 
         # Process to complete the task
-        results = thread_pool.process_futures(timeout=0.1)
+        results, _ = thread_pool.process_futures(timeout=0.1)
         assert len(results) == 1
         assert results[0].job_id == "j-1"
 
@@ -472,7 +472,7 @@ def test_submit_multiple_task_types(self, thread_pool):
 
     def test_process_futures_updates_empty(self, thread_pool):
         """Test process futures with no pools."""
-        results = thread_pool.process_futures(timeout=0)
+        results, _ = thread_pool.process_futures(timeout=0)
         assert results == []
 
     def test_process_futures_updates_multiple_pools(self, thread_pool):
@@ -482,7 +482,7 @@ def test_process_futures_updates_multiple_pools(self, thread_pool):
         thread_pool.submit_task(NopTask(job_id="j-2", df_idx=2))  # NopTask pool
         thread_pool.submit_task(DummyTask(job_id="j-3", df_idx=3))  # DummyTask pool
 
-        results = thread_pool.process_futures(timeout=0.1)
+        results, _ = thread_pool.process_futures(timeout=0.1)
 
         assert len(results) == 3
 
@@ -508,7 +508,7 @@ def test_process_futures_updates_partial_completion(self):
         pool.submit_task(quick_task, "quick")     # NopTask pool
 
         # Process with timeout=0 - only quick task should complete
-        results = pool.process_futures(timeout=0)
+        results, _ = pool.process_futures(timeout=0)
 
         # Only quick task completed
         assert len(results) == 1
@@ -520,7 +520,7 @@ def test_process_futures_updates_partial_completion(self):
 
         # Release blocking task and process again
         event.set()
-        results2 = pool.process_futures(timeout=0.1)
+        results2, _ = pool.process_futures(timeout=0.1)
 
         assert len(results2) == 1
         assert results2[0].job_id == "j-block"
@@ -609,7 +609,7 @@ def execute(self) -> _TaskResult:
         assert pool.number_pending_tasks() == 1
 
         # Process it
-        results = pool.process_futures(timeout=0.1)
+        results, _ = pool.process_futures(timeout=0.1)
         assert len(results) == 1
         assert results[0].job_id == "j-1"
 
@@ -633,7 +633,7 @@ def submit_tasks(start_idx: int):
         assert thread_pool.number_pending_tasks() == 15
 
         # Process them all
-        results = thread_pool.process_futures(timeout=0.5)
+        results, _ = thread_pool.process_futures(timeout=0.5)
 
         assert len(results) == 15
 
@@ -661,7 +661,7 @@ def test_pool_parallelism_with_blocking_tasks(self):
         for event in events:
             event.set()
 
-        results = pool.process_futures(timeout=0.5)        
+        results, _ = pool.process_futures(timeout=0.5)        
         assert len(results) == 5
 
         for result in results:
@@ -675,7 +675,7 @@ def test_task_with_error_handling(self, thread_pool):
         thread_pool.submit_task(DummyTask(job_id="j-666", df_idx=0))
 
         # Process it
-        results = thread_pool.process_futures(timeout=0.1)
+        results, _ = thread_pool.process_futures(timeout=0.1)
 
         # Should get error result
         assert len(results) == 1
@@ -692,7 +692,7 @@ def test_mixed_success_and_error_tasks(self, thread_pool):
         thread_pool.submit_task(DummyTask(job_id="j-3", df_idx=3))   # Success
 
         # Process all
-        results = thread_pool.process_futures(timeout=0.1)
+        results, _ = thread_pool.process_futures(timeout=0.1)
 
         # Should get 3 results
         assert len(results) == 3