Issue #21326: Add a new is_closed() method to asyncio.BaseEventLoop

Add BaseEventLoop._closed attribute and use it to check if the event loop was
closed or not, instead of checking different attributes in each subclass of
BaseEventLoop.

run_forever() and run_until_complete() methods now raise a RuntimeError('Event loop is
closed') exception if the event loop was closed.

BaseProactorEventLoop.close() now also cancels "accept futures".

Author: vstinner

Doc/library/asyncio-eventloop.rst

@@ -119,6 +119,12 @@ Run an event loop
    Callback scheduled after :meth:`stop` is called won't.  However, those
    callbacks will run if :meth:`run_forever` is called again later.
 
+.. method:: BaseEventLoop.is_closed()
+
+   Returns ``True`` if the event loop was closed.
+
+   .. versionadded:: 3.4.2
+
 .. method:: BaseEventLoop.close()
 
    Close the event loop. The loop should not be running.

Lib/asyncio/base_events.py

@@ -119,6 +119,7 @@ def wait_closed(self):
 class BaseEventLoop(events.AbstractEventLoop):
 
     def __init__(self):
+        self._closed = False
         self._ready = collections.deque()
         self._scheduled = []
         self._default_executor = None
@@ -128,6 +129,11 @@ def __init__(self):
         self._exception_handler = None
         self._debug = False
 
+    def __repr__(self):
+        return ('<%s running=%s closed=%s debug=%s>'
+                % (self.__class__.__name__, self.is_running(),
+                   self.is_closed(), self.get_debug()))
+
     def _make_socket_transport(self, sock, protocol, waiter=None, *,
                                extra=None, server=None):
         """Create socket transport."""
@@ -173,8 +179,13 @@ def _process_events(self, event_list):
         """Process selector events."""
         raise NotImplementedError
 
+    def _check_closed(self):
+        if self._closed:
+            raise RuntimeError('Event loop is closed')
+
     def run_forever(self):
         """Run until stop() is called."""
+        self._check_closed()
         if self._running:
             raise RuntimeError('Event loop is running.')
         self._running = True
@@ -198,6 +209,7 @@ def run_until_complete(self, future):
 
         Return the Future's result, or raise its exception.
         """
+        self._check_closed()
         future = tasks.async(future, loop=self)
         future.add_done_callback(_raise_stop_error)
         self.run_forever()
@@ -222,13 +234,20 @@ def close(self):
         This clears the queues and shuts down the executor,
         but does not wait for the executor to finish.
         """
+        if self._closed:
+            return
+        self._closed = True
         self._ready.clear()
         self._scheduled.clear()
         executor = self._default_executor
         if executor is not None:
             self._default_executor = None
             executor.shutdown(wait=False)
 
+    def is_closed(self):
+        """Returns True if the event loop was closed."""
+        return self._closed
+
     def is_running(self):
         """Returns running status of event loop."""
         return self._running

Lib/asyncio/proactor_events.py

@@ -353,13 +353,14 @@ def _make_write_pipe_transport(self, sock, protocol, waiter=None,
                                            sock, protocol, waiter, extra)
 
     def close(self):
-        if self._proactor is not None:
-            self._close_self_pipe()
-            self._proactor.close()
-            self._proactor = None
-            self._selector = None
-            super().close()
-        self._accept_futures.clear()
+        if self.is_closed():
+            return
+        self._stop_accept_futures()
+        self._close_self_pipe()
+        self._proactor.close()
+        self._proactor = None
+        self._selector = None
+        super().close()
 
     def sock_recv(self, sock, n):
         return self._proactor.recv(sock, n)
@@ -428,6 +429,8 @@ def loop(f=None):
                     self._make_socket_transport(
                         conn, protocol,
                         extra={'peername': addr}, server=server)
+                if self.is_closed():
+                    return
                 f = self._proactor.accept(sock)
             except OSError as exc:
                 if sock.fileno() != -1:
@@ -448,8 +451,12 @@ def loop(f=None):
     def _process_events(self, event_list):
         pass    # XXX hard work currently done in poll
 
-    def _stop_serving(self, sock):
+    def _stop_accept_futures(self):
         for future in self._accept_futures.values():
             future.cancel()
+        self._accept_futures.clear()
+
+    def _stop_serving(self, sock):
+        self._stop_accept_futures()
         self._proactor._stop_serving(sock)
         sock.close()

Lib/asyncio/selector_events.py

@@ -55,11 +55,13 @@ def _make_datagram_transport(self, sock, protocol,
         return _SelectorDatagramTransport(self, sock, protocol, address, extra)
 
     def close(self):
+        if self.is_closed():
+            return
+        self._close_self_pipe()
         if self._selector is not None:
-            self._close_self_pipe()
             self._selector.close()
             self._selector = None
-            super().close()
+        super().close()
 
     def _socketpair(self):
         raise NotImplementedError
@@ -143,8 +145,7 @@ def _accept_connection(self, protocol_factory, sock,
 
     def add_reader(self, fd, callback, *args):
         """Add a reader callback."""
-        if self._selector is None:
-            raise RuntimeError('Event loop is closed')
+        self._check_closed()
         handle = events.Handle(callback, args, self)
         try:
             key = self._selector.get_key(fd)
@@ -160,7 +161,7 @@ def add_reader(self, fd, callback, *args):
 
     def remove_reader(self, fd):
         """Remove a reader callback."""
-        if self._selector is None:
+        if self.is_closed():
             return False
         try:
             key = self._selector.get_key(fd)
@@ -182,8 +183,7 @@ def remove_reader(self, fd):
 
     def add_writer(self, fd, callback, *args):
         """Add a writer callback.."""
-        if self._selector is None:
-            raise RuntimeError('Event loop is closed')
+        self._check_closed()
         handle = events.Handle(callback, args, self)
         try:
             key = self._selector.get_key(fd)
@@ -199,7 +199,7 @@ def add_writer(self, fd, callback, *args):
 
     def remove_writer(self, fd):
         """Remove a writer callback."""
-        if self._selector is None:
+        if self.is_closed():
             return False
         try:
             key = self._selector.get_key(fd)

Lib/test/test_asyncio/test_base_events.py

@@ -52,6 +52,20 @@ def test_not_implemented(self):
         gen = self.loop._make_subprocess_transport(m, m, m, m, m, m, m)
         self.assertRaises(NotImplementedError, next, iter(gen))
 
+    def test_close(self):
+        self.assertFalse(self.loop.is_closed())
+        self.loop.close()
+        self.assertTrue(self.loop.is_closed())
+
+        # it should be possible to call close() more than once
+        self.loop.close()
+        self.loop.close()
+
+        # operation blocked when the loop is closed
+        f = asyncio.Future(loop=self.loop)
+        self.assertRaises(RuntimeError, self.loop.run_forever)
+        self.assertRaises(RuntimeError, self.loop.run_until_complete, f)
+
     def test__add_callback_handle(self):
         h = asyncio.Handle(lambda: False, (), self.loop)
 

Lib/test/test_asyncio/test_selector_events.py

@@ -80,7 +80,10 @@ def test_close(self):
 
         self.loop._selector.close()
         self.loop._selector = selector = mock.Mock()
+        self.assertFalse(self.loop.is_closed())
+
         self.loop.close()
+        self.assertTrue(self.loop.is_closed())
         self.assertIsNone(self.loop._selector)
         self.assertIsNone(self.loop._csock)
         self.assertIsNone(self.loop._ssock)
@@ -89,9 +92,20 @@ def test_close(self):
         csock.close.assert_called_with()
         remove_reader.assert_called_with(7)
 
+        # it should be possible to call close() more than once
         self.loop.close()
         self.loop.close()
 
+        # operation blocked when the loop is closed
+        f = asyncio.Future(loop=self.loop)
+        self.assertRaises(RuntimeError, self.loop.run_forever)
+        self.assertRaises(RuntimeError, self.loop.run_until_complete, f)
+        fd = 0
+        def callback():
+            pass
+        self.assertRaises(RuntimeError, self.loop.add_reader, fd, callback)
+        self.assertRaises(RuntimeError, self.loop.add_writer, fd, callback)
+
     def test_close_no_selector(self):
         ssock = self.loop._ssock
         csock = self.loop._csock
@@ -101,9 +115,6 @@ def test_close_no_selector(self):
         self.loop._selector = None
         self.loop.close()
         self.assertIsNone(self.loop._selector)
-        self.assertFalse(ssock.close.called)
-        self.assertFalse(csock.close.called)
-        self.assertFalse(remove_reader.called)
 
     def test_socketpair(self):
         self.assertRaises(NotImplementedError, self.loop._socketpair)

Misc/NEWS

@@ -22,6 +22,10 @@ Core and Builtins
 Library
 -------
 
+- Issue #21326: Add a new is_closed() method to asyncio.BaseEventLoop.
+  run_forever() and run_until_complete() methods of asyncio.BaseEventLoop now
+  raise an exception if the event loop was closed.
+
 - Issue #21310: Fixed possible resource leak in failed open().
 
 - Issue #21677: Fixed chaining nonnormalized exceptions in io close() methods.