Zipping files for faster transfer

Author: kks32

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## v0.5.4
+
+### New features
+
+- **Input bundling for SimCenter apps** (`ds.jobs.prepare_inputs`): the Tapis transfers service pays a fixed per-file scheduling cost (measured ~40s/file under tenant load), so staging many small input files dominates job wall time. For `simcenter-*` apps, `prepare_inputs` now zips the prepared `tmp.SimCenter/` tree into a single `tmpSimCenter.zip` — which the app wrapper unpacks natively on the execution system — inside a staging directory (`<input_dir>_staged`). Staging transfers ~2 objects instead of one per file.
+  - Default **on** for SimCenter apps (matching what quoFEM desktop submits); pass `bundle=False` to stage the loose tree as before. Apps without a profile are unaffected.
+  - Every `prepare_inputs` summary now includes `staged_dir` — the directory to pass to `ds.files.to_uri()`; the original input directory is never modified beyond the workflow-JSON patch.
+
 ## v0.5.3
 
 ### New features

dapi/__init__.py

@@ -52,7 +52,7 @@
     TAPIS_TERMINAL_STATES,
 )
 
-__version__ = "0.5.3"
+__version__ = "0.5.4"
 
 __all__ = [
     "DSClient",

dapi/client.py

@@ -713,19 +713,24 @@ def prepare_inputs(
         Dispatches to the app's profile when one is registered; a no-op
         beyond existence checking for apps that need no preparation. For
         SimCenter apps (``simcenter-*``) this rewrites the workflow JSON's
-        backend paths and reports the UQ engine, random variables, and EDPs.
+        backend paths, reports the UQ engine, random variables, and EDPs,
+        and by default bundles the inputs into a single ``tmpSimCenter.zip``
+        (unpacked natively by the app wrapper) so staging transfers one
+        file instead of many. Stage the returned ``staged_dir``.
 
         Args:
             app_id (str): The Tapis app id the inputs are being prepared for.
             input_dir (str): Local path to the job input directory.
             **options: Profile-specific options (e.g. ``backend_dir``,
-                ``input_filename`` for SimCenter apps).
+                ``input_filename``, ``bundle=False`` for SimCenter apps).
 
         Returns:
-            Dict[str, Any]: Summary of the preparation performed.
+            Dict[str, Any]: Summary of the preparation performed. Always
+                includes ``staged_dir`` — the directory to stage.
 
         Example:
-            >>> ds.jobs.prepare_inputs("simcenter-uq-stampede3", "./DS_input")
+            >>> info = ds.jobs.prepare_inputs("simcenter-uq-stampede3", "./DS_input")
+            >>> input_uri = ds.files.to_uri(info["staged_dir"])
         """
         return jobs_module.prepare_job_inputs(app_id, input_dir, **options)
 

dapi/jobs.py

@@ -535,16 +535,19 @@ def prepare_job_inputs(app_id: str, input_dir: str, **options: Any) -> Dict[str,
 
     For SimCenter apps (``simcenter-*``) this rewrites the workflow JSON's
     ``remoteAppDir``/``remoteAppWorkingDir`` to the backend installation on
-    the execution system and returns a summary of the UQ workflow.
+    the execution system, and by default bundles the inputs into a single
+    ``tmpSimCenter.zip`` (unpacked natively by the app wrapper) so staging
+    transfers one file instead of many. Stage the returned ``staged_dir``.
 
     Args:
         app_id (str): The Tapis app id the inputs are being prepared for.
         input_dir (str): Local path to the job input directory.
         **options: Profile-specific options (e.g. ``backend_dir``,
-            ``input_filename`` for SimCenter apps).
+            ``input_filename``, ``bundle=False`` for SimCenter apps).
 
     Returns:
-        Dict[str, Any]: Summary of the preparation performed.
+        Dict[str, Any]: Summary of the preparation performed. Always
+            includes ``staged_dir`` — the directory callers should stage.
 
     Raises:
         ValueError: If input_dir does not exist, or a profile rejects the
@@ -559,7 +562,12 @@ def prepare_job_inputs(app_id: str, input_dir: str, **options: Any) -> Dict[str,
     profile = profiles.find(app_id)
     if profile is None:
         print(f"App '{app_id}' requires no input preparation.")
-        return {"app_id": app_id, "input_dir": input_dir, "prepared": False}
+        return {
+            "app_id": app_id,
+            "input_dir": input_dir,
+            "prepared": False,
+            "staged_dir": input_dir,
+        }
     print(f"Preparing inputs with '{profile.name}' app profile.")
     return profile.prepare_inputs(input_dir, app_id=app_id, **options)
 

dapi/profiles.py

@@ -53,9 +53,15 @@ def prepare_inputs(
             **options: Profile-specific options.
 
         Returns:
-            Dict[str, Any]: Summary of what was prepared.
+            Dict[str, Any]: Summary of what was prepared. Always includes
+                ``staged_dir`` — the directory callers should stage.
         """
-        return {"app_id": app_id, "input_dir": input_dir, "prepared": False}
+        return {
+            "app_id": app_id,
+            "input_dir": input_dir,
+            "prepared": False,
+            "staged_dir": input_dir,
+        }
 
     @classmethod
     def finalize_job_request(cls, job_request: Dict[str, Any]) -> Dict[str, Any]:

dapi/simcenter.py

@@ -26,6 +26,7 @@
 import io
 import json
 import os
+import shutil
 import zipfile
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional, Union
@@ -40,6 +41,10 @@
 # The env variable name the wrapper script expects for the staged input dir.
 INPUT_DIR_ENV_KEY = "inputDirectory"
 
+# Archive name the SimCenter wrapper script unpacks natively on the exec
+# system ("unzip tmpSimCenter.zip" is its first action in the input dir).
+BUNDLE_NAME = "tmpSimCenter.zip"
+
 # SimCenter backend installations per app id. These paths change when
 # SimCenter redeploys the backend; update alongside a dapi release.
 DEFAULT_BACKEND_DIRS: Dict[str, str] = {
@@ -54,38 +59,55 @@ def prepare_inputs(
     input_filename: str = "scInput.json",
     backend_dir: Optional[str] = None,
     app_id: str = "simcenter-uq-stampede3",
+    bundle: bool = True,
+    staged_dir: Optional[str] = None,
 ) -> Dict[str, Any]:
-    """Point a SimCenter workflow JSON at the backend on the execution system.
+    """Point a SimCenter workflow JSON at the backend and bundle the inputs.
 
     The SimCenter desktop applications write local paths into the workflow
     JSON (``remoteAppDir``/``remoteAppWorkingDir``). Before submission these
     must be rewritten to the backend installation on the execution system,
     which the app wrapper reads back with ``jq`` to locate the UQ engine.
 
+    By default the prepared ``tmp.SimCenter/`` tree is then zipped into a
+    single ``tmpSimCenter.zip`` inside a staging directory. The SimCenter
+    wrapper unpacks this archive natively on the execution system, and
+    staging one file instead of many avoids the Tapis transfers service's
+    per-file overhead (which can add ~40s per file under tenant load).
+    Stage the returned ``staged_dir``, not ``input_dir``, when bundling.
+
     Args:
         input_dir (str): Local path to the job input directory (the folder
-            containing ``tmp.SimCenter/templatedir/``).
+            containing ``tmp.SimCenter/templatedir/``). Never modified
+            beyond the workflow JSON patch.
         input_filename (str, optional): Workflow JSON filename inside
             ``tmp.SimCenter/templatedir/``. Defaults to "scInput.json".
         backend_dir (str, optional): SimCenter backend installation path on
             the execution system. If None, looked up from
             ``DEFAULT_BACKEND_DIRS`` by app_id.
         app_id (str, optional): Tapis app id used for the backend lookup.
             Defaults to "simcenter-uq-stampede3".
+        bundle (bool, optional): Zip ``tmp.SimCenter/`` into
+            ``tmpSimCenter.zip`` in a staging directory. Defaults to True.
+            Pass False to stage the loose directory tree as before.
+        staged_dir (str, optional): Where to place the bundle. Defaults to
+            ``<input_dir>_staged`` next to the input directory.
 
     Returns:
         Dict[str, Any]: Summary of the prepared workflow with keys
             ``workflow_json`` (path), ``backend_dir``, ``uq_engine``,
-            ``uq_type``, ``random_variables``, and ``edps``.
+            ``uq_type``, ``random_variables``, ``edps``, and ``staged_dir``
+            (the directory to stage: the bundle directory when bundling,
+            otherwise input_dir). When bundling, also ``bundle`` (zip path)
+            and ``bundled_files`` (count).
 
     Raises:
         ValueError: If no backend directory is known for app_id and none given.
         FileNotFoundError: If the workflow JSON cannot be found.
 
     Example:
         >>> info = ds.jobs.prepare_inputs("simcenter-uq-stampede3", "./DS_input")
-        >>> info["random_variables"]
-        ['Dr', 'G0', 'hpo']
+        >>> input_uri = ds.files.to_uri(info["staged_dir"])
     """
     if backend_dir is None:
         backend_dir = DEFAULT_BACKEND_DIRS.get(app_id)
@@ -123,14 +145,54 @@ def prepare_inputs(
         "uq_type": uq.get("uqType"),
         "random_variables": [rv["name"] for rv in workflow.get("randomVariables", [])],
         "edps": [edp["name"] for edp in workflow.get("EDP", [])],
+        "staged_dir": input_dir,
     }
     print(f"Updated remoteAppDir in {workflow_path}")
     print(f"UQ engine: {summary['uq_engine']} ({summary['uq_type']})")
     print(f"Random variables: {summary['random_variables']}")
     print(f"EDPs: {summary['edps']}")
+
+    if bundle:
+        summary.update(_bundle_inputs(input_dir, staged_dir))
     return summary
 
 
+def _bundle_inputs(input_dir: str, staged_dir: Optional[str]) -> Dict[str, Any]:
+    """Zip ``tmp.SimCenter/`` into ``tmpSimCenter.zip`` in a staging dir.
+
+    The archive preserves the ``tmp.SimCenter/...`` layout the wrapper
+    expects when it runs ``unzip tmpSimCenter.zip``. Any other top-level
+    entries of input_dir are copied into the staging dir unchanged.
+    """
+    staged = staged_dir or input_dir.rstrip("/") + "_staged"
+    os.makedirs(staged, exist_ok=True)
+
+    bundle_src = os.path.join(input_dir, "tmp.SimCenter")
+    zip_path = os.path.join(staged, BUNDLE_NAME)
+    bundled_files = 0
+    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        for root, _dirs, files in os.walk(bundle_src):
+            for name in files:
+                full_path = os.path.join(root, name)
+                zf.write(full_path, os.path.relpath(full_path, input_dir))
+                bundled_files += 1
+
+    # Preserve any sibling entries of tmp.SimCenter/ the app may expect.
+    for entry in os.listdir(input_dir):
+        if entry == "tmp.SimCenter":
+            continue
+        src = os.path.join(input_dir, entry)
+        dst = os.path.join(staged, entry)
+        if os.path.isfile(src):
+            shutil.copy2(src, dst)
+        elif os.path.isdir(src) and os.path.abspath(src) != os.path.abspath(staged):
+            shutil.copytree(src, dst, dirs_exist_ok=True)
+
+    print(f"Bundled {bundled_files} files into {zip_path}")
+    print(f"Stage this directory: {staged}")
+    return {"staged_dir": staged, "bundle": zip_path, "bundled_files": bundled_files}
+
+
 def finalize_job_request(job_request: Dict[str, Any]) -> Dict[str, Any]:
     """Apply the SimCenter wrapper contract to a generated job request.