GitLab: https://gitlab.com/jsnow/qemu/-/commits/python-async-qmp-aqmp
CI: https://gitlab.com/jsnow/qemu/-/pipelines/338508045
Docs: https://people.redhat.com/~jsnow/sphinx/html/qemu.aqmp.html

Hi!

This patch series adds an Asynchronous QMP package to the Python
library. It offers a few improvements over the previous library:

- out-of-band support
- true asynchronous event support
- avoids undocumented interfaces abusing non-blocking sockets
- unit tests!
- documentation!

This library serves as the basis for a new qmp-shell program that will
offer improved reconnection support, true asynchronous display of
events, VM and job status update notifiers, and so on.

My intent is to eventually publish this library directly to PyPI as a
standalone package. I would like to phase out our usage of the old QMP
library over time; eventually replacing it entirely with this one.

This series looks big by line count, but it's *mostly*
docstrings. Seriously!

This package has *no* external dependencies whatsoever.

Notes & Design
==============

Here are some notes on the design of how the library works, to serve as
a primer for review; however I also **highly recommend** browsing the
generated Sphinx documentation for this series.

Here's that link again:
https://people.redhat.com/~jsnow/sphinx/html/qemu.aqmp.html

The core machinery is split between the AsyncProtocol and QMP
classes. AsyncProtocol provides the generic machinery, while QMP
provides the QMP-specific details.

The design uses two independent coroutines that act as the "bottom
half", a writer task and a reader task. These tasks run for the duration
of the connection and independently send and receive messages,
respectively.

A third task, disconnect, is scheduled asynchronously whenever an
unrecoverable error occurs and facilitates coalescing of the other two
tasks.

This diagram for how execute() operates may be helpful for understanding
how AsyncProtocol is laid out. The arrows indicate the direction of a
QMP message; the long horizontal dash indicates the separation between
the upper and lower half of the event loop. The queue mechanisms between
both dashes serve as the intermediaries between the upper and lower
half.

                       +---------+
                       | caller  |
                       +---------+
                           ^ |
                           | v
                       +---------+
     +---------------> |execute()| -----------+
     |                 +---------+            |
     |                                        |
[-----------------------------------------------------------]
     |                                        |
     |                                        v
+----+------+    +----------------+    +------+-------+
| ExecQueue |    | EventListeners |    |Outbound Queue|
+----+------+    +----+-----------+    +------+-------+
     ^                ^                       |
     |                |                       |
[-----------------------------------------------------------]
     |                |                       |
     |                |                       v
  +--+----------------+---+       +-----------+-----------+
  | Reader Task/Coroutine |       | Writer Task/Coroutine |
  +-----------+-----------+       +-----------+-----------+
              ^                               |
              |                               v
        +-----+------+                  +-----+------+
        |StreamReader|                  |StreamWriter|
        +------------+                  +------------+

The caller will invoke execute(), which in turn will deposit a message
in the outbound send queue. This will wake up the writer task, which
well send the message over the wire.

The execute() method will then yield to wait for a reply delivered to an
execution queue created solely for that execute statement.

When a message arrives, the Reader task will unblock and route the
message either to the EventListener subsystem, or place it in the
appropriate pending execution queue.

Once a message is placed in the pending execution queue, execute() will
unblock and the execution will conclude, returning the result of the RPC
call to the caller.

Patch Layout
============

Patches 1-4   add tiny pre-requisites, utilities, etc.
Patches 5-12  add a generic async message-based protocol class,
              AsyncProtocol. They are split fairly small and should
              be reasonably self-contained.
Patches 13-15 check in more QMP-centric components.
Patches 16-21 add qmp_client.py, with a new 'QMPClient()' class.
              They're split into reasonably tiny pieces here.
Patches 22-23 add a few finishing touches, they are small patches.
Patch 24      adds unit tests. They're maybe a little messy yet, but
              they've been quite helpful to me so far. Coverage of
              protocol.py is at about 86%.

Future Work
===========

These items are in progress:

- A synchronous QMP wrapper that allows this library to be easily used
  from non-async code; this will also allow me to prove it works well by
  demoing its replacement throughout iotests. I have all of iotests
  passing locally, but I am still seeing some failures on gitlab CI I
  need to diagnose, possibly a race condition somewhere.

- A QMP server class; to facilitate writing of unit tests. It's done,
  but needs some polish and tests.

- More unit tests for qmp_client.py, qmp_server.py and other modules.

Changelog
=========

V2:

Renamed classes/methods:

- Renamed qmp_protocol.py to qmp_client.py
- Renamed 'QMP' class to 'QMPClient'
- Renamed _begin_new_session() to _establish_session()
- Split _establish_connection() out from _new_session().
- Removed _results() method

Bugfixes:

- Suppress duplicate Exceptions when attempting to drain the
  StreamWriter
- Delay initialization of asyncio.Queue and asyncio.Event variables to
  _new_session or later -- they must not be created outside of the loop,
  even if they are not async functions.
- Rework runstate_changed events to guarantee visibility of events to
  waiters
- Improve connect()/accept() cleanup to work with
  asyncio.CancelledError, asyncio.TimeoutError
- No-argument form of Message() now succeeds properly.
- flush utility will correctly yield when data is below the "high water
  mark", giving the stream a chance to actually flush.
- Increase read buffer size to accommodate query-qmp-schema (Thanks
  Niteesh)

Ugly bits from V1 removed:

- Remove tertiary filtering from EventListener (for now), accompanying
  documentation removed from events.py
- Use asyncio.wait() instead of custom wait_task_done()
- MultiException is removed in favor of just raising the first Exception
  that occurs in the bottom half; other Exceptions if any are logged
  instead.

Improvements:

- QMPClient now allows ID-less execution statements via the _raw()
  interface.
- Add tests that grant ~86% coverage of protocol.py to the avocado test
  suite.
- Removed 'force' parameter from _bh_disconnect; the disconnection
  routine determines for itself if we are in the error pathway or not
  instead now.  This removes any chance of duplicate calls to
  _schedule_disconnect accidentally dropping the 'force' setting.

Debugging/Testing changes:

- Add debug: bool parameter to asyncio_run utility wrapper
- Improve error messages for '@require' decorator
- Add debugging message for state change events
- Avoid flushing the StreamWriter if we don't have one (This
  circumstance only arises in testing, but it's helpful.)
- Improved __repr__ method for AsyncProtocol, and removed __str__
  method.  enforcing eval(__repr__(x)) == x does not make sense for
  AsyncProtocol.
- Misc logging message changes
- Add a suite of fancy Task debugging utilities.
- Most tracebacks now log at the DEBUG level instead of
  CRITICAL/ERROR/WARNING; In those error cases, a one-line summary is
  logged instead.

Misc. aesthetic changes:

- Misc docstring fixes, whitespace, etc.
- Reordered the definition of some methods to try and keep similar
  methods near each other (Moved _cleanup near _bh_disconnect in
  QMPClient.)

~ Shucks Howdy, Gee Golly!

John Snow (24):
  python/aqmp: add asynchronous QMP (AQMP) subpackage
  python/aqmp: add error classes
  python/pylint: Add exception for TypeVar names ('T')
  python/aqmp: add asyncio compatibility wrappers
  python/aqmp: add generic async message-based protocol support
  python/aqmp: add runstate state machine to AsyncProtocol
  python/aqmp: Add logging utility helpers
  python/aqmp: add logging to AsyncProtocol
  python/aqmp: add AsyncProtocol.accept() method
  python/aqmp: add configurable read buffer limit
  python/aqmp: add _cb_inbound and _cb_inbound logging hooks
  python/aqmp: add AsyncProtocol._readline() method
  python/aqmp: add QMP Message format
  python/aqmp: add well-known QMP object models
  python/aqmp: add QMP event support
  python/pylint: disable too-many-function-args
  python/aqmp: add QMP protocol support
  python/pylint: disable no-member check
  python/aqmp: Add message routing to QMP protocol
  python/aqmp: add execute() interfaces
  python/aqmp: add _raw() execution interface
  python/aqmp: add asyncio_run compatibility wrapper
  python/aqmp: add scary message
  python/aqmp: add AsyncProtocol unit tests

 python/qemu/aqmp/__init__.py   |  58 +++
 python/qemu/aqmp/error.py      |  50 ++
 python/qemu/aqmp/events.py     | 706 ++++++++++++++++++++++++++
 python/qemu/aqmp/message.py    | 209 ++++++++
 python/qemu/aqmp/models.py     | 133 +++++
 python/qemu/aqmp/protocol.py   | 882 +++++++++++++++++++++++++++++++++
 python/qemu/aqmp/py.typed      |   0
 python/qemu/aqmp/qmp_client.py | 621 +++++++++++++++++++++++
 python/qemu/aqmp/util.py       | 207 ++++++++
 python/setup.cfg               |   5 +-
 python/tests/null_proto.py     |  67 +++
 python/tests/protocol.py       | 458 +++++++++++++++++
 12 files changed, 3395 insertions(+), 1 deletion(-)
 create mode 100644 python/qemu/aqmp/__init__.py
 create mode 100644 python/qemu/aqmp/error.py
 create mode 100644 python/qemu/aqmp/events.py
 create mode 100644 python/qemu/aqmp/message.py
 create mode 100644 python/qemu/aqmp/models.py
 create mode 100644 python/qemu/aqmp/protocol.py
 create mode 100644 python/qemu/aqmp/py.typed
 create mode 100644 python/qemu/aqmp/qmp_client.py
 create mode 100644 python/qemu/aqmp/util.py
 create mode 100644 python/tests/null_proto.py
 create mode 100644 python/tests/protocol.py

-- 
2.31.1

Hello all,

I recently rebased(incrementally) my TUI on this V2 patch and faced an
issue.
https://gitlab.com/niteesh.gs/qemu/-/commits/aqmp-tui-prototype-v3
I decided to rebase incrementally so that I can address some of the
comments posted
in my patch series. While testing out, the initial draft of TUI
which worked fine in the V1
version of AQMP failed in this version.

Disconnecting from a fully connected state doesn't exit cleanly.
---------------------------------------------------------------------------------
To reproduce the issue:
1) Initiate a QMP server
2) Connect the TUI to the server using aqmp-tui localhost:1234 --log-file
log.txt
3) Once the TUI is connected and running, press 'Esc' to exit the app. This
should result
in the following exception.
--------------------------------------------------------------------------------------------------------------------------------------------
Transitioning from 'Runstate.IDLE' to 'Runstate.CONNECTING'.
Connecting to ('localhost', 1234) ...
Connected.
Awaiting greeting ...
Response: {
  "QMP": {
    .......... Skipping
  }
}
Negotiating capabilities ...
Request: {
  "execute": "qmp_capabilities",
    .......... Skipping
  }
}
Response: {
  "return": {}
}
Transitioning from 'Runstate.CONNECTING' to 'Runstate.RUNNING'.
Transitioning from 'Runstate.RUNNING' to 'Runstate.DISCONNECTING'.
Scheduling disconnect.
Draining the outbound queue ...
Flushing the StreamWriter ...
Cancelling writer task ...
Task.Writer: cancelled.
Task.Writer: exiting.
Cancelling reader task ...
Task.Reader: cancelled.
Task.Reader: exiting.
Closing StreamWriter.
Waiting for StreamWriter to close ...
QMP Disconnected.
Transitioning from 'Runstate.DISCONNECTING' to 'Runstate.IDLE'.
_kill_app: Connection lost
Connection lost
  | Traceback (most recent call last):
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/aqmp_tui.py",
line 246, in run
  |     main_loop.run()
  |   File
"/home/niteesh/development/qemu/python/.venv/lib/python3.6/site-packages/urwid/main_loop.py",
line 287, in run
  |     self._run()
  |   File
"/home/niteesh/development/qemu/python/.venv/lib/python3.6/site-packages/urwid/main_loop.py",
line 385, in _run
  |     self.event_loop.run()
  |   File
"/home/niteesh/development/qemu/python/.venv/lib/python3.6/site-packages/urwid/main_loop.py",
line 1494, in run
  |     reraise(*exc_info)
  |   File
"/home/niteesh/development/qemu/python/.venv/lib/python3.6/site-packages/urwid/compat.py",
line 58, in reraise
  |     raise value
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/aqmp_tui.py",
line 206, in _kill_app
  |     raise err
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/aqmp_tui.py",
line 201, in _kill_app
  |     await self.disconnect()
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/protocol.py",
line 303, in disconnect
  |     await self._wait_disconnect()
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/protocol.py",
line 573, in _wait_disconnect
  |     await self._dc_task
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/qmp_client.py",
line 316, in _bh_disconnect
  |     await super()._bh_disconnect()
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/protocol.py",
line 644, in _bh_disconnect
  |     await wait_closed(self._writer)
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/util.py", line
137, in wait_closed
  |     await flush(writer)
  |   File "/home/niteesh/development/qemu/python/qemu/aqmp/util.py", line
49, in flush
  |     await writer.drain()
  |   File "/usr/lib/python3.6/asyncio/streams.py", line 339, in drain
  |     yield from self._protocol._drain_helper()
  |   File "/usr/lib/python3.6/asyncio/streams.py", line 210, in
_drain_helper
  |     raise ConnectionResetError('Connection lost')
  | ConnectionResetError: Connection lost
--------------------------------------------------------------------------------------------------------------------------------------------


On Sat, Jul 17, 2021 at 6:03 AM John Snow <jsnow@redhat.com> wrote:

> GitLab: https://gitlab.com/jsnow/qemu/-/commits/python-async-qmp-aqmp
> CI: https://gitlab.com/jsnow/qemu/-/pipelines/338508045
> Docs: https://people.redhat.com/~jsnow/sphinx/html/qemu.aqmp.html
>
> Hi!
>
> This patch series adds an Asynchronous QMP package to the Python
> library. It offers a few improvements over the previous library:
>
> - out-of-band support
> - true asynchronous event support
> - avoids undocumented interfaces abusing non-blocking sockets
> - unit tests!
> - documentation!
>
> This library serves as the basis for a new qmp-shell program that will
> offer improved reconnection support, true asynchronous display of
> events, VM and job status update notifiers, and so on.
>
> My intent is to eventually publish this library directly to PyPI as a
> standalone package. I would like to phase out our usage of the old QMP
> library over time; eventually replacing it entirely with this one.
>
> This series looks big by line count, but it's *mostly*
> docstrings. Seriously!
>
> This package has *no* external dependencies whatsoever.
>
> Notes & Design
> ==============
>
> Here are some notes on the design of how the library works, to serve as
> a primer for review; however I also **highly recommend** browsing the
> generated Sphinx documentation for this series.
>
> Here's that link again:
> https://people.redhat.com/~jsnow/sphinx/html/qemu.aqmp.html
>
> The core machinery is split between the AsyncProtocol and QMP
> classes. AsyncProtocol provides the generic machinery, while QMP
> provides the QMP-specific details.
>
> The design uses two independent coroutines that act as the "bottom
> half", a writer task and a reader task. These tasks run for the duration
> of the connection and independently send and receive messages,
> respectively.
>
> A third task, disconnect, is scheduled asynchronously whenever an
> unrecoverable error occurs and facilitates coalescing of the other two
> tasks.
>
> This diagram for how execute() operates may be helpful for understanding
> how AsyncProtocol is laid out. The arrows indicate the direction of a
> QMP message; the long horizontal dash indicates the separation between
> the upper and lower half of the event loop. The queue mechanisms between
> both dashes serve as the intermediaries between the upper and lower
> half.
>
>                        +---------+
>                        | caller  |
>                        +---------+
>                            ^ |
>                            | v
>                        +---------+
>      +---------------> |execute()| -----------+
>      |                 +---------+            |
>      |                                        |
> [-----------------------------------------------------------]
>      |                                        |
>      |                                        v
> +----+------+    +----------------+    +------+-------+
> | ExecQueue |    | EventListeners |    |Outbound Queue|
> +----+------+    +----+-----------+    +------+-------+
>      ^                ^                       |
>      |                |                       |
> [-----------------------------------------------------------]
>      |                |                       |
>      |                |                       v
>   +--+----------------+---+       +-----------+-----------+
>   | Reader Task/Coroutine |       | Writer Task/Coroutine |
>   +-----------+-----------+       +-----------+-----------+
>               ^                               |
>               |                               v
>         +-----+------+                  +-----+------+
>         |StreamReader|                  |StreamWriter|
>         +------------+                  +------------+
>
> The caller will invoke execute(), which in turn will deposit a message
> in the outbound send queue. This will wake up the writer task, which
> well send the message over the wire.
>
> The execute() method will then yield to wait for a reply delivered to an
> execution queue created solely for that execute statement.
>
> When a message arrives, the Reader task will unblock and route the
> message either to the EventListener subsystem, or place it in the
> appropriate pending execution queue.
>
> Once a message is placed in the pending execution queue, execute() will
> unblock and the execution will conclude, returning the result of the RPC
> call to the caller.
>
> Patch Layout
> ============
>
> Patches 1-4   add tiny pre-requisites, utilities, etc.
> Patches 5-12  add a generic async message-based protocol class,
>               AsyncProtocol. They are split fairly small and should
>               be reasonably self-contained.
> Patches 13-15 check in more QMP-centric components.
> Patches 16-21 add qmp_client.py, with a new 'QMPClient()' class.
>               They're split into reasonably tiny pieces here.
> Patches 22-23 add a few finishing touches, they are small patches.
> Patch 24      adds unit tests. They're maybe a little messy yet, but
>               they've been quite helpful to me so far. Coverage of
>               protocol.py is at about 86%.
>
> Future Work
> ===========
>
> These items are in progress:
>
> - A synchronous QMP wrapper that allows this library to be easily used
>   from non-async code; this will also allow me to prove it works well by
>   demoing its replacement throughout iotests. I have all of iotests
>   passing locally, but I am still seeing some failures on gitlab CI I
>   need to diagnose, possibly a race condition somewhere.
>
> - A QMP server class; to facilitate writing of unit tests. It's done,
>   but needs some polish and tests.
>
> - More unit tests for qmp_client.py, qmp_server.py and other modules.
>
> Changelog
> =========
>
> V2:
>
> Renamed classes/methods:
>
> - Renamed qmp_protocol.py to qmp_client.py
> - Renamed 'QMP' class to 'QMPClient'
> - Renamed _begin_new_session() to _establish_session()
> - Split _establish_connection() out from _new_session().
> - Removed _results() method
>
> Bugfixes:
>
> - Suppress duplicate Exceptions when attempting to drain the
>   StreamWriter
> - Delay initialization of asyncio.Queue and asyncio.Event variables to
>   _new_session or later -- they must not be created outside of the loop,
>   even if they are not async functions.
> - Rework runstate_changed events to guarantee visibility of events to
>   waiters
> - Improve connect()/accept() cleanup to work with
>   asyncio.CancelledError, asyncio.TimeoutError
> - No-argument form of Message() now succeeds properly.
> - flush utility will correctly yield when data is below the "high water
>   mark", giving the stream a chance to actually flush.
> - Increase read buffer size to accommodate query-qmp-schema (Thanks
>   Niteesh)
>
> Ugly bits from V1 removed:
>
> - Remove tertiary filtering from EventListener (for now), accompanying
>   documentation removed from events.py
> - Use asyncio.wait() instead of custom wait_task_done()
> - MultiException is removed in favor of just raising the first Exception
>   that occurs in the bottom half; other Exceptions if any are logged
>   instead.
>
> Improvements:
>
> - QMPClient now allows ID-less execution statements via the _raw()
>   interface.
> - Add tests that grant ~86% coverage of protocol.py to the avocado test
>   suite.
> - Removed 'force' parameter from _bh_disconnect; the disconnection
>   routine determines for itself if we are in the error pathway or not
>   instead now.  This removes any chance of duplicate calls to
>   _schedule_disconnect accidentally dropping the 'force' setting.
>
> Debugging/Testing changes:
>
> - Add debug: bool parameter to asyncio_run utility wrapper
> - Improve error messages for '@require' decorator
> - Add debugging message for state change events
> - Avoid flushing the StreamWriter if we don't have one (This
>   circumstance only arises in testing, but it's helpful.)
> - Improved __repr__ method for AsyncProtocol, and removed __str__
>   method.  enforcing eval(__repr__(x)) == x does not make sense for
>   AsyncProtocol.
> - Misc logging message changes
> - Add a suite of fancy Task debugging utilities.
> - Most tracebacks now log at the DEBUG level instead of
>   CRITICAL/ERROR/WARNING; In those error cases, a one-line summary is
>   logged instead.
>
> Misc. aesthetic changes:
>
> - Misc docstring fixes, whitespace, etc.
> - Reordered the definition of some methods to try and keep similar
>   methods near each other (Moved _cleanup near _bh_disconnect in
>   QMPClient.)
>
> ~ Shucks Howdy, Gee Golly!
>
> John Snow (24):
>   python/aqmp: add asynchronous QMP (AQMP) subpackage
>   python/aqmp: add error classes
>   python/pylint: Add exception for TypeVar names ('T')
>   python/aqmp: add asyncio compatibility wrappers
>   python/aqmp: add generic async message-based protocol support
>   python/aqmp: add runstate state machine to AsyncProtocol
>   python/aqmp: Add logging utility helpers
>   python/aqmp: add logging to AsyncProtocol
>   python/aqmp: add AsyncProtocol.accept() method
>   python/aqmp: add configurable read buffer limit
>   python/aqmp: add _cb_inbound and _cb_inbound logging hooks
>   python/aqmp: add AsyncProtocol._readline() method
>   python/aqmp: add QMP Message format
>   python/aqmp: add well-known QMP object models
>   python/aqmp: add QMP event support
>   python/pylint: disable too-many-function-args
>   python/aqmp: add QMP protocol support
>   python/pylint: disable no-member check
>   python/aqmp: Add message routing to QMP protocol
>   python/aqmp: add execute() interfaces
>   python/aqmp: add _raw() execution interface
>   python/aqmp: add asyncio_run compatibility wrapper
>   python/aqmp: add scary message
>   python/aqmp: add AsyncProtocol unit tests
>
>  python/qemu/aqmp/__init__.py   |  58 +++
>  python/qemu/aqmp/error.py      |  50 ++
>  python/qemu/aqmp/events.py     | 706 ++++++++++++++++++++++++++
>  python/qemu/aqmp/message.py    | 209 ++++++++
>  python/qemu/aqmp/models.py     | 133 +++++
>  python/qemu/aqmp/protocol.py   | 882 +++++++++++++++++++++++++++++++++
>  python/qemu/aqmp/py.typed      |   0
>  python/qemu/aqmp/qmp_client.py | 621 +++++++++++++++++++++++
>  python/qemu/aqmp/util.py       | 207 ++++++++
>  python/setup.cfg               |   5 +-
>  python/tests/null_proto.py     |  67 +++
>  python/tests/protocol.py       | 458 +++++++++++++++++
>  12 files changed, 3395 insertions(+), 1 deletion(-)
>  create mode 100644 python/qemu/aqmp/__init__.py
>  create mode 100644 python/qemu/aqmp/error.py
>  create mode 100644 python/qemu/aqmp/events.py
>  create mode 100644 python/qemu/aqmp/message.py
>  create mode 100644 python/qemu/aqmp/models.py
>  create mode 100644 python/qemu/aqmp/protocol.py
>  create mode 100644 python/qemu/aqmp/py.typed
>  create mode 100644 python/qemu/aqmp/qmp_client.py
>  create mode 100644 python/qemu/aqmp/util.py
>  create mode 100644 python/tests/null_proto.py
>  create mode 100644 python/tests/protocol.py
>
> --
> 2.31.1
>
>
>