From dd4707299fec2d3b7c51480cabd983c95ad05bd1 Mon Sep 17 00:00:00 2001 From: Bas Zalmstra <4995967+baszalmstra@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:20:06 +0200 Subject: [PATCH] feat: add dependency-aware workflow batching --- README.md | 50 ++ pyproject.toml | 1 + vinca/generate_gha.py | 116 ++++- vinca/test_generate_gha.py | 38 ++ vinca/test_workflow_batching.py | 347 ++++++++++++++ vinca/test_workflow_timings.py | 117 +++++ vinca/workflow_batching.py | 783 ++++++++++++++++++++++++++++++++ vinca/workflow_timings.py | 409 +++++++++++++++++ 8 files changed, 1856 insertions(+), 5 deletions(-) create mode 100644 vinca/test_workflow_batching.py create mode 100644 vinca/test_workflow_timings.py create mode 100644 vinca/workflow_batching.py create mode 100644 vinca/workflow_timings.py diff --git a/README.md b/README.md index 16cf6e7..bdf6f61 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,56 @@ pixi_version: v0.78.0 Pinning a release or full action commit hash is recommended for reproducible workflows. +## Configuring GitHub Actions batching + +GitHub Actions workflow generation keeps the existing stage-based batching behavior by default. To test dependency-aware batching, configure a strategy in `vinca.yaml`: + +```yaml +github_actions: + batching: + strategy: schedule-and-isolation + max_jobs: 120 + runner_count: 8 + job_overhead: 5.0 + maximum_batch_size: 30 + schedule_tolerance: 0.01 + failure_isolation_weight: 1.0 + build_backend_weights: + empty: 0.1 + ament_python: 1.0 + other: 2.0 + cmake: 3.0 + catkin: 4.0 + ament_cmake: 4.0 + package_weights: + ros2-rclcpp: 12.0 +``` + +The available strategies are: + +- `legacy` (default): preserves stage barriers and uses `vinca-gha --batch_size` as the maximum number of packages per batch. +- `schedule-and-isolation`: contracts the package dependency DAG to at most `max_jobs`, evaluates candidate contractions against `runner_count`, and penalizes artificial failure propagation. Generated jobs use the transitive reduction of the batch DAG for exact `needs` dependencies instead of stage-wide barriers. + +`maximum_batch_size` is optional. Vinca reports an error when the requested job limit cannot be reached without exceeding it or merging a package listed in `build_in_own_azure_stage`. `job_overhead` is an estimated fixed cost per job. `schedule_tolerance` groups scheduling estimates that differ by less than the configured fraction so failure isolation can decide between them. `failure_isolation_weight` controls that secondary preference. + +The default cost model treats empty compatibility recipes as cheapest, followed by `ament_python`, generic builds, CMake, and then `catkin`/`ament_cmake`. Requirement count, patches, and vendor packages adjust that base cost. `build_backend_weights` can calibrate the backend ratios, while `package_weights` takes precedence for packages with measured CI durations. + +Recipe-derived weights are only relative estimates. Compare generated workflows and measured CI timings before making the experimental strategy the default. New strategies can be added with `register_batching_strategy` from `vinca.workflow_batching`. + +Historical GitHub Actions timings can be converted into calibrated package weights from a generated distribution repository: + +```bash +vinca-gha-timings \ + --repository RoboStack/ros-jazzy \ + --workflow linux.yml \ + --branch buildbranch_linux \ + --runs 30 \ + --recipes recipes \ + --output workflow-weights.yaml +``` + +The helper reads GitHub's job and step timestamps, excludes failed or incomplete build steps, measures job overhead outside the generated build step, and fits positive additive package durations in minutes. Fixed setup and upload work inside that step is distributed across the package estimates. Multi-package jobs do not expose individual package timings, so the fit is regularized toward the recipe backend estimates. The resulting `github_actions.batching` section can be copied into `vinca.yaml`; `workflow_timing_metadata` records the source and sample counts. + ## Managing conda-forge pinning RoboStack repositories can keep their global pins reproducible without copying and diff --git a/pyproject.toml b/pyproject.toml index 50ab86b..5413b93 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,6 +50,7 @@ GitHub = "https://github.com/RoboStack/vinca" vinca = "vinca.main:main" vinca-glab = "vinca.generate_gitlab:main" vinca-gha = "vinca.generate_gha:main" +vinca-gha-timings = "vinca.workflow_timings:main" vinca-azure = "vinca.generate_azure:main" vinca-migrate = "vinca.migrate:main" vinca-snapshot = "vinca.snapshot:main" diff --git a/vinca/generate_gha.py b/vinca/generate_gha.py index 39ba6ee..d53b84c 100644 --- a/vinca/generate_gha.py +++ b/vinca/generate_gha.py @@ -7,7 +7,7 @@ import argparse from importlib import resources from distutils.dir_util import copy_tree -from typing import Any +from typing import Any, Sequence from rich import print @@ -27,6 +27,11 @@ get_conda_subdir, ) from vinca import config +from vinca.workflow_batching import ( + estimate_recipe_weights, + parse_workflow_batching_config, + plan_workflow_batches, +) # Use the v0 version of setup-pixi by default, which should give you the last major release @@ -99,7 +104,7 @@ def normalize_name(s): def batch_stages(stages, max_batch_size=5): with open("vinca.yaml", "r") as vinca_yaml: - vinca_conf = yaml.safe_load(vinca_yaml) + vinca_conf = yaml.safe_load(vinca_yaml) or {} # this reduces the number of individual builds to try to save some time stage_lengths = [len(s) for s in stages] @@ -254,6 +259,28 @@ def dump_for_gha(doc, f): ) +def _resolve_batch_needs( + batch_dependencies: Sequence[Sequence[int]], + batch_index: int, + emitted_batch_keys: Sequence[str], +) -> list[str]: + dependencies = batch_dependencies[batch_index] + needs = [] + for dependency in dependencies: + if ( + not isinstance(dependency, int) + or isinstance(dependency, bool) + or dependency < 0 + or dependency >= batch_index + ): + raise ValueError( + f"Batch {batch_index} has invalid dependency index {dependency!r}; " + "dependencies must refer to earlier batches" + ) + needs.append(emitted_batch_keys[dependency]) + return needs + + def get_stage_name(batch): legacy_prefix = f"ros-{config.ros_distro}-" stage_name = [] @@ -309,13 +336,24 @@ def build_unix_pipeline( target="", setup_pixi_version: str = DEFAULT_SETUP_PIXI_VERSION, pixi_version: str = DEFAULT_PIXI_VERSION, + batch_dependencies=None, ): + """Render a Unix workflow from package batches and optional exact dependencies.""" blurb = {"jobs": {}, "name": pipeline_name} if azure_template is None: azure_template = blurb + expected_batch_count = sum(len(stage) for stage in stages) + if ( + batch_dependencies is not None + and len(batch_dependencies) != expected_batch_count + ): + raise ValueError("batch_dependencies must contain one entry per generated job") + prev_batch_keys = [] + emitted_batch_keys = [] + batch_index = 0 for i, s in enumerate(stages): stage_name = f"stage_{i}" @@ -345,11 +383,18 @@ def build_unix_pipeline( }, ] + if batch_dependencies is None: + needs = prev_batch_keys + else: + needs = _resolve_batch_needs( + batch_dependencies, batch_index, emitted_batch_keys + ) + job = { "name": pretty_stage_name, "runs-on": runs_on, "strategy": {"fail-fast": False}, - "needs": prev_batch_keys, + "needs": needs, "steps": steps, } @@ -359,6 +404,8 @@ def build_unix_pipeline( } azure_template["jobs"][batch_key] = job + emitted_batch_keys.append(batch_key) + batch_index += 1 prev_batch_keys = batch_keys @@ -380,7 +427,9 @@ def build_linux_pipeline( pipeline_name="build_linux", setup_pixi_version: str = DEFAULT_SETUP_PIXI_VERSION, pixi_version: str = DEFAULT_PIXI_VERSION, + batch_dependencies=None, ): + """Render the Linux workflow with optional exact batch dependencies.""" build_unix_pipeline( stages, trigger_branch, @@ -392,6 +441,7 @@ def build_linux_pipeline( target="linux-64", setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) @@ -406,7 +456,9 @@ def build_osx_pipeline( pipeline_name="build_osx64", setup_pixi_version: str = DEFAULT_SETUP_PIXI_VERSION, pixi_version: str = DEFAULT_PIXI_VERSION, + batch_dependencies=None, ): + """Render the macOS workflow with optional exact batch dependencies.""" build_unix_pipeline( stages, trigger_branch, @@ -418,6 +470,7 @@ def build_osx_pipeline( pipeline_name=pipeline_name, setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) @@ -428,7 +481,9 @@ def build_win_pipeline( azure_template=None, setup_pixi_version: str = DEFAULT_SETUP_PIXI_VERSION, pixi_version: str = DEFAULT_PIXI_VERSION, + batch_dependencies=None, ): + """Render a Windows workflow from package batches and optional dependencies.""" vm_imagename = "windows-2022" # Build Win pipeline blurb = {"jobs": {}, "name": "build_win"} @@ -443,7 +498,16 @@ def build_win_pipeline( with open(".scripts/build_win.bat", "r") as fi: script = lu(fi.read()) + expected_batch_count = sum(len(stage) for stage in stages) + if ( + batch_dependencies is not None + and len(batch_dependencies) != expected_batch_count + ): + raise ValueError("batch_dependencies must contain one entry per generated job") + prev_batch_keys = [] + emitted_batch_keys = [] + batch_index = 0 for i, s in enumerate(stages): stage_name = f"stage_{i}" batch_keys = [] @@ -481,11 +545,18 @@ def build_win_pipeline( }, ] + if batch_dependencies is None: + needs = prev_batch_keys + else: + needs = _resolve_batch_needs( + batch_dependencies, batch_index, emitted_batch_keys + ) + job = { "name": pretty_stage_name, "runs-on": vm_imagename, "strategy": {"fail-fast": False}, - "needs": prev_batch_keys, + "needs": needs, "env": { "CONDA_BLD_PATH": "C:\\\\bld\\\\", "VINCA_CUSTOM_CMAKE_BUILD_DIR": "C:\\\\x\\\\", @@ -499,6 +570,8 @@ def build_win_pipeline( } azure_template["jobs"][batch_key] = job + emitted_batch_keys.append(batch_key) + batch_index += 1 prev_batch_keys = batch_keys @@ -624,7 +697,34 @@ def main(): if len(filtered): filtered_stages.append(filtered) - stages = batch_stages(filtered_stages, args.batch_size) + with open("vinca.yaml", "r") as vinca_yaml: + vinca_conf = yaml.safe_load(vinca_yaml) + batching_config = parse_workflow_batching_config(vinca_conf) + batch_dependencies = None + if batching_config.strategy == "legacy": + stages = batch_stages(filtered_stages, args.batch_size) + else: + names_to_build = {package for stage in filtered_stages for package in stage} + dependency_graph = nx.DiGraph() + dependency_graph.add_nodes_from(sorted(names_to_build)) + for consumer in sorted(names_to_build): + for dependency in requirements.get(consumer, ()): + if dependency in names_to_build: + dependency_graph.add_edge(dependency, consumer) + batch_plan = plan_workflow_batches( + filtered_stages, + dependency_graph, + estimate_recipe_weights( + metas, + batching_config.build_backend_weights, + batching_config.package_weights, + ), + vinca_conf.get("build_in_own_azure_stage", []), + args.batch_size, + batching_config, + ) + stages = batch_plan.as_stages() + batch_dependencies = batch_plan.dependencies print(stages) with open("buildorder.txt", "w") as fo: @@ -644,6 +744,7 @@ def main(): pipeline_name="build_linux64", setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) if args.platform == "osx-64": @@ -652,6 +753,7 @@ def main(): args.trigger_branch, setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) if args.platform == "osx-arm64": @@ -665,6 +767,7 @@ def main(): pipeline_name="build_osx_arm64", setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) if args.platform == "linux-aarch64": @@ -678,6 +781,7 @@ def main(): pipeline_name="build_linux_aarch64", setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) # windows @@ -688,6 +792,7 @@ def main(): outfile="win.yml", setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) if args.platform == "emscripten-wasm32": @@ -699,4 +804,5 @@ def main(): target="emscripten-wasm32", setup_pixi_version=setup_pixi_version, pixi_version=pixi_version, + batch_dependencies=batch_dependencies, ) diff --git a/vinca/test_generate_gha.py b/vinca/test_generate_gha.py index 4745084..df5c629 100644 --- a/vinca/test_generate_gha.py +++ b/vinca/test_generate_gha.py @@ -5,6 +5,7 @@ from vinca import config from vinca.generate_gha import ( build_unix_pipeline, + build_win_pipeline, get_setup_pixi_step, get_stage_name, ) @@ -38,6 +39,43 @@ def test_get_stage_name_joins_a_batch(): assert get_stage_name(batch) == "rclcpp ros2-ament-package" +@pytest.mark.parametrize("builder", [build_unix_pipeline, build_win_pipeline]) +def test_pipeline_uses_exact_batch_dependencies(builder, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + outfile = tmp_path / "workflow.yml" + + builder( + [ + [["ros-rolling-foundation"]], + [["ros-rolling-independent"]], + [["ros-rolling-consumer"]], + ], + "buildbranch", + outfile=outfile, + batch_dependencies=((), (), (0,)), + ) + + workflow = pytest.importorskip("yaml").safe_load(outfile.read_text()) + assert workflow["jobs"]["stage_0_job_0"]["needs"] == [] + assert workflow["jobs"]["stage_1_job_1"]["needs"] == [] + assert workflow["jobs"]["stage_2_job_2"]["needs"] == ["stage_0_job_0"] + + +@pytest.mark.parametrize("builder", [build_unix_pipeline, build_win_pipeline]) +def test_pipeline_rejects_dependencies_on_unemitted_jobs( + builder, tmp_path, monkeypatch +): + monkeypatch.chdir(tmp_path) + + with pytest.raises(ValueError, match="dependencies must refer to earlier batches"): + builder( + [[["ros-rolling-package"]]], + "buildbranch", + outfile=tmp_path / "workflow.yml", + batch_dependencies=((1,),), + ) + + def test_unix_pipeline_sets_up_pixi(tmp_path): outfile = tmp_path / "linux.yml" diff --git a/vinca/test_workflow_batching.py b/vinca/test_workflow_batching.py new file mode 100644 index 0000000..cc3f528 --- /dev/null +++ b/vinca/test_workflow_batching.py @@ -0,0 +1,347 @@ +"""Tests for configurable workflow batching strategies.""" + +import networkx as nx +import pytest + +from vinca.generate_gha import build_unix_pipeline, build_win_pipeline +from vinca.workflow_batching import ( + WorkflowBatchingConfig, + available_batching_strategies, + estimate_recipe_weights, + parse_workflow_batching_config, + plan_workflow_batches, +) + + +@pytest.mark.parametrize("vinca_config", [{}, None]) +def test_legacy_strategy_is_the_default(vinca_config): + batching_config = parse_workflow_batching_config(vinca_config) + + assert batching_config.strategy == "legacy" + assert available_batching_strategies() == ("legacy", "schedule-and-isolation") + + +def test_batching_config_rejects_a_non_mapping_root(): + with pytest.raises(ValueError, match="vinca.yaml must contain a mapping"): + parse_workflow_batching_config([]) + + +def test_legacy_strategy_preserves_stage_barriers(): + graph = nx.DiGraph() + plan = plan_workflow_batches( + [["one", "two"], ["three"]], + graph, + {}, + [], + 2, + WorkflowBatchingConfig(), + ) + + assert plan.batches == (("one", "two"), ("three",)) + assert plan.dependencies == ((), (0,)) + + +def test_recipe_weights_account_for_build_backend_and_overrides(): + recipes = [ + {"package": {"name": "compat"}}, + {"package": {"name": "explicit-empty"}, "build": {"script": ""}}, + { + "package": {"name": "python-package"}, + "build": {"script": "$RECIPE_DIR/build_ament_python.sh"}, + }, + { + "package": {"name": "cpp-package"}, + "build": {"script": "$RECIPE_DIR/build_ament_cmake.sh"}, + }, + { + "package": {"name": "cpp-vendor"}, + "build": {"script": "$RECIPE_DIR/build_ament_cmake.sh"}, + "source": {"patches": ["one.patch"]}, + }, + ] + + weights = estimate_recipe_weights( + recipes, + build_backend_weights={"ament_python": 1.5}, + package_weights={"cpp-package": 12.0}, + ) + + assert weights["compat"] == weights["explicit-empty"] + assert weights["compat"] < weights["python-package"] < weights["cpp-vendor"] + assert weights["cpp-package"] == 12.0 + + +def test_batching_config_accepts_weight_calibration(): + batching_config = parse_workflow_batching_config( + { + "github_actions": { + "batching": { + "strategy": "schedule-and-isolation", + "max_jobs": 10, + "build_backend_weights": {"ament_python": 2.5}, + "package_weights": {"ros2-rclcpp": 17.0}, + } + } + } + ) + + assert batching_config.build_backend_weights["ament_python"] == 2.5 + assert batching_config.package_weights == {"ros2-rclcpp": 17.0} + + +def test_schedule_strategy_requires_a_job_limit(): + with pytest.raises(ValueError, match="max_jobs is required"): + parse_workflow_batching_config( + {"github_actions": {"batching": {"strategy": "schedule-and-isolation"}}} + ) + + +def test_unknown_strategy_reports_available_choices(): + with pytest.raises( + ValueError, match="choose one of: legacy, schedule-and-isolation" + ): + parse_workflow_batching_config( + {"github_actions": {"batching": {"strategy": "unknown"}}} + ) + + +def test_schedule_strategy_caps_jobs_and_preserves_dependency_order(): + graph = nx.DiGraph( + [ + ("foundation", "left"), + ("foundation", "right"), + ("left", "join"), + ("right", "join"), + ("join", "consumer"), + ("independent", "consumer"), + ] + ) + batching_config = WorkflowBatchingConfig( + strategy="schedule-and-isolation", + max_jobs=3, + runner_count=2, + ) + + first = plan_workflow_batches( + [["foundation", "independent"], ["left", "right"], ["join"], ["consumer"]], + graph, + {package: 1.0 for package in graph}, + [], + 5, + batching_config, + ) + second = plan_workflow_batches( + [["foundation", "independent"], ["left", "right"], ["join"], ["consumer"]], + graph, + {package: 1.0 for package in graph}, + [], + 5, + batching_config, + ) + + assert first == second + assert len(first.batches) == 3 + assert sorted(package for batch in first.batches for package in batch) == sorted( + graph.nodes + ) + + package_to_batch = { + package: batch_index + for batch_index, batch in enumerate(first.batches) + for package in batch + } + batch_graph = nx.DiGraph() + batch_graph.add_nodes_from(range(len(first.batches))) + for batch_index, dependencies in enumerate(first.dependencies): + assert all(dependency < batch_index for dependency in dependencies) + batch_graph.add_edges_from( + (dependency, batch_index) for dependency in dependencies + ) + assert nx.is_directed_acyclic_graph(batch_graph) + + for dependency, consumer in graph.edges: + dependency_batch = package_to_batch[dependency] + consumer_batch = package_to_batch[consumer] + if dependency_batch != consumer_batch: + assert nx.has_path(batch_graph, dependency_batch, consumer_batch) + + for batch in first.batches: + positions = {package: index for index, package in enumerate(batch)} + for dependency, consumer in graph.subgraph(batch).edges: + assert positions[dependency] < positions[consumer] + + +def test_schedule_strategy_rejects_stage_graph_mismatches(): + batching_config = WorkflowBatchingConfig( + strategy="schedule-and-isolation", + max_jobs=1, + ) + + with pytest.raises(ValueError, match="different packages"): + plan_workflow_batches( + [["one"]], + nx.DiGraph(), + {"one": 1.0}, + [], + 5, + batching_config, + ) + + +def test_schedule_strategy_keeps_configured_packages_isolated(): + graph = nx.DiGraph([("foundation", "consumer"), ("side", "consumer")]) + batching_config = WorkflowBatchingConfig( + strategy="schedule-and-isolation", + max_jobs=2, + runner_count=2, + ) + + plan = plan_workflow_batches( + [["foundation", "side"], ["consumer"]], + graph, + {package: 1.0 for package in graph}, + ["foundation"], + 5, + batching_config, + ) + + assert ("foundation",) in plan.batches + + +def test_schedule_strategy_can_batch_disconnected_packages(): + graph = nx.DiGraph() + graph.add_nodes_from(("one", "two", "three")) + batching_config = WorkflowBatchingConfig( + strategy="schedule-and-isolation", + max_jobs=1, + runner_count=8, + ) + + plan = plan_workflow_batches( + [["one", "two", "three"]], + graph, + {package: 1.0 for package in graph}, + [], + 5, + batching_config, + ) + + assert plan.batches == (("one", "three", "two"),) + assert plan.dependencies == ((),) + + +@pytest.mark.parametrize("workflow_builder", [build_unix_pipeline, build_win_pipeline]) +def test_schedule_plan_renders_exact_unix_and_windows_needs( + workflow_builder, tmp_path, monkeypatch +): + monkeypatch.chdir(tmp_path) + yaml = pytest.importorskip("yaml") + graph = nx.DiGraph( + [ + ("foundation", "python-package"), + ("foundation", "cpp-package"), + ("python-package", "consumer"), + ("cpp-package", "consumer"), + ("independent", "consumer"), + ] + ) + recipes = [ + { + "package": {"name": "python-package"}, + "build": {"script": "$RECIPE_DIR/build_ament_python.sh"}, + }, + { + "package": {"name": "cpp-package"}, + "build": {"script": "$RECIPE_DIR/build_ament_cmake.sh"}, + }, + ] + batching_config = parse_workflow_batching_config( + { + "github_actions": { + "batching": { + "strategy": "schedule-and-isolation", + "max_jobs": 4, + "runner_count": 2, + "maximum_batch_size": 3, + "build_backend_weights": {"ament_cmake": 9.0}, + } + } + } + ) + weights = {package: 1.0 for package in graph} + weights.update( + estimate_recipe_weights( + recipes, + batching_config.build_backend_weights, + batching_config.package_weights, + ) + ) + plan = plan_workflow_batches( + [ + ["foundation", "independent"], + ["python-package", "cpp-package"], + ["consumer"], + ], + graph, + weights, + ["foundation"], + 5, + batching_config, + ) + + assert len(plan.batches) <= batching_config.max_jobs + assert ("foundation",) in plan.batches + assert weights["cpp-package"] > weights["python-package"] + + outfile = tmp_path / "workflow.yml" + workflow_builder( + plan.as_stages(), + "buildbranch", + outfile=outfile, + batch_dependencies=plan.dependencies, + ) + workflow = yaml.safe_load(outfile.read_text()) + job_keys = [f"stage_{index}_job_{index}" for index in range(len(plan.batches))] + for batch_index, dependencies in enumerate(plan.dependencies): + assert workflow["jobs"][job_keys[batch_index]]["needs"] == [ + job_keys[dependency] for dependency in dependencies + ] + + batch_graph = nx.DiGraph() + batch_graph.add_nodes_from(range(len(plan.batches))) + batch_graph.add_edges_from( + (dependency, batch_index) + for batch_index, dependencies in enumerate(plan.dependencies) + for dependency in dependencies + ) + package_to_batch = { + package: batch_index + for batch_index, batch in enumerate(plan.batches) + for package in batch + } + assert nx.is_directed_acyclic_graph(batch_graph) + for dependency, consumer in graph.edges: + if package_to_batch[dependency] != package_to_batch[consumer]: + assert nx.has_path( + batch_graph, package_to_batch[dependency], package_to_batch[consumer] + ) + + +def test_schedule_strategy_reports_an_impossible_batch_size_limit(): + graph = nx.DiGraph() + graph.add_nodes_from(("one", "two", "three")) + batching_config = WorkflowBatchingConfig( + strategy="schedule-and-isolation", + max_jobs=1, + maximum_batch_size=2, + ) + + with pytest.raises(ValueError, match="maximum_batch_size"): + plan_workflow_batches( + [["one", "two", "three"]], + graph, + {package: 1.0 for package in graph}, + [], + 5, + batching_config, + ) diff --git a/vinca/test_workflow_timings.py b/vinca/test_workflow_timings.py new file mode 100644 index 0000000..d5e4b4e --- /dev/null +++ b/vinca/test_workflow_timings.py @@ -0,0 +1,117 @@ +from collections import Counter + +import pytest +import ruamel.yaml + +from .workflow_timings import ( + BuildTiming, + estimate_job_overhead, + extract_build_timing, + fit_package_weights, + write_timing_config, +) + + +def _job(conclusion="success", build_conclusion="success"): + return { + "name": "python-package cpp-package", + "conclusion": conclusion, + "started_at": "2026-01-01T00:00:00Z", + "completed_at": "2026-01-01T00:12:00Z", + "steps": [ + { + "name": "Set up job", + "conclusion": "success", + "started_at": "2026-01-01T00:00:00Z", + "completed_at": "2026-01-01T00:01:00Z", + }, + { + "name": "Build python-package cpp-package", + "conclusion": build_conclusion, + "started_at": "2026-01-01T00:01:00Z", + "completed_at": "2026-01-01T00:11:00Z", + }, + ], + } + + +def test_extract_build_timing_uses_successful_build_step(): + timing = extract_build_timing(_job(), 42) + + assert timing == BuildTiming( + packages=("python-package", "cpp-package"), + build_seconds=600, + overhead_seconds=120, + run_id=42, + job_name="python-package cpp-package", + ) + + +@pytest.mark.parametrize( + ("job_conclusion", "build_conclusion"), + [("failure", "success"), ("success", "failure")], +) +def test_extract_build_timing_rejects_incomplete_builds( + job_conclusion, build_conclusion +): + assert extract_build_timing(_job(job_conclusion, build_conclusion), 42) is None + + +def test_estimate_job_overhead_uses_time_outside_build_step(): + timings = [ + BuildTiming(("one",), 240, 60, 1, "one"), + BuildTiming(("two",), 420, 60, 1, "two"), + BuildTiming(("one", "two"), 600, 60, 1, "one two"), + ] + + assert estimate_job_overhead(timings) == pytest.approx(1.0) + + +def test_fit_package_weights_uses_timings_and_backend_priors(): + timings = [ + BuildTiming(("python",), 60, 10, 1, "python"), + BuildTiming(("cpp",), 600, 10, 1, "cpp"), + BuildTiming(("python", "cpp"), 660, 10, 2, "python cpp"), + ] + + weights, sample_counts = fit_package_weights( + timings, + {"python": 1.0, "cpp": 4.0}, + regularization=0, + job_overhead=10 / 60, + ) + + assert weights == pytest.approx({"python": 1.0, "cpp": 10.0}) + assert sample_counts == {"cpp": 2, "python": 2} + + +def test_write_timing_config_emits_batching_configuration(tmp_path): + timing = BuildTiming(("python",), 60, 30, 7, "python") + output = tmp_path / "weights.yaml" + + write_timing_config( + output, + [timing], + {"python": 1.0}, + {"python": 1}, + "RoboStack/ros-jazzy", + "linux.yml", + 8, + 120, + Counter(success=1), + Counter(success=1), + 0.5, + ) + + yaml = ruamel.yaml.YAML(typ="safe") + document = yaml.load(output) + batching = document["github_actions"]["batching"] + assert batching == { + "strategy": "schedule-and-isolation", + "max_jobs": 120, + "runner_count": 8, + "job_overhead": 0.5, + "package_weights": {"python": 1.0}, + } + assert document["workflow_timing_metadata"]["runs"] == 1 + assert document["workflow_timing_metadata"]["run_conclusions"] == {"success": 1} diff --git a/vinca/workflow_batching.py b/vinca/workflow_batching.py new file mode 100644 index 0000000..10996c2 --- /dev/null +++ b/vinca/workflow_batching.py @@ -0,0 +1,783 @@ +"""Configurable package batching strategies for generated workflows.""" + +from __future__ import annotations + +import heapq +import json +import math +from collections import defaultdict +from dataclasses import dataclass, field +from typing import Callable, Mapping, Optional, Sequence + +import networkx as nx + + +DEFAULT_BUILD_BACKEND_WEIGHTS = { + "empty": 0.1, + "ament_python": 1.0, + "other": 2.0, + "cmake": 3.0, + "catkin": 4.0, + "ament_cmake": 4.0, +} + + +@dataclass(frozen=True) +class WorkflowBatchingConfig: + """Configure how generated recipes are grouped into workflow jobs.""" + + strategy: str = "legacy" + max_jobs: Optional[int] = None + runner_count: int = 8 + job_overhead: float = 5.0 + maximum_batch_size: Optional[int] = None + schedule_tolerance: float = 0.01 + failure_isolation_weight: float = 1.0 + build_backend_weights: Mapping[str, float] = field( + default_factory=lambda: dict(DEFAULT_BUILD_BACKEND_WEIGHTS) + ) + package_weights: Mapping[str, float] = field(default_factory=dict) + + +@dataclass(frozen=True) +class BatchPlan: + """Describe ordered package batches and their direct batch dependencies.""" + + batches: tuple[tuple[str, ...], ...] + dependencies: tuple[tuple[int, ...], ...] + + def as_stages(self) -> list[list[list[str]]]: + """Return one compatibility stage per batch for pipeline rendering.""" + return [[list(batch)] for batch in self.batches] + + +@dataclass(frozen=True) +class _StrategyInput: + stages: tuple[tuple[str, ...], ...] + dependency_graph: nx.DiGraph + weights: Mapping[str, float] + isolated_packages: frozenset[str] + legacy_batch_size: int + config: WorkflowBatchingConfig + + +BatchingStrategy = Callable[[_StrategyInput], BatchPlan] +_BATCHING_STRATEGIES: dict[str, BatchingStrategy] = {} + + +def register_batching_strategy( + name: str, +) -> Callable[[BatchingStrategy], BatchingStrategy]: + """Register a workflow batching strategy under a configuration name.""" + + def register(strategy: BatchingStrategy) -> BatchingStrategy: + if name in _BATCHING_STRATEGIES: + raise ValueError( + f"The workflow batching strategy {name!r} is already registered" + ) + _BATCHING_STRATEGIES[name] = strategy + return strategy + + return register + + +def available_batching_strategies() -> tuple[str, ...]: + """Return the deterministic list of configured workflow batching strategies.""" + return tuple(sorted(_BATCHING_STRATEGIES)) + + +def parse_workflow_batching_config( + vinca_config: Optional[Mapping[str, object]], +) -> WorkflowBatchingConfig: + """Parse and validate the GitHub Actions batching section of ``vinca.yaml``.""" + if vinca_config is None: + vinca_config = {} + if not isinstance(vinca_config, Mapping): + raise ValueError("vinca.yaml must contain a mapping") + github_actions = vinca_config.get("github_actions", {}) + if github_actions is None: + github_actions = {} + if not isinstance(github_actions, Mapping): + raise ValueError("github_actions must be a mapping") + + batching = github_actions.get("batching", {}) + if batching is None: + batching = {} + if not isinstance(batching, Mapping): + raise ValueError("github_actions.batching must be a mapping") + + strategy = batching.get("strategy", "legacy") + if not isinstance(strategy, str) or strategy not in _BATCHING_STRATEGIES: + choices = ", ".join(available_batching_strategies()) + raise ValueError( + f"Unknown GitHub Actions batching strategy {strategy!r}; choose one of: {choices}" + ) + + max_jobs = _optional_positive_integer(batching, "max_jobs") + runner_count = _positive_integer(batching, "runner_count", 8) + maximum_batch_size = _optional_positive_integer(batching, "maximum_batch_size") + job_overhead = _non_negative_number(batching, "job_overhead", 5.0) + schedule_tolerance = _non_negative_number(batching, "schedule_tolerance", 0.01) + failure_isolation_weight = _non_negative_number( + batching, "failure_isolation_weight", 1.0 + ) + build_backend_weights = dict(DEFAULT_BUILD_BACKEND_WEIGHTS) + build_backend_weights.update(_weight_mapping(batching, "build_backend_weights")) + unknown_backends = set(build_backend_weights) - set(DEFAULT_BUILD_BACKEND_WEIGHTS) + if unknown_backends: + names = ", ".join(sorted(unknown_backends)) + raise ValueError(f"Unknown build backend weight names: {names}") + package_weights = _weight_mapping(batching, "package_weights") + + if strategy != "legacy" and max_jobs is None: + raise ValueError( + f"github_actions.batching.max_jobs is required for strategy {strategy!r}" + ) + + return WorkflowBatchingConfig( + strategy=strategy, + max_jobs=max_jobs, + runner_count=runner_count, + job_overhead=job_overhead, + maximum_batch_size=maximum_batch_size, + schedule_tolerance=schedule_tolerance, + failure_isolation_weight=failure_isolation_weight, + build_backend_weights=build_backend_weights, + package_weights=package_weights, + ) + + +def plan_workflow_batches( + stages: Sequence[Sequence[str]], + dependency_graph: nx.DiGraph, + weights: Mapping[str, float], + isolated_packages: Sequence[str], + legacy_batch_size: int, + batching_config: WorkflowBatchingConfig, +) -> BatchPlan: + """Create a deterministic batch plan with the selected registered strategy.""" + if batching_config.strategy != "legacy": + if not nx.is_directed_acyclic_graph(dependency_graph): + raise ValueError("The package dependency graph must be acyclic") + stage_packages = [package for stage in stages for package in stage] + if len(stage_packages) != len(set(stage_packages)): + raise ValueError("Workflow stages must not contain duplicate packages") + graph_packages = set(dependency_graph) + if set(stage_packages) != graph_packages: + missing = sorted(set(stage_packages) - graph_packages) + unexpected = sorted(graph_packages - set(stage_packages)) + raise ValueError( + "Workflow stages and dependency graph contain different packages: " + f"missing from graph={missing}, missing from stages={unexpected}" + ) + + strategy_input = _StrategyInput( + stages=tuple(tuple(stage) for stage in stages), + dependency_graph=dependency_graph.copy(), + weights=weights, + isolated_packages=frozenset(isolated_packages), + legacy_batch_size=legacy_batch_size, + config=batching_config, + ) + return _BATCHING_STRATEGIES[batching_config.strategy](strategy_input) + + +def estimate_recipe_weights( + recipes: Sequence[Mapping[str, object]], + build_backend_weights: Optional[Mapping[str, float]] = None, + package_weights: Optional[Mapping[str, float]] = None, +) -> dict[str, float]: + """Estimate build costs by backend, recipe complexity, and explicit overrides.""" + backend_weights = dict(DEFAULT_BUILD_BACKEND_WEIGHTS) + if build_backend_weights is not None: + backend_weights.update(build_backend_weights) + explicit_weights = package_weights or {} + weights = {} + for recipe in recipes: + package = recipe.get("package") + if not isinstance(package, Mapping): + continue + name = package.get("name") + if not isinstance(name, str): + continue + if name in explicit_weights: + weights[name] = float(explicit_weights[name]) + continue + + backend = _recipe_build_backend(recipe) + requirement_count = _recipe_requirement_count(recipe) + patch_count = _recipe_patch_count(recipe) + backend_cost = backend_weights.get(backend, backend_weights["other"]) + vendor_multiplier = 1.5 if "vendor" in name else 1.0 + weights[name] = ( + backend_cost * vendor_multiplier + + min(requirement_count, 40) * 0.03 + + patch_count * 0.2 + ) + return weights + + +def _recipe_build_backend(recipe: Mapping[str, object]) -> str: + build = recipe.get("build") + if build is None: + return "empty" + if not isinstance(build, Mapping): + return "other" + script = build.get("script") + if script is None or script == "" or script == []: + return "empty" + script_text = json.dumps(script, sort_keys=True).lower() + for backend in ("ament_python", "ament_cmake", "catkin", "cmake"): + if backend in script_text: + return backend + return "other" + + +def _recipe_requirement_count(recipe: Mapping[str, object]) -> int: + requirement_sections = [] + requirements = recipe.get("requirements") + if isinstance(requirements, Mapping): + requirement_sections.append(requirements) + outputs = recipe.get("outputs") + if isinstance(outputs, Sequence) and not isinstance(outputs, (str, bytes)): + for output in outputs: + if isinstance(output, Mapping) and isinstance( + output.get("requirements"), Mapping + ): + requirement_sections.append(output["requirements"]) + + count = 0 + for requirement_section in requirement_sections: + for section_name in ("host", "run"): + values = requirement_section.get(section_name, ()) + if isinstance(values, Sequence) and not isinstance(values, (str, bytes)): + count += len(values) + return count + + +def _recipe_patch_count(recipe: Mapping[str, object]) -> int: + sources = recipe.get("source", ()) + if isinstance(sources, Mapping): + sources = (sources,) + if not isinstance(sources, Sequence) or isinstance(sources, (str, bytes)): + return 0 + count = 0 + for source in sources: + if not isinstance(source, Mapping): + continue + patches = source.get("patches", ()) + if isinstance(patches, Sequence) and not isinstance(patches, (str, bytes)): + count += len(patches) + return count + + +def _positive_integer(config: Mapping[str, object], key: str, default: int) -> int: + value = config.get(key, default) + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise ValueError(f"github_actions.batching.{key} must be a positive integer") + return value + + +def _optional_positive_integer(config: Mapping[str, object], key: str) -> Optional[int]: + value = config.get(key) + if value is None: + return None + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise ValueError(f"github_actions.batching.{key} must be a positive integer") + return value + + +def _weight_mapping(config: Mapping[str, object], key: str) -> dict[str, float]: + value = config.get(key, {}) + if not isinstance(value, Mapping): + raise ValueError(f"github_actions.batching.{key} must be a mapping") + weights = {} + for name, weight in value.items(): + if not isinstance(name, str): + raise ValueError(f"github_actions.batching.{key} keys must be strings") + if ( + not isinstance(weight, (int, float)) + or isinstance(weight, bool) + or not math.isfinite(weight) + or weight <= 0 + ): + raise ValueError( + f"github_actions.batching.{key}.{name} must be a positive number" + ) + weights[name] = float(weight) + return weights + + +def _non_negative_number( + config: Mapping[str, object], key: str, default: float +) -> float: + value = config.get(key, default) + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(value) + or value < 0 + ): + raise ValueError(f"github_actions.batching.{key} must be non-negative") + return float(value) + + +def _plan_from_stages(stages: Sequence[Sequence[Sequence[str]]]) -> BatchPlan: + batches = [] + dependencies = [] + previous_stage = [] + for stage in stages: + current_stage = [] + for batch in stage: + current_stage.append(len(batches)) + batches.append(tuple(batch)) + dependencies.append(tuple(previous_stage)) + previous_stage = current_stage + return BatchPlan(tuple(batches), tuple(dependencies)) + + +@register_batching_strategy("legacy") +def _legacy_strategy(strategy_input: _StrategyInput) -> BatchPlan: + stages = [list(stage) for stage in strategy_input.stages] + stage_lengths = [len(stage) for stage in stages] + merged_stages = [] + current_stage = [] + + def chunks(values: list[str], size: int): + for index in range(0, len(values), size): + yield values[index : index + size] + + for index, stage in enumerate(stages): + for package in sorted(strategy_input.isolated_packages): + if package in stage: + merged_stages.append([[package]]) + stage.remove(package) + + if ( + stage_lengths[index] < strategy_input.legacy_batch_size + and len(current_stage) + stage_lengths[index] + < strategy_input.legacy_batch_size + ): + current_stage += stage + else: + if current_stage: + merged_stages.append([current_stage]) + current_stage = [] + if stage_lengths[index] < strategy_input.legacy_batch_size: + current_stage += stage + else: + merged_stages.append( + list(chunks(stage, strategy_input.legacy_batch_size)) + ) + if current_stage: + merged_stages.append([current_stage]) + return _plan_from_stages(merged_stages) + + +def _list_schedule( + graph: nx.DiGraph, weights: Mapping[int, float], runner_count: int +) -> tuple[dict[int, float], dict[int, float], dict[int, list[int]], float]: + order = list(nx.lexicographical_topological_sort(graph, key=lambda node: node)) + upward_rank = {} + for node in reversed(order): + upward_rank[node] = weights[node] + max( + (upward_rank[successor] for successor in graph.successors(node)), + default=0.0, + ) + + indegrees = {node: graph.in_degree(node) for node in graph} + ready = [(-upward_rank[node], node) for node in graph if indegrees[node] == 0] + heapq.heapify(ready) + idle_runners = list(range(runner_count)) + heapq.heapify(idle_runners) + running = [] + starts = {} + finishes = {} + assignments = defaultdict(list) + current_time = 0.0 + + while ready or running: + while ready and idle_runners: + _, node = heapq.heappop(ready) + runner = heapq.heappop(idle_runners) + starts[node] = current_time + finishes[node] = current_time + weights[node] + assignments[runner].append(node) + heapq.heappush(running, (finishes[node], runner, node)) + if not running: + raise RuntimeError("The workflow scheduler stalled on an invalid graph") + current_time = running[0][0] + completed = [] + while running and running[0][0] <= current_time + 1e-9: + _, runner, node = heapq.heappop(running) + heapq.heappush(idle_runners, runner) + completed.append(node) + for node in completed: + for successor in graph.successors(node): + indegrees[successor] -= 1 + if indegrees[successor] == 0: + heapq.heappush(ready, (-upward_rank[successor], successor)) + + return starts, finishes, assignments, current_time + + +def _schedule_metrics( + graph: nx.DiGraph, weights: Mapping[int, float] +) -> tuple[dict[int, float], dict[int, float]]: + order = list(nx.lexicographical_topological_sort(graph, key=lambda node: node)) + finish = {} + for node in order: + start = max((finish[pred] for pred in graph.predecessors(node)), default=0.0) + finish[node] = start + weights[node] + tail = {} + for node in reversed(order): + tail[node] = max( + ( + weights[successor] + tail[successor] + for successor in graph.successors(node) + ), + default=0.0, + ) + return finish, tail + + +def _is_safe_contraction(graph: nx.DiGraph, source: int, target: int) -> bool: + source_reaches_target = nx.has_path(graph, source, target) + target_reaches_source = nx.has_path(graph, target, source) + if source_reaches_target and not graph.has_edge(source, target): + return False + if target_reaches_source and not graph.has_edge(target, source): + return False + if source_reaches_target: + graph.remove_edge(source, target) + alternate_path = nx.has_path(graph, source, target) + graph.add_edge(source, target) + return not alternate_path + if target_reaches_source: + graph.remove_edge(target, source) + alternate_path = nx.has_path(graph, target, source) + graph.add_edge(target, source) + return not alternate_path + return True + + +def _merge_graph_nodes( + graph: nx.DiGraph, source: int, target: int, merged: int +) -> None: + predecessors = ( + set(graph.predecessors(source)) | set(graph.predecessors(target)) + ) - { + source, + target, + } + successors = (set(graph.successors(source)) | set(graph.successors(target))) - { + source, + target, + } + graph.remove_nodes_from((source, target)) + graph.add_node(merged) + graph.add_edges_from((predecessor, merged) for predecessor in predecessors) + graph.add_edges_from((merged, successor) for successor in successors) + + +@register_batching_strategy("schedule-and-isolation") +def _schedule_and_isolation_strategy(strategy_input: _StrategyInput) -> BatchPlan: + config = strategy_input.config + if config.max_jobs is None: + raise ValueError("schedule-and-isolation requires max_jobs") + + package_names = tuple( + nx.lexicographical_topological_sort( + strategy_input.dependency_graph, key=lambda package: package + ) + ) + if len(package_names) <= config.max_jobs: + batches = tuple((package,) for package in package_names) + package_to_batch = { + package: index for index, package in enumerate(package_names) + } + return _finalize_plan( + batches, strategy_input.dependency_graph, package_to_batch + ) + + package_indexes = {package: index for index, package in enumerate(package_names)} + graph = nx.relabel_nodes( + strategy_input.dependency_graph, package_indexes, copy=True + ) + graph = nx.transitive_reduction(graph) + package_count = len(package_names) + original_ancestors = { + package_indexes[package]: { + package_indexes[ancestor] + for ancestor in nx.ancestors(strategy_input.dependency_graph, package) + } + for package in package_names + } + original_descendants = { + package_indexes[package]: { + package_indexes[descendant] + for descendant in nx.descendants(strategy_input.dependency_graph, package) + } + for package in package_names + } + source_hubs = { + node + for node in graph + if graph.in_degree(node) >= 12 + or graph.out_degree(node) >= 12 + or len(original_descendants[node]) >= package_count // 2 + } + weights = { + package_indexes[package]: float(strategy_input.weights.get(package, 1.0)) + for package in package_names + } + members = {node: frozenset((node,)) for node in graph} + ancestor_signatures = { + node: frozenset(original_ancestors[node] | {node}) for node in graph + } + descendant_signatures = { + node: frozenset(original_descendants[node] | {node}) for node in graph + } + contains_hub = {node: node in source_hubs for node in graph} + contains_isolated = { + node: package_names[node] in strategy_input.isolated_packages for node in graph + } + next_node = package_count + + runner_counts = tuple( + sorted( + { + min(config.max_jobs, max(1, config.runner_count // 2)), + min(config.max_jobs, config.runner_count), + min(config.max_jobs, config.runner_count * 2), + } + ) + ) + + while graph.number_of_nodes() > config.max_jobs: + reduced = nx.transitive_reduction(graph) + scheduled_weights = { + node: weights[node] + config.job_overhead for node in reduced + } + critical_finish, tail = _schedule_metrics(reduced, scheduled_weights) + schedules = {} + candidate_kinds = {} + + for runner_count in runner_counts: + starts, finishes, assignments, makespan = _list_schedule( + reduced, scheduled_weights, runner_count + ) + schedules[runner_count] = (starts, finishes, makespan) + for runner_jobs in assignments.values(): + for source, target in zip(runner_jobs, runner_jobs[1:]): + pair = tuple(sorted((source, target))) + candidate_kinds[pair] = min(candidate_kinds.get(pair, 3), 2) + + for source, target in reduced.edges: + pair = tuple(sorted((source, target))) + is_series = ( + reduced.out_degree(source) == 1 and reduced.in_degree(target) == 1 + ) + candidate_kinds[pair] = min( + candidate_kinds.get(pair, 3), 0 if is_series else 1 + ) + + by_sole_successor = defaultdict(list) + for node in reduced: + successors = tuple(reduced.successors(node)) + if len(successors) == 1: + by_sole_successor[successors[0]].append(node) + for group in by_sole_successor.values(): + ordered = sorted(group, key=lambda node: (weights[node], node)) + for source, target in zip(ordered, ordered[1:]): + pair = tuple(sorted((source, target))) + candidate_kinds[pair] = min(candidate_kinds.get(pair, 3), 2) + + candidates = [] + reach_fallback_candidates = [] + for (source, target), kind in candidate_kinds.items(): + if contains_isolated[source] or contains_isolated[target]: + continue + merged_size = len(members[source]) + len(members[target]) + if ( + config.maximum_batch_size is not None + and merged_size > config.maximum_batch_size + ): + continue + if not _is_safe_contraction(reduced, source, target): + continue + + source_descendants = descendant_signatures[source] + target_descendants = descendant_signatures[target] + descendant_similarity = _jaccard_similarity( + source_descendants, target_descendants + ) + source_ancestors = ancestor_signatures[source] + target_ancestors = ancestor_signatures[target] + ancestor_similarity = _jaccard_similarity( + source_ancestors, target_ancestors + ) + comparable = reduced.has_edge(source, target) or reduced.has_edge( + target, source + ) + requires_reach_fallback = not comparable and ( + descendant_similarity < 0.5 or ancestor_similarity < 0.5 + ) + + external_predecessors = ( + set(reduced.predecessors(source)) | set(reduced.predecessors(target)) + ) - {source, target} + external_successors = ( + set(reduced.successors(source)) | set(reduced.successors(target)) + ) - {source, target} + merged_start = max( + (critical_finish[pred] for pred in external_predecessors), default=0.0 + ) + merged_finish = ( + merged_start + weights[source] + weights[target] + config.job_overhead + ) + merged_tail = max( + ( + scheduled_weights[successor] + tail[successor] + for successor in external_successors + ), + default=0.0, + ) + candidate_path = merged_finish + merged_tail + schedule_penalty = 0.0 + for starts, finishes, makespan in schedules.values(): + overlap = max( + 0.0, + min(finishes[source], finishes[target]) + - max(starts[source], starts[target]), + ) + predicted_delay = max(0.0, candidate_path - makespan) + schedule_penalty = max( + schedule_penalty, + (predicted_delay + overlap) / max(1.0, makespan), + ) + + reach_distance = 1.0 - (descendant_similarity + ancestor_similarity) / 2.0 + artificial_pairs = sum( + len(members[target] - original_descendants[package] - {package}) + for package in members[source] + ) + sum( + len(members[source] - original_descendants[package] - {package}) + for package in members[target] + ) + blast_penalty = reach_distance + artificial_pairs / max(1, merged_size) + spurious_waits = sum( + not (members[predecessor] & original_ancestors[package]) + for package in members[source] | members[target] + for predecessor in external_predecessors + ) + possible_waits = merged_size * len(external_predecessors) + spurious_wait_severity = ( + spurious_waits / possible_waits if possible_waits else 0.0 + ) + isolation_penalty = config.failure_isolation_weight * ( + blast_penalty + 2.0 * spurious_wait_severity + ) + schedule_bucket = math.floor( + schedule_penalty / max(config.schedule_tolerance, 1e-9) + ) + candidate = ( + 0 if kind == 0 else 1, + 0 if kind == 0 else int(contains_hub[source] or contains_hub[target]), + schedule_bucket, + isolation_penalty, + schedule_penalty, + len(external_predecessors) * len(external_successors), + weights[source] + weights[target], + source, + target, + ) + if requires_reach_fallback: + reach_fallback_candidates.append(candidate) + else: + candidates.append(candidate) + + if not candidates: + candidates = reach_fallback_candidates + candidates.sort() + merged_this_round = 0 + for candidate in candidates: + source, target = candidate[-2:] + if graph.number_of_nodes() <= config.max_jobs: + break + if source not in graph or target not in graph: + continue + if not _is_safe_contraction(graph, source, target): + continue + _merge_graph_nodes(graph, source, target, next_node) + weights[next_node] = weights[source] + weights[target] + members[next_node] = members[source] | members[target] + ancestor_signatures[next_node] = ( + ancestor_signatures[source] | ancestor_signatures[target] + ) + descendant_signatures[next_node] = ( + descendant_signatures[source] | descendant_signatures[target] + ) + contains_hub[next_node] = contains_hub[source] or contains_hub[target] + contains_isolated[next_node] = False + next_node += 1 + merged_this_round += 1 + + if merged_this_round == 0: + raise ValueError( + "Cannot satisfy github_actions.batching.max_jobs without exceeding " + "maximum_batch_size or merging an isolated package" + ) + + batches_by_node = { + node: tuple( + nx.lexicographical_topological_sort( + strategy_input.dependency_graph.subgraph( + package_names[index] for index in members[node] + ), + key=lambda package: package, + ) + ) + for node in graph + } + quotient_order = tuple( + nx.lexicographical_topological_sort( + graph, key=lambda node: batches_by_node[node] + ) + ) + batches = tuple(batches_by_node[node] for node in quotient_order) + package_to_batch = { + package: batch_index + for batch_index, batch in enumerate(batches) + for package in batch + } + return _finalize_plan(batches, strategy_input.dependency_graph, package_to_batch) + + +def _jaccard_similarity(left: frozenset[int], right: frozenset[int]) -> float: + union = left | right + if not union: + return 1.0 + return len(left & right) / len(union) + + +def _finalize_plan( + batches: tuple[tuple[str, ...], ...], + dependency_graph: nx.DiGraph, + package_to_batch: Mapping[str, int], +) -> BatchPlan: + quotient = nx.DiGraph() + quotient.add_nodes_from(range(len(batches))) + quotient.add_edges_from( + (package_to_batch[source], package_to_batch[target]) + for source, target in dependency_graph.edges + if package_to_batch[source] != package_to_batch[target] + ) + if not nx.is_directed_acyclic_graph(quotient): + raise RuntimeError("The workflow batching strategy produced a cyclic plan") + reduced = nx.transitive_reduction(quotient) + dependencies = tuple( + tuple(sorted(reduced.predecessors(batch_index))) + for batch_index in range(len(batches)) + ) + return BatchPlan(batches, dependencies) diff --git a/vinca/workflow_timings.py b/vinca/workflow_timings.py new file mode 100644 index 0000000..c4a9507 --- /dev/null +++ b/vinca/workflow_timings.py @@ -0,0 +1,409 @@ +"""Extract empirical package build weights from GitHub Actions history.""" + +from __future__ import annotations + +import argparse +import os +import statistics +import subprocess +import time +from collections import Counter, defaultdict +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Mapping, Optional, Sequence +from urllib.parse import quote + +import requests +import ruamel.yaml + +from vinca.workflow_batching import estimate_recipe_weights + + +@dataclass(frozen=True) +class BuildTiming: + """Represent one successful workflow build step and its measured durations.""" + + packages: tuple[str, ...] + build_seconds: float + overhead_seconds: float + run_id: int + job_name: str + + +class GitHubActionsClient: + """Read paginated workflow run and job metadata from the GitHub REST API.""" + + def __init__(self, repository: str, token: Optional[str] = None): + if repository.count("/") != 1: + raise ValueError("repository must use the OWNER/NAME form") + self.repository = repository + self.session = requests.Session() + self.session.headers.update( + { + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "vinca-workflow-timings", + } + ) + if token: + self.session.headers["Authorization"] = f"Bearer {token}" + + def workflow_runs( + self, workflow: str, limit: int, branch: Optional[str] = None + ) -> list[Mapping[str, object]]: + """Return the newest completed runs for one workflow file or ID.""" + params = {"status": "completed"} + if branch: + params["branch"] = branch + endpoint = f"/repos/{self.repository}/actions/workflows/{quote(workflow, safe='')}/runs" + return self._paginated(endpoint, "workflow_runs", limit, params) + + def run_jobs(self, run_id: int) -> list[Mapping[str, object]]: + """Return every job from the latest attempt of a workflow run.""" + endpoint = f"/repos/{self.repository}/actions/runs/{run_id}/jobs" + return self._paginated(endpoint, "jobs", None, {"filter": "latest"}) + + def _paginated( + self, + endpoint: str, + collection_key: str, + limit: Optional[int], + params: Mapping[str, object], + ) -> list[Mapping[str, object]]: + results = [] + page = 1 + while limit is None or len(results) < limit: + request_params = dict(params) + request_params.update({"page": page, "per_page": 100}) + response = self._get(endpoint, request_params) + payload = response.json() + page_results = payload.get(collection_key, []) + if not isinstance(page_results, list): + raise RuntimeError( + f"GitHub returned an invalid {collection_key!r} collection" + ) + results.extend(page_results) + if len(page_results) < 100: + break + page += 1 + return results if limit is None else results[:limit] + + def _get(self, endpoint: str, params: Mapping[str, object]) -> requests.Response: + response = None + for attempt in range(5): + response = self.session.get( + f"https://api.github.com{endpoint}", + params=params, + timeout=60, + ) + if response.status_code not in {429, 500, 502, 503, 504}: + response.raise_for_status() + return response + time.sleep(2**attempt) + if response is None: + raise RuntimeError("GitHub request was not attempted") + response.raise_for_status() + return response + + +def extract_build_timing( + job: Mapping[str, object], run_id: int +) -> Optional[BuildTiming]: + """Extract a successful ``Build …`` step from one GitHub Actions job.""" + if job.get("conclusion") != "success": + return None + steps = job.get("steps") + if not isinstance(steps, Sequence) or isinstance(steps, (str, bytes)): + return None + build_step = next( + ( + step + for step in steps + if isinstance(step, Mapping) + and isinstance(step.get("name"), str) + and step["name"].startswith("Build ") + ), + None, + ) + if build_step is None or build_step.get("conclusion") != "success": + return None + packages = tuple(build_step["name"].removeprefix("Build ").split()) + if not packages: + return None + build_seconds = _duration_seconds(build_step) + job_seconds = _duration_seconds(job) + if build_seconds is None or job_seconds is None or build_seconds <= 0: + return None + return BuildTiming( + packages=packages, + build_seconds=build_seconds, + overhead_seconds=max(0.0, job_seconds - build_seconds), + run_id=run_id, + job_name=str(job.get("name", "")), + ) + + +def collect_build_timings( + client: GitHubActionsClient, + workflow: str, + run_limit: int, + branch: Optional[str] = None, +) -> tuple[list[BuildTiming], Counter, Counter]: + """Collect successful build observations and run and job conclusion counts.""" + timings = [] + run_conclusions = Counter() + job_conclusions = Counter() + for run in client.workflow_runs(workflow, run_limit, branch): + run_conclusions[str(run.get("conclusion"))] += 1 + run_id = run.get("id") + if not isinstance(run_id, int): + continue + for job in client.run_jobs(run_id): + job_conclusions[str(job.get("conclusion"))] += 1 + timing = extract_build_timing(job, run_id) + if timing is not None: + timings.append(timing) + return timings, run_conclusions, job_conclusions + + +def estimate_job_overhead(timings: Sequence[BuildTiming]) -> float: + """Estimate fixed job minutes spent outside the generated build step.""" + if not timings: + raise ValueError("No successful build timings were found") + return statistics.median(timing.overhead_seconds / 60.0 for timing in timings) + + +def fit_package_weights( + timings: Sequence[BuildTiming], + prior_weights: Mapping[str, float], + regularization: float = 3.0, + iterations: int = 60, + job_overhead: float = 0.0, +) -> tuple[dict[str, float], dict[str, int]]: + """Fit positive additive package minutes with ridge-regularized coordinate descent.""" + if not timings: + raise ValueError("No successful build timings were found") + if regularization < 0: + raise ValueError("regularization must be non-negative") + + grouped_durations = defaultdict(list) + for timing in timings: + grouped_durations[tuple(sorted(set(timing.packages)))].append( + max( + 0.01, + (timing.build_seconds + timing.overhead_seconds) / 60.0 - job_overhead, + ) + ) + observations = [ + (packages, statistics.median(durations), len(durations)) + for packages, durations in sorted(grouped_durations.items()) + ] + observed_packages = sorted( + {package for packages, _, _ in observations for package in packages} + ) + scale_candidates = [] + for packages, duration, sample_count in observations: + prior_sum = sum( + max(0.01, prior_weights.get(package, 1.0)) for package in packages + ) + scale_candidates.extend([duration / prior_sum] * sample_count) + prior_scale = statistics.median(scale_candidates) + scaled_priors = { + package: max(0.01, prior_weights.get(package, 1.0) * prior_scale) + for package in observed_packages + } + weights = dict(scaled_priors) + + observations_by_package = defaultdict(list) + for observation_index, (packages, _, _) in enumerate(observations): + for package in packages: + observations_by_package[package].append(observation_index) + + for _ in range(iterations): + for package in observed_packages: + numerator = regularization * scaled_priors[package] + denominator = regularization + for observation_index in observations_by_package[package]: + packages, duration, sample_count = observations[observation_index] + other_weight = sum( + weights[other] for other in packages if other != package + ) + numerator += sample_count * (duration - other_weight) + denominator += sample_count + weights[package] = max(0.01, numerator / max(denominator, 1e-9)) + + sample_counts = { + package: sum( + observations[index][2] for index in observations_by_package[package] + ) + for package in observed_packages + } + return weights, sample_counts + + +def load_recipe_priors(recipes_directory: Path) -> dict[str, float]: + """Load recipe files and derive backend-aware prior package weights.""" + yaml = ruamel.yaml.YAML(typ="safe") + recipes = [] + for path in sorted(recipes_directory.glob("**/recipe.yaml")): + with path.open("r", encoding="utf-8") as stream: + recipe = yaml.load(stream) + if isinstance(recipe, Mapping): + recipes.append(recipe) + return estimate_recipe_weights(recipes) + + +def write_timing_config( + output_path: Path, + timings: Sequence[BuildTiming], + package_weights: Mapping[str, float], + sample_counts: Mapping[str, int], + repository: str, + workflow: str, + runner_count: int, + max_jobs: int, + run_conclusions: Mapping[str, int], + job_conclusions: Mapping[str, int], + job_overhead: float, +) -> None: + """Write a config-compatible YAML snippet and bounded provenance metadata.""" + document = { + "github_actions": { + "batching": { + "strategy": "schedule-and-isolation", + "max_jobs": max_jobs, + "runner_count": runner_count, + "job_overhead": round(job_overhead, 4), + "package_weights": { + package: round(weight, 4) + for package, weight in sorted(package_weights.items()) + }, + } + }, + "workflow_timing_metadata": { + "repository": repository, + "workflow": workflow, + "observations": len(timings), + "runs": len({timing.run_id for timing in timings}), + "run_ids": sorted({timing.run_id for timing in timings}), + "run_conclusions": dict(sorted(run_conclusions.items())), + "job_conclusions": dict(sorted(job_conclusions.items())), + "package_samples": dict(sorted(sample_counts.items())), + }, + } + yaml = ruamel.yaml.YAML() + yaml.indent(mapping=2, sequence=4, offset=2) + with output_path.open("w", encoding="utf-8") as stream: + yaml.dump(document, stream) + + +def _duration_seconds(item: Mapping[str, object]) -> Optional[float]: + started_at = item.get("started_at") + completed_at = item.get("completed_at") + if not isinstance(started_at, str) or not isinstance(completed_at, str): + return None + start = datetime.fromisoformat(started_at.replace("Z", "+00:00")) + end = datetime.fromisoformat(completed_at.replace("Z", "+00:00")) + return (end - start).total_seconds() + + +def _github_token() -> Optional[str]: + token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") + if token: + return token + try: + result = subprocess.run( + ["gh", "auth", "token"], + check=True, + capture_output=True, + text=True, + timeout=15, + ) + except ( + FileNotFoundError, + subprocess.CalledProcessError, + subprocess.TimeoutExpired, + ): + return None + return result.stdout.strip() or None + + +def parse_command_line(arguments: Optional[Sequence[str]] = None) -> argparse.Namespace: + """Parse command-line options for historical workflow timing extraction.""" + parser = argparse.ArgumentParser( + description="Fit Vinca package weights from GitHub Actions job timings" + ) + parser.add_argument("--repository", required=True, help="GitHub OWNER/NAME") + parser.add_argument( + "--workflow", + default="linux.yml", + help="Workflow file name or numeric workflow ID", + ) + parser.add_argument("--branch", help="Optional workflow run branch") + parser.add_argument( + "--runs", type=int, default=30, help="Completed runs to inspect" + ) + parser.add_argument( + "--recipes", type=Path, default=Path("recipes"), help="Generated recipes" + ) + parser.add_argument("--output", type=Path, default=Path("workflow-weights.yaml")) + parser.add_argument("--runner-count", type=int, default=8) + parser.add_argument("--max-jobs", type=int, default=120) + parser.add_argument("--regularization", type=float, default=3.0) + parsed = parser.parse_args(arguments) + for name in ("runs", "runner_count", "max_jobs"): + if getattr(parsed, name) <= 0: + parser.error(f"--{name.replace('_', '-')} must be positive") + if parsed.regularization < 0: + parser.error("--regularization must be non-negative") + return parsed + + +def main(arguments: Optional[Sequence[str]] = None) -> None: + """Extract workflow timings and write package weights for Vinca batching.""" + parsed = parse_command_line(arguments) + client = GitHubActionsClient(parsed.repository, _github_token()) + timings, run_conclusions, job_conclusions = collect_build_timings( + client, parsed.workflow, parsed.runs, parsed.branch + ) + priors = load_recipe_priors(parsed.recipes) + job_overhead = estimate_job_overhead(timings) + package_weights, sample_counts = fit_package_weights( + timings, + priors, + parsed.regularization, + job_overhead=job_overhead, + ) + if priors: + package_weights = { + package: weight + for package, weight in package_weights.items() + if package in priors + } + sample_counts = { + package: count + for package, count in sample_counts.items() + if package in priors + } + write_timing_config( + parsed.output, + timings, + package_weights, + sample_counts, + parsed.repository, + parsed.workflow, + parsed.runner_count, + parsed.max_jobs, + run_conclusions, + job_conclusions, + job_overhead, + ) + print( + f"Wrote {len(package_weights)} package weights from {len(timings)} " + f"successful jobs to {parsed.output}" + ) + + +if __name__ == "__main__": + main()