DDP lanes: train one dataset across several GPUs (--gpus-per-job)

docs/rf100vl-operator-runbook.md

@@ -324,6 +324,42 @@ headroom set it; re-measure after stack changes). The campaign prints a
 Monitor line and a Profile line — use `libreyolo profile phases` on a slow
 dataset before inventing theories.
 
+### Multi-GPU lanes (`--gpus-per-job`), for models that do not fit one card
+
+The inverse of packing. When the protocol batch of a model does not fit a
+single GPU (measured 2026-08: rfdetr-m and rfdetr-l at batch 16 need ~31 GB,
+and batch 8 already OOMs a 16 GB card), group the cards into DDP lanes
+instead of renting a bigger one:
+
+```bash
+va-bench rf100vl-campaign --model rfdetr-m   --data-dir /root/rf100-vl --weights-root /root/rf100vl-weights   --gpus 0,1,2,3,4,5,6,7 --gpus-per-job 4
+```
+
+That is 2 lanes of 4 GPUs; each dataset trains on a whole lane. What it does
+and does not change:
+
+- The recipe's GLOBAL batch is untouched. LibreYOLO splits it across the
+  lane's ranks (`batch // world_size` per GPU: 16 over 4 GPUs is 4 per card,
+  ~7 GB for rfdetr-m) and derives accumulation from `nbs` against the global
+  batch, so effective batch 16 holds at the optimizer step.
+- RF-DETR's per-step multi-scale draw is seeded by the step counter, so all
+  ranks resize to the SAME resolution each step: closer to the reference
+  single-GPU batch-16 semantics than the grad-accum fallback, which averages
+  a different scale per micro-batch.
+- It is NOT bit-identical to a single-GPU run (sampler sharding and per-rank
+  augmentation seeds differ), so the lane width is part of the run
+  signature: a dataset started at one width resumes only at that width, and
+  the solo OOM drain retries at full lane width. Single-GPU signatures are
+  unchanged, so banked checkpoints from earlier campaigns stay resumable.
+- `--gpus-per-job` and `--jobs-per-gpu` are mutually exclusive, and the GPU
+  count must divide into whole lanes.
+- `rfdetr-l` still carries a `physical_batch: 2` size override from the
+  16 GB era. Running l at the protocol batch over a lane needs a recipe
+  change (which re-signs the campaign); decide that before dataset one.
+
+Shake down a lane exactly like a card: one dense dataset, `--smoke-epochs 2`,
+read the true per-GPU peak with the lane you will actually run.
+
 Detach (Ctrl-b d). If ANYTHING dies, up to and including the box, re-running
 this *identical* command resumes only if the recipe hash is unchanged:
 finished datasets are skipped, interrupted ones continue from `last.pt`.

tests/test_rf100vl_recovery.py

@@ -419,3 +419,93 @@ def test_capture_race_with_graphs_already_disabled_is_final(tmp_path, monkeypatc
     assert summary["completed"] == []
     assert summary["failed"] == ["racy"]
     assert len(launcher.calls) == 2
+
+
+# ---------------------------------------------------------------------------
+# DDP lanes (gpus_per_job > 1)
+# ---------------------------------------------------------------------------
+
+
+def test_ddp_lanes_group_gpus_and_record_the_width(tmp_path, monkeypatch):
+    monkeypatch.setattr(rf100vl_train, "_libreyolo_supports_ddp", lambda: True)
+    names = ["alpha", "beta"]
+    summary, launcher = _orchestrate(
+        tmp_path,
+        monkeypatch,
+        {"alpha": ["done"], "beta": ["done"]},
+        names,
+        gpus=["0", "1", "2", "3"],
+        gpus_per_job=2,
+    )
+
+    assert sorted(summary["completed"]) == names
+    assert {call["gpu"] for call in launcher.calls} <= {"0,1", "2,3"}
+
+    for name in names:
+        config = json.loads(
+            (tmp_path / "state" / "jobs" / f"{name}.json").read_text(encoding="utf-8")
+        )
+        assert config["batch_plan"]["ddp_world_size"] == 2
+        assert config["batch_plan"]["per_gpu_physical_batch"] == 8
+
+
+def test_ddp_lane_oom_retries_at_full_width(tmp_path, monkeypatch):
+    """The solo drain must hand an OOM'd dataset its lane back at the SAME
+    width: shrinking to one GPU would change the run signature and turn the
+    resume into a hard refusal."""
+    monkeypatch.setattr(rf100vl_train, "_libreyolo_supports_ddp", lambda: True)
+    names = ["alpha", "densy"]
+    summary, launcher = _orchestrate(
+        tmp_path,
+        monkeypatch,
+        {"alpha": ["done"], "densy": ["oom", "done"]},
+        names,
+        gpus=["0", "1"],
+        gpus_per_job=2,
+    )
+
+    assert sorted(summary["completed"]) == names
+    densy_calls = [call for call in launcher.calls if call["dataset"] == "densy"]
+    assert len(densy_calls) == 2
+    assert all(call["gpu"] == "0,1" for call in densy_calls)
+
+
+def test_ddp_lane_argument_validation(tmp_path, monkeypatch):
+    monkeypatch.setattr(rf100vl_train, "_libreyolo_supports_ddp", lambda: True)
+    names = ["alpha"]
+    root_a = tmp_path / "a"
+    root_b = tmp_path / "b"
+    root_a.mkdir()
+    root_b.mkdir()
+    with pytest.raises(ValueError, match="divide into lanes"):
+        _orchestrate(
+            root_a,
+            monkeypatch,
+            {"alpha": ["done"]},
+            names,
+            gpus=["0", "1", "2"],
+            gpus_per_job=2,
+        )
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        _orchestrate(
+            root_b,
+            monkeypatch,
+            {"alpha": ["done"]},
+            names,
+            gpus=["0", "1"],
+            gpus_per_job=2,
+            jobs_per_gpu=2,
+        )
+
+
+def test_ddp_lanes_refuse_a_libreyolo_without_ddp(tmp_path, monkeypatch):
+    monkeypatch.setattr(rf100vl_train, "_libreyolo_supports_ddp", lambda: False)
+    with pytest.raises(RuntimeError, match="ddp_spawn"):
+        _orchestrate(
+            tmp_path,
+            monkeypatch,
+            {"alpha": ["done"]},
+            ["alpha"],
+            gpus=["0", "1"],
+            gpus_per_job=2,
+        )

tests/test_rf100vl_train.py

@@ -708,3 +708,122 @@ def write(name, sha):
 
     assert artifacts._submission_recipe_sha(tmp_path, "ec-s") == mine
     assert artifacts._submission_recipe_sha(tmp_path, "yolox-m") == theirs
+
+
+# ---------------------------------------------------------------------------
+# DDP lanes (gpus_per_job > 1)
+# ---------------------------------------------------------------------------
+
+
+def test_lane_world_size_reads_the_lane_spec():
+    assert rf100vl_train._lane_world_size("0") == 1
+    assert rf100vl_train._lane_world_size("4,5") == 2
+    assert rf100vl_train._lane_world_size(" 4 , 5 , 6 , 7 ") == 4
+
+
+def test_ddp_lane_keeps_the_global_batch_and_names_every_device(tmp_path):
+    """A 2-GPU lane must not change the protocol: the plan keeps the recipe's
+    global physical batch, the split is derived, and the worker's device spec
+    names one index per rank of the lane."""
+    spec = get_spec("rfdetr-m")
+    recipe = rf100vl_train.load_recipe(
+        rf100vl_train.recipe_path_for_family("rfdetr"),
+        family="rfdetr",
+    )
+    facts = {"max_annotations_per_image": 1, "num_train_images": 20}
+
+    plan = rf100vl_train.select_batch_plan(recipe, spec, facts, world_size=2)
+    assert plan["physical_batch"] == 16
+    assert plan["effective_batch"] == 16
+    assert plan["gradient_accumulation_steps"] == 1
+    assert plan["ddp_world_size"] == 2
+    assert plan["per_gpu_physical_batch"] == 8
+
+    kwargs = rf100vl_train.build_train_kwargs(
+        recipe,
+        spec,
+        plan,
+        data_yaml=tmp_path / "data.yaml",
+        run_dir=tmp_path / "run",
+        resume=False,
+    )
+    assert kwargs["device"] == "0,1"
+    assert kwargs["batch"] == 16
+    assert kwargs["nbs"] == 16
+
+    solo = rf100vl_train.select_batch_plan(recipe, spec, facts)
+    solo_kwargs = rf100vl_train.build_train_kwargs(
+        recipe,
+        spec,
+        solo,
+        data_yaml=tmp_path / "data.yaml",
+        run_dir=tmp_path / "run",
+        resume=False,
+    )
+    assert solo_kwargs["device"] == "0"
+
+
+def test_ddp_lane_rejects_a_batch_that_does_not_split():
+    recipe = rf100vl_train.load_recipe(
+        rf100vl_train.recipe_path_for_family("rfdetr"),
+        family="rfdetr",
+    )
+    facts = {"max_annotations_per_image": 1, "num_train_images": 20}
+    with pytest.raises(ValueError, match="split evenly"):
+        rf100vl_train.select_batch_plan(
+            recipe, get_spec("rfdetr-m"), facts, world_size=3
+        )
+    # rfdetr-l's recipe override is physical batch 2, which cannot feed 4 ranks.
+    with pytest.raises(ValueError, match="split evenly"):
+        rf100vl_train.select_batch_plan(
+            recipe, get_spec("rfdetr-l"), facts, world_size=4
+        )
+
+
+def test_run_signature_is_unchanged_for_single_gpu_and_widens_for_ddp(tmp_path):
+    """Signatures of every existing single-GPU campaign must keep hashing as
+    before this change (banked checkpoints stay resumable), while a DDP lane
+    signs its width so a checkpoint cannot silently resume at another."""
+    spec = get_spec("rfdetr-m")
+    recipe = rf100vl_train.load_recipe(
+        rf100vl_train.recipe_path_for_family("rfdetr"),
+        family="rfdetr",
+    )
+    facts = {
+        "max_annotations_per_image": 1,
+        "num_train_images": 20,
+        "train_annotations_sha256": "d" * 64,
+    }
+
+    def signature(plan):
+        kwargs = rf100vl_train.build_train_kwargs(
+            recipe,
+            spec,
+            plan,
+            data_yaml=tmp_path / "data.yaml",
+            run_dir=tmp_path / "run",
+            resume=False,
+        )
+        return rf100vl_train._run_signature(
+            model_key=spec.key,
+            recipe_sha256="a" * 64,
+            versions_sha256="b" * 64,
+            dataset_name="aerial-cows",
+            dataset_version_id=1,
+            dataset_facts=facts,
+            batch_plan=plan,
+            train_kwargs=kwargs,
+        )
+
+    modern = rf100vl_train.select_batch_plan(recipe, spec, facts)
+    # A plan recorded before ddp_world_size existed: same fields, no new keys.
+    legacy = {
+        key: value
+        for key, value in modern.items()
+        if key not in ("ddp_world_size", "per_gpu_physical_batch")
+    }
+    wide = rf100vl_train.select_batch_plan(recipe, spec, facts, world_size=2)
+
+    assert signature(modern) == signature(legacy)
+    assert signature(wide) != signature(modern)
+

va_bench/cli.py

@@ -262,6 +262,7 @@ def cmd_rf100vl_train(args: argparse.Namespace) -> None:
         num_shards=args.num_shards,
         timeout_hours=args.timeout_hours,
         jobs_per_gpu=args.jobs_per_gpu,
+        gpus_per_job=args.gpus_per_job,
         runs_root=args.runs_root,
         state_root=args.state_root,
         smoke_epochs=args.smoke_epochs,
@@ -514,10 +515,12 @@ def cmd_rf100vl_campaign(args: argparse.Namespace) -> None:
         num_shards=1,
         timeout_hours=args.timeout_hours,
         jobs_per_gpu=args.jobs_per_gpu,
+        gpus_per_job=args.gpus_per_job,
         runs_root=args.runs_root,
         state_root=args.state_root,
         smoke_epochs=args.smoke_epochs,
         force=args.force,
+        keep_cache=args.keep_cache,
     )
     # Stop the sampler and write its per-dataset traces BEFORE the final sync,
     # or the telemetry lands on disk after the last upload and never leaves the
@@ -1066,6 +1069,24 @@ def main(argv: list[str] | None = None) -> None:
         "run at the recipe's batch, so results are unchanged; this only "
         "fills a card that one small model cannot.",
     )
+    rft.add_argument(
+        "--gpus-per-job",
+        type=int,
+        default=1,
+        help="GPUs per training lane (DDP). --gpus is grouped into lanes of "
+        "this width and each dataset trains on a whole lane, with the "
+        "recipe's global batch split across the lane's ranks. For models "
+        "whose protocol batch does not fit one card. Mutually exclusive "
+        "with --jobs-per-gpu > 1.",
+    )
+    rft.add_argument(
+        "--keep-cache",
+        action="store_true",
+        help="Keep each dataset's post-resize .npy cache and its last.pt once "
+        "the dataset finishes. Off by default: across 100 datasets those are "
+        "the two largest consumers on a campaign box and neither is read again "
+        "after a dataset is done.",
+    )
     rft.add_argument(
         "--datasets",
         nargs="+",
@@ -1209,6 +1230,16 @@ def main(argv: list[str] | None = None) -> None:
         "run at the recipe's batch, so results are unchanged; this only "
         "fills a card that one small model cannot.",
     )
+    rc.add_argument(
+        "--gpus-per-job",
+        type=int,
+        default=1,
+        help="GPUs per training lane (DDP). --gpus is grouped into lanes of "
+        "this width and each dataset trains on a whole lane, with the "
+        "recipe's global batch split across the lane's ranks. For models "
+        "whose protocol batch does not fit one card. Mutually exclusive "
+        "with --jobs-per-gpu > 1.",
+    )
     rc.add_argument(
         "--no-gpu-trace",
         action="store_true",

va_bench/gpu_trace.py

@@ -100,11 +100,17 @@ def _gpu_to_datasets(state_dir: Path) -> dict[int, list[str]]:
         except Exception:
             continue
         if status.get("state") == "running" and status.get("gpu") is not None:
-            try:
-                index = int(status["gpu"])
-            except (TypeError, ValueError):
-                continue
-            mapping.setdefault(index, []).append(str(status.get("dataset", "?")))
+            # The gpu field is a lane spec: one index, or a comma-joined DDP
+            # group like "4,5". A DDP dataset belongs on every card it spans,
+            # so its trace attributes the whole lane rather than vanishing on
+            # the int() parse.
+            for part in str(status["gpu"]).split(","):
+                part = part.strip()
+                try:
+                    index = int(part)
+                except (TypeError, ValueError):
+                    continue
+                mapping.setdefault(index, []).append(str(status.get("dataset", "?")))
     return {index: sorted(names) for index, names in mapping.items()}