bpo-32528: Make asyncio.CancelledError a BaseException.

This will address the common mistake many asyncio users make:
an "except Exception" clause breaking Tasks cancellation.

In addition to this change, we stop inheriting asyncio.TimeoutError
and asyncio.InvalidStateError from their concurrent.futures.*
counterparts.  There's no point for these exceptions to share the
inheritance chain.

Author: Yury Selivanov

Lib/asyncio/base_events.py

@@ -186,7 +186,7 @@ def _interleave_addrinfos(addrinfos, first_address_family_count=1):
 def _run_until_complete_cb(fut):
     if not fut.cancelled():
         exc = fut.exception()
-        if isinstance(exc, BaseException) and not isinstance(exc, Exception):
+        if isinstance(exc, (SystemExit, KeyboardInterrupt)):
             # Issue #22429: run_forever() already finished, no need to
             # stop it.
             return
@@ -1196,7 +1196,7 @@ async def start_tls(self, transport, protocol, sslcontext, *,
 
         try:
             await waiter
-        except Exception:
+        except BaseException:
             transport.close()
             conmade_cb.cancel()
             resume_cb.cancel()
@@ -1710,7 +1710,9 @@ def call_exception_handler(self, context):
         if self._exception_handler is None:
             try:
                 self.default_exception_handler(context)
-            except Exception:
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except BaseException:
                 # Second protection layer for unexpected errors
                 # in the default implementation, as well as for subclassed
                 # event loops with overloaded "default_exception_handler".
@@ -1719,7 +1721,9 @@ def call_exception_handler(self, context):
         else:
             try:
                 self._exception_handler(self, context)
-            except Exception as exc:
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except BaseException as exc:
                 # Exception in the user set custom exception handler.
                 try:
                     # Let's try default handler.
@@ -1728,7 +1732,9 @@ def call_exception_handler(self, context):
                         'exception': exc,
                         'context': context,
                     })
-                except Exception:
+                except (SystemExit, KeyboardInterrupt):
+                    raise
+                except BaseException:
                     # Guard 'default_exception_handler' in case it is
                     # overloaded.
                     logger.error('Exception in default exception handler '

Lib/asyncio/base_subprocess.py

@@ -182,7 +182,9 @@ async def _connect_pipes(self, waiter):
             for callback, data in self._pending_calls:
                 loop.call_soon(callback, *data)
             self._pending_calls = None
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             if waiter is not None and not waiter.cancelled():
                 waiter.set_exception(exc)
         else:

Lib/asyncio/events.py

@@ -79,7 +79,9 @@ def cancelled(self):
     def _run(self):
         try:
             self._context.run(self._callback, *self._args)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             cb = format_helpers._format_callback_source(
                 self._callback, self._args)
             msg = f'Exception in callback {cb}'

Lib/asyncio/exceptions.py

@@ -5,19 +5,16 @@
            'IncompleteReadError', 'LimitOverrunError',
            'SendfileNotAvailableError')
 
-import concurrent.futures
-from . import base_futures
 
-
-class CancelledError(concurrent.futures.CancelledError):
+class CancelledError(BaseException):
     """The Future or Task was cancelled."""
 
 
-class TimeoutError(concurrent.futures.TimeoutError):
+class TimeoutError(Exception):
     """The operation exceeded the given deadline."""
 
 
-class InvalidStateError(concurrent.futures.InvalidStateError):
+class InvalidStateError(Exception):
     """The operation is not allowed in this state."""
 
 

Lib/asyncio/proactor_events.py

@@ -212,7 +212,9 @@ def _eof_received(self):
 
         try:
             keep_open = self._protocol.eof_received()
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(
                 exc, 'Fatal error: protocol.eof_received() call failed.')
             return
@@ -235,7 +237,9 @@ def _data_received(self, data):
         if isinstance(self._protocol, protocols.BufferedProtocol):
             try:
                 protocols._feed_data_to_buffered_proto(self._protocol, data)
-            except Exception as exc:
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except BaseException as exc:
                 self._fatal_error(exc,
                                   'Fatal error: protocol.buffer_updated() '
                                   'call failed.')
@@ -625,7 +629,9 @@ def _loop_self_reading(self, f=None):
         except exceptions.CancelledError:
             # _close_self_pipe() has been called, stop waiting for data
             return
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self.call_exception_handler({
                 'message': 'Error on reading from the event loop self pipe',
                 'exception': exc,

Lib/asyncio/selector_events.py

@@ -208,12 +208,14 @@ async def _accept_connection2(
 
             try:
                 await waiter
-            except:
+            except BaseException:
                 transport.close()
                 raise
+                # It's now up to the protocol to handle the connection.
 
-            # It's now up to the protocol to handle the connection.
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             if self._debug:
                 context = {
                     'message':
@@ -370,7 +372,9 @@ def _sock_recv(self, fut, sock, n):
             data = sock.recv(n)
         except (BlockingIOError, InterruptedError):
             return  # try again next time
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             fut.set_exception(exc)
         else:
             fut.set_result(data)
@@ -404,7 +408,9 @@ def _sock_recv_into(self, fut, sock, buf):
             nbytes = sock.recv_into(buf)
         except (BlockingIOError, InterruptedError):
             return  # try again next time
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             fut.set_exception(exc)
         else:
             fut.set_result(nbytes)
@@ -447,7 +453,9 @@ def _sock_sendall(self, fut, sock, view, pos):
             n = sock.send(view[start:])
         except (BlockingIOError, InterruptedError):
             return
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             fut.set_exception(exc)
             return
 
@@ -487,7 +495,9 @@ def _sock_connect(self, fut, sock, address):
             fut.add_done_callback(
                 functools.partial(self._sock_write_done, fd))
             self.add_writer(fd, self._sock_connect_cb, fut, sock, address)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             fut.set_exception(exc)
         else:
             fut.set_result(None)
@@ -507,7 +517,9 @@ def _sock_connect_cb(self, fut, sock, address):
         except (BlockingIOError, InterruptedError):
             # socket is still registered, the callback will be retried later
             pass
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             fut.set_exception(exc)
         else:
             fut.set_result(None)
@@ -537,7 +549,9 @@ def _sock_accept(self, fut, registered, sock):
             conn.setblocking(False)
         except (BlockingIOError, InterruptedError):
             self.add_reader(fd, self._sock_accept, fut, True, sock)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             fut.set_exception(exc)
         else:
             fut.set_result((conn, address))
@@ -785,7 +799,9 @@ def _read_ready__get_buffer(self):
             buf = self._protocol.get_buffer(-1)
             if not len(buf):
                 raise RuntimeError('get_buffer() returned an empty buffer')
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(
                 exc, 'Fatal error: protocol.get_buffer() call failed.')
             return
@@ -794,7 +810,9 @@ def _read_ready__get_buffer(self):
             nbytes = self._sock.recv_into(buf)
         except (BlockingIOError, InterruptedError):
             return
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(exc, 'Fatal read error on socket transport')
             return
 
@@ -804,7 +822,9 @@ def _read_ready__get_buffer(self):
 
         try:
             self._protocol.buffer_updated(nbytes)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(
                 exc, 'Fatal error: protocol.buffer_updated() call failed.')
 
@@ -815,7 +835,9 @@ def _read_ready__data_received(self):
             data = self._sock.recv(self.max_size)
         except (BlockingIOError, InterruptedError):
             return
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(exc, 'Fatal read error on socket transport')
             return
 
@@ -825,7 +847,9 @@ def _read_ready__data_received(self):
 
         try:
             self._protocol.data_received(data)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(
                 exc, 'Fatal error: protocol.data_received() call failed.')
 
@@ -835,7 +859,9 @@ def _read_ready__on_eof(self):
 
         try:
             keep_open = self._protocol.eof_received()
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(
                 exc, 'Fatal error: protocol.eof_received() call failed.')
             return
@@ -871,7 +897,9 @@ def write(self, data):
                 n = self._sock.send(data)
             except (BlockingIOError, InterruptedError):
                 pass
-            except Exception as exc:
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except BaseException as exc:
                 self._fatal_error(exc, 'Fatal write error on socket transport')
                 return
             else:
@@ -894,7 +922,9 @@ def _write_ready(self):
             n = self._sock.send(self._buffer)
         except (BlockingIOError, InterruptedError):
             pass
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._loop._remove_writer(self._sock_fd)
             self._buffer.clear()
             self._fatal_error(exc, 'Fatal write error on socket transport')
@@ -970,7 +1000,9 @@ def _read_ready(self):
             pass
         except OSError as exc:
             self._protocol.error_received(exc)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             self._fatal_error(exc, 'Fatal read error on datagram transport')
         else:
             self._protocol.datagram_received(data, addr)
@@ -1007,7 +1039,9 @@ def sendto(self, data, addr=None):
             except OSError as exc:
                 self._protocol.error_received(exc)
                 return
-            except Exception as exc:
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except BaseException as exc:
                 self._fatal_error(
                     exc, 'Fatal write error on datagram transport')
                 return
@@ -1030,7 +1064,9 @@ def _sendto_ready(self):
             except OSError as exc:
                 self._protocol.error_received(exc)
                 return
-            except Exception as exc:
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except BaseException as exc:
                 self._fatal_error(
                     exc, 'Fatal write error on datagram transport')
                 return

Lib/asyncio/sslproto.py

@@ -527,7 +527,9 @@ def data_received(self, data):
 
         try:
             ssldata, appdata = self._sslpipe.feed_ssldata(data)
-        except Exception as e:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as e:
             self._fatal_error(e, 'SSL error in data received')
             return
 
@@ -542,7 +544,9 @@ def data_received(self, data):
                             self._app_protocol, chunk)
                     else:
                         self._app_protocol.data_received(chunk)
-                except Exception as ex:
+                except (SystemExit, KeyboardInterrupt):
+                    raise
+                except BaseException as ex:
                     self._fatal_error(
                         ex, 'application protocol failed to receive SSL data')
                     return
@@ -628,7 +632,9 @@ def _on_handshake_complete(self, handshake_exc):
                 raise handshake_exc
 
             peercert = sslobj.getpeercert()
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             if isinstance(exc, ssl.CertificateError):
                 msg = 'SSL handshake failed on verifying the certificate'
             else:
@@ -691,7 +697,9 @@ def _process_write_backlog(self):
                 # delete it and reduce the outstanding buffer size.
                 del self._write_backlog[0]
                 self._write_buffer_size -= len(data)
-        except Exception as exc:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as exc:
             if self._in_handshake:
                 # Exceptions will be re-raised in _on_handshake_complete.
                 self._on_handshake_complete(exc)

Lib/asyncio/staggered.py

@@ -105,7 +105,9 @@ async def run_one_coro(
 
         try:
             result = await coro_fn()
-        except Exception as e:
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as e:
             exceptions[this_index] = e
             this_failed.set()  # Kickstart the next coroutine
         else: