diff --git a/vinca/generate_gha.py b/vinca/generate_gha.py index 27f553a..248fe17 100644 --- a/vinca/generate_gha.py +++ b/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-- compatibility package only + depends on ros2- (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: diff --git a/vinca/pipeline.py b/vinca/pipeline.py index ab532d8..2c7e150 100644 --- a/vinca/pipeline.py +++ b/vinca/pipeline.py @@ -8,7 +8,10 @@ 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 @@ -16,25 +19,150 @@ 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 diff --git a/vinca/test_generate_gha.py b/vinca/test_generate_gha.py index 7226414..7a605ff 100644 --- a/vinca/test_generate_gha.py +++ b/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) diff --git a/vinca/test_pipeline.py b/vinca/test_pipeline.py index 17ceb15..670aa10 100644 --- a/vinca/test_pipeline.py +++ b/vinca/test_pipeline.py @@ -1,5 +1,5 @@ from vinca import generate_gha -from vinca.pipeline import batch_stages, get_all_ancestors +from vinca.pipeline import BuildTimings, batch_stages, get_all_ancestors def test_github_generator_loads_packaged_build_scripts(): @@ -81,3 +81,102 @@ def test_batch_stages_splits_large_stages(tmp_path): assert batch_stages([["a", "b", "c", "d", "e"]], 2, config_path=config) == [ [["a", "b"], ["c", "d"], ["e"]] ] + + +def test_batch_stages_packs_by_time(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + timings = BuildTimings( + recipes={"heavy": 600, "big": 300, "mid": 200}, + default=10, + job_overhead=0, + runners=4, + ) + stage = ["a", "heavy", "b", "big", "c", "mid", "d", "e"] + + # in order and by count, heavy and big would share a job; by time, heavy gets its + # own job and the rest fill the others (each job keeps the stage order) + batches = batch_stages([stage], 25, config_path=config, timings=timings) + + assert len(batches) == 1 + jobs = batches[0] + assert ["heavy"] in jobs + assert sorted(p for job in jobs for p in job) == sorted(stage) + longest = max(sum(timings.seconds(p) for p in job) for job in jobs) + assert longest == 600 + for job in jobs: + assert job == [p for p in stage if p in job] + + +def test_batch_stages_with_timings_respects_the_batch_size(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + timings = BuildTimings(recipes={}, default=1, job_overhead=30, runners=2) + stage = [f"p{i}" for i in range(10)] + + batches = batch_stages([stage], 4, config_path=config, timings=timings) + + assert len(batches) == 1 + assert all(len(job) <= 4 for job in batches[0]) + assert sorted(p for job in batches[0] for p in job) == sorted(stage) + + +def test_batch_stages_with_timings_shares_jobs_between_short_stages(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + timings = BuildTimings( + recipes={"long": 1000}, default=10, job_overhead=30, runners=8 + ) + + batches = batch_stages( + [["a"], ["b", "c"], ["long"], ["d"]], 25, config_path=config, timings=timings + ) + + # a, b and c are short, so they share one job; long has its own, then d + assert batches == [[["a", "b", "c"]], [["long"]], [["d"]]] + + +def test_build_timings_from_file(tmp_path): + path = tmp_path / "timings.json" + path.write_text('{"recipes": {"a": 5, "b": 7, "c": 100}, "job_overhead": 12}') + + timings = BuildTimings.from_file(path, runners=3) + + assert timings.seconds("c") == 100 + assert timings.seconds("unknown") == 7 # the median + assert (timings.job_overhead, timings.runners) == (12, 3) + # GitHub's default limits on concurrent jobs + assert BuildTimings.from_file(path, platform="linux-64").runners == 20 + assert BuildTimings.from_file(path, platform="osx-arm64").runners == 5 + + +def test_batch_stages_builds_followers_right_after_their_leader(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + timings = BuildTimings( + recipes={"ros2-a": 400, "ros2-b": 400}, default=1, job_overhead=0, runners=2 + ) + + # the compatibility packages leave their stage, which then disappears + batches = batch_stages( + [["ros2-a", "ros2-b"], ["ros-x-a", "ros-x-b"]], + 25, + config_path=config, + timings=timings, + followers={ + "ros2-a": ["ros-x-a"], + "ros2-b": ["ros-x-b"], + "ros2-gone": ["ros-x-gone"], + }, + ) + + assert batches == [[["ros2-a", "ros-x-a"], ["ros2-b", "ros-x-b"]]] + + +def test_batch_stages_keeps_followers_whose_leader_is_not_built(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + + assert batch_stages( + [["ros-x-a"]], 25, config_path=config, followers={"ros2-a": ["ros-x-a"]} + ) == [[["ros-x-a"]]]