python/qemu/qmp/legacy.py

   1 """
   2 (Legacy) Sync QMP Wrapper
   3
   4 This module provides the `QEMUMonitorProtocol` class, which is a
   5 synchronous wrapper around `QMPClient`.
   6
   7 Its design closely resembles that of the original QEMUMonitorProtocol
   8 class, originally written by Luiz Capitulino. It is provided here for
   9 compatibility with scripts inside the QEMU source tree that expect the
  10 old interface.
  11 """
  12
  13 #
  14 # Copyright (C) 2009-2022 Red Hat Inc.
  15 #
  16 # Authors:
  17 #  Luiz Capitulino <[email protected]>
  18 #  John Snow <[email protected]>
  19 #
  20 # This work is licensed under the terms of the GNU GPL, version 2.  See
  21 # the COPYING file in the top-level directory.
  22 #
  23
  24 import asyncio
  25 from types import TracebackType
  26 from typing import (
  27     Any,
  28     Awaitable,
  29     Dict,
  30     List,
  31     Optional,
  32     Type,
  33     TypeVar,
  34     Union,
  35 )
  36
  37 from .error import QMPError
  38 from .protocol import Runstate, SocketAddrT
  39 from .qmp_client import QMPClient
  40
  41
  42 #: QMPMessage is an entire QMP message of any kind.
  43 QMPMessage = Dict[str, Any]
  44
  45 #: QMPReturnValue is the 'return' value of a command.
  46 QMPReturnValue = object
  47
  48 #: QMPObject is any object in a QMP message.
  49 QMPObject = Dict[str, object]
  50
  51 # QMPMessage can be outgoing commands or incoming events/returns.
  52 # QMPReturnValue is usually a dict/json object, but due to QAPI's
  53 # 'returns-whitelist', it can actually be anything.
  54 #
  55 # {'return': {}} is a QMPMessage,
  56 # {} is the QMPReturnValue.
  57
  58
  59 class QMPBadPortError(QMPError):
  60     """
  61     Unable to parse socket address: Port was non-numerical.
  62     """
  63
  64
  65 class QEMUMonitorProtocol:
  66     """
  67     Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP)
  68     and then allow to handle commands and events.
  69
  70     :param address:  QEMU address, can be either a unix socket path (string)
  71                      or a tuple in the form ( address, port ) for a TCP
  72                      connection
  73     :param server:   Act as the socket server. (See 'accept')
  74     :param nickname: Optional nickname used for logging.
  75     """
  76
  77     def __init__(self, address: SocketAddrT,
  78                  server: bool = False,
  79                  nickname: Optional[str] = None):
  80
  81         self._qmp = QMPClient(nickname)
  82         self._aloop = asyncio.get_event_loop()
  83         self._address = address
  84         self._timeout: Optional[float] = None
  85
  86         if server:
  87             self._sync(self._qmp.start_server(self._address))
  88
  89     _T = TypeVar('_T')
  90
  91     def _sync(
  92             self, future: Awaitable[_T], timeout: Optional[float] = None
  93     ) -> _T:
  94         return self._aloop.run_until_complete(
  95             asyncio.wait_for(future, timeout=timeout)
  96         )
  97
  98     def _get_greeting(self) -> Optional[QMPMessage]:
  99         if self._qmp.greeting is not None:
 100             # pylint: disable=protected-access
 101             return self._qmp.greeting._asdict()
 102         return None
 103
 104     def __enter__(self: _T) -> _T:
 105         # Implement context manager enter function.
 106         return self
 107
 108     def __exit__(self,
 109                  # pylint: disable=duplicate-code
 110                  # see https://github.com/PyCQA/pylint/issues/3619
 111                  exc_type: Optional[Type[BaseException]],
 112                  exc_val: Optional[BaseException],
 113                  exc_tb: Optional[TracebackType]) -> None:
 114         # Implement context manager exit function.
 115         self.close()
 116
 117     @classmethod
 118     def parse_address(cls, address: str) -> SocketAddrT:
 119         """
 120         Parse a string into a QMP address.
 121
 122         Figure out if the argument is in the port:host form.
 123         If it's not, it's probably a file path.
 124         """
 125         components = address.split(':')
 126         if len(components) == 2:
 127             try:
 128                 port = int(components[1])
 129             except ValueError:
 130                 msg = f"Bad port: '{components[1]}' in '{address}'."
 131                 raise QMPBadPortError(msg) from None
 132             return (components[0], port)
 133
 134         # Treat as filepath.
 135         return address
 136
 137     def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
 138         """
 139         Connect to the QMP Monitor and perform capabilities negotiation.
 140
 141         :return: QMP greeting dict, or None if negotiate is false
 142         :raise ConnectError: on connection errors
 143         """
 144         self._qmp.await_greeting = negotiate
 145         self._qmp.negotiate = negotiate
 146
 147         self._sync(
 148             self._qmp.connect(self._address)
 149         )
 150         return self._get_greeting()
 151
 152     def accept(self, timeout: Optional[float] = 15.0) -> QMPMessage:
 153         """
 154         Await connection from QMP Monitor and perform capabilities negotiation.
 155
 156         :param timeout:
 157             timeout in seconds (nonnegative float number, or None).
 158             If None, there is no timeout, and this may block forever.
 159
 160         :return: QMP greeting dict
 161         :raise ConnectError: on connection errors
 162         """
 163         self._qmp.await_greeting = True
 164         self._qmp.negotiate = True
 165
 166         self._sync(self._qmp.accept(), timeout)
 167
 168         ret = self._get_greeting()
 169         assert ret is not None
 170         return ret
 171
 172     def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage:
 173         """
 174         Send a QMP command to the QMP Monitor.
 175
 176         :param qmp_cmd: QMP command to be sent as a Python dict
 177         :return: QMP response as a Python dict
 178         """
 179         return dict(
 180             self._sync(
 181                 # pylint: disable=protected-access
 182
 183                 # _raw() isn't a public API, because turning off
 184                 # automatic ID assignment is discouraged. For
 185                 # compatibility with iotests *only*, do it anyway.
 186                 self._qmp._raw(qmp_cmd, assign_id=False),
 187                 self._timeout
 188             )
 189         )
 190
 191     def cmd(self, name: str,
 192             args: Optional[Dict[str, object]] = None,
 193             cmd_id: Optional[object] = None) -> QMPMessage:
 194         """
 195         Build a QMP command and send it to the QMP Monitor.
 196
 197         :param name: command name (string)
 198         :param args: command arguments (dict)
 199         :param cmd_id: command id (dict, list, string or int)
 200         """
 201         qmp_cmd: QMPMessage = {'execute': name}
 202         if args:
 203             qmp_cmd['arguments'] = args
 204         if cmd_id:
 205             qmp_cmd['id'] = cmd_id
 206         return self.cmd_obj(qmp_cmd)
 207
 208     def command(self, cmd: str, **kwds: object) -> QMPReturnValue:
 209         """
 210         Build and send a QMP command to the monitor, report errors if any
 211         """
 212         return self._sync(
 213             self._qmp.execute(cmd, kwds),
 214             self._timeout
 215         )
 216
 217     def pull_event(self,
 218                    wait: Union[bool, float] = False) -> Optional[QMPMessage]:
 219         """
 220         Pulls a single event.
 221
 222         :param wait:
 223             If False or 0, do not wait. Return None if no events ready.
 224             If True, wait forever until the next event.
 225             Otherwise, wait for the specified number of seconds.
 226
 227         :raise asyncio.TimeoutError:
 228             When a timeout is requested and the timeout period elapses.
 229
 230         :return: The first available QMP event, or None.
 231         """
 232         if not wait:
 233             # wait is False/0: "do not wait, do not except."
 234             if self._qmp.events.empty():
 235                 return None
 236
 237         # If wait is 'True', wait forever. If wait is False/0, the events
 238         # queue must not be empty; but it still needs some real amount
 239         # of time to complete.
 240         timeout = None
 241         if wait and isinstance(wait, float):
 242             timeout = wait
 243
 244         return dict(
 245             self._sync(
 246                 self._qmp.events.get(),
 247                 timeout
 248             )
 249         )
 250
 251     def get_events(self, wait: Union[bool, float] = False) -> List[QMPMessage]:
 252         """
 253         Get a list of QMP events and clear all pending events.
 254
 255         :param wait:
 256             If False or 0, do not wait. Return None if no events ready.
 257             If True, wait until we have at least one event.
 258             Otherwise, wait for up to the specified number of seconds for at
 259             least one event.
 260
 261         :raise asyncio.TimeoutError:
 262             When a timeout is requested and the timeout period elapses.
 263
 264         :return: A list of QMP events.
 265         """
 266         events = [dict(x) for x in self._qmp.events.clear()]
 267         if events:
 268             return events
 269
 270         event = self.pull_event(wait)
 271         return [event] if event is not None else []
 272
 273     def clear_events(self) -> None:
 274         """Clear current list of pending events."""
 275         self._qmp.events.clear()
 276
 277     def close(self) -> None:
 278         """Close the connection."""
 279         self._sync(
 280             self._qmp.disconnect()
 281         )
 282
 283     def settimeout(self, timeout: Optional[float]) -> None:
 284         """
 285         Set the timeout for QMP RPC execution.
 286
 287         This timeout affects the `cmd`, `cmd_obj`, and `command` methods.
 288         The `accept`, `pull_event` and `get_event` methods have their
 289         own configurable timeouts.
 290
 291         :param timeout:
 292             timeout in seconds, or None.
 293             None will wait indefinitely.
 294         """
 295         self._timeout = timeout
 296
 297     def send_fd_scm(self, fd: int) -> None:
 298         """
 299         Send a file descriptor to the remote via SCM_RIGHTS.
 300         """
 301         self._qmp.send_fd_scm(fd)
 302
 303     def __del__(self) -> None:
 304         if self._qmp.runstate == Runstate.IDLE:
 305             return
 306
 307         if not self._aloop.is_running():
 308             self.close()
 309         else:
 310             # Garbage collection ran while the event loop was running.
 311             # Nothing we can do about it now, but if we don't raise our
 312             # own error, the user will be treated to a lot of traceback
 313             # they might not understand.
 314             raise QMPError(
 315                 "QEMUMonitorProtocol.close()"
 316                 " was not called before object was garbage collected"
 317             )