python/qemu/aqmp/protocol.py

   1 """
   2 Generic Asynchronous Message-based Protocol Support
   3
   4 This module provides a generic framework for sending and receiving
   5 messages over an asyncio stream. `AsyncProtocol` is an abstract class
   6 that implements the core mechanisms of a simple send/receive protocol,
   7 and is designed to be extended.
   8
   9 In this package, it is used as the implementation for the `QMPClient`
  10 class.
  11 """
  12
  13 import asyncio
  14 from asyncio import StreamReader, StreamWriter
  15 from enum import Enum
  16 from functools import wraps
  17 import logging
  18 import socket
  19 from ssl import SSLContext
  20 from typing import (
  21     Any,
  22     Awaitable,
  23     Callable,
  24     Generic,
  25     List,
  26     Optional,
  27     Tuple,
  28     TypeVar,
  29     Union,
  30     cast,
  31 )
  32
  33 from .error import QMPError
  34 from .util import (
  35     bottom_half,
  36     create_task,
  37     exception_summary,
  38     flush,
  39     is_closing,
  40     pretty_traceback,
  41     upper_half,
  42     wait_closed,
  43 )
  44
  45
  46 T = TypeVar('T')
  47 _U = TypeVar('_U')
  48 _TaskFN = Callable[[], Awaitable[None]]  # aka ``async def func() -> None``
  49
  50 InternetAddrT = Tuple[str, int]
  51 UnixAddrT = str
  52 SocketAddrT = Union[UnixAddrT, InternetAddrT]
  53
  54
  55 class Runstate(Enum):
  56     """Protocol session runstate."""
  57
  58     #: Fully quiesced and disconnected.
  59     IDLE = 0
  60     #: In the process of connecting or establishing a session.
  61     CONNECTING = 1
  62     #: Fully connected and active session.
  63     RUNNING = 2
  64     #: In the process of disconnecting.
  65     #: Runstate may be returned to `IDLE` by calling `disconnect()`.
  66     DISCONNECTING = 3
  67
  68
  69 class ConnectError(QMPError):
  70     """
  71     Raised when the initial connection process has failed.
  72
  73     This Exception always wraps a "root cause" exception that can be
  74     interrogated for additional information.
  75
  76     :param error_message: Human-readable string describing the error.
  77     :param exc: The root-cause exception.
  78     """
  79     def __init__(self, error_message: str, exc: Exception):
  80         super().__init__(error_message)
  81         #: Human-readable error string
  82         self.error_message: str = error_message
  83         #: Wrapped root cause exception
  84         self.exc: Exception = exc
  85
  86     def __str__(self) -> str:
  87         cause = str(self.exc)
  88         if not cause:
  89             # If there's no error string, use the exception name.
  90             cause = exception_summary(self.exc)
  91         return f"{self.error_message}: {cause}"
  92
  93
  94 class StateError(QMPError):
  95     """
  96     An API command (connect, execute, etc) was issued at an inappropriate time.
  97
  98     This error is raised when a command like
  99     :py:meth:`~AsyncProtocol.connect()` is issued at an inappropriate
 100     time.
 101
 102     :param error_message: Human-readable string describing the state violation.
 103     :param state: The actual `Runstate` seen at the time of the violation.
 104     :param required: The `Runstate` required to process this command.
 105     """
 106     def __init__(self, error_message: str,
 107                  state: Runstate, required: Runstate):
 108         super().__init__(error_message)
 109         self.error_message = error_message
 110         self.state = state
 111         self.required = required
 112
 113
 114 F = TypeVar('F', bound=Callable[..., Any])  # pylint: disable=invalid-name
 115
 116
 117 # Don't Panic.
 118 def require(required_state: Runstate) -> Callable[[F], F]:
 119     """
 120     Decorator: protect a method so it can only be run in a certain `Runstate`.
 121
 122     :param required_state: The `Runstate` required to invoke this method.
 123     :raise StateError: When the required `Runstate` is not met.
 124     """
 125     def _decorator(func: F) -> F:
 126         # _decorator is the decorator that is built by calling the
 127         # require() decorator factory; e.g.:
 128         #
 129         # @require(Runstate.IDLE) def foo(): ...
 130         # will replace 'foo' with the result of '_decorator(foo)'.
 131
 132         @wraps(func)
 133         def _wrapper(proto: 'AsyncProtocol[Any]',
 134                      *args: Any, **kwargs: Any) -> Any:
 135             # _wrapper is the function that gets executed prior to the
 136             # decorated method.
 137
 138             name = type(proto).__name__
 139
 140             if proto.runstate != required_state:
 141                 if proto.runstate == Runstate.CONNECTING:
 142                     emsg = f"{name} is currently connecting."
 143                 elif proto.runstate == Runstate.DISCONNECTING:
 144                     emsg = (f"{name} is disconnecting."
 145                             " Call disconnect() to return to IDLE state.")
 146                 elif proto.runstate == Runstate.RUNNING:
 147                     emsg = f"{name} is already connected and running."
 148                 elif proto.runstate == Runstate.IDLE:
 149                     emsg = f"{name} is disconnected and idle."
 150                 else:
 151                     assert False
 152                 raise StateError(emsg, proto.runstate, required_state)
 153             # No StateError, so call the wrapped method.
 154             return func(proto, *args, **kwargs)
 155
 156         # Return the decorated method;
 157         # Transforming Func to Decorated[Func].
 158         return cast(F, _wrapper)
 159
 160     # Return the decorator instance from the decorator factory. Phew!
 161     return _decorator
 162
 163
 164 class AsyncProtocol(Generic[T]):
 165     """
 166     AsyncProtocol implements a generic async message-based protocol.
 167
 168     This protocol assumes the basic unit of information transfer between
 169     client and server is a "message", the details of which are left up
 170     to the implementation. It assumes the sending and receiving of these
 171     messages is full-duplex and not necessarily correlated; i.e. it
 172     supports asynchronous inbound messages.
 173
 174     It is designed to be extended by a specific protocol which provides
 175     the implementations for how to read and send messages. These must be
 176     defined in `_do_recv()` and `_do_send()`, respectively.
 177
 178     Other callbacks have a default implementation, but are intended to be
 179     either extended or overridden:
 180
 181      - `_establish_session`:
 182          The base implementation starts the reader/writer tasks.
 183          A protocol implementation can override this call, inserting
 184          actions to be taken prior to starting the reader/writer tasks
 185          before the super() call; actions needing to occur afterwards
 186          can be written after the super() call.
 187      - `_on_message`:
 188          Actions to be performed when a message is received.
 189      - `_cb_outbound`:
 190          Logging/Filtering hook for all outbound messages.
 191      - `_cb_inbound`:
 192          Logging/Filtering hook for all inbound messages.
 193          This hook runs *before* `_on_message()`.
 194
 195     :param name:
 196         Name used for logging messages, if any. By default, messages
 197         will log to 'qemu.aqmp.protocol', but each individual connection
 198         can be given its own logger by giving it a name; messages will
 199         then log to 'qemu.aqmp.protocol.${name}'.
 200     """
 201     # pylint: disable=too-many-instance-attributes
 202
 203     #: Logger object for debugging messages from this connection.
 204     logger = logging.getLogger(__name__)
 205
 206     # Maximum allowable size of read buffer
 207     _limit = (64 * 1024)
 208
 209     # -------------------------
 210     # Section: Public interface
 211     # -------------------------
 212
 213     def __init__(self, name: Optional[str] = None) -> None:
 214         #: The nickname for this connection, if any.
 215         self.name: Optional[str] = name
 216         if self.name is not None:
 217             self.logger = self.logger.getChild(self.name)
 218
 219         # stream I/O
 220         self._reader: Optional[StreamReader] = None
 221         self._writer: Optional[StreamWriter] = None
 222
 223         # Outbound Message queue
 224         self._outgoing: asyncio.Queue[T]
 225
 226         # Special, long-running tasks:
 227         self._reader_task: Optional[asyncio.Future[None]] = None
 228         self._writer_task: Optional[asyncio.Future[None]] = None
 229
 230         # Aggregate of the above two tasks, used for Exception management.
 231         self._bh_tasks: Optional[asyncio.Future[Tuple[None, None]]] = None
 232
 233         #: Disconnect task. The disconnect implementation runs in a task
 234         #: so that asynchronous disconnects (initiated by the
 235         #: reader/writer) are allowed to wait for the reader/writers to
 236         #: exit.
 237         self._dc_task: Optional[asyncio.Future[None]] = None
 238
 239         self._runstate = Runstate.IDLE
 240         self._runstate_changed: Optional[asyncio.Event] = None
 241
 242         # Workaround for bind()
 243         self._sock: Optional[socket.socket] = None
 244
 245     def __repr__(self) -> str:
 246         cls_name = type(self).__name__
 247         tokens = []
 248         if self.name is not None:
 249             tokens.append(f"name={self.name!r}")
 250         tokens.append(f"runstate={self.runstate.name}")
 251         return f"<{cls_name} {' '.join(tokens)}>"
 252
 253     @property  # @upper_half
 254     def runstate(self) -> Runstate:
 255         """The current `Runstate` of the connection."""
 256         return self._runstate
 257
 258     @upper_half
 259     async def runstate_changed(self) -> Runstate:
 260         """
 261         Wait for the `runstate` to change, then return that runstate.
 262         """
 263         await self._runstate_event.wait()
 264         return self.runstate
 265
 266     @upper_half
 267     @require(Runstate.IDLE)
 268     async def accept(self, address: SocketAddrT,
 269                      ssl: Optional[SSLContext] = None) -> None:
 270         """
 271         Accept a connection and begin processing message queues.
 272
 273         If this call fails, `runstate` is guaranteed to be set back to `IDLE`.
 274
 275         :param address:
 276             Address to listen to; UNIX socket path or TCP address/port.
 277         :param ssl: SSL context to use, if any.
 278
 279         :raise StateError: When the `Runstate` is not `IDLE`.
 280         :raise ConnectError: If a connection could not be accepted.
 281         """
 282         await self._new_session(address, ssl, accept=True)
 283
 284     @upper_half
 285     @require(Runstate.IDLE)
 286     async def connect(self, address: SocketAddrT,
 287                       ssl: Optional[SSLContext] = None) -> None:
 288         """
 289         Connect to the server and begin processing message queues.
 290
 291         If this call fails, `runstate` is guaranteed to be set back to `IDLE`.
 292
 293         :param address:
 294             Address to connect to; UNIX socket path or TCP address/port.
 295         :param ssl: SSL context to use, if any.
 296
 297         :raise StateError: When the `Runstate` is not `IDLE`.
 298         :raise ConnectError: If a connection cannot be made to the server.
 299         """
 300         await self._new_session(address, ssl)
 301
 302     @upper_half
 303     async def disconnect(self) -> None:
 304         """
 305         Disconnect and wait for all tasks to fully stop.
 306
 307         If there was an exception that caused the reader/writers to
 308         terminate prematurely, it will be raised here.
 309
 310         :raise Exception: When the reader or writer terminate unexpectedly.
 311         """
 312         self.logger.debug("disconnect() called.")
 313         self._schedule_disconnect()
 314         await self._wait_disconnect()
 315
 316     # --------------------------
 317     # Section: Session machinery
 318     # --------------------------
 319
 320     async def _session_guard(self, coro: Awaitable[None], emsg: str) -> None:
 321         """
 322         Async guard function used to roll back to `IDLE` on any error.
 323
 324         On any Exception, the state machine will be reset back to
 325         `IDLE`. Most Exceptions will be wrapped with `ConnectError`, but
 326         `BaseException` events will be left alone (This includes
 327         asyncio.CancelledError, even prior to Python 3.8).
 328
 329         :param error_message:
 330             Human-readable string describing what connection phase failed.
 331
 332         :raise BaseException:
 333             When `BaseException` occurs in the guarded block.
 334         :raise ConnectError:
 335             When any other error is encountered in the guarded block.
 336         """
 337         # Note: After Python 3.6 support is removed, this should be an
 338         # @asynccontextmanager instead of accepting a callback.
 339         try:
 340             await coro
 341         except BaseException as err:
 342             self.logger.error("%s: %s", emsg, exception_summary(err))
 343             self.logger.debug("%s:\n%s\n", emsg, pretty_traceback())
 344             try:
 345                 # Reset the runstate back to IDLE.
 346                 await self.disconnect()
 347             except:
 348                 # We don't expect any Exceptions from the disconnect function
 349                 # here, because we failed to connect in the first place.
 350                 # The disconnect() function is intended to perform
 351                 # only cannot-fail cleanup here, but you never know.
 352                 emsg = (
 353                     "Unexpected bottom half exception. "
 354                     "This is a bug in the QMP library. "
 355                     "Please report it to <[email protected]> and "
 356                     "CC: John Snow <[email protected]>."
 357                 )
 358                 self.logger.critical("%s:\n%s\n", emsg, pretty_traceback())
 359                 raise
 360
 361             # CancelledError is an Exception with special semantic meaning;
 362             # We do NOT want to wrap it up under ConnectError.
 363             # NB: CancelledError is not a BaseException before Python 3.8
 364             if isinstance(err, asyncio.CancelledError):
 365                 raise
 366
 367             # Any other kind of error can be treated as some kind of connection
 368             # failure broadly. Inspect the 'exc' field to explore the root
 369             # cause in greater detail.
 370             if isinstance(err, Exception):
 371                 raise ConnectError(emsg, err) from err
 372
 373             # Raise BaseExceptions un-wrapped, they're more important.
 374             raise
 375
 376     @property
 377     def _runstate_event(self) -> asyncio.Event:
 378         # asyncio.Event() objects should not be created prior to entrance into
 379         # an event loop, so we can ensure we create it in the correct context.
 380         # Create it on-demand *only* at the behest of an 'async def' method.
 381         if not self._runstate_changed:
 382             self._runstate_changed = asyncio.Event()
 383         return self._runstate_changed
 384
 385     @upper_half
 386     @bottom_half
 387     def _set_state(self, state: Runstate) -> None:
 388         """
 389         Change the `Runstate` of the protocol connection.
 390
 391         Signals the `runstate_changed` event.
 392         """
 393         if state == self._runstate:
 394             return
 395
 396         self.logger.debug("Transitioning from '%s' to '%s'.",
 397                           str(self._runstate), str(state))
 398         self._runstate = state
 399         self._runstate_event.set()
 400         self._runstate_event.clear()
 401
 402     @upper_half
 403     async def _new_session(self,
 404                            address: SocketAddrT,
 405                            ssl: Optional[SSLContext] = None,
 406                            accept: bool = False) -> None:
 407         """
 408         Establish a new connection and initialize the session.
 409
 410         Connect or accept a new connection, then begin the protocol
 411         session machinery. If this call fails, `runstate` is guaranteed
 412         to be set back to `IDLE`.
 413
 414         :param address:
 415             Address to connect to/listen on;
 416             UNIX socket path or TCP address/port.
 417         :param ssl: SSL context to use, if any.
 418         :param accept: Accept a connection instead of connecting when `True`.
 419
 420         :raise ConnectError:
 421             When a connection or session cannot be established.
 422
 423             This exception will wrap a more concrete one. In most cases,
 424             the wrapped exception will be `OSError` or `EOFError`. If a
 425             protocol-level failure occurs while establishing a new
 426             session, the wrapped error may also be an `QMPError`.
 427         """
 428         assert self.runstate == Runstate.IDLE
 429
 430         await self._session_guard(
 431             self._establish_connection(address, ssl, accept),
 432             'Failed to establish connection')
 433
 434         await self._session_guard(
 435             self._establish_session(),
 436             'Failed to establish session')
 437
 438         assert self.runstate == Runstate.RUNNING
 439
 440     @upper_half
 441     async def _establish_connection(
 442             self,
 443             address: SocketAddrT,
 444             ssl: Optional[SSLContext] = None,
 445             accept: bool = False
 446     ) -> None:
 447         """
 448         Establish a new connection.
 449
 450         :param address:
 451             Address to connect to/listen on;
 452             UNIX socket path or TCP address/port.
 453         :param ssl: SSL context to use, if any.
 454         :param accept: Accept a connection instead of connecting when `True`.
 455         """
 456         assert self.runstate == Runstate.IDLE
 457         self._set_state(Runstate.CONNECTING)
 458
 459         # Allow runstate watchers to witness 'CONNECTING' state; some
 460         # failures in the streaming layer are synchronous and will not
 461         # otherwise yield.
 462         await asyncio.sleep(0)
 463
 464         if accept:
 465             await self._do_accept(address, ssl)
 466         else:
 467             await self._do_connect(address, ssl)
 468
 469     def _bind_hack(self, address: Union[str, Tuple[str, int]]) -> None:
 470         """
 471         Used to create a socket in advance of accept().
 472
 473         This is a workaround to ensure that we can guarantee timing of
 474         precisely when a socket exists to avoid a connection attempt
 475         bouncing off of nothing.
 476
 477         Python 3.7+ adds a feature to separate the server creation and
 478         listening phases instead, and should be used instead of this
 479         hack.
 480         """
 481         if isinstance(address, tuple):
 482             family = socket.AF_INET
 483         else:
 484             family = socket.AF_UNIX
 485
 486         sock = socket.socket(family, socket.SOCK_STREAM)
 487         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 488
 489         try:
 490             sock.bind(address)
 491         except:
 492             sock.close()
 493             raise
 494
 495         self._sock = sock
 496
 497     @upper_half
 498     async def _do_accept(self, address: SocketAddrT,
 499                          ssl: Optional[SSLContext] = None) -> None:
 500         """
 501         Acting as the transport server, accept a single connection.
 502
 503         :param address:
 504             Address to listen on; UNIX socket path or TCP address/port.
 505         :param ssl: SSL context to use, if any.
 506
 507         :raise OSError: For stream-related errors.
 508         """
 509         self.logger.debug("Awaiting connection on %s ...", address)
 510         connected = asyncio.Event()
 511         server: Optional[asyncio.AbstractServer] = None
 512
 513         async def _client_connected_cb(reader: asyncio.StreamReader,
 514                                        writer: asyncio.StreamWriter) -> None:
 515             """Used to accept a single incoming connection, see below."""
 516             nonlocal server
 517             nonlocal connected
 518
 519             # A connection has been accepted; stop listening for new ones.
 520             assert server is not None
 521             server.close()
 522             await server.wait_closed()
 523             server = None
 524
 525             # Register this client as being connected
 526             self._reader, self._writer = (reader, writer)
 527
 528             # Signal back: We've accepted a client!
 529             connected.set()
 530
 531         if isinstance(address, tuple):
 532             coro = asyncio.start_server(
 533                 _client_connected_cb,
 534                 host=None if self._sock else address[0],
 535                 port=None if self._sock else address[1],
 536                 ssl=ssl,
 537                 backlog=1,
 538                 limit=self._limit,
 539                 sock=self._sock,
 540             )
 541         else:
 542             coro = asyncio.start_unix_server(
 543                 _client_connected_cb,
 544                 path=None if self._sock else address,
 545                 ssl=ssl,
 546                 backlog=1,
 547                 limit=self._limit,
 548                 sock=self._sock,
 549             )
 550
 551         server = await coro     # Starts listening
 552         await connected.wait()  # Waits for the callback to fire (and finish)
 553         assert server is None
 554         self._sock = None
 555
 556         self.logger.debug("Connection accepted.")
 557
 558     @upper_half
 559     async def _do_connect(self, address: SocketAddrT,
 560                           ssl: Optional[SSLContext] = None) -> None:
 561         """
 562         Acting as the transport client, initiate a connection to a server.
 563
 564         :param address:
 565             Address to connect to; UNIX socket path or TCP address/port.
 566         :param ssl: SSL context to use, if any.
 567
 568         :raise OSError: For stream-related errors.
 569         """
 570         self.logger.debug("Connecting to %s ...", address)
 571
 572         if isinstance(address, tuple):
 573             connect = asyncio.open_connection(
 574                 address[0],
 575                 address[1],
 576                 ssl=ssl,
 577                 limit=self._limit,
 578             )
 579         else:
 580             connect = asyncio.open_unix_connection(
 581                 path=address,
 582                 ssl=ssl,
 583                 limit=self._limit,
 584             )
 585         self._reader, self._writer = await connect
 586
 587         self.logger.debug("Connected.")
 588
 589     @upper_half
 590     async def _establish_session(self) -> None:
 591         """
 592         Establish a new session.
 593
 594         Starts the readers/writer tasks; subclasses may perform their
 595         own negotiations here. The Runstate will be RUNNING upon
 596         successful conclusion.
 597         """
 598         assert self.runstate == Runstate.CONNECTING
 599
 600         self._outgoing = asyncio.Queue()
 601
 602         reader_coro = self._bh_loop_forever(self._bh_recv_message, 'Reader')
 603         writer_coro = self._bh_loop_forever(self._bh_send_message, 'Writer')
 604
 605         self._reader_task = create_task(reader_coro)
 606         self._writer_task = create_task(writer_coro)
 607
 608         self._bh_tasks = asyncio.gather(
 609             self._reader_task,
 610             self._writer_task,
 611         )
 612
 613         self._set_state(Runstate.RUNNING)
 614         await asyncio.sleep(0)  # Allow runstate_event to process
 615
 616     @upper_half
 617     @bottom_half
 618     def _schedule_disconnect(self) -> None:
 619         """
 620         Initiate a disconnect; idempotent.
 621
 622         This method is used both in the upper-half as a direct
 623         consequence of `disconnect()`, and in the bottom-half in the
 624         case of unhandled exceptions in the reader/writer tasks.
 625
 626         It can be invoked no matter what the `runstate` is.
 627         """
 628         if not self._dc_task:
 629             self._set_state(Runstate.DISCONNECTING)
 630             self.logger.debug("Scheduling disconnect.")
 631             self._dc_task = create_task(self._bh_disconnect())
 632
 633     @upper_half
 634     async def _wait_disconnect(self) -> None:
 635         """
 636         Waits for a previously scheduled disconnect to finish.
 637
 638         This method will gather any bottom half exceptions and re-raise
 639         the one that occurred first; presuming it to be the root cause
 640         of any subsequent Exceptions. It is intended to be used in the
 641         upper half of the call chain.
 642
 643         :raise Exception:
 644             Arbitrary exception re-raised on behalf of the reader/writer.
 645         """
 646         assert self.runstate == Runstate.DISCONNECTING
 647         assert self._dc_task
 648
 649         aws: List[Awaitable[object]] = [self._dc_task]
 650         if self._bh_tasks:
 651             aws.insert(0, self._bh_tasks)
 652         all_defined_tasks = asyncio.gather(*aws)
 653
 654         # Ensure disconnect is done; Exception (if any) is not raised here:
 655         await asyncio.wait((self._dc_task,))
 656
 657         try:
 658             await all_defined_tasks  # Raise Exceptions from the bottom half.
 659         finally:
 660             self._cleanup()
 661             self._set_state(Runstate.IDLE)
 662
 663     @upper_half
 664     def _cleanup(self) -> None:
 665         """
 666         Fully reset this object to a clean state and return to `IDLE`.
 667         """
 668         def _paranoid_task_erase(task: Optional['asyncio.Future[_U]']
 669                                  ) -> Optional['asyncio.Future[_U]']:
 670             # Help to erase a task, ENSURING it is fully quiesced first.
 671             assert (task is None) or task.done()
 672             return None if (task and task.done()) else task
 673
 674         assert self.runstate == Runstate.DISCONNECTING
 675         self._dc_task = _paranoid_task_erase(self._dc_task)
 676         self._reader_task = _paranoid_task_erase(self._reader_task)
 677         self._writer_task = _paranoid_task_erase(self._writer_task)
 678         self._bh_tasks = _paranoid_task_erase(self._bh_tasks)
 679
 680         self._reader = None
 681         self._writer = None
 682
 683         # NB: _runstate_changed cannot be cleared because we still need it to
 684         # send the final runstate changed event ...!
 685
 686     # ----------------------------
 687     # Section: Bottom Half methods
 688     # ----------------------------
 689
 690     @bottom_half
 691     async def _bh_disconnect(self) -> None:
 692         """
 693         Disconnect and cancel all outstanding tasks.
 694
 695         It is designed to be called from its task context,
 696         :py:obj:`~AsyncProtocol._dc_task`. By running in its own task,
 697         it is free to wait on any pending actions that may still need to
 698         occur in either the reader or writer tasks.
 699         """
 700         assert self.runstate == Runstate.DISCONNECTING
 701
 702         def _done(task: Optional['asyncio.Future[Any]']) -> bool:
 703             return task is not None and task.done()
 704
 705         # Are we already in an error pathway? If either of the tasks are
 706         # already done, or if we have no tasks but a reader/writer; we
 707         # must be.
 708         #
 709         # NB: We can't use _bh_tasks to check for premature task
 710         # completion, because it may not yet have had a chance to run
 711         # and gather itself.
 712         tasks = tuple(filter(None, (self._writer_task, self._reader_task)))
 713         error_pathway = _done(self._reader_task) or _done(self._writer_task)
 714         if not tasks:
 715             error_pathway |= bool(self._reader) or bool(self._writer)
 716
 717         try:
 718             # Try to flush the writer, if possible.
 719             # This *may* cause an error and force us over into the error path.
 720             if not error_pathway:
 721                 await self._bh_flush_writer()
 722         except BaseException as err:
 723             error_pathway = True
 724             emsg = "Failed to flush the writer"
 725             self.logger.error("%s: %s", emsg, exception_summary(err))
 726             self.logger.debug("%s:\n%s\n", emsg, pretty_traceback())
 727             raise
 728         finally:
 729             # Cancel any still-running tasks (Won't raise):
 730             if self._writer_task is not None and not self._writer_task.done():
 731                 self.logger.debug("Cancelling writer task.")
 732                 self._writer_task.cancel()
 733             if self._reader_task is not None and not self._reader_task.done():
 734                 self.logger.debug("Cancelling reader task.")
 735                 self._reader_task.cancel()
 736
 737             # Close out the tasks entirely (Won't raise):
 738             if tasks:
 739                 self.logger.debug("Waiting for tasks to complete ...")
 740                 await asyncio.wait(tasks)
 741
 742             # Lastly, close the stream itself. (*May raise*!):
 743             await self._bh_close_stream(error_pathway)
 744             self.logger.debug("Disconnected.")
 745
 746     @bottom_half
 747     async def _bh_flush_writer(self) -> None:
 748         if not self._writer_task:
 749             return
 750
 751         self.logger.debug("Draining the outbound queue ...")
 752         await self._outgoing.join()
 753         if self._writer is not None:
 754             self.logger.debug("Flushing the StreamWriter ...")
 755             await flush(self._writer)
 756
 757     @bottom_half
 758     async def _bh_close_stream(self, error_pathway: bool = False) -> None:
 759         # NB: Closing the writer also implcitly closes the reader.
 760         if not self._writer:
 761             return
 762
 763         if not is_closing(self._writer):
 764             self.logger.debug("Closing StreamWriter.")
 765             self._writer.close()
 766
 767         self.logger.debug("Waiting for StreamWriter to close ...")
 768         try:
 769             await wait_closed(self._writer)
 770         except Exception:  # pylint: disable=broad-except
 771             # It's hard to tell if the Stream is already closed or
 772             # not. Even if one of the tasks has failed, it may have
 773             # failed for a higher-layered protocol reason. The
 774             # stream could still be open and perfectly fine.
 775             # I don't know how to discern its health here.
 776
 777             if error_pathway:
 778                 # We already know that *something* went wrong. Let's
 779                 # just trust that the Exception we already have is the
 780                 # better one to present to the user, even if we don't
 781                 # genuinely *know* the relationship between the two.
 782                 self.logger.debug(
 783                     "Discarding Exception from wait_closed:\n%s\n",
 784                     pretty_traceback(),
 785                 )
 786             else:
 787                 # Oops, this is a brand-new error!
 788                 raise
 789         finally:
 790             self.logger.debug("StreamWriter closed.")
 791
 792     @bottom_half
 793     async def _bh_loop_forever(self, async_fn: _TaskFN, name: str) -> None:
 794         """
 795         Run one of the bottom-half methods in a loop forever.
 796
 797         If the bottom half ever raises any exception, schedule a
 798         disconnect that will terminate the entire loop.
 799
 800         :param async_fn: The bottom-half method to run in a loop.
 801         :param name: The name of this task, used for logging.
 802         """
 803         try:
 804             while True:
 805                 await async_fn()
 806         except asyncio.CancelledError:
 807             # We have been cancelled by _bh_disconnect, exit gracefully.
 808             self.logger.debug("Task.%s: cancelled.", name)
 809             return
 810         except BaseException as err:
 811             self.logger.log(
 812                 logging.INFO if isinstance(err, EOFError) else logging.ERROR,
 813                 "Task.%s: %s",
 814                 name, exception_summary(err)
 815             )
 816             self.logger.debug("Task.%s: failure:\n%s\n",
 817                               name, pretty_traceback())
 818             self._schedule_disconnect()
 819             raise
 820         finally:
 821             self.logger.debug("Task.%s: exiting.", name)
 822
 823     @bottom_half
 824     async def _bh_send_message(self) -> None:
 825         """
 826         Wait for an outgoing message, then send it.
 827
 828         Designed to be run in `_bh_loop_forever()`.
 829         """
 830         msg = await self._outgoing.get()
 831         try:
 832             await self._send(msg)
 833         finally:
 834             self._outgoing.task_done()
 835
 836     @bottom_half
 837     async def _bh_recv_message(self) -> None:
 838         """
 839         Wait for an incoming message and call `_on_message` to route it.
 840
 841         Designed to be run in `_bh_loop_forever()`.
 842         """
 843         msg = await self._recv()
 844         await self._on_message(msg)
 845
 846     # --------------------
 847     # Section: Message I/O
 848     # --------------------
 849
 850     @upper_half
 851     @bottom_half
 852     def _cb_outbound(self, msg: T) -> T:
 853         """
 854         Callback: outbound message hook.
 855
 856         This is intended for subclasses to be able to add arbitrary
 857         hooks to filter or manipulate outgoing messages. The base
 858         implementation does nothing but log the message without any
 859         manipulation of the message.
 860
 861         :param msg: raw outbound message
 862         :return: final outbound message
 863         """
 864         self.logger.debug("--> %s", str(msg))
 865         return msg
 866
 867     @upper_half
 868     @bottom_half
 869     def _cb_inbound(self, msg: T) -> T:
 870         """
 871         Callback: inbound message hook.
 872
 873         This is intended for subclasses to be able to add arbitrary
 874         hooks to filter or manipulate incoming messages. The base
 875         implementation does nothing but log the message without any
 876         manipulation of the message.
 877
 878         This method does not "handle" incoming messages; it is a filter.
 879         The actual "endpoint" for incoming messages is `_on_message()`.
 880
 881         :param msg: raw inbound message
 882         :return: processed inbound message
 883         """
 884         self.logger.debug("<-- %s", str(msg))
 885         return msg
 886
 887     @upper_half
 888     @bottom_half
 889     async def _readline(self) -> bytes:
 890         """
 891         Wait for a newline from the incoming reader.
 892
 893         This method is provided as a convenience for upper-layer
 894         protocols, as many are line-based.
 895
 896         This method *may* return a sequence of bytes without a trailing
 897         newline if EOF occurs, but *some* bytes were received. In this
 898         case, the next call will raise `EOFError`. It is assumed that
 899         the layer 5 protocol will decide if there is anything meaningful
 900         to be done with a partial message.
 901
 902         :raise OSError: For stream-related errors.
 903         :raise EOFError:
 904             If the reader stream is at EOF and there are no bytes to return.
 905         :return: bytes, including the newline.
 906         """
 907         assert self._reader is not None
 908         msg_bytes = await self._reader.readline()
 909
 910         if not msg_bytes:
 911             if self._reader.at_eof():
 912                 raise EOFError
 913
 914         return msg_bytes
 915
 916     @upper_half
 917     @bottom_half
 918     async def _do_recv(self) -> T:
 919         """
 920         Abstract: Read from the stream and return a message.
 921
 922         Very low-level; intended to only be called by `_recv()`.
 923         """
 924         raise NotImplementedError
 925
 926     @upper_half
 927     @bottom_half
 928     async def _recv(self) -> T:
 929         """
 930         Read an arbitrary protocol message.
 931
 932         .. warning::
 933             This method is intended primarily for `_bh_recv_message()`
 934             to use in an asynchronous task loop. Using it outside of
 935             this loop will "steal" messages from the normal routing
 936             mechanism. It is safe to use prior to `_establish_session()`,
 937             but should not be used otherwise.
 938
 939         This method uses `_do_recv()` to retrieve the raw message, and
 940         then transforms it using `_cb_inbound()`.
 941
 942         :return: A single (filtered, processed) protocol message.
 943         """
 944         message = await self._do_recv()
 945         return self._cb_inbound(message)
 946
 947     @upper_half
 948     @bottom_half
 949     def _do_send(self, msg: T) -> None:
 950         """
 951         Abstract: Write a message to the stream.
 952
 953         Very low-level; intended to only be called by `_send()`.
 954         """
 955         raise NotImplementedError
 956
 957     @upper_half
 958     @bottom_half
 959     async def _send(self, msg: T) -> None:
 960         """
 961         Send an arbitrary protocol message.
 962
 963         This method will transform any outgoing messages according to
 964         `_cb_outbound()`.
 965
 966         .. warning::
 967             Like `_recv()`, this method is intended to be called by
 968             the writer task loop that processes outgoing
 969             messages. Calling it directly may circumvent logic
 970             implemented by the caller meant to correlate outgoing and
 971             incoming messages.
 972
 973         :raise OSError: For problems with the underlying stream.
 974         """
 975         msg = self._cb_outbound(msg)
 976         self._do_send(msg)
 977
 978     @bottom_half
 979     async def _on_message(self, msg: T) -> None:
 980         """
 981         Called to handle the receipt of a new message.
 982
 983         .. caution::
 984             This is executed from within the reader loop, so be advised
 985             that waiting on either the reader or writer task will lead
 986             to deadlock. Additionally, any unhandled exceptions will
 987             directly cause the loop to halt, so logic may be best-kept
 988             to a minimum if at all possible.
 989
 990         :param msg: The incoming message, already logged/filtered.
 991         """
 992         # Nothing to do in the abstract case.