added argument for expected overall time

Author: sanderegg

packages/service-library/src/servicelib/redis/_semaphore_decorator.py

@@ -38,6 +38,7 @@ async def _managed_semaphore_execution(
     semaphore_key: str,
     ttl: datetime.timedelta,
     execution_context: str,
+    expected_lock_overall_time: datetime.timedelta,
 ) -> AsyncIterator:
     """Common semaphore management logic with auto-renewal."""
     # Acquire the semaphore first
@@ -106,14 +107,14 @@ async def _periodic_renewer() -> None:
         finally:
             lock_release_time = arrow.utcnow()
             locking_time = lock_release_time - lock_acquisition_time
-            if locking_time > DEFAULT_EXPECTED_LOCK_OVERALL_TIME:
+            if locking_time > expected_lock_overall_time:
                 _logger.warning(
                     "Semaphore '%s' was held for %s which is longer than expected (%s). "
                     "TIP: consider reducing the locking time by optimizing the code inside "
                     "the critical section or increasing the default locking time",
                     semaphore_key,
                     locking_time,
-                    DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
+                    expected_lock_overall_time,
                 )
 
 
@@ -157,6 +158,7 @@ def with_limited_concurrency(
     ttl: datetime.timedelta = DEFAULT_SEMAPHORE_TTL,
     blocking: bool = True,
     blocking_timeout: datetime.timedelta | None = DEFAULT_SOCKET_TIMEOUT,
+    expected_lock_overall_time: datetime.timedelta = DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
 ) -> Callable[
     [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]
 ]:
@@ -174,6 +176,7 @@ def with_limited_concurrency(
         ttl: Time-to-live for semaphore entries (default: 5 minutes)
         blocking: Whether to block when semaphore is full (default: True)
         blocking_timeout: Maximum time to wait when blocking (default: socket timeout)
+        expected_lock_overall_time: helper for logging warnings if lock is held longer than expected
 
     Example:
         @with_limited_concurrency(
@@ -209,7 +212,11 @@ async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
             )
 
             async with _managed_semaphore_execution(
-                semaphore, semaphore_key, ttl, f"coroutine_{coro.__name__}"
+                semaphore,
+                semaphore_key,
+                ttl,
+                f"coroutine_{coro.__name__}",
+                expected_lock_overall_time,
             ):
                 return await coro(*args, **kwargs)
 
@@ -226,6 +233,7 @@ def with_limited_concurrency_cm(
     ttl: datetime.timedelta = DEFAULT_SEMAPHORE_TTL,
     blocking: bool = True,
     blocking_timeout: datetime.timedelta | None = DEFAULT_SOCKET_TIMEOUT,
+    expected_lock_overall_time: datetime.timedelta = DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
 ) -> Callable[
     [Callable[P, AbstractAsyncContextManager[R]]],
     Callable[P, AbstractAsyncContextManager[R]],
@@ -244,6 +252,7 @@ def with_limited_concurrency_cm(
         ttl: Time-to-live for semaphore entries (default: 5 minutes)
         blocking: Whether to block when semaphore is full (default: True)
         blocking_timeout: Maximum time to wait when blocking (default: socket timeout)
+        expected_lock_overall_time: helper for logging warnings if lock is held longer than expected
 
     Example:
         @asynccontextmanager
@@ -281,7 +290,11 @@ async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[R]:
 
             async with (
                 _managed_semaphore_execution(
-                    semaphore, semaphore_key, ttl, f"context_manager_{cm_func.__name__}"
+                    semaphore,
+                    semaphore_key,
+                    ttl,
+                    f"context_manager_{cm_func.__name__}",
+                    expected_lock_overall_time,
                 ),
                 cm_func(*args, **kwargs) as value,
             ):

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

@@ -13,6 +13,7 @@
 
 import pytest
 from pytest_mock import MockerFixture
+from pytest_simcore.helpers.logging_tools import log_context
 from servicelib.redis import RedisClientSDK
 from servicelib.redis._constants import (
     SEMAPHORE_HOLDER_KEY_PREFIX,
@@ -365,11 +366,11 @@ async def test_with_large_capacity(
     redis_client_sdk: RedisClientSDK,
     semaphore_name: str,
 ):
-    large_capacity = 100
+    large_capacity = 1
     concurrent_count = 0
     max_concurrent = 0
-    sleep_time_s = 5
-    num_tasks = 1000
+    sleep_time_s = 50
+    num_tasks = 10
 
     @with_limited_concurrency(
         redis_client_sdk,
@@ -382,9 +383,8 @@ async def limited_function() -> None:
         nonlocal concurrent_count, max_concurrent
         concurrent_count += 1
         max_concurrent = max(max_concurrent, concurrent_count)
-        logging.info("Started task, current concurrent: %d", concurrent_count)
-        await asyncio.sleep(sleep_time_s)
-        logging.info("Done task, current concurrent: %d", concurrent_count)
+        with log_context(logging.INFO, f"task with {concurrent_count=}"):
+            await asyncio.sleep(sleep_time_s)
         concurrent_count -= 1
 
     # Start tasks equal to the large capacity
@@ -400,6 +400,30 @@ async def limited_function() -> None:
     assert max_concurrent <= large_capacity
 
 
+async def test_long_locking_logs_warning(
+    redis_client_sdk: RedisClientSDK,
+    semaphore_name: str,
+    caplog: pytest.LogCaptureFixture,
+    mocker: MockerFixture,
+):
+    @with_limited_concurrency(
+        redis_client_sdk,
+        key=semaphore_name,
+        capacity=1,
+        blocking=True,
+        blocking_timeout=None,
+        expected_lock_overall_time=datetime.timedelta(milliseconds=200),
+    )
+    async def limited_function() -> None:
+        with log_context(logging.INFO, "task"):
+            await asyncio.sleep(0.4)
+
+    with caplog.at_level(logging.WARNING):
+        await limited_function()
+        assert caplog.records
+        assert "longer than expected" in caplog.messages[-1]
+
+
 async def test_context_manager_basic_functionality(
     redis_client_sdk: RedisClientSDK,
     semaphore_name: str,