exp: use celery threadpool worker instead of prefork

dvc/repo/experiments/queue/local.py

@@ -93,7 +93,24 @@ class LocalCeleryQueue(BaseStashQueue):
     def worker(self) -> "TemporaryWorker":
         from dvc_task.worker import TemporaryWorker
 
-        return TemporaryWorker(self.celery, concurrency=1, timeout=10)
+        # NOTE: Use thread pool with concurrency 1 and disabled prefetch.
+        # Worker scaling should be handled by running additional workers,
+        # rather than increasing pool concurrency.
+        #
+        # We use "threads" over "solo" (inline single-threaded) execution so
+        # that we still have access to the control/broadcast API (which
+        # requires a separate message handling thread in the worker).
+        #
+        # Disabled prefetch ensures that each worker will can only schedule and
+        # execute up to one experiment at a time (and a worker cannot prefetch
+        # additional experiments from the queue).
+        return TemporaryWorker(
+            self.celery,
+            pool="threads",
+            concurrency=1,
+            prefetch_multiplier=1,
+            timeout=10,
+        )
 
     def spawn_worker(self):
         from dvc_task.proc.process import ManagedProcess

setup.cfg

@@ -67,7 +67,7 @@ install_requires =
     aiohttp-retry>=2.4.5
     scmrepo==0.0.25
     dvc-render==0.0.6
-    dvc-task@git+https://github.com/iterative/[email protected].6
+    dvc-task@git+https://github.com/iterative/[email protected].7
     dvclive>=0.7.3
     dvc-data==0.0.23