diff --git a/Doc/library/unittest.mock.rst b/Doc/library/unittest.mock.rst
index b768557e6075f6..6c96999be53256 100644
--- a/Doc/library/unittest.mock.rst
+++ b/Doc/library/unittest.mock.rst
@@ -863,7 +863,7 @@ object::
   call is an awaitable.
 
     >>> mock = AsyncMock()
-    >>> asyncio.iscoroutinefunction(mock)
+    >>> inspect.iscoroutinefunction(mock)
     True
     >>> inspect.isawaitable(mock())  # doctest: +SKIP
     True
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index fa00bf9a2ca090..ab9647da50c1d7 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -18,6 +18,7 @@
 import concurrent.futures
 import functools
 import heapq
+import inspect
 import itertools
 import os
 import socket
@@ -766,7 +767,7 @@ def call_soon(self, callback, *args, context=None):
 
     def _check_callback(self, callback, method):
         if (coroutines.iscoroutine(callback) or
-                coroutines.iscoroutinefunction(callback)):
+                inspect.iscoroutinefunction(callback)):
             raise TypeError(
                 f"coroutines cannot be used with {method}()")
         if not callable(callback):
diff --git a/Lib/asyncio/coroutines.py b/Lib/asyncio/coroutines.py
index 7fda0e449d500a..b337df7d77ab94 100644
--- a/Lib/asyncio/coroutines.py
+++ b/Lib/asyncio/coroutines.py
@@ -5,6 +5,7 @@
 import os
 import sys
 import types
+import warnings
 
 
 def _is_debug_mode():
@@ -12,15 +13,21 @@ def _is_debug_mode():
     return sys.flags.dev_mode or (not sys.flags.ignore_environment and
                                   bool(os.environ.get('PYTHONASYNCIODEBUG')))
 
-
 # A marker for iscoroutinefunction.
+# slated for removal in 3.14 see https://github.com/python/cpython/pull/94923/
 _is_coroutine = object()
 
 
 def iscoroutinefunction(func):
     """Return True if func is a decorated coroutine function."""
-    return (inspect.iscoroutinefunction(func) or
-            getattr(func, '_is_coroutine', None) is _is_coroutine)
+    if inspect.iscoroutinefunction(func):
+        return True
+
+    if getattr(func, '_is_coroutine', None) is _is_coroutine:
+        warnings._deprecated("asyncio.coroutines._is_coroutine", remove=(3, 14))
+        return True
+
+    return False
 
 
 # Prioritize native coroutine check to speed-up
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 3952b5f2a7743d..ec410753ce7d79 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -24,7 +24,6 @@
 from . import events
 from . import exceptions
 from . import futures
-from .coroutines import _is_coroutine
 
 # Helper to generate new task names
 # This uses itertools.count() instead of a "+= 1" operation because the latter
@@ -664,11 +663,9 @@ def _ensure_future(coro_or_future, *, loop=None):
             raise ValueError('The future belongs to a different loop than '
                             'the one specified as the loop argument')
         return coro_or_future
-    called_wrap_awaitable = False
     if not coroutines.iscoroutine(coro_or_future):
         if inspect.isawaitable(coro_or_future):
             coro_or_future = _wrap_awaitable(coro_or_future)
-            called_wrap_awaitable = True
         else:
             raise TypeError('An asyncio.Future, a coroutine or an awaitable '
                             'is required')
@@ -678,21 +675,17 @@ def _ensure_future(coro_or_future, *, loop=None):
     try:
         return loop.create_task(coro_or_future)
     except RuntimeError:
-        if not called_wrap_awaitable:
-            coro_or_future.close()
+        coro_or_future.close()
         raise
 
 
-@types.coroutine
-def _wrap_awaitable(awaitable):
+async def _wrap_awaitable(awaitable):
     """Helper for asyncio.ensure_future().
 
     Wraps awaitable (an object with __await__) into a coroutine
     that will later be wrapped in a Task by ensure_future().
     """
-    return (yield from awaitable.__await__())
-
-_wrap_awaitable._is_coroutine = _is_coroutine
+    return await awaitable
 
 
 class _GatheringFuture(futures.Future):
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index cf7683fee64621..aa17341ebb875c 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -2,6 +2,7 @@
 
 import errno
 import io
+import inspect
 import itertools
 import os
 import selectors
@@ -92,7 +93,7 @@ def add_signal_handler(self, sig, callback, *args):
         Raise RuntimeError if there is a problem setting up the handler.
         """
         if (coroutines.iscoroutine(callback) or
-                coroutines.iscoroutinefunction(callback)):
+                inspect.iscoroutinefunction(callback)):
             raise TypeError("coroutines cannot be used "
                             "with add_signal_handler()")
         self._check_signal(sig)
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
index bde4defdf0129e..38a2ff4d7a80ed 100644
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -1,6 +1,6 @@
 """Tests for tasks.py."""
 
-import collections
+import collections.abc
 import contextvars
 import gc
 import io
@@ -286,6 +286,21 @@ async def coro():
         loop.run_until_complete(fut)
         self.assertEqual(fut.result(), 'ok')
 
+    def test_ensure_future_virtual_awaitable(self):
+        @collections.abc.Awaitable.register
+        class Aw:
+            def __init__(self, coro):
+                self.__await__ = coro.__await__
+
+        async def coro():
+            return 'ok'
+
+        loop = asyncio.new_event_loop()
+        self.set_event_loop(loop)
+        fut = asyncio.ensure_future(Aw(coro()), loop=loop)
+        with self.assertRaises(TypeError):
+            loop.run_until_complete(fut)
+
     def test_ensure_future_neither(self):
         with self.assertRaises(TypeError):
             asyncio.ensure_future('ok')
diff --git a/Lib/unittest/mock.py b/Lib/unittest/mock.py
index cd46fea5162aba..0dbfa874bc76dd 100644
--- a/Lib/unittest/mock.py
+++ b/Lib/unittest/mock.py
@@ -31,7 +31,7 @@
 import sys
 import builtins
 import pkgutil
-from asyncio import iscoroutinefunction
+from inspect import iscoroutinefunction
 from types import CodeType, ModuleType, MethodType
 from unittest.util import safe_repr
 from functools import wraps, partial
@@ -197,6 +197,33 @@ def checksig(*args, **kwargs):
     _setup_func(funcopy, mock, sig)
     return funcopy
 
+def _set_async_signature(mock, original, instance=False, is_async_mock=False):
+    # creates an async function with signature (*args, **kwargs) that delegates to a
+    # mock. It still does signature checking by calling a lambda with the same
+    # signature as the original.
+
+    skipfirst = isinstance(original, type)
+    result = _get_signature_object(original, instance, skipfirst)
+    if result is None:
+        return mock
+    func, sig = result
+    def checksig(*args, **kwargs):
+        sig.bind(*args, **kwargs)
+    _copy_func_details(func, checksig)
+
+    name = original.__name__
+    if not name.isidentifier():
+        name = 'funcopy'
+    context = {'_checksig_': checksig, 'mock': mock}
+    src = """async def %s(*args, **kwargs):
+    _checksig_(*args, **kwargs)
+    return await mock(*args, **kwargs)""" % name
+    exec (src, context)
+    funcopy = context[name]
+    _setup_func(funcopy, mock, sig)
+    _setup_async_mock(funcopy)
+    return funcopy
+
 
 def _setup_func(funcopy, mock, sig):
     funcopy.mock = mock
@@ -248,7 +275,6 @@ def reset_mock():
 
 
 def _setup_async_mock(mock):
-    mock._is_coroutine = asyncio.coroutines._is_coroutine
     mock.await_count = 0
     mock.await_args = None
     mock.await_args_list = _CallList()
@@ -2162,13 +2188,6 @@ class AsyncMockMixin(Base):
 
     def __init__(self, /, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        # iscoroutinefunction() checks _is_coroutine property to say if an
-        # object is a coroutine. Without this check it looks to see if it is a
-        # function/method, which in this case it is not (since it is an
-        # AsyncMock).
-        # It is set through __dict__ because when spec_set is True, this
-        # attribute is likely undefined.
-        self.__dict__['_is_coroutine'] = asyncio.coroutines._is_coroutine
         self.__dict__['_mock_await_count'] = 0
         self.__dict__['_mock_await_args'] = None
         self.__dict__['_mock_await_args_list'] = _CallList()
@@ -2681,9 +2700,10 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
     if isinstance(spec, FunctionTypes):
         # should only happen at the top level because we don't
         # recurse for functions
-        mock = _set_signature(mock, spec)
         if is_async_func:
-            _setup_async_mock(mock)
+            mock = _set_async_signature(mock, spec)
+        else:
+            mock = _set_signature(mock, spec)
     else:
         _check_signature(spec, mock, is_type, instance)
 
diff --git a/Misc/NEWS.d/next/Library/2022-07-17-11-06-27.gh-issue-94912.jc4y1V.rst b/Misc/NEWS.d/next/Library/2022-07-17-11-06-27.gh-issue-94912.jc4y1V.rst
new file mode 100644
index 00000000000000..3e844471f07578
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2022-07-17-11-06-27.gh-issue-94912.jc4y1V.rst
@@ -0,0 +1 @@
+make ``asyncio.iscoroutinefunction`` a deprecated alias of ``inspect.iscoroutinefunction`` and remove ``asyncio.coroutines._is_coroutine``