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".
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index 1c7073c..5ee21d1 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -119,6 +119,7 @@
class BaseEventLoop(events.AbstractEventLoop):
def __init__(self):
+ self._closed = False
self._ready = collections.deque()
self._scheduled = []
self._default_executor = None
@@ -128,6 +129,11 @@
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 @@
"""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 @@
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,6 +234,9 @@
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
@@ -229,6 +244,10 @@
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
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index d99e8ce..757a22e 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -353,13 +353,14 @@
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 @@
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 _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()
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index 86a8d23..1f8e5c8 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -55,11 +55,13 @@
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 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 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 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 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)