Skip to content

Commit

Permalink
Merge pull request #373 from projectcaluma/fix_misc
Browse files Browse the repository at this point in the history
Fix: misc small issues
  • Loading branch information
luytena authored Nov 20, 2024
2 parents 14c11cd + d0fc075 commit 1877532
Showing 1 changed file with 21 additions and 21 deletions.
42 changes: 21 additions & 21 deletions manabi/lock.py
Original file line number Diff line number Diff line change
@@ -47,6 +47,9 @@ class ManabiContextLockMixin(ABC):
def acquire(self):
pass

def __init__(self):
self._id = -1

@abstractmethod
def release(self):
pass
@@ -57,6 +60,13 @@ def __enter__(self):
def __exit__(self, exc_type, exc_value, traceback):
self.release()

def _check_and_set_tid(self):
tid: int = threading.get_ident()
tid_is_set = self._id != -1
if tid_is_set and self._id != tid:
_logger.error("Do not use from multiple threads")
self._id = tid


class ManabiShelfLock(ManabiContextLockMixin):
_storage_object: Callable[[], "ManabiShelfLockLockStorage"]
@@ -69,35 +79,29 @@ def __init__(self, storage_path, storage_object: "ManabiShelfLockLockStorage"):
self._lock_file = open(f"{storage_path}.lock", "wb+")
self._fd = self._lock_file.fileno()
self.acquire_write = self.acquire_read = self.acquire
self._id = -1
super().__init__()

def acquire(self):
tid = threading.get_ident()
if self._id and self._id != tid:
_logger.error("Do not use from multiple threads")
self._id = tid
self._check_and_set_tid()
if self._semaphore == 0:
fcntl.flock(self._fd, fcntl.LOCK_EX)
self._storage_object()._dict = shelve.open(str(self._storage_path))
self._semaphore += 1

def release(self):
tid = threading.get_ident()
if self._semaphore == 0:
_logger.error(
f"Inconsistent use of lock. {''.join(traceback.format_stack())}"
)
if self._id and self._id != tid:
_logger.error("Do not use from multiple threads")
self._id = tid

self._check_and_set_tid()
self._semaphore -= 1
if self._semaphore == 0:
storage_object = self._storage_object()
if storage_object._dict is not None:
storage_object._dict.close()
storage_object._dict = None
fcntl.flock(self._fd, fcntl.LOCK_UN)
self._id = threading.get_ident()


class ManabiShelfLockLockStorage(LockStorageDict, ManabiTimeoutMixin):
@@ -150,18 +154,15 @@ class ManabiPostgresLock(ManabiContextLockMixin):
_storage_object: Callable[[], "ManabiDbLockStorage"]

def __init__(self, storage_object: "ManabiDbLockStorage"):
self._id = None
# type manually checked
self._storage_object = weakref.ref(storage_object) # type: ignore
self.acquire_write = self.acquire_read = self.acquire
self._semaphore = 0
self._id = -1
super().__init__()

def acquire(self):
tid = threading.get_ident()
if self._id and self._id != tid:
_logger.error("Do not use from multiple threads")
self._id = tid
self._check_and_set_tid()
if self._semaphore == 0:
_logger.info(f"{tid} acquire")
self._storage_object().execute(
@@ -175,9 +176,7 @@ def release(self):
_logger.error(
f"Inconsistent use of lock. {''.join(traceback.format_stack())}"
)
if self._id and self._id != tid:
_logger.error("Do not use from multiple threads")
self._id = tid
self._check_and_set_tid()
self._semaphore -= 1
if self._semaphore == 0:
_logger.info(f"{tid} release")
@@ -258,10 +257,11 @@ def __getitem__(self, token):
cursor = self._storage_object().execute(
"SELECT data FROM manabi_lock WHERE token = %s", (str(token),)
)
lock = cursor.fetchone()
if lock is None:
locks = cursor.fetchmany(1)
if not len(locks):
raise KeyError(f"{token} not found")
lock = lock[0]

lock = locks[0][0] # first row, first col
self.encode_lock(lock)
return lock

0 comments on commit 1877532

Please sign in to comment.