Feat/cache reuse

verl/single_controller/ray/base.py

@@ -648,6 +648,35 @@ def _create_worker(self, rank, pg_idx, pg, local_rank, resource_pool, ray_cls_wi
         cia_name = match.group(1) if match else cia_name  # "ActorClass(Obj)" -> "Obj"
         name = f"{self.name_prefix}{cia_name}_{pg_idx}:{local_rank}"  # e.g. Worker_2:5
 
+        # Isolate compile/cache paths per worker to avoid concurrent file races while
+        # still reusing persistent cache roots on shared filesystem.
+        worker_key = pg_idx * 1024 + local_rank
+        cache_env_keys = [
+            "TORCHINDUCTOR_CACHE_DIR",
+            "TRITON_CACHE_DIR",
+            "XDG_CACHE_HOME",
+            "VLLM_CACHE_ROOT",
+        ]
+        resolved_cache_env = {}
+        for cache_key in cache_env_keys:
+            base_cache_dir = env_vars.get(cache_key) or os.environ.get(cache_key)
+            if not base_cache_dir:
+                continue
+            worker_cache_dir = os.path.join(base_cache_dir, cia_name, f"rank_{worker_key}")
+            os.makedirs(worker_cache_dir, exist_ok=True)
+            env_vars[cache_key] = worker_cache_dir
+            resolved_cache_env[cache_key] = worker_cache_dir
+
+        if resolved_cache_env:
+            logging.warning(
+                "Worker cache isolation: rank=%s pg_idx=%s local_rank=%s class=%s caches=%s",
+                rank,
+                pg_idx,
+                local_rank,
+                cia_name,
+                resolved_cache_env,
+            )
+
         if self.profile_steps and self.device_name == "cuda":
             ray_cls_with_init.update_options(
                 {

verl/trainer/main_ppo.py

@@ -17,6 +17,8 @@
 
 import os
 import socket
+import time
+from datetime import datetime
 
 import hydra
 import ray
@@ -276,16 +278,44 @@ def run(self, config):
             config: Training configuration object containing all parameters needed
                    for setting up and running the PPO training process.
         """
-        # Print the initial configuration. `resolve=True` will evaluate symbolic values.
-        from pprint import pprint
-
         from omegaconf import OmegaConf
 
         from verl.utils.fs import copy_to_local
 
         print(f"TaskRunner hostname: {socket.gethostname()}, PID: {os.getpid()}")
-        pprint(OmegaConf.to_container(config, resolve=True))
+        setup_start_ts = time.perf_counter()
+        setup_stage_ts = setup_start_ts
+
+        def log_setup_stage(stage: str) -> None:
+            nonlocal setup_stage_ts
+            now = time.perf_counter()
+            wall_clock = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            print(
+                f"[SETUP][{wall_clock}] {stage} | "
+                f"stage_s={now - setup_stage_ts:.2f} | elapsed_s={now - setup_start_ts:.2f}"
+            )
+            setup_stage_ts = now
+
+        print_full_config = os.getenv("VERL_PRINT_FULL_CONFIG", "0") == "1"
+        if print_full_config:
+            # resolve=True will evaluate symbolic values.
+            from pprint import pprint
+
+            pprint(OmegaConf.to_container(config, resolve=True))
+        else:
+            # Keep startup logging concise to avoid blocking on giant pretty-print output.
+            print(
+                "TaskRunner config summary: "
+                f"trainer.nnodes={config.trainer.nnodes}, "
+                f"trainer.n_gpus_per_node={config.trainer.n_gpus_per_node}, "
+                f"actor.strategy={config.actor_rollout_ref.actor.strategy}, "
+                f"rollout.name={config.actor_rollout_ref.rollout.name}, "
+                f"model.path={config.actor_rollout_ref.model.path}, "
+                f"train_files={config.data.train_files}, "
+                f"val_files={config.data.val_files}"
+            )
         OmegaConf.resolve(config)
+        log_setup_stage("resolved config")
 
         actor_rollout_cls, ray_worker_group_cls = self.add_actor_rollout_worker(config)
         self.add_critic_worker(config)
@@ -294,19 +324,22 @@ def run(self, config):
 
         # Add a reference policy worker if KL loss or KL reward is used.
         self.add_ref_policy_worker(config, actor_rollout_cls)
+        log_setup_stage("registered role workers and mappings")
 
         # validate config
         validate_config(
             config=config,
             use_reference_policy=need_reference_policy(config),
             use_critic=need_critic(config),
         )
+        log_setup_stage("validated config")
 
         # Download the checkpoint from HDFS to the local machine.
         # `use_shm` determines whether to use shared memory, which could lead to faster model loading if turned on
         local_path = copy_to_local(
             config.actor_rollout_ref.model.path, use_shm=config.actor_rollout_ref.model.get("use_shm", False)
         )
+        log_setup_stage("prepared local model path")
 
         # Instantiate the tokenizer and processor.
         from verl.utils import hf_processor, hf_tokenizer
@@ -315,8 +348,10 @@ def run(self, config):
         tokenizer = hf_tokenizer(local_path, trust_remote_code=trust_remote_code)
         # Used for multimodal LLM, could be None
         processor = hf_processor(local_path, trust_remote_code=trust_remote_code, use_fast=True)
+        log_setup_stage("initialized tokenizer/processor")
 
         resource_pool_manager = self.init_resource_pool_mgr(config)
+        log_setup_stage("initialized resource pool manager")
 
         from verl.utils.dataset.rl_dataset import collate_fn
 
@@ -338,6 +373,7 @@ def run(self, config):
             max_samples=config.data.get("val_max_samples", -1),
         )
         train_sampler = create_rl_sampler(config.data, train_dataset)
+        log_setup_stage("built datasets and sampler")
 
         # Initialize the PPO trainer.
         trainer = RayPPOTrainer(
@@ -352,8 +388,15 @@ def run(self, config):
             collate_fn=collate_fn,
             train_sampler=train_sampler,
         )
+        log_setup_stage("constructed trainer")
         # Initialize the workers of the trainer.
         trainer.init_workers()
+        log_setup_stage("initialized trainer workers")
+        wall_clock = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        print(
+            f"[SETUP][{wall_clock}] pre-training setup complete | "
+            f"total_setup_s={time.perf_counter() - setup_start_ts:.2f}"
+        )
 
         # Start the training process.
         trainer.fit()

verl/utils/tracking.py

@@ -184,20 +184,23 @@ def log(self, data, step, backend=None):
                 logger_instance.log(data=data, step=step)
 
     def __del__(self):
-        if "wandb" in self.logger:
-            self.logger["wandb"].finish(exit_code=0)
-        if "swanlab" in self.logger:
-            self.logger["swanlab"].finish()
-        if "vemlp_wandb" in self.logger:
-            self.logger["vemlp_wandb"].finish(exit_code=0)
-        if "tensorboard" in self.logger:
-            self.logger["tensorboard"].finish()
-        if "clearml" in self.logger:
-            self.logger["clearml"].finish()
-        if "trackio" in self.logger:
-            self.logger["trackio"].finish()
-        if "file" in self.logger:
-            self.logger["file"].finish()
+        # Destructor can run while asyncio loop is still active or interpreter is shutting down.
+        # Guard finalizers to avoid noisy teardown-time exceptions.
+        for backend, finish_call in (
+            ("wandb", lambda: self.logger["wandb"].finish(exit_code=0)),
+            ("swanlab", lambda: self.logger["swanlab"].finish()),
+            ("vemlp_wandb", lambda: self.logger["vemlp_wandb"].finish(exit_code=0)),
+            ("tensorboard", lambda: self.logger["tensorboard"].finish()),
+            ("clearml", lambda: self.logger["clearml"].finish()),
+            ("trackio", lambda: self.logger["trackio"].finish()),
+            ("file", lambda: self.logger["file"].finish()),
+        ):
+            if backend not in self.logger:
+                continue
+            try:
+                finish_call()
+            except Exception as e:
+                print(f"[Tracking] Ignored teardown error for {backend}: {e}")
 
 
 class ClearMLLogger:

verl/workers/rollout/vllm_rollout/bucketed_weight_transfer.py

@@ -156,6 +156,16 @@ async def async_send_weights(self, weights):
     def _init_socket(self):
         """Initialize ZMQ REQ socket and bind."""
         self.socket = self.zmq_context.socket(zmq.REQ)
+
+        if self.zmq_handle.startswith("ipc://"):
+            socket_path = self.zmq_handle[len("ipc://") :]
+            if os.path.exists(socket_path):
+                try:
+                    os.remove(socket_path)
+                    logger.warning(f"Removed stale ZMQ IPC socket path before bind: {socket_path}")
+                except OSError as e:
+                    logger.warning(f"Failed to remove stale ZMQ IPC socket path {socket_path}: {e}")
+
         self.socket.bind(self.zmq_handle)
 
     def _init_buffer(self):

verl/workers/rollout/vllm_rollout/utils.py

@@ -64,6 +64,17 @@ def get_device_uuid(device_id: int) -> str:
         return current_platform.get_device_uuid(device_id)
 
 
+def build_zmq_ipc_handle(device_uuid: str) -> str:
+    """Build a run-scoped IPC socket path to avoid cross-run collisions."""
+    namespace = os.environ.get("VERL_ZMQ_NAMESPACE") or os.environ.get("JOB_ID") or ""
+    if not namespace and os.environ.get("PBS_JOBID"):
+        namespace = os.environ["PBS_JOBID"].split(".", 1)[0]
+
+    safe_namespace = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in namespace)
+    suffix = f"-{safe_namespace}" if safe_namespace else ""
+    return f"ipc:///tmp/rl-colocate-zmq-{device_uuid}{suffix}.sock"
+
+
 def get_vllm_max_lora_rank(lora_rank: int):
     """
     For vLLM, automatically adjusts the `max_lora_rank` to the nearest allowed value.
@@ -233,7 +244,7 @@ def _get_zmq_handle(self) -> str:
         """Get ZMQ handle for communication."""
         if not hasattr(self, "device_uuid") or not self.device_uuid:
             self.device_uuid = get_device_uuid(self.device.index)
-        return f"ipc:///tmp/rl-colocate-zmq-{self.device_uuid}.sock"
+        return build_zmq_ipc_handle(self.device_uuid)
 
 
 class SuppressSignalInThread:

verl/workers/rollout/vllm_rollout/vllm_async_server.py

@@ -108,6 +108,20 @@ def __init__(
         """
         os.environ[get_visible_devices_keyword()] = cuda_visible_devices
 
+        cache_env_keys = [
+            "VLLM_CACHE_ROOT",
+            "TORCHINDUCTOR_CACHE_DIR",
+            "TRITON_CACHE_DIR",
+            "XDG_CACHE_HOME",
+        ]
+        resolved_cache_env = {}
+        for cache_key in cache_env_keys:
+            cache_dir = os.environ.get(cache_key)
+            if not cache_dir:
+                continue
+            os.makedirs(cache_dir, exist_ok=True)
+            resolved_cache_env[cache_key] = cache_dir
+
         self.config: RolloutConfig = omega_conf_to_dataclass(config)
         self.model_config: HFModelConfig = omega_conf_to_dataclass(model_config, dataclass_type=HFModelConfig)
         max_position_embeddings = get_max_position_embeddings(self.model_config.hf_config)
@@ -169,6 +183,13 @@ def __init__(
             f"master_address: {self._master_address}, master_port: {self._master_port}, "
             f"data_parallel_rpc_port: {self._dp_rpc_port}, data_parallel_master_port: {self._dp_master_port}"
         )
+        if resolved_cache_env:
+            logger.warning(
+                "vLLM server cache isolation: replica_rank=%s node_rank=%s caches=%s",
+                self.replica_rank,
+                self.node_rank,
+                resolved_cache_env,
+            )
 
     def get_master_address(self):
         """Get master address and port for data parallel.
@@ -843,23 +864,36 @@ async def launch_servers(self):
                 if not self.is_reward_model
                 else f"vllm_server_reward_{self.replica_rank}_{node_rank}"
             )
+            server_seed = self.replica_rank * 1024 + node_rank
+            server_env_vars = {
+                "RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES": "1",
+                "RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES": "1",
+                # To prevent hanging or crash during synchronization of weights between actor and rollout
+                # in disaggregated mode. See:
+                # https://docs.vllm.ai/en/latest/usage/troubleshooting.html?h=nccl_cumem_enable#known-issues
+                # https://github.com/vllm-project/vllm/blob/c6b0a7d3ba03ca414be1174e9bd86a97191b7090/vllm/worker/worker_base.py#L445
+                "NCCL_CUMEM_ENABLE": "0",
+            }
+            cache_env_keys = [
+                "VLLM_CACHE_ROOT",
+                "TORCHINDUCTOR_CACHE_DIR",
+                "TRITON_CACHE_DIR",
+                "XDG_CACHE_HOME",
+            ]
+            for cache_key in cache_env_keys:
+                base_cache_dir = os.environ.get(cache_key)
+                if not base_cache_dir:
+                    continue
+                server_cache_dir = os.path.join(base_cache_dir, "vllm_server", f"seed_{server_seed}")
+                os.makedirs(server_cache_dir, exist_ok=True)
+                server_env_vars[cache_key] = server_cache_dir
 
             server = self.server_class.options(
                 scheduling_strategy=ray.util.scheduling_strategies.NodeAffinitySchedulingStrategy(
                     node_id=node_id,
                     soft=False,
                 ),
-                runtime_env={
-                    "env_vars": {
-                        "RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES": "1",
-                        "RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES": "1",
-                        # To prevent hanging or crash during synchronization of weights between actor and rollout
-                        # in disaggregated mode. See:
-                        # https://docs.vllm.ai/en/latest/usage/troubleshooting.html?h=nccl_cumem_enable#known-issues
-                        # https://github.com/vllm-project/vllm/blob/c6b0a7d3ba03ca414be1174e9bd86a97191b7090/vllm/worker/worker_base.py#L445
-                        "NCCL_CUMEM_ENABLE": "0",
-                    }
-                },
+                runtime_env={"env_vars": server_env_vars},
                 name=name,
                 max_concurrency=self.max_concurrency,
             ).remote(

verl/workers/rollout/vllm_rollout/vllm_rollout.py

@@ -42,7 +42,7 @@
 from verl.workers.config import HFModelConfig, RolloutConfig
 from verl.workers.rollout.base import BaseRollout
 from verl.workers.rollout.vllm_rollout.bucketed_weight_transfer import BucketedWeightSender
-from verl.workers.rollout.vllm_rollout.utils import get_device_uuid
+from verl.workers.rollout.vllm_rollout.utils import build_zmq_ipc_handle, get_device_uuid
 
 logger = logging.getLogger(__file__)
 logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "INFO"))
@@ -95,7 +95,7 @@ def __init__(
             self.sleep_level = VLLM_SLEEP_LEVEL
 
         self.device_uuid = get_device_uuid(get_device_id())
-        self.zmq_handle = f"ipc:///tmp/rl-colocate-zmq-{self.device_uuid}.sock"
+        self.zmq_handle = build_zmq_ipc_handle(self.device_uuid)
 
         self.use_shm = not is_support_ipc()
         if self.use_shm: