bpo-36373: Deprecate explicit loop in task and subprocess API (GH-16033)

(cherry picked from commit a488879cbaf4b8b52699cadccf73bb4c271bcb29)

Co-authored-by: Andrew Svetlov <andrew.svetlov@gmail.com>

7 files changed, 178 insertions, 101 deletions

diff --git a/Doc/library/asyncio-subprocess.rst b/Doc/library/asyncio-subprocess.rst
index 444fb6361b..bd92257c82 100644
--- a/Doc/library/asyncio-subprocess.rst
+++ b/Doc/library/asyncio-subprocess.rst
@@ -71,6 +71,10 @@ Creating Subprocesses
    See the documentation of :meth:`loop.subprocess_exec` for other
    parameters.
 
+   .. deprecated-removed:: 3.8 3.10
+
+      The *loop* parameter.
+
 .. coroutinefunction:: create_subprocess_shell(cmd, stdin=None, \
                           stdout=None, stderr=None, loop=None, \
                           limit=None, \*\*kwds)
@@ -95,6 +99,10 @@ Creating Subprocesses
    escape whitespace and special shell characters in strings that are going
    to be used to construct shell commands.
 
+   .. deprecated-removed:: 3.8 3.10
+
+      The *loop* parameter.
+
 .. note::
 
    The default asyncio event loop implementation on **Windows** does not
diff --git a/Doc/library/asyncio-task.rst b/Doc/library/asyncio-task.rst
index 1fcdcb985d..57e0e07ad3 100644
--- a/Doc/library/asyncio-task.rst
+++ b/Doc/library/asyncio-task.rst
@@ -334,6 +334,9 @@ Running Tasks Concurrently
    cancellation of one submitted Task/Future to cause other
    Tasks/Futures to be cancelled.
 
+   .. deprecated-removed:: 3.8 3.10
+      The *loop* parameter.
+
    .. _asyncio_example_gather:
 
    Example::
@@ -411,6 +414,9 @@ Shielding From Cancellation
        except CancelledError:
            res = None
 
+   .. deprecated-removed:: 3.8 3.10
+      The *loop* parameter.
+
 
 Timeouts
 ========
@@ -478,22 +484,12 @@ Waiting Primitives
    set concurrently and block until the condition specified
    by *return_when*.
 
-   .. deprecated:: 3.8
-
-      If any awaitable in *aws* is a coroutine, it is automatically
-      scheduled as a Task.  Passing coroutines objects to
-      ``wait()`` directly is deprecated as it leads to
-      :ref:`confusing behavior <asyncio_example_wait_coroutine>`.
-
    Returns two sets of Tasks/Futures: ``(done, pending)``.
 
    Usage::
 
         done, pending = await asyncio.wait(aws)
 
-   .. deprecated-removed:: 3.8 3.10
-      The *loop* parameter.
-
    *timeout* (a float or int), if specified, can be used to control
    the maximum number of seconds to wait before returning.
 
@@ -525,6 +521,17 @@ Waiting Primitives
    Unlike :func:`~asyncio.wait_for`, ``wait()`` does not cancel the
    futures when a timeout occurs.
 
+   .. deprecated:: 3.8
+
+      If any awaitable in *aws* is a coroutine, it is automatically
+      scheduled as a Task.  Passing coroutines objects to
+      ``wait()`` directly is deprecated as it leads to
+      :ref:`confusing behavior <asyncio_example_wait_coroutine>`.
+
+   .. deprecated-removed:: 3.8 3.10
+
+      The *loop* parameter.
+
    .. _asyncio_example_wait_coroutine:
    .. note::
 
@@ -568,6 +575,9 @@ Waiting Primitives
    Raises :exc:`asyncio.TimeoutError` if the timeout occurs before
    all Futures are done.
 
+   .. deprecated-removed:: 3.8 3.10
+      The *loop* parameter.
+
    Example::
 
        for f in as_completed(aws):
@@ -694,6 +704,9 @@ Task Object
    .. versionchanged:: 3.8
       Added the ``name`` parameter.
 
+   .. deprecated-removed:: 3.8 3.10
+      The *loop* parameter.
+
    .. method:: cancel()
 
       Request the Task to be cancelled.
diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py
index e6c1c8b61c..ce504b8b0c 100644
--- a/Lib/asyncio/subprocess.py
+++ b/Lib/asyncio/subprocess.py
@@ -224,6 +224,13 @@ async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
                                   **kwds):
     if loop is None:
         loop = events.get_event_loop()
+    else:
+        warnings.warn("The loop argument is deprecated since Python 3.8 "
+                      "and scheduled for removal in Python 3.10.",
+                      DeprecationWarning,
+                      stacklevel=2
+        )
+
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
                                                         loop=loop,
                                                         _asyncio_internal=True)
@@ -239,6 +246,12 @@ async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
                                  limit=streams._DEFAULT_LIMIT, **kwds):
     if loop is None:
         loop = events.get_event_loop()
+    else:
+        warnings.warn("The loop argument is deprecated since Python 3.8 "
+                      "and scheduled for removal in Python 3.10.",
+                      DeprecationWarning,
+                      stacklevel=2
+        )
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
                                                         loop=loop,
                                                         _asyncio_internal=True)
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index cd4832cfee..a0cb884eac 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -573,10 +573,17 @@ def as_completed(fs, *, loop=None, timeout=None):
     """
     if futures.isfuture(fs) or coroutines.iscoroutine(fs):
         raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
-    loop = loop if loop is not None else events.get_event_loop()
-    todo = {ensure_future(f, loop=loop) for f in set(fs)}
+
     from .queues import Queue  # Import here to avoid circular import problem.
     done = Queue(loop=loop)
+
+    if loop is None:
+        loop = events.get_event_loop()
+    else:
+        warnings.warn("The loop argument is deprecated since Python 3.8, "
+                      "and scheduled for removal in Python 3.10.",
+                      DeprecationWarning, stacklevel=2)
+    todo = {ensure_future(f, loop=loop) for f in set(fs)}
     timeout_handle = None
 
     def _on_timeout():
@@ -733,6 +740,10 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False):
     if not coros_or_futures:
         if loop is None:
             loop = events.get_event_loop()
+        else:
+            warnings.warn("The loop argument is deprecated since Python 3.8, "
+                          "and scheduled for removal in Python 3.10.",
+                          DeprecationWarning, stacklevel=2)
         outer = loop.create_future()
         outer.set_result([])
         return outer
@@ -842,6 +853,10 @@ def shield(arg, *, loop=None):
         except CancelledError:
             res = None
     """
+    if loop is not None:
+        warnings.warn("The loop argument is deprecated since Python 3.8, "
+                      "and scheduled for removal in Python 3.10.",
+                      DeprecationWarning, stacklevel=2)
     inner = ensure_future(arg, loop=loop)
     if inner.done():
         # Shortcut.
diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py
index 60fe52fa4b..6325ee3983 100644
--- a/Lib/test/test_asyncio/test_streams.py
+++ b/Lib/test/test_asyncio/test_streams.py
@@ -855,9 +855,10 @@ os.close(fd)
         watcher.attach_loop(self.loop)
         try:
             asyncio.set_child_watcher(watcher)
-            create = asyncio.create_subprocess_exec(*args,
-                                                    pass_fds={wfd},
-                                                    loop=self.loop)
+            create = asyncio.create_subprocess_exec(
+                *args,
+                pass_fds={wfd},
+            )
             proc = self.loop.run_until_complete(create)
             self.loop.run_until_complete(proc.wait())
         finally:
diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py
index 2cfe6576d6..3ad18e5c51 100644
--- a/Lib/test/test_asyncio/test_subprocess.py
+++ b/Lib/test/test_asyncio/test_subprocess.py
@@ -109,10 +109,10 @@ class SubprocessMixin:
 
         async def run(data):
             proc = await asyncio.create_subprocess_exec(
-                                          *args,
-                                          stdin=subprocess.PIPE,
-                                          stdout=subprocess.PIPE,
-                                          loop=self.loop)
+                *args,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+            )
 
             # feed data
             proc.stdin.write(data)
@@ -135,10 +135,10 @@ class SubprocessMixin:
 
         async def run(data):
             proc = await asyncio.create_subprocess_exec(
-                                          *args,
-                                          stdin=subprocess.PIPE,
-                                          stdout=subprocess.PIPE,
-                                          loop=self.loop)
+                *args,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+            )
             stdout, stderr = await proc.communicate(data)
             return proc.returncode, stdout
 
@@ -149,25 +149,28 @@ class SubprocessMixin:
         self.assertEqual(stdout, b'some data')
 
     def test_shell(self):
-        create = asyncio.create_subprocess_shell('exit 7',
-                                                 loop=self.loop)
-        proc = self.loop.run_until_complete(create)
+        proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_shell('exit 7')
+        )
         exitcode = self.loop.run_until_complete(proc.wait())
         self.assertEqual(exitcode, 7)
 
     def test_start_new_session(self):
         # start the new process in a new session
-        create = asyncio.create_subprocess_shell('exit 8',
-                                                 start_new_session=True,
-                                                 loop=self.loop)
-        proc = self.loop.run_until_complete(create)
+        proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_shell(
+                'exit 8',
+                start_new_session=True,
+            )
+        )
         exitcode = self.loop.run_until_complete(proc.wait())
         self.assertEqual(exitcode, 8)
 
     def test_kill(self):
         args = PROGRAM_BLOCKED
-        create = asyncio.create_subprocess_exec(*args, loop=self.loop)
-        proc = self.loop.run_until_complete(create)
+        proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_exec(*args)
+        )
         proc.kill()
         returncode = self.loop.run_until_complete(proc.wait())
         if sys.platform == 'win32':
@@ -178,8 +181,9 @@ class SubprocessMixin:
 
     def test_terminate(self):
         args = PROGRAM_BLOCKED
-        create = asyncio.create_subprocess_exec(*args, loop=self.loop)
-        proc = self.loop.run_until_complete(create)
+        proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_exec(*args)
+        )
         proc.terminate()
         returncode = self.loop.run_until_complete(proc.wait())
         if sys.platform == 'win32':
@@ -197,10 +201,12 @@ class SubprocessMixin:
         try:
             code = 'import time; print("sleeping", flush=True); time.sleep(3600)'
             args = [sys.executable, '-c', code]
-            create = asyncio.create_subprocess_exec(*args,
-                                                    stdout=subprocess.PIPE,
-                                                    loop=self.loop)
-            proc = self.loop.run_until_complete(create)
+            proc = self.loop.run_until_complete(
+                asyncio.create_subprocess_exec(
+                    *args,
+                    stdout=subprocess.PIPE,
+                )
+            )
 
             async def send_signal(proc):
                 # basic synchronization to wait until the program is sleeping
@@ -221,11 +227,13 @@ class SubprocessMixin:
         large_data = b'x' * support.PIPE_MAX_SIZE
 
         # the program ends before the stdin can be feeded
-        create = asyncio.create_subprocess_exec(
-                             sys.executable, '-c', 'pass',
-                             stdin=subprocess.PIPE,
-                             loop=self.loop)
-        proc = self.loop.run_until_complete(create)
+        proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_exec(
+                sys.executable, '-c', 'pass',
+                stdin=subprocess.PIPE,
+            )
+        )
+
         return (proc, large_data)
 
     def test_stdin_broken_pipe(self):
@@ -273,11 +281,11 @@ class SubprocessMixin:
             self.loop.connect_read_pipe = connect_read_pipe_mock
 
             proc = await asyncio.create_subprocess_exec(
-                                         sys.executable, '-c', code,
-                                         stdin=asyncio.subprocess.PIPE,
-                                         stdout=asyncio.subprocess.PIPE,
-                                         limit=limit,
-                                         loop=self.loop)
+                sys.executable, '-c', code,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                limit=limit,
+            )
             stdout_transport = proc._transport.get_pipe_transport(1)
 
             stdout, stderr = await proc.communicate()
@@ -301,12 +309,12 @@ class SubprocessMixin:
         async def len_message(message):
             code = 'import sys; data = sys.stdin.read(); print(len(data))'
             proc = await asyncio.create_subprocess_exec(
-                                          sys.executable, '-c', code,
-                                          stdin=asyncio.subprocess.PIPE,
-                                          stdout=asyncio.subprocess.PIPE,
-                                          stderr=asyncio.subprocess.PIPE,
-                                          close_fds=False,
-                                          loop=self.loop)
+                sys.executable, '-c', code,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                close_fds=False,
+            )
             stdout, stderr = await proc.communicate(message)
             exitcode = await proc.wait()
             return (stdout, exitcode)
@@ -320,12 +328,12 @@ class SubprocessMixin:
         async def empty_input():
             code = 'import sys; data = sys.stdin.read(); print(len(data))'
             proc = await asyncio.create_subprocess_exec(
-                                          sys.executable, '-c', code,
-                                          stdin=asyncio.subprocess.PIPE,
-                                          stdout=asyncio.subprocess.PIPE,
-                                          stderr=asyncio.subprocess.PIPE,
-                                          close_fds=False,
-                                          loop=self.loop)
+                sys.executable, '-c', code,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                close_fds=False,
+            )
             stdout, stderr = await proc.communicate(b'')
             exitcode = await proc.wait()
             return (stdout, exitcode)
@@ -339,12 +347,12 @@ class SubprocessMixin:
         async def empty_input():
             code = 'import sys; data = sys.stdin.read(); print(len(data))'
             proc = await asyncio.create_subprocess_exec(
-                                          sys.executable, '-c', code,
-                                          stdin=asyncio.subprocess.DEVNULL,
-                                          stdout=asyncio.subprocess.PIPE,
-                                          stderr=asyncio.subprocess.PIPE,
-                                          close_fds=False,
-                                          loop=self.loop)
+                sys.executable, '-c', code,
+                stdin=asyncio.subprocess.DEVNULL,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                close_fds=False,
+            )
             stdout, stderr = await proc.communicate()
             exitcode = await proc.wait()
             return (stdout, exitcode)
@@ -358,12 +366,12 @@ class SubprocessMixin:
         async def empty_output():
             code = 'import sys; data = sys.stdin.read(); print(len(data))'
             proc = await asyncio.create_subprocess_exec(
-                                          sys.executable, '-c', code,
-                                          stdin=asyncio.subprocess.PIPE,
-                                          stdout=asyncio.subprocess.DEVNULL,
-                                          stderr=asyncio.subprocess.PIPE,
-                                          close_fds=False,
-                                          loop=self.loop)
+                sys.executable, '-c', code,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.DEVNULL,
+                stderr=asyncio.subprocess.PIPE,
+                close_fds=False,
+            )
             stdout, stderr = await proc.communicate(b"abc")
             exitcode = await proc.wait()
             return (stdout, exitcode)
@@ -377,12 +385,12 @@ class SubprocessMixin:
         async def empty_error():
             code = 'import sys; data = sys.stdin.read(); print(len(data))'
             proc = await asyncio.create_subprocess_exec(
-                                          sys.executable, '-c', code,
-                                          stdin=asyncio.subprocess.PIPE,
-                                          stdout=asyncio.subprocess.PIPE,
-                                          stderr=asyncio.subprocess.DEVNULL,
-                                          close_fds=False,
-                                          loop=self.loop)
+                sys.executable, '-c', code,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.DEVNULL,
+                close_fds=False,
+            )
             stdout, stderr = await proc.communicate(b"abc")
             exitcode = await proc.wait()
             return (stderr, exitcode)
@@ -395,9 +403,7 @@ class SubprocessMixin:
         # Issue #23140: cancel Process.wait()
 
         async def cancel_wait():
-            proc = await asyncio.create_subprocess_exec(
-                                          *PROGRAM_BLOCKED,
-                                          loop=self.loop)
+            proc = await asyncio.create_subprocess_exec(*PROGRAM_BLOCKED)
 
             # Create an internal future waiting on the process exit
             task = self.loop.create_task(proc.wait())
@@ -419,8 +425,7 @@ class SubprocessMixin:
     def test_cancel_make_subprocess_transport_exec(self):
 
         async def cancel_make_transport():
-            coro = asyncio.create_subprocess_exec(*PROGRAM_BLOCKED,
-                                                  loop=self.loop)
+            coro = asyncio.create_subprocess_exec(*PROGRAM_BLOCKED)
             task = self.loop.create_task(coro)
 
             self.loop.call_soon(task.cancel)
@@ -524,7 +529,7 @@ class SubprocessMixin:
                 isinstance(self, SubprocessFastWatcherTests)):
             asyncio.get_child_watcher()._callbacks.clear()
 
-    def _test_popen_error(self, stdin):
+    async def _test_popen_error(self, stdin):
         if sys.platform == 'win32':
             target = 'asyncio.windows_utils.Popen'
         else:
@@ -533,23 +538,26 @@ class SubprocessMixin:
             exc = ZeroDivisionError
             popen.side_effect = exc
 
-            create = asyncio.create_subprocess_exec(sys.executable, '-c',
-                                                    'pass', stdin=stdin,
-                                                    loop=self.loop)
             with warnings.catch_warnings(record=True) as warns:
                 with self.assertRaises(exc):
-                    self.loop.run_until_complete(create)
+                    await asyncio.create_subprocess_exec(
+                        sys.executable,
+                        '-c',
+                        'pass',
+                        stdin=stdin
+                    )
                 self.assertEqual(warns, [])
 
     def test_popen_error(self):
         # Issue #24763: check that the subprocess transport is closed
         # when BaseSubprocessTransport fails
-        self._test_popen_error(stdin=None)
+        self.loop.run_until_complete(self._test_popen_error(stdin=None))
 
     def test_popen_error_with_stdin_pipe(self):
         # Issue #35721: check that newly created socket pair is closed when
         # Popen fails
-        self._test_popen_error(stdin=subprocess.PIPE)
+        self.loop.run_until_complete(
+            self._test_popen_error(stdin=subprocess.PIPE))
 
     def test_read_stdout_after_process_exit(self):
 
@@ -560,12 +568,11 @@ class SubprocessMixin:
                               'sys.stdout.flush()',
                               'sys.exit(1)'])
 
-            fut = asyncio.create_subprocess_exec(
+            process = await asyncio.create_subprocess_exec(
                 sys.executable, '-c', code,
                 stdout=asyncio.subprocess.PIPE,
-                loop=self.loop)
+            )
 
-            process = await fut
             while True:
                 data = await process.stdout.read(65536)
                 if data:
@@ -620,7 +627,6 @@ class SubprocessMixin:
 
         self.loop.run_until_complete(execute())
 
-
     def test_create_subprocess_exec_with_path(self):
         async def execute():
             p = await subprocess.create_subprocess_exec(
@@ -632,6 +638,26 @@ class SubprocessMixin:
 
         self.assertIsNone(self.loop.run_until_complete(execute()))
 
+    def test_exec_loop_deprecated(self):
+        async def go():
+            with self.assertWarns(DeprecationWarning):
+                proc = await asyncio.create_subprocess_exec(
+                    sys.executable, '-c', 'pass',
+                    loop=self.loop,
+                )
+            await proc.wait()
+        self.loop.run_until_complete(go())
+
+    def test_shell_loop_deprecated(self):
+        async def go():
+            with self.assertWarns(DeprecationWarning):
+                proc = await asyncio.create_subprocess_shell(
+                    "exit 0",
+                    loop=self.loop,
+                )
+            await proc.wait()
+        self.loop.run_until_complete(go())
+
 
 if sys.platform != 'win32':
     # Unix
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
index 3fbb2a1a16..6e832eab6d 100644
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -1810,7 +1810,7 @@ class BaseTaskTests:
 
         async def outer():
             nonlocal proof
-            await asyncio.shield(inner(), loop=self.loop)
+            await asyncio.shield(inner())
             proof += 100
 
         f = asyncio.ensure_future(outer(), loop=self.loop)
@@ -1825,8 +1825,8 @@ class BaseTaskTests:
     def test_shield_gather(self):
         child1 = self.new_future(self.loop)
         child2 = self.new_future(self.loop)
-        parent = asyncio.gather(child1, child2, loop=self.loop)
-        outer = asyncio.shield(parent, loop=self.loop)
+        parent = asyncio.gather(child1, child2)
+        outer = asyncio.shield(parent)
         test_utils.run_briefly(self.loop)
         outer.cancel()
         test_utils.run_briefly(self.loop)
@@ -1839,9 +1839,9 @@ class BaseTaskTests:
     def test_gather_shield(self):
         child1 = self.new_future(self.loop)
         child2 = self.new_future(self.loop)
-        inner1 = asyncio.shield(child1, loop=self.loop)
-        inner2 = asyncio.shield(child2, loop=self.loop)
-        parent = asyncio.gather(inner1, inner2, loop=self.loop)
+        inner1 = asyncio.shield(child1)
+        inner2 = asyncio.shield(child2)
+        parent = asyncio.gather(inner1, inner2)
         test_utils.run_briefly(self.loop)
         parent.cancel()
         # This should cancel inner1 and inner2 but bot child1 and child2.
@@ -2981,7 +2981,8 @@ class FutureGatherTests(GatherTestsBase, test_utils.TestCase):
         self._run_loop(self.one_loop)
         self.assertTrue(fut.done())
         self.assertEqual(fut.result(), [])
-        fut = asyncio.gather(*seq_or_iter, loop=self.other_loop)
+        with self.assertWarns(DeprecationWarning):
+            fut = asyncio.gather(*seq_or_iter, loop=self.other_loop)
         self.assertIs(fut._loop, self.other_loop)
 
     def test_constructor_empty_sequence(self):