this now works

Author: sanderegg

packages/common-library/src/common_library/async_tools.py

@@ -111,8 +111,6 @@ async def cancel_wait_task(
             # owner function is being cancelled -> propagate cancellation
             raise
 
-        # else: task cancellation is complete, we can safely ignore it
-
 
 def delayed_start(
     delay: datetime.timedelta,

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

@@ -177,7 +177,7 @@ async def acquire(self) -> bool:
                 else stop_never
             ),
             retry=retry_if_not_result(lambda acquired: acquired),
-            before_sleep=before_sleep_log(_logger, logging.INFO),
+            before_sleep=before_sleep_log(_logger, logging.DEBUG),
         )
         async def _blocking_acquire() -> bool:
             return await self._try_acquire()
@@ -287,23 +287,31 @@ async def reacquire(self) -> None:
 
         # Lua script returns: 'renewed' or status message
         if status == "renewed":
+            assert success == 1  # nosec
             _logger.debug(
-                "Renewed semaphore '%s' (instance: %s)",
+                "Renewed semaphore '%s' (instance: %s, count: %s, expired: %s)",
                 self.key,
                 self.instance_id,
+                current_count,
+                expired_count,
             )
         else:
+            assert success == 0  # nosec
             if status == "expired":
                 _logger.warning(
-                    "Semaphore '%s' holder key expired for instance %s",
+                    "Semaphore '%s' holder key expired (instance: %s, count: %s, expired: %s)",
                     self.key,
                     self.instance_id,
+                    current_count,
+                    expired_count,
                 )
             elif status == "not_held":
                 _logger.warning(
-                    "Semaphore '%s' not held by instance %s",
+                    "Semaphore '%s' not held (instance: %s, count: %s, expired: %s)",
                     self.key,
                     self.instance_id,
+                    current_count,
+                    expired_count,
                 )
 
             raise SemaphoreLostError(name=self.key, instance_id=self.instance_id)

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

@@ -6,6 +6,8 @@
 from collections.abc import Callable, Coroutine
 from typing import Any, ParamSpec, TypeVar
 
+from common_library.async_tools import cancel_wait_task
+
 from ..background_task import periodic
 from ..logging_errors import create_troubleshootting_log_kwargs
 from ._client import RedisClientSDK
@@ -113,7 +115,8 @@ async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
                     @periodic(interval=ttl / 3, raise_on_error=True)
                     async def _periodic_renewer() -> None:
                         await semaphore.reacquire()
-                        started_event.set()
+                        if not started_event.is_set():
+                            started_event.set()
 
                     # Start the renewal task
                     renewal_task = tg.create_task(
@@ -129,11 +132,10 @@ async def _periodic_renewer() -> None:
                         coro(*args, **kwargs),
                         name=f"semaphore/work_{coro.__module__}.{coro.__name__}",
                     )
-
                     result = await work_task
 
                     # Cancel renewal task (work is done)
-                    renewal_task.cancel()
+                    await cancel_wait_task(renewal_task, max_delay=None)
 
                 return result
 

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

@@ -351,16 +351,15 @@ async def function_raising_cancelled_error():
         await function_raising_cancelled_error()
 
 
-# @pytest.mark.skip
 async def test_with_large_capacity(
     redis_client_sdk: RedisClientSDK,
     semaphore_name: str,
 ):
     large_capacity = 100
     concurrent_count = 0
     max_concurrent = 0
-    sleep_time_s = 10
-    num_tasks = 400
+    sleep_time_s = 5
+    num_tasks = 1000
 
     @with_limited_concurrency(
         redis_client_sdk,
@@ -380,10 +379,12 @@ async def limited_function() -> None:
 
     # Start tasks equal to the large capacity
     tasks = [asyncio.create_task(limited_function()) for _ in range(num_tasks)]
-    async with asyncio.timeout(
-        float(num_tasks) / float(large_capacity) * 10.0 * float(sleep_time_s)
-    ):
-        await asyncio.gather(*tasks)
+    done, pending = await asyncio.wait(
+        tasks,
+        timeout=float(num_tasks) / float(large_capacity) * 10.0 * float(sleep_time_s),
+    )
+    assert not pending, f"Some tasks did not complete: {len(pending)} pending"
+    assert len(done) == num_tasks
 
     # Should never exceed the large capacity
     assert max_concurrent <= large_capacity