Add support for concurrency via threads to fs mode

docs/sections/user_guide/cli/tools/fs/copy-help.out

@@ -1,6 +1,7 @@
 usage: uw fs copy [-h] [--version] [--config-file PATH] [--target-dir PATH]
                   [--cycle CYCLE] [--leadtime LEADTIME] [--dry-run]
-                  [--key-path KEY[.KEY...]] [--report] [--quiet] [--verbose]
+                  [--threads NUM] [--key-path KEY[.KEY...]] [--report]
+                  [--quiet] [--verbose]
 
 Copy files
 
@@ -19,6 +20,8 @@ Optional arguments:
       The leadtime as hours[:minutes[:seconds]]
   --dry-run
       Only log info, making no changes
+  --threads NUM, -n NUM
+      Number of concurrent threads to use (default: 1)
   --key-path KEY[.KEY...]
       Dot-separated path of keys to config block to use
   --report

docs/sections/user_guide/cli/tools/fs/hardlink-help.out

@@ -1,6 +1,6 @@
 usage: uw fs hardlink [-h] [--version] [--config-file PATH]
                       [--target-dir PATH] [--cycle CYCLE]
-                      [--leadtime LEADTIME] [--dry-run]
+                      [--leadtime LEADTIME] [--dry-run] [--threads NUM]
                       [--key-path KEY[.KEY...]] [--report] [--quiet]
                       [--verbose] [--fallback {copy,symlink}]
 
@@ -21,6 +21,8 @@ Optional arguments:
       The leadtime as hours[:minutes[:seconds]]
   --dry-run
       Only log info, making no changes
+  --threads NUM, -n NUM
+      Number of concurrent threads to use (default: 1)
   --key-path KEY[.KEY...]
       Dot-separated path of keys to config block to use
   --report

docs/sections/user_guide/cli/tools/fs/link-help.out

@@ -1,6 +1,7 @@
 usage: uw fs link [-h] [--version] [--config-file PATH] [--target-dir PATH]
                   [--cycle CYCLE] [--leadtime LEADTIME] [--dry-run]
-                  [--key-path KEY[.KEY...]] [--report] [--quiet] [--verbose]
+                  [--threads NUM] [--key-path KEY[.KEY...]] [--report]
+                  [--quiet] [--verbose]
 
 Create symlinks
 
@@ -19,6 +20,8 @@ Optional arguments:
       The leadtime as hours[:minutes[:seconds]]
   --dry-run
       Only log info, making no changes
+  --threads NUM, -n NUM
+      Number of concurrent threads to use (default: 1)
   --key-path KEY[.KEY...]
       Dot-separated path of keys to config block to use
   --report

docs/sections/user_guide/cli/tools/fs/makedirs-help.out

@@ -1,6 +1,6 @@
 usage: uw fs makedirs [-h] [--version] [--config-file PATH]
                       [--target-dir PATH] [--cycle CYCLE]
-                      [--leadtime LEADTIME] [--dry-run]
+                      [--leadtime LEADTIME] [--dry-run] [--threads NUM]
                       [--key-path KEY[.KEY...]] [--report] [--quiet]
                       [--verbose]
 
@@ -21,6 +21,8 @@ Optional arguments:
       The leadtime as hours[:minutes[:seconds]]
   --dry-run
       Only log info, making no changes
+  --threads NUM, -n NUM
+      Number of concurrent threads to use (default: 1)
   --key-path KEY[.KEY...]
       Dot-separated path of keys to config block to use
   --report

src/uwtools/api/fs.py

@@ -24,6 +24,7 @@ def copy(
     leadtime: dt.timedelta | None = None,
     key_path: list[YAMLKey] | None = None,
     dry_run: bool = False,
+    threads: int = 1,
     stdin_ok: bool = False,
 ) -> dict[str, list[str]]:
     """
@@ -35,6 +36,7 @@ def copy(
     :param leadtime: A timedelta object to make available for use in the config.
     :param key_path: Path of keys to config block to use.
     :param dry_run: Do not copy files.
+    :param threads: Number of concurrent threads to use.
     :param stdin_ok: OK to read from ``stdin``?
     :return: A report on files copied / not copied.
     """
@@ -45,7 +47,7 @@ def copy(
         leadtime=leadtime,
         key_path=key_path,
     )
-    assets = cast(list, stager.go(dry_run=dry_run).asset)
+    assets = cast(list, stager.go(dry_run=dry_run, threads=threads).asset)
     ready = lambda state: [str(asset.ref) for asset in assets if asset.ready() is state]
     return {STR.ready: ready(True), STR.notready: ready(False)}
 
@@ -58,6 +60,7 @@ def link(
     leadtime: dt.timedelta | None = None,
     key_path: list[YAMLKey] | None = None,
     dry_run: bool = False,
+    threads: int = 1,
     stdin_ok: bool = False,
     fallback: str | None = None,
 ) -> dict[str, list[str]]:
@@ -75,6 +78,7 @@ def link(
     :param leadtime: A timedelta object to make available for use in the config.
     :param key_path: Path of keys to config block to use.
     :param dry_run: Do not link files.
+    :param threads: Number of concurrent threads to use.
     :param stdin_ok: OK to read from ``stdin``?
     :param fallback: Alternative if hardlink fails (choices: ``copy``, ``symlink``).
     :return: A report on files linked / not linked.
@@ -88,7 +92,7 @@ def link(
         key_path=key_path,
         fallback=fallback,
     )
-    assets = cast(list, stager.go(dry_run=dry_run).asset)
+    assets = cast(list, stager.go(dry_run=dry_run, threads=threads).asset)
     ready = lambda state: [str(asset.ref) for asset in assets if asset.ready() is state]
     return {STR.ready: ready(True), STR.notready: ready(False)}
 
@@ -100,6 +104,7 @@ def makedirs(
     leadtime: dt.timedelta | None = None,
     key_path: list[YAMLKey] | None = None,
     dry_run: bool = False,
+    threads: int = 1,
     stdin_ok: bool = False,
 ) -> dict[str, list[str]]:
     """
@@ -111,6 +116,7 @@ def makedirs(
     :param leadtime: A timedelta object to make available for use in the config.
     :param key_path: Path of keys to config block to use.
     :param dry_run: Do not create directories.
+    :param threads: Number of concurrent threads to use.
     :param stdin_ok: OK to read from ``stdin``?
     :return: A report on directories created / not created.
     """
@@ -121,7 +127,7 @@ def makedirs(
         leadtime=leadtime,
         key_path=key_path,
     )
-    assets = cast(list, stager.go(dry_run=dry_run).asset)
+    assets = cast(list, stager.go(dry_run=dry_run, threads=threads).asset)
     ready = lambda state: [str(asset.ref) for asset in assets if asset.ready() is state]
     return {STR.ready: ready(True), STR.notready: ready(False)}
 

src/uwtools/cli.py

@@ -387,6 +387,7 @@ def _add_subparser_fs_common(parser: Parser) -> tuple[ActionChecks, Group]:
     _add_arg_cycle(optional)
     _add_arg_leadtime(optional)
     _add_arg_dry_run(optional)
+    _add_arg_threads(optional)
     _add_arg_key_path(optional, helpmsg="Dot-separated path of keys to config block to use")
     _add_arg_report(optional)
     return _add_args_verbosity(optional), optional
@@ -465,6 +466,7 @@ def _dispatch_fs_copy(args: Args) -> bool:
         leadtime=args[STR.leadtime],
         key_path=args[STR.keypath],
         dry_run=args[STR.dryrun],
+        threads=args[STR.threads],
         stdin_ok=True,
     )
     return _dispatch_fs_report(report=report if args[STR.report] else None)
@@ -484,6 +486,7 @@ def _dispatch_fs_hardlink(args: Args) -> bool:
         leadtime=args[STR.leadtime],
         key_path=args[STR.keypath],
         dry_run=args[STR.dryrun],
+        threads=args[STR.threads],
         stdin_ok=True,
         fallback=args[STR.fallback],
     )
@@ -503,6 +506,7 @@ def _dispatch_fs_link(args: Args) -> bool:
         leadtime=args[STR.leadtime],
         key_path=args[STR.keypath],
         dry_run=args[STR.dryrun],
+        threads=args[STR.threads],
         stdin_ok=True,
     )
     return _dispatch_fs_report(report=report if args[STR.report] else None)
@@ -521,6 +525,7 @@ def _dispatch_fs_makedirs(args: Args) -> bool:
         leadtime=args[STR.leadtime],
         key_path=args[STR.keypath],
         dry_run=args[STR.dryrun],
+        threads=args[STR.threads],
         stdin_ok=True,
     )
     return _dispatch_fs_report(report=report if args[STR.report] else None)
@@ -1026,6 +1031,19 @@ def _add_arg_task(group: Group, required: bool = True) -> None:
     )
 
 
+def _add_arg_threads(group: Group) -> None:
+    default = 1
+    group.add_argument(
+        _switch(STR.threads),
+        "-n",
+        default=default,
+        help="Number of concurrent threads to use (default: %s)" % default,
+        metavar="NUM",
+        required=False,
+        type=int,
+    )
+
+
 def _add_arg_total(group: Group) -> None:
     group.add_argument(
         _switch(STR.total),

src/uwtools/tests/test_cli.py

@@ -57,6 +57,7 @@ def args_dispatch_fs(utc):
         "leadtime": timedelta(hours=6),
         "key_path": ["a", "b"],
         "dry_run": False,
+        "threads": 3,
         "report": True,
         "stdin_ok": True,
     }
@@ -426,6 +427,7 @@ def test_cli__dispatch_fs_action(action, args_dispatch_fs):
         "leadtime": args_actual["leadtime"],
         "key_path": args_actual["key_path"],
         "dry_run": args_actual["dry_run"],
+        "threads": args_actual["threads"],
         "stdin_ok": args_actual["stdin_ok"],
     }
     if action == "hardlink":