exp: use dvc.daemon to run worker on windows

dvc/daemon.py

@@ -6,6 +6,7 @@ import os
 import platform
 import sys
 from subprocess import Popen
+from typing import List
 
 from dvc.env import DVC_DAEMON
 from dvc.utils import fix_env, is_binary
@@ -99,12 +100,14 @@ def daemon(args):
     Args:
         args (list): list of arguments to append to `dvc daemon` command.
     """
+    daemonize(["daemon", "-q"] + args)
+
+
+def daemonize(cmd: List[str]):
     if os.environ.get(DVC_DAEMON):
         logger.debug("skipping launching a new daemon.")
         return
 
-    cmd = ["daemon", "-q"] + args
-
     env = fix_env()
     if not is_binary():
         file_path = os.path.abspath(inspect.stack()[0][1])

dvc/repo/experiments/queue/local.py

@@ -16,6 +16,7 @@ from typing import (
 from funcy import cached_property, first
 from kombu.message import Message
 
+from dvc.daemon import daemonize
 from dvc.exceptions import DvcException
 
 from ..exceptions import ExpQueueEmptyError
@@ -92,11 +93,13 @@ class LocalCeleryQueue(BaseStashQueue):
         logger.debug("Spawning exp queue worker")
         wdir_hash = hashlib.sha256(self.wdir.encode("utf-8")).hexdigest()[:6]
         node_name = f"dvc-exp-{wdir_hash}-1@localhost"
-        ManagedProcess.spawn(
-            ["dvc", "exp", "queue-worker", node_name],
-            wdir=self.wdir,
-            name="dvc-exp-worker",
-        )
+        cmd = ["exp", "queue-worker", node_name]
+        if os.name == "nt":
+            daemonize(cmd)
+        else:
+            ManagedProcess.spawn(
+                ["dvc"] + cmd, wdir=self.wdir, name="dvc-exp-worker"
+            )
 
     def put(self, *args, **kwargs) -> QueueEntry:
         """Stash an experiment and add it to the queue."""