python/qemu/machine/machine.py

   1 """
   2 QEMU machine module:
   3
   4 The machine module primarily provides the QEMUMachine class,
   5 which provides facilities for managing the lifetime of a QEMU VM.
   6 """
   7
   8 # Copyright (C) 2015-2016 Red Hat Inc.
   9 # Copyright (C) 2012 IBM Corp.
  10 #
  11 # Authors:
  12 #  Fam Zheng <famz@redhat.com>
  13 #
  14 # This work is licensed under the terms of the GNU GPL, version 2.  See
  15 # the COPYING file in the top-level directory.
  16 #
  17 # Based on qmp.py.
  18 #
  19
  20 import errno
  21 from itertools import chain
  22 import locale
  23 import logging
  24 import os
  25 import shutil
  26 import signal
  27 import socket
  28 import subprocess
  29 import tempfile
  30 from types import TracebackType
  31 from typing import (
  32     Any,
  33     BinaryIO,
  34     Dict,
  35     List,
  36     Optional,
  37     Sequence,
  38     Tuple,
  39     Type,
  40     TypeVar,
  41 )
  42
  43 from qemu.qmp import SocketAddrT
  44 from qemu.qmp.legacy import (
  45     QEMUMonitorProtocol,
  46     QMPMessage,
  47     QMPReturnValue,
  48 )
  49
  50 from . import console_socket
  51
  52
  53 LOG = logging.getLogger(__name__)
  54
  55
  56 class QEMUMachineError(Exception):
  57     """
  58     Exception called when an error in QEMUMachine happens.
  59     """
  60
  61
  62 class QEMUMachineAddDeviceError(QEMUMachineError):
  63     """
  64     Exception raised when a request to add a device can not be fulfilled
  65
  66     The failures are caused by limitations, lack of information or conflicting
  67     requests on the QEMUMachine methods.  This exception does not represent
  68     failures reported by the QEMU binary itself.
  69     """
  70
  71
  72 class VMLaunchFailure(QEMUMachineError):
  73     """
  74     Exception raised when a VM launch was attempted, but failed.
  75     """
  76     def __init__(self, exitcode: Optional[int],
  77                  command: str, output: Optional[str]):
  78         super().__init__(exitcode, command, output)
  79         self.exitcode = exitcode
  80         self.command = command
  81         self.output = output
  82
  83     def __str__(self) -> str:
  84         ret = ''
  85         if self.__cause__ is not None:
  86             name = type(self.__cause__).__name__
  87             reason = str(self.__cause__)
  88             if reason:
  89                 ret += f"{name}: {reason}"
  90             else:
  91                 ret += f"{name}"
  92         ret += '\n'
  93
  94         if self.exitcode is not None:
  95             ret += f"\tExit code: {self.exitcode}\n"
  96         ret += f"\tCommand: {self.command}\n"
  97         ret += f"\tOutput: {self.output}\n"
  98         return ret
  99
 100
 101 class AbnormalShutdown(QEMUMachineError):
 102     """
 103     Exception raised when a graceful shutdown was requested, but not performed.
 104     """
 105
 106
 107 _T = TypeVar('_T', bound='QEMUMachine')
 108
 109
 110 class QEMUMachine:
 111     """
 112     A QEMU VM.
 113
 114     Use this object as a context manager to ensure
 115     the QEMU process terminates::
 116
 117         with VM(binary) as vm:
 118             ...
 119         # vm is guaranteed to be shut down here
 120     """
 121     # pylint: disable=too-many-instance-attributes, too-many-public-methods
 122
 123     def __init__(self,
 124                  binary: str,
 125                  args: Sequence[str] = (),
 126                  wrapper: Sequence[str] = (),
 127                  name: Optional[str] = None,
 128                  base_temp_dir: str = "/var/tmp",
 129                  monitor_address: Optional[SocketAddrT] = None,
 130                  drain_console: bool = False,
 131                  console_log: Optional[str] = None,
 132                  log_dir: Optional[str] = None,
 133                  qmp_timer: Optional[float] = 30):
 134         '''
 135         Initialize a QEMUMachine
 136
 137         @param binary: path to the qemu binary
 138         @param args: list of extra arguments
 139         @param wrapper: list of arguments used as prefix to qemu binary
 140         @param name: prefix for socket and log file names (default: qemu-PID)
 141         @param base_temp_dir: default location where temp files are created
 142         @param monitor_address: address for QMP monitor
 143         @param drain_console: (optional) True to drain console socket to buffer
 144         @param console_log: (optional) path to console log file
 145         @param log_dir: where to create and keep log files
 146         @param qmp_timer: (optional) default QMP socket timeout
 147         @note: Qemu process is not started until launch() is used.
 148         '''
 149         # pylint: disable=too-many-arguments
 150
 151         # Direct user configuration
 152
 153         self._binary = binary
 154         self._args = list(args)
 155         self._wrapper = wrapper
 156         self._qmp_timer = qmp_timer
 157
 158         self._name = name or f"{id(self):x}"
 159         self._sock_pair: Optional[Tuple[socket.socket, socket.socket]] = None
 160         self._cons_sock_pair: Optional[
 161             Tuple[socket.socket, socket.socket]] = None
 162         self._temp_dir: Optional[str] = None
 163         self._base_temp_dir = base_temp_dir
 164         self._log_dir = log_dir
 165
 166         self._monitor_address = monitor_address
 167
 168         self._console_log_path = console_log
 169         if self._console_log_path:
 170             # In order to log the console, buffering needs to be enabled.
 171             self._drain_console = True
 172         else:
 173             self._drain_console = drain_console
 174
 175         # Runstate
 176         self._qemu_log_path: Optional[str] = None
 177         self._qemu_log_file: Optional[BinaryIO] = None
 178         self._popen: Optional['subprocess.Popen[bytes]'] = None
 179         self._events: List[QMPMessage] = []
 180         self._iolog: Optional[str] = None
 181         self._qmp_set = True   # Enable QMP monitor by default.
 182         self._qmp_connection: Optional[QEMUMonitorProtocol] = None
 183         self._qemu_full_args: Tuple[str, ...] = ()
 184         self._launched = False
 185         self._machine: Optional[str] = None
 186         self._console_index = 0
 187         self._console_set = False
 188         self._console_device_type: Optional[str] = None
 189         self._console_socket: Optional[socket.socket] = None
 190         self._console_file: Optional[socket.SocketIO] = None
 191         self._remove_files: List[str] = []
 192         self._user_killed = False
 193         self._quit_issued = False
 194
 195     def __enter__(self: _T) -> _T:
 196         return self
 197
 198     def __exit__(self,
 199                  exc_type: Optional[Type[BaseException]],
 200                  exc_val: Optional[BaseException],
 201                  exc_tb: Optional[TracebackType]) -> None:
 202         self.shutdown()
 203
 204     def add_monitor_null(self) -> None:
 205         """
 206         This can be used to add an unused monitor instance.
 207         """
 208         self._args.append('-monitor')
 209         self._args.append('null')
 210
 211     def add_fd(self: _T, fd: int, fdset: int,
 212                opaque: str, opts: str = '') -> _T:
 213         """
 214         Pass a file descriptor to the VM
 215         """
 216         options = ['fd=%d' % fd,
 217                    'set=%d' % fdset,
 218                    'opaque=%s' % opaque]
 219         if opts:
 220             options.append(opts)
 221
 222         # This did not exist before 3.4, but since then it is
 223         # mandatory for our purpose
 224         if hasattr(os, 'set_inheritable'):
 225             os.set_inheritable(fd, True)
 226
 227         self._args.append('-add-fd')
 228         self._args.append(','.join(options))
 229         return self
 230
 231     def send_fd_scm(self, fd: Optional[int] = None,
 232                     file_path: Optional[str] = None) -> int:
 233         """
 234         Send an fd or file_path to the remote via SCM_RIGHTS.
 235
 236         Exactly one of fd and file_path must be given.  If it is
 237         file_path, the file will be opened read-only and the new file
 238         descriptor will be sent to the remote.
 239         """
 240         if file_path is not None:
 241             assert fd is None
 242             with open(file_path, "rb") as passfile:
 243                 fd = passfile.fileno()
 244                 self._qmp.send_fd_scm(fd)
 245         else:
 246             assert fd is not None
 247             self._qmp.send_fd_scm(fd)
 248
 249         return 0
 250
 251     @staticmethod
 252     def _remove_if_exists(path: str) -> None:
 253         """
 254         Remove file object at path if it exists
 255         """
 256         try:
 257             os.remove(path)
 258         except OSError as exception:
 259             if exception.errno == errno.ENOENT:
 260                 return
 261             raise
 262
 263     def is_running(self) -> bool:
 264         """Returns true if the VM is running."""
 265         return self._popen is not None and self._popen.poll() is None
 266
 267     @property
 268     def _subp(self) -> 'subprocess.Popen[bytes]':
 269         if self._popen is None:
 270             raise QEMUMachineError('Subprocess pipe not present')
 271         return self._popen
 272
 273     def exitcode(self) -> Optional[int]:
 274         """Returns the exit code if possible, or None."""
 275         if self._popen is None:
 276             return None
 277         return self._popen.poll()
 278
 279     def get_pid(self) -> Optional[int]:
 280         """Returns the PID of the running process, or None."""
 281         if not self.is_running():
 282             return None
 283         return self._subp.pid
 284
 285     def _load_io_log(self) -> None:
 286         # Assume that the output encoding of QEMU's terminal output is
 287         # defined by our locale. If indeterminate, allow open() to fall
 288         # back to the platform default.
 289         _, encoding = locale.getlocale()
 290         if self._qemu_log_path is not None:
 291             with open(self._qemu_log_path, "r", encoding=encoding) as iolog:
 292                 self._iolog = iolog.read()
 293
 294     @property
 295     def _base_args(self) -> List[str]:
 296         args = ['-display', 'none', '-vga', 'none']
 297
 298         if self._qmp_set:
 299             if self._sock_pair:
 300                 moncdev = f"socket,id=mon,fd={self._sock_pair[0].fileno()}"
 301             elif isinstance(self._monitor_address, tuple):
 302                 moncdev = "socket,id=mon,host={},port={}".format(
 303                     *self._monitor_address
 304                 )
 305             else:
 306                 moncdev = f"socket,id=mon,path={self._monitor_address}"
 307             args.extend(['-chardev', moncdev, '-mon',
 308                          'chardev=mon,mode=control'])
 309
 310         if self._machine is not None:
 311             args.extend(['-machine', self._machine])
 312         for _ in range(self._console_index):
 313             args.extend(['-serial', 'null'])
 314         if self._console_set:
 315             assert self._cons_sock_pair is not None
 316             fd = self._cons_sock_pair[0].fileno()
 317             chardev = f"socket,id=console,fd={fd}"
 318             args.extend(['-chardev', chardev])
 319             if self._console_device_type is None:
 320                 args.extend(['-serial', 'chardev:console'])
 321             else:
 322                 device = '%s,chardev=console' % self._console_device_type
 323                 args.extend(['-device', device])
 324         return args
 325
 326     @property
 327     def args(self) -> List[str]:
 328         """Returns the list of arguments given to the QEMU binary."""
 329         return self._args
 330
 331     @property
 332     def binary(self) -> str:
 333         """Returns path to the QEMU binary"""
 334         return self._binary
 335
 336     def _pre_launch(self) -> None:
 337         if self._qmp_set:
 338             sock = None
 339             if self._monitor_address is None:
 340                 self._sock_pair = socket.socketpair()
 341                 os.set_inheritable(self._sock_pair[0].fileno(), True)
 342                 sock = self._sock_pair[1]
 343             if isinstance(self._monitor_address, str):
 344                 self._remove_files.append(self._monitor_address)
 345
 346             sock_or_addr = self._monitor_address or sock
 347             assert sock_or_addr is not None
 348
 349             self._qmp_connection = QEMUMonitorProtocol(
 350                 sock_or_addr,
 351                 server=bool(self._monitor_address),
 352                 nickname=self._name
 353             )
 354
 355         if self._console_set:
 356             self._cons_sock_pair = socket.socketpair()
 357             os.set_inheritable(self._cons_sock_pair[0].fileno(), True)
 358
 359         # NOTE: Make sure any opened resources are *definitely* freed in
 360         # _post_shutdown()!
 361         # pylint: disable=consider-using-with
 362         self._qemu_log_path = os.path.join(self.log_dir, self._name + ".log")
 363         self._qemu_log_file = open(self._qemu_log_path, 'wb')
 364
 365         self._iolog = None
 366         self._qemu_full_args = tuple(chain(
 367             self._wrapper,
 368             [self._binary],
 369             self._base_args,
 370             self._args
 371         ))
 372
 373     def _post_launch(self) -> None:
 374         if self._sock_pair:
 375             self._sock_pair[0].close()
 376         if self._cons_sock_pair:
 377             self._cons_sock_pair[0].close()
 378
 379         if self._qmp_connection:
 380             if self._sock_pair:
 381                 self._qmp.connect()
 382             else:
 383                 self._qmp.accept(self._qmp_timer)
 384
 385     def _close_qemu_log_file(self) -> None:
 386         if self._qemu_log_file is not None:
 387             self._qemu_log_file.close()
 388             self._qemu_log_file = None
 389
 390     def _post_shutdown(self) -> None:
 391         """
 392         Called to cleanup the VM instance after the process has exited.
 393         May also be called after a failed launch.
 394         """
 395         LOG.debug("Cleaning up after VM process")
 396         try:
 397             self._close_qmp_connection()
 398         except Exception as err:  # pylint: disable=broad-except
 399             LOG.warning(
 400                 "Exception closing QMP connection: %s",
 401                 str(err) if str(err) else type(err).__name__
 402             )
 403         finally:
 404             assert self._qmp_connection is None
 405
 406         if self._sock_pair:
 407             self._sock_pair[0].close()
 408             self._sock_pair[1].close()
 409             self._sock_pair = None
 410
 411         self._close_qemu_log_file()
 412
 413         self._load_io_log()
 414
 415         self._qemu_log_path = None
 416
 417         if self._temp_dir is not None:
 418             shutil.rmtree(self._temp_dir)
 419             self._temp_dir = None
 420
 421         while len(self._remove_files) > 0:
 422             self._remove_if_exists(self._remove_files.pop())
 423
 424         exitcode = self.exitcode()
 425         if (exitcode is not None and exitcode < 0
 426                 and not (self._user_killed and exitcode == -signal.SIGKILL)):
 427             msg = 'qemu received signal %i; command: "%s"'
 428             if self._qemu_full_args:
 429                 command = ' '.join(self._qemu_full_args)
 430             else:
 431                 command = ''
 432             LOG.warning(msg, -int(exitcode), command)
 433
 434         self._quit_issued = False
 435         self._user_killed = False
 436         self._launched = False
 437
 438     def launch(self) -> None:
 439         """
 440         Launch the VM and make sure we cleanup and expose the
 441         command line/output in case of exception
 442         """
 443
 444         if self._launched:
 445             raise QEMUMachineError('VM already launched')
 446
 447         try:
 448             self._launch()
 449         except BaseException as exc:
 450             # We may have launched the process but it may
 451             # have exited before we could connect via QMP.
 452             # Assume the VM didn't launch or is exiting.
 453             # If we don't wait for the process, exitcode() may still be
 454             # 'None' by the time control is ceded back to the caller.
 455             if self._launched:
 456                 self.wait()
 457             else:
 458                 self._post_shutdown()
 459
 460             if isinstance(exc, Exception):
 461                 raise VMLaunchFailure(
 462                     exitcode=self.exitcode(),
 463                     command=' '.join(self._qemu_full_args),
 464                     output=self._iolog
 465                 ) from exc
 466
 467             # Don't wrap 'BaseException'; doing so would downgrade
 468             # that exception. However, we still want to clean up.
 469             raise
 470
 471     def _launch(self) -> None:
 472         """
 473         Launch the VM and establish a QMP connection
 474         """
 475         self._pre_launch()
 476         LOG.debug('VM launch command: %r', ' '.join(self._qemu_full_args))
 477
 478         # Cleaning up of this subprocess is guaranteed by _do_shutdown.
 479         # pylint: disable=consider-using-with
 480         self._popen = subprocess.Popen(self._qemu_full_args,
 481                                        stdin=subprocess.DEVNULL,
 482                                        stdout=self._qemu_log_file,
 483                                        stderr=subprocess.STDOUT,
 484                                        shell=False,
 485                                        close_fds=False)
 486         self._launched = True
 487         self._post_launch()
 488
 489     def _close_qmp_connection(self) -> None:
 490         """
 491         Close the underlying QMP connection, if any.
 492
 493         Dutifully report errors that occurred while closing, but assume
 494         that any error encountered indicates an abnormal termination
 495         process and not a failure to close.
 496         """
 497         if self._qmp_connection is None:
 498             return
 499
 500         try:
 501             self._qmp.close()
 502         except EOFError:
 503             # EOF can occur as an Exception here when using the Async
 504             # QMP backend. It indicates that the server closed the
 505             # stream. If we successfully issued 'quit' at any point,
 506             # then this was expected. If the remote went away without
 507             # our permission, it's worth reporting that as an abnormal
 508             # shutdown case.
 509             if not (self._user_killed or self._quit_issued):
 510                 raise
 511         finally:
 512             self._qmp_connection = None
 513
 514     def _early_cleanup(self) -> None:
 515         """
 516         Perform any cleanup that needs to happen before the VM exits.
 517
 518         This method may be called twice upon shutdown, once each by soft
 519         and hard shutdown in failover scenarios.
 520         """
 521         # If we keep the console socket open, we may deadlock waiting
 522         # for QEMU to exit, while QEMU is waiting for the socket to
 523         # become writable.
 524         if self._console_file is not None:
 525             LOG.debug("Closing console file")
 526             self._console_file.close()
 527             self._console_file = None
 528
 529         if self._console_socket is not None:
 530             LOG.debug("Closing console socket")
 531             self._console_socket.close()
 532             self._console_socket = None
 533
 534         if self._cons_sock_pair:
 535             self._cons_sock_pair[0].close()
 536             self._cons_sock_pair[1].close()
 537             self._cons_sock_pair = None
 538
 539     def _hard_shutdown(self) -> None:
 540         """
 541         Perform early cleanup, kill the VM, and wait for it to terminate.
 542
 543         :raise subprocess.Timeout: When timeout is exceeds 60 seconds
 544             waiting for the QEMU process to terminate.
 545         """
 546         LOG.debug("Performing hard shutdown")
 547         self._early_cleanup()
 548         self._subp.kill()
 549         self._subp.wait(timeout=60)
 550
 551     def _soft_shutdown(self, timeout: Optional[int]) -> None:
 552         """
 553         Perform early cleanup, attempt to gracefully shut down the VM, and wait
 554         for it to terminate.
 555
 556         :param timeout: Timeout in seconds for graceful shutdown.
 557                         A value of None is an infinite wait.
 558
 559         :raise ConnectionReset: On QMP communication errors
 560         :raise subprocess.TimeoutExpired: When timeout is exceeded waiting for
 561             the QEMU process to terminate.
 562         """
 563         LOG.debug("Attempting graceful termination")
 564
 565         self._early_cleanup()
 566
 567         if self._quit_issued:
 568             LOG.debug(
 569                 "Anticipating QEMU termination due to prior 'quit' command, "
 570                 "or explicit call to wait()"
 571             )
 572         else:
 573             LOG.debug("Politely asking QEMU to terminate")
 574
 575         if self._qmp_connection:
 576             try:
 577                 if not self._quit_issued:
 578                     # May raise ExecInterruptedError or StateError if the
 579                     # connection dies or has *already* died.
 580                     self.qmp('quit')
 581             finally:
 582                 # Regardless, we want to quiesce the connection.
 583                 self._close_qmp_connection()
 584         elif not self._quit_issued:
 585             LOG.debug(
 586                 "Not anticipating QEMU quit and no QMP connection present, "
 587                 "issuing SIGTERM"
 588             )
 589             self._subp.terminate()
 590
 591         # May raise subprocess.TimeoutExpired
 592         LOG.debug(
 593             "Waiting (timeout=%s) for QEMU process (pid=%s) to terminate",
 594             timeout, self._subp.pid
 595         )
 596         self._subp.wait(timeout=timeout)
 597
 598     def _do_shutdown(self, timeout: Optional[int]) -> None:
 599         """
 600         Attempt to shutdown the VM gracefully; fallback to a hard shutdown.
 601
 602         :param timeout: Timeout in seconds for graceful shutdown.
 603                         A value of None is an infinite wait.
 604
 605         :raise AbnormalShutdown: When the VM could not be shut down gracefully.
 606             The inner exception will likely be ConnectionReset or
 607             subprocess.TimeoutExpired. In rare cases, non-graceful termination
 608             may result in its own exceptions, likely subprocess.TimeoutExpired.
 609         """
 610         try:
 611             self._soft_shutdown(timeout)
 612         except Exception as exc:
 613             if isinstance(exc, subprocess.TimeoutExpired):
 614                 LOG.debug("Timed out waiting for QEMU process to exit")
 615             LOG.debug("Graceful shutdown failed", exc_info=True)
 616             LOG.debug("Falling back to hard shutdown")
 617             self._hard_shutdown()
 618             raise AbnormalShutdown("Could not perform graceful shutdown") \
 619                 from exc
 620
 621     def shutdown(self,
 622                  hard: bool = False,
 623                  timeout: Optional[int] = 30) -> None:
 624         """
 625         Terminate the VM (gracefully if possible) and perform cleanup.
 626         Cleanup will always be performed.
 627
 628         If the VM has not yet been launched, or shutdown(), wait(), or kill()
 629         have already been called, this method does nothing.
 630
 631         :param hard: When true, do not attempt graceful shutdown, and
 632                      suppress the SIGKILL warning log message.
 633         :param timeout: Optional timeout in seconds for graceful shutdown.
 634                         Default 30 seconds, A `None` value is an infinite wait.
 635         """
 636         if not self._launched:
 637             return
 638
 639         LOG.debug("Shutting down VM appliance; timeout=%s", timeout)
 640         if hard:
 641             LOG.debug("Caller requests immediate termination of QEMU process.")
 642
 643         try:
 644             if hard:
 645                 self._user_killed = True
 646                 self._hard_shutdown()
 647             else:
 648                 self._do_shutdown(timeout)
 649         finally:
 650             self._post_shutdown()
 651
 652     def kill(self) -> None:
 653         """
 654         Terminate the VM forcefully, wait for it to exit, and perform cleanup.
 655         """
 656         self.shutdown(hard=True)
 657
 658     def wait(self, timeout: Optional[int] = 30) -> None:
 659         """
 660         Wait for the VM to power off and perform post-shutdown cleanup.
 661
 662         :param timeout: Optional timeout in seconds. Default 30 seconds.
 663                         A value of `None` is an infinite wait.
 664         """
 665         self._quit_issued = True
 666         self.shutdown(timeout=timeout)
 667
 668     def set_qmp_monitor(self, enabled: bool = True) -> None:
 669         """
 670         Set the QMP monitor.
 671
 672         @param enabled: if False, qmp monitor options will be removed from
 673                         the base arguments of the resulting QEMU command
 674                         line. Default is True.
 675
 676         .. note:: Call this function before launch().
 677         """
 678         self._qmp_set = enabled
 679
 680     @property
 681     def _qmp(self) -> QEMUMonitorProtocol:
 682         if self._qmp_connection is None:
 683             raise QEMUMachineError("Attempt to access QMP with no connection")
 684         return self._qmp_connection
 685
 686     @classmethod
 687     def _qmp_args(cls, conv_keys: bool,
 688                   args: Dict[str, Any]) -> Dict[str, object]:
 689         if conv_keys:
 690             return {k.replace('_', '-'): v for k, v in args.items()}
 691
 692         return args
 693
 694     def qmp(self, cmd: str,
 695             args_dict: Optional[Dict[str, object]] = None,
 696             conv_keys: Optional[bool] = None,
 697             **args: Any) -> QMPMessage:
 698         """
 699         Invoke a QMP command and return the response dict
 700         """
 701         if args_dict is not None:
 702             assert not args
 703             assert conv_keys is None
 704             args = args_dict
 705             conv_keys = False
 706
 707         if conv_keys is None:
 708             conv_keys = True
 709
 710         qmp_args = self._qmp_args(conv_keys, args)
 711         ret = self._qmp.cmd_raw(cmd, args=qmp_args)
 712         if cmd == 'quit' and 'error' not in ret and 'return' in ret:
 713             self._quit_issued = True
 714         return ret
 715
 716     def cmd(self, cmd: str,
 717             args_dict: Optional[Dict[str, object]] = None,
 718             conv_keys: Optional[bool] = None,
 719             **args: Any) -> QMPReturnValue:
 720         """
 721         Invoke a QMP command.
 722         On success return the response dict.
 723         On failure raise an exception.
 724         """
 725         if args_dict is not None:
 726             assert not args
 727             assert conv_keys is None
 728             args = args_dict
 729             conv_keys = False
 730
 731         if conv_keys is None:
 732             conv_keys = True
 733
 734         qmp_args = self._qmp_args(conv_keys, args)
 735         ret = self._qmp.cmd(cmd, **qmp_args)
 736         if cmd == 'quit':
 737             self._quit_issued = True
 738         return ret
 739
 740     def get_qmp_event(self, wait: bool = False) -> Optional[QMPMessage]:
 741         """
 742         Poll for one queued QMP events and return it
 743         """
 744         if self._events:
 745             return self._events.pop(0)
 746         return self._qmp.pull_event(wait=wait)
 747
 748     def get_qmp_events(self, wait: bool = False) -> List[QMPMessage]:
 749         """
 750         Poll for queued QMP events and return a list of dicts
 751         """
 752         events = self._qmp.get_events(wait=wait)
 753         events.extend(self._events)
 754         del self._events[:]
 755         return events
 756
 757     @staticmethod
 758     def event_match(event: Any, match: Optional[Any]) -> bool:
 759         """
 760         Check if an event matches optional match criteria.
 761
 762         The match criteria takes the form of a matching subdict. The event is
 763         checked to be a superset of the subdict, recursively, with matching
 764         values whenever the subdict values are not None.
 765
 766         This has a limitation that you cannot explicitly check for None values.
 767
 768         Examples, with the subdict queries on the left:
 769          - None matches any object.
 770          - {"foo": None} matches {"foo": {"bar": 1}}
 771          - {"foo": None} matches {"foo": 5}
 772          - {"foo": {"abc": None}} does not match {"foo": {"bar": 1}}
 773          - {"foo": {"rab": 2}} matches {"foo": {"bar": 1, "rab": 2}}
 774         """
 775         if match is None:
 776             return True
 777
 778         try:
 779             for key in match:
 780                 if key in event:
 781                     if not QEMUMachine.event_match(event[key], match[key]):
 782                         return False
 783                 else:
 784                     return False
 785             return True
 786         except TypeError:
 787             # either match or event wasn't iterable (not a dict)
 788             return bool(match == event)
 789
 790     def event_wait(self, name: str,
 791                    timeout: float = 60.0,
 792                    match: Optional[QMPMessage] = None) -> Optional[QMPMessage]:
 793         """
 794         event_wait waits for and returns a named event from QMP with a timeout.
 795
 796         name: The event to wait for.
 797         timeout: QEMUMonitorProtocol.pull_event timeout parameter.
 798         match: Optional match criteria. See event_match for details.
 799         """
 800         return self.events_wait([(name, match)], timeout)
 801
 802     def events_wait(self,
 803                     events: Sequence[Tuple[str, Any]],
 804                     timeout: float = 60.0) -> Optional[QMPMessage]:
 805         """
 806         events_wait waits for and returns a single named event from QMP.
 807         In the case of multiple qualifying events, this function returns the
 808         first one.
 809
 810         :param events: A sequence of (name, match_criteria) tuples.
 811                        The match criteria are optional and may be None.
 812                        See event_match for details.
 813         :param timeout: Optional timeout, in seconds.
 814                         See QEMUMonitorProtocol.pull_event.
 815
 816         :raise asyncio.TimeoutError:
 817             If timeout was non-zero and no matching events were found.
 818
 819         :return: A QMP event matching the filter criteria.
 820                  If timeout was 0 and no event matched, None.
 821         """
 822         def _match(event: QMPMessage) -> bool:
 823             for name, match in events:
 824                 if event['event'] == name and self.event_match(event, match):
 825                     return True
 826             return False
 827
 828         event: Optional[QMPMessage]
 829
 830         # Search cached events
 831         for event in self._events:
 832             if _match(event):
 833                 self._events.remove(event)
 834                 return event
 835
 836         # Poll for new events
 837         while True:
 838             event = self._qmp.pull_event(wait=timeout)
 839             if event is None:
 840                 # NB: None is only returned when timeout is false-ish.
 841                 # Timeouts raise asyncio.TimeoutError instead!
 842                 break
 843             if _match(event):
 844                 return event
 845             self._events.append(event)
 846
 847         return None
 848
 849     def get_log(self) -> Optional[str]:
 850         """
 851         After self.shutdown or failed qemu execution, this returns the output
 852         of the qemu process.
 853         """
 854         return self._iolog
 855
 856     def add_args(self, *args: str) -> None:
 857         """
 858         Adds to the list of extra arguments to be given to the QEMU binary
 859         """
 860         self._args.extend(args)
 861
 862     def set_machine(self, machine_type: str) -> None:
 863         """
 864         Sets the machine type
 865
 866         If set, the machine type will be added to the base arguments
 867         of the resulting QEMU command line.
 868         """
 869         self._machine = machine_type
 870
 871     def set_console(self,
 872                     device_type: Optional[str] = None,
 873                     console_index: int = 0) -> None:
 874         """
 875         Sets the device type for a console device
 876
 877         If set, the console device and a backing character device will
 878         be added to the base arguments of the resulting QEMU command
 879         line.
 880
 881         This is a convenience method that will either use the provided
 882         device type, or default to a "-serial chardev:console" command
 883         line argument.
 884
 885         The actual setting of command line arguments will be be done at
 886         machine launch time, as it depends on the temporary directory
 887         to be created.
 888
 889         @param device_type: the device type, such as "isa-serial".  If
 890                             None is given (the default value) a "-serial
 891                             chardev:console" command line argument will
 892                             be used instead, resorting to the machine's
 893                             default device type.
 894         @param console_index: the index of the console device to use.
 895                               If not zero, the command line will create
 896                               'index - 1' consoles and connect them to
 897                               the 'null' backing character device.
 898         """
 899         self._console_set = True
 900         self._console_device_type = device_type
 901         self._console_index = console_index
 902
 903     @property
 904     def console_socket(self) -> socket.socket:
 905         """
 906         Returns a socket connected to the console
 907         """
 908         if self._console_socket is None:
 909             LOG.debug("Opening console socket")
 910             if not self._console_set:
 911                 raise QEMUMachineError(
 912                     "Attempt to access console socket with no connection")
 913             assert self._cons_sock_pair is not None
 914             # os.dup() is used here for sock_fd because otherwise we'd
 915             # have two rich python socket objects that would each try to
 916             # close the same underlying fd when either one gets garbage
 917             # collected.
 918             self._console_socket = console_socket.ConsoleSocket(
 919                 sock_fd=os.dup(self._cons_sock_pair[1].fileno()),
 920                 file=self._console_log_path,
 921                 drain=self._drain_console)
 922             self._cons_sock_pair[1].close()
 923         return self._console_socket
 924
 925     @property
 926     def console_file(self) -> socket.SocketIO:
 927         """
 928         Returns a file associated with the console socket
 929         """
 930         if self._console_file is None:
 931             LOG.debug("Opening console file")
 932             self._console_file = self.console_socket.makefile(mode='rb',
 933                                                               buffering=0,
 934                                                               encoding='utf-8')
 935         return self._console_file
 936
 937     @property
 938     def temp_dir(self) -> str:
 939         """
 940         Returns a temporary directory to be used for this machine
 941         """
 942         if self._temp_dir is None:
 943             self._temp_dir = tempfile.mkdtemp(prefix="qemu-machine-",
 944                                               dir=self._base_temp_dir)
 945         return self._temp_dir
 946
 947     @property
 948     def log_dir(self) -> str:
 949         """
 950         Returns a directory to be used for writing logs
 951         """
 952         if self._log_dir is None:
 953             return self.temp_dir
 954         return self._log_dir