🎨 Refactor garbage collection task: Update to use new background task creation method

Author: pcrespov

services/web/server/src/simcore_service_webserver/garbage_collector/_tasks_core.py

@@ -1,103 +1,59 @@
-""" Setup and running of periodic background task
+"""Setup and running of periodic background task
 
 
 Specifics of the gc implementation should go into garbage_collector_core.py
 """
 
 import asyncio
 import logging
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncIterator, Callable
+from datetime import timedelta
 
 from aiohttp import web
+from servicelib.async_utils import cancel_wait_task
+from servicelib.background_task_utils import exclusive_periodic
 from servicelib.logging_utils import log_context
+from simcore_service_webserver.redis import get_redis_lock_manager_client_sdk
 
 from ._core import collect_garbage
 from .settings import GarbageCollectorSettings, get_plugin_settings
 
 _logger = logging.getLogger(__name__)
 
+CleanupContextFunc = Callable[[web.Application], AsyncIterator[None]]
 
-_GC_TASK_NAME = f"background-task.{__name__}.collect_garbage_periodically"
-_GC_TASK_CONFIG = f"{_GC_TASK_NAME}.config"
-_GC_TASK = f"{_GC_TASK_NAME}.task"
 
+def create_background_task_for_garbage_collection() -> CleanupContextFunc:
 
-async def run_background_task(app: web.Application) -> AsyncGenerator:
-    # SETUP ------
-    # create a background task to collect garbage periodically
-    assert not any(  # nosec
-        t.get_name() == _GC_TASK_NAME for t in asyncio.all_tasks()
-    ), "Garbage collector task already running. ONLY ONE expected"  # nosec
+    async def _cleanup_ctx_fun(app: web.Application) -> AsyncIterator[None]:
+        settings: GarbageCollectorSettings = get_plugin_settings(app)
 
-    gc_bg_task = asyncio.create_task(
-        _collect_garbage_periodically(app), name=_GC_TASK_NAME
-    )
-    # attaches variable to the app's lifetime
-    app[_GC_TASK] = gc_bg_task
+        @exclusive_periodic(
+            # Function-exclusiveness is required to avoid multiple tasks like thisone running concurrently
+            get_redis_lock_manager_client_sdk(app),
+            task_interval=timedelta(seconds=settings.GARBAGE_COLLECTOR_INTERVAL_S),
+            retry_after=timedelta(minutes=5),
+        )
+        async def _collect_garbage_periodically() -> None:
+            with log_context(_logger, logging.INFO, "Garbage collect cycle"):
+                await collect_garbage(app)
 
-    # FIXME: added this config to overcome the state in which the
-    # task cancelation is ignored and the exceptions enter in a loop
-    # that never stops the background task. This flag is an additional
-    # mechanism to enforce stopping the background task
-    #
-    # Implemented with a mutable dict to avoid
-    #   DeprecationWarning: Changing state of started or joined application is deprecated
-    #
-    app[_GC_TASK_CONFIG] = {"force_stop": False, "name": _GC_TASK_NAME}
+        # setup
+        task_name = _collect_garbage_periodically.__name__
 
-    yield
+        task = asyncio.create_task(
+            _collect_garbage_periodically(),
+            name=task_name,
+        )
 
-    # TEAR-DOWN -----
-    # controlled cancelation of the gc task
-    try:
-        _logger.info("Stopping garbage collector...")
+        # prevents premature garbage collection of the task
+        app_task_key = f"tasks.{task_name}"
+        app[app_task_key] = task
 
-        ack = gc_bg_task.cancel()
-        assert ack  # nosec
+        yield
 
-        app[_GC_TASK_CONFIG]["force_stop"] = True
+        # tear-down
+        await cancel_wait_task(task)
+        app.pop(app_task_key, None)
 
-        await gc_bg_task
-
-    except asyncio.CancelledError:
-        assert gc_bg_task.cancelled()  # nosec
-
-
-async def _collect_garbage_periodically(app: web.Application):
-    settings: GarbageCollectorSettings = get_plugin_settings(app)
-    interval = settings.GARBAGE_COLLECTOR_INTERVAL_S
-
-    while True:
-        try:
-            while True:
-                with log_context(_logger, logging.INFO, "Garbage collect cycle"):
-                    await collect_garbage(app)
-
-                    if app[_GC_TASK_CONFIG].get("force_stop", False):
-                        msg = "Forced to stop garbage collection"
-                        raise RuntimeError(msg)
-
-                _logger.info("Garbage collect cycle pauses %ss", interval)
-                await asyncio.sleep(interval)
-
-        except asyncio.CancelledError:  # EXIT  # noqa: PERF203
-            _logger.info(
-                "Stopped: Garbage collection task was cancelled, it will not restart!"
-            )
-            # do not catch Cancellation errors
-            raise
-
-        except Exception:  # RESILIENT restart # pylint: disable=broad-except
-            _logger.warning(
-                "Stopped: There was an error during garbage collection, restarting...",
-                exc_info=True,
-            )
-
-            if app[_GC_TASK_CONFIG].get("force_stop", False):
-                _logger.warning("Forced to stop garbage collection")
-                break
-
-            # will wait 5 seconds to recover before restarting to avoid restart loops
-            # - it might be that db/redis is down, etc
-            #
-            await asyncio.sleep(5)
+    return _cleanup_ctx_fun

services/web/server/src/simcore_service_webserver/garbage_collector/plugin.py

@@ -34,7 +34,7 @@ def setup_garbage_collector(app: web.Application) -> None:
 
     settings = get_plugin_settings(app)
 
-    app.cleanup_ctx.append(_tasks_core.run_background_task)
+    app.cleanup_ctx.append(_tasks_core.create_background_task_for_garbage_collection())
 
     set_parent_module_log_level(
         _logger.name, min(logging.INFO, get_application_settings(app).log_level)