]> Git Repo - qemu.git/blob - python/qemu/qmp/events.py
works with less than base ISA qemu-system-riscv32 -M virt -bios none -kernel output...
[qemu.git] / python / qemu / qmp / events.py
1 """
2 QMP Events and EventListeners
3
4 Asynchronous QMP uses `EventListener` objects to listen for events. An
5 `EventListener` is a FIFO event queue that can be pre-filtered to listen
6 for only specific events. Each `EventListener` instance receives its own
7 copy of events that it hears, so events may be consumed without fear or
8 worry for depriving other listeners of events they need to hear.
9
10
11 EventListener Tutorial
12 ----------------------
13
14 In all of the following examples, we assume that we have a `QMPClient`
15 instantiated named ``qmp`` that is already connected.
16
17
18 `listener()` context blocks with one name
19 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
20
21 The most basic usage is by using the `listener()` context manager to
22 construct them:
23
24 .. code:: python
25
26    with qmp.listener('STOP') as listener:
27        await qmp.execute('stop')
28        await listener.get()
29
30 The listener is active only for the duration of the ‘with’ block. This
31 instance listens only for ‘STOP’ events.
32
33
34 `listener()` context blocks with two or more names
35 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
36
37 Multiple events can be selected for by providing any ``Iterable[str]``:
38
39 .. code:: python
40
41    with qmp.listener(('STOP', 'RESUME')) as listener:
42        await qmp.execute('stop')
43        event = await listener.get()
44        assert event['event'] == 'STOP'
45
46        await qmp.execute('cont')
47        event = await listener.get()
48        assert event['event'] == 'RESUME'
49
50
51 `listener()` context blocks with no names
52 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
53
54 By omitting names entirely, you can listen to ALL events.
55
56 .. code:: python
57
58    with qmp.listener() as listener:
59        await qmp.execute('stop')
60        event = await listener.get()
61        assert event['event'] == 'STOP'
62
63 This isn’t a very good use case for this feature: In a non-trivial
64 running system, we may not know what event will arrive next. Grabbing
65 the top of a FIFO queue returning multiple kinds of events may be prone
66 to error.
67
68
69 Using async iterators to retrieve events
70 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
71
72 If you’d like to simply watch what events happen to arrive, you can use
73 the listener as an async iterator:
74
75 .. code:: python
76
77    with qmp.listener() as listener:
78        async for event in listener:
79            print(f"Event arrived: {event['event']}")
80
81 This is analogous to the following code:
82
83 .. code:: python
84
85    with qmp.listener() as listener:
86        while True:
87            event = listener.get()
88            print(f"Event arrived: {event['event']}")
89
90 This event stream will never end, so these blocks will never terminate.
91
92
93 Using asyncio.Task to concurrently retrieve events
94 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
95
96 Since a listener’s event stream will never terminate, it is not likely
97 useful to use that form in a script. For longer-running clients, we can
98 create event handlers by using `asyncio.Task` to create concurrent
99 coroutines:
100
101 .. code:: python
102
103    async def print_events(listener):
104        try:
105            async for event in listener:
106                print(f"Event arrived: {event['event']}")
107        except asyncio.CancelledError:
108            return
109
110    with qmp.listener() as listener:
111        task = asyncio.Task(print_events(listener))
112        await qmp.execute('stop')
113        await qmp.execute('cont')
114        task.cancel()
115        await task
116
117 However, there is no guarantee that these events will be received by the
118 time we leave this context block. Once the context block is exited, the
119 listener will cease to hear any new events, and becomes inert.
120
121 Be mindful of the timing: the above example will *probably*– but does
122 not *guarantee*– that both STOP/RESUMED events will be printed. The
123 example below outlines how to use listeners outside of a context block.
124
125
126 Using `register_listener()` and `remove_listener()`
127 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
128
129 To create a listener with a longer lifetime, beyond the scope of a
130 single block, create a listener and then call `register_listener()`:
131
132 .. code:: python
133
134    class MyClient:
135        def __init__(self, qmp):
136            self.qmp = qmp
137            self.listener = EventListener()
138
139        async def print_events(self):
140            try:
141                async for event in self.listener:
142                    print(f"Event arrived: {event['event']}")
143            except asyncio.CancelledError:
144                return
145
146        async def run(self):
147            self.task = asyncio.Task(self.print_events)
148            self.qmp.register_listener(self.listener)
149            await qmp.execute('stop')
150            await qmp.execute('cont')
151
152        async def stop(self):
153            self.task.cancel()
154            await self.task
155            self.qmp.remove_listener(self.listener)
156
157 The listener can be deactivated by using `remove_listener()`. When it is
158 removed, any possible pending events are cleared and it can be
159 re-registered at a later time.
160
161
162 Using the built-in all events listener
163 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
164
165 The `QMPClient` object creates its own default listener named
166 :py:obj:`~Events.events` that can be used for the same purpose without
167 having to create your own:
168
169 .. code:: python
170
171    async def print_events(listener):
172        try:
173            async for event in listener:
174                print(f"Event arrived: {event['event']}")
175        except asyncio.CancelledError:
176            return
177
178    task = asyncio.Task(print_events(qmp.events))
179
180    await qmp.execute('stop')
181    await qmp.execute('cont')
182
183    task.cancel()
184    await task
185
186
187 Using both .get() and async iterators
188 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
189
190 The async iterator and `get()` methods pull events from the same FIFO
191 queue. If you mix the usage of both, be aware: Events are emitted
192 precisely once per listener.
193
194 If multiple contexts try to pull events from the same listener instance,
195 events are still emitted only precisely once.
196
197 This restriction can be lifted by creating additional listeners.
198
199
200 Creating multiple listeners
201 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
202
203 Additional `EventListener` objects can be created at-will. Each one
204 receives its own copy of events, with separate FIFO event queues.
205
206 .. code:: python
207
208    my_listener = EventListener()
209    qmp.register_listener(my_listener)
210
211    await qmp.execute('stop')
212    copy1 = await my_listener.get()
213    copy2 = await qmp.events.get()
214
215    assert copy1 == copy2
216
217 In this example, we await an event from both a user-created
218 `EventListener` and the built-in events listener. Both receive the same
219 event.
220
221
222 Clearing listeners
223 ~~~~~~~~~~~~~~~~~~
224
225 `EventListener` objects can be cleared, clearing all events seen thus far:
226
227 .. code:: python
228
229    await qmp.execute('stop')
230    qmp.events.clear()
231    await qmp.execute('cont')
232    event = await qmp.events.get()
233    assert event['event'] == 'RESUME'
234
235 `EventListener` objects are FIFO queues. If events are not consumed,
236 they will remain in the queue until they are witnessed or discarded via
237 `clear()`. FIFO queues will be drained automatically upon leaving a
238 context block, or when calling `remove_listener()`.
239
240
241 Accessing listener history
242 ~~~~~~~~~~~~~~~~~~~~~~~~~~
243
244 `EventListener` objects record their history. Even after being cleared,
245 you can obtain a record of all events seen so far:
246
247 .. code:: python
248
249    await qmp.execute('stop')
250    await qmp.execute('cont')
251    qmp.events.clear()
252
253    assert len(qmp.events.history) == 2
254    assert qmp.events.history[0]['event'] == 'STOP'
255    assert qmp.events.history[1]['event'] == 'RESUME'
256
257 The history is updated immediately and does not require the event to be
258 witnessed first.
259
260
261 Using event filters
262 ~~~~~~~~~~~~~~~~~~~
263
264 `EventListener` objects can be given complex filtering criteria if names
265 are not sufficient:
266
267 .. code:: python
268
269    def job1_filter(event) -> bool:
270        event_data = event.get('data', {})
271        event_job_id = event_data.get('id')
272        return event_job_id == "job1"
273
274    with qmp.listener('JOB_STATUS_CHANGE', job1_filter) as listener:
275        await qmp.execute('blockdev-backup', arguments={'job-id': 'job1', ...})
276        async for event in listener:
277            if event['data']['status'] == 'concluded':
278                break
279
280 These filters might be most useful when parameterized. `EventListener`
281 objects expect a function that takes only a single argument (the raw
282 event, as a `Message`) and returns a bool; True if the event should be
283 accepted into the stream. You can create a function that adapts this
284 signature to accept configuration parameters:
285
286 .. code:: python
287
288    def job_filter(job_id: str) -> EventFilter:
289        def filter(event: Message) -> bool:
290            return event['data']['id'] == job_id
291        return filter
292
293    with qmp.listener('JOB_STATUS_CHANGE', job_filter('job2')) as listener:
294        await qmp.execute('blockdev-backup', arguments={'job-id': 'job2', ...})
295        async for event in listener:
296            if event['data']['status'] == 'concluded':
297                break
298
299
300 Activating an existing listener with `listen()`
301 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
302
303 Listeners with complex, long configurations can also be created manually
304 and activated temporarily by using `listen()` instead of `listener()`:
305
306 .. code:: python
307
308    listener = EventListener(('BLOCK_JOB_COMPLETED', 'BLOCK_JOB_CANCELLED',
309                              'BLOCK_JOB_ERROR', 'BLOCK_JOB_READY',
310                              'BLOCK_JOB_PENDING', 'JOB_STATUS_CHANGE'))
311
312    with qmp.listen(listener):
313        await qmp.execute('blockdev-backup', arguments={'job-id': 'job3', ...})
314        async for event in listener:
315            print(event)
316            if event['event'] == 'BLOCK_JOB_COMPLETED':
317                break
318
319 Any events that are not witnessed by the time the block is left will be
320 cleared from the queue; entering the block is an implicit
321 `register_listener()` and leaving the block is an implicit
322 `remove_listener()`.
323
324
325 Activating multiple existing listeners with `listen()`
326 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
327
328 While `listener()` is only capable of creating a single listener,
329 `listen()` is capable of activating multiple listeners simultaneously:
330
331 .. code:: python
332
333    def job_filter(job_id: str) -> EventFilter:
334        def filter(event: Message) -> bool:
335            return event['data']['id'] == job_id
336        return filter
337
338    jobA = EventListener('JOB_STATUS_CHANGE', job_filter('jobA'))
339    jobB = EventListener('JOB_STATUS_CHANGE', job_filter('jobB'))
340
341    with qmp.listen(jobA, jobB):
342        qmp.execute('blockdev-create', arguments={'job-id': 'jobA', ...})
343        qmp.execute('blockdev-create', arguments={'job-id': 'jobB', ...})
344
345        async for event in jobA.get():
346            if event['data']['status'] == 'concluded':
347                break
348        async for event in jobB.get():
349            if event['data']['status'] == 'concluded':
350                break
351
352
353 Extending the `EventListener` class
354 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
355
356 In the case that a more specialized `EventListener` is desired to
357 provide either more functionality or more compact syntax for specialized
358 cases, it can be extended.
359
360 One of the key methods to extend or override is
361 :py:meth:`~EventListener.accept()`. The default implementation checks an
362 incoming message for:
363
364 1. A qualifying name, if any :py:obj:`~EventListener.names` were
365    specified at initialization time
366 2. That :py:obj:`~EventListener.event_filter()` returns True.
367
368 This can be modified however you see fit to change the criteria for
369 inclusion in the stream.
370
371 For convenience, a ``JobListener`` class could be created that simply
372 bakes in configuration so it does not need to be repeated:
373
374 .. code:: python
375
376    class JobListener(EventListener):
377        def __init__(self, job_id: str):
378            super().__init__(('BLOCK_JOB_COMPLETED', 'BLOCK_JOB_CANCELLED',
379                              'BLOCK_JOB_ERROR', 'BLOCK_JOB_READY',
380                              'BLOCK_JOB_PENDING', 'JOB_STATUS_CHANGE'))
381            self.job_id = job_id
382
383        def accept(self, event) -> bool:
384            if not super().accept(event):
385                return False
386            if event['event'] in ('BLOCK_JOB_PENDING', 'JOB_STATUS_CHANGE'):
387                return event['data']['id'] == job_id
388            return event['data']['device'] == job_id
389
390 From here on out, you can conjure up a custom-purpose listener that
391 listens only for job-related events for a specific job-id easily:
392
393 .. code:: python
394
395    listener = JobListener('job4')
396    with qmp.listener(listener):
397        await qmp.execute('blockdev-backup', arguments={'job-id': 'job4', ...})
398        async for event in listener:
399            print(event)
400            if event['event'] == 'BLOCK_JOB_COMPLETED':
401                break
402
403
404 Experimental Interfaces & Design Issues
405 ---------------------------------------
406
407 These interfaces are not ones I am sure I will keep or otherwise modify
408 heavily.
409
410 qmp.listener()’s type signature
411 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
412
413 `listener()` does not return anything, because it was assumed the caller
414 already had a handle to the listener. However, for
415 ``qmp.listener(EventListener())`` forms, the caller will not have saved
416 a handle to the listener.
417
418 Because this function can accept *many* listeners, I found it hard to
419 accurately type in a way where it could be used in both “one” or “many”
420 forms conveniently and in a statically type-safe manner.
421
422 Ultimately, I removed the return altogether, but perhaps with more time
423 I can work out a way to re-add it.
424
425
426 API Reference
427 -------------
428
429 """
430
431 import asyncio
432 from contextlib import contextmanager
433 import logging
434 from typing import (
435     AsyncIterator,
436     Callable,
437     Iterable,
438     Iterator,
439     List,
440     Optional,
441     Set,
442     Tuple,
443     Union,
444 )
445
446 from .error import QMPError
447 from .message import Message
448
449
450 EventNames = Union[str, Iterable[str], None]
451 EventFilter = Callable[[Message], bool]
452
453
454 class ListenerError(QMPError):
455     """
456     Generic error class for `EventListener`-related problems.
457     """
458
459
460 class EventListener:
461     """
462     Selectively listens for events with runtime configurable filtering.
463
464     This class is designed to be directly usable for the most common cases,
465     but it can be extended to provide more rigorous control.
466
467     :param names:
468         One or more names of events to listen for.
469         When not provided, listen for ALL events.
470     :param event_filter:
471         An optional event filtering function.
472         When names are also provided, this acts as a secondary filter.
473
474     When ``names`` and ``event_filter`` are both provided, the names
475     will be filtered first, and then the filter function will be called
476     second. The event filter function can assume that the format of the
477     event is a known format.
478     """
479     def __init__(
480         self,
481         names: EventNames = None,
482         event_filter: Optional[EventFilter] = None,
483     ):
484         # Queue of 'heard' events yet to be witnessed by a caller.
485         self._queue: 'asyncio.Queue[Message]' = asyncio.Queue()
486
487         # Intended as a historical record, NOT a processing queue or backlog.
488         self._history: List[Message] = []
489
490         #: Primary event filter, based on one or more event names.
491         self.names: Set[str] = set()
492         if isinstance(names, str):
493             self.names.add(names)
494         elif names is not None:
495             self.names.update(names)
496
497         #: Optional, secondary event filter.
498         self.event_filter: Optional[EventFilter] = event_filter
499
500     @property
501     def history(self) -> Tuple[Message, ...]:
502         """
503         A read-only history of all events seen so far.
504
505         This represents *every* event, including those not yet witnessed
506         via `get()` or ``async for``. It persists between `clear()`
507         calls and is immutable.
508         """
509         return tuple(self._history)
510
511     def accept(self, event: Message) -> bool:
512         """
513         Determine if this listener accepts this event.
514
515         This method determines which events will appear in the stream.
516         The default implementation simply checks the event against the
517         list of names and the event_filter to decide if this
518         `EventListener` accepts a given event. It can be
519         overridden/extended to provide custom listener behavior.
520
521         User code is not expected to need to invoke this method.
522
523         :param event: The event under consideration.
524         :return: `True`, if this listener accepts this event.
525         """
526         name_ok = (not self.names) or (event['event'] in self.names)
527         return name_ok and (
528             (not self.event_filter) or self.event_filter(event)
529         )
530
531     async def put(self, event: Message) -> None:
532         """
533         Conditionally put a new event into the FIFO queue.
534
535         This method is not designed to be invoked from user code, and it
536         should not need to be overridden. It is a public interface so
537         that `QMPClient` has an interface by which it can inform
538         registered listeners of new events.
539
540         The event will be put into the queue if
541         :py:meth:`~EventListener.accept()` returns `True`.
542
543         :param event: The new event to put into the FIFO queue.
544         """
545         if not self.accept(event):
546             return
547
548         self._history.append(event)
549         await self._queue.put(event)
550
551     async def get(self) -> Message:
552         """
553         Wait for the very next event in this stream.
554
555         If one is already available, return that one.
556         """
557         return await self._queue.get()
558
559     def empty(self) -> bool:
560         """
561         Return `True` if there are no pending events.
562         """
563         return self._queue.empty()
564
565     def clear(self) -> List[Message]:
566         """
567         Clear this listener of all pending events.
568
569         Called when an `EventListener` is being unregistered, this clears the
570         pending FIFO queue synchronously. It can be also be used to
571         manually clear any pending events, if desired.
572
573         :return: The cleared events, if any.
574
575         .. warning::
576             Take care when discarding events. Cleared events will be
577             silently tossed on the floor. All events that were ever
578             accepted by this listener are visible in `history()`.
579         """
580         events = []
581         while True:
582             try:
583                 events.append(self._queue.get_nowait())
584             except asyncio.QueueEmpty:
585                 break
586
587         return events
588
589     def __aiter__(self) -> AsyncIterator[Message]:
590         return self
591
592     async def __anext__(self) -> Message:
593         """
594         Enables the `EventListener` to function as an async iterator.
595
596         It may be used like this:
597
598         .. code:: python
599
600             async for event in listener:
601                 print(event)
602
603         These iterators will never terminate of their own accord; you
604         must provide break conditions or otherwise prepare to run them
605         in an `asyncio.Task` that can be cancelled.
606         """
607         return await self.get()
608
609
610 class Events:
611     """
612     Events is a mix-in class that adds event functionality to the QMP class.
613
614     It's designed specifically as a mix-in for `QMPClient`, and it
615     relies upon the class it is being mixed into having a 'logger'
616     property.
617     """
618     def __init__(self) -> None:
619         self._listeners: List[EventListener] = []
620
621         #: Default, all-events `EventListener`.
622         self.events: EventListener = EventListener()
623         self.register_listener(self.events)
624
625         # Parent class needs to have a logger
626         self.logger: logging.Logger
627
628     async def _event_dispatch(self, msg: Message) -> None:
629         """
630         Given a new event, propagate it to all of the active listeners.
631
632         :param msg: The event to propagate.
633         """
634         for listener in self._listeners:
635             await listener.put(msg)
636
637     def register_listener(self, listener: EventListener) -> None:
638         """
639         Register and activate an `EventListener`.
640
641         :param listener: The listener to activate.
642         :raise ListenerError: If the given listener is already registered.
643         """
644         if listener in self._listeners:
645             raise ListenerError("Attempted to re-register existing listener")
646         self.logger.debug("Registering %s.", str(listener))
647         self._listeners.append(listener)
648
649     def remove_listener(self, listener: EventListener) -> None:
650         """
651         Unregister and deactivate an `EventListener`.
652
653         The removed listener will have its pending events cleared via
654         `clear()`. The listener can be re-registered later when
655         desired.
656
657         :param listener: The listener to deactivate.
658         :raise ListenerError: If the given listener is not registered.
659         """
660         if listener == self.events:
661             raise ListenerError("Cannot remove the default listener.")
662         self.logger.debug("Removing %s.", str(listener))
663         listener.clear()
664         self._listeners.remove(listener)
665
666     @contextmanager
667     def listen(self, *listeners: EventListener) -> Iterator[None]:
668         r"""
669         Context manager: Temporarily listen with an `EventListener`.
670
671         Accepts one or more `EventListener` objects and registers them,
672         activating them for the duration of the context block.
673
674         `EventListener` objects will have any pending events in their
675         FIFO queue cleared upon exiting the context block, when they are
676         deactivated.
677
678         :param \*listeners: One or more EventListeners to activate.
679         :raise ListenerError: If the given listener(s) are already active.
680         """
681         _added = []
682
683         try:
684             for listener in listeners:
685                 self.register_listener(listener)
686                 _added.append(listener)
687
688             yield
689
690         finally:
691             for listener in _added:
692                 self.remove_listener(listener)
693
694     @contextmanager
695     def listener(
696         self,
697         names: EventNames = (),
698         event_filter: Optional[EventFilter] = None
699     ) -> Iterator[EventListener]:
700         """
701         Context manager: Temporarily listen with a new `EventListener`.
702
703         Creates an `EventListener` object and registers it, activating
704         it for the duration of the context block.
705
706         :param names:
707             One or more names of events to listen for.
708             When not provided, listen for ALL events.
709         :param event_filter:
710             An optional event filtering function.
711             When names are also provided, this acts as a secondary filter.
712
713         :return: The newly created and active `EventListener`.
714         """
715         listener = EventListener(names, event_filter)
716         with self.listen(listener):
717             yield listener
This page took 0.061366 seconds and 4 git commands to generate.