exp run --run-all: re-enable --jobs concurrency (#8608)

* re-enable --jobs concurrency

* use `start_workers` instead of `spawn_worker`

* make `spawn_worker` a private method.

* update tests: make `spawn_worker` a private method.

dvc/repo/experiments/__init__.py

@@ -180,8 +180,7 @@ class Experiments:
         if not entries:
             return results
 
-        # TODO: re-enable --jobs concurrency
-        self.celery_queue.spawn_worker()
+        self.celery_queue.start_workers(count=kwargs.get("jobs", 1))
         failed = []
         try:
             ui.write(

dvc/repo/experiments/queue/celery.py

@@ -116,7 +116,7 @@ class LocalCeleryQueue(BaseStashQueue):
             timeout=10,
         )
 
-    def spawn_worker(self, num: int = 1):
+    def _spawn_worker(self, num: int = 1):
         """spawn one single worker to process to queued tasks.
 
         Argument:
@@ -159,7 +159,7 @@ class LocalCeleryQueue(BaseStashQueue):
             if node_name in active_worker:
                 logger.debug(f"Exp queue worker {node_name} already exist")
                 continue
-            self.spawn_worker(num)
+            self._spawn_worker(num)
             started += 1
 
         return started

tests/unit/command/test_queue.py

@@ -114,7 +114,7 @@ def test_experiments_start(dvc, scm, mocker):
 
     cmd = cli_args.func(cli_args)
     m = mocker.patch(
-        "dvc.repo.experiments.queue.celery.LocalCeleryQueue.spawn_worker",
+        "dvc.repo.experiments.queue.celery.LocalCeleryQueue._spawn_worker",
     )
 
     assert cmd.run() == 0

tests/unit/repo/experiments/conftest.py

@@ -132,7 +132,7 @@ def test_queue(tmp_dir, dvc, scm, mocker) -> LocalCeleryQueue:
     import celery
 
     queue = dvc.experiments.celery_queue
-    mocker.patch.object(queue, "spawn_worker")
+    mocker.patch.object(queue, "_spawn_worker")
 
     f = partial(
         _thread_worker,