Add caching support for CWL (#5187)

* Add cachedir argument and make sure directories exist

* Add a test

* Do a copy when not bypassing filestore and when caching is enabled

* Fix test

* Add cache test

* Bump mypy from 1.13.0 to 1.14.1 (#5193)

* Bump mypy from 1.13.0 to 1.14.1

Bumps [mypy](https://github.com/python/mypy) from 1.13.0 to 1.14.1.
- [Changelog](https://github.com/python/mypy/blob/master/CHANGELOG.md)
- [Commits](python/mypy@v1.13.0...v1.14.1)

---
updated-dependencies:
- dependency-name: mypy
  dependency-type: direct:development
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

* type stubs: upgrade for newer stricter mypy 1.14

https://typing.readthedocs.io/en/latest/spec/enums.html#defining-members

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Michael R. Crusoe <michael.crusoe@gmail.com>

* Update cwltool

* Add cachedir argument and make sure directories exist

* Add a test

* Do a copy when not bypassing filestore and when caching is enabled

* Fix test

* Add cache test

* Update options documentation and enable bypassing file store when CWL task caching is enabled

* Fix CWL cachedir test

* Remove dead code and add clarifications

* Apply suggestions from code review

Co-authored-by: Adam Novak <anovak@soe.ucsc.edu>

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Michael R. Crusoe <michael.crusoe@gmail.com>
Co-authored-by: Adam Novak <anovak@soe.ucsc.edu>

Author: 4 people

docs/running/cliOptions.rst

@@ -300,10 +300,9 @@ Allows configuring Toil's data storage.
                         copies the files into the output directory. Applies to
                         filesystem-based job stores only. (Default=False)
   --caching BOOL        
-                        Set caching options. This must be set to "false"
-                        to use a batch system that does not support
-                        cleanup. Set to "true" if caching
-                        is desired.
+                        Enable or disable worker level file caching. Set to "true" if
+                        caching is desired. By default, caching is enabled on supported
+                        batch systems. Does not affect CWL or WDL task caching.
   --symlinkJobStoreReads BOOL
                         Allow reads and container mounts from a JobStore's
                         shared filesystem directly via symlink. Can be turned

src/toil/cwl/cwltoil.py

@@ -3599,6 +3599,7 @@ def fill_in_files(
         """
         Given a mapping of filenames to Toil file IDs, replace the filename with the file IDs throughout the CWL object.
         """
+
         def fill_in_file(filename: str) -> FileID:
             """
             Return the file name's associated Toil file ID
@@ -4261,8 +4262,18 @@ def main(args: Optional[list[str]] = None, stdout: TextIO = sys.stdout) -> int:
     runtime_context.move_outputs = "leave"
     runtime_context.rm_tmpdir = False
     runtime_context.streaming_allowed = not options.disable_streaming
+    if options.cachedir is not None:
+        runtime_context.cachedir = os.path.abspath(options.cachedir)
+        # Automatically bypass the file store to be compatible with cwltool caching
+        # Otherwise, the CWL caching code makes links to temporary local copies
+        # of filestore files and caches those.
+        logger.debug("CWL task caching is turned on. Bypassing file store.")
+        options.bypass_file_store = True
     if options.mpi_config_file is not None:
         runtime_context.mpi_config = MpiConfig.load(options.mpi_config_file)
+    if cwltool.main.check_working_directories(runtime_context) is not None:
+        logger.error("Failed to create directory. If using tmpdir_prefix, tmpdir_outdir_prefix, or cachedir, consider changing directory locations.")
+        return 1
     setattr(runtime_context, "bypass_file_store", options.bypass_file_store)
     if options.bypass_file_store and options.destBucket:
         # We use the file store to write to buckets, so we can't do this (yet?)

src/toil/options/common.py

@@ -466,7 +466,8 @@ def is_within(x: Union[int, float]) -> bool:
     )
 
     caching = file_store_options.add_mutually_exclusive_group()
-    caching_help = "Enable or disable caching for your workflow, specifying this overrides default from job store"
+    caching_help = ("Enable or disable worker level file caching for your workflow, specifying this overrides default from batch system. "
+                    "Does not affect CWL or WDL task caching.")
     caching.add_argument(
         "--caching",
         dest="caching",

src/toil/options/cwl.py

@@ -419,3 +419,13 @@ def add_cwl_options(parser: ArgumentParser, suppress: bool = True) -> None:
         type=str,
         help=suppress_help or "Specify a cloud bucket endpoint for output files.",
     )
+    parser.add_argument(
+        "--cachedir",
+        type=str,
+        help=suppress_help
+        or "Directory to cache intermediate workflow outputs to avoid "
+        "recomputing steps. Can be very helpful in the development and "
+        "troubleshooting of CWL documents. This automatically bypasses the file store."
+        " Not to be confused with --caching.",
+        dest="cachedir"
+    )

src/toil/test/cwl/cwlTest.py

@@ -703,6 +703,50 @@ def path_with_bogus_rev() -> str:
         except subprocess.CalledProcessError:
             pass
 
+    def test_caching(self) -> None:
+        log.info("Running CWL caching test.")
+        from toil.cwl import cwltoil
+
+        outDir = self._createTempDir()
+        cacheDir = self._createTempDir()
+
+        cwlDir = os.path.join(self._projectRootPath(), "src", "toil", "test", "cwl")
+        log_path = os.path.join(outDir, "log")
+        cmd = [
+            "--outdir",
+            outDir,
+            "--jobStore",
+            os.path.join(outDir, "jobStore"),
+            "--clean=always",
+            "--no-container",
+            "--cachedir",
+            cacheDir,
+            os.path.join(cwlDir, "revsort.cwl"),
+            os.path.join(cwlDir, "revsort-job.json"),
+        ]
+        st = StringIO()
+        ret = cwltoil.main(cmd, stdout=st)
+        assert ret == 0
+        # cwltool hashes certain steps into directories, ensure it exists
+        # since cwltool caches per task and revsort has 2 cwl tasks, there should be 2 directories and 2 status files
+        assert (len(os.listdir(cacheDir)) == 4)
+
+        # Rerun the workflow to ensure there is a cache hit and that we don't rerun the tools
+        st = StringIO()
+        cmd = [
+                  "--writeLogsFromAllJobs=True",
+                  "--writeLogs",
+                  log_path
+              ] + cmd
+        ret = cwltoil.main(cmd, stdout=st)
+        assert ret == 0
+
+        # Ensure all of the worker logs are using their cached outputs
+        for file in os.listdir(log_path):
+            assert "Using cached output" in open(os.path.join(log_path, file), encoding="utf-8").read()
+
+
+
     @needs_aws_s3
     def test_streamable(self, extra_args: Optional[list[str]] = None) -> None:
         """
@@ -1203,6 +1247,14 @@ def test_run_conformance_with_caching(self) -> None:
             junit_file=os.path.join(self.rootDir, "caching-conformance-1.2.junit.xml"),
         )
 
+    @slow
+    @pytest.mark.timeout(CONFORMANCE_TEST_TIMEOUT)
+    def test_run_conformance_with_task_caching(self) -> None:
+        self.test_run_conformance(
+            junit_file=os.path.join(self.rootDir, "task-caching-conformance-1.2.junit.xml"),
+            extra_args=["--cachedir", self._createTempDir("task_cache")]
+        )
+
     @slow
     @pytest.mark.timeout(CONFORMANCE_TEST_TIMEOUT)
     def test_run_conformance_with_in_place_update(self) -> None: