Repository: tornadoweb/tornado
Pull request #2179 (2017-10-24 09:11:04)

Fetch IOLoop in PeriodicCallback.start(), not __init__()
See motivation at https://github.com/dask/distributed/pull/1172


Patch:
diff --git a/tornado/ioloop.py b/tornado/ioloop.py
index a60632e4..631f15a2 100644
--- a/tornado/ioloop.py
+++ b/tornado/ioloop.py
@@ -44,7 +44,7 @@ import time
 import traceback
 import math
 
-from tornado.concurrent import TracebackFuture, is_future, chain_future
+from tornado.concurrent import TracebackFuture, is_future
 from tornado.log import app_log, gen_log
 from tornado.platform.auto import set_close_exec, Waker
 from tornado import stack_context
@@ -655,12 +655,8 @@ class IOLoop(Configurable):
                 from tornado.process import cpu_count
                 self._executor = ThreadPoolExecutor(max_workers=(cpu_count() * 5))
             executor = self._executor
-        c_future = executor.submit(func, *args)
-        # Concurrent Futures are not usable with await. Wrap this in a
-        # Tornado Future instead, using self.add_future for thread-safety.
-        t_future = TracebackFuture()
-        self.add_future(c_future, lambda f: chain_future(f, t_future))
-        return t_future
+
+        return executor.submit(func, *args)
 
     def set_default_executor(self, executor):
         """Sets the default executor to use with :meth:`run_in_executor`."""
@@ -1086,12 +1082,15 @@ class PeriodicCallback(object):
         if callback_time <= 0:
             raise ValueError("Periodic callback must have a positive callback_time")
         self.callback_time = callback_time
-        self.io_loop = IOLoop.current()
         self._running = False
         self._timeout = None
 
     def start(self):
         """Starts the timer."""
+        # Looking up the IOLoop here allows to first instantiate the
+        # PeriodicCallback in another thread, then start it using
+        # IOLoop.add_callback().
+        self.io_loop = IOLoop.current()
         self._running = True
         self._next_timeout = self.io_loop.time()
         self._schedule_next()


Test patch:
diff --git a/tornado/test/ioloop_test.py b/tornado/test/ioloop_test.py
index 603eb202..bde351f6 100644
--- a/tornado/test/ioloop_test.py
+++ b/tornado/test/ioloop_test.py
@@ -604,70 +604,56 @@ class TestIOLoopFutures(AsyncTestCase):
         event1 = threading.Event()
         event2 = threading.Event()
 
-        def sync_func(self_event, other_event):
+        def callback(self_event, other_event):
             self_event.set()
-            other_event.wait()
-            # Note that return value doesn't actually do anything,
-            # it is just passed through to our final assertion to
-            # make sure it is passed through properly.
+            time.sleep(0.01)
+            self.assertTrue(other_event.is_set())
             return self_event
 
-        # Run two synchronous functions, which would deadlock if not
-        # run in parallel.
         res = yield [
-            IOLoop.current().run_in_executor(None, sync_func, event1, event2),
-            IOLoop.current().run_in_executor(None, sync_func, event2, event1)
+            IOLoop.current().run_in_executor(None, callback, event1, event2),
+            IOLoop.current().run_in_executor(None, callback, event2, event1)
         ]
 
         self.assertEqual([event1, event2], res)
 
     @skipBefore35
-    @gen_test
     def test_run_in_executor_native(self):
         event1 = threading.Event()
         event2 = threading.Event()
 
-        def sync_func(self_event, other_event):
+        def callback(self_event, other_event):
             self_event.set()
+            time.sleep(0.01)
+            self.assertTrue(other_event.is_set())
             other_event.wait()
             return self_event
 
-        # Go through an async wrapper to ensure that the result of
-        # run_in_executor works with await and not just gen.coroutine
-        # (simply passing the underlying concurrrent future would do that).
         namespace = exec_test(globals(), locals(), """
-            async def async_wrapper(self_event, other_event):
-                return await IOLoop.current().run_in_executor(
-                    None, sync_func, self_event, other_event)
+            async def main():
+                res = await gen.multi([
+                    IOLoop.current().run_in_executor(None, callback, event1, event2),
+                    IOLoop.current().run_in_executor(None, callback, event2, event1)
+                ])
+                self.assertEqual([event1, event2], res)
         """)
+        IOLoop.current().run_sync(namespace['main'])
 
-        res = yield [
-            namespace["async_wrapper"](event1, event2),
-            namespace["async_wrapper"](event2, event1)
-            ]
-
-        self.assertEqual([event1, event2], res)
-
-    @gen_test
     def test_set_default_executor(self):
-        count = [0]
-
-        class MyExecutor(futures.ThreadPoolExecutor):
+        class MyExecutor(futures.Executor):
             def submit(self, func, *args):
-                count[0] += 1
-                return super(MyExecutor, self).submit(func, *args)
+                return Future()
 
         event = threading.Event()
 
-        def sync_func():
+        def future_func():
             event.set()
 
-        executor = MyExecutor(1)
+        executor = MyExecutor()
         loop = IOLoop.current()
         loop.set_default_executor(executor)
-        yield loop.run_in_executor(None, sync_func)
-        self.assertEqual(1, count[0])
-        self.assertTrue(event.is_set())
+        loop.run_in_executor(None, future_func)
+        loop.add_timeout(0.01, lambda: self.assertFalse(event.is_set()))
 
 
 class TestIOLoopRunSync(unittest.TestCase):
@@ -761,6 +747,21 @@ class TestPeriodicCallback(unittest.TestCase):
         self.io_loop.start()
         self.assertEqual(calls, expected)
 
+    def test_io_loop_set_at_start(self):
+        # Check PeriodicCallback uses the current IOLoop at start() time,
+        # not at instantiation time.
+        calls = []
+        io_loop = FakeTimeIOLoop()
+
+        def cb():
+            calls.append(io_loop.time())
+        pc = PeriodicCallback(cb, 10000)
+        io_loop.make_current()
+        pc.start()
+        io_loop.call_later(50, io_loop.stop)
+        io_loop.start()
+        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])
+
 
 class TestIOLoopConfiguration(unittest.TestCase):
     def run_python(self, *statements):