diff --git a/lib/entropy/db/skel.py b/lib/entropy/db/skel.py index e2490f6a1..9538013d6 100644 --- a/lib/entropy/db/skel.py +++ b/lib/entropy/db/skel.py @@ -437,11 +437,12 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): If this is not the case, please synchronize using the Entropy Resources General Lock (entropy.tools.acquire_entropy_locks()). """ - acquired = False + opaque = None try: - lock_path = self.lock_path() - acquired = self.try_acquire_shared() - if not acquired: + opaque = self.try_acquire_shared() + if opaque is None: + lock_path = self.lock_path() + self.output( "%s %s ..." % ( darkred(_("Acquiring shared lock on")), @@ -451,8 +452,7 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): back = True, importance = 0) - self.acquire_shared() - acquired = True + opaque = self.acquire_shared() self.output( "%s %s" % ( @@ -466,8 +466,8 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): yield finally: - if acquired: - self.release_shared() + if opaque is not None: + self.release_shared(opaque) @contextlib.contextmanager def exclusive(self): @@ -480,11 +480,12 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): If this is not the case, please synchronize using the Entropy Resources General Lock (entropy.tools.acquire_entropy_locks()). """ - acquired = False + opaque = None try: - lock_path = self.lock_path() - acquired = self.try_acquire_exclusive() - if not acquired: + opaque = self.try_acquire_exclusive() + if opaque is None: + lock_path = self.lock_path() + self.output( "%s %s ..." % ( darkred(_("Acquiring exclusive lock on")), @@ -494,8 +495,7 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): back = True, importance = 0) - self.acquire_exclusive() - acquired = True + opaque = self.acquire_exclusive() self.output( "%s %s" % ( @@ -509,8 +509,8 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): yield finally: - if acquired: - self.release_exclusive() + if opaque is not None: + self.release_exclusive(opaque) def acquire_shared(self): """ @@ -521,6 +521,9 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): object instantiation are valid throughout the whole object lifecycle. If this is not the case, please synchronize using the Entropy Resources General Lock (entropy.tools.acquire_entropy_locks()). + + @return: an opaque that must be used to release the lock. + @rtype: object """ raise NotImplementedError() @@ -533,6 +536,9 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): object instantiation are valid throughout the whole object lifecycle. If this is not the case, please synchronize using the Entropy Resources General Lock (entropy.tools.acquire_entropy_locks()). + + @return: an opaque that must be used to release the lock. + @rtype: object """ raise NotImplementedError() @@ -546,8 +552,9 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): If this is not the case, please synchronize using the Entropy Resources General Lock (entropy.tools.acquire_entropy_locks()). - @return: True, if acquired, False otherwise. - @rtype: bool + @return: an opaque object that must be used to release the lock, None + otherwise. + @rtype: object or None """ raise NotImplementedError() @@ -561,26 +568,33 @@ class EntropyRepositoryBase(TextInterface, EntropyRepositoryPluginStore): If this is not the case, please synchronize using the Entropy Resources General Lock (entropy.tools.acquire_entropy_locks()). - @return: True, if acquired, False otherwise. - @rtype: bool + @return: an opaque object that must be used to release the lock, None + otherwise. + @rtype: object or None """ raise NotImplementedError() - def release_shared(self): + def release_shared(self, opaque): """ Release the previously acquired shared file lock for this repository. This is used for inter-process synchronization only. Make sure to commit any pending transaction before releasing the lock. + + @param opaque: the opaque object returned by *acquire_shared methods. + @type opaque: object """ raise NotImplementedError() - def release_exclusive(self): + def release_exclusive(self, opaque): """ Release the previously acquired exclusive file lock for this repository. This is used for inter-process synchronization only. Make sure to commit any pending transaction before releasing the lock. + + @param opaque: the opaque object returned by *acquire_exclusive methods. + @type opaque: object """ raise NotImplementedError() diff --git a/lib/entropy/db/sqlite.py b/lib/entropy/db/sqlite.py index f458a56fb..147ab2b2c 100644 --- a/lib/entropy/db/sqlite.py +++ b/lib/entropy/db/sqlite.py @@ -199,9 +199,6 @@ class EntropySQLiteRepository(EntropySQLRepository): on close() @type temporary: bool """ - self._flock_lock = threading.RLock() - self._flock = None - self._sqlite = self.ModuleProxy.get() EntropySQLRepository.__init__( @@ -477,107 +474,155 @@ class EntropySQLiteRepository(EntropySQLRepository): self._discardLiveCache() return self._live_cacher.get(self._getLiveCacheKey() + key) - def _get_flock(self): + def _get_flock(self, mode): """ Get the lock object used for locking. """ - flock = None - with self._flock_lock: - if not self._flock: + lock_path = self.lock_path() + lock_dir = os.path.dirname(lock_path) - lock_path = self.lock_path() - lock_dir = os.path.dirname(lock_path) - try: - const_setup_directory(lock_dir) - except (OSError, IOError): - # best effort, hope not to fail - # on FlockFile() - pass + try: + const_setup_directory(lock_dir) + except (OSError, IOError): + # best effort, hope not to fail + # on FlockFile() + pass - flock = FlockFile(lock_path) - self._flock = flock - else: - flock = self._flock - return flock + class RepositoryFlockFile(FlockFile): + + def __init__(self, lock_path, mode): + super(RepositoryFlockFile, self).__init__(lock_path) + self._mode = mode + + return RepositoryFlockFile(lock_path, mode) def acquire_shared(self): """ Reimplemented from EntropyBaseRepository. """ - if not self._is_memory(): - flock = self._get_flock() - flock.acquire_shared() + if self._is_memory(): + return True + else: + flock = None + acquired = False + try: + flock = self._get_flock(False) - # in-RAM cached data may have become stale - self.clearCache() + flock.acquire_shared() + acquired = True + + # in-RAM cached data may have become stale + self.clearCache() + return flock + + finally: + if not acquired and flock is not None: + flock.close() def try_acquire_shared(self): """ Reimplemented from EntropyBaseRepository. """ - acquired = None if self._is_memory(): - acquired = True + return True else: - flock = self._get_flock() - acquired = flock.try_acquire_shared() - if acquired: - # in-RAM cached data may have become stale - self.clearCache() + acquired = False + flock = None + try: - return acquired + flock = self._get_flock(False) + acquired = flock.try_acquire_shared() + if acquired: + # in-RAM cached data may have become stale + self.clearCache() + return flock + + else: + return None + + finally: + if not acquired and flock is not None: + flock.close() def acquire_exclusive(self): """ Reimplemented from EntropyBaseRepository. """ - if not self._is_memory(): - flock = self._get_flock() - flock.acquire_exclusive() + if self._is_memory(): + return True + else: + flock = None + acquired = False + try: + flock = self._get_flock(True) + flock.acquire_exclusive() + acquired = True - # in-RAM cached data may have become stale - self.clearCache() + # in-RAM cached data may have become stale + self.clearCache() + return flock + + finally: + if not acquired and flock is not None: + flock.close() def try_acquire_exclusive(self): """ Reimplemented from EntropyBaseRepository. """ - acquired = None if self._is_memory(): - acquired = True + return True else: - flock = self._get_flock() - acquired = flock.try_acquire_exclusive() - if acquired: - # in-RAM cached data may have become stale - self.clearCache() + acquired = False + flock = None + try: - return acquired + flock = self._get_flock(True) + acquired = flock.try_acquire_exclusive() + if acquired: + # in-RAM cached data may have become stale + self.clearCache() + return flock - def _release_flock(self): + else: + return None + + finally: + if not acquired and flock is not None: + flock.close() + + def _release_flock(self, flock, mode): """ Release the resource associated with the FlockFile object. """ - with self._flock_lock: - if not self._flock: - raise RuntimeError("releasing a lock that wasn't acquired") - self._flock.release() + if self._is_memory(): + return + else: + if flock._mode != mode: + raise RuntimeError( + "Programming error: acquired lock in a different mode") + flock.release() + flock.close() - def release_shared(self): + def release_shared(self, opaque): """ Reimplemented from EntropyBaseRepository. """ - if not self._is_memory(): + if self._is_memory(): + return + else: self.commit() - self._release_flock() + self._release_flock(opaque, False) - def release_exclusive(self): + def release_exclusive(self, opaque): """ Reimplemented from EntropyBaseRepository. """ - if not self._is_memory(): + if self._is_memory(): + return + else: self.commit() - self._release_flock() + self._release_flock(opaque, True) def close(self, safe=False): """ @@ -586,11 +631,6 @@ class EntropySQLiteRepository(EntropySQLRepository): """ super(EntropySQLiteRepository, self).close(safe=safe) - with self._flock_lock: - if self._flock: - self._flock.close() - self._flock = None - self._cleanup_all(_cleanup_main_thread=not safe) if self._temporary and (not self._is_memory()) and \ os.path.isfile(self._db): diff --git a/lib/tests/db.py b/lib/tests/db.py index 31dda5349..d55550555 100644 --- a/lib/tests/db.py +++ b/lib/tests/db.py @@ -1027,35 +1027,36 @@ class EntropyRepositoryTest(unittest.TestCase): test_db.initializeRepository() with test_db.shared(): - self.assert_(test_db._flock is not None) + self.assertEqual(test_db.try_acquire_exclusive(), + None) - test_db.close() - self.assert_(test_db._flock is None) + with test_db.exclusive(): + self.assertEqual(test_db.try_acquire_shared(), + None) - with test_db.shared(): - self.assert_(test_db._flock is not None) + opaque_shared = test_db.try_acquire_shared() + self.assert_(opaque_shared is not None) - acquired = test_db.try_acquire_shared() - self.assertEqual(acquired, True) + opaque_exclusive = test_db.try_acquire_exclusive() + self.assert_(opaque_exclusive is None) - acquired = test_db.try_acquire_exclusive() - self.assertEqual(acquired, True) + test_db.release_shared(opaque_shared) - pid = os.fork() - if pid == 0: - test_db.close() - acquired = test_db.try_acquire_exclusive() - if acquired: - os._exit(1) + opaque_exclusive = test_db.try_acquire_exclusive() + self.assert_(opaque_exclusive is not None) - acquired = test_db.try_acquire_shared() - if acquired: - os._exit(1) + opaque_exclusive2 = test_db.try_acquire_exclusive() + self.assert_(opaque_exclusive2 is None) - os._exit(0) - else: - child_pid, exit_st = os.waitpid(pid, 0) - self.assertEqual(exit_st, 0) + test_db.release_exclusive(opaque_exclusive) + + opaque_exclusive = test_db.try_acquire_exclusive() + self.assert_(opaque_exclusive is not None) + + self.assertRaises(RuntimeError, test_db.release_shared, + opaque_exclusive) + + test_db.release_exclusive(opaque_exclusive) finally: if test_db is not None: