sablink/entropy
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[entropy.db] rewrite locking infrastructure, handle multithreading use case

Browse Source
This commit is contained in:
Fabio Erculiani
2013-12-03 14:30:12 +01:00
parent 8c2b9c4841
commit d7f22534b2
3 changed files with 161 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/entropy/db/skel.py
+36 -22
View File
@@ -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()
lib/entropy/db/sqlite.py
+102 -62
View File
@@ -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):
lib/tests/db.py
+23 -22
View File
@@ -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: