feat: New primitives (#42)

* feat: debuggable Event primitive

* feat: CounterLock primitive

* feat: add new primitives to main module

a_sync/__init__.py

@@ -3,7 +3,8 @@
 from a_sync.base import ASyncGenericBase
 from a_sync.decorator import a_sync
 from a_sync.modifiers.semaphores import (DummySemaphore, ThreadsafeSemaphore,
-                               apply_semaphore)
+                                         apply_semaphore)
+from a_sync.primitives import *
 from a_sync.singleton import ASyncGenericSingleton
 
 # I alias the aliases for your convenience.
@@ -12,5 +13,4 @@
 aka = alias = aliases
 
 # alias for backward-compatability, will be removed eventually, probably in 0.1.0
-ASyncBase = ASyncGenericBase    
-
+ASyncBase = ASyncGenericBase

a_sync/primitives/__init__.py

@@ -0,0 +1,6 @@
+
+"""
+While not the focus of this lib, this module includes some new primitives and some modified versions of standard asyncio primitives.
+"""
+
+from a_sync.primitives.locks import *

a_sync/primitives/_loggable.py

@@ -0,0 +1,9 @@
+
+from functools import cached_property
+from logging import Logger, getLogger
+
+
+class _Loggable:
+    @cached_property
+    def logger(self) -> Logger:
+        return getLogger(f"a_sync.{self.__class__.__name__}")

a_sync/primitives/locks/__init__.py

@@ -0,0 +1,3 @@
+
+from a_sync.primitives.locks.counter import CounterLock
+from a_sync.primitives.locks.event import Event

a_sync/primitives/locks/counter.py

@@ -0,0 +1,59 @@
+import asyncio
+from collections import defaultdict
+from typing import Iterable
+
+
+class CounterLock:
+    """
+    A asyncio primative that blocks until the internal counter has reached a specific value.
+    
+    counter = CounterLock()
+    A coroutine can now `await counter.wait_for(3)` and it will block until the internal counter >= 3.
+    Now if some other task executes `counter.value = 5` or `counter.set(5)`, the first coroutine will unblock as 5 >= 3.
+    
+    The internal counter can only increase.
+    """
+    def __init__(self, start_value: int = 0):
+        self._value = start_value
+        self._conditions = defaultdict(asyncio.Event)
+        self.is_ready = lambda v: self._value >= v
+
+    async def wait_for(self, value: int) -> bool:
+        if not self.is_ready(value):
+            await self._conditions[value].wait()
+        return True
+
+    def set(self, value: int) -> None:
+        self.value = value
+
+    @property
+    def value(self) -> int:
+        return self._value
+
+    @value.setter
+    def value(self, value: int) -> None:
+        if value > self._value:
+            self._value = value
+            ready = [
+                self._conditions.pop(key)
+                for key in list(self._conditions.keys())
+                if key <= self._value
+            ]
+            for event in ready:
+                event.set()
+        elif value < self._value:
+            raise ValueError("You cannot decrease the value.")
+
+class CounterLockCluster:
+    """
+    An asyncio primitive that represents 2 or more CounterLock objects.
+    
+    `wait_for(i)` will block until the value of all CounterLock objects is >= i.
+    """
+    def __init__(self, counter_locks: Iterable[CounterLock]) -> None:
+        self.locks = list(counter_locks)
+
+    async def wait_for(self, value: int) -> bool:
+        await asyncio.gather(*[counter_lock.wait_for(value) for counter_lock in self.locks])
+        return True
+

a_sync/primitives/locks/event.py

@@ -0,0 +1,23 @@
+
+import asyncio
+from a_sync.primitives._loggable import _Loggable
+
+class Event(asyncio.Event, _Loggable):
+    """asyncio.Event but with some additional debug logging to help detect deadlocks."""
+    def __init__(self):
+        self._task = None
+        self._counter = 0
+        super().__init__()
+
+    async def wait(self) -> bool:
+        if self.is_set():
+            return True
+        if self._task is None:
+            self._task = asyncio.create_task(self._debug_helper())
+        return await super().wait()
+
+    async def _debug_helper(self) -> None:
+        while not self.is_set():
+            self.logger.debug(f"Waiting for {self}")
+            await asyncio.sleep(5)
+        self._task = None