From bec5d507cf38bc500ad29e64fad5ce2af4b8ce5f Mon Sep 17 00:00:00 2001 From: Yuma Takino <36617106+tappe9@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:25:16 +0900 Subject: [PATCH 1/4] chore: prepare isolated read-only Flask investigation workspace --- .github/workflows/flask-investigation.yml | 31 +++++++++++++++++++++++ 1 file changed, 31 insertions(+) create mode 100644 .github/workflows/flask-investigation.yml diff --git a/.github/workflows/flask-investigation.yml b/.github/workflows/flask-investigation.yml new file mode 100644 index 0000000..6dead63 --- /dev/null +++ b/.github/workflows/flask-investigation.yml @@ -0,0 +1,31 @@ +name: Flask investigation +on: + push: + branches: [investigate/flask-json-70] +permissions: + contents: read +concurrency: + group: flask-investigation-70 + cancel-in-progress: false +jobs: + snapshot: + if: github.repository == 'tappe9/simple-api-benchmark' + runs-on: ubuntu-24.04 + timeout-minutes: 5 + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 + with: + fetch-depth: 0 + persist-credentials: false + - name: Export exact source for offline verification + run: | + mkdir -p .cache/flask-investigation-source + git bundle create .cache/flask-investigation-source/source.bundle HEAD + git rev-parse HEAD HEAD^{tree} > .cache/flask-investigation-source/identity.txt + - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + with: + name: flask-investigation-source-${{ github.run_id }}-${{ github.run_attempt }} + path: .cache/flask-investigation-source/ + include-hidden-files: true + retention-days: 7 + if-no-files-found: error From cbebd385b0f78503d797fcc73167f1ecce1b8611 Mon Sep 17 00:00:00 2001 From: Yuma Takino <36617106+tappe9@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:33:52 +0900 Subject: [PATCH 2/4] test: add bounded non-publishing Flask hypothesis diagnostic Exercise unchanged Flask against thread-switch and queue-logging probes in two reversed blocks. Preserve every sample, enforce contract and cleanup, and retain raw evidence outside public results. Local diagnostic tests were observed red before implementation and pass after implementation; real Docker evidence is collected separately by a temporary read-only workflow. Refs #70. --- .github/workflows/flask-investigation.yml | 33 ++- benchmark/flask_diagnostic.py | 272 ++++++++++++++++++++++ tests/test_benchmark_flask_diagnostic.py | 187 +++++++++++++++ 3 files changed, 483 insertions(+), 9 deletions(-) create mode 100644 benchmark/flask_diagnostic.py create mode 100644 tests/test_benchmark_flask_diagnostic.py diff --git a/.github/workflows/flask-investigation.yml b/.github/workflows/flask-investigation.yml index 6dead63..2ac1f2e 100644 --- a/.github/workflows/flask-investigation.yml +++ b/.github/workflows/flask-investigation.yml @@ -8,24 +8,39 @@ concurrency: group: flask-investigation-70 cancel-in-progress: false jobs: - snapshot: + diagnose: if: github.repository == 'tappe9/simple-api-benchmark' runs-on: ubuntu-24.04 - timeout-minutes: 5 + timeout-minutes: 20 + env: + PYTHONDONTWRITEBYTECODE: '1' steps: - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 with: fetch-depth: 0 persist-credentials: false - - name: Export exact source for offline verification + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + with: + python-version-file: apps/python-fastapi/.python-version + - name: Diagnostic regression tests + run: python -m unittest discover -s tests -p test_benchmark_flask_diagnostic.py -v + - name: Fixed non-publishing hypothesis matrix + run: python -m benchmark.flask_diagnostic + - name: Confirm production inputs and published outputs stay unchanged + run: | + git diff --exit-code HEAD + test -z "$(git ls-files --others --exclude-standard)" + - name: Export exact source for offline review + if: always() run: | - mkdir -p .cache/flask-investigation-source - git bundle create .cache/flask-investigation-source/source.bundle HEAD - git rev-parse HEAD HEAD^{tree} > .cache/flask-investigation-source/identity.txt + mkdir -p .cache/flask-diagnostic/source + git bundle create .cache/flask-diagnostic/source/source.bundle HEAD + git rev-parse HEAD HEAD^{tree} > .cache/flask-diagnostic/source/identity.txt - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + if: always() with: - name: flask-investigation-source-${{ github.run_id }}-${{ github.run_attempt }} - path: .cache/flask-investigation-source/ + name: flask-diagnostic-${{ github.run_id }}-${{ github.run_attempt }} + path: .cache/flask-diagnostic/ include-hidden-files: true - retention-days: 7 + retention-days: 30 if-no-files-found: error diff --git a/benchmark/flask_diagnostic.py b/benchmark/flask_diagnostic.py new file mode 100644 index 0000000..781fac5 --- /dev/null +++ b/benchmark/flask_diagnostic.py @@ -0,0 +1,272 @@ +"""Counterbalanced Flask hypothesis probes; never an official performance result.""" + +import copy +import hashlib +import json +import re +import signal +import sys +import uuid +from pathlib import Path + +from .contract_runner import protect_cleanup +from .contract_test import ContractFailure, load_cases, run_contract +from .environment import DockerEnvironment +from .healthcheck import EXTERNAL_READINESS +from .process import ROOT, execute +from .registry import implementation +from .results import BenchmarkFailure, atomic_json, require, select_run, strict_json, validate_run +from .run import now + +CACHE_ROOT = ROOT / ".cache" / "flask-diagnostic" +IMPLEMENTATION = "python-flask" +ARMS = ("baseline", "switch-1ms", "queue-quiet") + + +def diagnostic_plan() -> list[dict]: + """Fix order before collecting data; reverse arms and endpoints in block two.""" + return [ + { + "id": f"block-{block}-{arm}", + "block": block, + "arm": arm, + "endpoints": ["/json", "/db/42"] if block == 1 else ["/db/42", "/json"], + } + for block, arms in ((1, ARMS), (2, tuple(reversed(ARMS)))) + for arm in arms + ] + + +def validate_output_path(path: Path) -> Path: + candidate = path.absolute() + root = CACHE_ROOT.absolute() + require(candidate.is_relative_to(root), "output must stay below Flask diagnostic cache") + require(candidate.resolve().is_relative_to(root.resolve()), "output escapes diagnostic cache") + require(candidate.name == "diagnostic.json", "expected diagnostic.json output") + for component in (candidate, *candidate.parents): + require(not component.is_symlink(), "symlink in diagnostic output path") + return candidate + + +def hook_source(arm: str) -> str: + """Only fixed, reviewed interventions are allowed; baseline has no startup hook.""" + require(arm in ARMS, "unknown diagnostic arm") + if arm == "baseline": + return "" + intervention = ( + "sys.setswitchinterval(0.001)" + if arm == "switch-1ms" + else 'logging.getLogger("waitress.queue").disabled = True' + ) + return ( + "import json, logging, os, sys\n" + + intervention + + '\nif os.getpid() == 1:\n' + + ' print("FLASK_DIAGNOSTIC " + json.dumps({"arm": ' + + repr(arm) + + ', "pid": os.getpid(), "interval": sys.getswitchinterval(), ' + + '"queue_disabled": logging.getLogger("waitress.queue").disabled}), flush=True)\n' + ) + + +def compose_override(arm: str, hook: Path) -> dict: + hook_source(arm) # Validate before producing any container configuration. + if arm == "baseline": + return {} + return { + "services": { + IMPLEMENTATION: { + "environment": {"PYTHONPATH": "/flask-diagnostic"}, + "volumes": [ + { + "type": "bind", + "source": str(hook.resolve()), + "target": "/flask-diagnostic", + "read_only": True, + } + ], + } + } + } + + +class FlaskEnvironment(DockerEnvironment): + """Keep production image/entrypoint intact and instrument only between windows.""" + + def __init__(self, oha: Path, directory: Path, arm: str): + super().__init__(oha, directory, health_policy=EXTERNAL_READINESS) + self.arm = arm + self.since = now() + self.hook_evidence = None + if arm != "baseline": + hook = self.artifacts / "hook" + hook.mkdir() + content = hook_source(arm) + (hook / "sitecustomize.py").write_text(content, encoding="utf-8") + override = self.artifacts / "diagnostic.compose.json" + atomic_json(override, compose_override(arm, hook)) + self.prefix = self.prefix[:-2] + ["-f", str(override), *self.prefix[-2:]] + self.hook_evidence = {"sha256": hashlib.sha256(content.encode()).hexdigest()} + + def start(self, identifier: str) -> dict: + result = super().start(identifier) + if self.arm != "baseline": + logs = execute(["docker", "logs", "--tail", "100", self.container], timeout=10) + markers = [line.removeprefix("FLASK_DIAGNOSTIC ") for line in logs.splitlines() + if line.startswith("FLASK_DIAGNOSTIC ")] + require(len(markers) == 1, "diagnostic startup hook was not observed exactly once") + observed = strict_json(markers[0].encode()) + require(observed.get("arm") == self.arm and observed.get("pid") == 1, + "hook belongs to wrong process/arm") + require(observed.get("queue_disabled") is (self.arm == "queue-quiet"), + "queue logger intervention mismatch") + require(observed.get("interval") == (0.001 if self.arm == "switch-1ms" else 0.005), + "thread interval intervention mismatch") + self.hook_evidence["observed"] = observed + result["diagnostic_hook"] = self.hook_evidence + return result + + def observe(self, label: str) -> dict: + """Snapshots include collection overhead; no exec/polling runs during oha.""" + self.check() + require(re.fullmatch(r"[a-z0-9-]+", label) is not None, "invalid observation label") + script = ( + "import json; from pathlib import Path; " + "paths=['/sys/fs/cgroup/cpu.stat','/proc/1/stat','/proc/1/status']; " + "print(json.dumps({p:Path(p).read_text()[:16384] if Path(p).is_file() " + "else None for p in paths}))" + ) + raw = execute(["docker", "exec", self.container, "python", "-c", script], timeout=10) + counters = strict_json(raw.encode()) + logs = execute(["docker", "logs", "--timestamps", "--since", self.since, + "--tail", "10000", self.container], timeout=10) + encoded = logs.encode() + limited = encoded[-2 * 1024 * 1024:] + log_path = self.artifacts / f"{label}.server.log" + log_path.write_bytes(limited) + snapshot = { + "at": now(), "container_id": self.container, "counters": counters, + "server_log": log_path.name, "log_tail_limit_lines": 10000, + "log_tail_may_be_truncated": len(logs.splitlines()) >= 10000 or len(limited) != len(encoded), + "queue_warning_lines_in_tail": limited.count(b"Task queue depth is"), + } + atomic_json(self.artifacts / f"{label}.observation.json", snapshot) + self.check() + return snapshot + + +def run_cell(environment, cell: dict, *, contract=run_contract) -> dict: + failure = None + try: + require(environment.health_policy == EXTERNAL_READINESS, "diagnostic health policy differs") + require(environment.connections == 50, "diagnostic connections must be 50") + require(environment.request_timeout == 15, "diagnostic request timeout must be 15s") + environment.build(IMPLEMENTATION) + container = environment.start(IMPLEMENTATION) + checks = contract("http://127.0.0.1:8080", implementation=IMPLEMENTATION) + require(type(checks) is int and checks == 2 * len(load_cases()), "incomplete shared contract") + environment.check() + endpoints = [] + for endpoint in cell["endpoints"]: + label = endpoint.strip("/").replace("/", "-") + before = environment.observe(label + "-before") + warmup = environment.measure(endpoint, 5, 0) + validate_run(warmup) + runs = [] + for index in (1, 2, 3): + observed = environment.measure(endpoint, 10, index) + validate_run(observed) + runs.append(observed) + after = environment.observe(label + "-after") + endpoints.append({"endpoint": endpoint, "warmup": warmup, "runs": runs, + "selected": select_run(runs), "before": before, "after": after}) + environment.check() + readiness = copy.deepcopy(environment.readiness) + except BaseException as error: + failure = error + raise + finally: + try: + with protect_cleanup(): + environment.cleanup() + except (BenchmarkFailure, OSError) as error: + if failure is not None: + raise BenchmarkFailure(f"{failure}; cleanup also failed: {error}") from failure + raise + return {**cell, "container": container, "readiness": readiness, "contract_checks": checks, + "artifact_directory": str(environment.artifacts), "endpoints": endpoints} + + +def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contract) -> dict: + output = validate_output_path(output) + for key in ("source_commit", "source_tree"): + require(type(metadata.get(key)) is str and re.fullmatch(r"[0-9a-f]{40}", metadata[key]), + "diagnostic source metadata missing " + key) + versions = metadata.get("versions", {}) + require(set(versions) == {IMPLEMENTATION} and + set(implementation(IMPLEMENTATION)["version_fields"]) <= set(versions[IMPLEMENTATION]), + "diagnostic requires Flask version metadata") + plan = diagnostic_plan() + started_at = now() + cells = [] + progress = output.parent / "progress.json" + for cell in plan: + atomic_json(progress, {"status": "running", "official": False, "publishable": False, + "metadata": metadata, "plan": plan, "active_cell": cell["id"], "cells": cells}) + print(f"Flask diagnostic {cell['id']}", flush=True) + try: + environment = factory(cell) + cells.append(run_cell(environment, cell, contract=contract)) + except BaseException as error: + atomic_json(progress, {"status": "failed", "official": False, "publishable": False, + "metadata": metadata, "plan": plan, "failed_cell": cell["id"], + "error_type": type(error).__name__, "cells": cells}) + raise + result = { + "schema_version": 1, "mode": "flask-diagnostic", "status": "verified", + "official": False, "publishable": False, "implementation": IMPLEMENTATION, + "started_at": started_at, "completed_at": now(), "metadata": copy.deepcopy(metadata), + "conditions": {"connections": 50, "warmup_seconds": 5, "duration_seconds": 10, + "runs": 3, "api_health_policy": EXTERNAL_READINESS}, + "limitations": ["Non-publishing diagnostic, not an official result or framework ranking.", + "CPU snapshots span warm-up and measurement plus observation overhead.", + "Server logs are bounded tails; warning counts may be lower bounds.", + "Two blocks do not establish cross-host or universal performance effects."], + "plan": plan, "cells": cells, + } + atomic_json(output, result) + atomic_json(progress, {"status": "completed", "official": False, "publishable": False, + "metadata": metadata, "plan": plan, "cells": cells}) + return result + + +def main() -> int: + from .environment import provenance, registered_pinned_versions + from .install_oha import ensure_oha + + def interrupted(signum, _frame): + raise KeyboardInterrupt(f"received signal {signum}") + + previous = {sig: signal.signal(sig, interrupted) for sig in (signal.SIGINT, signal.SIGTERM)} + try: + directory = CACHE_ROOT / uuid.uuid4().hex + output = validate_output_path(directory / "diagnostic.json") + oha = ensure_oha() + metadata = provenance(oha) + metadata["versions"] = {IMPLEMENTATION: registered_pinned_versions()[IMPLEMENTATION]} + metadata["diagnostic_plan_sha256"] = hashlib.sha256( + json.dumps(diagnostic_plan(), sort_keys=True).encode()).hexdigest() + run_diagnostic(lambda cell: FlaskEnvironment(oha, directory / cell["id"], cell["arm"]), + output, metadata=metadata) + print(f"Flask diagnostic complete: {output}; no results published.") + return 0 + except (BenchmarkFailure, ContractFailure, OSError, ValueError, KeyboardInterrupt) as error: + print(f"Flask diagnostic failed: {error}", file=sys.stderr) + return 1 + finally: + for sig, handler in previous.items(): + signal.signal(sig, handler) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_benchmark_flask_diagnostic.py b/tests/test_benchmark_flask_diagnostic.py new file mode 100644 index 0000000..36fdc8c --- /dev/null +++ b/tests/test_benchmark_flask_diagnostic.py @@ -0,0 +1,187 @@ +"""Flask investigation must keep all arms non-publishing and clean every cell.""" + +import importlib +import importlib.util +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from test_benchmark_runner import Environment + +from benchmark.environment import registered_pinned_versions +from benchmark.healthcheck import EXTERNAL_READINESS +from benchmark.report import validate_report +from benchmark.results import BenchmarkFailure + + +class DiagnosticEnvironment(Environment): + def __init__(self, root, cell, fail=None): + super().__init__(fail) + self.artifacts = root / cell["id"] + self.artifacts.mkdir() + self.health_policy = EXTERNAL_READINESS + self.connections = 50 + self.request_timeout = 15 + self.readiness = {"attempts": 1} + + def observe(self, label): + self.event("observe") + return {"label": label, "cpu_stat": "usage_usec 1"} + + +class FlaskDiagnosticTests(unittest.TestCase): + def setUp(self): + self.assertIsNotNone( + importlib.util.find_spec("benchmark.flask_diagnostic"), + "Flask needs the bounded non-publishing investigation runner", + ) + self.module = importlib.import_module("benchmark.flask_diagnostic") + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.cache = self.root / ".cache" / "flask-diagnostic" + self.cache.mkdir(parents=True) + patcher = patch.object(self.module, "CACHE_ROOT", self.cache) + patcher.start() + self.addCleanup(patcher.stop) + self.output = self.cache / "diagnostic.json" + self.output.write_bytes(b"previous diagnostic") + self.published = self.root / "results" / "latest.json" + self.published.parent.mkdir() + self.published.write_bytes(b"previous official result") + self.metadata = { + "source_commit": "a" * 40, + "source_tree": "b" * 40, + "versions": {"python-flask": registered_pinned_versions()["python-flask"]}, + } + self.environments = [] + + def run_diagnostic(self, fail=None, alter=None): + def factory(cell): + environment = DiagnosticEnvironment(self.cache, cell, fail) + if alter is not None: + alter(environment) + self.environments.append(environment) + return environment + + def contract(url, *, implementation): + return self.environments[-1].contract(url, implementation=implementation) + + return self.module.run_diagnostic(factory, self.output, metadata=self.metadata, contract=contract) + + def test_fixed_counterbalanced_plan_and_every_sample_are_preserved(self): + result = self.run_diagnostic() + self.assertEqual([cell["arm"] for cell in result["cells"]], + ["baseline", "switch-1ms", "queue-quiet", "queue-quiet", "switch-1ms", "baseline"]) + self.assertEqual(len(self.environments), 6) + for position, (cell, environment) in enumerate(zip(result["cells"], self.environments)): + endpoints = ["/json", "/db/42"] if position < 3 else ["/db/42", "/json"] + self.assertEqual([item["endpoint"] for item in cell["endpoints"]], endpoints) + measured = [event for event in environment.events if isinstance(event, tuple)] + self.assertEqual(measured, [("python-flask", endpoint, 5 if index == 0 else 10, index) + for endpoint in endpoints for index in (0, 1, 2, 3)]) + self.assertEqual(environment.events.count("cleanup"), 1) + self.assertIsNone(environment.active) + self.assertEqual(cell["readiness"], {"attempts": 1}) + self.assertFalse(result["official"]) + self.assertFalse(result["publishable"]) + self.assertEqual(result["mode"], "flask-diagnostic") + self.assertEqual(json.loads(self.output.read_bytes()), result) + self.assertEqual(self.published.read_bytes(), b"previous official result") + with self.assertRaises(BenchmarkFailure): + validate_report(result) + + def test_all_failure_stages_clean_and_never_write_partial_success(self): + for stage in ("build", "startup", "contract", "state", "warmup", "measurement", "metric", "observe", "cleanup"): + for folder in self.cache.iterdir(): + if folder.is_dir(): + import shutil + shutil.rmtree(folder) + with self.subTest(stage=stage), self.assertRaises(BenchmarkFailure): + self.run_diagnostic(stage) + environment = self.environments[-1] + self.assertEqual(environment.events.count("cleanup"), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + self.assertEqual(self.published.read_bytes(), b"previous official result") + + def test_cleanup_precedes_successful_output_and_retains_readiness(self): + def alter(environment): + original = environment.cleanup + def cleanup(): + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + original() + environment.readiness = None + environment.cleanup = cleanup + self.run_diagnostic(alter=alter) + + def test_interruption_cleans_and_does_not_retry(self): + def alter(environment): + def interrupted(*args): + raise KeyboardInterrupt + environment.measure = interrupted + with self.assertRaises(KeyboardInterrupt): + self.run_diagnostic(alter=alter) + self.assertEqual(len(self.environments), 1) + self.assertEqual(self.environments[0].events.count("cleanup"), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + + def test_incomplete_contract_cannot_start_load(self): + def alter(environment): + environment.contract = lambda *args, **kwargs: True + with self.assertRaisesRegex(BenchmarkFailure, "contract"): + self.run_diagnostic(alter=alter) + self.assertEqual(self.environments[0].measures, 0) + + def test_invalid_profile_or_metadata_prevents_load(self): + for field, value in (("connections", 2), ("request_timeout", 1), ("health_policy", "container-healthcheck")): + with self.subTest(field=field), self.assertRaises(BenchmarkFailure): + self.run_diagnostic(alter=lambda environment: setattr(environment, field, value)) + self.assertNotIn("build", self.environments[-1].events) + self.environments[-1].artifacts.rmdir() + self.metadata["source_commit"] = "unknown" + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic() + + def test_output_and_symlinks_cannot_escape_cache(self): + for path in (self.published, self.cache / ".." / "diagnostic.json"): + with self.assertRaises(BenchmarkFailure): + self.module.validate_output_path(path) + linked = self.cache / "linked" + linked.symlink_to(self.published.parent, target_is_directory=True) + with self.assertRaises(BenchmarkFailure): + self.module.validate_output_path(linked / "diagnostic.json") + self.output.unlink() + self.output.symlink_to(self.published) + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic() + self.assertEqual(self.published.read_bytes(), b"previous official result") + + def test_startup_hooks_change_only_the_declared_factor(self): + for arm in ("baseline", "switch-1ms", "queue-quiet"): + code = self.module.hook_source(arm) + probe = ('import sys, logging, json\n' + code + + '\nprint(json.dumps({"interval":sys.getswitchinterval(),"queue":logging.getLogger("waitress.queue").disabled,"root":logging.getLogger().disabled}))') + output = subprocess.check_output([sys.executable, "-c", probe], text=True, timeout=5) + observed = json.loads(output.splitlines()[-1]) + self.assertEqual(observed["interval"], 0.001 if arm == "switch-1ms" else sys.getswitchinterval()) + self.assertEqual(observed["queue"], arm == "queue-quiet") + self.assertFalse(observed["root"]) + with self.assertRaises(BenchmarkFailure): + self.module.hook_source("arbitrary-code") + + def test_baseline_override_does_not_change_command_or_environment(self): + self.assertEqual(self.module.compose_override("baseline", self.cache), {}) + for arm in ("switch-1ms", "queue-quiet"): + config = self.module.compose_override(arm, self.cache) + service = config["services"]["python-flask"] + self.assertEqual(set(service), {"environment", "volumes"}) + self.assertTrue(service["volumes"][0]["read_only"]) + self.assertEqual(service["environment"], {"PYTHONPATH": "/flask-diagnostic"}) + + +if __name__ == "__main__": + unittest.main() From a153a103f445af8cd571687fb095cc0d2f4cb9a3 Mon Sep 17 00:00:00 2001 From: Yuma Takino <36617106+tappe9@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:51:08 +0900 Subject: [PATCH 3/4] test: probe Flask scheduling with bracketing affinity controls Keep the first 36-window diagnostic immutable. Add a separately predeclared baseline/one-CPU-affinity/baseline phase with per-thread CPU observations outside load, preserving quota and production inputs. Add CLI argument safety and regression coverage; 14 focused local tests pass. Refs #70. --- .github/workflows/flask-investigation.yml | 9 ++- benchmark/flask_diagnostic.py | 73 +++++++++++++++------ tests/test_benchmark_flask_diagnostic.py | 78 ++++++++++++++++++++++- 3 files changed, 135 insertions(+), 25 deletions(-) diff --git a/.github/workflows/flask-investigation.yml b/.github/workflows/flask-investigation.yml index 2ac1f2e..f6762b4 100644 --- a/.github/workflows/flask-investigation.yml +++ b/.github/workflows/flask-investigation.yml @@ -25,7 +25,7 @@ jobs: - name: Diagnostic regression tests run: python -m unittest discover -s tests -p test_benchmark_flask_diagnostic.py -v - name: Fixed non-publishing hypothesis matrix - run: python -m benchmark.flask_diagnostic + run: python -m benchmark.flask_diagnostic --phase scheduling - name: Confirm production inputs and published outputs stay unchanged run: | git diff --exit-code HEAD @@ -36,6 +36,13 @@ jobs: mkdir -p .cache/flask-diagnostic/source git bundle create .cache/flask-diagnostic/source/source.bundle HEAD git rev-parse HEAD HEAD^{tree} > .cache/flask-diagnostic/source/identity.txt + - name: Export pinned formatter for offline review + if: always() + run: | + python -m pip install --require-hashes --only-binary=:all: -r apps/python-fastapi/requirements-dev.lock + mkdir -p .cache/flask-diagnostic/review + cp "$(command -v ruff)" .cache/flask-diagnostic/review/ruff + sha256sum .cache/flask-diagnostic/review/ruff > .cache/flask-diagnostic/review/ruff.sha256 - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 if: always() with: diff --git a/benchmark/flask_diagnostic.py b/benchmark/flask_diagnostic.py index 781fac5..092a938 100644 --- a/benchmark/flask_diagnostic.py +++ b/benchmark/flask_diagnostic.py @@ -1,5 +1,6 @@ """Counterbalanced Flask hypothesis probes; never an official performance result.""" +import argparse import copy import hashlib import json @@ -20,11 +21,17 @@ CACHE_ROOT = ROOT / ".cache" / "flask-diagnostic" IMPLEMENTATION = "python-flask" -ARMS = ("baseline", "switch-1ms", "queue-quiet") +INITIAL_ARMS = ("baseline", "switch-1ms", "queue-quiet") +ARMS = (*INITIAL_ARMS, "affinity-one") -def diagnostic_plan() -> list[dict]: +def diagnostic_plan(phase: str = "initial") -> list[dict]: """Fix order before collecting data; reverse arms and endpoints in block two.""" + require(phase in ("initial", "scheduling"), "unknown diagnostic phase") + if phase == "scheduling": + return [{"id": f"block-{index}-{arm}", "block": index, "arm": arm, + "endpoints": ["/json", "/db/42"]} + for index, arm in enumerate(("baseline", "affinity-one", "baseline"), 1)] return [ { "id": f"block-{block}-{arm}", @@ -32,7 +39,7 @@ def diagnostic_plan() -> list[dict]: "arm": arm, "endpoints": ["/json", "/db/42"] if block == 1 else ["/db/42", "/json"], } - for block, arms in ((1, ARMS), (2, tuple(reversed(ARMS)))) + for block, arms in ((1, INITIAL_ARMS), (2, tuple(reversed(INITIAL_ARMS)))) for arm in arms ] @@ -53,19 +60,22 @@ def hook_source(arm: str) -> str: require(arm in ARMS, "unknown diagnostic arm") if arm == "baseline": return "" - intervention = ( - "sys.setswitchinterval(0.001)" - if arm == "switch-1ms" - else 'logging.getLogger("waitress.queue").disabled = True' - ) + intervention = { + "switch-1ms": "sys.setswitchinterval(0.001)", + "queue-quiet": 'logging.getLogger("waitress.queue").disabled = True', + "affinity-one": "os.sched_setaffinity(0, {allowed_before[0]})", + }[arm] return ( "import json, logging, os, sys\n" + "allowed_before = sorted(os.sched_getaffinity(0))\n" + intervention + '\nif os.getpid() == 1:\n' + ' print("FLASK_DIAGNOSTIC " + json.dumps({"arm": ' + repr(arm) + ', "pid": os.getpid(), "interval": sys.getswitchinterval(), ' - + '"queue_disabled": logging.getLogger("waitress.queue").disabled}), flush=True)\n' + + '"queue_disabled": logging.getLogger("waitress.queue").disabled, ' + + '"allowed_before": allowed_before, "allowed_after": sorted(os.sched_getaffinity(0)), ' + + '"gil_enabled": getattr(sys, "_is_gil_enabled", lambda: None)()}), flush=True)\n' ) @@ -122,6 +132,10 @@ def start(self, identifier: str) -> dict: "queue logger intervention mismatch") require(observed.get("interval") == (0.001 if self.arm == "switch-1ms" else 0.005), "thread interval intervention mismatch") + before = observed.get("allowed_before") + require(type(before) is list and bool(before), "missing allowed CPU set") + expected = before[:1] if self.arm == "affinity-one" else before + require(observed.get("allowed_after") == expected, "CPU affinity intervention mismatch") self.hook_evidence["observed"] = observed result["diagnostic_hook"] = self.hook_evidence return result @@ -130,12 +144,25 @@ def observe(self, label: str) -> dict: """Snapshots include collection overhead; no exec/polling runs during oha.""" self.check() require(re.fullmatch(r"[a-z0-9-]+", label) is not None, "invalid observation label") - script = ( - "import json; from pathlib import Path; " - "paths=['/sys/fs/cgroup/cpu.stat','/proc/1/stat','/proc/1/status']; " - "print(json.dumps({p:Path(p).read_text()[:16384] if Path(p).is_file() " - "else None for p in paths}))" - ) + script = """import json, os, sys +from pathlib import Path +paths = ['/sys/fs/cgroup/cpu.stat', '/proc/1/stat', '/proc/1/status'] +result = {p: Path(p).read_text()[:16384] if Path(p).is_file() else None for p in paths} +threads = {} +vanished = [] +for path in sorted(Path('/proc/1/task').glob('[0-9]*'))[:64]: + try: + threads[path.name] = {name: (path / name).read_text()[:16384] + for name in ('comm', 'stat', 'schedstat', 'status')} + except FileNotFoundError: + vanished.append(path.name) +result['threads'] = threads +result['vanished_threads'] = vanished +result['helper_runtime_not_pid1'] = {'version': sys.version, + 'gil_enabled': getattr(sys, '_is_gil_enabled', lambda: None)(), + 'affinity': sorted(os.sched_getaffinity(0))} +print(json.dumps(result)) +""" raw = execute(["docker", "exec", self.container, "python", "-c", script], timeout=10) counters = strict_json(raw.encode()) logs = execute(["docker", "logs", "--timestamps", "--since", self.since, @@ -197,7 +224,7 @@ def run_cell(environment, cell: dict, *, contract=run_contract) -> dict: "artifact_directory": str(environment.artifacts), "endpoints": endpoints} -def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contract) -> dict: +def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contract, phase="initial") -> dict: output = validate_output_path(output) for key in ("source_commit", "source_tree"): require(type(metadata.get(key)) is str and re.fullmatch(r"[0-9a-f]{40}", metadata[key]), @@ -206,7 +233,7 @@ def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contra require(set(versions) == {IMPLEMENTATION} and set(implementation(IMPLEMENTATION)["version_fields"]) <= set(versions[IMPLEMENTATION]), "diagnostic requires Flask version metadata") - plan = diagnostic_plan() + plan = diagnostic_plan(phase) started_at = now() cells = [] progress = output.parent / "progress.json" @@ -225,7 +252,7 @@ def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contra result = { "schema_version": 1, "mode": "flask-diagnostic", "status": "verified", "official": False, "publishable": False, "implementation": IMPLEMENTATION, - "started_at": started_at, "completed_at": now(), "metadata": copy.deepcopy(metadata), + "started_at": started_at, "completed_at": now(), "metadata": copy.deepcopy(metadata), "phase": phase, "conditions": {"connections": 50, "warmup_seconds": 5, "duration_seconds": 10, "runs": 3, "api_health_policy": EXTERNAL_READINESS}, "limitations": ["Non-publishing diagnostic, not an official result or framework ranking.", @@ -240,10 +267,14 @@ def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contra return result -def main() -> int: +def main(argv: list[str] | None = None) -> int: from .environment import provenance, registered_pinned_versions from .install_oha import ensure_oha + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--phase", choices=("initial", "scheduling"), default="initial") + args = parser.parse_args(argv) + def interrupted(signum, _frame): raise KeyboardInterrupt(f"received signal {signum}") @@ -255,9 +286,9 @@ def interrupted(signum, _frame): metadata = provenance(oha) metadata["versions"] = {IMPLEMENTATION: registered_pinned_versions()[IMPLEMENTATION]} metadata["diagnostic_plan_sha256"] = hashlib.sha256( - json.dumps(diagnostic_plan(), sort_keys=True).encode()).hexdigest() + json.dumps(diagnostic_plan(args.phase), sort_keys=True).encode()).hexdigest() run_diagnostic(lambda cell: FlaskEnvironment(oha, directory / cell["id"], cell["arm"]), - output, metadata=metadata) + output, metadata=metadata, phase=args.phase) print(f"Flask diagnostic complete: {output}; no results published.") return 0 except (BenchmarkFailure, ContractFailure, OSError, ValueError, KeyboardInterrupt) as error: diff --git a/tests/test_benchmark_flask_diagnostic.py b/tests/test_benchmark_flask_diagnostic.py index 36fdc8c..6039993 100644 --- a/tests/test_benchmark_flask_diagnostic.py +++ b/tests/test_benchmark_flask_diagnostic.py @@ -2,7 +2,10 @@ import importlib import importlib.util +import inspect import json +import os +import shutil import subprocess import sys import tempfile @@ -60,7 +63,7 @@ def setUp(self): } self.environments = [] - def run_diagnostic(self, fail=None, alter=None): + def run_diagnostic(self, fail=None, alter=None, phase="initial"): def factory(cell): environment = DiagnosticEnvironment(self.cache, cell, fail) if alter is not None: @@ -71,7 +74,7 @@ def factory(cell): def contract(url, *, implementation): return self.environments[-1].contract(url, implementation=implementation) - return self.module.run_diagnostic(factory, self.output, metadata=self.metadata, contract=contract) + return self.module.run_diagnostic(factory, self.output, metadata=self.metadata, contract=contract, phase=phase) def test_fixed_counterbalanced_plan_and_every_sample_are_preserved(self): result = self.run_diagnostic() @@ -99,7 +102,6 @@ def test_all_failure_stages_clean_and_never_write_partial_success(self): for stage in ("build", "startup", "contract", "state", "warmup", "measurement", "metric", "observe", "cleanup"): for folder in self.cache.iterdir(): if folder.is_dir(): - import shutil shutil.rmtree(folder) with self.subTest(stage=stage), self.assertRaises(BenchmarkFailure): self.run_diagnostic(stage) @@ -146,6 +148,42 @@ def test_invalid_profile_or_metadata_prevents_load(self): with self.assertRaises(BenchmarkFailure): self.run_diagnostic() + def test_later_cell_failure_preserves_completed_evidence_without_final_result(self): + count = 0 + + def alter(environment): + nonlocal count + count += 1 + if count == 2: + environment.fail = "measurement" + + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic(alter=alter) + self.assertEqual(len(self.environments), 2) + self.assertTrue(all(e.events.count("cleanup") == 1 for e in self.environments)) + progress = json.loads((self.cache / "progress.json").read_bytes()) + self.assertEqual(progress["status"], "failed") + self.assertEqual(progress["failed_cell"], "block-1-switch-1ms") + self.assertEqual(len(progress["cells"]), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + + def test_invalid_measurement_is_not_retried_or_selected(self): + def alter(environment): + original = environment.measure + + def measure(*args): + value = original(*args) + value["requests_per_second"] = float("nan") + return value + + environment.measure = measure + + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic(alter=alter) + self.assertEqual(len(self.environments), 1) + self.assertEqual(self.environments[0].events.count("cleanup"), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + def test_output_and_symlinks_cannot_escape_cache(self): for path in (self.published, self.cache / ".." / "diagnostic.json"): with self.assertRaises(BenchmarkFailure): @@ -173,6 +211,40 @@ def test_startup_hooks_change_only_the_declared_factor(self): with self.assertRaises(BenchmarkFailure): self.module.hook_source("arbitrary-code") + def test_scheduling_phase_has_bracketing_baselines_and_unchanged_load(self): + self.assertIn("phase", inspect.signature(self.module.diagnostic_plan).parameters) + plan = self.module.diagnostic_plan("scheduling") + self.assertEqual([cell["arm"] for cell in plan], ["baseline", "affinity-one", "baseline"]) + self.assertTrue(all(cell["endpoints"] == ["/json", "/db/42"] for cell in plan)) + with self.assertRaises(BenchmarkFailure): + self.module.diagnostic_plan("arbitrary") + result = self.run_diagnostic(phase="scheduling") + self.assertEqual(len(result["cells"]), 3) + self.assertEqual(sum(e.measures for e in self.environments), 18) + self.assertEqual(result["phase"], "scheduling") + + def test_affinity_hook_changes_only_the_child_allowed_cpu_set(self): + self.assertIn("affinity-one", self.module.ARMS) + before = sorted(os.sched_getaffinity(0)) + probe = self.module.hook_source("affinity-one") + ( + '\nprint(json.dumps({"affinity":sorted(os.sched_getaffinity(0)), ' + '"interval":sys.getswitchinterval(),"queue":logging.getLogger("waitress.queue").disabled}))' + ) + output = subprocess.check_output([sys.executable, "-c", probe], text=True, timeout=5) + observed = json.loads(output.splitlines()[-1]) + self.assertEqual(observed["affinity"], [before[0]]) + self.assertEqual(observed["interval"], 0.005) + self.assertFalse(observed["queue"]) + self.assertEqual(sorted(os.sched_getaffinity(0)), before) + + def test_help_and_invalid_arguments_never_start_measurement(self): + for args, code in ((["--help"], 0), (["--unknown"], 2)): + with self.subTest(args=args), patch.object(sys, "argv", ["flask-diagnostic", *args]): + with patch("benchmark.install_oha.ensure_oha", side_effect=AssertionError("load must not start")): + with self.assertRaises(SystemExit) as stopped: + self.module.main() + self.assertEqual(stopped.exception.code, code) + def test_baseline_override_does_not_change_command_or_environment(self): self.assertEqual(self.module.compose_override("baseline", self.cache), {}) for arm in ("switch-1ms", "queue-quiet"): From 1ee3a87cda97a280e54745e425579ca5ee3222b8 Mon Sep 17 00:00:00 2001 From: Yuma Takino <36617106+tappe9@users.noreply.github.com> Date: Fri, 18 Sep 2026 14:31:06 +0900 Subject: [PATCH 4/4] test: finalize Flask scheduling investigation and preserve evidence Retain the two predeclared non-publishing diagnostic phases and their exact-source evidence. Document the JSON/DB inversion, negative switch-interval and queue-logging probes, and the CPU-placement-sensitive bracketing comparison without claiming a universal speedup or an isolated GIL/kernel root cause. Add a fail-closed image-drift guard with verified red/green regression coverage. Format the diagnostic and tests, link the durable public guide, and remove the temporary branch-only investigation workflow. Verified final tree e0298668618fa31124a823502cc1886d8c8264be: 15 focused tests, 21 workflow tests, scoped Ruff, registry and README checks. Independently audited the original 72 measured windows and both diagnostic archives (54 measured windows plus 18 warm-ups). Full local benchmark/site suites did not complete within tool deadlines and local Docker acceptance is unavailable; normal PR CI remains required. Review is self-review, not independent approval. Preserve applications, dependency pins, Compose, active and frozen cohorts, official conditions, publication workflows and published results. No new official run, publication, Pages rollout, production tuning or protection change. Refs #70. --- .github/workflows/flask-investigation.yml | 53 ------ benchmark/flask_diagnostic.py | 211 ++++++++++++++++------ docs/FLASK-DIAGNOSTIC.md | 189 +++++++++++++++++++ docs/FLASK.md | 7 + tests/test_benchmark_flask_diagnostic.py | 88 +++++++-- 5 files changed, 433 insertions(+), 115 deletions(-) delete mode 100644 .github/workflows/flask-investigation.yml create mode 100644 docs/FLASK-DIAGNOSTIC.md diff --git a/.github/workflows/flask-investigation.yml b/.github/workflows/flask-investigation.yml deleted file mode 100644 index f6762b4..0000000 --- a/.github/workflows/flask-investigation.yml +++ /dev/null @@ -1,53 +0,0 @@ -name: Flask investigation -on: - push: - branches: [investigate/flask-json-70] -permissions: - contents: read -concurrency: - group: flask-investigation-70 - cancel-in-progress: false -jobs: - diagnose: - if: github.repository == 'tappe9/simple-api-benchmark' - runs-on: ubuntu-24.04 - timeout-minutes: 20 - env: - PYTHONDONTWRITEBYTECODE: '1' - steps: - - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 - with: - fetch-depth: 0 - persist-credentials: false - - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 - with: - python-version-file: apps/python-fastapi/.python-version - - name: Diagnostic regression tests - run: python -m unittest discover -s tests -p test_benchmark_flask_diagnostic.py -v - - name: Fixed non-publishing hypothesis matrix - run: python -m benchmark.flask_diagnostic --phase scheduling - - name: Confirm production inputs and published outputs stay unchanged - run: | - git diff --exit-code HEAD - test -z "$(git ls-files --others --exclude-standard)" - - name: Export exact source for offline review - if: always() - run: | - mkdir -p .cache/flask-diagnostic/source - git bundle create .cache/flask-diagnostic/source/source.bundle HEAD - git rev-parse HEAD HEAD^{tree} > .cache/flask-diagnostic/source/identity.txt - - name: Export pinned formatter for offline review - if: always() - run: | - python -m pip install --require-hashes --only-binary=:all: -r apps/python-fastapi/requirements-dev.lock - mkdir -p .cache/flask-diagnostic/review - cp "$(command -v ruff)" .cache/flask-diagnostic/review/ruff - sha256sum .cache/flask-diagnostic/review/ruff > .cache/flask-diagnostic/review/ruff.sha256 - - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 - if: always() - with: - name: flask-diagnostic-${{ github.run_id }}-${{ github.run_attempt }} - path: .cache/flask-diagnostic/ - include-hidden-files: true - retention-days: 30 - if-no-files-found: error diff --git a/benchmark/flask_diagnostic.py b/benchmark/flask_diagnostic.py index 092a938..6abd045 100644 --- a/benchmark/flask_diagnostic.py +++ b/benchmark/flask_diagnostic.py @@ -29,9 +29,15 @@ def diagnostic_plan(phase: str = "initial") -> list[dict]: """Fix order before collecting data; reverse arms and endpoints in block two.""" require(phase in ("initial", "scheduling"), "unknown diagnostic phase") if phase == "scheduling": - return [{"id": f"block-{index}-{arm}", "block": index, "arm": arm, - "endpoints": ["/json", "/db/42"]} - for index, arm in enumerate(("baseline", "affinity-one", "baseline"), 1)] + return [ + { + "id": f"block-{index}-{arm}", + "block": index, + "arm": arm, + "endpoints": ["/json", "/db/42"], + } + for index, arm in enumerate(("baseline", "affinity-one", "baseline"), 1) + ] return [ { "id": f"block-{block}-{arm}", @@ -69,7 +75,7 @@ def hook_source(arm: str) -> str: "import json, logging, os, sys\n" "allowed_before = sorted(os.sched_getaffinity(0))\n" + intervention - + '\nif os.getpid() == 1:\n' + + "\nif os.getpid() == 1:\n" + ' print("FLASK_DIAGNOSTIC " + json.dumps({"arm": ' + repr(arm) + ', "pid": os.getpid(), "interval": sys.getswitchinterval(), ' @@ -122,16 +128,25 @@ def start(self, identifier: str) -> dict: result = super().start(identifier) if self.arm != "baseline": logs = execute(["docker", "logs", "--tail", "100", self.container], timeout=10) - markers = [line.removeprefix("FLASK_DIAGNOSTIC ") for line in logs.splitlines() - if line.startswith("FLASK_DIAGNOSTIC ")] + markers = [ + line.removeprefix("FLASK_DIAGNOSTIC ") + for line in logs.splitlines() + if line.startswith("FLASK_DIAGNOSTIC ") + ] require(len(markers) == 1, "diagnostic startup hook was not observed exactly once") observed = strict_json(markers[0].encode()) - require(observed.get("arm") == self.arm and observed.get("pid") == 1, - "hook belongs to wrong process/arm") - require(observed.get("queue_disabled") is (self.arm == "queue-quiet"), - "queue logger intervention mismatch") - require(observed.get("interval") == (0.001 if self.arm == "switch-1ms" else 0.005), - "thread interval intervention mismatch") + require( + observed.get("arm") == self.arm and observed.get("pid") == 1, + "hook belongs to wrong process/arm", + ) + require( + observed.get("queue_disabled") is (self.arm == "queue-quiet"), + "queue logger intervention mismatch", + ) + require( + observed.get("interval") == (0.001 if self.arm == "switch-1ms" else 0.005), + "thread interval intervention mismatch", + ) before = observed.get("allowed_before") require(type(before) is list and bool(before), "missing allowed CPU set") expected = before[:1] if self.arm == "affinity-one" else before @@ -165,16 +180,31 @@ def observe(self, label: str) -> dict: """ raw = execute(["docker", "exec", self.container, "python", "-c", script], timeout=10) counters = strict_json(raw.encode()) - logs = execute(["docker", "logs", "--timestamps", "--since", self.since, - "--tail", "10000", self.container], timeout=10) + logs = execute( + [ + "docker", + "logs", + "--timestamps", + "--since", + self.since, + "--tail", + "10000", + self.container, + ], + timeout=10, + ) encoded = logs.encode() - limited = encoded[-2 * 1024 * 1024:] + limited = encoded[-2 * 1024 * 1024 :] log_path = self.artifacts / f"{label}.server.log" log_path.write_bytes(limited) snapshot = { - "at": now(), "container_id": self.container, "counters": counters, - "server_log": log_path.name, "log_tail_limit_lines": 10000, - "log_tail_may_be_truncated": len(logs.splitlines()) >= 10000 or len(limited) != len(encoded), + "at": now(), + "container_id": self.container, + "counters": counters, + "server_log": log_path.name, + "log_tail_limit_lines": 10000, + "log_tail_may_be_truncated": len(logs.splitlines()) >= 10000 + or len(limited) != len(encoded), "queue_warning_lines_in_tail": limited.count(b"Task queue depth is"), } atomic_json(self.artifacts / f"{label}.observation.json", snapshot) @@ -182,7 +212,7 @@ def observe(self, label: str) -> dict: return snapshot -def run_cell(environment, cell: dict, *, contract=run_contract) -> dict: +def run_cell(environment, cell: dict, *, contract=run_contract, expected_image=None) -> dict: failure = None try: require(environment.health_policy == EXTERNAL_READINESS, "diagnostic health policy differs") @@ -190,8 +220,15 @@ def run_cell(environment, cell: dict, *, contract=run_contract) -> dict: require(environment.request_timeout == 15, "diagnostic request timeout must be 15s") environment.build(IMPLEMENTATION) container = environment.start(IMPLEMENTATION) + if expected_image is not None: + require( + container.get("image_id") == expected_image, + "diagnostic image changed between cells", + ) checks = contract("http://127.0.0.1:8080", implementation=IMPLEMENTATION) - require(type(checks) is int and checks == 2 * len(load_cases()), "incomplete shared contract") + require( + type(checks) is int and checks == 2 * len(load_cases()), "incomplete shared contract" + ) environment.check() endpoints = [] for endpoint in cell["endpoints"]: @@ -205,8 +242,16 @@ def run_cell(environment, cell: dict, *, contract=run_contract) -> dict: validate_run(observed) runs.append(observed) after = environment.observe(label + "-after") - endpoints.append({"endpoint": endpoint, "warmup": warmup, "runs": runs, - "selected": select_run(runs), "before": before, "after": after}) + endpoints.append( + { + "endpoint": endpoint, + "warmup": warmup, + "runs": runs, + "selected": select_run(runs), + "before": before, + "after": after, + } + ) environment.check() readiness = copy.deepcopy(environment.readiness) except BaseException as error: @@ -220,50 +265,109 @@ def run_cell(environment, cell: dict, *, contract=run_contract) -> dict: if failure is not None: raise BenchmarkFailure(f"{failure}; cleanup also failed: {error}") from failure raise - return {**cell, "container": container, "readiness": readiness, "contract_checks": checks, - "artifact_directory": str(environment.artifacts), "endpoints": endpoints} + return { + **cell, + "container": container, + "readiness": readiness, + "contract_checks": checks, + "artifact_directory": str(environment.artifacts), + "endpoints": endpoints, + } -def run_diagnostic(factory, output: Path, *, metadata: dict, contract=run_contract, phase="initial") -> dict: +def run_diagnostic( + factory, output: Path, *, metadata: dict, contract=run_contract, phase="initial" +) -> dict: output = validate_output_path(output) for key in ("source_commit", "source_tree"): - require(type(metadata.get(key)) is str and re.fullmatch(r"[0-9a-f]{40}", metadata[key]), - "diagnostic source metadata missing " + key) + require( + type(metadata.get(key)) is str and re.fullmatch(r"[0-9a-f]{40}", metadata[key]), + "diagnostic source metadata missing " + key, + ) versions = metadata.get("versions", {}) - require(set(versions) == {IMPLEMENTATION} and - set(implementation(IMPLEMENTATION)["version_fields"]) <= set(versions[IMPLEMENTATION]), - "diagnostic requires Flask version metadata") + require( + set(versions) == {IMPLEMENTATION} + and set(implementation(IMPLEMENTATION)["version_fields"]) <= set(versions[IMPLEMENTATION]), + "diagnostic requires Flask version metadata", + ) plan = diagnostic_plan(phase) started_at = now() cells = [] progress = output.parent / "progress.json" for cell in plan: - atomic_json(progress, {"status": "running", "official": False, "publishable": False, - "metadata": metadata, "plan": plan, "active_cell": cell["id"], "cells": cells}) + atomic_json( + progress, + { + "status": "running", + "official": False, + "publishable": False, + "metadata": metadata, + "plan": plan, + "active_cell": cell["id"], + "cells": cells, + }, + ) print(f"Flask diagnostic {cell['id']}", flush=True) try: environment = factory(cell) - cells.append(run_cell(environment, cell, contract=contract)) + expected_image = cells[0]["container"]["image_id"] if cells else None + cells.append( + run_cell(environment, cell, contract=contract, expected_image=expected_image) + ) except BaseException as error: - atomic_json(progress, {"status": "failed", "official": False, "publishable": False, - "metadata": metadata, "plan": plan, "failed_cell": cell["id"], - "error_type": type(error).__name__, "cells": cells}) + atomic_json( + progress, + { + "status": "failed", + "official": False, + "publishable": False, + "metadata": metadata, + "plan": plan, + "failed_cell": cell["id"], + "error_type": type(error).__name__, + "cells": cells, + }, + ) raise result = { - "schema_version": 1, "mode": "flask-diagnostic", "status": "verified", - "official": False, "publishable": False, "implementation": IMPLEMENTATION, - "started_at": started_at, "completed_at": now(), "metadata": copy.deepcopy(metadata), "phase": phase, - "conditions": {"connections": 50, "warmup_seconds": 5, "duration_seconds": 10, - "runs": 3, "api_health_policy": EXTERNAL_READINESS}, - "limitations": ["Non-publishing diagnostic, not an official result or framework ranking.", - "CPU snapshots span warm-up and measurement plus observation overhead.", - "Server logs are bounded tails; warning counts may be lower bounds.", - "Two blocks do not establish cross-host or universal performance effects."], - "plan": plan, "cells": cells, + "schema_version": 1, + "mode": "flask-diagnostic", + "status": "verified", + "official": False, + "publishable": False, + "implementation": IMPLEMENTATION, + "started_at": started_at, + "completed_at": now(), + "metadata": copy.deepcopy(metadata), + "phase": phase, + "conditions": { + "connections": 50, + "warmup_seconds": 5, + "duration_seconds": 10, + "runs": 3, + "api_health_policy": EXTERNAL_READINESS, + }, + "limitations": [ + "Non-publishing diagnostic, not an official result or framework ranking.", + "CPU snapshots span warm-up and measurement plus observation overhead.", + "Server logs are bounded tails; warning counts may be lower bounds.", + "A bounded diagnostic matrix does not establish cross-host or universal performance effects.", + ], + "plan": plan, + "cells": cells, } atomic_json(output, result) - atomic_json(progress, {"status": "completed", "official": False, "publishable": False, - "metadata": metadata, "plan": plan, "cells": cells}) + atomic_json( + progress, + { + "status": "completed", + "official": False, + "publishable": False, + "metadata": metadata, + "plan": plan, + "cells": cells, + }, + ) return result @@ -286,9 +390,14 @@ def interrupted(signum, _frame): metadata = provenance(oha) metadata["versions"] = {IMPLEMENTATION: registered_pinned_versions()[IMPLEMENTATION]} metadata["diagnostic_plan_sha256"] = hashlib.sha256( - json.dumps(diagnostic_plan(args.phase), sort_keys=True).encode()).hexdigest() - run_diagnostic(lambda cell: FlaskEnvironment(oha, directory / cell["id"], cell["arm"]), - output, metadata=metadata, phase=args.phase) + json.dumps(diagnostic_plan(args.phase), sort_keys=True).encode() + ).hexdigest() + run_diagnostic( + lambda cell: FlaskEnvironment(oha, directory / cell["id"], cell["arm"]), + output, + metadata=metadata, + phase=args.phase, + ) print(f"Flask diagnostic complete: {output}; no results published.") return 0 except (BenchmarkFailure, ContractFailure, OSError, ValueError, KeyboardInterrupt) as error: diff --git a/docs/FLASK-DIAGNOSTIC.md b/docs/FLASK-DIAGNOSTIC.md new file mode 100644 index 0000000..9a1834c --- /dev/null +++ b/docs/FLASK-DIAGNOSTIC.md @@ -0,0 +1,189 @@ +# Flask JSON diagnostic + +## Conclusion and boundary + +Issue #70 found a reproducible **CPU-placement-sensitive scheduling effect** in +this repository's pinned CPython / Flask / Waitress stack under a one-CPU quota. +It did not establish that Flask's JSON handler is intrinsically slow, that the +published numbers were fabricated, or that a particular GIL/kernel function is +the root cause. + +In one same-job baseline / one-CPU-affinity / baseline diagnostic, JSON throughput +was approximately 102 / 1,883 / 103 requests/s. DB throughput stayed approximately +1,190 / 1,204 / 1,207 requests/s. The CPU quota, memory limit, production image, +Waitress worker count and DB pool limit were unchanged. Affinity selected one +already-allowed logical CPU; it did not reserve a dedicated physical core. + +These are **non-publishing diagnostic observations**, not replacement official +results, an approved production optimization, or a cross-framework ranking. +Existing applications, pins, official cohorts, methodology and `results/**` were +not changed. Neither #52's deferred publishing/protection work nor #59's next +justified official-run verification is completed by these experiments. + +## Retained evidence + +All sources below are immutable measured snapshots, not whichever branch is +currently checked out. The corresponding run artifacts were downloaded and their +ZIP SHA-256 values checked during the 2026-09-18 investigation. + +| Evidence | Source commit | Run / attempt | Artifact ID | +| --- | --- | --- | --- | +| Original eight-stack publication | `c8e033ea176366644f5bed5ded12cab79e6fcbeb` | [34925168324 / 1](https://github.com/tappe9/simple-api-benchmark/actions/runs/34925168324) | `10380696186` | +| Initial hypothesis matrix | `cbebd385b0f78503d797fcc73167f1ecce1b8611` | [35299848616 / 1](https://github.com/tappe9/simple-api-benchmark/actions/runs/35299848616) | `10529950752` | +| Scheduling probe | `a153a103f445af8cd571687fb095cc0d2f4cb9a3` | [35300974388 / 1](https://github.com/tappe9/simple-api-benchmark/actions/runs/35300974388) | `10529473210` | + +ZIP SHA-256 values, in the same order: + +```text +cc376a99e573d65c5cedc165ee00fda294795f45ed491a5586ef419f96b9989b +4df46b3f0843e1486d9fe3382c2962efe952dd414283c2f46ba731f6de0e9bf8 +ae16af92916f62f198ed509d57273831b5b118c8de4c2f054d57037e42123bec +``` + +The original selected JSON matched the published file byte-for-byte at audit +main `9b92c74cba34bd5a4bfc389075b9bef135131034`. All 72 original measured windows +passed `benchmark.report.audit_raw`. The original Flask JSON run 2 had mean +response time 315.451675 ms and mean first-byte time 315.450804 ms: downloading +the small body does not explain the observed latency. + +Both diagnostic archives were independently replayed against `parse_oha`, the +memory samples, selected whole-run values, observation files and source/tree +identities: **36 + 18 measured windows and 12 + 6 warm-ups**, with no transport +errors or non-200 responses. Every cell used a fresh container; all cells within +each job used one identical image ID. Image IDs between separate jobs need not +match and those jobs must not be pooled as paired measurements. The original +publication does not contain the later per-thread or throttling observations. + +Artifacts have finite retention; their availability was verified on the audit +date, not guaranteed forever. An expired original artifact is not recreated by +rerunning a diagnostic. See [Issue #70](https://github.com/tappe9/simple-api-benchmark/issues/70) +for the predeclared plans and review record. + +## Experiment 1: switch interval and queue logging + +The predefined arms were baseline, Python switch interval 1 ms, and disabled +`waitress.queue` warning output. The second block reversed both arm order and +JSON/DB endpoint order. Each cell recreated PostgreSQL and the API container. +Only the named intervention was applied through a read-only diagnostic startup +hook; baseline had no hook. + +Common settings were HTTP/1.1, 50 connections, external readiness stopped before +load, a five-second warm-up and three ten-second measured windows per endpoint. +The unchanged environment enforced one CPU, 512 MiB, one Waitress worker thread +and DB pool maximum 10. These ten-second windows are not the official thirty- +second profile. + +| Block | Arm | JSON requests/s | DB requests/s | +| --- | --- | ---: | ---: | +| 1 | baseline | 95.615 | 1,196.952 | +| 1 | switch-1ms | 92.682 | 1,215.155 | +| 1 | queue-quiet | 77.811 | 1,425.894 | +| 2 | queue-quiet | 81.356 | 1,388.971 | +| 2 | switch-1ms | 92.198 | 1,203.858 | +| 2 | baseline | 103.816 | 1,205.640 | + +The JSON/DB inversion reproduced. Neither tested intervention improved JSON in +this experiment. Suppressing warnings changed DB throughput, but did not explain +or fix the JSON anomaly. This does not exclude every possible GIL or logging +interaction; it rejects these two simple proposed remedies under the tested +conditions. + +## Experiment 2: bracketing CPU-affinity probe + +The follow-up plan was recorded before execution. On a new job it ran baseline, +`affinity-one`, then baseline, with fresh environments and JSON then DB in every +cell. Load settings stayed as above. A startup marker confirmed the intervention +in PID 1; `/proc/1/task/*/status` snapshots confirmed the observed threads' CPU +masks. The helper interpreter's metadata is separately labeled, not passed off +as PID 1's runtime configuration. + +| Arm | JSON requests/s | JSON mean ms | DB requests/s | +| --- | ---: | ---: | ---: | +| baseline before | 101.553 | 491.030 | 1,190.344 | +| affinity-one | 1,882.871 | 26.508 | 1,204.494 | +| baseline after | 102.701 | 485.357 | 1,206.691 | + +JSON throughput in the affinity arm was 18.54 and 18.33 times the two bracketing +baselines. These are within-diagnostic contrasts, not an improvement applied to +an existing official publication. The baselines could run on logical CPUs 0-3; +the affinity arm and its observed threads used CPU 0, still with the one-CPU +quota and identical production image. + +JSON observation-span counter deltas were: + +| Arm | Throttled periods / periods | Accumulated throttled seconds | Main-thread CPU seconds | Waitress-worker CPU seconds | Worker scheduler slices | +| --- | ---: | ---: | ---: | ---: | ---: | +| baseline before | 349 / 363 | 13.613507 | 26.358 | 9.030 | 1,081,401 | +| affinity-one | 25 / 360 | 0.003450 | 17.314 | 17.768 | 28,258 | +| baseline after | 350 / 363 | 13.393709 | 26.353 | 9.038 | 1,066,044 | + +These cumulative spans include warm-up, measured windows and observation overhead. +They are not per-request profiles. `schedstat` scheduler slices are not a count +of GIL handoffs, and accumulated throttling time is not a latency decomposition. +DB-pool helper threads showed no material CPU growth in these snapshots. + +## Interpretation and next decision + +[Waitress's documented design](https://docs.pylonsproject.org/projects/waitress/en/stable/design.html) +separates main-thread client I/O from worker-thread application execution. +[CPU bandwidth control](https://docs.kernel.org/scheduler/sched-bwc.html) limits +aggregate runtime using quota and period; a one-CPU quota is not the same setting +as restricting execution to one logical CPU. Those distinctions explain why +"one process / one worker / one CPU" does not specify every scheduling condition. + +The affinity intervention, bracketing baselines and thread/cgroup observations +support a scheduling/CPU-quota interaction in this particular stack. GIL +contention, I/O wakeups, lock contention and kernel scheduling are plausible +lower-level mechanisms, but these experiments do not isolate one of them. No +stack sampling or controlled quota-only experiment was performed. In particular, +it would overstate the evidence to say that the GIL alone, CPU throttling alone, +or a particular Waitress defect was proved responsible. Host noise, one treatment +cell and lack of independent-host replication limit generalization. + +The supported next action is to decide and document common CPU-allocation +semantics before changing any official comparison. Keeping the existing quota- +only baseline with an explanation is valid. Adopting affinity would require a +separately reviewed common-methodology decision across the complete cohort, not a +Flask-only boost or a silent overwrite of history. Any later official run still +requires explicit authorization for measurement, audited publication and Pages. + +## Reproducing the diagnostics + +Use a clean committed checkout, Python 3.10+, Git, curl, Make and a local Linux +Docker Compose v2 daemon reachable through a Unix socket. Leave port 8080 free +and do not run other benchmark services concurrently. No host Flask installation +is needed; the existing pinned production container is built and exercised. + +```bash +# No Docker or load: regression tests and CLI help. +python -m unittest discover -s tests -p test_benchmark_flask_diagnostic.py -v +python -m benchmark.flask_diagnostic --help + +# Explicit non-publishing invocations; do not run both concurrently. +python -m benchmark.flask_diagnostic --phase initial +python -m benchmark.flask_diagnostic --phase scheduling +``` + +The initial phase has 36 measured windows and 12 warm-ups; scheduling has 18 +measured windows and six warm-ups. Both retain all attempted samples without +selective retries under `.cache/flask-diagnostic//`. The final +`diagnostic.json` is written only after every cell and its scoped cleanup succeed; +`progress.json` and raw files retain failure evidence. Each invocation uses new +owned Compose projects. CLI help and invalid arguments exit before tool +installation or load. An image change between cells now fails before the next +cell's contract/load; the already measured archives had identical images and +are unchanged by this additional guard. + +The report records source/tree, versions, locks, conditions, plan digest, +container identities, readiness, interventions, raw run summaries, memory, +bounded server-log tails and outside-load observations. Tail warning counts may +be lower bounds. Collection introduces overhead outside individual oha windows. +SIGINT/SIGTERM attempt bounded cleanup; SIGKILL/host failure cannot guarantee it. +Inspect the printed owned project name before manually removing leftover +resources, and never use broad Docker cleanup commands for this investigation. + +There is no permanent diagnostic workflow or new automatic measurement. The +branch-only investigation workflow was removed after its evidence was retained. +Diagnostics cannot feed the official publisher or Pages. Investigation scratch +notes and Superpowers plans remain untracked under the paths in `AGENTS.md`; +this guide is the durable public explanation, not an implementation plan. diff --git a/docs/FLASK.md b/docs/FLASK.md index 26b5db6..54724fe 100644 --- a/docs/FLASK.md +++ b/docs/FLASK.md @@ -51,3 +51,10 @@ The gates cover unit behavior, dependency/hash locks, real PostgreSQL and Docker ## Publication boundary Python / Flask is registered and CI-covered, but Issue #32 does not activate a new official benchmark cohort. `active_cohort` remains the frozen `four-stack-v1`; expanded-cohort activation and new official measurements are separate work. + +## JSON performance diagnostic + +The [Flask diagnostic guide](FLASK-DIAGNOSTIC.md) records Issue #70's controlled +non-publishing experiments, CPU-placement sensitivity, evidence limits and +reproduction commands. Diagnostic values do not replace the official results or +approve a Flask-specific runtime/CPU tuning change. diff --git a/tests/test_benchmark_flask_diagnostic.py b/tests/test_benchmark_flask_diagnostic.py index 6039993..23b64d3 100644 --- a/tests/test_benchmark_flask_diagnostic.py +++ b/tests/test_benchmark_flask_diagnostic.py @@ -74,19 +74,29 @@ def factory(cell): def contract(url, *, implementation): return self.environments[-1].contract(url, implementation=implementation) - return self.module.run_diagnostic(factory, self.output, metadata=self.metadata, contract=contract, phase=phase) + return self.module.run_diagnostic( + factory, self.output, metadata=self.metadata, contract=contract, phase=phase + ) def test_fixed_counterbalanced_plan_and_every_sample_are_preserved(self): result = self.run_diagnostic() - self.assertEqual([cell["arm"] for cell in result["cells"]], - ["baseline", "switch-1ms", "queue-quiet", "queue-quiet", "switch-1ms", "baseline"]) + self.assertEqual( + [cell["arm"] for cell in result["cells"]], + ["baseline", "switch-1ms", "queue-quiet", "queue-quiet", "switch-1ms", "baseline"], + ) self.assertEqual(len(self.environments), 6) for position, (cell, environment) in enumerate(zip(result["cells"], self.environments)): endpoints = ["/json", "/db/42"] if position < 3 else ["/db/42", "/json"] self.assertEqual([item["endpoint"] for item in cell["endpoints"]], endpoints) measured = [event for event in environment.events if isinstance(event, tuple)] - self.assertEqual(measured, [("python-flask", endpoint, 5 if index == 0 else 10, index) - for endpoint in endpoints for index in (0, 1, 2, 3)]) + self.assertEqual( + measured, + [ + ("python-flask", endpoint, 5 if index == 0 else 10, index) + for endpoint in endpoints + for index in (0, 1, 2, 3) + ], + ) self.assertEqual(environment.events.count("cleanup"), 1) self.assertIsNone(environment.active) self.assertEqual(cell["readiness"], {"attempts": 1}) @@ -99,7 +109,17 @@ def test_fixed_counterbalanced_plan_and_every_sample_are_preserved(self): validate_report(result) def test_all_failure_stages_clean_and_never_write_partial_success(self): - for stage in ("build", "startup", "contract", "state", "warmup", "measurement", "metric", "observe", "cleanup"): + for stage in ( + "build", + "startup", + "contract", + "state", + "warmup", + "measurement", + "metric", + "observe", + "cleanup", + ): for folder in self.cache.iterdir(): if folder.is_dir(): shutil.rmtree(folder) @@ -113,18 +133,23 @@ def test_all_failure_stages_clean_and_never_write_partial_success(self): def test_cleanup_precedes_successful_output_and_retains_readiness(self): def alter(environment): original = environment.cleanup + def cleanup(): self.assertEqual(self.output.read_bytes(), b"previous diagnostic") original() environment.readiness = None + environment.cleanup = cleanup + self.run_diagnostic(alter=alter) def test_interruption_cleans_and_does_not_retry(self): def alter(environment): def interrupted(*args): raise KeyboardInterrupt + environment.measure = interrupted + with self.assertRaises(KeyboardInterrupt): self.run_diagnostic(alter=alter) self.assertEqual(len(self.environments), 1) @@ -134,12 +159,17 @@ def interrupted(*args): def test_incomplete_contract_cannot_start_load(self): def alter(environment): environment.contract = lambda *args, **kwargs: True + with self.assertRaisesRegex(BenchmarkFailure, "contract"): self.run_diagnostic(alter=alter) self.assertEqual(self.environments[0].measures, 0) def test_invalid_profile_or_metadata_prevents_load(self): - for field, value in (("connections", 2), ("request_timeout", 1), ("health_policy", "container-healthcheck")): + for field, value in ( + ("connections", 2), + ("request_timeout", 1), + ("health_policy", "container-healthcheck"), + ): with self.subTest(field=field), self.assertRaises(BenchmarkFailure): self.run_diagnostic(alter=lambda environment: setattr(environment, field, value)) self.assertNotIn("build", self.environments[-1].events) @@ -167,6 +197,34 @@ def alter(environment): self.assertEqual(len(progress["cells"]), 1) self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + def test_image_drift_stops_before_loading_the_next_cell(self): + count = 0 + + def alter(environment): + nonlocal count + count += 1 + if count == 2: + original = environment.start + + def changed_image(identifier): + result = original(identifier) + result["image_id"] = "sha256:changed-test-image" + return result + + environment.start = changed_image + + with self.assertRaisesRegex(BenchmarkFailure, "image changed"): + self.run_diagnostic(alter=alter, phase="scheduling") + self.assertEqual(len(self.environments), 2) + self.assertEqual(self.environments[0].measures, 6) + self.assertEqual(self.environments[1].measures, 0) + self.assertTrue(all(e.events.count("cleanup") == 1 for e in self.environments)) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + self.assertEqual(self.published.read_bytes(), b"previous official result") + progress = json.loads((self.cache / "progress.json").read_bytes()) + self.assertEqual(progress["status"], "failed") + self.assertEqual(len(progress["cells"]), 1) + def test_invalid_measurement_is_not_retried_or_selected(self): def alter(environment): original = environment.measure @@ -201,11 +259,16 @@ def test_output_and_symlinks_cannot_escape_cache(self): def test_startup_hooks_change_only_the_declared_factor(self): for arm in ("baseline", "switch-1ms", "queue-quiet"): code = self.module.hook_source(arm) - probe = ('import sys, logging, json\n' + code + - '\nprint(json.dumps({"interval":sys.getswitchinterval(),"queue":logging.getLogger("waitress.queue").disabled,"root":logging.getLogger().disabled}))') + probe = ( + "import sys, logging, json\n" + + code + + '\nprint(json.dumps({"interval":sys.getswitchinterval(),"queue":logging.getLogger("waitress.queue").disabled,"root":logging.getLogger().disabled}))' + ) output = subprocess.check_output([sys.executable, "-c", probe], text=True, timeout=5) observed = json.loads(output.splitlines()[-1]) - self.assertEqual(observed["interval"], 0.001 if arm == "switch-1ms" else sys.getswitchinterval()) + self.assertEqual( + observed["interval"], 0.001 if arm == "switch-1ms" else sys.getswitchinterval() + ) self.assertEqual(observed["queue"], arm == "queue-quiet") self.assertFalse(observed["root"]) with self.assertRaises(BenchmarkFailure): @@ -240,7 +303,10 @@ def test_affinity_hook_changes_only_the_child_allowed_cpu_set(self): def test_help_and_invalid_arguments_never_start_measurement(self): for args, code in ((["--help"], 0), (["--unknown"], 2)): with self.subTest(args=args), patch.object(sys, "argv", ["flask-diagnostic", *args]): - with patch("benchmark.install_oha.ensure_oha", side_effect=AssertionError("load must not start")): + with patch( + "benchmark.install_oha.ensure_oha", + side_effect=AssertionError("load must not start"), + ): with self.assertRaises(SystemExit) as stopped: self.module.main() self.assertEqual(stopped.exception.code, code)