perf(tests): replace behave-parallel subprocess model with in-process parallelism #490

Closed
brent.edwards wants to merge 1 commits from perf/in-process-parallel-behave into perf/optimize-medium-features
+292 -215
View File
@@ -1,8 +1,6 @@
import json
import os
import sys
import tarfile
import urllib.request
from pathlib import Path
import nox
@@ -13,11 +11,7 @@ SUPPORTED_PYTHONS = ["3.13"]
nox.options.reuse_existing_virtualenvs = True
nox.options.error_on_external_run = True
BEHAVE_PARALLEL_VERSION = "1.2.4a1"
BEHAVE_PARALLEL_URL = (
"https://files.pythonhosted.org/packages/05/9d/22f74dd77bc4fa85d391564a232c49b4e99cfdeac7bfdee8151ea4606632/"
f"behave-parallel-{BEHAVE_PARALLEL_VERSION}.tar.gz"
)
BEHAVE_PARALLEL_VERSION = "2.0.0"
def _default_processes() -> int:
@@ -88,58 +82,193 @@ def _create_template_db(session: nox.Session) -> str:
def _install_behave_parallel(session: nox.Session) -> None:
"""Install behave-parallel with a custom CLI wrapper."""
"""Install behave-parallel with an in-process parallel runner.
Instead of spawning one subprocess per feature file (the old model),
the new CLI runs features in-process via behave's Python API. Step
definitions and environment hooks are loaded once; each feature file
is then parsed and executed without interpreter startup overhead.
Parallel execution uses ``multiprocessing.Pool`` with the ``fork``
start method so that already-imported modules are shared (copy-on-write)
across workers.
"""
tmp_dir = Path(session.create_tmp())
archive_path = tmp_dir / "behave-parallel.tar.gz"
with (
urllib.request.urlopen(BEHAVE_PARALLEL_URL) as response,
archive_path.open("wb") as target,
):
target.write(response.read())
source_dir = tmp_dir / "behave-parallel-inprocess"
source_dir.mkdir(parents=True, exist_ok=True)
with tarfile.open(archive_path, "r:gz") as tar:
tar.extractall(tmp_dir)
source_dir = next(tmp_dir.glob("behave-parallel-*/"))
# Recreate package with a parallel-aware CLI that aggregates summaries
pkg_dir = source_dir / "behave_parallel"
pkg_dir.mkdir(parents=True, exist_ok=True)
(pkg_dir / "__init__.py").write_text("\n")
(pkg_dir / "cli.py").write_text(
"""
(pkg_dir / "cli.py").write_text(_BEHAVE_PARALLEL_CLI_SOURCE)
setup_path = source_dir / "setup.py"
setup_path.write_text(
"from setuptools import find_packages, setup\n"
"setup(\n"
' name="behave-parallel",\n'
f' version="{BEHAVE_PARALLEL_VERSION}",\n'
" packages=find_packages(),\n"
' install_requires=["behave>=1.2.6"],\n'
" entry_points={\n"
' "console_scripts": '
'["behave-parallel=behave_parallel.cli:main"],\n'
" },\n"
")\n"
)
session.install("setuptools", "wheel")
session.install(str(source_dir))
# The in-process behave-parallel CLI source is kept as a module-level
# constant so the noxfile itself stays readable and ``ruff`` can still
# lint the surrounding Python without tripping over a giant raw string
# embedded inside a function body.
_BEHAVE_PARALLEL_CLI_SOURCE = r'''
"""In-process parallel behave runner.
Replaces the old subprocess-per-feature model with direct use of
behave's ``Runner`` API. Step definitions and environment hooks are
loaded once per process; feature files are parsed and executed without
Python interpreter startup overhead.
Parallelism modes
-----------------
* **Sequential** (``--processes 1`` or ``BEHAVE_PARALLEL_COVERAGE=1``):
All features run in a single ``Runner.run()`` call.
* **Parallel** (``--processes N``, N > 1, no coverage):
Features are split into *N* equal-size chunks. A
``multiprocessing.Pool`` with the ``fork`` start method dispatches
each chunk to a worker that creates its own ``Runner`` (hooks and
step definitions are re-loaded cheaply because all heavy modules are
already in memory from the parent).
"""
from __future__ import annotations
import argparse
import concurrent.futures
import io
import multiprocessing
import os
import re
import subprocess
import sys
import uuid
import time
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
from typing import Dict, Iterable, List, Tuple
DEFAULT_FEATURE_ROOT = "features/"
SUMMARY_PATTERNS = {
"features": re.compile(
r"(\\d+)\\s+feature(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
),
"scenarios": re.compile(
r"(\\d+)\\s+scenario(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
),
"steps": re.compile(
r"(\\d+)\\s+step(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
),
}
DURATION_PATTERN = re.compile(r"Took\\s+(?:(\\d+)m\\s*)?([\\d\\.]+)s")
def _extract_features_and_args(argv: List[str]) -> Tuple[List[str], List[str]]:
positional: List[str] = []
options: List[str] = []
# ---------------------------------------------------------------------------
# Summary helpers
# ---------------------------------------------------------------------------
def _empty_summary():
return {
"features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"duration": 0.0,
}
def _extract_summary(runner):
"""Build a summary dict from a completed behave Runner."""
summary = _empty_summary()
for feature in runner.features:
status_name = feature.status.name
if status_name == "passed":
summary["features"]["passed"] += 1
elif status_name == "skipped":
summary["features"]["skipped"] += 1
else:
summary["features"]["failed"] += 1
summary["duration"] += feature.duration or 0.0
for scenario in feature.walk_scenarios():
sname = scenario.status.name
if sname == "passed":
summary["scenarios"]["passed"] += 1
elif sname == "skipped":
summary["scenarios"]["skipped"] += 1
elif sname == "failed":
summary["scenarios"]["failed"] += 1
else:
summary["scenarios"]["errors"] += 1
for step in scenario.steps:
stname = step.status.name
if stname == "passed":
summary["steps"]["passed"] += 1
elif stname == "skipped":
summary["steps"]["skipped"] += 1
elif stname == "failed":
summary["steps"]["failed"] += 1
else:
summary["steps"]["errors"] += 1
return summary
def _merge_summaries(summaries):
total = _empty_summary()
for s in summaries:
for bucket in ("features", "scenarios", "steps"):
for field in ("passed", "failed", "errors", "skipped"):
total[bucket][field] += s.get(bucket, {}).get(field, 0)
total["duration"] += float(s.get("duration", 0.0))
return total
def _format_duration(seconds):
minutes = int(seconds // 60)
remainder = seconds % 60
if minutes:
return f"{minutes}m {remainder:.3f}s"
return f"{remainder:.3f}s"
def _print_overall_summary(total, wall_seconds=None):
print("\nOverall summary:")
for bucket in ("features", "scenarios", "steps"):
b = total[bucket]
print(
f"{b['passed']} {bucket} passed, {b['failed']} failed, "
f"{b['errors']} errored, {b['skipped']} skipped"
)
if total["duration"]:
print(f"Took {_format_duration(float(total['duration']))}")
if wall_seconds is not None:
print(f"Wall time: {_format_duration(wall_seconds)}")
def _has_failures(total):
return (
total["features"]["failed"] > 0
or total["features"]["errors"] > 0
or total["scenarios"]["failed"] > 0
or total["scenarios"]["errors"] > 0
)
# ---------------------------------------------------------------------------
# Feature discovery
# ---------------------------------------------------------------------------
def _iter_features(paths):
collected = []
for path in paths:
p = Path(path)
if p.is_dir():
collected.extend(sorted(str(fp) for fp in p.rglob("*.feature")))
else:
collected.append(str(p))
return collected
def _extract_features_and_args(argv):
positional = []
options = []
for arg in argv:
if arg.startswith("-"):
options.append(arg)
@@ -150,128 +279,82 @@ def _extract_features_and_args(argv: List[str]) -> Tuple[List[str], List[str]]:
return [DEFAULT_FEATURE_ROOT], options
def _empty_summary() -> Dict[str, Dict[str, int] | float]:
return {
"features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"duration": 0.0,
}
# ---------------------------------------------------------------------------
# In-process behave execution
# ---------------------------------------------------------------------------
def _make_runner(feature_paths, behave_args):
"""Create a behave Runner with proper configuration defaults.
Mirrors the format-defaulting logic from ``behave.__main__.run_behave``
so that ``-q`` and bare invocations get a sensible formatter instead
of crashing on ``config.format is None``.
"""
from behave.configuration import Configuration
from behave.runner import Runner
from behave.runner_util import reset_runtime
reset_runtime()
args = list(behave_args) + [str(p) for p in feature_paths]
config = Configuration(command_args=args)
if not config.format:
config.format = [config.default_format]
return Runner(config)
def _parse_summary(text: str) -> Dict[str, Dict[str, int] | float]:
summary = _empty_summary()
for key, pattern in SUMMARY_PATTERNS.items():
match = pattern.search(text)
if match:
summary[key]["passed"] = int(match.group(1))
summary[key]["failed"] = int(match.group(2))
summary[key]["errors"] = int(match.group(3) or 0)
summary[key]["skipped"] = int(match.group(4))
duration_match = DURATION_PATTERN.search(text)
if duration_match:
minutes = duration_match.group(1)
seconds = float(duration_match.group(2))
if minutes:
seconds += int(minutes) * 60
summary["duration"] = seconds
return summary
def _run_features_inprocess(feature_paths, behave_args):
"""Run *all* feature_paths in a single behave Runner invocation.
Returns ``(failed: bool, summary: dict)``.
"""
runner = _make_runner(feature_paths, behave_args)
failed = runner.run()
summary = _extract_summary(runner)
return failed, summary
def _worker_run_features(payload):
"""Entry point for multiprocessing workers.
def _merge_summaries(
summaries: List[Dict[str, Dict[str, int] | float]],
) -> Dict[str, Dict[str, int] | float]:
total = _empty_summary()
for summary in summaries:
for bucket in ("features", "scenarios", "steps"):
for field in ("passed", "failed", "errors", "skipped"):
total[bucket][field] += summary.get(bucket, {}).get(field, 0)
total["duration"] += float(summary.get("duration", 0.0))
return total
Runs a chunk of feature files in a forked child process. Heavy
Python modules (cleveragents, behave, SQLAlchemy, ...) are already
loaded in the parent and shared via copy-on-write after ``fork``.
``load_step_definitions()`` and ``load_hooks()`` still execute
inside each worker (they ``exec()`` the step .py files and
environment.py), but every ``import`` they trigger is a cache hit.
"""
feature_paths, behave_args = payload
stdout_buf = io.StringIO()
stderr_buf = io.StringIO()
runner = _make_runner(feature_paths, behave_args)
with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
failed = runner.run()
summary = _extract_summary(runner)
return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary
def _format_duration(seconds: float) -> str:
minutes = int(seconds // 60)
remainder = seconds % 60
if minutes:
return f"{minutes}m {remainder:.3f}s"
return f"{remainder:.3f}s"
# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------
def _print_overall_summary(total: Dict[str, Dict[str, int] | float]) -> None:
print("\\nOverall summary:")
for bucket in ("features", "scenarios", "steps"):
b = total[bucket]
print(
f"{b['passed']} {bucket} passed, {b['failed']} failed, "
f"{b['errors']} errored, {b['skipped']} skipped"
)
if total["duration"]:
print(f"Took {_format_duration(float(total['duration']))} (sum across workers)")
def _run_feature(
base_args: List[str], feature: str
) -> Tuple[int, str, str, str, Dict[str, Dict[str, int] | float]]:
args = list(base_args)
if "__SLIPCOVER_OUT__" in args:
output_dir = os.environ.get("SLIPCOVER_OUTPUT_DIR", "build")
output_name = f".slipcover.{uuid.uuid4().hex}.json"
output_path = os.path.join(output_dir, output_name)
args = [output_path if arg == "__SLIPCOVER_OUT__" else arg for arg in args]
proc = subprocess.run([*args, feature], capture_output=True, text=True)
combined_output = (proc.stdout or "") + "\\n" + (proc.stderr or "")
summary = _parse_summary(combined_output)
return proc.returncode, feature, proc.stdout or "", proc.stderr or "", summary
def _build_base_args(other_args: List[str]) -> List[str]:
if os.environ.get("BEHAVE_PARALLEL_COVERAGE"):
source = os.environ.get("SLIPCOVER_SOURCE")
omit = os.environ.get("SLIPCOVER_OMIT")
output_path = "__SLIPCOVER_OUT__"
base_args = [
sys.executable,
"-m",
"slipcover",
"--json",
"--out",
output_path,
]
if source:
base_args.extend(["--source", source])
if omit:
base_args.extend(["--omit", omit])
base_args.extend(["-m", "behave", *other_args])
return base_args
return [sys.executable, "-m", "behave", *other_args]
def _iter_features(paths: Iterable[str]) -> List[str]:
collected: List[str] = []
for path in paths:
p = Path(path)
if p.is_dir():
collected.extend(str(fp) for fp in p.rglob("*.feature"))
else:
collected.append(str(p))
return collected
def main(argv: List[str] | None = None) -> None:
def main(argv=None):
argv = list(sys.argv[1:] if argv is None else argv)
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument("--processes", "-j", type=int, default=None)
known, remaining = parser.parse_known_args(argv)
processes = known.processes or os.cpu_count() or 1
feature_args, other_args = _extract_features_and_args(remaining)
# Pass-through --help / --version to behave
if any(flag in remaining for flag in ("-h", "--help", "--version")):
import subprocess
code = subprocess.run(
[sys.executable, "-m", "behave", *other_args, *feature_args]
).returncode
@@ -282,61 +365,60 @@ def main(argv: List[str] | None = None) -> None:
print("No feature files found", file=sys.stderr)
sys.exit(0)
base_args = _build_base_args(other_args)
coverage_mode = bool(os.environ.get("BEHAVE_PARALLEL_COVERAGE"))
summaries: List[Dict[str, Dict[str, int] | float]] = []
failures: List[Tuple[int, str]] = []
with concurrent.futures.ThreadPoolExecutor(max_workers=processes) as executor:
futures = [
executor.submit(_run_feature, base_args, feature)
for feature in feature_paths
start = time.monotonic()
if processes <= 1 or coverage_mode or len(feature_paths) == 1:
# ---- sequential in-process mode ----
failed, total = _run_features_inprocess(feature_paths, other_args)
else:
# ---- parallel in-process mode (multiprocessing fork) ----
# Pre-import heavy modules so forked children get them for free.
try:
import cleveragents # noqa: F401
except ImportError:
pass
try:
import behave # noqa: F401
except ImportError:
pass
# Split features into roughly equal chunks.
chunk_size = max(1, (len(feature_paths) + processes - 1) // processes)
chunks = [
feature_paths[i : i + chunk_size]
for i in range(0, len(feature_paths), chunk_size)
]
for fut in concurrent.futures.as_completed(futures):
code, feature, stdout, stderr, summary = fut.result()
ctx = multiprocessing.get_context("fork")
with ctx.Pool(processes=min(processes, len(chunks))) as pool:
results = pool.map(
_worker_run_features,
[(chunk, other_args) for chunk in chunks],
)
failed = False
summaries = []
for worker_failed, stdout, stderr, summary in results:
if stdout:
print(f"=== Output for {feature} ===")
print(stdout, end="")
if stderr:
print(f"=== Stderr for {feature} ===", file=sys.stderr)
print(stderr, file=sys.stderr, end="")
print(stderr, end="", file=sys.stderr)
failed = failed or worker_failed
summaries.append(summary)
if code:
failures.append((code, feature))
total = _merge_summaries(summaries)
total = _merge_summaries(summaries)
_print_overall_summary(total)
wall = time.monotonic() - start
_print_overall_summary(total, wall_seconds=wall)
if failures:
for code, feature in failures:
print(f"behave failed for {feature} with code {code}", file=sys.stderr)
if failed or _has_failures(total):
sys.exit(1)
if __name__ == "__main__":
main()
"""
)
setup_path = source_dir / "setup.py"
setup_path.write_text(
"""
from setuptools import find_packages, setup
setup(
name="behave-parallel",
version="1.2.4a1",
packages=find_packages(),
include_package_data=True,
install_requires=["behave>=1.2.6"],
entry_points={
"console_scripts": ["behave-parallel=behave_parallel.cli:main"],
},
)
"""
)
session.install("setuptools", "wheel")
session.install(str(source_dir))
'''
# =============================================================================
@@ -526,10 +608,10 @@ COVERAGE_THRESHOLD = 97
def coverage_report(session: nox.Session):
"""Generate coverage report from Behave tests.
Runs behave tests in parallel via behave-parallel (same approach as
unit_tests) with each worker collecting slipcover JSON coverage data.
After all workers finish, the data files are merged with
``slipcover --merge``.
Runs all behave features in a single process under slipcover.
The in-process behave-parallel runner avoids subprocess overhead,
so a single slipcover invocation collects coverage for the entire
suite -- no per-worker files or merge step required.
Coverage threshold is enforced at >=97%.
@@ -565,56 +647,51 @@ def coverage_report(session: nox.Session):
"src/cleveragents/discovery/*",
]
)
session.env["SLIPCOVER_SOURCE"] = source_paths
session.env["SLIPCOVER_OMIT"] = omit_patterns
session.env["SLIPCOVER_OUTPUT_DIR"] = "build"
# Tell behave-parallel workers to run under slipcover
# Force sequential mode so slipcover wraps the entire process.
session.env["BEHAVE_PARALLEL_COVERAGE"] = "1"
# Clean up any existing slipcover data
for path in Path("build").glob(".slipcover.*.json"):
path.unlink()
# Run behave tests in parallel; each worker collects coverage separately
# Build behave-parallel args (sequential for coverage).
behave_cmd = session.bin + "/behave-parallel"
parallel_args = _behave_parallel_args(session.posargs)
if session.posargs and session.posargs[0].endswith(".feature"):
args = [
behave_args = [
behave_cmd,
"-q",
"--tags=-discovery",
"--no-capture",
*parallel_args,
*session.posargs,
]
else:
args = [
behave_args = [
behave_cmd,
"-q",
"--tags=-discovery",
"--no-capture",
*parallel_args,
"features/",
*session.posargs,
]
session.run(*args)
data_files = sorted(Path("build").glob(".slipcover.*.json"))
if not data_files:
session.error("No slipcover data files found in build/")
# Merge per-worker coverage data into a single JSON report
# Wrap the entire behave-parallel run under slipcover.
# A single process produces a single JSON output file directly.
# Allow exit code 1 (test failures) — coverage data is still produced.
session.run(
"python",
"-m",
"slipcover",
"--merge",
*[str(path) for path in data_files],
"--json",
"--out",
"build/coverage.json",
"--source",
source_paths,
"--omit",
omit_patterns,
"--",
*behave_args,
success_codes=[0, 1],
)
# Generate XML report for CI