Schedule synchronous function to separate thread

Waiting for the server to send the result of the API calls, allows
dealing with failures/exceptions:

- QubesVM.shutdown -> .shutdown(wait=True)
- QubesVM.run_service -> .run_service_for_stdio

It also does not block the widget from being opened again even if an
action is still running. This allows, for example:

- To create disposable qube and attach device to it (which takes 10s),
  but not block the widget from opening
- To detach and shutdown disposable that is hanging, without hanging the
  widget

For qui/tray/domains.py, the "widget.destroy()" is called earlier, cause
it's not needed after the response is received. Else it hangs until the
"react_to_question" finishes.

For: QubesOS/qubes-issues#10648
For: QubesOS/qubes-issues#10651
For: QubesOS/qubes-issues#10835
Requires: QubesOS/qubes-core-admin#807
Requires: QubesOS/qubes-core-admin-client#469

Author: ben-grande

qui/devices/actionable_widgets.py

@@ -26,19 +26,20 @@
 import asyncio
 import functools
 import pathlib
+import time
 from typing import Iterable, Callable, Optional, List
 
 import qubesadmin
 import qubesadmin.devices
 import qubesadmin.vm
+from qubesadmin.utils import async_thread
 
 import gi
 
 gi.require_version("Gtk", "3.0")  # isort:skip
 from gi.repository import Gtk, GdkPixbuf, GLib  # isort:skip
 
 from . import backend
-import time
 
 
 def load_icon(icon_name: str, backup_name: str, size: int = 24):
@@ -247,7 +248,7 @@ def __init__(self, vm: backend.VM, device: backend.Device):
             self.actionable = False
 
     async def widget_action(self, *_args):
-        self.device.attach_to_vm(self.vm)
+        await async_thread(self.device.attach_to_vm, self.vm)
 
 
 class DetachWidget(ActionableWidget, SimpleActionWidget):
@@ -259,7 +260,7 @@ def __init__(self, vm: backend.VM, device: backend.Device, variant: str = "dark"
         self.device = device
 
     async def widget_action(self, *_args):
-        self.device.detach_from_vm(self.vm, False)
+        await async_thread(self.device.detach_from_vm, self.vm, False)
 
 
 class DetachWithWidget(ActionableWidget, SimpleActionWidget):
@@ -278,7 +279,7 @@ def __init__(self, vm: backend.VM, device: backend.Device, variant: str = "dark"
         self.device = device
 
     async def widget_action(self, *_args):
-        self.device.detach_from_vm(self.vm, True)
+        await async_thread(self.device.detach_from_vm, self.vm, True)
 
 
 class DetachAndShutdownWidget(ActionableWidget, SimpleActionWidget):
@@ -292,8 +293,8 @@ def __init__(self, vm: backend.VM, device: backend.Device, variant: str = "dark"
         self.device = device
 
     async def widget_action(self, *_args):
-        self.device.detach_from_vm(self.vm, True)
-        self.vm.vm_object.shutdown()
+        await async_thread(self.device.detach_from_vm, self.vm, True)
+        await async_thread(self.vm.vm_object.shutdown)
 
 
 class DetachAndAttachWidget(ActionableWidget, VMWithIcon):
@@ -306,8 +307,8 @@ def __init__(self, vm: backend.VM, device: backend.Device, variant: str = "dark"
 
     async def widget_action(self, *_args):
         for vm in self.device.attachments:
-            self.device.detach_from_vm(vm, True)
-        self.device.attach_to_vm(self.vm)
+            await async_thread(self.device.detach_from_vm, vm, True)
+        await async_thread(self.device.attach_to_vm, self.vm)
 
 
 class AttachDisposableWidget(ActionableWidget, VMWithIcon):
@@ -320,9 +321,9 @@ def __init__(self, vm: backend.VM, device: backend.Device, variant: str = "dark"
 
     async def widget_action(self, *_args):
         new_dispvm = qubesadmin.vm.DispVM.from_appvm(self.vm.vm_object.app, self.vm)
-        new_dispvm.start()
+        await async_thread(new_dispvm.start)
 
-        self.device.attach_to_vm(backend.VM(new_dispvm))
+        await async_thread(self.device.attach_to_vm, backend.VM(new_dispvm))
 
 
 class DetachAndAttachDisposableWidget(ActionableWidget, VMWithIcon):
@@ -334,11 +335,11 @@ def __init__(self, vm: backend.VM, device: backend.Device, variant: str = "dark"
         self.device = device
 
     async def widget_action(self, *_args):
-        self.device.detach_from_vm(self.vm)
+        await async_thread(self.device.detach_from_vm, self.vm)
         new_dispvm = qubesadmin.vm.DispVM.from_appvm(self.vm.vm_object.app, self.vm)
-        new_dispvm.start()
+        await async_thread(new_dispvm.start)
 
-        self.device.attach_to_vm(backend.VM(new_dispvm))
+        await async_thread(self.device.attach_to_vm, backend.VM(new_dispvm))
 
 
 class ToggleFeatureItem(ActionableWidget, SimpleActionWidget):
@@ -379,7 +380,7 @@ def __init__(self, usbvm: backend.VM, variant: str = "dark"):
         self.usbvm = usbvm
 
     async def widget_action(self, *_args):
-        self.usbvm.vm_object.start()
+        await async_thread(self.usbvm.vm_object.start)
 
 
 #### Configuration-related actions

qui/tray/domains.py

@@ -18,6 +18,7 @@
 import qubesadmin
 import qubesadmin.events
 from qubesadmin import exc
+from qubesadmin.utils import async_thread
 
 import qui.decorators
 import qui.utils
@@ -157,7 +158,7 @@ def __init__(self, vm, icon_cache):
 
     async def perform_action(self):
         try:
-            self.vm.pause()
+            await async_thread(self.vm.pause)
         except exc.QubesException as ex:
             show_error(
                 _("Error pausing qube"),
@@ -178,7 +179,7 @@ def __init__(self, vm, icon_cache):
 
     async def perform_action(self):
         try:
-            self.vm.unpause()
+            await async_thread(self.vm.unpause)
         except exc.QubesException as ex:
             show_error(
                 _("Error unpausing qube"),
@@ -215,7 +216,7 @@ def set_force(self, force):
 
     async def perform_action(self):
         try:
-            self.vm.shutdown(force=self.force)
+            await async_thread(self.vm.shutdown, force=self.force, wait=True)
         except exc.QubesException as ex:
             if self.force:
                 show_error(
@@ -271,19 +272,22 @@ async def perform_action(self):
             GLib.idle_add(dialog.show)
 
     def react_to_question(self, widget, response, action):
+        asyncio.create_task(self.react_to_question_async(widget, response, action))
+
+    async def react_to_question_async(self, widget, response, action):
         if response not in (Gtk.ResponseType.OK, Gtk.ResponseType.YES):
             widget.destroy()
             return
         try:
             if action == "force":
-                self.vm.shutdown(force=True)
+                await async_thread(self.vm.shutdown, force=True, wait=True)
             elif action == "timeout":
                 if response == Gtk.ResponseType.YES:
-                    self.vm.kill()
+                    await async_thread(self.vm.kill)
                 elif response == Gtk.ResponseType.OK:
-                    self.vm.shutdown(force=False)
+                    await async_thread(self.vm.shutdown, force=False, wait=True)
             elif action == "kill" and response == Gtk.ResponseType.OK:
-                self.vm.kill()
+                await async_thread(self.vm.kill)
         except exc.QubesException as ex:
             show_error(
                 _("Error shutting down qube"),
@@ -320,7 +324,7 @@ def set_force(self, force):
 
     async def perform_action(self, *_args, **_kwargs):
         try:
-            self.vm.shutdown(force=self.force)
+            await async_thread(self.vm.shutdown, force=self.force, wait=True)
         except exc.QubesException as ex:
             if self.force:
                 # we already tried forcing it, let's just give up
@@ -351,12 +355,7 @@ async def perform_action(self, *_args, **_kwargs):
                 if self.give_up:
                     return
                 await asyncio.sleep(1)
-            proc = await asyncio.create_subprocess_exec(
-                "qvm-start", self.vm.name, stderr=asyncio.subprocess.PIPE
-            )
-            _stdout, stderr = await proc.communicate()
-            if proc.returncode != 0:
-                raise exc.QubesException(stderr)
+            await async_thread(self.vm.start)
         except exc.QubesException as ex:
             show_error(
                 _("Error restarting qube"),
@@ -367,9 +366,12 @@ async def perform_action(self, *_args, **_kwargs):
             )
 
     def react_to_question(self, widget, response):
+        asyncio.create_task(self.react_to_question_async(widget, response))
+
+    async def react_to_question_async(self, widget, response):
         if response == Gtk.ResponseType.OK:
             try:
-                self.vm.shutdown(force=True)
+                await async_thread(self.vm.shutdown, force=True, wait=True)
             except exc.QubesException as ex:
                 show_error(
                     _("Error shutting down qube"),
@@ -392,13 +394,13 @@ def __init__(self, vm, icon_cache):
 
     async def perform_action(self, *_args, **_kwargs):
         try:
-            self.vm.kill()
+            await async_thread(self.vm.kill)
         except exc.QubesException as ex:
             show_error(
                 _("Error shutting down qube"),
                 _(
-                    "The following error occurred while attempting to shut"
-                    "down qube {0}:\n{1}"
+                    "The following error occurred while attempting to kill"
+                    "qube {0}:\n{1}"
                 ).format(self.vm.name, str(ex)),
             )
 
@@ -459,7 +461,11 @@ async def perform_action(self):
         if self.as_root:
             service_args["user"] = "root"
         try:
-            self.vm.run_service("qubes.StartApp+qubes-run-terminal", **service_args)
+            await async_thread(
+                self.vm.run_service_for_stdio,
+                "qubes.StartApp+qubes-run-terminal",
+                **service_args,
+            )
         except exc.QubesException as ex:
             show_error(
                 _("Error starting terminal"),
@@ -508,7 +514,9 @@ def __init__(self, vm, icon_cache):
 
     async def perform_action(self):
         try:
-            self.vm.run_service("qubes.StartApp+qubes-open-file-manager")
+            await async_thread(
+                self.vm.run_service_for_stdio, "qubes.StartApp+qubes-open-file-manager"
+            )
         except exc.QubesException as ex:
             show_error(
                 _("Error opening file manager"),

qui/updater/summary_page.py

@@ -31,7 +31,7 @@
 from typing import Optional, Any
 
 import qubesadmin
-from qubesadmin.events.utils import wait_for_domain_shutdown
+from qubesadmin.utils import async_thread
 
 from qubes_config.widgets.gtk_utils import (
     load_icon,
@@ -308,25 +308,26 @@ def shutdown_domains(self, to_shutdown):
         """
         Try to shut down vms and wait to finish.
         """
-        wait_for = []
-        for vm in to_shutdown:
-            try:
-                vm.shutdown(force=True)
-                wait_for.append(vm)
-                self.log.info("Shutdown %s", vm.name)
-            except qubesadmin.exc.QubesVMError as err:
-                self.err += vm.name + " cannot shutdown: " + str(err) + "\n"
-                self.log.error("Cannot shutdown %s because %s", vm.name, str(err))
-                self.status = RestartStatus.ERROR_TMPL_DOWN
         try:
             loop = asyncio.get_event_loop()
         except RuntimeError:
             # changes between GLib versions and python versions mean that the above
             # can fail on some dom0/gui domain configurations
             loop = asyncio.new_event_loop()
-        loop.run_until_complete(wait_for_domain_shutdown(wait_for))
-
-        return wait_for
+        tasks = [async_thread(vm.shutdown, force=True, wait=True) for vm in to_shutdown]
+        results = loop.run_until_complete(
+            asyncio.gather(*tasks, return_exceptions=True)
+        )
+        done = []
+        for vm, res in zip(to_shutdown, results):
+            if not isinstance(res, qubesadmin.exc.QubesVMError):
+                self.log.info("Shutdown %s", vm.name)
+                done.append(vm)
+                continue
+            self.err += vm.name + " cannot shutdown: " + str(res) + "\n"
+            self.log.error("Cannot shutdown %s because %s", vm.name, str(res))
+            self.status = RestartStatus.ERROR_TMPL_DOWN
+        return done
 
     def restart_vms(self, to_restart):
         """
@@ -335,14 +336,23 @@ def restart_vms(self, to_restart):
         shutdowns = self.shutdown_domains(to_restart)
 
         # restart shutdown qubes
-        for vm in shutdowns:
-            try:
-                vm.start()
+        try:
+            loop = asyncio.get_event_loop()
+        except RuntimeError:
+            # changes between GLib versions and python versions mean that the above
+            # can fail on some dom0/gui domain configurations
+            loop = asyncio.new_event_loop()
+        tasks = [async_thread(vm.start) for vm in shutdowns]
+        results = loop.run_until_complete(
+            asyncio.gather(*tasks, return_exceptions=True)
+        )
+        for vm, res in zip(shutdowns, results):
+            if not isinstance(res, qubesadmin.exc.QubesVMError):
                 self.log.info("Restart %s", vm.name)
-            except qubesadmin.exc.QubesVMError as err:
-                self.err += vm.name + " cannot start: " + str(err) + "\n"
-                self.log.error("Cannot start %s because %s", vm.name, str(err))
-                self.status = RestartStatus.ERROR_APP_DOWN
+                continue
+            self.err += vm.name + " cannot start: " + str(res) + "\n"
+            self.log.error("Cannot start %s because %s", vm.name, str(res))
+            self.status = RestartStatus.ERROR_APP_DOWN
 
     def _show_status_dialog(self, show_only_error: bool):
         if self.status == RestartStatus.OK and not show_only_error: