concurrent: run_on_executor returns async futures, not concurrent ones

Fixes #2276

tornado/concurrent.py

@@ -421,6 +421,10 @@ def run_on_executor(*args, **kwargs):
 
     .. versionchanged:: 5.0
        Always uses the current IOLoop instead of ``self.io_loop``.
+
+    .. versionchanged:: 5.1
+       Returns a `.Future` compatible with ``await`` instead of a
+       `concurrent.futures.Future`.
     """
     def run_on_executor_decorator(fn):
         executor = kwargs.get("executor", "executor")
@@ -428,12 +432,14 @@ def run_on_executor(*args, **kwargs):
         @functools.wraps(fn)
         def wrapper(self, *args, **kwargs):
             callback = kwargs.pop("callback", None)
-            future = getattr(self, executor).submit(fn, self, *args, **kwargs)
+            async_future = Future()
+            conc_future = getattr(self, executor).submit(fn, self, *args, **kwargs)
+            chain_future(conc_future, async_future)
             if callback:
                 from tornado.ioloop import IOLoop
                 IOLoop.current().add_future(
-                    future, lambda future: callback(future.result()))
-            return future
+                    async_future, lambda future: callback(future.result()))
+            return async_future
         return wrapper
     if args and kwargs:
         raise ValueError("cannot combine positional and keyword args")

tornado/test/concurrent_test.py

@@ -31,7 +31,7 @@ from tornado.log import app_log
 from tornado import stack_context
 from tornado.tcpserver import TCPServer
 from tornado.testing import AsyncTestCase, ExpectLog, bind_unused_port, gen_test
-from tornado.test.util import unittest
+from tornado.test.util import unittest, skipBefore35, exec_test
 
 
 try:
@@ -429,6 +429,26 @@ class RunOnExecutorTest(AsyncTestCase):
         answer = yield o.f()
         self.assertEqual(answer, 42)
 
+    @skipBefore35
+    @gen_test
+    def test_async_await(self):
+        class Object(object):
+            def __init__(self):
+                self.executor = futures.thread.ThreadPoolExecutor(1)
+
+            @run_on_executor()
+            def f(self):
+                return 42
+
+        o = Object()
+        namespace = exec_test(globals(), locals(), """
+        async def f():
+            answer = await o.f()
+            return answer
+        """)
+        result = yield namespace['f']()
+        self.assertEqual(result, 42)
+
 
 if __name__ == '__main__':
     unittest.main()