remove usage of lock_context

Author: sanderegg

services/clusters-keeper/src/simcore_service_clusters_keeper/rpc/clusters.py

@@ -1,10 +1,13 @@
+import datetime
+
 from aws_library.ec2 import EC2InstanceData
 from aws_library.ec2._errors import EC2InstanceNotFoundError
 from fastapi import FastAPI
 from models_library.api_schemas_clusters_keeper.clusters import OnDemandCluster
 from models_library.users import UserID
 from models_library.wallets import WalletID
 from servicelib.rabbitmq import RPCRouter
+from servicelib.redis._decorators import exclusive
 
 from ..core.settings import get_application_settings
 from ..modules import clusters
@@ -29,11 +32,14 @@ async def get_or_create_cluster(
     redis = get_redis_client(app)
     dask_scheduler_ready = False
     cluster_auth = get_scheduler_auth(app)
-    async with redis.lock_context(
-        f"get_or_create_cluster-{user_id=}-{wallet_id=}",
+
+    @exclusive(
+        redis,
+        lock_key=f"get_or_create_cluster-{user_id=}-{wallet_id=}",
         blocking=True,
-        blocking_timeout_s=10,
-    ):
+        blocking_timeout=datetime.timedelta(seconds=10),
+    )
+    async def _locked_get_or_create_cluster() -> None:
         try:
             ec2_instance = await clusters.get_cluster(
                 app, user_id=user_id, wallet_id=wallet_id
@@ -52,6 +58,9 @@ async def get_or_create_cluster(
         )
         if dask_scheduler_ready:
             await clusters.cluster_heartbeat(app, user_id=user_id, wallet_id=wallet_id)
+
+    await _locked_get_or_create_cluster()
+
     assert ec2_instance is not None  # nosec
     app_settings = get_application_settings(app)
     assert app_settings.CLUSTERS_KEEPER_PRIMARY_EC2_INSTANCES  # nosec

services/efs-guardian/src/simcore_service_efs_guardian/services/process_messages.py

@@ -1,3 +1,4 @@
+import datetime
 import logging
 
 from fastapi import FastAPI
@@ -8,6 +9,7 @@
 from servicelib.rabbitmq.rpc_interfaces.dynamic_sidecar.disk_usage import (
     update_disk_usage,
 )
+from servicelib.redis._decorators import exclusive
 from servicelib.utils import fire_and_forget_task
 
 from ..core.settings import get_application_settings
@@ -77,13 +79,13 @@ async def process_dynamic_service_running_message(app: FastAPI, data: bytes) ->
         msg = f"Removing write permissions inside of EFS starts for project ID: {rabbit_message.project_id}, node ID: {rabbit_message.node_id}, current user: {rabbit_message.user_id}, size: {size}, upper limit: {settings.EFS_DEFAULT_USER_SERVICE_SIZE_BYTES}"
         with log_context(_logger, logging.WARNING, msg=msg):
             redis = get_redis_lock_client(app)
-            async with redis.lock_context(
-                f"efs_remove_write_permissions-{rabbit_message.project_id=}-{rabbit_message.node_id=}",
+            await exclusive(
+                redis,
+                lock_key=f"efs_remove_write_permissions-{rabbit_message.project_id=}-{rabbit_message.node_id=}",
                 blocking=True,
-                blocking_timeout_s=10,
-            ):
-                await efs_manager.remove_project_node_data_write_permissions(
-                    project_id=rabbit_message.project_id, node_id=rabbit_message.node_id
-                )
+                blocking_timeout=datetime.timedelta(seconds=10),
+            )(efs_manager.remove_project_node_data_write_permissions)(
+                project_id=rabbit_message.project_id, node_id=rabbit_message.node_id
+            )
 
     return True

services/web/server/src/simcore_service_webserver/projects/projects_api.py

@@ -1,10 +1,10 @@
-""" Interface to other subsystems
+"""Interface to other subsystems
 
-    - Data validation
-    - Operations on projects
-        - are NOT handlers, therefore do not return web.Response
-        - return data and successful HTTP responses (or raise them)
-        - upon failure raise errors that can be also HTTP reponses
+- Data validation
+- Operations on projects
+    - are NOT handlers, therefore do not return web.Response
+    - return data and successful HTTP responses (or raise them)
+    - upon failure raise errors that can be also HTTP reponses
 """
 
 import asyncio
@@ -81,6 +81,7 @@
     ServiceWaitingForManualInterventionError,
     ServiceWasNotFoundError,
 )
+from servicelib.redis._decorators import exclusive
 from servicelib.utils import fire_and_forget_task, logged_gather
 from simcore_postgres_database.models.users import UserRole
 from simcore_postgres_database.utils_projects_nodes import (
@@ -488,11 +489,7 @@ def _by_type_name(ec2: EC2InstanceTypeGet) -> bool:
 
     except KeyError as exc:
         raise InvalidKeysInResourcesSpecsError(missing_key=f"{exc}") from exc
-    except (
-        RemoteMethodNotRegisteredError,
-        RPCServerError,
-        asyncio.TimeoutError,
-    ) as exc:
+    except (TimeoutError, RemoteMethodNotRegisteredError, RPCServerError) as exc:
         raise ClustersKeeperNotAvailableError from exc
 
 
@@ -503,7 +500,6 @@ async def _check_project_node_has_all_required_inputs(
     project_uuid: ProjectID,
     node_id: NodeID,
 ) -> None:
-
     product_name = await db.get_project_product(project_uuid)
     await check_user_project_permission(
         app,
@@ -603,13 +599,17 @@ async def _start_dynamic_service(
     redis_client_sdk = get_redis_lock_manager_client_sdk(request.app)
     project_settings: ProjectsSettings = get_plugin_settings(request.app)
 
-    async with redis_client_sdk.lock_context(
-        lock_key,
+    @exclusive(
+        redis_client_sdk,
+        lock_key=lock_key,
         blocking=True,
-        blocking_timeout_s=_nodes_api.get_total_project_dynamic_nodes_creation_interval(
-            project_settings.PROJECTS_MAX_NUM_RUNNING_DYNAMIC_NODES
+        blocking_timeout=datetime.timedelta(
+            seconds=_nodes_api.get_total_project_dynamic_nodes_creation_interval(
+                project_settings.PROJECTS_MAX_NUM_RUNNING_DYNAMIC_NODES
+            )
         ),
-    ):
+    )
+    async def _() -> None:
         project_running_nodes = await dynamic_scheduler_api.list_dynamic_services(
             request.app, user_id=user_id, project_id=project_uuid
         )
@@ -759,6 +759,8 @@ async def _start_dynamic_service(
             ),
         )
 
+    await _()
+
 
 async def add_project_node(
     request: web.Request,