Add available memory threshold to preload event

.pylintrc

@@ -58,10 +58,10 @@ const-rgx=(([A-Za-z_][A-Za-z0-9_]*)|(__.*__))$
 class-rgx=([A-Z_][a-zA-Z0-9]+|TC_\d\d_[a-zA-Z0-9_]+)$
 
 # Regular expression which should only match correct function names
-function-rgx=(test_[0-9]{3}_[a-z0-9_]{2,50}|[a-z_][a-z0-9_]{2,40})$
+function-rgx=(test_[0-9]{3}_[a-z0-9_]{2,50}|[a-z_][a-z0-9_]{2,50})$
 
 # Regular expression which should only match correct method names
-method-rgx=(test_[0-9]{3}_[a-z0-9_]{2,50}|[a-z_][a-z0-9_]{2,40})$
+method-rgx=(test_[0-9]{3}_[a-z0-9_]{2,50}|[a-z_][a-z0-9_]{2,50})$
 
 # Regular expression which should only match correct instance attribute names
 attr-rgx=[a-z_][a-z0-9_]{2,40}$

qubes/tests/integ/dispvm.py

@@ -498,20 +498,38 @@ async def _test_012_preload_low_mem(self):
         # pylint: disable=unspecified-encoding
         logger.info("start")
         unpatched_open = open
+        memory = int(getattr(self.disp_base, "memory", 0) * 1024**2)
 
         def mock_open_mem(file, *args, **kwargs):
             if file == qubes.config.qmemman_avail_mem_file:
-                memory = str(getattr(self.disp_base, "memory", 0) * 1024 * 1024)
-                return mock_open(read_data=memory)()
+                return mock_open(read_data=str(memory))()
             return unpatched_open(file, *args, **kwargs)
 
+        def mock_open_mem_threshold(file, *args, **kwargs):
+            if file == qubes.config.qmemman_avail_mem_file:
+                return mock_open(read_data=str(memory * 2))()
+            return unpatched_open(file, *args, **kwargs)
+
+        preload_max = 2
         with patch("builtins.open", side_effect=mock_open_mem):
-            preload_max = 2
+            logger.info("low mem standard")
+            self.disp_base.features["preload-dispvm-max"] = str(preload_max)
+            await self.wait_preload(
+                preload_max, fail_on_timeout=False, timeout=15
+            )
+            self.assertEqual(1, len(self.disp_base.get_feat_preload()))
+            # Nothing will be done here, just to prepare to the next test.
+            self.disp_base.features["preload-dispvm-max"] = str(preload_max - 1)
+
+        with patch("builtins.open", side_effect=mock_open_mem_threshold):
+            logger.info("low mem threshold")
+            self.adminvm.features["preload-dispvm-threshold"] = memory
             self.disp_base.features["preload-dispvm-max"] = str(preload_max)
             await self.wait_preload(
                 preload_max, fail_on_timeout=False, timeout=15
             )
             self.assertEqual(1, len(self.disp_base.get_feat_preload()))
+
         logger.info("end")
 
     def test_013_preload_gui(self):

qubes/tests/vm/adminvm.py

@@ -279,3 +279,14 @@ def test_801_preload_del_max(self):
             mock_events.assert_called_once_with(
                 "domain-preload-dispvm-start", reason=unittest.mock.ANY
             )
+
+    def test_802_preload_set_threshold(self):
+        cases_valid = ["", "0", "1"]
+        cases_invalid = ["a", "-1", "1 1"]
+        for value in cases_invalid:
+            with self.subTest(value=value):
+                with self.assertRaises(qubes.exc.QubesValueError):
+                    self.vm.features["preload-dispvm-threshold"] = value
+        for value in cases_valid:
+            with self.subTest(value=value):
+                self.vm.features["preload-dispvm-threshold"] = value

qubes/tests/vm/mix/dvmtemplate.py

@@ -250,3 +250,12 @@ def test_012_dvm_preload_set_max(self, mock_events):
         mock_events.assert_called_once_with(
             "domain-preload-dispvm-start", reason=mock.ANY
         )
+
+    def test_013_dvm_preload_get_treshold(self):
+        cases = [None, False, "0", "2", "1000"]
+        self.assertEqual(self.appvm.get_feat_preload_threshold(), 0)
+        for value in cases:
+            with self.subTest(value=value):
+                self.adminvm.features["preload-dispvm-threshold"] = value
+                threshold = self.appvm.get_feat_preload_threshold()
+                self.assertEqual(threshold, int(value or 0) * 1024**2)

qubes/vm/adminvm.py

@@ -348,6 +348,16 @@ async def run_service_for_stdio(self, *args, input=None, **kwargs):
 
         return stdouterr
 
+    @qubes.events.handler("domain-feature-pre-set:preload-dispvm-threshold")
+    def on_feature_pre_set_preload_dispvm_threshold(
+        self, event, feature, value, oldvalue=None
+    ):  # pylint: disable=unused-argument
+        value = value or "0"
+        if not value.isdigit():
+            raise qubes.exc.QubesValueError(
+                "Invalid preload-dispvm-threshold value: not a digit"
+            )
+
     @qubes.events.handler("domain-feature-delete:preload-dispvm-max")
     def on_feature_delete_preload_dispvm_max(
         self, event, feature

qubes/vm/mix/dvmtemplate.py

@@ -282,12 +282,14 @@ async def on_domain_preload_dispvm_used(
         available_memory = None
         try:
             with open(avail_mem_file, "r", encoding="ascii") as file:
-                available_memory = int(file.read())
+                available_memory = max(
+                    0, int(file.read()) - self.get_feat_preload_threshold()
+                )
         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
+            memory = getattr(self, "memory", 0) * 1024**2
             unrestricted_preload = int(available_memory / memory)
             can_preload = min(unrestricted_preload, want_preload)
             if skip_preload := want_preload - can_preload:
@@ -308,6 +310,14 @@ async def on_domain_preload_dispvm_used(
                     qubes.vm.dispvm.DispVM.from_appvm(self, preload=True)
                 )
 
+    def get_feat_preload_threshold(self) -> int:
+        """Get the ``preload-dispvm-threshold`` feature as int (bytes unit)."""
+        assert isinstance(self, qubes.vm.BaseVM)
+        feature = "preload-dispvm-threshold"
+        global_features = self.app.domains["dom0"].features
+        value = int(global_features.get(feature) or 0)
+        return value * 1024**2
+
     def get_feat_preload(self) -> list[str]:
         """Get the ``preload-dispvm`` feature as a list."""
         assert isinstance(self, qubes.vm.BaseVM)