Untitled
unknown
plain_text
a year ago
1.2 kB
9
Indexable
from collections import deque
from in_memory_db import InMemoryDB
class InMemoryDBImpl(InMemoryDB):
# ... (rest of the code is the same as before)
def unlock(self, key: str, caller_id: str = None) -> str | None:
if key not in self.locks:
return "invalid_request" # Key is not locked
if caller_id is not None and self.locks[key] != caller_id:
return None # Caller doesn't own the lock
released_user = self.locks.pop(key)
if key not in self.lock_queues or not self.lock_queues[key]:
# No one is waiting in the queue
return "released"
# Assign the lock to the next user in the queue
next_user = self.lock_queues[key].popleft()
self.locks[key] = next_user
return "released"
def set_or_inc(self, key: str, field: str, value: int, caller_id: str = None) -> int | None:
if key in self.locks and (caller_id is None or self.locks[key] != caller_id):
return None # Ignore if locked by another user or no caller_id provided
return self._set_or_inc_internal(key, field, value)
# ... (rest of the code is the same as before)Editor is loading...
Leave a Comment