bpo-34622: Extract asyncio exceptions into a separate module

Lib/asyncio/__init__.py

@@ -8,6 +8,7 @@
 from .base_events import *
 from .coroutines import *
 from .events import *
+from .exceptions import *
 from .futures import *
 from .locks import *
 from .protocols import *
@@ -25,6 +26,7 @@
 __all__ = (base_events.__all__ +
            coroutines.__all__ +
            events.__all__ +
+           exceptions.__all__ +
            futures.__all__ +
            locks.__all__ +
            protocols.__all__ +

Lib/asyncio/base_events.py

@@ -37,6 +37,7 @@
 from . import constants
 from . import coroutines
 from . import events
+from . import exceptions
 from . import futures
 from . import protocols
 from . import sslproto
@@ -327,7 +328,7 @@ async def serve_forever(self):
 
         try:
             await self._serving_forever_fut
-        except futures.CancelledError:
+        except exceptions.CancelledError:
             try:
                 self.close()
                 await self.wait_closed()
@@ -800,7 +801,7 @@ async def sock_sendfile(self, sock, file, offset=0, count=None,
         try:
             return await self._sock_sendfile_native(sock, file,
                                                     offset, count)
-        except events.SendfileNotAvailableError as exc:
+        except exceptions.SendfileNotAvailableError as exc:
             if not fallback:
                 raise
         return await self._sock_sendfile_fallback(sock, file,
@@ -809,7 +810,7 @@ async def sock_sendfile(self, sock, file, offset=0, count=None,
     async def _sock_sendfile_native(self, sock, file, offset, count):
         # NB: sendfile syscall is not supported for SSL sockets and
         # non-mmap files even if sendfile is supported by OS
-        raise events.SendfileNotAvailableError(
+        raise exceptions.SendfileNotAvailableError(
             f"syscall sendfile is not available for socket {sock!r} "
             "and file {file!r} combination")
 
@@ -1053,7 +1054,7 @@ async def sendfile(self, transport, file, offset=0, count=None,
             try:
                 return await self._sendfile_native(transport, file,
                                                    offset, count)
-            except events.SendfileNotAvailableError as exc:
+            except exceptions.SendfileNotAvailableError as exc:
                 if not fallback:
                     raise
 
@@ -1066,7 +1067,7 @@ async def sendfile(self, transport, file, offset=0, count=None,
                                              offset, count)
 
     async def _sendfile_native(self, transp, file, offset, count):
-        raise events.SendfileNotAvailableError(
+        raise exceptions.SendfileNotAvailableError(
             "sendfile syscall is not supported")
 
     async def _sendfile_fallback(self, transp, file, offset, count):

Lib/asyncio/base_futures.py

@@ -1,15 +1,9 @@
 __all__ = ()
 
-import concurrent.futures
 import reprlib
 
 from . import format_helpers
 
-CancelledError = concurrent.futures.CancelledError
-TimeoutError = concurrent.futures.TimeoutError
-InvalidStateError = concurrent.futures.InvalidStateError
-
-
 # States for Future.
 _PENDING = 'PENDING'
 _CANCELLED = 'CANCELLED'

Lib/asyncio/events.py

@@ -3,7 +3,7 @@
 __all__ = (
     'AbstractEventLoopPolicy',
     'AbstractEventLoop', 'AbstractServer',
-    'Handle', 'TimerHandle', 'SendfileNotAvailableError',
+    'Handle', 'TimerHandle',
     'get_event_loop_policy', 'set_event_loop_policy',
     'get_event_loop', 'set_event_loop', 'new_event_loop',
     'get_child_watcher', 'set_child_watcher',
@@ -19,14 +19,7 @@
 import threading
 
 from . import format_helpers
-
-
-class SendfileNotAvailableError(RuntimeError):
-    """Sendfile syscall is not available.
-
-    Raised if OS does not support sendfile syscall for given socket or
-    file type.
-    """
+from . import exceptions
 
 
 class Handle:

Lib/asyncio/exceptions.py

@@ -0,0 +1,60 @@
+"""asyncio exceptions."""
+
+
+__all__ = ('CancelledError', 'InvalidStateError', 'TimeoutError',
+           'IncompleteReadError', 'LimitOverrunError',
+           'SendfileNotAvailableError')
+
+import concurrent.futures
+from . import base_futures
+
+
+class CancelledError(concurrent.futures.CancelledError):
+    """The Future or Task was cancelled."""
+
+
+class TimeoutError(concurrent.futures.TimeoutError):
+    """The operation exceeded the given deadline."""
+
+
+class InvalidStateError(concurrent.futures.InvalidStateError):
+    """The operation is not allowed in this state."""
+
+
+class SendfileNotAvailableError(RuntimeError):
+    """Sendfile syscall is not available.
+
+    Raised if OS does not support sendfile syscall for given socket or
+    file type.
+    """
+
+
+class IncompleteReadError(EOFError):
+    """
+    Incomplete read error. Attributes:
+
+    - partial: read bytes string before the end of stream was reached
+    - expected: total number of expected bytes (or None if unknown)
+    """
+    def __init__(self, partial, expected):
+        super().__init__(f'{len(partial)} bytes read on a total of '
+                         f'{expected!r} expected bytes')
+        self.partial = partial
+        self.expected = expected
+
+    def __reduce__(self):
+        return type(self), (self.partial, self.expected)
+
+
+class LimitOverrunError(Exception):
+    """Reached the buffer limit while looking for a separator.
+
+    Attributes:
+    - consumed: total number of to be consumed bytes.
+    """
+    def __init__(self, message, consumed):
+        super().__init__(message)
+        self.consumed = consumed
+
+    def __reduce__(self):
+        return type(self), (self.args[0], self.consumed)

Lib/asyncio/futures.py

@@ -1,7 +1,6 @@
 """A Future class similar to the one in PEP 3148."""
 
 __all__ = (
-    'CancelledError', 'TimeoutError', 'InvalidStateError',
     'Future', 'wrap_future', 'isfuture',
 )
 
@@ -12,12 +11,10 @@
 
 from . import base_futures
 from . import events
+from . import exceptions
 from . import format_helpers
 
 
-CancelledError = base_futures.CancelledError
-InvalidStateError = base_futures.InvalidStateError
-TimeoutError = base_futures.TimeoutError
 isfuture = base_futures.isfuture
 
 
@@ -170,9 +167,9 @@ def result(self):
         the future is done and has an exception set, this exception is raised.
         """
         if self._state == _CANCELLED:
-            raise CancelledError
+            raise exceptions.CancelledError
         if self._state != _FINISHED:
-            raise InvalidStateError('Result is not ready.')
+            raise exceptions.InvalidStateError('Result is not ready.')
         self.__log_traceback = False
         if self._exception is not None:
             raise self._exception
@@ -187,9 +184,9 @@ def exception(self):
         InvalidStateError.
         """
         if self._state == _CANCELLED:
-            raise CancelledError
+            raise exceptions.CancelledError
         if self._state != _FINISHED:
-            raise InvalidStateError('Exception is not set.')
+            raise exceptions.InvalidStateError('Exception is not set.')
         self.__log_traceback = False
         return self._exception
 
@@ -231,7 +228,7 @@ def set_result(self, result):
         InvalidStateError.
         """
         if self._state != _PENDING:
-            raise InvalidStateError('{}: {!r}'.format(self._state, self))
+            raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
         self._result = result
         self._state = _FINISHED
         self.__schedule_callbacks()
@@ -243,7 +240,7 @@ def set_exception(self, exception):
         InvalidStateError.
         """
         if self._state != _PENDING:
-            raise InvalidStateError('{}: {!r}'.format(self._state, self))
+            raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
         if isinstance(exception, type):
             exception = exception()
         if type(exception) is StopIteration:
@@ -288,6 +285,18 @@ def _set_result_unless_cancelled(fut, result):
     fut.set_result(result)
 
 
+def _convert_future_exc(exc):
+    exc_class = type(exc)
+    if exc_class is concurrent.futures.CancelledError:
+        return exceptions.CancelledError(*exc.args)
+    elif exc_class is concurrent.futures.TimeoutError:
+        return exceptions.TimeoutError(*exc.args)
+    elif exc_class is concurrent.futures.InvalidStateError:
+        return exceptions.InvalidStateError(*exc.args)
+    else:
+        return exc
+
+
 def _set_concurrent_future_state(concurrent, source):
     """Copy state from a future to a concurrent.futures.Future."""
     assert source.done()
@@ -297,7 +306,7 @@ def _set_concurrent_future_state(concurrent, source):
         return
     exception = source.exception()
     if exception is not None:
-        concurrent.set_exception(exception)
+        concurrent.set_exception(_convert_future_exc(exception))
     else:
         result = source.result()
         concurrent.set_result(result)
@@ -317,7 +326,7 @@ def _copy_future_state(source, dest):
     else:
         exception = source.exception()
         if exception is not None:
-            dest.set_exception(exception)
+            dest.set_exception(_convert_future_exc(exception))
         else:
             result = source.result()
             dest.set_result(result)

Lib/asyncio/locks.py

@@ -7,6 +7,7 @@
 
 from . import events
 from . import futures
+from . import exceptions
 from .coroutines import coroutine
 
 
@@ -192,7 +193,7 @@ async def acquire(self):
                 await fut
             finally:
                 self._waiters.remove(fut)
-        except futures.CancelledError:
+        except exceptions.CancelledError:
             if not self._locked:
                 self._wake_up_first()
             raise
@@ -363,11 +364,11 @@ async def wait(self):
                 try:
                     await self.acquire()
                     break
-                except futures.CancelledError:
+                except exceptions.CancelledError:
                     cancelled = True
 
             if cancelled:
-                raise futures.CancelledError
+                raise exceptions.CancelledError
 
     async def wait_for(self, predicate):
         """Wait until a predicate becomes true.

Lib/asyncio/proactor_events.py

@@ -15,6 +15,7 @@
 from . import constants
 from . import events
 from . import futures
+from . import exceptions
 from . import protocols
 from . import sslproto
 from . import transports
@@ -282,7 +283,7 @@ def _loop_reading(self, fut=None):
             self._force_close(exc)
         except OSError as exc:
             self._fatal_error(exc, 'Fatal read error on pipe transport')
-        except futures.CancelledError:
+        except exceptions.CancelledError:
             if not self._closing:
                 raise
         else:
@@ -555,11 +556,11 @@ async def _sock_sendfile_native(self, sock, file, offset, count):
         try:
             fileno = file.fileno()
         except (AttributeError, io.UnsupportedOperation) as err:
-            raise events.SendfileNotAvailableError("not a regular file")
+            raise exceptions.SendfileNotAvailableError("not a regular file")
         try:
             fsize = os.fstat(fileno).st_size
         except OSError as err:
-            raise events.SendfileNotAvailableError("not a regular file")
+            raise exceptions.SendfileNotAvailableError("not a regular file")
         blocksize = count if count else fsize
         if not blocksize:
             return 0  # empty file
@@ -615,7 +616,7 @@ def _loop_self_reading(self, f=None):
             if f is not None:
                 f.result()  # may raise
             f = self._proactor.recv(self._ssock, 4096)
-        except futures.CancelledError:
+        except exceptions.CancelledError:
             # _close_self_pipe() has been called, stop waiting for data
             return
         except Exception as exc:
@@ -666,7 +667,7 @@ def loop(f=None):
                 elif self._debug:
                     logger.debug("Accept failed on socket %r",
                                  sock, exc_info=True)
-            except futures.CancelledError:
+            except exceptions.CancelledError:
                 sock.close()
             else:
                 self._accept_futures[sock.fileno()] = f