removed useless code

Author: sanderegg

packages/service-library/src/servicelib/background_task.py

@@ -3,15 +3,14 @@
 import datetime
 import functools
 import logging
-from collections.abc import AsyncIterator, Awaitable, Callable
-from typing import Any, Coroutine, Final, ParamSpec, TypeVar
+from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
+from typing import Any, Final, ParamSpec, TypeVar
 
 from tenacity import TryAgain, before_sleep_log, retry, retry_if_exception_type
-from tenacity.asyncio import AsyncRetrying
 from tenacity.wait import wait_fixed
 
 from .async_utils import retried_cancel_task, with_delay
-from .logging_utils import log_catch, log_context
+from .logging_utils import log_context
 
 _logger = logging.getLogger(__name__)
 
@@ -43,6 +42,21 @@ def periodic(
 ) -> Callable[
     [Callable[P, Coroutine[Any, Any, None]]], Callable[P, Coroutine[Any, Any, None]]
 ]:
+    """Calls the function periodically with a given interval.
+
+    Arguments:
+        interval -- the interval between calls
+
+    Keyword Arguments:
+        raise_on_error -- If False the function will be retried indefinitely unless cancelled.
+                          If True the function will be retried indefinitely unless cancelled
+                          or an exception is raised. (default: {False})
+        early_wake_up_event -- allows to awaken the function before the interval has passed. (default: {None})
+
+    Returns:
+        _description_
+    """
+
     def _decorator(
         func: Callable[P, Coroutine[Any, Any, None]],
     ) -> Callable[P, Coroutine[Any, Any, None]]:
@@ -71,50 +85,6 @@ async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> None:
     return _decorator
 
 
-async def _periodic_scheduled_task(
-    task: Callable[..., Awaitable[None]],
-    *,
-    interval: datetime.timedelta,
-    task_name: str,
-    raise_on_error: bool,
-    early_wake_up_event: asyncio.Event | None,
-    **task_kwargs,
-) -> None:
-    """periodically runs task with a given interval.
-    If raise_on_error is False, the task will be retried indefinitely unless cancelled.
-    If raise_on_error is True, the task will be retried indefinitely unless cancelled or an exception is raised.
-    If early_wake_up_event is set, the task might be woken up earlier than interval when the event is set.
-
-    Raises:
-        task exception if raise_on_error is True
-    """
-    nap = (
-        asyncio.sleep
-        if early_wake_up_event is None
-        else SleepUsingAsyncioEvent(early_wake_up_event)
-    )
-    async for attempt in AsyncRetrying(
-        sleep=nap,
-        wait=wait_fixed(interval.total_seconds()),
-        reraise=True,
-        retry=retry_if_exception_type(TryAgain)
-        if raise_on_error
-        else retry_if_exception_type(),
-    ):
-        with attempt:
-            with (
-                log_context(
-                    _logger,
-                    logging.DEBUG,
-                    msg=f"iteration {attempt.retry_state.attempt_number} of '{task_name}'",
-                ),
-                log_catch(_logger),
-            ):
-                await task(**task_kwargs)
-
-            raise TryAgain
-
-
 def create_periodic_task(
     task: Callable[..., Awaitable[None]],
     *,
@@ -125,34 +95,19 @@ def create_periodic_task(
     early_wake_up_event: asyncio.Event | None = None,
     **kwargs,
 ) -> asyncio.Task:
-    with log_context(
-        _logger, logging.DEBUG, msg=f"create periodic background task '{task_name}'"
-    ):
-        delayed_periodic_scheduled_task = with_delay(wait_before_running)(
-            _periodic_scheduled_task
-        )
-        return asyncio.create_task(
-            delayed_periodic_scheduled_task(
-                task,
-                interval=interval,
-                task_name=task_name,
-                raise_on_error=raise_on_error,
-                early_wake_up_event=early_wake_up_event,
-                **kwargs,
-            ),
-            name=task_name,
-        )
-
+    @with_delay(wait_before_running)
+    @periodic(
+        interval=interval,
+        raise_on_error=raise_on_error,
+        early_wake_up_event=early_wake_up_event,
+    )
+    async def _() -> None:
+        await task(**kwargs)
 
-async def stop_periodic_task(
-    asyncio_task: asyncio.Task, *, timeout: float | None = None
-) -> None:
     with log_context(
-        _logger,
-        logging.DEBUG,
-        msg=f"cancel periodic background task '{asyncio_task.get_name()}'",
+        _logger, logging.DEBUG, msg=f"create periodic background task '{task_name}'"
     ):
-        await retried_cancel_task(asyncio_task, timeout=timeout)
+        return asyncio.create_task(_(), name=task_name)
 
 
 @contextlib.asynccontextmanager
@@ -179,4 +134,6 @@ async def periodic_task(
         if asyncio_task is not None:
             # NOTE: this stopping is shielded to prevent the cancellation to propagate
             # into the stopping procedure
-            await asyncio.shield(stop_periodic_task(asyncio_task, timeout=stop_timeout))
+            await asyncio.shield(
+                retried_cancel_task(asyncio_task, timeout=stop_timeout)
+            )

packages/service-library/tests/redis/test_decorators.py

@@ -11,7 +11,7 @@
 import arrow
 import pytest
 from faker import Faker
-from servicelib.background_task import stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
 from servicelib.redis import (
     CouldNotAcquireLockError,
     RedisClientSDK,
@@ -217,7 +217,7 @@ async def _assert_task_completes_once(
 
         await _assert_on_sleep_done(sleep_events, stop_after=stop_after)
 
-        await stop_periodic_task(started_task, timeout=5)
+        await retried_cancel_task(started_task, timeout=5)
 
         events_timestamps: tuple[float, ...] = tuple(
             x.args[0].timestamp() for x in sleep_events.call_args_list

packages/service-library/tests/test_background_task.py

@@ -14,11 +14,11 @@
 import pytest
 from faker import Faker
 from pytest_mock.plugin import MockerFixture
+from servicelib.async_utils import retried_cancel_task
 from servicelib.background_task import (  # Assuming the module is imported correctly
     create_periodic_task,
     periodic,
     periodic_task,
-    stop_periodic_task,
 )
 
 _FAST_POLL_INTERVAL: Final[int] = 1
@@ -73,7 +73,7 @@ async def _creator(
     yield _creator
     # cleanup
     await asyncio.gather(
-        *(stop_periodic_task(t, timeout=stop_task_timeout) for t in created_tasks)
+        *(retried_cancel_task(t, timeout=stop_task_timeout) for t in created_tasks)
     )
 
 

services/agent/src/simcore_service_agent/services/volumes_manager.py

@@ -9,7 +9,8 @@
 from fastapi import FastAPI
 from models_library.projects_nodes_io import NodeID
 from pydantic import NonNegativeFloat
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.logging_utils import log_context
 from servicelib.rabbitmq.rpc_interfaces.agent.errors import (
@@ -60,10 +61,10 @@ async def shutdown(self) -> None:
         await self.docker.close()
 
         if self._task_bookkeeping:
-            await stop_periodic_task(self._task_bookkeeping)
+            await retried_cancel_task(self._task_bookkeeping)
 
         if self._task_periodic_volume_cleanup:
-            await stop_periodic_task(self._task_periodic_volume_cleanup)
+            await retried_cancel_task(self._task_periodic_volume_cleanup)
 
     async def _bookkeeping_task(self) -> None:
         with log_context(_logger, logging.DEBUG, "volume bookkeeping"):

services/api-server/src/simcore_service_api_server/core/_prometheus_instrumentation.py

@@ -6,7 +6,8 @@
 from fastapi import FastAPI
 from prometheus_client import CollectorRegistry, Gauge
 from pydantic import PositiveInt
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.prometheus_instrumentation import (
     setup_prometheus_instrumentation as setup_rest_instrumentation,
 )
@@ -79,7 +80,7 @@ async def on_startup() -> None:
     async def on_shutdown() -> None:
         assert app.state.instrumentation_task  # nosec
         with log_catch(_logger, reraise=False):
-            await stop_periodic_task(app.state.instrumentation_task)
+            await retried_cancel_task(app.state.instrumentation_task)
 
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)

services/api-server/src/simcore_service_api_server/core/health_checker.py

@@ -9,7 +9,8 @@
 from models_library.rabbitmq_messages import LoggerRabbitMessage
 from models_library.users import UserID
 from pydantic import NonNegativeInt, PositiveFloat, PositiveInt
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.dependencies import get_app
 from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RabbitMQClient
@@ -62,7 +63,7 @@ async def setup(self, health_check_task_period_seconds: PositiveFloat):
     async def teardown(self):
         if self._background_task:
             with log_catch(_logger, reraise=False):
-                await stop_periodic_task(
+                await retried_cancel_task(
                     self._background_task, timeout=self._timeout_seconds
                 )
         await self._log_distributor.deregister(job_id=self._dummy_job_id)

services/autoscaling/src/simcore_service_autoscaling/modules/auto_scaling_task.py

@@ -3,7 +3,8 @@
 from typing import Final
 
 from fastapi import FastAPI
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.redis import exclusive
 
 from ..core.settings import ApplicationSettings
@@ -43,7 +44,7 @@ async def _startup() -> None:
 
 def on_app_shutdown(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _stop() -> None:
-        await stop_periodic_task(app.state.autoscaler_task)
+        await retried_cancel_task(app.state.autoscaler_task)
 
     return _stop
 

services/autoscaling/src/simcore_service_autoscaling/modules/buffer_machines_pool_task.py

@@ -3,7 +3,8 @@
 from typing import Final
 
 from fastapi import FastAPI
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.redis import exclusive
 
 from ..core.settings import ApplicationSettings
@@ -43,7 +44,7 @@ async def _startup() -> None:
 def on_app_shutdown(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _stop() -> None:
         if hasattr(app.state, "buffers_pool_task"):
-            await stop_periodic_task(app.state.buffers_pool_task)
+            await retried_cancel_task(app.state.buffers_pool_task)
 
     return _stop
 

services/clusters-keeper/src/simcore_service_clusters_keeper/modules/clusters_management_task.py

@@ -3,7 +3,8 @@
 from collections.abc import Awaitable, Callable
 
 from fastapi import FastAPI
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.redis import exclusive
 
 from .._meta import APP_NAME
@@ -36,7 +37,7 @@ async def _startup() -> None:
 
 def on_app_shutdown(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _stop() -> None:
-        await stop_periodic_task(app.state.clusters_cleaning_task, timeout=5)
+        await retried_cancel_task(app.state.clusters_cleaning_task, timeout=5)
 
     return _stop
 

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_manager.py

@@ -6,7 +6,8 @@
 from fastapi import FastAPI
 from models_library.projects import ProjectID
 from models_library.users import UserID
-from servicelib.background_task import create_periodic_task, stop_periodic_task
+from servicelib.async_utils import retried_cancel_task
+from servicelib.background_task import create_periodic_task
 from servicelib.exception_utils import silence_exceptions
 from servicelib.logging_utils import log_context
 from servicelib.redis import CouldNotAcquireLockError, exclusive
@@ -164,4 +165,4 @@ async def setup_manager(app: FastAPI) -> None:
 
 
 async def shutdown_manager(app: FastAPI) -> None:
-    await stop_periodic_task(app.state.scheduler_manager)
+    await retried_cancel_task(app.state.scheduler_manager)