diff --git a/qubes/app.py b/qubes/app.py
index a3864189e..69694b842 100644
--- a/qubes/app.py
+++ b/qubes/app.py
@@ -1732,9 +1732,12 @@ def on_property_set_default_netvm(
                     "property-reset:netvm", name="netvm", oldvalue=oldvalue
                 )
 
-    @qubes.events.handler("property-set:default_dispvm")
+    @qubes.events.handler(
+        "property-set:default_dispvm",
+        "property-reset:default_dispvm",
+    )
     def on_property_set_default_dispvm(
-        self, event, name, newvalue, oldvalue=None
+        self, event, name, newvalue=None, oldvalue=None
     ):
         # pylint: disable=unused-argument
         for vm in self.domains:
@@ -1749,6 +1752,39 @@ def on_property_set_default_dispvm(
                     oldvalue=oldvalue,
                 )
 
+        if not newvalue:
+            newvalue = self.default_dispvm
+        if newvalue == oldvalue:
+            return
+        old_appvm = None
+        appvm = None
+        if oldvalue:
+            old_appvm = self.domains[oldvalue]
+        if newvalue:
+            appvm = self.domains[newvalue]
+        if (
+            old_appvm
+            and old_appvm.get_feat_global_preload_max() is not None
+            and old_appvm.is_global_preload_distinct()
+        ):
+            reason = "it was the default_dispvm"
+            asyncio.ensure_future(
+                old_appvm.fire_event_async(
+                    "domain-preload-dispvm-start", reason=reason
+                )
+            )
+        if (
+            appvm
+            and appvm.get_feat_global_preload_max() is not None
+            and appvm.is_global_preload_distinct()
+        ):
+            reason = "it became the default_dispvm"
+            asyncio.ensure_future(
+                appvm.fire_event_async(
+                    "domain-preload-dispvm-start", reason=reason
+                )
+            )
+
     @qubes.events.handler("property-pre-set:default_kernel")
     # pylint: disable-next=invalid-name
     def on_property_pre_set_default_kernel(
diff --git a/qubes/tests/app.py b/qubes/tests/app.py
index f64306fc6..1c17e7d7e 100644
--- a/qubes/tests/app.py
+++ b/qubes/tests/app.py
@@ -689,13 +689,6 @@ def test_101_property_migrate_label(self):
 
 
 class TC_90_Qubes(qubes.tests.QubesTestCase):
-    def tearDown(self):
-        try:
-            os.unlink("/tmp/qubestest.xml")
-        except:
-            pass
-        super().tearDown()
-
     def setUp(self):
         super(TC_90_Qubes, self).setUp()
         self.app = qubes.Qubes(
@@ -707,12 +700,43 @@ def setUp(self):
         self.template = self.app.add_new_vm(
             "TemplateVM", name="test-template", label="green"
         )
+        self.template.features["qrexec"] = True
+        self.template.features["supported-rpc.qubes.WaitForRunningSystem"] = (
+            True
+        )
+        self.appvm = self.app.add_new_vm(
+            "AppVM",
+            name="test-dvm",
+            template=self.template,
+            label="red",
+        )
+        self.appvm_alt = self.app.add_new_vm(
+            qubes.vm.appvm.AppVM,
+            name="test-alt-dvm",
+            template=self.template,
+            label="red",
+        )
+        for qube in [self.appvm, self.appvm_alt]:
+            qube.features["gui"] = False
+            qube.template_for_dispvms = True
+        self.app.save()
+        self.emitter = qubes.tests.TestEmitter()
+
+    def tearDown(self):
+        try:
+            os.unlink("/tmp/qubestest.xml")
+        except:
+            pass
+        del self.emitter
+        super().tearDown()
 
     def cleanup_qubes(self):
         self.app.close()
         del self.app
         try:
             del self.template
+            del self.appvm
+            del self.appvm_alt
         except AttributeError:
             pass
 
@@ -932,11 +956,7 @@ def test_116_remotevm_add_and_remove(self):
 
         assert remotevm1 in self.app.domains
         del self.app.domains["remote-vm1"]
-
-        self.assertCountEqual(
-            {d.name for d in self.app.domains},
-            {"dom0", "test-template", "test-vm", "remote-vm2"},
-        )
+        assert "remote-vm1" not in self.app.domains
 
     def test_117_remotevm_status(self):
         remotevm1 = self.app.add_new_vm(
@@ -1096,6 +1116,90 @@ def test_207_default_kernel(self):
                 existence.side_effect = [True, False]
                 self.app.default_kernel = "unittest_GNU_Hurd_1.0.0"
 
+    def test_300_preload_default_dispvm(self):
+        """Fire event for new setting from no previous one."""
+        self.appvm.features["preload-dispvm-max"] = "1"
+        with mock.patch.object(self.appvm, "fire_event_async") as mock_events:
+            self.app.default_dispvm = self.appvm
+            mock_events.assert_not_called()
+
+        self.app.default_dispvm = None
+        del self.appvm.features["preload-dispvm-max"]
+        self.app.domains["dom0"].features["preload-dispvm-max"] = "1"
+        with mock.patch.object(self.appvm, "fire_event_async") as mock_events:
+            self.app.default_dispvm = self.appvm
+            mock_events.assert_called_once_with(
+                "domain-preload-dispvm-start", reason=mock.ANY
+            )
+
+    def test_301_preload_default_dispvm_change(self):
+        """Fire event for old and new setting."""
+        self.appvm.features["preload-dispvm-max"] = "1"
+        self.app.default_dispvm = self.appvm
+        self.appvm_alt.features["preload-dispvm-max"] = "1"
+        # Global is not set and thus there are no events.
+        with mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_old, mock.patch.object(
+            self.appvm_alt, "fire_event_async"
+        ) as mock_new:
+            self.app.default_dispvm = self.appvm_alt
+            mock_old.assert_not_called()
+            mock_new.assert_not_called()
+
+        self.app.domains["dom0"].features["preload-dispvm-max"] = "1"
+        self.app.default_dispvm = self.appvm
+        self.appvm.features["preload-dispvm-max"] = "2"
+        self.appvm_alt.features["preload-dispvm-max"] = "2"
+        with mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_old, mock.patch.object(
+            self.appvm_alt, "fire_event_async"
+        ) as mock_new:
+            self.app.default_dispvm = self.appvm_alt
+            mock_old.assert_called_once_with(
+                "domain-preload-dispvm-start", reason=mock.ANY
+            )
+            mock_new.assert_called_once_with(
+                "domain-preload-dispvm-start", reason=mock.ANY
+            )
+
+    def test_302_preload_default_dispvm_change_noop(self):
+        """If global feature has the same value as the new and old setting,
+        don't fire the event.
+        """
+        self.appvm.features["preload-dispvm-max"] = "1"
+        self.appvm_alt.features["preload-dispvm-max"] = "1"
+        self.app.domains["dom0"].features["preload-dispvm-max"] = "1"
+        self.app.default_dispvm = self.appvm
+        with mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_old, mock.patch.object(
+            self.appvm_alt, "fire_event_async"
+        ) as mock_new:
+            self.app.default_dispvm = self.appvm_alt
+            mock_old.assert_not_called()
+            mock_new.assert_not_called()
+
+    def test_303_preload_default_dispvm_change_partial_noop(self):
+        """If global feature has the same value as the new or old setting,
+        don't fire the event.
+        """
+        self.appvm.features["preload-dispvm-max"] = "2"
+        self.appvm_alt.features["preload-dispvm-max"] = "1"
+        self.app.domains["dom0"].features["preload-dispvm-max"] = "2"
+        self.app.default_dispvm = self.appvm
+        with mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_old, mock.patch.object(
+            self.appvm_alt, "fire_event_async"
+        ) as mock_new:
+            self.app.default_dispvm = self.appvm_alt
+            mock_old.assert_not_called()
+            mock_new.assert_called_once_with(
+                "domain-preload-dispvm-start", reason=mock.ANY
+            )
+
     @qubes.tests.skipUnlessGit
     def test_900_example_xml_in_doc(self):
         self.assertXMLIsValid(
diff --git a/qubes/tests/integ/dispvm.py b/qubes/tests/integ/dispvm.py
index 4b96bdd28..a8b97df18 100644
--- a/qubes/tests/integ/dispvm.py
+++ b/qubes/tests/integ/dispvm.py
@@ -72,7 +72,7 @@ def tearDown(self):
         self.app.default_dispvm = None
         super(TC_04_DispVM, self).tearDown()
 
-    def wait_for_dispvm_destroy(self, dispvm_name):
+    def wait_for_dispvm_destroy(self, dispvm_name: list):
         timeout = 20
         while dispvm_name in self.app.domains:
             self.loop.run_until_complete(asyncio.sleep(1))
@@ -204,6 +204,7 @@ def setUp(self):  # pylint: disable=invalid-name
         self.addCleanup(
             self.app.remove_handler, "domain-add", self._on_domain_add
         )
+        self.adminvm = self.app.domains["dom0"]
         self.init_default_template(self.template)
         self.disp_base = self.app.add_new_vm(
             qubes.vm.appvm.AppVM,
@@ -212,7 +213,18 @@ def setUp(self):  # pylint: disable=invalid-name
             template_for_dispvms=True,
         )
         self.loop.run_until_complete(self.disp_base.create_on_disk())
-        self.app.default_dispvm = self.disp_base
+        self.disp_base_alt = self.app.add_new_vm(
+            qubes.vm.appvm.AppVM,
+            name=self.make_vm_name("dvm-alt"),
+            label="red",
+            template_for_dispvms=True,
+        )
+        self.loop.run_until_complete(self.disp_base_alt.create_on_disk())
+        # Setting "default_dispvm" fires the preload event before patches of
+        # each test function is applied.
+        if "_preload_" not in self._testMethodName:
+            self.app.default_dispvm = self.disp_base
+        self.cleanup_preload()
         self.app.save()
         self.preload_cmd = [
             "qvm-run",
@@ -229,11 +241,11 @@ def tearDown(self):  # pylint: disable=invalid-name
         logger.info("start")
         if "gui" in self.disp_base.features:
             del self.disp_base.features["gui"]
-        old_preload = self.disp_base.get_feat_preload()
-        self.app.default_dispvm = None
-        tasks = [self.app.domains[x].cleanup() for x in old_preload]
-        self.loop.run_until_complete(asyncio.gather(*tasks))
-        self.disp_base.features["preload-dispvm-max"] = False
+        self.cleanup_preload()
+        # See comment in setUp().
+        if "_preload_" not in self._testMethodName:
+            self.app.default_dispvm = None
+        self.app.save()
         super(TC_20_DispVMMixin, self).tearDown()
         logger.info("end")
 
@@ -246,7 +258,6 @@ def _test_event_handler(
         self.event_handler.setdefault(vm.name, {}).setdefault(event, 0)
         self.event_handler[vm.name][event] += 1
 
-
     def _test_event_handler_remove(self, vm, event):
         if not hasattr(self, "event_handler"):
             self.event_handler = {}
@@ -271,6 +282,8 @@ def _register_handlers(self, vm):  # pylint: disable=unused-argument
             "domain-paused",
             "domain-unpaused",
             "domain-feature-delete:internal",
+            # debug
+            "domain-shutdown",
         ]
         for event in events:
             vm.add_handler(event, self._test_event_handler)
@@ -278,22 +291,66 @@ def _register_handlers(self, vm):  # pylint: disable=unused-argument
     def _on_domain_add(self, app, event, vm):  # pylint: disable=unused-argument
         self._register_handlers(vm)
 
+    async def cleanup_preload_run(self, qube):
+        old_preload = qube.get_feat_preload()
+        tasks = [self.app.domains[x].cleanup() for x in old_preload]
+        await asyncio.gather(*tasks)
+
+    def cleanup_preload(self):
+        logger.info("start")
+        if "preload-dispvm-max" in self.app.domains[self.disp_base].features:
+            logger.info("has local preload")
+            self.loop.run_until_complete(
+                self.cleanup_preload_run(self.disp_base)
+            )
+            logger.info("deleting local feature")
+            del self.disp_base.features["preload-dispvm-max"]
+        if "preload-dispvm-max" in self.app.domains["dom0"].features:
+            logger.info("has global preload")
+            default_dispvm = self.app.default_dispvm
+            if default_dispvm:
+                self.loop.run_until_complete(
+                    self.cleanup_preload_run(default_dispvm)
+                )
+            logger.info("deleting global feature")
+            del self.app.domains["dom0"].features["preload-dispvm-max"]
+        logger.info("end")
+
     async def no_preload(self):
         # Trick to gather this function as an async task.
         await asyncio.sleep(0)
         self.disp_base.features["preload-dispvm-max"] = False
 
+    def log_preload(self):
+        preload_dict = {}
+        default_dispvm = self.app.default_dispvm
+        global_preload_max = None
+        if default_dispvm:
+            global_preload_max = default_dispvm.get_feat_global_preload_max()
+        preload_dict["global"] = {
+            "name": default_dispvm.name if default_dispvm else None,
+            "max": global_preload_max,
+        }
+        for qube in [self.disp_base, self.disp_base_alt]:
+            preload = qube.get_feat_preload()
+            preload_max = qube.get_feat_preload_max()
+            preload_dict[qube.name] = {"max": preload_max, "list": preload}
+        logger.info(preload_dict)
+
     async def wait_preload(
         self,
         preload_max,
+        appvm=None,
         wait_completion=True,
         fail_on_timeout=True,
         timeout=60,
     ):
         """Waiting for completion avoids coroutine objects leaking."""
         logger.info("start")
+        if not appvm:
+            appvm = self.disp_base
         for _ in range(timeout):
-            preload_dispvm = self.disp_base.get_feat_preload()
+            preload_dispvm = appvm.get_feat_preload()
             if len(preload_dispvm) == preload_max:
                 break
             await asyncio.sleep(1)
@@ -303,7 +360,7 @@ async def wait_preload(
         if not wait_completion:
             logger.info("end")
             return
-        preload_dispvm = self.disp_base.get_feat_preload()
+        preload_dispvm = appvm.get_feat_preload()
         preload_unfinished = preload_dispvm
         for _ in range(timeout):
             for qube in preload_unfinished.copy():
@@ -319,6 +376,18 @@ async def wait_preload(
                 self.fail("last preloaded didn't complete in time")
         logger.info("end")
 
+    def wait_for_dispvm_destroy(self, dispvm_names):
+        logger.info("start")
+        timeout = 20
+        while True:
+            if set(dispvm_names).isdisjoint(self.app.domains):
+                break
+            self.loop.run_until_complete(asyncio.sleep(1))
+            timeout -= 1
+            if timeout <= 0:
+                self.fail("didn't destroy dispvm(s) in time")
+        logger.info("end")
+
     async def run_preload_proc(self):
         logger.info("start")
         proc = await asyncio.create_subprocess_exec(
@@ -414,18 +483,18 @@ def test_010_dvm_run_simple(self):
         finally:
             self.loop.run_until_complete(dispvm.cleanup())
 
-    def test_011_dvm_run_preload_reject_max(self):
+    def test_011_preload_reject_max(self):
         """Test preloading when max has been reached"""
         self.loop.run_until_complete(
             qubes.vm.dispvm.DispVM.from_appvm(self.disp_base, preload=True)
         )
         self.assertEqual(0, len(self.disp_base.get_feat_preload()))
 
-    def test_012_dvm_run_preload_low_mem(self):
+    def test_012_preload_low_mem(self):
         """Test preloading with low memory"""
-        self.loop.run_until_complete(self._test_012_dvm_run_preload_low_mem())
+        self.loop.run_until_complete(self._test_012_preload_low_mem())
 
-    async def _test_012_dvm_run_preload_low_mem(self):
+    async def _test_012_preload_low_mem(self):
         # pylint: disable=unspecified-encoding
         logger.info("start")
         unpatched_open = open
@@ -445,12 +514,12 @@ def mock_open_mem(file, *args, **kwargs):
             self.assertEqual(1, len(self.disp_base.get_feat_preload()))
         logger.info("end")
 
-    def test_013_dvm_run_preload_gui(self):
+    def test_013_preload_gui(self):
         """Test preloading with GUI feature enabled and use after
         completion."""
-        self.loop.run_until_complete(self._test_013_dvm_run_preload_gui())
+        self.loop.run_until_complete(self._test_013_preload_gui())
 
-    async def _test_013_dvm_run_preload_gui(self):
+    async def _test_013_preload_gui(self):
         logger.info("start")
         preload_max = 1
         self.disp_base.features["gui"] = True
@@ -459,12 +528,12 @@ async def _test_013_dvm_run_preload_gui(self):
         await self.run_preload()
         logger.info("end")
 
-    def test_014_dvm_run_preload_nogui(self):
+    def test_014_preload_nogui(self):
         """Test preloading with GUI feature disabled and use before
         completion."""
-        self.loop.run_until_complete(self._test_014_dvm_run_preload_nogui())
+        self.loop.run_until_complete(self._test_014_preload_nogui())
 
-    async def _test_014_dvm_run_preload_nogui(self):
+    async def _test_014_preload_nogui(self):
         logger.info("start")
         preload_max = 1
         self.disp_base.features["gui"] = False
@@ -474,22 +543,22 @@ async def _test_014_dvm_run_preload_nogui(self):
         await self.run_preload()
         logger.info("end")
 
-    def test_015_dvm_run_preload_race_more(self):
+    def test_015_preload_race_more(self):
         """Test race requesting multiple preloaded qubes"""
-        self.loop.run_until_complete(self._test_015_dvm_run_preload_race_more())
+        self.loop.run_until_complete(self._test_015_preload_race_more())
 
-    async def _test_015_dvm_run_preload_race_more(self):
+    async def _test_015_preload_race_more(self):
         # The limiting factor is how much memory is available on OpenQA and the
         # unreasonable memory allocated before the qube is paused due to:
         #   https://github.com/QubesOS/qubes-issues/issues/9917
-        # Whonix (Kicksecure) 17 fail more due to memory consumption. From the
-        # templates deployed by default, only Debian and Fedora survives due to
-        # using less memory than the other OSes.
+        # Whonix (Kicksecure) 17 fail more due to higher memory consumption.
+        # From the templates deployed by default, only Debian and Fedora
+        # survives due to using less memory than the other OSes.
         logger.info("start")
-        preload_max = 4
-        os_dist = self.disp_base.features.check_with_template("os-distribution")
-        if os_dist in ["whonix", "kicksecure"]:
-            preload_max -= 1
+        preload_max = 3
+        # dist = self.disp_base.features.check_with_template("os-distribution")
+        # if dist in ["whonix", "kicksecure"]:
+        #     preload_max -= 1
         self.disp_base.features["preload-dispvm-max"] = str(preload_max)
         await self.wait_preload(preload_max)
         old_preload = self.disp_base.get_feat_preload()
@@ -502,11 +571,11 @@ async def _test_015_dvm_run_preload_race_more(self):
         self.assertEqual(len(targets), len(set(targets)))
         logger.info("end")
 
-    def test_016_dvm_run_preload_race_less(self):
+    def test_016_preload_race_less(self):
         """Test race requesting preloaded qube while the maximum is zeroed."""
-        self.loop.run_until_complete(self._test_016_dvm_run_preload_race_less())
+        self.loop.run_until_complete(self._test_016_preload_race_less())
 
-    async def _test_016_dvm_run_preload_race_less(self):
+    async def _test_016_preload_race_less(self):
         logger.info("start")
         preload_max = 1
         self.disp_base.features["preload-dispvm-max"] = str(preload_max)
@@ -517,7 +586,7 @@ async def _test_016_dvm_run_preload_race_less(self):
         self.assertTrue(target_dispvm.startswith("disp"))
         logger.info("end")
 
-    def test_017_dvm_run_preload_autostart(self):
+    def test_017_preload_autostart(self):
         logger.info("start")
         preload_max = 1
         proc = self.loop.run_until_complete(
@@ -543,6 +612,77 @@ def test_017_dvm_run_preload_autostart(self):
         )
         logger.info("end")
 
+    def test_018_preload_global(self):
+        """Tweak global preload setting and global dispvm."""
+        self.loop.run_until_complete(self._test_018_preload_global())
+
+    async def _test_018_preload_global(self):
+        logger.info("start")
+        self.log_preload()
+        preload_max = 1
+
+        logger.info("set global dispvm")
+        self.app.default_dispvm = self.disp_base
+        logger.info("set global feat, state must change")
+        self.adminvm.features["preload-dispvm-max"] = str(preload_max)
+        await self.wait_preload(preload_max)
+
+        self.log_preload()
+        logger.info("set local feat, state must not change")
+        preload_max += 1
+        self.disp_base.features["preload-dispvm-max"] = str(preload_max)
+        await self.wait_preload(preload_max, fail_on_timeout=False, timeout=15)
+        self.assertEqual(len(self.disp_base.get_feat_preload()), 1)
+
+        self.log_preload()
+        logger.info("del local feat, state must not change")
+        del self.disp_base.features["preload-dispvm-max"]
+        await asyncio.sleep(5)
+        self.assertEqual(len(self.disp_base.get_feat_preload()), 1)
+
+        self.log_preload()
+        logger.info("set local feat and del global feat, state must change")
+        self.disp_base.features["preload-dispvm-max"] = str(preload_max)
+        del self.adminvm.features["preload-dispvm-max"]
+        await self.wait_preload(preload_max)
+
+        self.log_preload()
+        logger.info("del local feat and set global feat, state must change")
+        preload_max -= 1
+        preload_remove = self.app.default_dispvm.get_feat_preload()
+        self.disp_base.features["preload-dispvm-max"] = ""
+        self.wait_for_dispvm_destroy(preload_remove)
+        self.adminvm.features["preload-dispvm-max"] = str(preload_max)
+        await self.wait_preload(preload_max)
+        self.assertEqual(len(self.disp_base.get_feat_preload()), preload_max)
+
+        self.log_preload()
+        logger.info("switch global dispvm, state must change")
+        self.app.default_dispvm = self.disp_base_alt
+        await self.wait_preload(preload_max, appvm=self.disp_base_alt)
+
+        self.log_preload()
+        logger.info("set local feat, state must not change")
+        preload_max += 1
+        self.disp_base.features["preload-dispvm-max"] = str(preload_max)
+        await self.wait_preload(preload_max)
+
+        self.log_preload()
+        logger.info("switch back global dispvm, state must change")
+        preload_remove = self.app.default_dispvm.get_feat_preload()
+        self.app.default_dispvm = self.disp_base
+        self.wait_for_dispvm_destroy(preload_remove)
+        await self.wait_preload(preload_max)
+
+        self.log_preload()
+        logger.info("unset global dispvm, state must change")
+        preload_remove = self.app.default_dispvm.get_feat_preload()
+        self.app.default_dispvm = None
+        self.wait_for_dispvm_destroy(preload_remove)
+
+        self.log_preload()
+        logger.info("end")
+
     @unittest.skipUnless(
         spawn.find_executable("xdotool"), "xdotool not installed"
     )
diff --git a/qubes/tests/vm/adminvm.py b/qubes/tests/vm/adminvm.py
index 2e9ade46e..e8152107a 100644
--- a/qubes/tests/vm/adminvm.py
+++ b/qubes/tests/vm/adminvm.py
@@ -18,6 +18,7 @@
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 #
 import grp
+import os
 import subprocess
 import unittest
 import unittest.mock
@@ -37,44 +38,44 @@
 class TC_00_AdminVM(qubes.tests.QubesTestCase):
     def setUp(self):
         super().setUp()
-        try:
-            self.app = qubes.tests.vm.TestApp()
-            with unittest.mock.patch.object(
-                qubes.vm.adminvm.AdminVM, "start_qdb_watch"
-            ) as mock_qdb:
-                self.vm = qubes.vm.adminvm.AdminVM(self.app, xml=None)
-                mock_qdb.assert_called_once_with()
-                self.addCleanup(self.cleanup_adminvm)
-        except:  # pylint: disable=bare-except
-            if self.id().endswith(".test_000_init"):
-                raise
-            self.skipTest("setup failed")
+        self.app = qubes.Qubes(
+            "/tmp/qubestest.xml", load=False, offline_mode=True
+        )
+        self.app.load_initial_values()
+        self.vm = self.app.domains["dom0"]
+        self.template = self.app.add_new_vm(
+            "TemplateVM", name="test-template", label="green"
+        )
+        self.template.features["qrexec"] = True
+        self.template.features["supported-rpc.qubes.WaitForRunningSystem"] = (
+            True
+        )
+        self.appvm = self.app.add_new_vm(
+            "AppVM",
+            name="test-dvm",
+            template=self.template,
+            label="red",
+        )
+        self.appvm.features["gui"] = False
+        self.appvm.template_for_dispvms = True
+        self.emitter = qubes.tests.TestEmitter()
 
     def tearDown(self) -> None:
-        self.app.domains.clear()
-
-    def add_vm(self, name, cls=qubes.vm.qubesvm.QubesVM, **kwargs):
-        vm = cls(
-            self.app,
-            None,
-            qid=kwargs.pop("qid", 1),
-            name=qubes.tests.VMPREFIX + name,
-            **kwargs
-        )
-        self.app.domains[vm.qid] = vm
-        self.app.domains[vm.uuid] = vm
-        self.app.domains[vm.name] = vm
-        self.app.domains[vm] = vm
-        self.addCleanup(vm.close)
-        return vm
+        del self.appvm
+        del self.template
+        del self.vm
+        self.app.close()
+        del self.app
+        del self.emitter
+        try:
+            os.unlink("/tmp/qubestest.xml")
+        except:
+            pass
+        super().tearDown()
 
     async def coroutine_mock(self, mock, *args, **kwargs):
         return mock(*args, **kwargs)
 
-    def cleanup_adminvm(self):
-        self.vm.close()
-        del self.vm
-
     def test_000_init(self):
         pass
 
@@ -118,8 +119,6 @@ def test_311_suspend(self):
 
     @unittest.mock.patch("asyncio.create_subprocess_exec")
     def test_700_run_service(self, mock_subprocess):
-        self.add_vm("vm")
-
         # if there is a user in 'qubes' group, it should be used by default
         try:
             gr = grp.getgrnam("qubes")
@@ -159,13 +158,13 @@ def test_700_run_service(self, mock_subprocess):
             mock_subprocess.reset_mock()
         with self.subTest("other_source"):
             self.loop.run_until_complete(
-                self.vm.run_service("test.service", source="test-inst-vm")
+                self.vm.run_service("test.service", source=self.appvm.name)
             )
             mock_subprocess.assert_called_once_with(
                 *command_prefix,
                 "/usr/lib/qubes/qubes-rpc-multiplexer",
                 "test.service",
-                "test-inst-vm",
+                self.appvm.name,
                 "name",
                 "dom0"
             )
@@ -227,3 +226,56 @@ def test_711_adminvm_ordering(self):
         assert self.vm < qubes.vm.qubesvm.QubesVM(
             self.app, None, qid=1, name="dom0"
         )
+
+    def test_800_preload_set_max(self):
+        self.app.default_dispvm = None
+        with unittest.mock.patch.object(
+            self.appvm, "fire_event"
+        ) as mock_events:
+            self.vm.features["preload-dispvm-max"] = "1"
+            mock_events.assert_not_called()
+        del self.vm.features["preload-dispvm-max"]
+
+        self.app.default_dispvm = self.appvm
+        with unittest.mock.patch.object(
+            self.appvm, "fire_event"
+        ) as mock_sync, unittest.mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_async:
+            self.vm.features["preload-dispvm-max"] = "1"
+            mock_sync.assert_called_once_with(
+                "domain-feature-pre-set:preload-dispvm-max",
+                pre_event=True,
+                feature="preload-dispvm-max",
+                value="1",
+                oldvalue=None,
+            )
+            mock_async.assert_called_once_with(
+                "domain-preload-dispvm-start", reason=unittest.mock.ANY
+            )
+
+        # Setting the feature to the same value it has skips firing the event.
+        with unittest.mock.patch.object(
+            self.appvm, "fire_event"
+        ) as mock_events:
+            self.vm.features["preload-dispvm-max"] = "1"
+            mock_events.assert_not_called()
+
+    def test_801_preload_del_max(self):
+        self.vm.features["preload-dispvm-max"] = "1"
+        self.app.default_dispvm = None
+        with unittest.mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_events:
+            del self.vm.features["preload-dispvm-max"]
+            mock_events.assert_not_called()
+
+        self.vm.features["preload-dispvm-max"] = "1"
+        self.app.default_dispvm = self.appvm
+        with unittest.mock.patch.object(
+            self.appvm, "fire_event_async"
+        ) as mock_events:
+            del self.vm.features["preload-dispvm-max"]
+            mock_events.assert_called_once_with(
+                "domain-preload-dispvm-start", reason=unittest.mock.ANY
+            )
diff --git a/qubes/tests/vm/dispvm.py b/qubes/tests/vm/dispvm.py
index 118899014..28847aec3 100644
--- a/qubes/tests/vm/dispvm.py
+++ b/qubes/tests/vm/dispvm.py
@@ -95,11 +95,13 @@ def _test_event_handler(
         self, vm, event, *args, **kwargs
     ):  # pylint: disable=unused-argument
         if not hasattr(self, "event_handler"):
+            # pylint: disable=attribute-defined-outside-init
             self.event_handler = {}
         self.event_handler.setdefault(vm.name, {})[event] = True
 
     def _test_event_was_handled(self, vm, event):
         if not hasattr(self, "event_handler"):
+            # pylint: disable=attribute-defined-outside-init
             self.event_handler = {}
         return self.event_handler.get(vm, {}).get(event)
 
@@ -217,6 +219,59 @@ def test_000_from_appvm_preload_use(
         # can use the same qube that was preloaded is enough for unit tests.
         self.assertEqual(dispvm.name, fresh_dispvm.name)
 
+    @mock.patch("qubes.vm.qubesvm.QubesVM.start")
+    @mock.patch("os.symlink")
+    @mock.patch("os.makedirs")
+    @mock.patch("qubes.storage.Storage")
+    def test_000_from_appvm_preload_fill_gap(
+        self,
+        mock_storage,
+        mock_makedirs,
+        mock_symlink,
+        mock_start,
+    ):
+        mock_storage.return_value.create.side_effect = self.mock_coro
+        mock_start.side_effect = self.mock_coro
+        self.appvm.template_for_dispvms = True
+        self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
+        orig_getitem = self.app.domains.__getitem__
+        with mock.patch("qubes.events.Emitter.fire_event_async") as mock_events:
+            self.appvm.features["preload-dispvm-max"] = "1"
+        with mock.patch.object(
+            self.app, "domains", wraps=self.app.domains
+        ) as mock_domains, mock.patch(
+            "qubes.events.Emitter.fire_event_async"
+        ) as mock_events:
+            mock_events.assert_not_called()
+            mock_qube = mock.Mock()
+            mock_qube.template = self.appvm
+            mock_qube.qrexec_timeout = self.appvm.qrexec_timeout
+            mock_qube.preload_complete = mock.Mock(spec=asyncio.Event)
+            mock_qube.preload_complete.is_set.return_value = True
+            mock_qube.preload_complete.set = self.mock_coro
+            mock_qube.preload_complete.clear = self.mock_coro
+            mock_qube.preload_complete.wait = self.mock_coro
+            mock_domains.configure_mock(
+                **{
+                    "get_new_unused_dispid": mock.Mock(return_value=42),
+                    "__contains__.return_value": True,
+                    "__getitem__.side_effect": lambda key: (
+                        mock_qube if key == "disp42" else orig_getitem(key)
+                    ),
+                }
+            )
+            self.loop.run_until_complete(
+                qubes.vm.dispvm.DispVM.from_appvm(self.appvm)
+            )
+            assert mock_events.mock_calls == [
+                mock.call(
+                    "domain-preload-dispvm-start", reason=unittest.mock.ANY
+                ),
+                mock.call("domain-create-on-disk"),
+            ]
+        mock_symlink.assert_not_called()
+        mock_makedirs.assert_called_once()
+
     def test_001_from_appvm_reject_not_allowed(self):
         with self.assertRaises(qubes.exc.QubesException):
             dispvm = self.loop.run_until_complete(
diff --git a/qubes/tests/vm/mix/dvmtemplate.py b/qubes/tests/vm/mix/dvmtemplate.py
index 652fa100b..a1f3492b4 100755
--- a/qubes/tests/vm/mix/dvmtemplate.py
+++ b/qubes/tests/vm/mix/dvmtemplate.py
@@ -34,12 +34,16 @@
 class TestApp(qubes.tests.vm.TestApp):
     def __init__(self):
         super(TestApp, self).__init__()
-        self.qid_counter = 1
+        self.qid_counter = 0
 
     def add_new_vm(self, cls, **kwargs):
         qid = self.qid_counter
-        self.qid_counter += 1
-        vm = cls(self, None, qid=qid, **kwargs)
+        if self.qid_counter == 0:
+            self.qid_counter += 1
+            vm = cls(self, None, **kwargs)
+        else:
+            self.qid_counter += 1
+            vm = cls(self, None, qid=qid, **kwargs)
         self.domains[vm.name] = vm
         self.domains[vm] = vm
         return vm
@@ -60,6 +64,8 @@ def setUp(self):
             name="linux-kernel"
         )
         self.app.vmm.offline_mode = True
+        self.adminvm = self.app.add_new_vm(qubes.vm.adminvm.AdminVM)
+        self.addCleanup(self.cleanup_adminvm)
         self.template = self.app.add_new_vm(
             qubes.vm.templatevm.TemplateVM, name="test-template", label="red"
         )
@@ -70,15 +76,24 @@ def setUp(self):
             label="red",
         )
         self.appvm.template_for_dispvms = True
+        self.appvm.features["qrexec"] = True
+        self.appvm.features["gui"] = False
+        self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
         self.app.domains[self.appvm.name] = self.appvm
         self.app.domains[self.appvm] = self.appvm
+        self.app.default_dispvm = self.appvm
         self.addCleanup(self.cleanup_dispvm)
         self.emitter = qubes.tests.TestEmitter()
 
     def tearDown(self):
+        self.app.default_dispvm = None
         del self.emitter
         super(TC_00_DVMTemplateMixin, self).tearDown()
 
+    def cleanup_adminvm(self):
+        self.adminvm.close()
+        del self.adminvm
+
     def cleanup_dispvm(self):
         if hasattr(self, "dispvm"):
             self.dispvm.close()
@@ -94,9 +109,6 @@ async def mock_coro(self, *args, **kwargs):
         pass
 
     def test_010_dvm_preload_get_max(self):
-        self.appvm.features["qrexec"] = True
-        self.appvm.features["gui"] = False
-        self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
         cases = [
             (None, 0),
             (False, 0),
@@ -128,6 +140,22 @@ def test_010_dvm_preload_get_max(self):
                 with self.assertRaises(qubes.exc.QubesValueError):
                     self.appvm.features["preload-dispvm-max"] = value
 
+        # Global setting from now on.
+        if "preload-dispvm-max" in self.adminvm.features:
+            del self.adminvm.features["preload-dispvm-max"]
+        self.appvm.features["preload-dispvm-max"] = "1"
+        self.assertEqual(self.appvm.get_feat_global_preload_max(), None)
+        self.assertEqual(self.appvm.get_feat_preload_max(), 1)
+
+        self.adminvm.features["preload-dispvm-max"] = ""
+        self.appvm.features["preload-dispvm-max"] = "1"
+        self.assertEqual(self.appvm.get_feat_global_preload_max(), 0)
+        self.assertEqual(self.appvm.get_feat_preload_max(), 0)
+
+        self.app.default_dispvm = None
+        self.assertEqual(self.appvm.get_feat_global_preload_max(), 0)
+        self.assertEqual(self.appvm.get_feat_preload_max(), 1)
+
     @mock.patch("os.symlink")
     @mock.patch("os.makedirs")
     @mock.patch("qubes.storage.Storage")
@@ -137,9 +165,6 @@ def test_010_dvm_preload_get_list(
         mock_storage.return_value.create.side_effect = self.mock_coro
         mock_makedirs.return_value.create.side_effect = self.mock_coro
         mock_symlink.return_value.create.side_effect = self.mock_coro
-        self.appvm.features["qrexec"] = True
-        self.appvm.features["gui"] = False
-        self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
         self.assertEqual(self.appvm.get_feat_preload(), [])
         orig_getitem = self.app.domains.__getitem__
         with mock.patch.object(
@@ -187,9 +212,6 @@ def test_010_dvm_preload_get_list(
                     )
 
     def test_010_dvm_preload_can(self):
-        self.appvm.features["qrexec"] = True
-        self.appvm.features["gui"] = False
-        self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
         self.assertFalse(self.appvm.can_preload())
         self.appvm.features["preload-dispvm-max"] = 1
         cases = [
@@ -206,3 +228,25 @@ def test_010_dvm_preload_can(self):
                 self.appvm.features["preload-dispvm-max"] = preload_max
                 self.appvm.features["preload-dispvm"] = preload_list
                 self.assertEqual(self.appvm.can_preload(), expected_value)
+
+    @mock.patch(
+        "qubes.vm.mix.dvmtemplate.DVMTemplateMixin.remove_preload_excess"
+    )
+    def test_011_dvm_preload_del_max(self, mock_remove_preload_excess):
+        self.appvm.features["preload-dispvm-max"] = ""
+        self.adminvm.features["preload-dispvm-max"] = ""
+        del self.appvm.features["preload-dispvm-max"]
+        mock_remove_preload_excess.assert_not_called()
+
+        del self.adminvm.features["preload-dispvm-max"]
+        self.appvm.features["preload-dispvm-max"] = ""
+        del self.appvm.features["preload-dispvm-max"]
+        mock_remove_preload_excess.assert_called_once_with(0)
+
+    @mock.patch("qubes.events.Emitter.fire_event_async")
+    def test_012_dvm_preload_set_max(self, mock_events):
+        mock_events.side_effect = self.mock_coro
+        self.appvm.features["preload-dispvm-max"] = "1"
+        mock_events.assert_called_once_with(
+            "domain-preload-dispvm-start", reason=mock.ANY
+        )
diff --git a/qubes/vm/adminvm.py b/qubes/vm/adminvm.py
index e3005ce0f..ff933346e 100644
--- a/qubes/vm/adminvm.py
+++ b/qubes/vm/adminvm.py
@@ -20,7 +20,7 @@
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 #
 
-""" This module contains the AdminVM implementation """
+"""This module contains the AdminVM implementation"""
 import asyncio
 import grp
 import subprocess
@@ -266,7 +266,7 @@ async def run_service(
         filter_esc=False,
         autostart=False,
         gui=False,
-        **kwargs
+        **kwargs,
     ):
         """Run service on this VM
 
@@ -347,3 +347,43 @@ async def run_service_for_stdio(self, *args, input=None, **kwargs):
             )
 
         return stdouterr
+
+    @qubes.events.handler("domain-feature-delete:preload-dispvm-max")
+    def on_feature_delete_preload_dispvm_max(
+        self, event, feature
+    ):  # pylint: disable=unused-argument
+        if not (appvm := getattr(self.app, "default_dispvm", None)):
+            return
+        reason = "global feature was deleted"
+        asyncio.ensure_future(
+            appvm.fire_event_async("domain-preload-dispvm-start", reason=reason)
+        )
+
+    @qubes.events.handler("domain-feature-pre-set:preload-dispvm-max")
+    def on_feature_pre_set_preload_dispvm_max(
+        self, event, feature, value, oldvalue=None
+    ):  # pylint: disable=unused-argument
+        if value == oldvalue:
+            return
+        if not (appvm := getattr(self.app, "default_dispvm", None)):
+            return
+        appvm.fire_event(
+            "domain-feature-pre-set:preload-dispvm-max",
+            pre_event=True,
+            feature="preload-dispvm-max",
+            value=value,
+            oldvalue=oldvalue,
+        )
+
+    @qubes.events.handler("domain-feature-set:preload-dispvm-max")
+    def on_feature_set_preload_dispvm_max(
+        self, event, feature, value, oldvalue=None
+    ):  # pylint: disable=unused-argument
+        if value == oldvalue:
+            return
+        if not (appvm := getattr(self.app, "default_dispvm", None)):
+            return
+        reason = "global feature was set to " + str(value)
+        asyncio.ensure_future(
+            appvm.fire_event_async("domain-preload-dispvm-start", reason=reason)
+        )
diff --git a/qubes/vm/dispvm.py b/qubes/vm/dispvm.py
index b37a36c65..3f162e72d 100644
--- a/qubes/vm/dispvm.py
+++ b/qubes/vm/dispvm.py
@@ -340,7 +340,7 @@ async def on_domain_started_dispvm(
         try:
             self.log.info(
                 "Preload startup waiting '%s' with '%d' seconds timeout",
-                service,
+                rpc,
                 timeout,
             )
             await asyncio.wait_for(
@@ -389,9 +389,7 @@ def on_domain_unpaused(
             self.use_preload()
 
     @qubes.events.handler("domain-shutdown")
-    async def on_domain_shutdown(
-        self, _event, **_kwargs
-    ):
+    async def on_domain_shutdown(self, _event, **_kwargs):
         """Do auto cleanup if enabled"""
         await self._auto_cleanup()
 
@@ -456,7 +454,9 @@ async def from_appvm(cls, appvm, preload=False, **kwargs):
             # Not necessary to await for this event as its intent is to fill
             # gaps and not relevant for this run.
             asyncio.ensure_future(
-                appvm.fire_event_async("domain-preload-dispvm-start")
+                appvm.fire_event_async(
+                    "domain-preload-dispvm-start", reason="there is a gap"
+                )
             )
 
         if not preload and (preload_dispvm := appvm.get_feat_preload()):
diff --git a/qubes/vm/mix/dvmtemplate.py b/qubes/vm/mix/dvmtemplate.py
index 3e574743e..20361514d 100644
--- a/qubes/vm/mix/dvmtemplate.py
+++ b/qubes/vm/mix/dvmtemplate.py
@@ -91,6 +91,8 @@ def on_domain_loaded(self, event):  # pylint: disable=unused-argument
     def on_feature_delete_preload_dispvm_max(
         self, event, feature
     ):  # pylint: disable=unused-argument
+        if self.is_global_preload_set():
+            return
         self.remove_preload_excess(0)
 
     @qubes.events.handler("domain-feature-pre-set:preload-dispvm-max")
@@ -117,8 +119,13 @@ def on_feature_pre_set_preload_dispvm_max(
     def on_feature_set_preload_dispvm_max(
         self, event, feature, value, oldvalue=None
     ):  # pylint: disable=unused-argument
+        if value == oldvalue:
+            return
+        if self.is_global_preload_set():
+            return
+        reason = "local feature was set to " + str(value)
         asyncio.ensure_future(
-            self.fire_event_async("domain-preload-dispvm-start")
+            self.fire_event_async("domain-preload-dispvm-start", reason=reason)
         )
 
     @qubes.events.handler("domain-feature-pre-set:preload-dispvm")
@@ -237,6 +244,8 @@ async def on_domain_preload_dispvm_used(
         event_log = "Received preload event '%s'" % str(event)
         if event == "used":
             event_log += " for dispvm '%s'" % str(kwargs.get("dispvm"))
+        if "reason" in kwargs:
+            event_log += " because %s" % str(kwargs.get("reason"))
         self.log.info(event_log)
 
         if event == "autostart":
@@ -259,6 +268,7 @@ async def on_domain_preload_dispvm_used(
                 available_memory = int(file.read())
         except FileNotFoundError:
             can_preload = want_preload
+            self.log.warning("File containing available memory was not found")
         if available_memory is not None:
             memory = getattr(self, "memory", 0) * 1024 * 1024
             unrestricted_preload = int(available_memory / memory)
@@ -271,7 +281,7 @@ async def on_domain_preload_dispvm_used(
                 )
             if can_preload == 0:
                 # The gap is filled when consuming a preloaded qube or
-                # requesting a disposable.
+                # requesting a non-preloaded disposable.
                 return
 
         self.log.info("Preloading '%d' qube(s)", can_preload)
@@ -283,17 +293,57 @@ async def on_domain_preload_dispvm_used(
 
     def get_feat_preload(self) -> list[str]:
         """Get the ``preload-dispvm`` feature as a list."""
-        feature = "preload-dispvm"
         assert isinstance(self, qubes.vm.BaseVM)
+        feature = "preload-dispvm"
         value = self.features.get(feature, "")
         return value.split(" ") if value else []
 
-    def get_feat_preload_max(self) -> int:
-        """Get the ``preload-dispvm-max`` feature as an integer."""
+    def get_feat_global_preload_max(self) -> Optional[int]:
+        """Get the global ``preload-dispvm-max`` feature as an integer if it is
+        set, None otherwise."""
+        assert isinstance(self, qubes.vm.BaseVM)
+        feature = "preload-dispvm-max"
+        value = None
+        global_features = self.app.domains["dom0"].features
+        if feature in global_features:
+            value = int(global_features.get(feature) or 0)
+        return value
+
+    def get_feat_preload_max(self, force_local=False) -> int:
+        """Get the ``preload-dispvm-max`` feature as an integer.
+
+        :param force_local: ignore global setting.
+        """
+        assert isinstance(self, qubes.vm.BaseVM)
         feature = "preload-dispvm-max"
+        value = None
+        if not force_local and self == getattr(
+            self.app, "default_dispvm", None
+        ):
+            value = self.get_feat_global_preload_max()
+        if value is None:
+            value = self.features.get(feature)
+        return int(value or 0)
+
+    def is_global_preload_set(self) -> bool:
+        """Returns ``True`` if this qube is the global default_dispvm and the
+        global preload feature is set."""
         assert isinstance(self, qubes.vm.BaseVM)
-        value = self.features.get(feature, 0)
-        return int(value) if value else 0
+        if (
+            self == getattr(self.app, "default_dispvm", None)
+            and "preload-dispvm-max" in self.app.domains["dom0"].features
+        ):
+            return True
+        return False
+
+    def is_global_preload_distinct(self):
+        """Returns ``True`` if global preload feature is distinct compared to
+        local one."""
+        if (
+            self.get_feat_global_preload_max() or 0
+        ) != self.get_feat_preload_max(force_local=True):
+            return True
+        return False
 
     def can_preload(self) -> bool:
         """Returns ``True`` if there is preload vacancy."""