diff --git a/Doc/library/asyncio-task.rst b/Doc/library/asyncio-task.rst
index a1297f5fb7f..e7cf39b2bcc 100644
--- a/Doc/library/asyncio-task.rst
+++ b/Doc/library/asyncio-task.rst
@@ -916,12 +916,13 @@ enforced.
         async def main():
             await old_style_coroutine()
 
-    This decorator is **deprecated** and is scheduled for removal in
-    Python 3.10.
-
     This decorator should not be used for :keyword:`async def`
     coroutines.
 
+    .. deprecated-removed:: 3.8 3.10
+
+       Use :keyword:`async def` instead.
+
 .. function:: iscoroutine(obj)
 
    Return ``True`` if *obj* is a :ref:`coroutine object <coroutine>`.
diff --git a/Lib/asyncio/coroutines.py b/Lib/asyncio/coroutines.py
index c665ebe33ee..9664ea74d75 100644
--- a/Lib/asyncio/coroutines.py
+++ b/Lib/asyncio/coroutines.py
@@ -7,6 +7,7 @@ import os
 import sys
 import traceback
 import types
+import warnings
 
 from . import base_futures
 from . import constants
@@ -107,6 +108,9 @@ def coroutine(func):
     If the coroutine is not yielded from before it is destroyed,
     an error message is logged.
     """
+    warnings.warn('"@coroutine" decorator is deprecated since Python 3.8, use "async def" instead',
+                  DeprecationWarning,
+                  stacklevel=2)
     if inspect.iscoroutinefunction(func):
         # In Python 3.5 that's all we need to do for coroutines
         # defined with "async def".
diff --git a/Lib/asyncio/locks.py b/Lib/asyncio/locks.py
index 639bd11bd06..d59eb8f210c 100644
--- a/Lib/asyncio/locks.py
+++ b/Lib/asyncio/locks.py
@@ -3,12 +3,13 @@
 __all__ = ('Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore')
 
 import collections
+import types
 import warnings
 
 from . import events
 from . import futures
 from . import exceptions
-from .coroutines import coroutine
+from .import coroutines
 
 
 class _ContextManager:
@@ -55,7 +56,7 @@ class _ContextManagerMixin:
         # always raises; that's how the with-statement works.
         pass
 
-    @coroutine
+    @types.coroutine
     def __iter__(self):
         # This is not a coroutine.  It is meant to enable the idiom:
         #
@@ -78,6 +79,9 @@ class _ContextManagerMixin:
         yield from self.acquire()
         return _ContextManager(self)
 
+    # The flag is needed for legacy asyncio.iscoroutine()
+    __iter__._is_coroutine = coroutines._is_coroutine
+
     async def __acquire_ctx(self):
         await self.acquire()
         return _ContextManager(self)
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 211b9126b01..b274b9bd332 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -23,7 +23,7 @@ from . import coroutines
 from . import events
 from . import exceptions
 from . import futures
-from .coroutines import coroutine
+from .coroutines import _is_coroutine
 
 # Helper to generate new task names
 # This uses itertools.count() instead of a "+= 1" operation because the latter
@@ -638,7 +638,7 @@ def ensure_future(coro_or_future, *, loop=None):
                         'required')
 
 
-@coroutine
+@types.coroutine
 def _wrap_awaitable(awaitable):
     """Helper for asyncio.ensure_future().
 
@@ -647,6 +647,8 @@ def _wrap_awaitable(awaitable):
     """
     return (yield from awaitable.__await__())
 
+_wrap_awaitable._is_coroutine = _is_coroutine
+
 
 class _GatheringFuture(futures.Future):
     """Helper for gather().
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
index 25420b2ff6f..27e21b33d36 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -575,9 +575,8 @@ class BaseEventLoopTests(test_utils.TestCase):
     def test_default_exc_handler_coro(self):
         self.loop._process_events = mock.Mock()
 
-        @asyncio.coroutine
-        def zero_error_coro():
-            yield from asyncio.sleep(0.01)
+        async def zero_error_coro():
+            await asyncio.sleep(0.01)
             1/0
 
         # Test Future.__del__
@@ -723,8 +722,7 @@ class BaseEventLoopTests(test_utils.TestCase):
         class MyTask(asyncio.Task):
             pass
 
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             pass
 
         factory = lambda loop, coro: MyTask(coro, loop=loop)
@@ -779,8 +777,7 @@ class BaseEventLoopTests(test_utils.TestCase):
         class MyTask(asyncio.Task):
             pass
 
-        @asyncio.coroutine
-        def test():
+        async def test():
             pass
 
         class EventLoop(base_events.BaseEventLoop):
@@ -830,8 +827,7 @@ class BaseEventLoopTests(test_utils.TestCase):
         # Python issue #22601: ensure that the temporary task created by
         # run_forever() consumes the KeyboardInterrupt and so don't log
         # a warning
-        @asyncio.coroutine
-        def raise_keyboard_interrupt():
+        async def raise_keyboard_interrupt():
             raise KeyboardInterrupt
 
         self.loop._process_events = mock.Mock()
@@ -849,8 +845,7 @@ class BaseEventLoopTests(test_utils.TestCase):
     def test_run_until_complete_baseexception(self):
         # Python issue #22429: run_until_complete() must not schedule a pending
         # call to stop() if the future raised a BaseException
-        @asyncio.coroutine
-        def raise_keyboard_interrupt():
+        async def raise_keyboard_interrupt():
             raise KeyboardInterrupt
 
         self.loop._process_events = mock.Mock()
@@ -1070,9 +1065,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
         class MyProto(asyncio.Protocol):
             pass
 
-        @asyncio.coroutine
-        def getaddrinfo(*args, **kw):
-            yield from []
+        async def getaddrinfo(*args, **kw):
             return [(2, 1, 6, '', ('107.6.106.82', 80)),
                     (2, 1, 6, '', ('107.6.106.82', 80))]
 
@@ -1191,9 +1184,8 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
         self.assertRaises(ValueError, self.loop.run_until_complete, coro)
 
     def test_create_connection_no_getaddrinfo(self):
-        @asyncio.coroutine
-        def getaddrinfo(*args, **kw):
-            yield from []
+        async def getaddrinfo(*args, **kw):
+            return []
 
         def getaddrinfo_task(*args, **kwds):
             return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
@@ -1219,8 +1211,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
             OSError, self.loop.run_until_complete, coro)
 
     def test_create_connection_multiple(self):
-        @asyncio.coroutine
-        def getaddrinfo(*args, **kw):
+        async def getaddrinfo(*args, **kw):
             return [(2, 1, 6, '', ('0.0.0.1', 80)),
                     (2, 1, 6, '', ('0.0.0.2', 80))]
 
@@ -1247,8 +1238,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
 
         m_socket.socket.return_value.bind = bind
 
-        @asyncio.coroutine
-        def getaddrinfo(*args, **kw):
+        async def getaddrinfo(*args, **kw):
             return [(2, 1, 6, '', ('0.0.0.1', 80)),
                     (2, 1, 6, '', ('0.0.0.2', 80))]
 
@@ -1349,8 +1339,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
                 self.loop.run_until_complete(coro)
 
     def test_create_connection_no_local_addr(self):
-        @asyncio.coroutine
-        def getaddrinfo(host, *args, **kw):
+        async def getaddrinfo(host, *args, **kw):
             if host == 'example.com':
                 return [(2, 1, 6, '', ('107.6.106.82', 80)),
                         (2, 1, 6, '', ('107.6.106.82', 80))]
@@ -1488,11 +1477,10 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
         # if host is empty string use None instead
         host = object()
 
-        @asyncio.coroutine
-        def getaddrinfo(*args, **kw):
+        async def getaddrinfo(*args, **kw):
             nonlocal host
             host = args[0]
-            yield from []
+            return []
 
         def getaddrinfo_task(*args, **kwds):
             return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
@@ -1854,9 +1842,10 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
             MyProto, sock, None, None, mock.ANY, mock.ANY)
 
     def test_call_coroutine(self):
-        @asyncio.coroutine
-        def simple_coroutine():
-            pass
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def simple_coroutine():
+                pass
 
         self.loop.set_debug(True)
         coro_func = simple_coroutine
@@ -1880,9 +1869,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
         def stop_loop_cb(loop):
             loop.stop()
 
-        @asyncio.coroutine
-        def stop_loop_coro(loop):
-            yield from ()
+        async def stop_loop_coro(loop):
             loop.stop()
 
         asyncio.set_event_loop(self.loop)
@@ -1909,8 +1896,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
 class RunningLoopTests(unittest.TestCase):
 
     def test_running_loop_within_a_loop(self):
-        @asyncio.coroutine
-        def runner(loop):
+        async def runner(loop):
             loop.run_forever()
 
         loop = asyncio.new_event_loop()
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index b46b614e556..0ae6eab1e1e 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -253,12 +253,10 @@ class EventLoopTestsMixin:
         super().tearDown()
 
     def test_run_until_complete_nesting(self):
-        @asyncio.coroutine
-        def coro1():
-            yield
+        async def coro1():
+            await asyncio.sleep(0)
 
-        @asyncio.coroutine
-        def coro2():
+        async def coro2():
             self.assertTrue(self.loop.is_running())
             self.loop.run_until_complete(coro1())
 
@@ -735,8 +733,7 @@ class EventLoopTestsMixin:
 
     @mock.patch('asyncio.base_events.socket')
     def create_server_multiple_hosts(self, family, hosts, mock_sock):
-        @asyncio.coroutine
-        def getaddrinfo(host, port, *args, **kw):
+        async def getaddrinfo(host, port, *args, **kw):
             if family == socket.AF_INET:
                 return [(family, socket.SOCK_STREAM, 6, '', (host, port))]
             else:
@@ -1662,8 +1659,7 @@ class EventLoopTestsMixin:
             loop.add_writer(w, callback)
 
     def test_close_running_event_loop(self):
-        @asyncio.coroutine
-        def close_loop(loop):
+        async def close_loop(loop):
             self.loop.close()
 
         coro = close_loop(self.loop)
@@ -1673,8 +1669,7 @@ class EventLoopTestsMixin:
     def test_close(self):
         self.loop.close()
 
-        @asyncio.coroutine
-        def test():
+        async def test():
             pass
 
         func = lambda: False
@@ -2142,7 +2137,8 @@ class HandleTests(test_utils.TestCase):
                         '<Handle cancelled>')
 
         # decorated function
-        cb = asyncio.coroutine(noop)
+        with self.assertWarns(DeprecationWarning):
+            cb = asyncio.coroutine(noop)
         h = asyncio.Handle(cb, (), self.loop)
         self.assertEqual(repr(h),
                         '<Handle noop() at %s:%s>'
diff --git a/Lib/test/test_asyncio/test_locks.py b/Lib/test/test_asyncio/test_locks.py
index b61cf743c3c..5063a1da448 100644
--- a/Lib/test/test_asyncio/test_locks.py
+++ b/Lib/test/test_asyncio/test_locks.py
@@ -44,10 +44,11 @@ class LockTests(test_utils.TestCase):
         self.assertTrue(repr(lock).endswith('[unlocked]>'))
         self.assertTrue(RGX_REPR.match(repr(lock)))
 
-        @asyncio.coroutine
-        def acquire_lock():
-            with self.assertWarns(DeprecationWarning):
-                yield from lock
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_lock():
+                with self.assertWarns(DeprecationWarning):
+                    yield from lock
 
         self.loop.run_until_complete(acquire_lock())
         self.assertTrue(repr(lock).endswith('[locked]>'))
@@ -56,10 +57,11 @@ class LockTests(test_utils.TestCase):
     def test_lock(self):
         lock = asyncio.Lock(loop=self.loop)
 
-        @asyncio.coroutine
-        def acquire_lock():
-            with self.assertWarns(DeprecationWarning):
-                return (yield from lock)
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_lock():
+                with self.assertWarns(DeprecationWarning):
+                    return (yield from lock)
 
         res = self.loop.run_until_complete(acquire_lock())
 
@@ -79,17 +81,18 @@ class LockTests(test_utils.TestCase):
             asyncio.BoundedSemaphore(loop=loop),
         ]
 
-        @asyncio.coroutine
-        def test(lock):
-            yield from asyncio.sleep(0.01)
-            self.assertFalse(lock.locked())
-            with self.assertWarns(DeprecationWarning):
-                with (yield from lock) as _lock:
-                    self.assertIs(_lock, None)
-                    self.assertTrue(lock.locked())
-                    yield from asyncio.sleep(0.01)
-                    self.assertTrue(lock.locked())
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def test(lock):
+                yield from asyncio.sleep(0.01)
                 self.assertFalse(lock.locked())
+                with self.assertWarns(DeprecationWarning):
+                    with (yield from lock) as _lock:
+                        self.assertIs(_lock, None)
+                        self.assertTrue(lock.locked())
+                        yield from asyncio.sleep(0.01)
+                        self.assertTrue(lock.locked())
+                    self.assertFalse(lock.locked())
 
         for primitive in primitives:
             loop.run_until_complete(test(primitive))
@@ -290,10 +293,11 @@ class LockTests(test_utils.TestCase):
     def test_context_manager(self):
         lock = asyncio.Lock(loop=self.loop)
 
-        @asyncio.coroutine
-        def acquire_lock():
-            with self.assertWarns(DeprecationWarning):
-                return (yield from lock)
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_lock():
+                with self.assertWarns(DeprecationWarning):
+                    return (yield from lock)
 
         with self.loop.run_until_complete(acquire_lock()):
             self.assertTrue(lock.locked())
@@ -303,10 +307,11 @@ class LockTests(test_utils.TestCase):
     def test_context_manager_cant_reuse(self):
         lock = asyncio.Lock(loop=self.loop)
 
-        @asyncio.coroutine
-        def acquire_lock():
-            with self.assertWarns(DeprecationWarning):
-                return (yield from lock)
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_lock():
+                with self.assertWarns(DeprecationWarning):
+                    return (yield from lock)
 
         # This spells "yield from lock" outside a generator.
         cm = self.loop.run_until_complete(acquire_lock())
@@ -773,10 +778,11 @@ class ConditionTests(test_utils.TestCase):
     def test_context_manager(self):
         cond = asyncio.Condition(loop=self.loop)
 
-        @asyncio.coroutine
-        def acquire_cond():
-            with self.assertWarns(DeprecationWarning):
-                return (yield from cond)
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_cond():
+                with self.assertWarns(DeprecationWarning):
+                    return (yield from cond)
 
         with self.loop.run_until_complete(acquire_cond()):
             self.assertTrue(cond.locked())
@@ -869,10 +875,11 @@ class SemaphoreTests(test_utils.TestCase):
         sem = asyncio.Semaphore(loop=self.loop)
         self.assertEqual(1, sem._value)
 
-        @asyncio.coroutine
-        def acquire_lock():
-            with self.assertWarns(DeprecationWarning):
-                return (yield from sem)
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_lock():
+                with self.assertWarns(DeprecationWarning):
+                    return (yield from sem)
 
         res = self.loop.run_until_complete(acquire_lock())
 
@@ -1012,10 +1019,11 @@ class SemaphoreTests(test_utils.TestCase):
     def test_context_manager(self):
         sem = asyncio.Semaphore(2, loop=self.loop)
 
-        @asyncio.coroutine
-        def acquire_lock():
-            with self.assertWarns(DeprecationWarning):
-                return (yield from sem)
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def acquire_lock():
+                with self.assertWarns(DeprecationWarning):
+                    return (yield from sem)
 
         with self.loop.run_until_complete(acquire_lock()):
             self.assertFalse(sem.locked())
diff --git a/Lib/test/test_asyncio/test_pep492.py b/Lib/test/test_asyncio/test_pep492.py
index 558e268415c..297a3b3901d 100644
--- a/Lib/test/test_asyncio/test_pep492.py
+++ b/Lib/test/test_asyncio/test_pep492.py
@@ -130,9 +130,10 @@ class CoroutineTests(BaseTest):
             def __await__(self):
                 return ('spam',)
 
-        @asyncio.coroutine
-        def func():
-            return Awaitable()
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def func():
+                return Awaitable()
 
         coro = func()
         self.assertEqual(coro.send(None), 'spam')
diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py
index 258d8a7f7fd..fed609816da 100644
--- a/Lib/test/test_asyncio/test_streams.py
+++ b/Lib/test/test_asyncio/test_streams.py
@@ -588,8 +588,7 @@ class StreamTests(test_utils.TestCase):
         stream = asyncio.StreamReader(loop=self.loop,
                                       _asyncio_internal=True)
 
-        @asyncio.coroutine
-        def set_err():
+        async def set_err():
             stream.set_exception(ValueError())
 
         t1 = asyncio.Task(stream.readline(), loop=self.loop)
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
index fa9783f2ff2..1c1f912ff8a 100644
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -28,8 +28,7 @@ def tearDownModule():
     asyncio.set_event_loop_policy(None)
 
 
-@asyncio.coroutine
-def coroutine_function():
+async def coroutine_function():
     pass
 
 
@@ -103,8 +102,7 @@ class BaseTaskTests:
             def __del__(self):
                 gc.collect()
 
-        @asyncio.coroutine
-        def run():
+        async def run():
             return Evil()
 
         self.loop.run_until_complete(
@@ -138,8 +136,7 @@ class BaseTaskTests:
             self.loop.run_until_complete(task)
 
     def test_task_class(self):
-        @asyncio.coroutine
-        def notmuch():
+        async def notmuch():
             return 'ok'
         t = self.new_task(self.loop, notmuch())
         self.loop.run_until_complete(t)
@@ -156,9 +153,10 @@ class BaseTaskTests:
         loop.close()
 
     def test_ensure_future_coroutine(self):
-        @asyncio.coroutine
-        def notmuch():
-            return 'ok'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def notmuch():
+                return 'ok'
         t = asyncio.ensure_future(notmuch(), loop=self.loop)
         self.loop.run_until_complete(t)
         self.assertTrue(t.done())
@@ -194,8 +192,7 @@ class BaseTaskTests:
         self.assertIs(f, f_orig)
 
     def test_ensure_future_task(self):
-        @asyncio.coroutine
-        def notmuch():
+        async def notmuch():
             return 'ok'
         t_orig = self.new_task(self.loop, notmuch())
         t = asyncio.ensure_future(t_orig)
@@ -222,9 +219,10 @@ class BaseTaskTests:
             def __await__(self):
                 return (yield from self.coro)
 
-        @asyncio.coroutine
-        def coro():
-            return 'ok'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro():
+                return 'ok'
 
         loop = asyncio.new_event_loop()
         self.set_event_loop(loop)
@@ -276,9 +274,7 @@ class BaseTaskTests:
     def test_task_repr(self):
         self.loop.set_debug(False)
 
-        @asyncio.coroutine
-        def notmuch():
-            yield from []
+        async def notmuch():
             return 'abc'
 
         # test coroutine function
@@ -327,8 +323,7 @@ class BaseTaskTests:
                          "<Task finished name='TestTask' %s result='abc'>" % coro)
 
     def test_task_repr_autogenerated(self):
-        @asyncio.coroutine
-        def notmuch():
+        async def notmuch():
             return 123
 
         t1 = self.new_task(self.loop, notmuch(), None)
@@ -346,8 +341,7 @@ class BaseTaskTests:
         self.loop.run_until_complete(t2)
 
     def test_task_repr_name_not_str(self):
-        @asyncio.coroutine
-        def notmuch():
+        async def notmuch():
             return 123
 
         t = self.new_task(self.loop, notmuch())
@@ -358,11 +352,12 @@ class BaseTaskTests:
     def test_task_repr_coro_decorator(self):
         self.loop.set_debug(False)
 
-        @asyncio.coroutine
-        def notmuch():
-            # notmuch() function doesn't use yield from: it will be wrapped by
-            # @coroutine decorator
-            return 123
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def notmuch():
+                # notmuch() function doesn't use yield from: it will be wrapped by
+                # @coroutine decorator
+                return 123
 
         # test coroutine function
         self.assertEqual(notmuch.__name__, 'notmuch')
@@ -440,7 +435,8 @@ class BaseTaskTests:
             async def func(x, y):
                 await asyncio.sleep(0)
 
-            partial_func = asyncio.coroutine(functools.partial(func, 1))
+            with self.assertWarns(DeprecationWarning):
+                partial_func = asyncio.coroutine(functools.partial(func, 1))
             task = self.loop.create_task(partial_func(2))
 
             # make warnings quiet
@@ -492,11 +488,12 @@ class BaseTaskTests:
         self.assertFalse(t.cancel())
 
     def test_cancel_yield(self):
-        @asyncio.coroutine
-        def task():
-            yield
-            yield
-            return 12
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def task():
+                yield
+                yield
+                return 12
 
         t = self.new_task(self.loop, task())
         test_utils.run_briefly(self.loop)  # start coro
@@ -618,8 +615,7 @@ class BaseTaskTests:
         loop = asyncio.new_event_loop()
         self.set_event_loop(loop)
 
-        @asyncio.coroutine
-        def task():
+        async def task():
             t.cancel()
             self.assertTrue(t._must_cancel)  # White-box test.
             return 12
@@ -736,8 +732,7 @@ class BaseTaskTests:
 
         foo_started = False
 
-        @asyncio.coroutine
-        def foo():
+        async def foo():
             nonlocal foo_started
             foo_started = True
 
@@ -814,8 +809,7 @@ class BaseTaskTests:
     def test_wait_for_blocking(self):
         loop = self.new_test_loop()
 
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             return 'done'
 
         res = loop.run_until_complete(asyncio.wait_for(coro(), timeout=None))
@@ -976,9 +970,10 @@ class BaseTaskTests:
 
     def test_wait_duplicate_coroutines(self):
 
-        @asyncio.coroutine
-        def coro(s):
-            return s
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro(s):
+                return s
         c = coro('test')
 
         task =self.new_task(
@@ -1036,14 +1031,12 @@ class BaseTaskTests:
         # there is possibility that some tasks in the pending list
         # became done but their callbacks haven't all been called yet
 
-        @asyncio.coroutine
-        def coro1():
-            yield
+        async def coro1():
+            await asyncio.sleep(0)
 
-        @asyncio.coroutine
-        def coro2():
-            yield
-            yield
+        async def coro2():
+            await asyncio.sleep(0)
+            await asyncio.sleep(0)
 
         a = self.new_task(self.loop, coro1())
         b = self.new_task(self.loop, coro2())
@@ -1070,8 +1063,7 @@ class BaseTaskTests:
         # first_exception, task already has exception
         a = self.new_task(loop, asyncio.sleep(10.0))
 
-        @asyncio.coroutine
-        def exc():
+        async def exc():
             raise ZeroDivisionError('err')
 
         b = self.new_task(loop, exc())
@@ -1131,9 +1123,8 @@ class BaseTaskTests:
 
         a = self.new_task(loop, asyncio.sleep(0.1))
 
-        @asyncio.coroutine
-        def sleeper():
-            yield from asyncio.sleep(0.15)
+        async def sleeper():
+            await asyncio.sleep(0.15)
             raise ZeroDivisionError('really')
 
         b = self.new_task(loop, sleeper())
@@ -1220,25 +1211,25 @@ class BaseTaskTests:
         completed = set()
         time_shifted = False
 
-        @asyncio.coroutine
-        def sleeper(dt, x):
-            nonlocal time_shifted
-            yield from asyncio.sleep(dt)
-            completed.add(x)
-            if not time_shifted and 'a' in completed and 'b' in completed:
-                time_shifted = True
-                loop.advance_time(0.14)
-            return x
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def sleeper(dt, x):
+                nonlocal time_shifted
+                yield from  asyncio.sleep(dt)
+                completed.add(x)
+                if not time_shifted and 'a' in completed and 'b' in completed:
+                    time_shifted = True
+                    loop.advance_time(0.14)
+                return x
 
         a = sleeper(0.01, 'a')
         b = sleeper(0.01, 'b')
         c = sleeper(0.15, 'c')
 
-        @asyncio.coroutine
-        def foo():
+        async def foo():
             values = []
             for f in asyncio.as_completed([b, c, a], loop=loop):
-                values.append((yield from f))
+                values.append(await f)
             return values
 
         res = loop.run_until_complete(self.new_task(loop, foo()))
@@ -1350,18 +1341,20 @@ class BaseTaskTests:
 
     def test_as_completed_duplicate_coroutines(self):
 
-        @asyncio.coroutine
-        def coro(s):
-            return s
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro(s):
+                return s
 
-        @asyncio.coroutine
-        def runner():
-            result = []
-            c = coro('ham')
-            for f in asyncio.as_completed([c, c, coro('spam')],
-                                          loop=self.loop):
-                result.append((yield from f))
-            return result
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def runner():
+                result = []
+                c = coro('ham')
+                for f in asyncio.as_completed([c, c, coro('spam')],
+                                              loop=self.loop):
+                    result.append((yield from f))
+                return result
 
         fut = self.new_task(self.loop, runner())
         self.loop.run_until_complete(fut)
@@ -1380,10 +1373,9 @@ class BaseTaskTests:
 
         loop = self.new_test_loop(gen)
 
-        @asyncio.coroutine
-        def sleeper(dt, arg):
-            yield from asyncio.sleep(dt/2)
-            res = yield from asyncio.sleep(dt/2, arg)
+        async def sleeper(dt, arg):
+            await asyncio.sleep(dt/2)
+            res = await asyncio.sleep(dt/2, arg)
             return res
 
         t = self.new_task(loop, sleeper(0.1, 'yeah'))
@@ -1431,16 +1423,14 @@ class BaseTaskTests:
 
         loop = self.new_test_loop(gen)
 
-        @asyncio.coroutine
-        def sleep(dt):
-            yield from asyncio.sleep(dt)
+        async def sleep(dt):
+            await asyncio.sleep(dt)
 
-        @asyncio.coroutine
-        def doit():
+        async def doit():
             sleeper = self.new_task(loop, sleep(5000))
             loop.call_later(0.1, sleeper.cancel)
             try:
-                yield from sleeper
+                await sleeper
             except asyncio.CancelledError:
                 return 'cancelled'
             else:
@@ -1453,9 +1443,8 @@ class BaseTaskTests:
     def test_task_cancel_waiter_future(self):
         fut = self.new_future(self.loop)
 
-        @asyncio.coroutine
-        def coro():
-            yield from fut
+        async def coro():
+            await fut
 
         task = self.new_task(self.loop, coro())
         test_utils.run_briefly(self.loop)
@@ -1469,8 +1458,7 @@ class BaseTaskTests:
         self.assertTrue(fut.cancelled())
 
     def test_task_set_methods(self):
-        @asyncio.coroutine
-        def notmuch():
+        async def notmuch():
             return 'ko'
 
         gen = notmuch()
@@ -1487,11 +1475,12 @@ class BaseTaskTests:
             'ko')
 
     def test_step_result(self):
-        @asyncio.coroutine
-        def notmuch():
-            yield None
-            yield 1
-            return 'ko'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def notmuch():
+                yield None
+                yield 1
+                return 'ko'
 
         self.assertRaises(
             RuntimeError, self.loop.run_until_complete, notmuch())
@@ -1511,10 +1500,9 @@ class BaseTaskTests:
         fut = Fut(loop=self.loop)
         result = None
 
-        @asyncio.coroutine
-        def wait_for_future():
+        async def wait_for_future():
             nonlocal result
-            result = yield from fut
+            result = await fut
 
         t = self.new_task(self.loop, wait_for_future())
         test_utils.run_briefly(self.loop)
@@ -1536,16 +1524,14 @@ class BaseTaskTests:
 
         loop = self.new_test_loop(gen)
 
-        @asyncio.coroutine
-        def sleeper():
-            yield from asyncio.sleep(10)
+        async def sleeper():
+            await asyncio.sleep(10)
 
         base_exc = BaseException()
 
-        @asyncio.coroutine
-        def notmutch():
+        async def notmutch():
             try:
-                yield from sleeper()
+                await sleeper()
             except asyncio.CancelledError:
                 raise base_exc
 
@@ -1571,9 +1557,10 @@ class BaseTaskTests:
             yield
         self.assertFalse(asyncio.iscoroutinefunction(fn1))
 
-        @asyncio.coroutine
-        def fn2():
-            yield
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def fn2():
+                yield
         self.assertTrue(asyncio.iscoroutinefunction(fn2))
 
         self.assertFalse(asyncio.iscoroutinefunction(mock.Mock()))
@@ -1581,9 +1568,10 @@ class BaseTaskTests:
     def test_yield_vs_yield_from(self):
         fut = self.new_future(self.loop)
 
-        @asyncio.coroutine
-        def wait_for_future():
-            yield fut
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def wait_for_future():
+                yield fut
 
         task = wait_for_future()
         with self.assertRaises(RuntimeError):
@@ -1592,17 +1580,19 @@ class BaseTaskTests:
         self.assertFalse(fut.done())
 
     def test_yield_vs_yield_from_generator(self):
-        @asyncio.coroutine
-        def coro():
-            yield
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro():
+                yield
 
-        @asyncio.coroutine
-        def wait_for_future():
-            gen = coro()
-            try:
-                yield gen
-            finally:
-                gen.close()
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def wait_for_future():
+                gen = coro()
+                try:
+                    yield gen
+                finally:
+                    gen.close()
 
         task = wait_for_future()
         self.assertRaises(
@@ -1610,9 +1600,10 @@ class BaseTaskTests:
             self.loop.run_until_complete, task)
 
     def test_coroutine_non_gen_function(self):
-        @asyncio.coroutine
-        def func():
-            return 'test'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def func():
+                return 'test'
 
         self.assertTrue(asyncio.iscoroutinefunction(func))
 
@@ -1625,12 +1616,12 @@ class BaseTaskTests:
     def test_coroutine_non_gen_function_return_future(self):
         fut = self.new_future(self.loop)
 
-        @asyncio.coroutine
-        def func():
-            return fut
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def func():
+                return fut
 
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             fut.set_result('test')
 
         t1 = self.new_task(self.loop, func())
@@ -1887,11 +1878,12 @@ class BaseTaskTests:
             # A function that asserts various things.
             # Called twice, with different debug flag values.
 
-            @asyncio.coroutine
-            def coro():
-                # The actual coroutine.
-                self.assertTrue(gen.gi_running)
-                yield from fut
+            with self.assertWarns(DeprecationWarning):
+                @asyncio.coroutine
+                def coro():
+                    # The actual coroutine.
+                    self.assertTrue(gen.gi_running)
+                    yield from fut
 
             # A completed Future used to run the coroutine.
             fut = self.new_future(self.loop)
@@ -1922,19 +1914,22 @@ class BaseTaskTests:
 
     def test_yield_from_corowrapper(self):
         with set_coroutine_debug(True):
-            @asyncio.coroutine
-            def t1():
-                return (yield from t2())
+            with self.assertWarns(DeprecationWarning):
+                @asyncio.coroutine
+                def t1():
+                    return (yield from t2())
 
-            @asyncio.coroutine
-            def t2():
-                f = self.new_future(self.loop)
-                self.new_task(self.loop, t3(f))
-                return (yield from f)
+            with self.assertWarns(DeprecationWarning):
+                @asyncio.coroutine
+                def t2():
+                    f = self.new_future(self.loop)
+                    self.new_task(self.loop, t3(f))
+                    return (yield from f)
 
-            @asyncio.coroutine
-            def t3(f):
-                f.set_result((1, 2, 3))
+            with self.assertWarns(DeprecationWarning):
+                @asyncio.coroutine
+                def t3(f):
+                    f.set_result((1, 2, 3))
 
             task = self.new_task(self.loop, t1())
             val = self.loop.run_until_complete(task)
@@ -2009,13 +2004,14 @@ class BaseTaskTests:
     def test_log_destroyed_pending_task(self):
         Task = self.__class__.Task
 
-        @asyncio.coroutine
-        def kill_me(loop):
-            future = self.new_future(loop)
-            yield from future
-            # at this point, the only reference to kill_me() task is
-            # the Task._wakeup() method in future._callbacks
-            raise Exception("code never reached")
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def kill_me(loop):
+                future = self.new_future(loop)
+                yield from future
+                # at this point, the only reference to kill_me() task is
+                # the Task._wakeup() method in future._callbacks
+                raise Exception("code never reached")
 
         mock_handler = mock.Mock()
         self.loop.set_debug(True)
@@ -2064,14 +2060,12 @@ class BaseTaskTests:
         loop = asyncio.new_event_loop()
         self.set_event_loop(loop)
 
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             raise TypeError
 
-        @asyncio.coroutine
-        def runner():
+        async def runner():
             task = self.new_task(loop, coro())
-            yield from asyncio.sleep(0.05)
+            await asyncio.sleep(0.05)
             task.cancel()
             task = None
 
@@ -2081,9 +2075,10 @@ class BaseTaskTests:
     @mock.patch('asyncio.coroutines.logger')
     def test_coroutine_never_yielded(self, m_log):
         with set_coroutine_debug(True):
-            @asyncio.coroutine
-            def coro_noop():
-                pass
+            with self.assertWarns(DeprecationWarning):
+                @asyncio.coroutine
+                def coro_noop():
+                    pass
 
         tb_filename = __file__
         tb_lineno = sys._getframe().f_lineno + 2
@@ -2112,13 +2107,15 @@ class BaseTaskTests:
         from @asyncio.coroutine()-wrapped function should have same effect as
         returning generator object or Future."""
         def check():
-            @asyncio.coroutine
-            def outer_coro():
+            with self.assertWarns(DeprecationWarning):
                 @asyncio.coroutine
-                def inner_coro():
-                    return 1
+                def outer_coro():
+                    with self.assertWarns(DeprecationWarning):
+                        @asyncio.coroutine
+                        def inner_coro():
+                            return 1
 
-                return inner_coro()
+                    return inner_coro()
 
             result = self.loop.run_until_complete(outer_coro())
             self.assertEqual(result, 1)
@@ -2147,11 +2144,10 @@ class BaseTaskTests:
         loop = asyncio.new_event_loop()
         self.addCleanup(loop.close)
 
-        @asyncio.coroutine
-        def blocking_coroutine():
+        async def blocking_coroutine():
             fut = self.new_future(loop)
             # Block: fut result is never set
-            yield from fut
+            await fut
 
         task = loop.create_task(blocking_coroutine())
 
@@ -2230,14 +2226,12 @@ class BaseTaskTests:
     def test_exception_traceback(self):
         # See http://bugs.python.org/issue28843
 
-        @asyncio.coroutine
-        def foo():
+        async def foo():
             1 / 0
 
-        @asyncio.coroutine
-        def main():
+        async def main():
             task = self.new_task(self.loop, foo())
-            yield  # skip one loop iteration
+            await asyncio.sleep(0)  # skip one loop iteration
             self.assertIsNotNone(task.exception().__traceback__)
 
         self.loop.run_until_complete(main())
@@ -2248,9 +2242,10 @@ class BaseTaskTests:
             raise ValueError
         self.loop.call_soon = call_soon
 
-        @asyncio.coroutine
-        def coro():
-            pass
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro():
+                pass
 
         self.assertFalse(m_log.error.called)
 
@@ -2280,9 +2275,10 @@ class BaseTaskTests:
 
     def test_create_task_with_oldstyle_coroutine(self):
 
-        @asyncio.coroutine
-        def coro():
-            pass
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro():
+                pass
 
         task = self.new_task(self.loop, coro())
         self.assertIsInstance(task, self.Task)
@@ -2553,8 +2549,7 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
     @support.refcount_test
     def test_refleaks_in_task___init__(self):
         gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             pass
         task = self.new_task(self.loop, coro())
         self.loop.run_until_complete(task)
@@ -2565,8 +2560,7 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
         self.assertAlmostEqual(gettotalrefcount() - refs_before, 0, delta=10)
 
     def test_del__log_destroy_pending_segfault(self):
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             pass
         task = self.new_task(self.loop, coro())
         self.loop.run_until_complete(task)
@@ -3054,15 +3048,13 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
     def wrap_futures(self, *futures):
         coros = []
         for fut in futures:
-            @asyncio.coroutine
-            def coro(fut=fut):
-                return (yield from fut)
+            async def coro(fut=fut):
+                return await fut
             coros.append(coro())
         return coros
 
     def test_constructor_loop_selection(self):
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             return 'abc'
         gen1 = coro()
         gen2 = coro()
@@ -3078,9 +3070,10 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
         self.other_loop.run_until_complete(fut2)
 
     def test_duplicate_coroutines(self):
-        @asyncio.coroutine
-        def coro(s):
-            return s
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro(s):
+                return s
         c = coro('abc')
         fut = asyncio.gather(c, c, coro('def'), c, loop=self.one_loop)
         self._run_loop(self.one_loop)
@@ -3091,21 +3084,19 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
         proof = 0
         waiter = asyncio.Future(loop=self.one_loop)
 
-        @asyncio.coroutine
-        def inner():
+        async def inner():
             nonlocal proof
-            yield from waiter
+            await waiter
             proof += 1
 
         child1 = asyncio.ensure_future(inner(), loop=self.one_loop)
         child2 = asyncio.ensure_future(inner(), loop=self.one_loop)
         gatherer = None
 
-        @asyncio.coroutine
-        def outer():
+        async def outer():
             nonlocal proof, gatherer
             gatherer = asyncio.gather(child1, child2, loop=self.one_loop)
-            yield from gatherer
+            await gatherer
             proof += 100
 
         f = asyncio.ensure_future(outer(), loop=self.one_loop)
@@ -3123,17 +3114,15 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
     def test_exception_marking(self):
         # Test for the first line marked "Mark exception retrieved."
 
-        @asyncio.coroutine
-        def inner(f):
-            yield from f
+        async def inner(f):
+            await f
             raise RuntimeError('should not be ignored')
 
         a = asyncio.Future(loop=self.one_loop)
         b = asyncio.Future(loop=self.one_loop)
 
-        @asyncio.coroutine
-        def outer():
-            yield from asyncio.gather(inner(a), inner(b), loop=self.one_loop)
+        async def outer():
+            await asyncio.gather(inner(a), inner(b), loop=self.one_loop)
 
         f = asyncio.ensure_future(outer(), loop=self.one_loop)
         test_utils.run_briefly(self.one_loop)
@@ -3152,15 +3141,14 @@ class RunCoroutineThreadsafeTests(test_utils.TestCase):
         self.loop = asyncio.new_event_loop()
         self.set_event_loop(self.loop) # Will cleanup properly
 
-    @asyncio.coroutine
-    def add(self, a, b, fail=False, cancel=False):
+    async def add(self, a, b, fail=False, cancel=False):
         """Wait 0.05 second and return a + b."""
-        yield from asyncio.sleep(0.05)
+        await asyncio.sleep(0.05)
         if fail:
             raise RuntimeError("Fail!")
         if cancel:
             asyncio.current_task(self.loop).cancel()
-            yield
+            await asyncio.sleep(0)
         return a + b
 
     def target(self, fail=False, cancel=False, timeout=None,
@@ -3261,11 +3249,10 @@ class SleepTests(test_utils.TestCase):
             nonlocal result
             result += num
 
-        @asyncio.coroutine
-        def coro():
+        async def coro():
             self.loop.call_soon(inc_result, 1)
             self.assertEqual(result, 0)
-            num = yield from asyncio.sleep(0, result=10)
+            num = await asyncio.sleep(0, result=10)
             self.assertEqual(result, 1) # inc'ed by call_soon
             inc_result(num) # num should be 11
 
@@ -3318,24 +3305,27 @@ class CompatibilityTests(test_utils.TestCase):
 
     def test_yield_from_awaitable(self):
 
-        @asyncio.coroutine
-        def coro():
-            yield from asyncio.sleep(0)
-            return 'ok'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro():
+                yield from asyncio.sleep(0)
+                return 'ok'
 
         result = self.loop.run_until_complete(coro())
         self.assertEqual('ok', result)
 
     def test_await_old_style_coro(self):
 
-        @asyncio.coroutine
-        def coro1():
-            return 'ok1'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro1():
+                return 'ok1'
 
-        @asyncio.coroutine
-        def coro2():
-            yield from asyncio.sleep(0)
-            return 'ok2'
+        with self.assertWarns(DeprecationWarning):
+            @asyncio.coroutine
+            def coro2():
+                yield from asyncio.sleep(0)
+                return 'ok2'
 
         async def inner():
             return await asyncio.gather(coro1(), coro2(), loop=self.loop)
diff --git a/Lib/test/test_typing.py b/Lib/test/test_typing.py
index 0d66ebbd184..a547fe274c8 100644
--- a/Lib/test/test_typing.py
+++ b/Lib/test/test_typing.py
@@ -1708,9 +1708,8 @@ class AsyncIteratorWrapper(typing.AsyncIterator[T_a]):
     def __aiter__(self) -> typing.AsyncIterator[T_a]:
         return self
 
-    @asyncio.coroutine
-    def __anext__(self) -> T_a:
-        data = yield from self.value
+    async def __anext__(self) -> T_a:
+        data = await self.value
         if data:
             return data
         else:
diff --git a/Misc/NEWS.d/next/Library/2019-05-15-21-35-23.bpo-36921.kA1306.rst b/Misc/NEWS.d/next/Library/2019-05-15-21-35-23.bpo-36921.kA1306.rst
new file mode 100644
index 00000000000..b443b379d8f
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2019-05-15-21-35-23.bpo-36921.kA1306.rst
@@ -0,0 +1 @@
+Deprecate ``@coroutine`` for sake of ``async def``.