Cancel timeouts in Condition.wait and Semaphore.acquire.

diff --git a/tornado/locks.py b/tornado/locks.py
index 4d2ab9b38331408464fac2e94ff1f3edce3a73d5..bdf15a2c274ce013899ab23b3e7ff42a82ead6a4 100644 (file)
--- a/tornado/locks.py
+++ b/tornado/locks.py
@@ -73,7 +73,10 @@ class Condition(_TimeoutGarbageCollector):
             def on_timeout():
                 waiter.set_result(False)
                 self._garbage_collect()
-            self.io_loop.add_timeout(timeout, on_timeout)
+            io_loop = ioloop.IOLoop.current()
+            timeout_handle = io_loop.add_timeout(timeout, on_timeout)
+            waiter.add_done_callback(
+                lambda _: io_loop.remove_timeout(timeout_handle))
         return waiter
 
     def notify(self, n=1):
@@ -223,7 +226,10 @@ class Semaphore(_TimeoutGarbageCollector):
                 def on_timeout():
                     waiter.set_exception(gen.TimeoutError())
                     self._garbage_collect()
-                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
+                io_loop = ioloop.IOLoop.current()
+                timeout_handle = io_loop.add_timeout(timeout, on_timeout)
+                waiter.add_done_callback(
+                    lambda _: io_loop.remove_timeout(timeout_handle))
         return waiter
 
     def __enter__(self):

diff --git a/tornado/test/locks_test.py b/tornado/test/locks_test.py
index 8eaa4236f4a84c86d246cb8bfdd8502a3211b218..49161287dc4daa0ed1f9332779edca6845852fab 100644 (file)
--- a/tornado/test/locks_test.py
+++ b/tornado/test/locks_test.py
@@ -87,7 +87,10 @@ class ConditionTest(AsyncTestCase):
     @gen_test
     def test_wait_timeout(self):
         c = locks.Condition()
-        self.assertFalse((yield c.wait(timedelta(seconds=0.01))))
+        wait = c.wait(timedelta(seconds=0.01))
+        self.io_loop.call_later(0.02, c.notify)  # Too late.
+        yield gen.sleep(0.03)
+        self.assertFalse((yield wait))
 
     @gen_test
     def test_wait_timeout_preempted(self):
@@ -95,7 +98,9 @@ class ConditionTest(AsyncTestCase):
 
         # This fires before the wait times out.
         self.io_loop.call_later(0.01, c.notify)
-        yield c.wait(timedelta(seconds=1))
+        wait = c.wait(timedelta(seconds=0.02))
+        yield gen.sleep(0.03)
+        yield wait  # No TimeoutError.
 
     @gen_test
     def test_notify_n_with_timeout(self):
@@ -255,13 +260,29 @@ class SemaphoreTest(AsyncTestCase):
         sem = locks.Semaphore(2)
         yield sem.acquire()
         yield sem.acquire()
+        acquire = sem.acquire(timedelta(seconds=0.01))
+        self.io_loop.call_later(0.02, sem.release)  # Too late.
+        yield gen.sleep(0.3)
         with self.assertRaises(gen.TimeoutError):
-            yield sem.acquire(timedelta(seconds=0.01))
+            yield acquire
 
+        sem.acquire()
         f = sem.acquire()
+        self.assertFalse(f.done())
         sem.release()
         self.assertTrue(f.done())
 
+    @gen_test
+    def test_acquire_timeout_preempted(self):
+        sem = locks.Semaphore(1)
+        yield sem.acquire()
+
+        # This fires before the wait times out.
+        self.io_loop.call_later(0.01, sem.release)
+        acquire = sem.acquire(timedelta(seconds=0.02))
+        yield gen.sleep(0.03)
+        yield acquire  # No TimeoutError.
+
     def test_release_unacquired(self):
         # Unbounded releases are allowed, and increment the semaphore's value.
         sem = locks.Semaphore()