vinca-gha --timings (pack each stage's recipes into jobs by measured build time) and --keep-file

vinca/generate_gha.py

@@ -13,7 +13,12 @@
 
 from vinca import config
 from vinca.main import get_conda_subdir, read_vinca_yaml
-from vinca.pipeline import batch_stages, get_all_ancestors, get_skip_existing
+from vinca.pipeline import (
+    BuildTimings,
+    batch_stages,
+    get_all_ancestors,
+    get_skip_existing,
+)
 from vinca.utils import (
     NoAliasDumper,
     add_test_requirements,
@@ -82,6 +87,32 @@ def parse_command_line(argv):
         help="How many packages to build at most per stage",
     )
 
+    parser.add_argument(
+        "--timings",
+        dest="timings",
+        default=None,
+        help='JSON file of measured build times ({"recipes": {name: seconds}, "default": '
+        'seconds, "job_overhead": seconds}): pack each stage\'s packages into jobs by '
+        "time instead of by count",
+    )
+
+    parser.add_argument(
+        "--keep-file",
+        dest="keep_file",
+        default=None,
+        help="Keep this file of each build job (relative to the checkout) as an artifact "
+        "named after the job, e.g. build times it recorded",
+    )
+
+    parser.add_argument(
+        "--runners",
+        dest="runners",
+        default=None,
+        type=int,
+        help="With --timings: how many jobs can run at once (default: the file's, else "
+        "GitHub's limit for hosted runners, 20, or 5 on macOS)",
+    )
+
     arguments = parser.parse_args(argv[1:])
     config.parsed_args = arguments
     return arguments
@@ -173,6 +204,40 @@ def get_stage_name(batch):
     return " ".join(stage_name)
 
 
+def keep_file_steps() -> list[dict[str, Any]]:
+    """With --keep-file: a last step that keeps that file of the job (e.g. the build
+    times it recorded) as an artifact, also when the build failed."""
+    path = getattr(config.parsed_args, "keep_file", None)
+    if not path:
+        return []
+    return [
+        {
+            "name": "Keep " + os.path.basename(path),
+            "if": "always()",
+            "uses": "actions/upload-artifact@v6",
+            "with": {
+                "name": "${{ github.job }}-${{ github.run_attempt }}",
+                "path": path,
+                "if-no-files-found": "ignore",
+                "retention-days": 30,
+            },
+        }
+    ]
+
+
+def compatibility_followers(stages: list[list[str]]) -> dict[str, list[str]]:
+    """With package_name_mode: both, each ros-<distro>-<name> compatibility package only
+    depends on ros2-<name> (and the mutex, which that needs as well): it is built right
+    after it, in the same job, instead of in a stage of its own."""
+    prefix = f"ros-{config.ros_distro}-"
+    packages = {package for stage in stages for package in stage}
+    return {
+        f"ros2-{package[len(prefix) :]}": [package]
+        for package in sorted(packages)
+        if package.startswith(prefix) and f"ros2-{package[len(prefix) :]}" in packages
+    }
+
+
 def normalize_version(version: str) -> str:
     """Normalize a release while leaving action refs such as commit SHAs intact."""
     version = str(version).strip()
@@ -263,6 +328,7 @@ def build_unix_pipeline(
                     "env": build_env,
                     "run": script,
                 },
+                *keep_file_steps(),
             ]
 
             job = {
@@ -403,6 +469,7 @@ def build_win_pipeline(
                     "env": build_env,
                     "name": f"Build {' '.join([pkg for pkg in batch])}",
                 },
+                *keep_file_steps(),
             ]
 
             job = {
@@ -548,7 +615,15 @@ def main():
         if len(filtered):
             filtered_stages.append(filtered)
 
-    stages = batch_stages(filtered_stages, args.batch_size)
+    timings = (
+        BuildTimings.from_file(args.timings, args.runners, args.platform)
+        if args.timings
+        else None
+    )
+    followers = compatibility_followers(filtered_stages) if timings else None
+    stages = batch_stages(
+        filtered_stages, args.batch_size, timings=timings, followers=followers
+    )
     print(stages)
 
     with open("buildorder.txt", "w", encoding="utf-8") as fo:

vinca/pipeline.py

@@ -8,33 +8,161 @@
 
 from __future__ import annotations
 
+import heapq
+import json
 from collections.abc import Iterable, Mapping
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any
 
 import yaml
 
 from vinca.utils import get_repodata
 
+# with timings: consecutive single-job stages share a job while it stays this short
+SHARED_JOB_SECONDS = 120.0
+
 
 def _chunks(items: list[str], size: int) -> list[list[str]]:
     return [items[index : index + size] for index in range(0, len(items), size)]
 
 
+def default_runners(platform: str) -> int:
+    """GitHub's default limit on concurrent jobs with its hosted runners (the Free plan):
+    20 in total, of which 5 macOS."""
+    return 5 if platform.startswith("osx") else 20
+
+
+@dataclass(frozen=True)
+class BuildTimings:
+    """Measured build times, for packing each stage's packages into jobs by time.
+
+    ``recipes`` maps package names to their build (and upload) seconds; packages
+    without a measurement count as ``default`` seconds. ``job_overhead`` is the fixed
+    cost of a CI job (checkout, environment setup, dispatch). ``runners`` is the number
+    of jobs that can run at once (default: GitHub's limit, see default_runners).
+    """
+
+    recipes: Mapping[str, float]
+    default: float = 20.0
+    job_overhead: float = 30.0
+    runners: int = 20
+
+    @classmethod
+    def from_file(
+        cls, path: str | Path, runners: int | None = None, platform: str = ""
+    ) -> BuildTimings:
+        """Read a JSON file ``{"recipes": {name: seconds}, "default": s, "job_overhead": s}``."""
+        with Path(path).open(encoding="utf-8") as stream:
+            data = json.load(stream)
+        recipes = {
+            name: float(seconds)
+            for name, seconds in (data.get("recipes") or {}).items()
+        }
+        known = sorted(recipes.values())
+        default = data.get("default", known[len(known) // 2] if known else cls.default)
+        return cls(
+            recipes=recipes,
+            default=float(default),
+            job_overhead=float(data.get("job_overhead", cls.job_overhead)),
+            runners=runners or int(data.get("runners", default_runners(platform))),
+        )
+
+    def seconds(self, package: str) -> float:
+        return self.recipes.get(package, self.default)
+
+
+def _makespan(durations: list[float], runners: int) -> float:
+    """When the last of these jobs ends, started longest first on this many runners."""
+    free = [0.0] * min(runners, len(durations))
+    heapq.heapify(free)
+    for duration in sorted(durations, reverse=True):
+        heapq.heappush(free, heapq.heappop(free) + duration)
+    return max(free, default=0.0)
+
+
+def _pack(stage: list[str], jobs: int, timings: BuildTimings) -> list[list[str]]:
+    """Longest-processing-time packing of the stage's packages into this many jobs, each
+    job's packages kept in stage order."""
+    order = sorted(range(len(stage)), key=lambda i: (-timings.seconds(stage[i]), i))
+    loads = [(0.0, job) for job in range(jobs)]
+    heapq.heapify(loads)
+    members: list[list[int]] = [[] for _ in range(jobs)]
+    for index in order:
+        load, job = heapq.heappop(loads)
+        members[job].append(index)
+        heapq.heappush(loads, (load + timings.seconds(stage[index]), job))
+    return [[stage[i] for i in sorted(job)] for job in members if job]
+
+
+def _pack_by_time(
+    stage: list[str], max_batch_size: int, timings: BuildTimings
+) -> list[list[str]]:
+    """The stage's packages in as few jobs as reach the shortest stage (within 1%),
+    given the runners and the overhead of each job; no job gets more than
+    max_batch_size packages."""
+    fewest = -(-len(stage) // max_batch_size)
+    most = max(fewest, min(len(stage), 4 * timings.runners))
+    candidates = []
+    for jobs in range(fewest, most + 1):
+        packed = _pack(stage, jobs, timings)
+        if any(len(job) > max_batch_size for job in packed):
+            continue
+        durations = [
+            timings.job_overhead + sum(map(timings.seconds, job)) for job in packed
+        ]
+        candidates.append((_makespan(durations, timings.runners), packed))
+    if not candidates:
+        return _chunks(stage, max_batch_size)
+    # fewer jobs (a smaller workflow) when that costs at most 1%
+    shortest = min(span for span, _ in candidates)
+    return next(packed for span, packed in candidates if span <= shortest * 1.01)
+
+
 def batch_stages(
     stages: Iterable[Iterable[str]],
     max_batch_size: int = 5,
     *,
     config_path: str | Path = "vinca.yaml",
+    timings: BuildTimings | None = None,
+    followers: Mapping[str, list[str]] | None = None,
 ) -> list[list[list[str]]]:
     """Group dependency stages into CI batches without mutating the input.
 
     The outer list preserves sequential dependency stages. Each inner list contains
     batches which can execute in parallel.
+
+    Without timings, a stage is cut into jobs of max_batch_size packages in order, and
+    consecutive small stages share a job. With timings, each stage is packed into jobs
+    by measured build time (longest first), so that no single job holds up the stage;
+    consecutive stages that need just one job share it while it stays short.
+
+    followers maps a package to packages that are built right after it in its job (e.g.
+    a compatibility package that only depends on it): they leave their own stage, and
+    with timings their time counts towards their leader's.
     """
     if max_batch_size < 1:
         raise ValueError("max_batch_size must be at least 1")
 
+    stages = [list(stage) for stage in stages]
+    present = {package for stage in stages for package in stage}
+    follow = {
+        leader: [f for f in members if f in present]
+        for leader, members in (followers or {}).items()
+        if leader in present
+    }
+    following = {f for members in follow.values() for f in members}
+    if following:
+        stages = [[p for p in stage if p not in following] for stage in stages]
+        stages = [stage for stage in stages if stage]
+        if timings is not None:
+            recipes = dict(timings.recipes)
+            for leader, members in follow.items():
+                recipes[leader] = timings.seconds(leader) + sum(
+                    map(timings.seconds, members)
+                )
+            timings = replace(timings, recipes=recipes)
+
     with Path(config_path).open(encoding="utf-8") as stream:
         vinca_conf = yaml.safe_load(stream) or {}
     build_individually = set(
@@ -58,6 +186,30 @@ def batch_stages(
             for package in own_stage:
                 result.append([[package]])
 
+        if timings is not None:
+            if not stage:
+                continue
+            seconds = sum(map(timings.seconds, stage))
+            if seconds <= SHARED_JOB_SECONDS and len(stage) <= max_batch_size:
+                # a short stage: one job, shared with the stages before it while that stays short
+                if (
+                    pending
+                    and len(pending) + len(stage) <= max_batch_size
+                    and sum(map(timings.seconds, pending)) + seconds
+                    <= SHARED_JOB_SECONDS
+                ):
+                    pending.extend(stage)
+                    continue
+                if pending:
+                    result.append([pending])
+                pending = stage
+                continue
+            if pending:
+                result.append([pending])
+                pending = []
+            result.append(_pack_by_time(stage, max_batch_size, timings))
+            continue
+
         if len(stage) >= max_batch_size:
             if pending:
                 result.append([pending])
@@ -72,6 +224,11 @@ def batch_stages(
 
     if pending:
         result.append([pending])
+    if following:
+        result = [
+            [[q for p in job for q in [p, *follow.get(p, [])]] for job in stage]
+            for stage in result
+        ]
     return result
 
 

vinca/test_generate_gha.py

@@ -3,6 +3,7 @@
 import sys
 
 import pytest
+import yaml
 
 from vinca import config, generate_gha
 from vinca.generate_gha import (
@@ -194,3 +195,42 @@ def test_only_later_stages_declare_needs(tmp_path, monkeypatch):
         assert "needs" not in jobs["stage_0_job_0"]
         assert "needs" not in jobs["stage_0_job_1"]
         assert jobs["stage_1_job_2"]["needs"] == ["stage_0_job_0", "stage_0_job_1"]
+
+
+def test_compatibility_followers():
+    from vinca.generate_gha import compatibility_followers
+
+    assert compatibility_followers(
+        [["ros2-a", "ros2-distro-mutex"], ["ros-rolling-a", "ros-rolling-b"]]
+    ) == {"ros2-a": ["ros-rolling-a"]}
+
+
+def test_keep_file_adds_an_artifact_step(tmp_path, monkeypatch):
+    import argparse
+
+    outfile = tmp_path / "linux.yml"
+    monkeypatch.setattr(
+        config, "parsed_args", argparse.Namespace(keep_file="work/times.jsonl")
+    )
+
+    build_unix_pipeline(
+        [[["ros-rolling-rclcpp"]]], "buildbranch_linux", outfile=str(outfile)
+    )
+
+    step = yaml.safe_load(outfile.read_text())["jobs"]["stage_0_job_0"]["steps"][-1]
+    assert step["uses"].startswith("actions/upload-artifact@")
+    assert step["if"] == "always()" and step["with"]["path"] == "work/times.jsonl"
+
+
+def test_no_artifact_step_without_keep_file(tmp_path, monkeypatch):
+    import argparse
+
+    outfile = tmp_path / "linux.yml"
+    monkeypatch.setattr(config, "parsed_args", argparse.Namespace())
+
+    build_unix_pipeline(
+        [[["ros-rolling-rclcpp"]]], "buildbranch_linux", outfile=str(outfile)
+    )
+
+    steps = yaml.safe_load(outfile.read_text())["jobs"]["stage_0_job_0"]["steps"]
+    assert not any("upload-artifact" in step.get("uses", "") for step in steps)