tests: lock

Author: phi-friday

src/tests/test_sync.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import threading
+import time
+
+import anyio
+import pytest
+
+import typed_diskcache
+
+pytestmark = pytest.mark.anyio
+
+
+def test_lock(cache):
+    state = {"num": 0}
+    lock = typed_diskcache.SyncLock(cache, "demo")
+
+    def worker() -> None:
+        state["num"] += 1
+        with lock:
+            assert lock.locked
+            state["num"] += 1
+            time.sleep(0.1)
+
+    with lock:
+        thread = threading.Thread(target=worker)
+        thread.start()
+        time.sleep(0.1)
+        assert state["num"] == 1
+    thread.join()
+    assert state["num"] == 2
+
+
+def test_rlock(cache):
+    state = {"num": 0}
+    rlock = typed_diskcache.SyncRLock(cache, "demo")
+
+    def worker() -> None:
+        state["num"] += 1
+        with rlock:
+            with rlock:
+                state["num"] += 1
+                time.sleep(0.1)
+
+    with rlock:
+        thread = threading.Thread(target=worker)
+        thread.start()
+        time.sleep(0.1)
+        assert state["num"] == 1
+    thread.join()
+    assert state["num"] == 2
+
+
+def test_semaphore(cache):
+    state = {"num": 0}
+    semaphore = typed_diskcache.SyncSemaphore(cache, "demo", value=3)
+
+    def worker() -> None:
+        state["num"] += 1
+        with semaphore:
+            state["num"] += 1
+            time.sleep(0.1)
+
+    semaphore.acquire()
+    semaphore.acquire()
+    with semaphore:
+        thread = threading.Thread(target=worker)
+        thread.start()
+        time.sleep(0.1)
+        assert state["num"] == 1
+    thread.join()
+    assert state["num"] == 2
+    semaphore.release()
+    semaphore.release()
+
+
+async def test_async_lock(cache):
+    state = {"num": 0}
+    lock = typed_diskcache.AsyncLock(cache, "demo")
+
+    async def worker() -> None:
+        state["num"] += 1
+        async with lock:
+            assert lock.locked
+            state["num"] += 1
+            await anyio.sleep(0.1)
+
+    async with lock:
+        thread = threading.Thread(target=anyio.run, args=(worker,))
+        thread.start()
+        await anyio.sleep(0.1)
+        assert state["num"] == 1
+    thread.join()
+    assert state["num"] == 2
+
+
+@pytest.mark.only
+async def test_async_rlock(cache):
+    state = {"num": 0}
+    rlock = typed_diskcache.AsyncRLock(cache, "demo")
+
+    async def worker() -> None:
+        state["num"] += 1
+        async with rlock:
+            async with rlock:
+                state["num"] += 1
+                await anyio.sleep(0.1)
+
+    async with rlock:
+        thread = threading.Thread(target=anyio.run, args=(worker,))
+        thread.start()
+        await anyio.sleep(0.1)
+        assert state["num"] == 1
+    thread.join()
+    assert state["num"] == 2
+
+
+async def test_async_semaphore(cache):
+    state = {"num": 0}
+    semaphore = typed_diskcache.AsyncSemaphore(cache, "demo", value=3)
+
+    async def worker() -> None:
+        state["num"] += 1
+        async with semaphore:
+            state["num"] += 1
+            await anyio.sleep(0.1)
+
+    await semaphore.acquire()
+    await semaphore.acquire()
+    async with semaphore:
+        thread = threading.Thread(target=anyio.run, args=(worker,))
+        thread.start()
+        await anyio.sleep(0.1)
+        assert state["num"] == 1
+    thread.join()
+    assert state["num"] == 2
+    await semaphore.release()
+    await semaphore.release()

src/typed_diskcache/implement/sync/lock.py

@@ -108,7 +108,7 @@ def acquire(self) -> None:
                 self.key, None, expire=self.expire, tags=self.tags, retry=True
             )
             if added:
-                break
+                return
             time.sleep(SPIN_LOCK_SLEEP)
             timeout = time.monotonic() - start
 
@@ -179,7 +179,11 @@ def acquire(self) -> None:
                     self._cache.get, self.key, default=("default", 0)
                 )
                 container_value = validate_lock_value(container.value)
-                if container.default or pid_tid == container_value[0]:
+                if (
+                    container.default
+                    or pid_tid == container_value[0]
+                    or container_value[1] <= 0
+                ):
                     value = 1 if container.default else container_value[1] + 1
                     logger.debug("acquired lock: %s, value: %d", pid_tid, value)
                     context.run(
@@ -393,7 +397,11 @@ async def acquire(self) -> None:
                         self._cache.aget, self.key, default=("default", 0)
                     )
                     container_value = validate_lock_value(container.value)
-                    if container.default or pid_tid == container_value[0]:
+                    if (
+                        container.default
+                        or pid_tid == container_value[0]
+                        or container_value[1] <= 0
+                    ):
                         value = 1 if container.default else container_value[1] + 1
                         logger.debug("acquired lock: %s, value: %d", pid_tid, value)
                         await context.run(