Initialize asyncio event loop before using it

doc/qubes-events.rst

@@ -177,8 +177,7 @@ handler (a coroutine) for synchronous event (the one fired with
 
    o = MyClass()
    o.events_enabled = True
-   loop = asyncio.get_event_loop()
-   loop.run_until_complete(o.fire_event_async('event1'))
+   asyncio.run(o.fire_event_async('event1'))
 
 Asynchronous event handlers can also return value - but only a collection, not
 yield individual values (because of python limitation):
@@ -207,7 +206,8 @@ yield individual values (because of python limitation):
 
    o = MyClass()
    o.events_enabled = True
-   loop = asyncio.get_event_loop()
+   loop = asyncio.new_event_loop()
+   asyncio.set_event_loop(loop)
    # returns ['sync result', 'result1', 'result2', 'result3', 'result4'],
    # possibly not in order
    effects = loop.run_until_complete(o.fire_event_async('event1'))

qubes/backup.py

@@ -256,7 +256,7 @@ class Backup:
     >>> }
     >>> backup_op = Backup(app, vms, exclude_vms, **options)
     >>> print(backup_op.get_backup_summary())
-    >>> asyncio.get_event_loop().run_until_complete(backup_op.backup_do())
+    >>> asyncio.run(backup_op.backup_do())
 
     See attributes of this object for all available options.
 

qubes/tests/__init__.py

@@ -494,11 +494,6 @@ def __init__(self, methodName="runTest"):
 
         self._success = True
 
-        global libvirt_event_impl
-
-        if in_dom0 and not libvirt_event_impl:
-            libvirt_event_impl = libvirtaio.virEventRegisterAsyncIOImpl()
-
     def set_result(self, success):
         self._success = success
 
@@ -513,7 +508,17 @@ def setUp(self):
         super().setUp()
         self.addCleanup(self.cleanup_gc)
 
-        self.loop = asyncio.get_event_loop()
+        try:
+            self.loop = asyncio.get_event_loop()
+        except RuntimeError:
+            self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+
+        global libvirt_event_impl
+
+        if in_dom0 and not libvirt_event_impl:
+            libvirt_event_impl = libvirtaio.virEventRegisterAsyncIOImpl()
+
         self.addCleanup(self.cleanup_loop)
 
         self.kernel_validator_original = qubes.app.validate_kernel
@@ -593,6 +598,25 @@ def cleanup_loop(self):
         self.loop.stop()
         self.loop.run_forever()
 
+        # and finally, cancel remaining tasks
+        to_cancel = asyncio.all_tasks(self.loop)
+        if to_cancel:
+            for task in to_cancel:
+                self.log.warning("Leftover task: %r", task)
+                task.cancel()
+
+        self.loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True))
+
+        for task in to_cancel:
+            if task.cancelled():
+                continue
+            if task.exception() is not None:
+                self.log.warning("Unhandled exception during test %r shutdown: %r",
+                    task,
+                    task.exception(),
+                )
+
+
         # Check there are no Tasks left.
         assert not self.loop._ready
         assert not self.loop._scheduled

qubes/tools/qubes_create.py

@@ -40,7 +40,7 @@ def main(args=None):
     """
 
     args = parser.parse_args(args)
-    asyncio.get_event_loop().run_until_complete(
+    asyncio.run(
         qubes.Qubes.create_empty_store(
             args.app, offline_mode=args.offline_mode
         ).setup_pools()

qubes/tools/qubesd.py

@@ -40,7 +40,8 @@ def sighandler(loop, signame, servers):
 
 
 def main(args=None):
-    loop = asyncio.get_event_loop()
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
     libvirtaio.virEventRegisterAsyncIOImpl(loop=loop)
     try:
         args = parser.parse_args(args)

qubes/tools/qubesd_query.py

@@ -112,7 +112,8 @@ async def qubesd_client(socket, payload, *args):
 # pylint: disable=too-many-return-statements
 def main(args=None):
     args = parser.parse_args(args)
-    loop = asyncio.get_event_loop()
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
     max_payload_size = 1024 if args.single_line else MAX_PAYLOAD_SIZE
 
     if args.max_bytes is not None: