gen: Correctly accept concurrent futures in gen.with_timeout.

The existing implementation happened to work for tornado's Future
implementation but breaks with the move to asyncio Futures.

diff --git a/tornado/concurrent.py b/tornado/concurrent.py
index 135f1e14aa7bf36066b18376c3263961fd6ca653..9fb06db0268dea2fc4f0630a825f661876fe51d5 100644 (file)
--- a/tornado/concurrent.py
+++ b/tornado/concurrent.py
@@ -508,6 +508,8 @@ def chain_future(a, b):
 
     The result (success or failure) of ``a`` will be copied to ``b``, unless
     ``b`` has already been completed or cancelled by the time ``a`` finishes.
+
+    Accepts both Tornado/asyncio `Future` objects and `concurrent.futures.Future`.
     """
     def copy(future):
         assert future is a
@@ -520,7 +522,12 @@ def chain_future(a, b):
             b.set_exception(a.exception())
         else:
             b.set_result(a.result())
-    future_add_done_callback(a, copy)
+    if isinstance(a, Future):
+        future_add_done_callback(a, copy)
+    else:
+        # concurrent.futures.Future
+        from tornado.ioloop import IOLoop
+        IOLoop.current().add_future(a, copy)
 
 
 def future_set_exc_info(future, exc_info):

diff --git a/tornado/test/gen_test.py b/tornado/test/gen_test.py
index 8183c164e59a67eec00607b0e84cfce7466384b4..1805bce5674e5b3d52d7af442f5d7ca79715d18d 100644 (file)
--- a/tornado/test/gen_test.py
+++ b/tornado/test/gen_test.py
@@ -1273,6 +1273,7 @@ class WithTimeoutTest(AsyncTestCase):
     @unittest.skipIf(futures is None, 'futures module not present')
     @gen_test
     def test_timeout_concurrent_future(self):
+        # A concurrent future that does not resolve before the timeout.
         with futures.ThreadPoolExecutor(1) as executor:
             with self.assertRaises(gen.TimeoutError):
                 yield gen.with_timeout(self.io_loop.time(),
@@ -1281,9 +1282,21 @@ class WithTimeoutTest(AsyncTestCase):
     @unittest.skipIf(futures is None, 'futures module not present')
     @gen_test
     def test_completed_concurrent_future(self):
+        # A concurrent future that is resolved before we even submit it
+        # to with_timeout.
+        with futures.ThreadPoolExecutor(1) as executor:
+            f = executor.submit(lambda: None)
+            f.result()  # wait for completion
+            yield gen.with_timeout(datetime.timedelta(seconds=3600), f)
+
+    @unittest.skipIf(futures is None, 'futures module not present')
+    @gen_test
+    def test_normal_concurrent_future(self):
+        # A conccurrent future that resolves while waiting for the timeout.
         with futures.ThreadPoolExecutor(1) as executor:
             yield gen.with_timeout(datetime.timedelta(seconds=3600),
-                                   executor.submit(lambda: None))
+                                   executor.submit(lambda: time.sleep(0.01)))
+
 
 
 class WaitIteratorTest(AsyncTestCase):