For additional reference see: https://stackoverflow.com/questions/58410610/calling-condition-wait-inside-thread-causes-retrieval-of-any-future-to-block-o
I created my own reentrant read write lock which uses Condtion.wait() to wait on pending locks.
This lock is shared with two tasks that are submitted to a GreenThreadPoolExecutor.
Since one of the tasks tries to acquire the read lock and the other the write lock one of the two will block on the Condition.wait().
When waiters.wait_for_any() is called the expected behavior is that future that has successfully ran be returned as its execution has finished while the other remains undone as it is blocked on condition.wait(). However, waiters.wait_for_any() blocks indefinitely and the timeout parameter is subsequently ignored.
I believe this behavior to be incorrect as one of the futures has successfully finished and as a result should be returned by wait_for_any.
I use the rest of the information to describe the classes, methods and calls that lead to this behavior. Starting with large parts of the ReentrantReadWriteLock:
class ReentrantReadWriteLock(object):
def __init__(self):
self._read_lock = RLock()
self._write_lock = RLock()
self._rcondition = Condition(lock=self._read_lock)
self._wcondition = Condition(lock=self._write_lock)
self._num_readers = 0
self._wants_write = False
def read_acquire(self, blocking=True, timeout=-1):
waitout = None if timeout == -1 else timeout
LOG.warning("read waitoit: {0}".format(waitout))
first_it = True
try:
if self._read_lock.acquire(blocking, timeout):
LOG.warning("read internal lock acquired")
while self._wants_write:
LOG.warning("read wants write true")
if not blocking or not first_it:
LOG.warning("read non blocking or second it")
return False
LOG.warning("read wait")
self._rcondition.wait(waitout)
first_it = False
LOG.warning("read acquired lock")
self._num_readers += 1
return True
LOG.warning("read internal lock failed")
return False
finally:
self._read_lock.release()
def write_acquire(self, blocking=True, timeout=-1):
waitout = None if timeout == -1 else timeout
LOG.warning("write waitoit: {0}".format(waitout))
first_it = True
try:
if self._write_lock.acquire(blocking, timeout):
LOG.warning("write internal lock acquired")
while self._num_readers > 0 or self._wants_write:
LOG.warning("write wants write true or num read")
if not blocking or not first_it:
LOG.warning("write non blocking or second it")
return False
LOG.warning("write wait")
self._wcondition.wait(waitout)
first_it = False
LOG.warning("write acquired lock")
self._wants_write = True
return True
LOG.warning("write internal lock failed")
return False
finally:
self._write_lock.release()
This is my test case, this test will block on results = waiters.wait_for_any(futures), once again
specifying the timeout parameter does not resolve the issue:
def get_read(self, rrwlock):
return rrwlock.read_acquire()
def get_write(self, rrwlock):
return rrwlock.write_acquire()
def test():
self._threadpool = futurist.GreenThreadPoolExecutor(max_workers=4)
rrwlock = ReentrantReadWriteLock()
futures = []
futures.append(self._threadpool.submit(self.get_read, rrwlock))
futures.append(self._threadpool.submit(self.get_write, rrwlock))
# Get the results and verify only one of the calls succeeded
# assert that the other call is still pending
results = waiters.wait_for_any(futures)
self.assertTrue(results[0].pop().result)
self.assertEqual(1, len(results[1]))
This bug seems very related even though closed: https:/
It describes how concurrent.
I have tracked that the blocking on the main thread occurs on waiter.