bpo-42392: Remove loop parameter form asyncio locks and Queue (#23420)

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

10 files changed, 304 insertions, 451 deletions

diff --git a/Lib/asyncio/locks.py b/Lib/asyncio/locks.py
index f1ce732..6f322c2 100644
--- a/Lib/asyncio/locks.py
+++ b/Lib/asyncio/locks.py
@@ -3,10 +3,9 @@
 __all__ = ('Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore')
 
 import collections
-import warnings
 
-from . import events
 from . import exceptions
+from . import mixins
 
 
 class _ContextManagerMixin:
@@ -20,7 +19,7 @@ class _ContextManagerMixin:
         self.release()
 
 
-class Lock(_ContextManagerMixin):
+class Lock(_ContextManagerMixin, mixins._LoopBoundedMixin):
     """Primitive lock objects.
 
     A primitive lock is a synchronization primitive that is not owned
@@ -74,16 +73,9 @@ class Lock(_ContextManagerMixin):
 
     """
 
-    def __init__(self, *, loop=None):
+    def __init__(self):
         self._waiters = None
         self._locked = False
-        if loop is None:
-            self._loop = events.get_event_loop()
-        else:
-            self._loop = loop
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
 
     def __repr__(self):
         res = super().__repr__()
@@ -109,7 +101,7 @@ class Lock(_ContextManagerMixin):
 
         if self._waiters is None:
             self._waiters = collections.deque()
-        fut = self._loop.create_future()
+        fut = self._get_loop().create_future()
         self._waiters.append(fut)
 
         # Finally block should be called before the CancelledError
@@ -161,7 +153,7 @@ class Lock(_ContextManagerMixin):
             fut.set_result(True)
 
 
-class Event:
+class Event(mixins._LoopBoundedMixin):
     """Asynchronous equivalent to threading.Event.
 
     Class implementing event objects. An event manages a flag that can be set
@@ -170,16 +162,9 @@ class Event:
     false.
     """
 
-    def __init__(self, *, loop=None):
+    def __init__(self):
         self._waiters = collections.deque()
         self._value = False
-        if loop is None:
-            self._loop = events.get_event_loop()
-        else:
-            self._loop = loop
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
 
     def __repr__(self):
         res = super().__repr__()
@@ -220,7 +205,7 @@ class Event:
         if self._value:
             return True
 
-        fut = self._loop.create_future()
+        fut = self._get_loop().create_future()
         self._waiters.append(fut)
         try:
             await fut
@@ -229,7 +214,7 @@ class Event:
             self._waiters.remove(fut)
 
 
-class Condition(_ContextManagerMixin):
+class Condition(_ContextManagerMixin, mixins._LoopBoundedMixin):
     """Asynchronous equivalent to threading.Condition.
 
     This class implements condition variable objects. A condition variable
@@ -239,18 +224,10 @@ class Condition(_ContextManagerMixin):
     A new Lock object is created and used as the underlying lock.
     """
 
-    def __init__(self, lock=None, *, loop=None):
-        if loop is None:
-            self._loop = events.get_event_loop()
-        else:
-            self._loop = loop
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
-
+    def __init__(self, lock=None):
         if lock is None:
-            lock = Lock(loop=loop)
-        elif lock._loop is not self._loop:
+            lock = Lock()
+        elif lock._loop is not self._get_loop():
             raise ValueError("loop argument must agree with lock")
 
         self._lock = lock
@@ -284,7 +261,7 @@ class Condition(_ContextManagerMixin):
 
         self.release()
         try:
-            fut = self._loop.create_future()
+            fut = self._get_loop().create_future()
             self._waiters.append(fut)
             try:
                 await fut
@@ -351,7 +328,7 @@ class Condition(_ContextManagerMixin):
         self.notify(len(self._waiters))
 
 
-class Semaphore(_ContextManagerMixin):
+class Semaphore(_ContextManagerMixin, mixins._LoopBoundedMixin):
     """A Semaphore implementation.
 
     A semaphore manages an internal counter which is decremented by each
@@ -366,18 +343,11 @@ class Semaphore(_ContextManagerMixin):
     ValueError is raised.
     """
 
-    def __init__(self, value=1, *, loop=None):
+    def __init__(self, value=1):
         if value < 0:
             raise ValueError("Semaphore initial value must be >= 0")
         self._value = value
         self._waiters = collections.deque()
-        if loop is None:
-            self._loop = events.get_event_loop()
-        else:
-            self._loop = loop
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
 
     def __repr__(self):
         res = super().__repr__()
@@ -407,7 +377,7 @@ class Semaphore(_ContextManagerMixin):
         True.
         """
         while self._value <= 0:
-            fut = self._loop.create_future()
+            fut = self._get_loop().create_future()
             self._waiters.append(fut)
             try:
                 await fut
@@ -436,14 +406,9 @@ class BoundedSemaphore(Semaphore):
     above the initial value.
     """
 
-    def __init__(self, value=1, *, loop=None):
-        if loop:
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
-
+    def __init__(self, value=1):
         self._bound_value = value
-        super().__init__(value, loop=loop)
+        super().__init__(value)
 
     def release(self):
         if self._value >= self._bound_value:
diff --git a/Lib/asyncio/mixins.py b/Lib/asyncio/mixins.py
new file mode 100644
index 0000000..dbc4b5f
--- /dev/null
+++ b/Lib/asyncio/mixins.py
@@ -0,0 +1,21 @@
+"""Event loop mixins."""
+
+import threading
+from . import events
+
+_global_lock = threading.Lock()
+
+
+class _LoopBoundedMixin:
+    _loop = None
+
+    def _get_loop(self):
+        loop = events._get_running_loop()
+
+        if self._loop is None:
+            with _global_lock:
+                if self._loop is None:
+                    self._loop = loop
+        if loop is not self._loop:
+            raise RuntimeError(f'{type(self).__name__} have already bounded to another loop')
+        return loop
diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py
index cd3f7c6..78ae9e9 100644
--- a/Lib/asyncio/queues.py
+++ b/Lib/asyncio/queues.py
@@ -2,10 +2,9 @@ __all__ = ('Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty')
 
 import collections
 import heapq
-import warnings
 
-from . import events
 from . import locks
+from . import mixins
 
 
 class QueueEmpty(Exception):
@@ -18,7 +17,7 @@ class QueueFull(Exception):
     pass
 
 
-class Queue:
+class Queue(mixins._LoopBoundedMixin):
     """A queue, useful for coordinating producer and consumer coroutines.
 
     If maxsize is less than or equal to zero, the queue size is infinite. If it
@@ -30,14 +29,7 @@ class Queue:
     interrupted between calling qsize() and doing an operation on the Queue.
     """
 
-    def __init__(self, maxsize=0, *, loop=None):
-        if loop is None:
-            self._loop = events.get_event_loop()
-        else:
-            self._loop = loop
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
+    def __init__(self, maxsize=0):
         self._maxsize = maxsize
 
         # Futures.
@@ -45,7 +37,7 @@ class Queue:
         # Futures.
         self._putters = collections.deque()
         self._unfinished_tasks = 0
-        self._finished = locks.Event(loop=loop)
+        self._finished = locks.Event()
         self._finished.set()
         self._init(maxsize)
 
@@ -122,7 +114,7 @@ class Queue:
         slot is available before adding item.
         """
         while self.full():
-            putter = self._loop.create_future()
+            putter = self._get_loop().create_future()
             self._putters.append(putter)
             try:
                 await putter
@@ -160,7 +152,7 @@ class Queue:
         If queue is empty, wait until an item is available.
         """
         while self.empty():
-            getter = self._loop.create_future()
+            getter = self._get_loop().create_future()
             self._getters.append(getter)
             try:
                 await getter
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index f486b67..03d8451 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -578,7 +578,7 @@ def as_completed(fs, *, loop=None, timeout=None):
         raise TypeError(f"expect an iterable of futures, not {type(fs).__name__}")
 
     from .queues import Queue  # Import here to avoid circular import problem.
-    done = Queue(loop=loop)
+    done = Queue()
 
     if loop is None:
         loop = events.get_event_loop()
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index b8fe466..7f76011 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -206,8 +206,8 @@ class MySubprocessProtocol(asyncio.SubprocessProtocol):
         self.disconnects = {fd: loop.create_future() for fd in range(3)}
         self.data = {1: b'', 2: b''}
         self.returncode = None
-        self.got_data = {1: asyncio.Event(loop=loop),
-                         2: asyncio.Event(loop=loop)}
+        self.got_data = {1: asyncio.Event(),
+                         2: asyncio.Event()}
 
     def connection_made(self, transport):
         self.transport = transport
@@ -1740,20 +1740,20 @@ class SubprocessTestsMixin:
         connect = self.loop.subprocess_exec(
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
-            self.assertEqual('CONNECTED', proto.state)
 
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'Python The Winner')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            with test_utils.disable_logger():
-                transp.close()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
-            self.assertEqual(b'Python The Winner', proto.data[1])
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
+        self.assertEqual('CONNECTED', proto.state)
+
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'Python The Winner')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        with test_utils.disable_logger():
+            transp.close()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
+        self.assertEqual(b'Python The Winner', proto.data[1])
 
     def test_subprocess_interactive(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo.py')
@@ -1762,51 +1762,48 @@ class SubprocessTestsMixin:
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
-            self.assertEqual('CONNECTED', proto.state)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
+        self.assertEqual('CONNECTED', proto.state)
 
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'Python ')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            proto.got_data[1].clear()
-            self.assertEqual(b'Python ', proto.data[1])
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'Python ')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        proto.got_data[1].clear()
+        self.assertEqual(b'Python ', proto.data[1])
 
-            stdin.write(b'The Winner')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            self.assertEqual(b'Python The Winner', proto.data[1])
+        stdin.write(b'The Winner')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        self.assertEqual(b'Python The Winner', proto.data[1])
 
-            with test_utils.disable_logger():
-                transp.close()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
+        with test_utils.disable_logger():
+            transp.close()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
 
     def test_subprocess_shell(self):
-        with self.assertWarns(DeprecationWarning):
-            connect = self.loop.subprocess_shell(
-                            functools.partial(MySubprocessProtocol, self.loop),
-                            'echo Python')
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
+        connect = self.loop.subprocess_shell(
+                        functools.partial(MySubprocessProtocol, self.loop),
+                        'echo Python')
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            transp.get_pipe_transport(0).close()
-            self.loop.run_until_complete(proto.completed)
-            self.assertEqual(0, proto.returncode)
-            self.assertTrue(all(f.done() for f in proto.disconnects.values()))
-            self.assertEqual(proto.data[1].rstrip(b'\r\n'), b'Python')
-            self.assertEqual(proto.data[2], b'')
-            transp.close()
+        transp.get_pipe_transport(0).close()
+        self.loop.run_until_complete(proto.completed)
+        self.assertEqual(0, proto.returncode)
+        self.assertTrue(all(f.done() for f in proto.disconnects.values()))
+        self.assertEqual(proto.data[1].rstrip(b'\r\n'), b'Python')
+        self.assertEqual(proto.data[2], b'')
+        transp.close()
 
     def test_subprocess_exitcode(self):
         connect = self.loop.subprocess_shell(
                         functools.partial(MySubprocessProtocol, self.loop),
                         'exit 7', stdin=None, stdout=None, stderr=None)
 
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
+        transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.completed)
         self.assertEqual(7, proto.returncode)
@@ -1816,8 +1813,8 @@ class SubprocessTestsMixin:
         connect = self.loop.subprocess_shell(
                         functools.partial(MySubprocessProtocol, self.loop),
                         'exit 7', stdin=None, stdout=None, stderr=None)
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
+
+        transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.assertIsNone(transp.get_pipe_transport(0))
         self.assertIsNone(transp.get_pipe_transport(1))
@@ -1833,15 +1830,14 @@ class SubprocessTestsMixin:
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            transp.kill()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
-            transp.close()
+        transp.kill()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
+        transp.close()
 
     def test_subprocess_terminate(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo.py')
@@ -1850,15 +1846,14 @@ class SubprocessTestsMixin:
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            transp.terminate()
-            self.loop.run_until_complete(proto.completed)
-            self.check_terminated(proto.returncode)
-            transp.close()
+        transp.terminate()
+        self.loop.run_until_complete(proto.completed)
+        self.check_terminated(proto.returncode)
+        transp.close()
 
     @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
     def test_subprocess_send_signal(self):
@@ -1873,15 +1868,15 @@ class SubprocessTestsMixin:
                             functools.partial(MySubprocessProtocol, self.loop),
                             sys.executable, prog)
 
-            with self.assertWarns(DeprecationWarning):
-                transp, proto = self.loop.run_until_complete(connect)
-                self.assertIsInstance(proto, MySubprocessProtocol)
-                self.loop.run_until_complete(proto.connected)
 
-                transp.send_signal(signal.SIGHUP)
-                self.loop.run_until_complete(proto.completed)
-                self.assertEqual(-signal.SIGHUP, proto.returncode)
-                transp.close()
+            transp, proto = self.loop.run_until_complete(connect)
+            self.assertIsInstance(proto, MySubprocessProtocol)
+            self.loop.run_until_complete(proto.connected)
+
+            transp.send_signal(signal.SIGHUP)
+            self.loop.run_until_complete(proto.completed)
+            self.assertEqual(-signal.SIGHUP, proto.returncode)
+            transp.close()
         finally:
             signal.signal(signal.SIGHUP, old_handler)
 
@@ -1892,20 +1887,19 @@ class SubprocessTestsMixin:
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'test')
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'test')
 
-            self.loop.run_until_complete(proto.completed)
+        self.loop.run_until_complete(proto.completed)
 
-            transp.close()
-            self.assertEqual(b'OUT:test', proto.data[1])
-            self.assertTrue(proto.data[2].startswith(b'ERR:test'), proto.data[2])
-            self.assertEqual(0, proto.returncode)
+        transp.close()
+        self.assertEqual(b'OUT:test', proto.data[1])
+        self.assertTrue(proto.data[2].startswith(b'ERR:test'), proto.data[2])
+        self.assertEqual(0, proto.returncode)
 
     def test_subprocess_stderr_redirect_to_stdout(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo2.py')
@@ -1914,23 +1908,23 @@ class SubprocessTestsMixin:
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog, stderr=subprocess.STDOUT)
 
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
 
-            stdin = transp.get_pipe_transport(0)
-            self.assertIsNotNone(transp.get_pipe_transport(1))
-            self.assertIsNone(transp.get_pipe_transport(2))
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            stdin.write(b'test')
-            self.loop.run_until_complete(proto.completed)
-            self.assertTrue(proto.data[1].startswith(b'OUT:testERR:test'),
-                            proto.data[1])
-            self.assertEqual(b'', proto.data[2])
+        stdin = transp.get_pipe_transport(0)
+        self.assertIsNotNone(transp.get_pipe_transport(1))
+        self.assertIsNone(transp.get_pipe_transport(2))
 
-            transp.close()
-            self.assertEqual(0, proto.returncode)
+        stdin.write(b'test')
+        self.loop.run_until_complete(proto.completed)
+        self.assertTrue(proto.data[1].startswith(b'OUT:testERR:test'),
+                        proto.data[1])
+        self.assertEqual(b'', proto.data[2])
+
+        transp.close()
+        self.assertEqual(0, proto.returncode)
 
     def test_subprocess_close_client_stream(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo3.py')
@@ -1938,33 +1932,33 @@ class SubprocessTestsMixin:
         connect = self.loop.subprocess_exec(
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
-        with self.assertWarns(DeprecationWarning):
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
 
-            stdin = transp.get_pipe_transport(0)
-            stdout = transp.get_pipe_transport(1)
-            stdin.write(b'test')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            self.assertEqual(b'OUT:test', proto.data[1])
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            stdout.close()
-            self.loop.run_until_complete(proto.disconnects[1])
-            stdin.write(b'xxx')
-            self.loop.run_until_complete(proto.got_data[2].wait())
-            if sys.platform != 'win32':
-                self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
-            else:
-                # After closing the read-end of a pipe, writing to the
-                # write-end using os.write() fails with errno==EINVAL and
-                # GetLastError()==ERROR_INVALID_NAME on Windows!?!  (Using
-                # WriteFile() we get ERROR_BROKEN_PIPE as expected.)
-                self.assertEqual(b'ERR:OSError', proto.data[2])
-            with test_utils.disable_logger():
-                transp.close()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
+        stdin = transp.get_pipe_transport(0)
+        stdout = transp.get_pipe_transport(1)
+        stdin.write(b'test')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        self.assertEqual(b'OUT:test', proto.data[1])
+
+        stdout.close()
+        self.loop.run_until_complete(proto.disconnects[1])
+        stdin.write(b'xxx')
+        self.loop.run_until_complete(proto.got_data[2].wait())
+        if sys.platform != 'win32':
+            self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
+        else:
+            # After closing the read-end of a pipe, writing to the
+            # write-end using os.write() fails with errno==EINVAL and
+            # GetLastError()==ERROR_INVALID_NAME on Windows!?!  (Using
+            # WriteFile() we get ERROR_BROKEN_PIPE as expected.)
+            self.assertEqual(b'ERR:OSError', proto.data[2])
+        with test_utils.disable_logger():
+            transp.close()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
 
     def test_subprocess_wait_no_same_group(self):
         # start the new process in a new session
diff --git a/Lib/test/test_asyncio/test_locks.py b/Lib/test/test_asyncio/test_locks.py
index 8c93fae..6c34ef6 100644
--- a/Lib/test/test_asyncio/test_locks.py
+++ b/Lib/test/test_asyncio/test_locks.py
@@ -26,24 +26,8 @@ class LockTests(test_utils.TestCase):
         super().setUp()
         self.loop = self.new_test_loop()
 
-    def test_ctor_loop(self):
-        loop = mock.Mock()
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=loop)
-        self.assertIs(lock._loop, loop)
-
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
-        self.assertIs(lock._loop, self.loop)
-
-    def test_ctor_noloop(self):
-        asyncio.set_event_loop(self.loop)
-        lock = asyncio.Lock()
-        self.assertIs(lock._loop, self.loop)
-
     def test_repr(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
         self.assertTrue(repr(lock).endswith('[unlocked]>'))
         self.assertTrue(RGX_REPR.match(repr(lock)))
 
@@ -52,9 +36,9 @@ class LockTests(test_utils.TestCase):
         self.assertTrue(RGX_REPR.match(repr(lock)))
 
     def test_lock(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
 
+        with self.assertWarns(DeprecationWarning):
             @asyncio.coroutine
             def acquire_lock():
                 return (yield from lock)
@@ -70,14 +54,14 @@ class LockTests(test_utils.TestCase):
     def test_lock_by_with_statement(self):
         loop = asyncio.new_event_loop()  # don't use TestLoop quirks
         self.set_event_loop(loop)
-        with self.assertWarns(DeprecationWarning):
-            primitives = [
-                asyncio.Lock(loop=loop),
-                asyncio.Condition(loop=loop),
-                asyncio.Semaphore(loop=loop),
-                asyncio.BoundedSemaphore(loop=loop),
-            ]
+        primitives = [
+            asyncio.Lock(),
+            asyncio.Condition(),
+            asyncio.Semaphore(),
+            asyncio.BoundedSemaphore(),
+        ]
 
+        with self.assertWarns(DeprecationWarning):
             @asyncio.coroutine
             def test(lock):
                 yield from asyncio.sleep(0.01)
@@ -95,8 +79,7 @@ class LockTests(test_utils.TestCase):
             self.assertFalse(primitive.locked())
 
     def test_acquire(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
         result = []
 
         self.assertTrue(self.loop.run_until_complete(lock.acquire()))
@@ -147,8 +130,7 @@ class LockTests(test_utils.TestCase):
         self.assertTrue(t3.result())
 
     def test_acquire_cancel(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
         self.assertTrue(self.loop.run_until_complete(lock.acquire()))
 
         task = self.loop.create_task(lock.acquire())
@@ -173,8 +155,7 @@ class LockTests(test_utils.TestCase):
         # B's waiter; instead, it should move on to C's waiter.
 
         # Setup: A has the lock, b and c are waiting.
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
 
         async def lockit(name, blocker):
             await lock.acquire()
@@ -210,8 +191,7 @@ class LockTests(test_utils.TestCase):
         # Issue 32734
         # Acquire 4 locks, cancel second, release first
         # and 2 locks are taken at once.
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
         lock_count = 0
         call_count = 0
 
@@ -256,8 +236,7 @@ class LockTests(test_utils.TestCase):
         self.assertTrue(t3.cancelled())
 
     def test_finished_waiter_cancelled(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
 
         ta = self.loop.create_task(lock.acquire())
         test_utils.run_briefly(self.loop)
@@ -279,14 +258,12 @@ class LockTests(test_utils.TestCase):
         self.assertTrue(tb.cancelled())
 
     def test_release_not_acquired(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
 
         self.assertRaises(RuntimeError, lock.release)
 
     def test_release_no_waiters(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+        lock = asyncio.Lock()
         self.loop.run_until_complete(lock.acquire())
         self.assertTrue(lock.locked())
 
@@ -312,24 +289,8 @@ class EventTests(test_utils.TestCase):
         super().setUp()
         self.loop = self.new_test_loop()
 
-    def test_ctor_loop(self):
-        loop = mock.Mock()
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=loop)
-        self.assertIs(ev._loop, loop)
-
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
-        self.assertIs(ev._loop, self.loop)
-
-    def test_ctor_noloop(self):
-        asyncio.set_event_loop(self.loop)
-        ev = asyncio.Event()
-        self.assertIs(ev._loop, self.loop)
-
     def test_repr(self):
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
+        ev = asyncio.Event()
         self.assertTrue(repr(ev).endswith('[unset]>'))
         match = RGX_REPR.match(repr(ev))
         self.assertEqual(match.group('extras'), 'unset')
@@ -343,8 +304,7 @@ class EventTests(test_utils.TestCase):
         self.assertTrue(RGX_REPR.match(repr(ev)))
 
     def test_wait(self):
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
+        ev = asyncio.Event()
         self.assertFalse(ev.is_set())
 
         result = []
@@ -381,16 +341,14 @@ class EventTests(test_utils.TestCase):
         self.assertIsNone(t3.result())
 
     def test_wait_on_set(self):
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
+        ev = asyncio.Event()
         ev.set()
 
         res = self.loop.run_until_complete(ev.wait())
         self.assertTrue(res)
 
     def test_wait_cancel(self):
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
+        ev = asyncio.Event()
 
         wait = self.loop.create_task(ev.wait())
         self.loop.call_soon(wait.cancel)
@@ -400,8 +358,7 @@ class EventTests(test_utils.TestCase):
         self.assertFalse(ev._waiters)
 
     def test_clear(self):
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
+        ev = asyncio.Event()
         self.assertFalse(ev.is_set())
 
         ev.set()
@@ -411,8 +368,7 @@ class EventTests(test_utils.TestCase):
         self.assertFalse(ev.is_set())
 
     def test_clear_with_waiters(self):
-        with self.assertWarns(DeprecationWarning):
-            ev = asyncio.Event(loop=self.loop)
+        ev = asyncio.Event()
         result = []
 
         async def c1(result):
@@ -446,23 +402,8 @@ class ConditionTests(test_utils.TestCase):
         super().setUp()
         self.loop = self.new_test_loop()
 
-    def test_ctor_loop(self):
-        loop = mock.Mock()
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=loop)
-            self.assertIs(cond._loop, loop)
-
-            cond = asyncio.Condition(loop=self.loop)
-            self.assertIs(cond._loop, self.loop)
-
-    def test_ctor_noloop(self):
-        asyncio.set_event_loop(self.loop)
-        cond = asyncio.Condition()
-        self.assertIs(cond._loop, self.loop)
-
     def test_wait(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         result = []
 
         async def c1(result):
@@ -525,8 +466,7 @@ class ConditionTests(test_utils.TestCase):
         self.assertTrue(t3.result())
 
     def test_wait_cancel(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         self.loop.run_until_complete(cond.acquire())
 
         wait = self.loop.create_task(cond.wait())
@@ -538,8 +478,7 @@ class ConditionTests(test_utils.TestCase):
         self.assertTrue(cond.locked())
 
     def test_wait_cancel_contested(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
 
         self.loop.run_until_complete(cond.acquire())
         self.assertTrue(cond.locked())
@@ -565,10 +504,11 @@ class ConditionTests(test_utils.TestCase):
 
     def test_wait_cancel_after_notify(self):
         # See bpo-32841
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
         waited = False
 
+        cond = asyncio.Condition()
+        cond._loop = self.loop
+
         async def wait_on_cond():
             nonlocal waited
             async with cond:
@@ -590,15 +530,13 @@ class ConditionTests(test_utils.TestCase):
         self.assertTrue(waited)
 
     def test_wait_unacquired(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         self.assertRaises(
             RuntimeError,
             self.loop.run_until_complete, cond.wait())
 
     def test_wait_for(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         presult = False
 
         def predicate():
@@ -635,8 +573,7 @@ class ConditionTests(test_utils.TestCase):
         self.assertTrue(t.result())
 
     def test_wait_for_unacquired(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
 
         # predicate can return true immediately
         res = self.loop.run_until_complete(cond.wait_for(lambda: [1, 2, 3]))
@@ -648,8 +585,7 @@ class ConditionTests(test_utils.TestCase):
             cond.wait_for(lambda: False))
 
     def test_notify(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         result = []
 
         async def c1(result):
@@ -701,8 +637,7 @@ class ConditionTests(test_utils.TestCase):
         self.assertTrue(t3.result())
 
     def test_notify_all(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
 
         result = []
 
@@ -738,18 +673,15 @@ class ConditionTests(test_utils.TestCase):
         self.assertTrue(t2.result())
 
     def test_notify_unacquired(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         self.assertRaises(RuntimeError, cond.notify)
 
     def test_notify_all_unacquired(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         self.assertRaises(RuntimeError, cond.notify_all)
 
     def test_repr(self):
-        with self.assertWarns(DeprecationWarning):
-            cond = asyncio.Condition(loop=self.loop)
+        cond = asyncio.Condition()
         self.assertTrue('unlocked' in repr(cond))
         self.assertTrue(RGX_REPR.match(repr(cond)))
 
@@ -775,9 +707,8 @@ class ConditionTests(test_utils.TestCase):
         self.loop.run_until_complete(f())
 
     def test_explicit_lock(self):
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
-            cond = asyncio.Condition(lock, loop=self.loop)
+        lock = asyncio.Lock()
+        cond = asyncio.Condition(lock)
 
         self.assertIs(cond._lock, lock)
         self.assertIs(cond._loop, lock._loop)
@@ -785,23 +716,27 @@ class ConditionTests(test_utils.TestCase):
     def test_ambiguous_loops(self):
         loop = self.new_test_loop()
         self.addCleanup(loop.close)
-        with self.assertWarns(DeprecationWarning):
-            lock = asyncio.Lock(loop=self.loop)
+
+        lock = asyncio.Lock()
+        lock._loop = loop
+
+        async def _create_condition():
             with self.assertRaises(ValueError):
-                asyncio.Condition(lock, loop=loop)
+                asyncio.Condition(lock)
+
+        self.loop.run_until_complete(_create_condition())
 
     def test_timeout_in_block(self):
         loop = asyncio.new_event_loop()
         self.addCleanup(loop.close)
 
         async def task_timeout():
-            condition = asyncio.Condition(loop=loop)
+            condition = asyncio.Condition()
             async with condition:
                 with self.assertRaises(asyncio.TimeoutError):
                     await asyncio.wait_for(condition.wait(), timeout=0.5)
 
-        with self.assertWarns(DeprecationWarning):
-            loop.run_until_complete(task_timeout())
+        loop.run_until_complete(task_timeout())
 
 
 class SemaphoreTests(test_utils.TestCase):
@@ -810,29 +745,12 @@ class SemaphoreTests(test_utils.TestCase):
         super().setUp()
         self.loop = self.new_test_loop()
 
-    def test_ctor_loop(self):
-        loop = mock.Mock()
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(loop=loop)
-        self.assertIs(sem._loop, loop)
-
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(loop=self.loop)
-        self.assertIs(sem._loop, self.loop)
-
-    def test_ctor_noloop(self):
-        asyncio.set_event_loop(self.loop)
-        sem = asyncio.Semaphore()
-        self.assertIs(sem._loop, self.loop)
-
     def test_initial_value_zero(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(0, loop=self.loop)
+        sem = asyncio.Semaphore(0)
         self.assertTrue(sem.locked())
 
     def test_repr(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(loop=self.loop)
+        sem = asyncio.Semaphore()
         self.assertTrue(repr(sem).endswith('[unlocked, value:1]>'))
         self.assertTrue(RGX_REPR.match(repr(sem)))
 
@@ -850,8 +768,7 @@ class SemaphoreTests(test_utils.TestCase):
         self.assertTrue(RGX_REPR.match(repr(sem)))
 
     def test_semaphore(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(loop=self.loop)
+        sem = asyncio.Semaphore()
         self.assertEqual(1, sem._value)
 
         with self.assertWarns(DeprecationWarning):
@@ -872,8 +789,7 @@ class SemaphoreTests(test_utils.TestCase):
         self.assertRaises(ValueError, asyncio.Semaphore, -1)
 
     def test_acquire(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(3, loop=self.loop)
+        sem = asyncio.Semaphore(3)
         result = []
 
         self.assertTrue(self.loop.run_until_complete(sem.acquire()))
@@ -934,8 +850,7 @@ class SemaphoreTests(test_utils.TestCase):
         self.loop.run_until_complete(asyncio.gather(*race_tasks))
 
     def test_acquire_cancel(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(loop=self.loop)
+        sem = asyncio.Semaphore()
         self.loop.run_until_complete(sem.acquire())
 
         acquire = self.loop.create_task(sem.acquire())
@@ -947,8 +862,7 @@ class SemaphoreTests(test_utils.TestCase):
                         all(waiter.done() for waiter in sem._waiters))
 
     def test_acquire_cancel_before_awoken(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(value=0, loop=self.loop)
+        sem = asyncio.Semaphore(value=0)
 
         t1 = self.loop.create_task(sem.acquire())
         t2 = self.loop.create_task(sem.acquire())
@@ -970,8 +884,7 @@ class SemaphoreTests(test_utils.TestCase):
         test_utils.run_briefly(self.loop)
 
     def test_acquire_hang(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(value=0, loop=self.loop)
+        sem = asyncio.Semaphore(value=0)
 
         t1 = self.loop.create_task(sem.acquire())
         t2 = self.loop.create_task(sem.acquire())
@@ -985,14 +898,12 @@ class SemaphoreTests(test_utils.TestCase):
         self.assertTrue(sem.locked())
 
     def test_release_not_acquired(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.BoundedSemaphore(loop=self.loop)
+        sem = asyncio.BoundedSemaphore()
 
         self.assertRaises(ValueError, sem.release)
 
     def test_release_no_waiters(self):
-        with self.assertWarns(DeprecationWarning):
-            sem = asyncio.Semaphore(loop=self.loop)
+        sem = asyncio.Semaphore()
         self.loop.run_until_complete(sem.acquire())
         self.assertTrue(sem.locked())
 
diff --git a/Lib/test/test_asyncio/test_pep492.py b/Lib/test/test_asyncio/test_pep492.py
index c5e3a5c..4bd50f4 100644
--- a/Lib/test/test_asyncio/test_pep492.py
+++ b/Lib/test/test_asyncio/test_pep492.py
@@ -43,13 +43,12 @@ class BaseTest(test_utils.TestCase):
 class LockTests(BaseTest):
 
     def test_context_manager_async_with(self):
-        with self.assertWarns(DeprecationWarning):
-            primitives = [
-                asyncio.Lock(loop=self.loop),
-                asyncio.Condition(loop=self.loop),
-                asyncio.Semaphore(loop=self.loop),
-                asyncio.BoundedSemaphore(loop=self.loop),
-            ]
+        primitives = [
+            asyncio.Lock(),
+            asyncio.Condition(),
+            asyncio.Semaphore(),
+            asyncio.BoundedSemaphore(),
+        ]
 
         async def test(lock):
             await asyncio.sleep(0.01)
@@ -66,13 +65,12 @@ class LockTests(BaseTest):
             self.assertFalse(primitive.locked())
 
     def test_context_manager_with_await(self):
-        with self.assertWarns(DeprecationWarning):
-            primitives = [
-                asyncio.Lock(loop=self.loop),
-                asyncio.Condition(loop=self.loop),
-                asyncio.Semaphore(loop=self.loop),
-                asyncio.BoundedSemaphore(loop=self.loop),
-            ]
+        primitives = [
+            asyncio.Lock(),
+            asyncio.Condition(),
+            asyncio.Semaphore(),
+            asyncio.BoundedSemaphore(),
+        ]
 
         async def test(lock):
             await asyncio.sleep(0.01)
diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py
index 5c9aaa8..dc97673 100644
--- a/Lib/test/test_asyncio/test_queues.py
+++ b/Lib/test/test_asyncio/test_queues.py
@@ -35,14 +35,13 @@ class QueueBasicTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=loop)
+        q = asyncio.Queue()
         self.assertTrue(fn(q).startswith('<Queue'), fn(q))
         id_is_present = hex(id(q)) in fn(q)
         self.assertEqual(expect_id, id_is_present)
 
         async def add_getter():
-            q = asyncio.Queue(loop=loop)
+            q = asyncio.Queue()
             # Start a task that waits to get.
             loop.create_task(q.get())
             # Let it start waiting.
@@ -51,11 +50,10 @@ class QueueBasicTests(_QueueTestBase):
             # resume q.get coroutine to finish generator
             q.put_nowait(0)
 
-        with self.assertWarns(DeprecationWarning):
-            loop.run_until_complete(add_getter())
+        loop.run_until_complete(add_getter())
 
         async def add_putter():
-            q = asyncio.Queue(maxsize=1, loop=loop)
+            q = asyncio.Queue(maxsize=1)
             q.put_nowait(1)
             # Start a task that waits to put.
             loop.create_task(q.put(2))
@@ -65,27 +63,11 @@ class QueueBasicTests(_QueueTestBase):
             # resume q.put coroutine to finish generator
             q.get_nowait()
 
-        with self.assertWarns(DeprecationWarning):
-            loop.run_until_complete(add_putter())
-            q = asyncio.Queue(loop=loop)
+        loop.run_until_complete(add_putter())
+        q = asyncio.Queue()
         q.put_nowait(1)
         self.assertTrue('_queue=[1]' in fn(q))
 
-    def test_ctor_loop(self):
-        loop = mock.Mock()
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=loop)
-        self.assertIs(q._loop, loop)
-
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
-        self.assertIs(q._loop, self.loop)
-
-    def test_ctor_noloop(self):
-        asyncio.set_event_loop(self.loop)
-        q = asyncio.Queue()
-        self.assertIs(q._loop, self.loop)
-
     def test_repr(self):
         self._test_repr_or_str(repr, True)
 
@@ -93,8 +75,7 @@ class QueueBasicTests(_QueueTestBase):
         self._test_repr_or_str(str, False)
 
     def test_empty(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         self.assertTrue(q.empty())
         q.put_nowait(1)
         self.assertFalse(q.empty())
@@ -102,18 +83,15 @@ class QueueBasicTests(_QueueTestBase):
         self.assertTrue(q.empty())
 
     def test_full(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         self.assertFalse(q.full())
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(maxsize=1, loop=self.loop)
+        q = asyncio.Queue(maxsize=1)
         q.put_nowait(1)
         self.assertTrue(q.full())
 
     def test_order(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         for i in [1, 3, 2]:
             q.put_nowait(i)
 
@@ -131,8 +109,7 @@ class QueueBasicTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(maxsize=2, loop=loop)
+        q = asyncio.Queue(maxsize=2)
         self.assertEqual(2, q.maxsize)
         have_been_put = []
 
@@ -166,8 +143,7 @@ class QueueBasicTests(_QueueTestBase):
 class QueueGetTests(_QueueTestBase):
 
     def test_blocking_get(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         q.put_nowait(1)
 
         async def queue_get():
@@ -177,8 +153,7 @@ class QueueGetTests(_QueueTestBase):
         self.assertEqual(1, res)
 
     def test_get_with_putters(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(1, loop=self.loop)
+        q = asyncio.Queue(1)
         q.put_nowait(1)
 
         waiter = self.loop.create_future()
@@ -198,9 +173,8 @@ class QueueGetTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=loop)
-            started = asyncio.Event(loop=loop)
+        q = asyncio.Queue()
+        started = asyncio.Event()
         finished = False
 
         async def queue_get():
@@ -224,14 +198,12 @@ class QueueGetTests(_QueueTestBase):
         self.assertAlmostEqual(0.01, loop.time())
 
     def test_nonblocking_get(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         q.put_nowait(1)
         self.assertEqual(1, q.get_nowait())
 
     def test_nonblocking_get_exception(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         self.assertRaises(asyncio.QueueEmpty, q.get_nowait)
 
     def test_get_cancelled(self):
@@ -245,8 +217,7 @@ class QueueGetTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=loop)
+        q = asyncio.Queue()
 
         async def queue_get():
             return await asyncio.wait_for(q.get(), 0.051)
@@ -261,8 +232,7 @@ class QueueGetTests(_QueueTestBase):
         self.assertAlmostEqual(0.06, loop.time())
 
     def test_get_cancelled_race(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
 
         t1 = self.loop.create_task(q.get())
         t2 = self.loop.create_task(q.get())
@@ -276,8 +246,7 @@ class QueueGetTests(_QueueTestBase):
         self.assertEqual(t2.result(), 'a')
 
     def test_get_with_waiting_putters(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop, maxsize=1)
+        q = asyncio.Queue(maxsize=1)
         self.loop.create_task(q.put('a'))
         self.loop.create_task(q.put('b'))
         test_utils.run_briefly(self.loop)
@@ -298,8 +267,12 @@ class QueueGetTests(_QueueTestBase):
         queue_size = 1
         producer_num_items = 5
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(queue_size, loop=self.loop)
+        async def create_queue():
+            queue = asyncio.Queue(queue_size)
+            queue._get_loop()
+            return queue
+
+        q = self.loop.run_until_complete(create_queue())
 
         self.loop.run_until_complete(
             asyncio.gather(producer(q, producer_num_items),
@@ -320,8 +293,7 @@ class QueueGetTests(_QueueTestBase):
             except asyncio.TimeoutError:
                 pass
 
-        with self.assertWarns(DeprecationWarning):
-            queue = asyncio.Queue(loop=self.loop, maxsize=5)
+        queue = asyncio.Queue(maxsize=5)
         self.loop.run_until_complete(self.loop.create_task(consumer(queue)))
         self.assertEqual(len(queue._getters), 0)
 
@@ -329,8 +301,7 @@ class QueueGetTests(_QueueTestBase):
 class QueuePutTests(_QueueTestBase):
 
     def test_blocking_put(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
 
         async def queue_put():
             # No maxsize, won't block.
@@ -347,9 +318,8 @@ class QueuePutTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(maxsize=1, loop=loop)
-            started = asyncio.Event(loop=loop)
+        q = asyncio.Queue(maxsize=1)
+        started = asyncio.Event()
         finished = False
 
         async def queue_put():
@@ -371,8 +341,7 @@ class QueuePutTests(_QueueTestBase):
         self.assertAlmostEqual(0.01, loop.time())
 
     def test_nonblocking_put(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         q.put_nowait(1)
         self.assertEqual(1, q.get_nowait())
 
@@ -383,8 +352,7 @@ class QueuePutTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=loop)
+        q = asyncio.Queue()
 
         reader = loop.create_task(q.get())
 
@@ -413,8 +381,7 @@ class QueuePutTests(_QueueTestBase):
         loop = self.new_test_loop(gen)
         loop.set_debug(True)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=loop)
+        q = asyncio.Queue()
 
         reader1 = loop.create_task(q.get())
         reader2 = loop.create_task(q.get())
@@ -444,8 +411,7 @@ class QueuePutTests(_QueueTestBase):
 
         loop = self.new_test_loop(gen)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(1, loop=loop)
+        q = asyncio.Queue(1)
 
         q.put_nowait(1)
 
@@ -469,21 +435,18 @@ class QueuePutTests(_QueueTestBase):
         self.assertEqual(q.qsize(), 0)
 
     def test_nonblocking_put_exception(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(maxsize=1, loop=self.loop)
+        q = asyncio.Queue(maxsize=1, )
         q.put_nowait(1)
         self.assertRaises(asyncio.QueueFull, q.put_nowait, 2)
 
     def test_float_maxsize(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(maxsize=1.3, loop=self.loop)
+        q = asyncio.Queue(maxsize=1.3, )
         q.put_nowait(1)
         q.put_nowait(2)
         self.assertTrue(q.full())
         self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)
 
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(maxsize=1.3, loop=self.loop)
+        q = asyncio.Queue(maxsize=1.3, )
 
         async def queue_put():
             await q.put(1)
@@ -492,8 +455,7 @@ class QueuePutTests(_QueueTestBase):
         self.loop.run_until_complete(queue_put())
 
     def test_put_cancelled(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
 
         async def queue_put():
             await q.put(1)
@@ -508,8 +470,7 @@ class QueuePutTests(_QueueTestBase):
         self.assertTrue(t.result())
 
     def test_put_cancelled_race(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop, maxsize=1)
+        q = asyncio.Queue(maxsize=1)
 
         put_a = self.loop.create_task(q.put('a'))
         put_b = self.loop.create_task(q.put('b'))
@@ -529,8 +490,7 @@ class QueuePutTests(_QueueTestBase):
         self.loop.run_until_complete(put_b)
 
     def test_put_with_waiting_getters(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.Queue(loop=self.loop)
+        q = asyncio.Queue()
         t = self.loop.create_task(q.get())
         test_utils.run_briefly(self.loop)
         self.loop.run_until_complete(q.put('a'))
@@ -539,8 +499,12 @@ class QueuePutTests(_QueueTestBase):
     def test_why_are_putters_waiting(self):
         # From issue #265.
 
-        with self.assertWarns(DeprecationWarning):
-            queue = asyncio.Queue(2, loop=self.loop)
+        async def create_queue():
+            q = asyncio.Queue(2)
+            q._get_loop()
+            return q
+
+        queue = self.loop.run_until_complete(create_queue())
 
         async def putter(item):
             await queue.put(item)
@@ -566,8 +530,7 @@ class QueuePutTests(_QueueTestBase):
         loop = self.new_test_loop(a_generator)
 
         # Full queue.
-        with self.assertWarns(DeprecationWarning):
-            queue = asyncio.Queue(loop=loop, maxsize=1)
+        queue = asyncio.Queue(maxsize=1)
         queue.put_nowait(1)
 
         # Task waiting for space to put an item in the queue.
@@ -590,8 +553,7 @@ class QueuePutTests(_QueueTestBase):
         loop = self.new_test_loop(gen)
 
         # Full Queue.
-        with self.assertWarns(DeprecationWarning):
-            queue = asyncio.Queue(1, loop=loop)
+        queue = asyncio.Queue(1)
         queue.put_nowait(1)
 
         # Task waiting for space to put a item in the queue.
@@ -614,8 +576,7 @@ class QueuePutTests(_QueueTestBase):
 class LifoQueueTests(_QueueTestBase):
 
     def test_order(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.LifoQueue(loop=self.loop)
+        q = asyncio.LifoQueue()
         for i in [1, 3, 2]:
             q.put_nowait(i)
 
@@ -626,8 +587,7 @@ class LifoQueueTests(_QueueTestBase):
 class PriorityQueueTests(_QueueTestBase):
 
     def test_order(self):
-        with self.assertWarns(DeprecationWarning):
-            q = asyncio.PriorityQueue(loop=self.loop)
+        q = asyncio.PriorityQueue()
         for i in [1, 3, 2]:
             q.put_nowait(i)
 
@@ -640,13 +600,11 @@ class _QueueJoinTestMixin:
     q_class = None
 
     def test_task_done_underflow(self):
-        with self.assertWarns(DeprecationWarning):
-            q = self.q_class(loop=self.loop)
+        q = self.q_class()
         self.assertRaises(ValueError, q.task_done)
 
     def test_task_done(self):
-        with self.assertWarns(DeprecationWarning):
-            q = self.q_class(loop=self.loop)
+        q = self.q_class()
         for i in range(100):
             q.put_nowait(i)
 
@@ -681,8 +639,7 @@ class _QueueJoinTestMixin:
         self.loop.run_until_complete(asyncio.wait(tasks))
 
     def test_join_empty_queue(self):
-        with self.assertWarns(DeprecationWarning):
-            q = self.q_class(loop=self.loop)
+        q = self.q_class()
 
         # Test that a queue join()s successfully, and before anything else
         # (done twice for insurance).
@@ -694,8 +651,7 @@ class _QueueJoinTestMixin:
         self.loop.run_until_complete(join())
 
     def test_format(self):
-        with self.assertWarns(DeprecationWarning):
-            q = self.q_class(loop=self.loop)
+        q = self.q_class()
         self.assertEqual(q._format(), 'maxsize=0')
 
         q._unfinished_tasks = 2
diff --git a/Lib/test/test_asyncio/utils.py b/Lib/test/test_asyncio/utils.py
index 34da739..aba90c9 100644
--- a/Lib/test/test_asyncio/utils.py
+++ b/Lib/test/test_asyncio/utils.py
@@ -546,7 +546,21 @@ class TestCase(unittest.TestCase):
 
     def setUp(self):
         self._get_running_loop = events._get_running_loop
-        events._get_running_loop = lambda: None
+
+        def _get_running_loop():
+            frame = sys._getframe(1)
+
+            if frame.f_globals['__name__'] == 'asyncio.mixins':
+                # When we called from LoopBoundedMixin we should
+                # fallback to default implementation of get_running_loop
+                try:
+                    return events.get_running_loop()
+                except RuntimeError:
+                    return None
+
+            return None
+
+        events._get_running_loop = _get_running_loop
         self._thread_cleanup = threading_helper.threading_setup()
 
     def tearDown(self):
diff --git a/Misc/NEWS.d/next/Library/2020-11-20-14-01-29.bpo-42392.-OUzvl.rst b/Misc/NEWS.d/next/Library/2020-11-20-14-01-29.bpo-42392.-OUzvl.rst
new file mode 100644
index 0000000..964bef1
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2020-11-20-14-01-29.bpo-42392.-OUzvl.rst
@@ -0,0 +1,2 @@
+Remove loop parameter from ``__init__`` in all ``asyncio.locks`` and
+``asyncio.Queue`` classes. Patch provided by Yurii Karabas.