Lock

Author: joente

thingsdb/misc/README.md

@@ -0,0 +1,69 @@
+# Lock
+
+This lock provides distributed mutual exclusion, allowing you to synchronize
+access to shared resources or critical sections of code across multiple
+independent Python programs or services, even if they are running on different
+machines.
+
+It functions similarly to `asyncio.Lock()`, which is designed for concurrency
+within a single process, but extends this capability to a multi-process,
+multi-host environment by leveraging ThingsDB as its backend. This ensures that
+only one client can acquire the lock at any given time, preventing race
+conditions and maintaining data integrity in a distributed system.
+
+The `timeout` parameter defines the maximum duration a lock can be held.
+If a client fails to explicitly release the lock (e.g., due to a crash),
+ThingsDB will automatically release it after this period, preventing deadlocks.
+Separately, the expression `queue_size * timeout` indicates the total maximum
+time a client will actively attempt to acquire the lock if it's currently
+unavailable.
+
+Example code:
+
+```python
+import asyncio
+from functools import partial
+from thingsdb.client import Client
+from thingsdb.misc import lock
+
+
+async def main():
+    # ThingsDB client
+    client = Client()
+
+    # Multiple locks may be created, make sure you give each lock a unique name
+    mylock = partial(lock.lock, client=client, name='my-lock', timeout=5)
+
+    await client.connect('localhost')
+    try:
+        await client.authenticate('admin', 'pass')
+
+        # This will set-up a lock collection
+        # It will only do work the first time, but no harm in keep calling
+        await lock.setup(client)
+
+        # Wait for a lock
+        async with mylock():
+            print('In here')
+            await asyncio.sleep(5.0)  # simulate some work
+            print('Done here')
+
+    finally:
+        await client.close_and_wait()
+
+
+if __name__ == '__main__':
+    asyncio.run(main())
+```
+
+To observe the distributed lock in action, you can execute the example Python
+script simultaneously in multiple separate terminal windows.
+
+You can determine if a specific distributed lock is currently held by using
+the `lock.locked()` asynchronous function.
+
+To check the lock's status:
+
+```python
+is_locked = await lock.locked(client, 'my-lock')
+```

thingsdb/misc/__init__.py

@@ -1 +0,0 @@
-from .lock import lock as ti_lock

thingsdb/misc/lock.py

@@ -1,5 +1,6 @@
 import asyncio
 from contextlib import asynccontextmanager
+from typing import AsyncGenerator
 from ..client import Client
 from ..room import Room, event
 
@@ -42,10 +43,12 @@ async def setup(client: Client, collection: str = 'lock'):
                 inner.room.emit('go');
             },
             acquire: |this, timeout| {
-                immediately = this.queue.len() == 0;
+                size = this.queue.len();
+                immediately = size == 0;
                 inner = Inner{timeout:,};
                 this.queue.push(inner);
-                immediately ? inner.set_task(this.id()) : inner.room.id();
+                immediately && inner.set_task(this.id());
+                [immediately, inner.room.id(), size];
             },
             release: |this, room_id| try({
                 if (this.queue.first().room.id() == room_id) {
@@ -74,12 +77,16 @@ async def setup(client: Client, collection: str = 'lock'):
             room(room_id).name() == 'go';
         });
 
+        new_procedure('locked', |name| {
+            bool(.lock[name].queue);
+        });
+
         new_procedure('release', |name, room_id| {
             wse(.lock[name].release(room_id));
         });
 
         .to_type('Root');
-    """)
+    """, scope=f'//{collection}')
 
 
 class _InnerRoom(Room):
@@ -93,28 +100,38 @@ async def on_join(self) -> None:
         # We might have missed the event during the join. If so, set the
         # future result to continue.
         ok = await self.client.run('test', self.id, scope=self.scope)
-        if ok:
+        if ok and not self.future.done():
             self.future.set_result(None)
 
     @event('go')
     def on_go(self):
-        self.future.set_result(None)
+        if not self.future.done():
+            self.future.set_result(None)
 
 
 @asynccontextmanager
 async def lock(client: Client, name: str,
                scope: str = '//lock',
-               timeout: int = 60):
+               timeout: int = 60) -> AsyncGenerator[int, None]:
 
-    room_id: int | None = \
+    res: tuple[bool, int, int] = \
         await client.run('acquire', name, timeout, scope=scope)
 
-    if room_id is not None:
+    immediately, room_id, size = res
+    if not immediately:
         room = _InnerRoom(room_id, scope=scope)
         await room.join(client, wait=None)
-        await room.future
+        try:
+            await asyncio.wait_for(room.future, timeout=timeout*size)
+        except asyncio.TimeoutError:
+            pass
 
     try:
         yield room_id  # Lock Id assigned to the 'as' target (not required)
     finally:
-        await client.run('release', room_id, scope=scope)
+        await client.run('release', name, room_id, scope=scope)
+
+
+async def locked(client: Client, name: str, scope: str = '//lock') -> bool:
+    res: bool = await client.run('locked', name, scope=scope)
+    return res