fix: always recalculate job resources before job is scheduled as inpu…

…t might have changed or not have been present initially (#1552)

snakemake/jobs.py

@@ -152,6 +152,7 @@ class Job(AbstractJob):
         "_group",
         "targetfile",
         "incomplete_input_expand",
+        "_params_and_resources_resetted",
     ]
 
     def __init__(
@@ -203,6 +204,7 @@ def __init__(
         self.shadow_dir = None
         self._inputsize = None
         self.is_updated = False
+        self._params_and_resources_resetted = False
 
         self._attempt = self.dag.workflow.attempt
 
@@ -340,8 +342,10 @@ def resources(self):
         return self._resources
 
     def reset_params_and_resources(self):
-        self._resources = None
-        self._params = None
+        if not self._params_and_resources_resetted:
+            self._resources = None
+            self._params = None
+            self._params_and_resources_resetted = True
 
     @property
     def conda_env_spec(self):

snakemake/scheduler.py

@@ -471,6 +471,12 @@ def schedule(self):
                 if self.dryrun:
                     run = needrun
                 else:
+                    # Reset params and resources because they might still contain TBDs
+                    # or old values from before files have been regenerated.
+                    # Now, they can be recalculated as all input is present and up to date.
+                    for job in needrun:
+                        job.reset_params_and_resources()
+
                     logger.debug(
                         "Resources before job selection: {}".format(self.resources)
                     )
@@ -497,11 +503,6 @@ def schedule(self):
                     # remove from ready_jobs
                     self.dag.register_running(run)
 
-                # reset params and resources because they might contain TBDs
-                if not self.dryrun:
-                    for job in run:
-                        job.reset_params_and_resources()
-
                 # actually run jobs
                 local_runjobs = [job for job in run if job.is_local]
                 runjobs = [job for job in run if not job.is_local]

tests/test_github_issue1550/Snakefile

@@ -0,0 +1,13 @@
+# Snakefile
+rule all:
+    input:
+        "a.out",
+
+
+rule a:
+    output:
+        "a.out",
+    resources:
+        mem_mb=2000,  # Note that resources are specified here, but not in all.
+    shell:
+        "touch {output}"

tests/tests.py

@@ -1591,6 +1591,18 @@ def test_github_issue1542():
     run(dpath("test_github_issue1542"), dryrun=True)
 
 
+def test_github_issue1550():
+    from snakemake.resources import DefaultResources
+
+    run(
+        dpath("test_github_issue1550"),
+        resources={"mem_mb": 4000},
+        default_resources=DefaultResources(
+            ["mem_mb=max(2*input.size, 1000)", "disk_mb=max(2*input.size, 1000)"]
+        ),
+    )
+
+
 def test_github_issue1498():
     run(dpath("test_github_issue1498"))