fix worker startup

Author: giancarloromeo

services/storage/docker/boot.sh

@@ -56,10 +56,10 @@ if [ "${SC_BOOT_MODE}" = "debug" ]; then
       --log-level \"${SERVER_LOG_LEVEL}\"
   "
 else
-  if [ "${STORAGE_WORKER_MODE}" = "1" ]; then
+  if [ "${STORAGE_WORKER_MODE}" = "true" ]; then
     exec celery \
       --app=simcore_service_storage.modules.celery.worker_main:app \
-      worker --pool=threads \
+      worker \
       --loglevel="${SERVER_LOG_LEVEL}" \
       --hostname="${HOSTNAME}"
   else

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

@@ -3,6 +3,8 @@
 from celery import Celery
 from fastapi import FastAPI
 
+from ...core.settings import ApplicationSettings
+from ._common import create_app
 from .client import CeleryTaskQueueClient
 from .worker import CeleryTaskQueueWorker
 
@@ -11,6 +13,12 @@
 _EVENT_LOOP_KEY = "loop"
 
 
+def create_celery_app_worker(settings: ApplicationSettings) -> Celery:
+    celery_app = create_app(settings)
+    celery_app.conf[_WORKER_KEY] = CeleryTaskQueueWorker(celery_app)
+    return celery_app
+
+
 def get_celery_app(fastapi: FastAPI) -> Celery:
     celery = fastapi.state.celery_app
     assert isinstance(celery, Celery)
@@ -41,10 +49,6 @@ def get_celery_worker(celery_app: Celery) -> CeleryTaskQueueWorker:
     return worker
 
 
-def set_celery_worker(celery_app: Celery, celery_worker: CeleryTaskQueueWorker) -> None:
-    celery_app.conf[_WORKER_KEY] = celery_worker
-
-
 def get_event_loop(celery_app: Celery) -> AbstractEventLoop:  # nosec
     loop = celery_app.conf[_EVENT_LOOP_KEY]
     assert isinstance(loop, AbstractEventLoop)

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

@@ -12,9 +12,7 @@
 
 from ...core.application import create_app
 from ...core.settings import ApplicationSettings
-from ._common import create_app as create_celery_app
-from ._utils import set_celery_worker
-from .worker import CeleryTaskQueueWorker
+from ._utils import create_celery_app_worker
 
 _settings = ApplicationSettings.create_from_envs()
 
@@ -77,9 +75,4 @@ async def shutdown():
     asyncio.run_coroutine_threadsafe(shutdown(), loop)
 
 
-celery_app = create_celery_app(_settings)
-
-celery_worker = CeleryTaskQueueWorker(celery_app)
-set_celery_worker(celery_app, CeleryTaskQueueWorker(celery_app))
-
-app = celery_app
+app = create_celery_app_worker(_settings)