Merge remote-tracking branch 'origin/pr/820'

* origin/pr/820:
  Fix mypy error my creating a Task
  Short loop.run_in_executor with asyncio.to_thread

Pull request description:

Doesn't require:

- finding the loop as it uses the default;
- passing "None" as executor as it uses the default;
- using partial to pass kwargs

---

~Didn't test at all~, but would be nice to use a shorter version, since we are not using the full capabilities of the executor, a wrapper is easier to read. If you look at `asyncio/threads.py`, it's just calling `loop.run_in_executor`.

Edit: tested manually the preloaded disposable early request before qmemman finishes and watching the logs. Didn't run integration tests, and that would be appropriate for the different files I changed.

Author: marmarek

qubes/storage/reflink.py

@@ -60,9 +60,7 @@ def _coroutinized(function):
 
     @functools.wraps(function)
     async def wrapper(*args, **kwargs):
-        return await asyncio.get_event_loop().run_in_executor(
-            None, functools.partial(function, *args, **kwargs)
-        )
+        return await asyncio.to_thread(function, *args, **kwargs)
 
     return wrapper
 

qubes/storage/zfs.py

@@ -2044,11 +2044,7 @@ async def _wipe_and_clone_from(self, source: qubes.storage.Volume) -> None:
                 self.log.debug("Source is a File volume")
                 # File volume export() does not actually return a coroutine.
                 # This isn't just a typing error.  The await() fails.
-                loop = asyncio.get_event_loop()
-                in_ = await loop.run_in_executor(
-                    None,
-                    source.export,
-                )  # type:ignore
+                in_ = await asyncio.to_thread(source.export)  # type:ignore
             else:
                 self.log.debug("Source is not a ZFS volume")
                 in_ = await source.export()  # type:ignore

qubes/tests/integ/backup.py

@@ -250,14 +250,11 @@ def restore_backup(
         client_app = qubesadmin.Qubes()
         if appvm:
             appvm = self.loop.run_until_complete(
-                self.loop.run_in_executor(
-                    None, client_app.domains.__getitem__, appvm.name
-                )
+                asyncio.to_thread(client_app.domains.__getitem__, appvm.name)
             )
         with self.assertNotRaises(qubesadmin.exc.QubesException):
             restore_op = self.loop.run_until_complete(
-                self.loop.run_in_executor(
-                    None,
+                asyncio.to_thread(
                     qubesadmin.backup.restore.BackupRestore,
                     client_app,
                     backupfile,
@@ -269,17 +266,15 @@ def restore_backup(
                 for key, value in options.items():
                     setattr(restore_op.options, key, value)
             restore_info = self.loop.run_until_complete(
-                self.loop.run_in_executor(None, restore_op.get_restore_info)
+                asyncio.to_thread(restore_op.get_restore_info)
             )
         if callable(manipulate_restore_info):
             restore_info = manipulate_restore_info(restore_info)
         self.log.debug(restore_op.get_restore_summary(restore_info))
 
         with self.assertNotRaises(qubesadmin.exc.QubesException):
             self.loop.run_until_complete(
-                self.loop.run_in_executor(
-                    None, restore_op.restore_do, restore_info
-                )
+                asyncio.to_thread(restore_op.restore_do, restore_info)
             )
 
         errors = []

qubes/tests/integ/backupdispvm.py

@@ -100,7 +100,7 @@ def restore_backup(
             restore_in_dispvm = RestoreInDisposableVM(args.app, args)
             try:
                 backup_log = self.loop.run_until_complete(
-                    self.loop.run_in_executor(None, restore_in_dispvm.run)
+                    asyncio.to_thread(restore_in_dispvm.run)
                 )
             except qubesadmin.exc.BackupRestoreError as e:
                 self.fail(str(e) + " backup log: " + e.backup_log.decode())

qubes/tests/integ/vm_qrexec_gui.py

@@ -812,8 +812,7 @@ async def _test_300_bug_1028_gui_memory_pinning(self):
         winid = await self.wait_for_window_coro(
             self.testvm1.name + ":xterm", search_class=True
         )
-        xprop = await asyncio.get_event_loop().run_in_executor(
-            None,
+        xprop = await asyncio.to_thread(
             subprocess.check_output,
             ["xprop", "-notype", "-id", winid, "_QUBES_VMWINDOWID"],
         )
@@ -850,8 +849,7 @@ async def _test_300_bug_1028_gui_memory_pinning(self):
             "gm import -window {} rgba:-".format(vm_winid)
         )
 
-        dom0_image = await asyncio.get_event_loop().run_in_executor(
-            None,
+        dom0_image = await asyncio.to_thread(
             subprocess.check_output,
             ["gm", "import", "-window", winid, "rgba:-"],
         )

qubes/vm/dispvm.py

@@ -626,10 +626,11 @@ async def on_domain_pre_paused(self, event, **kwargs) -> None:
             return
         break_task = asyncio.create_task(self.preload_requested_event.wait())
         qmemman_client = qubes.qmemman.client.QMemmanClient()
-        qmemman_task = asyncio.get_running_loop().run_in_executor(
-            None,
-            qmemman_client.set_mem,  # type: ignore[arg-type]
-            {self.xid: 0},
+        qmemman_task = asyncio.create_task(
+            asyncio.to_thread(
+                qmemman_client.set_mem,
+                {self.xid: 0},
+            )
         )
         tasks: list = [break_task, qmemman_task]
         result = None

qubes/vm/qubesvm.py

@@ -1453,8 +1453,8 @@ async def start(
                                 notify_function=notify_function,
                             )
 
-                qmemman_client = await asyncio.get_event_loop().run_in_executor(
-                    None, self.request_mem, mem_required
+                qmemman_client = await asyncio.to_thread(
+                    self.request_mem, mem_required
                 )
 
                 await self.storage.start()