perf: query updated input files in parallel (snakemake#3266)

johanneskoester · web-flow · commit bc4fceef0bbe · 2025-01-23T15:15:00.000+01:00
&lt;!--Add a description of your PR here--&gt;

### QC
&lt;!-- Make sure that you can tick the boxes below. --&gt;

* [x] The PR contains a test case for the changes or the changes are
already covered by an existing test case.
* [x] The documentation (`docs/`) is updated to reflect the changes or
this is not necessary (e.g. if the change does neither modify the
language nor the behavior or functionalities of Snakemake).


&lt;!-- This is an auto-generated comment: release notes by coderabbit.ai
--&gt;
## Summary by CodeRabbit

- **Refactor**
- Enhanced clarity of comments in the `update_needrun` method regarding
input checksums.
- Improved handling of asynchronous operations related to input file
checks.
- Adjusted overall structure of the `update_needrun` method for better
readability and maintainability.
&lt;!-- end of auto-generated comment: release notes by coderabbit.ai --&gt;
diff --git a/snakemake/dag.py b/snakemake/dag.py
@@ -1226,7 +1226,8 @@ async def is_same_checksum(f, job):
                     # no chance to compute checksum, cannot be assumed the same
                     is_same = False
                 else:
-                    # obtain the input checksums for the given file for all output files of the job
+                    # obtain the set of input checksums for the given file for all
+                    # output files of the job
                     checksums = self.workflow.persistence.input_checksums(job, f)
                     if len(checksums) > 1:
                         # more than one checksum recorded, cannot be all the same
@@ -1287,19 +1288,21 @@ async def update_needrun(job):
                     )
             if not reason:
                 output_mintime_ = output_mintime.get(job)
-                updated_input = None
+                reason.updated_input.clear()
                 if output_mintime_:
-                    # Input is updated if it is newer that the oldest output file
+                    # Input is updated if it is newer than the oldest output file
                     # and does not have the same checksum as the one previously recorded.
-                    updated_input = [
-                        f
-                        for f in job.input
-                        if await f.exists()
-                        and await f.is_newer(output_mintime_)
-                        and not await is_same_checksum(f, job)
-                    ]
-                    reason.updated_input.update(updated_input)
-                if not updated_input:
+                    async def updated_input():
+                        for f in job.input:
+                            if (
+                                await f.exists()
+                                and await f.is_newer(output_mintime_)
+                                and not await is_same_checksum(f, job)
+                            ):
+                                yield f
+
+                    reason.updated_input.update([f async for f in updated_input()])
+                if not reason.updated_input:
                     reason.unfinished_queue_input = job.has_unfinished_queue_input()
                     if not reason.unfinished_queue_input:
                         # check for other changes like parameters, set of input files, or code
