[entropy.misc] drop MasterSlaveLock, no longer used
This commit is contained in:
Fabio Erculiani
@@ -398,152 +398,6 @@ class ParallelTask(threading.Thread):
|
||||
return self.__rc
|
||||
|
||||
|
||||
class MasterSlaveLock(object):
|
||||
|
||||
DEBUG = False
|
||||
|
||||
def __init__(self):
|
||||
self.__locks = {}
|
||||
self.__ms_action_lock = threading.Lock()
|
||||
self.__locks_dict_lock = threading.Lock()
|
||||
self.__master_locked = False
|
||||
self.__master_is_locking = False
|
||||
self.__spin_count = 0
|
||||
|
||||
@staticmethod
|
||||
def log(*args):
|
||||
txt = ' '.join([str(x) for x in args])
|
||||
sys.stdout.write(txt + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
def slave_acquire(self, key, timeout = None):
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("calling slave_acquire", key)
|
||||
with self.__locks_dict_lock:
|
||||
lock = self.__locks.setdefault(key, threading.Lock())
|
||||
|
||||
# setup "else" case variables here to reduce transaction length below.
|
||||
acquired = False
|
||||
timeout_secs = 1
|
||||
if timeout is not None:
|
||||
timeout_secs = int(timeout)
|
||||
if timeout_secs < 1:
|
||||
# guard value
|
||||
timeout_secs = 1
|
||||
|
||||
with self.__ms_action_lock:
|
||||
if self.__master_locked or self.__master_is_locking:
|
||||
# master locked us, give up
|
||||
return False
|
||||
elif lock.locked() and not self.__master_locked:
|
||||
# slave locked, we're fine
|
||||
# NOTE: it can happen to allow one more acquire() on the lock
|
||||
# while master is locking, this is fine, the case is handled
|
||||
# in master_acquire().
|
||||
self.__spin_count += 1
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("DEBUG: new spin count (++1)",
|
||||
self.__spin_count)
|
||||
return True
|
||||
else:
|
||||
# if lock is not locked, first time slave gets here
|
||||
# spin_count here must be 0.
|
||||
|
||||
while timeout_secs > 0:
|
||||
# non blocking if timeout is not None
|
||||
# blocking otherwise.
|
||||
acquired = lock.acquire(timeout is None)
|
||||
if timeout is None:
|
||||
# if we get here, lock is acquired when in blocking mode
|
||||
acquired = True
|
||||
break
|
||||
if not acquired:
|
||||
timeout_secs -= 1
|
||||
time.sleep(1)
|
||||
continue
|
||||
break
|
||||
if acquired:
|
||||
self.__spin_count += 1
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("DEBUG: new spin count (++2)",
|
||||
self.__spin_count)
|
||||
return True
|
||||
else: # otherwise, give up.
|
||||
return False
|
||||
|
||||
def slave_release(self, key):
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("calling slave_release", key)
|
||||
with self.__locks_dict_lock:
|
||||
lock = self.__locks.setdefault(key, threading.Lock())
|
||||
|
||||
with self.__ms_action_lock:
|
||||
if (self.__spin_count - 1) < 1:
|
||||
# release only if master did not acquire
|
||||
if self.__master_locked:
|
||||
# don't do anything, don't release the lock!
|
||||
pass
|
||||
elif self.__master_is_locking:
|
||||
# if master is locking, release the lock
|
||||
# even if master already acquired it.
|
||||
# This race condition will be handled by master.
|
||||
lock.release()
|
||||
# otherwise do nothing
|
||||
self.__spin_count -= 1
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("DEBUG: new spin count (--)",
|
||||
self.__spin_count)
|
||||
|
||||
def master_acquire(self, key):
|
||||
# Acquire the lock, then wait for every slave to release before
|
||||
# returning
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("calling master_acquire", key)
|
||||
with self.__locks_dict_lock:
|
||||
lock = self.__locks.setdefault(key, threading.Lock())
|
||||
|
||||
self.__master_is_locking = True
|
||||
while True:
|
||||
acquired = lock.acquire(False) # non blocking, wait indefinitely
|
||||
if not acquired:
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log(
|
||||
"still waiting for master_acquire, lock acquire", key)
|
||||
time.sleep(1.0)
|
||||
continue
|
||||
break
|
||||
|
||||
# from now on, new requests from slaves won't accepted, since
|
||||
# the acquisition and the enable of master_locked are not atomic
|
||||
# we'll have to wait until all the slave requests are flushed out.
|
||||
# this is done below, in the while loop.
|
||||
self.__master_locked = True
|
||||
# -- race condition here, handled below --
|
||||
self.__master_is_locking = False
|
||||
# now make sure we still hold the lock, unsing a non-blocking acquire
|
||||
# if it fails, it's because we are still holding the lock
|
||||
# if it succeeds, it means that the last slave released a lock.
|
||||
lock.acquire(False)
|
||||
|
||||
# wait for slaves to terminate before returning
|
||||
while self.__spin_count > 0:
|
||||
time.sleep(0.4)
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log(
|
||||
"waiting for slaves to terminate, slaves:",
|
||||
self.__spin_count)
|
||||
|
||||
def master_release(self, key):
|
||||
# This can only happen when no slave is holding the lock
|
||||
# and master locked.
|
||||
if MasterSlaveLock.DEBUG:
|
||||
MasterSlaveLock.log("calling master_release", key)
|
||||
# must be available, otherwise KeyError is raised
|
||||
with self.__locks_dict_lock:
|
||||
self.__locks[key].release()
|
||||
self.__master_locked = False
|
||||
|
||||
|
||||
class FlockFile(object):
|
||||
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user