experiments: support local parallel execution in temp directories (#4257

)

* experiments: add initial local tmpdir executor

* fix detached head pull

* experiments: include unchanged (unrepro'd) stages in experiment hash

* experiments: repro single experiment using tmpdir executor by default

* stage experiments as stash commits before running

* include repro args/kwargs in stashed experiments

* support running arbitrary experiment commits (including stash commits)

* experiments: use ProcessPoolExecutor

* experiments: add `dvc exp` alias for `dvc experiments`

* experiments: add `dvc repro --experiment --queue`

* `--queue` can be used to stage an experiment for future execution

* experiments: show queued (unexecuted) experiments in `dvc exp show`

* revert ProcessPoolExecutor change

* experiments: add `dvc repro --run-all [--jobs]`

* `--run-all` can be used to run all queued experiments in parallel

* cleanup output

* experiments: use ProcessPoolExecutor

* fix returning unpicklable objects error

* update tests

* fix windows cleanup issue

* on windows tempdir cannot be removed if we are chdir'd into that
  directory

dvc/command/experiments.py

@@ -48,6 +48,10 @@ def _round(val):
         return val
 
     def _extend(row, names, items):
+        if not items:
+            row.extend(["-"] * len(names))
+            return
+
         for fname, item in items:
             if isinstance(item, dict):
                 item = flatten(item, ".")
@@ -62,13 +66,14 @@ def _extend(row, names, items):
     for i, (rev, exp) in enumerate(experiments.items()):
         row = []
         style = None
+        queued = "*" if exp.get("queued", False) else ""
         if rev == "baseline":
             row.append(f"{base_rev}")
             style = "bold"
         elif i < len(experiments) - 1:
-            row.append(f"├── {rev[:7]}")
+            row.append(f"├── {queued}{rev[:7]}")
         else:
-            row.append(f"└── {rev[:7]}")
+            row.append(f"└── {queued}{rev[:7]}")
 
         _extend(row, metric_names, exp.get("metrics", {}).items())
         _extend(row, param_names, exp.get("params", {}).items())
@@ -229,6 +234,7 @@ def add_parser(subparsers, parent_parser):
     experiments_parser = subparsers.add_parser(
         "experiments",
         parents=[parent_parser],
+        aliases=["exp"],
         description=append_doc_link(EXPERIMENTS_HELP, "experiments"),
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )

dvc/command/repro.py

@@ -41,6 +41,9 @@ def run(self):
                     recursive=self.args.recursive,
                     force_downstream=self.args.force_downstream,
                     experiment=self.args.experiment,
+                    queue=self.args.queue,
+                    run_all=self.args.run_all,
+                    jobs=self.args.jobs,
                 )
 
                 if len(stages) == 0:
@@ -174,4 +177,16 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help=argparse.SUPPRESS,
     )
+    repro_parser.add_argument(
+        "--queue", action="store_true", default=False, help=argparse.SUPPRESS
+    )
+    repro_parser.add_argument(
+        "--run-all",
+        action="store_true",
+        default=False,
+        help=argparse.SUPPRESS,
+    )
+    repro_parser.add_argument(
+        "-j", "--jobs", type=int, help=argparse.SUPPRESS, metavar="<number>"
+    )
     repro_parser.set_defaults(func=CmdRepro)