diff --git a/verl/single_controller/ray/base.py b/verl/single_controller/ray/base.py index 2f6ee47064f..d15b8b4ad97 100644 --- a/verl/single_controller/ray/base.py +++ b/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( { diff --git a/verl/trainer/main_ppo.py b/verl/trainer/main_ppo.py index 2c84374d245..c02b9dfa75b 100644 --- a/verl/trainer/main_ppo.py +++ b/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,6 +324,7 @@ 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( @@ -301,12 +332,14 @@ def run(self, 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() diff --git a/verl/utils/tracking.py b/verl/utils/tracking.py index f4c1e4a7f8d..c2223b14087 100644 --- a/verl/utils/tracking.py +++ b/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: diff --git a/verl/workers/rollout/vllm_rollout/bucketed_weight_transfer.py b/verl/workers/rollout/vllm_rollout/bucketed_weight_transfer.py index d4aa12490ca..8609dc6ac8c 100644 --- a/verl/workers/rollout/vllm_rollout/bucketed_weight_transfer.py +++ b/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): diff --git a/verl/workers/rollout/vllm_rollout/utils.py b/verl/workers/rollout/vllm_rollout/utils.py index 5754f17a81e..e05e8492067 100644 --- a/verl/workers/rollout/vllm_rollout/utils.py +++ b/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: diff --git a/verl/workers/rollout/vllm_rollout/vllm_async_server.py b/verl/workers/rollout/vllm_rollout/vllm_async_server.py index 5395668b1dd..26218d798ec 100644 --- a/verl/workers/rollout/vllm_rollout/vllm_async_server.py +++ b/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( diff --git a/verl/workers/rollout/vllm_rollout/vllm_rollout.py b/verl/workers/rollout/vllm_rollout/vllm_rollout.py index 90641b575a2..b8be8cdb62b 100644 --- a/verl/workers/rollout/vllm_rollout/vllm_rollout.py +++ b/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: