From 951cdb6925d8f66b53dc559489d78a520635b9ac Mon Sep 17 00:00:00 2001 From: Tobias Fischer Date: Fri, 23 Oct 2026 19:11:31 +1000 Subject: [PATCH] vinca-gha: spend a job budget where it shortens the workflow most; keep it under GitHub's size limit batch_stages now has one way of batching, with or without measured times: each stage first gets as few jobs as --batch_size allows (consecutive short stages, one job of at most 2 minutes each, share a job up to --batch_size packages: they run one after another anyway), then, as long as the job budget lasts, the stage whose duration more jobs cut the most (per added job) gets them, its packages packed longest first, at most one job per runner. Without --timings every package counts the same (BuildTimings' default), which splits the largest stages first. The budget comes from GitHub's 500 KiB workflow limit: when the generated workflow is too large (WorkflowTooLarge), vinca-gha batches again with a budget scaled down by how much it's over, so callers no longer retry with larger batches or fewer runners themselves. Smaller jobs, so that more fit: the Windows jobs call the repository's .scripts/build_win.bat instead of carrying its text (2-3 KiB per job), and the Build steps no longer repeat the job's package list in their name. Simulated on RoboStack's jazzy full rebuild with its measured times (20 runners): linux-64 6.41 -> 2.60 h (214 jobs, 302 KiB); win-64 25.0 h (90 jobs, over the limit at batch size 25) -> 8.87 h (181 jobs, 493 KiB). Co-Authored-By: Claude Opus 5.5 --- vinca/generate_gha.py | 57 ++++++++--- vinca/pipeline.py | 192 ++++++++++++++++++++----------------- vinca/test_generate_gha.py | 5 +- vinca/test_pipeline.py | 57 ++++++++++- 4 files changed, 204 insertions(+), 107 deletions(-) diff --git a/vinca/generate_gha.py b/vinca/generate_gha.py index 248fe17..8aeedbc 100644 --- a/vinca/generate_gha.py +++ b/vinca/generate_gha.py @@ -16,6 +16,7 @@ from vinca.pipeline import ( BuildTimings, batch_stages, + default_runners, get_all_ancestors, get_skip_existing, ) @@ -182,8 +183,16 @@ def dump_for_gha(doc, f): workflow_size = os.path.getsize(f) if workflow_size > MAX_WORKFLOW_SIZE_BYTES: - raise RuntimeError( - f"Generated workflow {f} is {workflow_size / 1024:.1f} KiB, exceeding " + raise WorkflowTooLarge(workflow_size, f) + + +class WorkflowTooLarge(RuntimeError): + """GitHub refuses workflow files over MAX_WORKFLOW_SIZE_BYTES.""" + + def __init__(self, size: int, path: str): + self.size = size + super().__init__( + f"Generated workflow {path} is {size / 1024:.1f} KiB, exceeding " f"the {MAX_WORKFLOW_SIZE_BYTES / 1024:.0f} KiB limit. Increase " "--batch_size to reduce the number of generated jobs." ) @@ -324,7 +333,7 @@ def build_unix_pipeline( DISABLE_GIT_MAINTENANCE_STEP, get_setup_pixi_step(setup_pixi_version, pixi_version), { - "name": f"Build {' '.join([pkg for pkg in batch])}", + "name": "Build", "env": build_env, "run": script, }, @@ -423,14 +432,14 @@ def build_win_pipeline( if workflow is None: workflow = blurb - # The build script is the repository's own; it is inlined into the workflow. + # The build script is the repository's own: each job calls it from its checkout + # (inlining its text into every job made the workflow too large for many jobs). if not os.path.exists(WINDOWS_BUILD_SCRIPT): raise FileNotFoundError( f"{WINDOWS_BUILD_SCRIPT} not found in {os.getcwd()}: the Windows workflow " "runs the repository's build script, run vinca-gha where it exists" ) - with open(WINDOWS_BUILD_SCRIPT, encoding="utf-8") as fi: - script = lu(fi.read()) + script = "call " + WINDOWS_BUILD_SCRIPT.replace("/", "\\") prev_batch_keys = [] for i, s in enumerate(stages): @@ -467,7 +476,7 @@ def build_win_pipeline( "shell": "cmd", "run": script, "env": build_env, - "name": f"Build {' '.join([pkg for pkg in batch])}", + "name": "Build", }, *keep_file_steps(), ] @@ -618,13 +627,9 @@ def main(): 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 + else BuildTimings(recipes={}, runners=args.runners or default_runners(platform)) ) - print(stages) + followers = compatibility_followers(filtered_stages) with open("buildorder.txt", "w", encoding="utf-8") as fo: order = [] @@ -635,6 +640,32 @@ def main(): fo.write("\n".join(order)) + # as many jobs as fit GitHub's workflow size limit: when the workflow is too large, + # batch again with a job budget scaled down by how much it's over + budget = None + for _ in range(8): + stages = batch_stages( + filtered_stages, + args.batch_size, + timings=timings, + followers=followers, + job_budget=budget, + ) + try: + write_pipeline(args, stages, platform, setup_pixi_version, pixi_version) + break + except WorkflowTooLarge as error: + jobs = sum(len(stage) for stage in stages) + if budget is not None and jobs > budget: + raise # can't get smaller: the batch size limits the jobs + budget = max(1, int(jobs * MAX_WORKFLOW_SIZE_BYTES / error.size * 0.95)) + print(f"{error}; batching again with at most {budget} jobs") + else: + raise RuntimeError("could not fit the workflow into GitHub's size limit") + + +def write_pipeline(args, stages, platform, setup_pixi_version, pixi_version): + """Write the platform's staged workflow for these stages.""" if args.platform == "linux-64": build_unix_pipeline( stages, diff --git a/vinca/pipeline.py b/vinca/pipeline.py index 2c7e150..185b7c1 100644 --- a/vinca/pipeline.py +++ b/vinca/pipeline.py @@ -19,7 +19,7 @@ from vinca.utils import get_repodata -# with timings: consecutive single-job stages share a job while it stays this short +# a stage this short is one job, shared with the short stages next to it SHARED_JOB_SECONDS = 120.0 @@ -95,28 +95,12 @@ def _pack(stage: list[str], jobs: int, timings: BuildTimings) -> list[list[str]] 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 _span(jobs: list[list[str]], timings: BuildTimings) -> float: + """How long a stage of these jobs takes on the runners.""" + return _makespan( + [timings.job_overhead + sum(map(timings.seconds, job)) for job in jobs], + timings.runners, + ) def batch_stages( @@ -126,23 +110,28 @@ def batch_stages( config_path: str | Path = "vinca.yaml", timings: BuildTimings | None = None, followers: Mapping[str, list[str]] | None = None, + job_budget: int | 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. + Each stage first gets as few jobs as max_batch_size (packages per job) allows, and + consecutive short stages (one job of at most SHARED_JOB_SECONDS each) share a job. + Then, as long as the job budget lasts, the stage whose duration one more job cuts the + most gets it, its packages packed into its jobs longest first: the jobs go where they + shorten the workflow most, and a smaller budget (e.g. for GitHub's workflow size + limit) still leaves the slowest stages split. A stage gets at most one job per + runner. Without timings every package counts the same (BuildTimings' default). 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. + their time counts towards their leader's. """ if max_batch_size < 1: raise ValueError("max_batch_size must be at least 1") + timings = timings or BuildTimings(recipes={}) stages = [list(stage) for stage in stages] present = {package for stage in stages for package in stage} @@ -155,13 +144,12 @@ def batch_stages( 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) + 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 {} @@ -172,64 +160,94 @@ def batch_stages( or [] ) - result: list[list[list[str]]] = [] - pending: list[str] = [] - for original_stage in stages: - stage = list(original_stage) - own_stage = [package for package in stage if package in build_individually] - stage = [package for package in stage if package not in build_individually] - - if own_stage: - if pending: - result.append([pending]) - pending = [] - 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 + # units in order: (packages, splittable); a unit that isn't splittable is one job + units: list[tuple[list[str], bool]] = [] + shared: list[str] = [] # consecutive short stages sharing one job + + def flush() -> None: + nonlocal shared + if shared: + units.append((shared, False)) + shared = [] + + for stage in stages: + own = [p for p in stage if p in build_individually] + stage = [p for p in stage if p not in build_individually] + if own: + flush() + units.extend(([p], False) for p in own) + if not stage: + continue + short = sum(map(timings.seconds, stage)) <= SHARED_JOB_SECONDS + if short and len(stage) <= max_batch_size: + if len(shared) + len(stage) > max_batch_size: + flush() + shared = shared + stage + continue + flush() + units.append((stage, True)) + flush() + + # as few jobs as the batch size allows, then more where they cut the most time + jobs: list[list[list[str]]] = [] + for packages, splittable in units: + if not splittable: + jobs.append([packages]) + continue + count = -(-len(packages) // max_batch_size) + packed = _pack(packages, count, timings) + jobs.append( + packed + if all(len(job) <= max_batch_size for job in packed) + else _chunks(packages, max_batch_size) + ) + + def more(index: int) -> tuple[float, list[list[str]]] | None: + """The unit's next packing with more jobs that is faster (e.g. 8 equal packages + only get faster again at 8 jobs once there are 4), and the time it saves per + added job.""" + packages, splittable = units[index] + current = jobs[index] + most = min(len(packages), timings.runners) + if not splittable: + return None + span = _span(current, timings) + for count in range(len(current) + 1, most + 1): + packed = _pack(packages, count, timings) + if not all(len(job) <= max_batch_size for job in packed): continue - if pending: - result.append([pending]) - pending = [] - result.append(_pack_by_time(stage, max_batch_size, timings)) + saved = span - _span(packed, timings) + # not for a negligible gain: every job makes the workflow larger + if saved > 0.01 * span: + return saved / (len(packed) - len(current)), packed + return None + + total = sum(len(stage) for stage in jobs) + candidates = [] + for index in range(len(units)): + if (option := more(index)) is not None: + heapq.heappush(candidates, (-option[0], index)) + while candidates and (job_budget is None or total < job_budget): + _, index = heapq.heappop(candidates) + option = more(index) # recomputed: the unit's jobs may have changed + if option is None: + continue + if ( + job_budget is not None + and total + len(option[1]) - len(jobs[index]) > job_budget + ): continue + total += len(option[1]) - len(jobs[index]) + jobs[index] = option[1] + if (option := more(index)) is not None: + heapq.heappush(candidates, (-option[0], index)) - if len(stage) >= max_batch_size: - if pending: - result.append([pending]) - pending = [] - result.append(_chunks(stage, max_batch_size)) - elif len(pending) + len(stage) < max_batch_size: - pending.extend(stage) - else: - if pending: - result.append([pending]) - pending = stage - - if pending: - result.append([pending]) if following: - result = [ + jobs = [ [[q for p in job for q in [p, *follow.get(p, [])]] for job in stage] - for stage in result + for stage in jobs ] - return result + return jobs def get_skip_existing(vinca_conf: Mapping[str, Any], platform: str) -> list[dict]: diff --git a/vinca/test_generate_gha.py b/vinca/test_generate_gha.py index 7a605ff..0c1c89e 100644 --- a/vinca/test_generate_gha.py +++ b/vinca/test_generate_gha.py @@ -106,7 +106,7 @@ def test_setup_pixi_versions_must_not_be_empty(): get_setup_pixi_step(setup_pixi_version=" ") -def test_win_pipeline_inlines_the_repository_build_script(tmp_path, monkeypatch): +def test_win_pipeline_calls_the_repository_build_script(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) (tmp_path / ".scripts").mkdir() (tmp_path / ".scripts" / "build_win.bat").write_text("echo repository script\n") @@ -116,7 +116,8 @@ def test_win_pipeline_inlines_the_repository_build_script(tmp_path, monkeypatch) workflow = pytest.importorskip("yaml").safe_load(outfile.read_text()) steps = workflow["jobs"]["stage_0_job_0"]["steps"] - assert "echo repository script" in steps[-1]["run"] + assert steps[-1]["run"] == "call .scripts\\build_win.bat" + assert steps[-1]["env"]["CURRENT_RECIPES"] == "ros2-rclcpp" def test_win_pipeline_requires_the_repository_build_script(tmp_path, monkeypatch): diff --git a/vinca/test_pipeline.py b/vinca/test_pipeline.py index 670aa10..d36eb6f 100644 --- a/vinca/test_pipeline.py +++ b/vinca/test_pipeline.py @@ -56,10 +56,10 @@ def test_batch_stages_does_not_mutate_input(tmp_path): batches = batch_stages(stages, 3, config_path=config) assert stages == original + # consecutive short stages share a job, up to the batch size assert batches == [ [["ros-special"]], - [["ros-a"]], - [["ros-b", "ros-c"]], + [["ros-a", "ros-b", "ros-c"]], [["ros-d", "ros-e"]], ] @@ -78,9 +78,10 @@ def test_batch_stages_splits_large_stages(tmp_path): config = tmp_path / "vinca.yaml" config.write_text("{}\n") - assert batch_stages([["a", "b", "c", "d", "e"]], 2, config_path=config) == [ - [["a", "b"], ["c", "d"], ["e"]] - ] + # no more than the batch size per job; without timings every package counts the same + batches = batch_stages([["a", "b", "c", "d", "e"]], 2, config_path=config) + assert len(batches) == 1 and all(len(job) <= 2 for job in batches[0]) + assert sorted(p for job in batches[0] for p in job) == ["a", "b", "c", "d", "e"] def test_batch_stages_packs_by_time(tmp_path): @@ -136,6 +137,19 @@ def test_batch_stages_with_timings_shares_jobs_between_short_stages(tmp_path): assert batches == [[["a", "b", "c"]], [["long"]], [["d"]]] +def test_batch_stages_with_timings_shares_a_job_between_many_short_stages(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + timings = BuildTimings(recipes={}, default=100, job_overhead=30, runners=8) + + # each stage is short (100 s), together far longer: they still share jobs, up to + # the batch size, since they run one after another anyway + stages = [[f"p{i}"] for i in range(7)] + batches = batch_stages(stages, 3, config_path=config, timings=timings) + + assert batches == [[["p0", "p1", "p2"]], [["p3", "p4", "p5"]], [["p6"]]] + + 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}') @@ -180,3 +194,36 @@ def test_batch_stages_keeps_followers_whose_leader_is_not_built(tmp_path): assert batch_stages( [["ros-x-a"]], 25, config_path=config, followers={"ros2-a": ["ros-x-a"]} ) == [[["ros-x-a"]]] + + +def test_batch_stages_spends_the_job_budget_on_the_slowest_stages(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + slow = [f"s{i}" for i in range(8)] + fast = [f"f{i}" for i in range(8)] + timings = BuildTimings( + recipes={**dict.fromkeys(slow, 600), **dict.fromkeys(fast, 30)}, + job_overhead=10, + runners=8, + ) + + unlimited = batch_stages([slow, fast], 25, config_path=config, timings=timings) + budget = batch_stages( + [slow, fast], 25, config_path=config, timings=timings, job_budget=5 + ) + + # without a budget both stages are split (the fast one less); with 5 jobs, the slow + # stage gets 4 of them, the fast one stays in one job + assert len(unlimited[0]) == 8 and len(unlimited[1]) > 1 + assert [len(stage) for stage in budget] == [4, 1] + + +def test_batch_stages_without_timings_splits_big_stages(tmp_path): + config = tmp_path / "vinca.yaml" + config.write_text("{}\n") + + batches = batch_stages([[f"p{i}" for i in range(100)]], 25, config_path=config) + + # every package counts the same: as many jobs as runners help (at most 20) + assert 4 < len(batches[0]) <= 20 + assert all(len(job) <= 25 for job in batches[0])