abstract task name in thread pool

Author: HansVRP

openeo/extra/job_management/_manager.py

@@ -658,7 +658,7 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
                                 df_idx=i,
                             )
                             _log.info(f"Submitting task {task} to thread pool")
-                            self._worker_pool.submit_task(task)
+                            self._worker_pool.submit_task(task=task, pool_name="job_start")
 
                             stats["job_queued_for_start"] += 1
                             df.loc[i, "status"] = "queued_for_start"
@@ -771,7 +771,7 @@ def on_job_done(self, job: BatchJob, row):
             if self._worker_pool is None:
                 self._worker_pool = _JobManagerWorkerThreadPool()
 
-            self._worker_pool.submit_task(task)
+            self._worker_pool.submit_task(task=task, pool_name="job_download")
 
     def on_job_error(self, job: BatchJob, row):
         """

openeo/extra/job_management/_thread_worker.py

@@ -268,7 +268,6 @@ class _JobManagerWorkerThreadPool:
 
     """
     Generic wrapper that manages multiple thread pools with a dict.
-    Uses task class names as pool names automatically.
     """
 
     def __init__(self, pool_configs: Optional[Dict[str, int]] = None):
@@ -285,17 +284,18 @@ def _get_pool_name_for_task(self, task: Task) -> str:
         """
         return task.__class__.__name__
 
-    def submit_task(self, task: Task) -> None:
+    def submit_task(self, task: Task, pool_name: str = "default") -> None:
         """
-        Submit a task to a pool named after its class.
+        Submit a task to a specific pool.
         Creates pool dynamically if it doesn't exist.
-        """
-        pool_name = self._get_pool_name_for_task(task)
         
+        :param task: The task to execute
+        :param pool_name: Which pool to use (default, download, etc.)
+        """
         if pool_name not in self._pools:
             # Create pool on-demand
             max_workers = self._pool_configs.get(pool_name, 1)  # Default 1 worker
-            self._pools[pool_name] = _TaskThreadPool(max_workers=max_workers, name=pool_name)
+            self._pools[pool_name] = _TaskThreadPool(max_workers=max_workers)
             _log.info(f"Created pool '{pool_name}' with {max_workers} workers")
 
         self._pools[pool_name].submit_task(task)