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])