Optimize embedded Sirius MO-scan batching and host preparation

optools/sirius_acceptance_test.py

@@ -228,6 +228,27 @@ def records(manifest, offset):
 
 
 class AcceptanceTest(unittest.TestCase):
+    def test_historical_v2_archive_preserves_missing_execution_profile(self):
+        inputs = evidence()
+        identities = [
+            inputs[0]["provenance"],
+            inputs[2]["provenance"],
+            *(item["provenance"] for item in inputs[4]),
+            *(item["provenance"] for item in inputs[5:]),
+        ]
+        for identity in identities:
+            identity["config"].pop("gomaxprocs", None)
+            identity["config"].pop("native_host_block_bytes", None)
+        original = copy.deepcopy(inputs)
+        self.assertTrue(acceptance.validate(*inputs)["local_acceptance_complete"])
+        self.assertEqual(original, inputs, "verification must not invent settings")
+
+        inputs = evidence()
+        for key in ("gomaxprocs", "native_host_block_bytes"):
+            del inputs[0]["provenance"]["config"][key]
+        with self.assertRaisesRegex(campaign.CampaignError, "semantic inputs differ"):
+            acceptance.validate(*inputs)
+
     def test_cli_binds_consumed_inputs_to_archive_checksums(self):
         inputs = evidence()
         with tempfile.TemporaryDirectory() as directory:

optools/sirius_campaign.py

@@ -854,7 +854,19 @@ def write_timing_tables(output, campaign, records, summary):
     write_private(output / "timings.md", "\n".join(lines) + "\n")
 
 
-def native_configuration(workspace):
+def execution_environment(gomaxprocs):
+    if type(gomaxprocs) is not int or not 0 < gomaxprocs <= v2.MAX_GOMAXPROCS:
+        raise v2.CampaignError(
+            "campaign Go CPU budget must be between 1 and 2147483647"
+        )
+    environment = os.environ.copy()
+    environment["GOMAXPROCS"] = str(gomaxprocs)
+    return environment
+
+
+def native_configuration(workspace, host_block_bytes=65536):
+    if type(host_block_bytes) is not int or host_block_bytes not in (65536, 1048576):
+        raise v2.CampaignError("unsupported campaign host block size")
     return f"""sirius:
   space:
     gpu:
@@ -863,6 +875,7 @@ def native_configuration(workspace):
     host:
       - numa_id: -1
         memory_capacity: {16 << 30}
+        block_size: {host_block_bytes}
   telemetry:
     output_directory: {json.dumps(str(workspace/'native-telemetry'))}
 """
@@ -1089,6 +1102,14 @@ def run_complete(arguments, *, provenance=None):
     ]
     if base_runner.get("session") != expected_session:
         raise v2.CampaignError("live campaign session differs from frozen provenance")
+    campaign = v2.build_campaign(
+        provenance,
+        resource=arguments.resource,
+        flight_profiles=[
+            {"id": route, "coverage": [], "status": "N/A", "reason": "not-provisioned"}
+            for route in ("flight-mo", "flight-tae")
+        ],
+    )
     sirius = tomllib.loads((workspace / "cn-2.toml").read_text())["cn"]["sirius"]
     if any(
         sirius.get(key) != value
@@ -1100,21 +1121,12 @@ def run_complete(arguments, *, provenance=None):
             "cleanup-timeout": "30s",
         }.items()
     ) or Path(sirius["native-config-path"]).read_text() != native_configuration(
-        workspace
+        workspace, provenance["config"]["native_host_block_bytes"]
     ):
         raise v2.CampaignError(
             "live campaign configuration differs from frozen profile"
         )
-    campaign = v2.build_campaign(
-        provenance,
-        resource=arguments.resource,
-        flight_profiles=[
-            {"id": route, "coverage": [], "status": "N/A", "reason": "not-provisioned"}
-            for route in ("flight-mo", "flight-tae")
-        ],
-    )
-    environment = os.environ.copy()
-    environment["GOMAXPROCS"] = "4"
+    environment = execution_environment(provenance["config"]["gomaxprocs"])
     run_workspace = workspace / f"campaign-{uuid.uuid4().hex}"
     run_workspace.mkdir(mode=0o700)
     (run_workspace / "scratch").mkdir(mode=0o700)
@@ -1521,13 +1533,12 @@ def excluded(specs, phase_name, base):
 
 
 def start_cluster(arguments):
+    environment = execution_environment(arguments.gomaxprocs)
     workspace = arguments.workspace.resolve()
     workspace.mkdir(mode=0o700, parents=True)
     native = workspace / "sirius.yaml"
-    write_private(native, native_configuration(workspace))
+    write_private(native, native_configuration(workspace, arguments.host_block_bytes))
     configs = make_configs(workspace, native, 2, arguments.port_base)
-    environment = os.environ.copy()
-    environment["GOMAXPROCS"] = "4"
     services = []
     previous_subreaper = v2.legacy._enter_exclusive_subreaper()
     try:
@@ -1578,6 +1589,8 @@ def freeze_provenance(arguments):
     """Capture actual binaries, guarded SDK, source data and local toolchain."""
     import sirius_sdk
 
+    execution_environment(arguments.gomaxprocs)
+    native_configuration(arguments.source_root, arguments.host_block_bytes)
     root = arguments.source_root.resolve()
     packaged = json.loads((root / "lib/sirius-provenance.json").read_text())
     if packaged["binary_sha256"] != sha256(arguments.service):
@@ -1668,6 +1681,8 @@ def command(*args):
             "gdb": command("gdb", "--version").splitlines()[0],
         },
         "config": {
+            "gomaxprocs": arguments.gomaxprocs,
+            "native_host_block_bytes": arguments.host_block_bytes,
             "gpu_memory_bytes": 6 << 30,
             "native_host_memory_bytes": 16 << 30,
             "input_window_bytes": 64 << 20,
@@ -1702,6 +1717,10 @@ def main(argv=None):
     serve.add_argument("--service", type=Path, required=True)
     serve.add_argument("--source-root", type=Path, required=True)
     serve.add_argument("--port-base", type=int, default=36001)
+    serve.add_argument("--gomaxprocs", type=int, required=True)
+    serve.add_argument(
+        "--host-block-bytes", type=int, choices=(65536, 1048576), default=65536
+    )
     run = sub.add_parser("run")
     run.add_argument("--workspace", type=Path, required=True)
     run.add_argument("--service", type=Path, required=True)
@@ -1724,6 +1743,10 @@ def main(argv=None):
     freeze.add_argument("--datasets", type=Path, required=True)
     freeze.add_argument("--loaded", type=Path, nargs=2, required=True)
     freeze.add_argument("--output", type=Path, required=True)
+    freeze.add_argument("--gomaxprocs", type=int, required=True)
+    freeze.add_argument(
+        "--host-block-bytes", type=int, choices=(65536, 1048576), default=65536
+    )
     args = parser.parse_args(argv)
 
     def interrupted(*_):

optools/sirius_campaign_test.py

@@ -36,6 +36,70 @@
 
 
 class ControllerTest(unittest.TestCase):
+    def test_execution_environment_replays_frozen_cpu_budget(self):
+        with patch.dict(os.environ, {"GOMAXPROCS": "99"}):
+            for budget in (16, (1 << 31) - 1):
+                environment = controller.execution_environment(budget)
+                self.assertEqual(environment["GOMAXPROCS"], str(budget))
+            self.assertEqual(os.environ["GOMAXPROCS"], "99")
+        for value in (0, -1, True, 1.5, "16", None, 1 << 31, 10**100):
+            with self.subTest(value=value), self.assertRaises(v2.CampaignError):
+                controller.execution_environment(value)
+
+    def test_native_configuration_records_supported_block_size(self):
+        workspace = Path("/tmp/campaign-profile-fixture")
+        self.assertIn("block_size: 65536", controller.native_configuration(workspace))
+        self.assertIn(
+            "block_size: 1048576", controller.native_configuration(workspace, 1048576)
+        )
+        for value in (0, -1, True, 65536.0, 32768, 2097152):
+            with self.subTest(value=value), self.assertRaises(v2.CampaignError):
+                controller.native_configuration(workspace, value)
+
+    def test_frozen_block_drift_rejects_before_service_launch(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            workspace = root / "workspace"
+            workspace.mkdir()
+            service, runner = root / "service", root / "runner"
+            service.write_text("fixture service")
+            runner.write_text("fixture runner")
+            identity = provenance()
+            identity["service_binary_sha256"] = controller.sha256(service)
+            identity["runner_binary_sha256"] = controller.sha256(runner)
+            identity["controller_sha256"] = controller.controller_sha256()
+            (workspace / "runner.json").write_text(
+                json.dumps(
+                    {
+                        "session": [
+                            "set session mo_sirius_campaign_evidence=1",
+                            "set session div_precision_increment=4",
+                            "set time_zone='+00:00'",
+                        ]
+                    }
+                )
+            )
+            native = workspace / "sirius.yaml"
+            native.write_text(controller.native_configuration(workspace, 1048576))
+            (workspace / "cn-2.toml").write_text(
+                '[cn.sirius]\nenabled=true\nbackend="embedded"\ninput-mode="mo"\n'
+                'max-waiting-queries=16\ncleanup-timeout="30s"\n'
+                + "native-config-path="
+                + json.dumps(str(native))
+                + "\n"
+            )
+            args = argparse.Namespace(
+                workspace=workspace,
+                output=root / "output",
+                service=service,
+                runner=runner,
+                resource=False,
+            )
+            with patch.object(controller, "Service") as launch:
+                with self.assertRaisesRegex(v2.CampaignError, "configuration differs"):
+                    controller.run_complete(args, provenance=identity)
+                launch.assert_not_called()
+
     def test_service_generation_failure_retains_spawn_cleanup(self):
         with tempfile.TemporaryDirectory() as directory:
             child = MagicMock(pid=123)

optools/sirius_campaign_v2.py

@@ -44,6 +44,10 @@
 ROW_BYTES = 1 << 20
 OUTPUT_BYTES = 1 << 20
 STDERR_BYTES = 64 << 10
+# The Go runtime parses GOMAXPROCS as a signed 32-bit integer. Larger values
+# are ignored, so accepting one would record a CPU budget the service never uses.
+MAX_GOMAXPROCS = (1 << 31) - 1
+EXECUTION_PROFILE_FIELDS = frozenset(("gomaxprocs", "native_host_block_bytes"))
 ROUTES = (
     {"id": "mo-native", "backend": "native", "scan_mode": "mo"},
     {"id": "embedded-mo", "backend": "embedded", "scan_mode": "mo"},
@@ -91,7 +95,24 @@ def query_uuid(value):
     return value
 
 
-def build_campaign(provenance, *, resource=False, flight_profiles=()):
+def validate_execution_profile(config):
+    if not isinstance(config, dict) or not EXECUTION_PROFILE_FIELDS <= config.keys():
+        raise CampaignError("campaign execution requires an explicit CPU and host profile")
+    if (
+        type(config["gomaxprocs"]) is not int
+        or not 0 < config["gomaxprocs"] <= MAX_GOMAXPROCS
+    ):
+        raise CampaignError("invalid campaign Go CPU budget")
+    if (
+        type(config["native_host_block_bytes"]) is not int
+        or config["native_host_block_bytes"] not in (65536, 1048576)
+    ):
+        raise CampaignError("invalid campaign host block size")
+
+
+def build_campaign(
+    provenance, *, resource=False, flight_profiles=(), require_execution_profile=True
+):
     required = (
         "matrixone_revision",
         "sirius_revision",
@@ -161,6 +182,11 @@ def build_campaign(provenance, *, resource=False, flight_profiles=()):
         ),
     ):
         value = provenance[field]
+        if field == "config" and isinstance(value, dict):
+            # V2 archives predating explicit profiles retain their original
+            # shape. Partial or invalid profiles never become historical data.
+            if require_execution_profile or EXECUTION_PROFILE_FIELDS & value.keys():
+                allowed |= EXECUTION_PROFILE_FIELDS
         if not isinstance(value, dict) or set(value) != allowed:
             raise CampaignError("invalid bounded provenance fields")
         if len(canonical(value)) > 4096:
@@ -173,6 +199,9 @@ def build_campaign(provenance, *, resource=False, flight_profiles=()):
         "time_zone": "+00:00",
     }:
         raise CampaignError("campaign session semantics differ")
+    config = provenance["config"]
+    if require_execution_profile or EXECUTION_PROFILE_FIELDS & config.keys():
+        validate_execution_profile(config)
     expected = {
         "gpu_memory_bytes": 6 << 30,
         "native_host_memory_bytes": 16 << 30,
@@ -1013,6 +1042,7 @@ def execute_campaign(campaign, executor, scratch):
     identities = set()
     produced = set()
     try:
+        validate_execution_profile(campaign["provenance"]["config"])
         for spec in campaign["schedule"]:
             request = copy.deepcopy(spec)
             request["spool_budget_bytes"] = min(
@@ -1101,6 +1131,7 @@ def summarize(campaign, records):
         campaign["provenance"],
         resource=campaign["resource_campaign"],
         flight_profiles=campaign["flight_profiles"],
+        require_execution_profile=False,
     )
     if campaign != trusted:
         raise CampaignError("summary rejects a partial or modified acceptance schedule")

optools/sirius_campaign_v2_test.py

@@ -50,6 +50,8 @@ def provenance():
             "gdb": "test",
         },
         "config": {
+            "gomaxprocs": 4,
+            "native_host_block_bytes": 65536,
             "gpu_memory_bytes": 6 << 30,
             "native_host_memory_bytes": 16 << 30,
             "input_window_bytes": 64 << 20,
@@ -147,6 +149,29 @@ def test_provenance_rejects_secrets_and_changed_limits(self):
         with self.assertRaises(campaign.CampaignError):
             campaign.build_campaign(value)
 
+    def test_execution_profile_requires_explicit_bounded_settings(self):
+        for budget, block in ((4, 1048576), (16, 65536), ((1 << 31) - 1, 65536)):
+            value = provenance()
+            value["config"].update(gomaxprocs=budget, native_host_block_bytes=block)
+            frozen = campaign.build_campaign(value)["provenance"]["config"]
+            self.assertEqual(frozen["gomaxprocs"], budget)
+            self.assertEqual(frozen["native_host_block_bytes"], block)
+        for key, values in (
+            ("gomaxprocs", (None, True, 0, -1, "16", 16.0, 1 << 31, 10**100)),
+            ("native_host_block_bytes", (None, True, 0, 32768, 65536.0)),
+        ):
+            for setting in values:
+                value = provenance()
+                value["config"][key] = setting
+                with self.subTest(key=key, setting=setting), self.assertRaises(
+                    campaign.CampaignError
+                ):
+                    campaign.build_campaign(value)
+            value = provenance()
+            del value["config"][key]
+            with self.subTest(missing=key), self.assertRaises(campaign.CampaignError):
+                campaign.build_campaign(value)
+
     def test_schedule_counts_and_sampling_separation(self):
         timing = campaign.build_campaign(provenance())
         resources = campaign.build_campaign(provenance(), resource=True)
@@ -159,6 +184,43 @@ def test_schedule_counts_and_sampling_separation(self):
         )
         self.assertEqual((0, 2, 1, 3), campaign.ORDER_KEYS[2])
 
+    def test_historical_profile_cannot_launch_an_executor(self):
+        manifest = campaign.build_campaign(provenance())
+        for key in ("gomaxprocs", "native_host_block_bytes"):
+            del manifest["provenance"]["config"][key]
+        with self.assertRaises(campaign.CampaignError):
+            campaign.build_campaign(manifest["provenance"])
+        calls = []
+
+        def executor(spec):
+            calls.append(spec)
+            return {}
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            root.chmod(0o700)
+            with self.assertRaises(campaign.CampaignError):
+                campaign.execute_campaign(manifest, executor, campaign.Scratch(root))
+            self.assertEqual([], calls, "historical verification cannot admit execution")
+            self.assertEqual([], list(root.iterdir()))
+
+    def test_archive_verification_rejects_partial_or_invalid_profiles(self):
+        for key, values in (
+            ("gomaxprocs", (None, True, 0, 1 << 31)),
+            ("native_host_block_bytes", (None, True, 32768, 65536.0)),
+        ):
+            for setting in values:
+                manifest = campaign.build_campaign(provenance())
+                manifest["provenance"]["config"][key] = setting
+                with self.subTest(key=key, setting=setting), self.assertRaises(
+                    campaign.CampaignError
+                ):
+                    campaign.summarize(manifest, [])
+            manifest = campaign.build_campaign(provenance())
+            del manifest["provenance"]["config"][key]
+            with self.subTest(missing=key), self.assertRaises(campaign.CampaignError):
+                campaign.summarize(manifest, [])
+
     def test_full_value_duplicate_and_empty_comparison(self):
         for a, b, keys, ok in (
             ([], [], [], True),