pipe clean

Signed-off-by: Youngeun Kwon <[email protected]>

Author: youngeunkwon0405

nemo_rl/algorithms/utils.py

@@ -463,7 +463,7 @@ def visualize_per_worker_timeline(
         zero_marker = "▁"
         # marker = {0: "□", 1: "⧅", 2: "⛝", 3: "■"}
 
-        max_value = max(max(v) for v in metric_dict.values())
+        max_value = max((max(v) if v else 0) for v in metric_dict.values())
         bin_width = (max_value + 1) / len(marker)
 
         print(f"  - {metric_name}:")
@@ -521,7 +521,8 @@ def visualize_per_worker_timeline(
             vllm_metrics_logger_interval,
         )
         max_num_pending_samples = max(
-            max(v) for v in vllm_logger_metrics["num_pending_samples"].values()
+            (max(v) if v else 0)
+            for v in vllm_logger_metrics["num_pending_samples"].values()
         )
         # If there is at least one pending sample, visualize the timeline
         if max_num_pending_samples > 0:

nemo_rl/models/generation/vllm/vllm_worker.py

@@ -332,6 +332,9 @@ def _patch_vllm_init_workers_ray():
         # Optionally start periodic vLLM metrics logging if the flag is set
         # NOTE: vLLM metrics logger is only supported with async engine enabled
         # Metrics logger only enabled for per-actor, model-owner only
+        # Thread synchronization for metrics access
+        self._vllm_metrics_lock = threading.Lock()
+
         if self.cfg["vllm_cfg"].get("enable_vllm_metrics_logger", False) and self.cfg[
             "vllm_cfg"
         ].get("async_engine", False):
@@ -366,9 +369,6 @@ def _start_vllm_metrics_logger(self) -> None:
         stop_event = threading.Event()
         self._vllm_metrics_logger_stop_event = stop_event
 
-        # Thread synchronization for metrics access
-        self._vllm_metrics_lock = threading.Lock()
-
         self.inflight_batch_sizes: list[int] = []
         self.num_pending_samples: list[int] = []