Index: Doc/library/threading.rst =================================================================== --- Doc/library/threading.rst (revision 85963) +++ Doc/library/threading.rst (working copy) @@ -580,6 +580,25 @@ .. versionchanged:: 3.2 Previously, the method always returned ``None``. + .. method:: wait_for(predicate, timeout=None) + + Wait until a condition evaluates to True. *predicate* should be a + callable whose result will be interpreted as a boolean value. + A *timeout* may be provided giving the maximum time to wait. + + This utility method may :meth:`wait` repeatedly until the predicate + evaluates to ``True``, or until a timeout occurs. The return value is + the last return value of the predicate and will evaluate to + ``False`` if the method timed out. + + Using this method, the consumer above can be written thus:: + + with cv: + cv.wait_for(an_item_is_available) + get_an_available_item() + + .. versionadded:: 3.2 + .. method:: notify() Wake up a thread waiting on this condition, if any. If the calling thread Index: Lib/test/lock_tests.py =================================================================== --- Lib/test/lock_tests.py (revision 85963) +++ Lib/test/lock_tests.py (working copy) @@ -445,6 +445,32 @@ # This makes it hard to verify the result value. # In practice, this implementation has no spurious wakeups. self.assertFalse(result) + + def test_waitfor(self): + cond = self.condtype() + state = 0 + def f(): + with cond: + result = cond.wait_for(lambda : state==4) + self.assertTrue(result) + self.assertEqual(state, 4) + b = Bunch(f, 1) + for i in range(5): + time.sleep(0.01) + with cond: + state += 1 + cond.notify() + b.wait_for_finished() + + def test_waitfor_timeout(self): + cond = self.condtype() + state = 0 + with cond: + dt = time.time() + result = cond.wait_for(lambda : state == 4, timeout=0.5) + dt = time.time() - dt + self.assertFalse(result) + self.assertTimeout(dt, 0.5) class BaseSemaphoreTests(BaseTestCase): Index: Lib/threading.py =================================================================== --- Lib/threading.py (revision 85963) +++ Lib/threading.py (working copy) @@ -254,6 +254,32 @@ finally: self._acquire_restore(saved_state) + def wait_for(self, predicate, timeout=None): + endtime = None + waittime = timeout + result = predicate() + while not result: + if waittime is not None: + if endtime is None: + endtime = _time() + waittime + else: + waittime = endtime - _time() + if waittime < 0: + if __debug__: + self._note("%s.wait_for(%r, %r): Timed out.", + self, predicate, timeout) + break + if __debug__: + self._note("%s.wait_for(%r, %r): Waiting with timeout=%s.", + self, predicate, timeout, waittime) + self.wait(waittime) + result = predicate() + else: + if __debug__: + self._note("%s.wait_for(%r, %r): Success.", + self, predicate, timeout) + return result + def notify(self, n=1): if not self._is_owned(): raise RuntimeError("cannot notify on un-acquired lock") @@ -482,13 +508,12 @@ # Wait in the barrier until we are relased. Raise an exception # if the barrier is reset or broken. def _wait(self, timeout): - while self._state == 0: - if self._cond.wait(timeout) is False: - #timed out. Break the barrier - self._break() - raise BrokenBarrierError - if self._state < 0: - raise BrokenBarrierError + if not self._cond.wait_for(lambda : self._state != 0, timeout): + #timed out. Break the barrier + self._break() + raise BrokenBarrierError + if self._state < 0: + raise BrokenBarrierError assert self._state == 1 # If we are the last thread to exit the barrier, signal any threads