spawn in setup

Author: casteryh

src/forge/actors/generator.py

@@ -18,7 +18,7 @@
 
 import torch
 import torchstore as ts
-from monarch.actor import current_rank, endpoint, ProcMesh
+from monarch.actor import current_rank, endpoint, ProcMesh, this_host
 
 from vllm.config import VllmConfig
 
@@ -174,8 +174,6 @@ async def launch(  # pyright: ignore[reportIncompatibleMethodOverride]
             "vllm_worker", GeneratorWorker, vllm_config=vllm_config, use_dcp=use_dcp
         )
 
-        weight_fetchers = worker_procs.spawn("weight_fetcher", _WeightFetcher)
-
         if isinstance(sampling_params, Mapping):
             sampling_params = SamplingParams.from_optional(**sampling_params)
             sampling_params.output_kind = RequestOutputKind.FINAL_ONLY
@@ -256,6 +254,9 @@ async def setup(self):
             log_stats=None,
         )
         self._start_processing()
+        fetcher_procs = this_host().spawn_procs(per_host={"procs": 8})
+        self._fetcher_procs = fetcher_procs
+        self.weight_fetchers = fetcher_procs.spawn("weight_fetcher", _WeightFetcher)
 
     def _start_processing(self):
         if self._run_task is None or self._run_task.done():
@@ -585,6 +586,7 @@ async def shutdown(  # pyright: ignore[reportIncompatibleMethodOverride]
         await actor._cleanup_shared_memory.call()
         await stop_proc_mesh(actor._worker_procs)
         await stop_proc_mesh(actor._generator_proc)
+        await stop_proc_mesh(actor._fetcher_procs)
 
     @endpoint
     async def _test_save_model_params(self):