fix(ckpt): master-only mkdir + barrier to avoid concurrent-mkdir race on parallel FS (#2699)

Path.mkdir(exist_ok=True) re-checks self.is_dir() after EEXIST and re-raises if
it returns False. On a parallel filesystem (beegfs) with concurrent mkdir from
every trainer rank, a non-master rank can hit EEXIST while its metadata cache
hasn't seen the new inode yet -> is_dir() returns False -> FileExistsError,
crashing the trainer mid-save (observed at weights/step_180).

Guard the three all-ranks mkdir sites behind world.is_master + a barrier so only
the master creates the dir and the others wait until it exists:
- CheckpointManager.save_to_path: dataloader_dir
- CheckpointManager.save: ckpt_path.parent
- WeightCheckpointManager.save: step_path

Mirrors the existing master-only guard in WeightCheckpointManager.save_to_path.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: snimu

src/prime_rl/trainer/ckpt.py

@@ -171,7 +171,12 @@ def save_to_path(
         # Checkpoint the local dataloader
         if dataloader is not None:
             dataloader_dir = path / "dataloader"
-            dataloader_dir.mkdir(parents=True, exist_ok=True)
+            # Only the master creates the dir; the rest wait at a barrier. On a
+            # parallel FS (beegfs), concurrent mkdir from every rank can re-raise
+            # FileExistsError (EEXIST + stale is_dir() metadata).
+            if self.world.is_master:
+                dataloader_dir.mkdir(parents=True, exist_ok=True)
+            torch.distributed.barrier()
             torch.save(dataloader.state_dict(), dataloader_dir / f"rank_{self.world.rank}.pt")
 
         # Save sharded state
@@ -239,7 +244,11 @@ def save(
     ) -> None:
         """Save the full checkpoint state for a specified step."""
         ckpt_path = self.get_ckpt_path(step)
-        ckpt_path.parent.mkdir(parents=True, exist_ok=True)
+        # Master-only mkdir + barrier: concurrent mkdir from every rank can
+        # re-raise FileExistsError on a parallel FS (see save_to_path).
+        if self.world.is_master:
+            ckpt_path.parent.mkdir(parents=True, exist_ok=True)
+        torch.distributed.barrier()
 
         self.save_to_path(ckpt_path, model, optimizers, scheduler, progress, dataloader)
         bisect.insort(self.ckpt_steps, step)
@@ -390,7 +399,11 @@ def save(
     ):
         """Save a HF-compatible weight-only checkpoint for a given step."""
         step_path = self.get_step_path(step)
-        step_path.mkdir(parents=True, exist_ok=True)
+        # Master-only mkdir + barrier: concurrent mkdir from every rank can
+        # re-raise FileExistsError on a parallel FS (EEXIST + stale is_dir()).
+        if self.world.is_master:
+            step_path.mkdir(parents=True, exist_ok=True)
+        torch.distributed.barrier()
 
         # Gather all weights on master rank
         self.logger.debug("Gathering weights on master rank for weight checkpoint")