fix queue

Author: giancarloromeo

services/docker-compose.yml

@@ -1199,15 +1199,15 @@ services:
       CELERY_CONCURRENCY: 100
     networks: *storage_networks
 
-  sto-worker-low-prio:
+  sto-worker-cpu-bound:
     image: ${DOCKER_REGISTRY:-itisfoundation}/storage:${DOCKER_IMAGE_TAG:-master-github-latest}
     init: true
-    hostname: "sto-worker-low-prio-{{.Node.Hostname}}-{{.Task.Slot}}"
+    hostname: "sto-worker-cpu-bound-{{.Node.Hostname}}-{{.Task.Slot}}"
     environment:
       <<: *storage_environment
       STORAGE_WORKER_MODE: "true"
       CELERY_CONCURRENCY: 1
-      CELERY_QUEUES: "low-prio"
+      CELERY_QUEUES: "cpu-bound"
     networks: *storage_networks
 
   rabbit:

services/storage/src/simcore_service_storage/api/_worker_tasks/tasks.py

@@ -20,7 +20,7 @@ def setup_worker_tasks(app: Celery) -> None:
         logging.INFO,
         msg="Storage setup Worker Tasks",
     ):
-        define_task(app, export_data, task_queue="low-prio")
+        define_task(app, export_data)
         define_task(app, compute_path_size)
         define_task(app, delete_paths)
         define_task(app, complete_upload_file)

services/storage/src/simcore_service_storage/api/rpc/_data_export.py

@@ -60,6 +60,7 @@ async def start_data_export(
         task_uuid = await get_celery_client(app).send_task(
             "export_data",
             task_context=job_id_data.model_dump(),
+            task_queue="cpu-bound",
             files=data_export_start.file_and_folder_ids,  # ANE: adapt here your signature
         )
     except CeleryError as exc:

services/storage/src/simcore_service_storage/modules/celery/_task.py

@@ -82,7 +82,6 @@ def define_task(
     app: Celery,
     fn: Callable[Concatenate[AbortableTask, TaskId, P], Coroutine[Any, Any, R]],
     task_name: str | None = None,
-    task_queue: str | None = None,
 ) -> None: ...
 
 
@@ -91,7 +90,6 @@ def define_task(
     app: Celery,
     fn: Callable[Concatenate[AbortableTask, P], R],
     task_name: str | None = None,
-    task_queue: str | None = None,
 ) -> None: ...
 
 
@@ -102,7 +100,6 @@ def define_task(  # type: ignore[misc]
         | Callable[Concatenate[AbortableTask, P], R]
     ),
     task_name: str | None = None,
-    task_queue: str | None = None,
 ) -> None:
     """Decorator to define a celery task with error handling and abortable support"""
     wrapped_fn: Callable[Concatenate[AbortableTask, P], R]
@@ -117,5 +114,4 @@ def define_task(  # type: ignore[misc]
         name=task_name or fn.__name__,
         bind=True,
         base=AbortableTask,
-        queue=task_queue,
     )(wrapped_fn)

services/storage/src/simcore_service_storage/modules/celery/client.py

@@ -1,6 +1,7 @@
 import contextlib
 import logging
 from dataclasses import dataclass
+import queue
 from typing import Any, Final
 from uuid import uuid4
 
@@ -46,7 +47,7 @@ class CeleryTaskQueueClient:
     _task_store: TaskStore
 
     async def send_task(
-        self, task_name: str, *, task_context: TaskContext, **task_params
+        self, task_name: str, *, task_context: TaskContext, task_queue: str = "default", **task_params
     ) -> TaskUUID:
         with log_context(
             _logger,
@@ -55,7 +56,7 @@ async def send_task(
         ):
             task_uuid = uuid4()
             task_id = build_task_id(task_context, task_uuid)
-            self._celery_app.send_task(task_name, task_id=task_id, kwargs=task_params)
+            self._celery_app.send_task(task_name, task_id=task_id, kwargs=task_params, queue=task_queue)
             await self._task_store.set_task(
                 task_id, TaskData(status=TaskState.PENDING.name)
             )