fix: 3342 faster touch runs and warning messages for non-existing files (#3398)

Marco-Masera · johanneskoester · web-flow · commit cd9c3c350b20 · 2025-03-14T10:43:40.000+01:00
### Description

This PR tries to optimize the --touch runs but running the *sleep*
command only for jobs that actually need to touch files.
The *sleep* command is moved into the async *touch* function and run
only if there are files to be created.

Sleeping inside this async should block the entire asyncio event loop,
which is the desired behavior here as the order of files creation is
preserved.

For jobs with non-existing files, a warning messsage is displayed to the
user. A new *log_warning* function is added in the Job class.


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

- **New Features**
- Introduced a configurable pause during file processing to enhance
operational flexibility.
  
- **Refactor**
- Streamlined the process for checking and processing output files,
ensuring clearer differentiation between file types.
- Improved logging to notify when certain expected files are not
present.
&lt;!-- end of auto-generated comment: release notes by coderabbit.ai --&gt;

---------

Co-authored-by: Johannes Köster &lt;johannes.koester@tu-dortmund.de&gt;
diff --git a/src/snakemake/executors/touch.py b/src/snakemake/executors/touch.py
@@ -15,7 +15,7 @@
 from snakemake_interface_executor_plugins.executors.base import SubmittedJobInfo
 from snakemake_interface_executor_plugins.settings import CommonSettings
 from snakemake.common import async_run
-
+from snakemake.logging import logger
 from snakemake.exceptions import print_exception
 
 
@@ -30,22 +30,43 @@
 
 
 class Executor(RealExecutor):
+    SLEEPING_TIME = 0.1
+
     def run_job(
         self,
         job: JobExecutorInterface,
     ):
         job_info = SubmittedJobInfo(job=job)
         try:
-            time.sleep(0.1)
 
             if job.output:
 
                 async def touch():
-                    for f in job.output:
-                        if f.is_storage and await f.exists_in_storage():
+                    touch_storage_and_local_files = {
+                        f
+                        for f in job.output
+                        if f.is_storage and await f.exists_in_storage()
+                    }
+                    touch_files = {
+                        f
+                        for f in job.output
+                        if f not in touch_storage_and_local_files
+                        and await f.exists_local()
+                    }
+                    non_existing_files = (
+                        set(job.output) - touch_storage_and_local_files - touch_files
+                    )
+
+                    if touch_files or touch_storage_and_local_files:
+                        time.sleep(Executor.SLEEPING_TIME)
+                        for f in touch_storage_and_local_files:
                             await f.touch_storage_and_local()
-                        elif await f.exists_local():
+                        for f in touch_files:
                             f.touch()
+                    if len(non_existing_files) > 0:
+                        logger.warning(
+                            f"Output files not touched because they don't exist: {', '.join(non_existing_files)}"
+                        )
 
                 async_run(touch())
 
