gen: Use the asyncio task runner for native coroutines

tornado/gen.py

@@ -1193,20 +1193,10 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-# Convert Awaitables into Futures. It is unfortunately possible
-# to have infinite recursion here if those Awaitables assume that
-# we're using a different coroutine runner and yield objects
-# we don't understand. If that happens, the solution is to
-# register that runner's yieldable objects with convert_yielded.
-if sys.version_info >= (3, 3):
-    exec(textwrap.dedent("""
-    @coroutine
-    def _wrap_awaitable(x):
-        if hasattr(x, '__await__'):
-            x = x.__await__()
-        return (yield from x)
-    """))
-else:
+# Convert Awaitables into Futures.
+try:
+    import asyncio
+except ImportError:
     # Py2-compatible version for use with Cython.
     # Copied from PEP 380.
     @coroutine
@@ -1253,6 +1243,8 @@ def _wrap_awaitable(x):
                         _r = _value_from_stopiteration(_e)
                         break
         raise Return(_r)
+else:
+    _wrap_awaitable = asyncio.ensure_future
 
 
 def convert_yielded(yielded):

tornado/ioloop.py

@@ -504,11 +504,18 @@ def run():
             self.add_future(future_cell[0], lambda future: self.stop())
         self.add_callback(run)
         if timeout is not None:
-            timeout_handle = self.add_timeout(self.time() + timeout, self.stop)
+            def timeout_callback():
+                # If we can cancel the future, do so and wait on it. If not,
+                # Just stop the loop and return with the task still pending.
+                # (If we neither cancel nor wait for the task, a warning
+                # will be logged).
+                if not future_cell[0].cancel():
+                    self.stop()
+            timeout_handle = self.add_timeout(self.time() + timeout, timeout_callback)
         self.start()
         if timeout is not None:
             self.remove_timeout(timeout_handle)
-        if not future_cell[0].done():
+        if future_cell[0].cancelled() or not future_cell[0].done():
             raise TimeoutError('Operation timed out after %s seconds' % timeout)
         return future_cell[0].result()
 

tornado/test/ioloop_test.py

@@ -405,7 +405,9 @@ def test_exception_logging_native_coro(self):
         """The IOLoop examines exceptions from awaitables and logs them."""
         namespace = exec_test(globals(), locals(), """
         async def callback():
-            self.io_loop.add_callback(self.stop)
+            # Stop the IOLoop two iterations after raising an exception
+            # to give the exception time to be logged.
+            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)
             1 / 0
         """)
         with NullContext():

tornado/testing.py

@@ -527,12 +527,17 @@ def post_coroutine(self, *args, **kwargs):
                     timeout=timeout)
             except TimeoutError as e:
                 # run_sync raises an error with an unhelpful traceback.
-                # Throw it back into the generator or coroutine so the stack
-                # trace is replaced by the point where the test is stopped.
-                self._test_generator.throw(e)
-                # In case the test contains an overly broad except clause,
-                # we may get back here.  In this case re-raise the original
-                # exception, which is better than nothing.
+                # If the underlying generator is still running, we can throw the
+                # exception back into it so the stack trace is replaced by the
+                # point where the test is stopped. The only reason the generator
+                # would not be running would be if it were cancelled, which means
+                # a native coroutine, so we can rely on the cr_running attribute.
+                if getattr(self._test_generator, 'cr_running', True):
+                    self._test_generator.throw(e)
+                    # In case the test contains an overly broad except
+                    # clause, we may get back here.
+                # Coroutine was stopped or didn't raise a useful stack trace,
+                # so re-raise the original exception which is better than nothing.
                 raise
         return post_coroutine