Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 10 additions & 3 deletions loadtest/multiprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import asyncio
import random
from concurrent.futures import ProcessPoolExecutor
from time import perf_counter
from uuid import UUID

from loadtest.harness import StrategyReport, violation_magnitude
Expand Down Expand Up @@ -75,17 +76,23 @@ async def multiprocess_sweep(
slices = [ids[p::processes] for p in range(processes)]
loop = asyncio.get_running_loop()
with ProcessPoolExecutor(max_workers=processes) as pool:
# Parent-side wall spans the whole parallel run — from dispatch until
# the last slice returns — so it is the correct denominator for an
# aggregate throughput across overlapping workers (summing per-slice
# walls would double-count concurrent time). The comparative table's
# published numbers still come from the single-process run; this is
# the multi-process path's own end-to-end throughput.
start = perf_counter()
futures = [
loop.run_in_executor(
pool, driven_slice, database_url, sl, f"mp-{p}", concurrency, seed + p
)
for p, sl in enumerate(slices)
]
results = await asyncio.gather(*futures)
wall = perf_counter() - start
samples = [s for slice_samples in results for s in slice_samples]
# Wall here is approximate (post-hoc); the comparative table's throughput
# comes from the single-process run. summarize over the union for tallies.
metrics = summarize("guarded-mp", samples, wall_seconds=0.0)
metrics = summarize("guarded-mp", samples, wall_seconds=wall)
oracle = await run_oracle(postgres_read_uow_factory(engine))
finally:
await engine.dispose()
Expand Down
3 changes: 3 additions & 0 deletions tests/integration/test_load_multiprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,3 +20,6 @@ async def test_multiprocess_guarded_stays_clean(postgres_url: str) -> None:
assert report.oversell == 0
assert report.metrics.count == 48
assert report.metrics.errors == 0
# The multi-process path measures its own end-to-end wall, so it reports a
# real aggregate throughput rather than a degenerate 0.0 (issue #96).
assert report.metrics.throughput > 0.0
Loading