Make DBus calls Async (2nd attempt)

packaging/Makefile

@@ -15,7 +15,7 @@ debian-package-code:
 	cp ../examples/instance_message_from_aleph.json ./aleph-vm/opt/aleph-vm/examples/instance_message_from_aleph.json
 	cp -r ../examples/data ./aleph-vm/opt/aleph-vm/examples/data
 	mkdir -p ./aleph-vm/opt/aleph-vm/examples/volumes
-	pip3 install --target ./aleph-vm/opt/aleph-vm/ 'aleph-message==0.4.4' 'jwskate==0.8.0' 'eth-account==0.9.0' 'sentry-sdk==1.31.0' 'qmp==1.1.0' 'superfluid==0.2.1' 'sqlalchemy[asyncio]' 'aiosqlite==0.19.0' 'alembic==1.13.1' 'aiohttp_cors==0.7.0' 'pyroute2==0.7.12'
+	pip3 install --target ./aleph-vm/opt/aleph-vm/ 'aleph-message==0.4.4' 'jwskate==0.8.0' 'eth-account==0.9.0' 'sentry-sdk==1.31.0' 'qmp==1.1.0' 'superfluid==0.2.1' 'sqlalchemy[asyncio]' 'aiosqlite==0.19.0' 'alembic==1.13.1' 'aiohttp_cors==0.7.0' 'pyroute2==0.7.12' 'dbus-fast==1.90.1'
 	python3 -m compileall ./aleph-vm/opt/aleph-vm/
 
 debian-package-resources: firecracker-bins vmlinux download-ipfs-kubo

packaging/aleph-vm/DEBIAN/control

@@ -3,6 +3,6 @@ Version: 0.1.8
 Architecture: all
 Maintainer: Aleph.im
 Description: Aleph.im VM execution engine
-Depends: python3,python3-pip,python3-aiohttp,python3-msgpack,python3-aiodns,python3-alembic,python3-sqlalchemy,python3-setproctitle,redis,python3-aioredis,python3-psutil,sudo,acl,curl,systemd-container,squashfs-tools,debootstrap,python3-packaging,python3-cpuinfo,python3-nftables,python3-jsonschema,cloud-image-utils,ndppd,python3-yaml,python3-dotenv,python3-schedule,qemu-system-x86,qemu-utils,python3-systemd,python3-dbus,btrfs-progs,nftables
+Depends: python3,python3-pip,python3-aiohttp,python3-msgpack,python3-aiodns,python3-alembic,python3-sqlalchemy,python3-setproctitle,redis,python3-aioredis,python3-psutil,sudo,acl,curl,systemd-container,squashfs-tools,debootstrap,python3-packaging,python3-cpuinfo,python3-nftables,python3-jsonschema,cloud-image-utils,ndppd,python3-yaml,python3-dotenv,python3-schedule,qemu-system-x86,qemu-utils,python3-systemd,btrfs-progs,nftables
 Section: aleph-im
 Priority: Extra

pyproject.toml

@@ -42,8 +42,7 @@ dependencies = [
   "packaging==23.2",
   "jsonschema==4.19.1",
   "qmp==0.0.1",
-  "dbus-python==1.3.2",
-  "systemd-python==235",
+  "dbus-fast==1.90.1",
   "systemd-python==235",
   "superfluid~=0.2.1",
   "sqlalchemy[asyncio]>=2.0",

src/aleph/vm/models.py

@@ -85,11 +85,7 @@
 
     @property
     def is_running(self) -> bool:
-        return (
-            bool(self.times.starting_at and not self.times.stopping_at)
-            if not self.persistent
-            else self.systemd_manager.is_service_active(self.controller_service)
-        )
+        return bool(self.times.starting_at and not self.times.stopping_at)
 
     @property
     def is_stopping(self) -> bool:

src/aleph/vm/orchestrator/supervisor.py

@@ -140,12 +140,16 @@
 
 def run():
     """Run the VM Supervisor."""
+    # Loop creation set here to avoid bug with Future on different loop
+
+    loop = asyncio.new_event_loop()
+    # apparently needed for Python 3.9 / Debian 11
+    asyncio.set_event_loop(loop)
     settings.check()
 
     engine = setup_engine()
     asyncio.run(create_tables(engine))
 
-    loop = asyncio.new_event_loop()
     pool = VmPool(loop)
     pool.setup()
 
@@ -170,10 +174,10 @@
             app.on_cleanup.append(stop_all_vms)
 
         logger.info("Loading existing executions ...")
-        asyncio.run(pool.load_persistent_executions())
+        loop.run_until_complete(pool.load_persistent_executions())
 
         logger.info(f"Starting the web server on http://{settings.SUPERVISOR_HOST}:{settings.SUPERVISOR_PORT}")
-        web.run_app(app, host=settings.SUPERVISOR_HOST, port=settings.SUPERVISOR_PORT)
+        web.run_app(app, host=settings.SUPERVISOR_HOST, port=settings.SUPERVISOR_PORT, loop=loop)
     except OSError as e:
         if e.errno == 98:
             logger.error(

src/aleph/vm/orchestrator/views/operator.py

@@ -174,7 +174,7 @@
     if execution.is_running:
         logger.info(f"Rebooting {execution.vm_hash}")
         if execution.persistent:
-            pool.systemd_manager.restart(execution.controller_service)
+            await pool.systemd_manager.restart(execution.controller_service)
         else:
             await pool.stop_vm(vm_hash)
             pool.forget_vm(vm_hash)

src/aleph/vm/pool.py

@@ -50,6 +50,7 @@
         self.executions = {}
         self.message_cache = {}
 
+        # apparently needed for Python 3.9 / Debian 11
         asyncio.set_event_loop(loop)
         self.creation_lock = asyncio.Lock()
 
@@ -124,7 +125,7 @@
 
                 # Start VM and snapshots automatically
                 if execution.persistent:
-                    self.systemd_manager.enable_and_start(execution.controller_service)
+                    await self.systemd_manager.enable_and_start(execution.controller_service)
                     await execution.wait_for_init()
                     if execution.is_program and execution.vm:
                         await execution.vm.load_configuration()
@@ -191,7 +192,7 @@
     async def stop_persistent_execution(self, execution: VmExecution):
         """Stop persistent VMs in the pool."""
         assert execution.persistent, "Execution isn't persistent"
-        self.systemd_manager.stop_and_disable(execution.controller_service)
+        await self.systemd_manager.stop_and_disable(execution.controller_service)
         await execution.stop()
 
     def forget_vm(self, vm_hash: ItemHash) -> None:
@@ -239,8 +240,10 @@
                 persistent=saved_execution.persistent,
             )
 
-            if execution.is_running:
-                # TODO: Improve the way that we re-create running execution
+            if await self.systemd_manager.is_service_active(
+                execution.controller_service
+            ):  # TODO: Improve the way that we re-create running execution
+                logger.debug("Execution %s is still running in systemd, reconnecting", execution.vm_hash)
                 await execution.prepare()
                 if self.network:
                     vm_type = VmType.from_message_content(execution.message)
@@ -251,16 +254,21 @@
                 vm = execution.create(vm_id=vm_id, tap_interface=tap_interface, prepare=False)
                 await vm.start_guest_api()
                 execution.ready_event.set()
+                execution.times.starting_at = execution.times.starting_at or datetime.now(tz=timezone.utc)
                 execution.times.started_at = datetime.now(tz=timezone.utc)
-
+                execution.times.stopping_at = None
+                execution.times.stopped_at = None
                 self._schedule_forget_on_stop(execution)
 
                 # Start the snapshot manager for the VM
                 if vm.support_snapshot and self.snapshot_manager:
                     await self.snapshot_manager.start_for(vm=execution.vm)
 
+                assert execution.is_running
                 self.executions[vm_hash] = execution
+
             else:
+                logger.debug(("Execution %s is not running in systemd, reconnecting", execution.vm_hash))
                 execution.uuid = saved_execution.uuid
                 await execution.record_usage()