tune_spsc.py

1#!/usr/bin/env python32"""Search measured configurations independently for throughput and write p50/p99/p99.9."""3import argparse4import csv5import hashlib6import itertools7import json8import math9import os10from pathlib import Path11import platform12import random13import statistics14import subprocess15import sys16import time1718TRANSPORTS = ("two_buf_queue", "rigtorp", "readerwriterqueue", "atomic_queue")19OBJECTIVES = ("throughput", "p50", "p99", "p999")20PUBLISH_COUNTS = tuple(1 << n for n in range(11)) + (2**32-1,)21FROZEN = {"transport", "mode", "messages", "warmup_messages", "payload_bytes", "callback_iterations",22          "verify_payload", "producer_interval_ns", "timer", "sample_every", "max_samples",23          "timer_calibration_samples", "random_seed", "timeout_seconds", "buffer_budget_bytes", "call_batch_size", "latency_batch_size",24          "round_storage_bytes", "producer_qos", "producer_cpu", "consumer_cpu", "prefault_producer_buffers"}25FROZEN.add("measure_callback_time")26FROZEN.update(("prepared_input", "variable_payload", "prepared_input_limit_bytes"))27FROZEN.update(("is_scan_current", "fixed_round_boundaries"))28SPSC_KEYS = {"readerwriterqueue_copy_after_zero", "rigtorp_write_prefetch", "spsc_capacity", "spsc_idle_spins", "spsc_idle_sleep_ns", "retry_yield_every", "producer_cpu", "consumer_cpu"}29ALIASES = {"block_bytes": ("source_block_size", "output_block_size"),30           "cache_block_bytes": ("source_block_size", "output_block_size", "source_pool_min", "source_pool_max",31                                 "output_pool_min", "output_pool_max", "source_queue_capacity", "output_queue_capacity"),32           "source_pool_blocks": ("source_pool_min", "source_pool_max"),33           "output_pool_blocks": ("output_pool_min", "output_pool_max")}343536def validate_calibration(settings):37    for key in ("target_ratio", "min_ratio", "max_ratio"):38        value = settings[key]39        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:40            raise ValueError("invalid callback ratio: " + key)41    if not settings["min_ratio"] <= settings["target_ratio"] <= settings["max_ratio"]:42        raise ValueError("callback target must be within min/max ratio")43    for key in ("repetitions", "messages", "max_iterations", "max_candidates"):44        if isinstance(settings[key], bool) or not isinstance(settings[key], int) or settings[key] < 1:45            raise ValueError("invalid callback calibration: " + key)46    if isinstance(settings["min_rounds"],bool) or not isinstance(settings["min_rounds"],int) or settings["min_rounds"]<0:47        raise ValueError("min_rounds must be nonnegative; zero keeps the requested calibration count")48    if settings["max_iterations"] > 2**32-1 or settings["reference_publish_count"] not in PUBLISH_COUNTS:49        raise ValueError("callback loop/publish count exceeds supported range")505152def choose_callback_iterations(probe, settings):53    """Positive integer loop search; never pretend an unreachable target was reached."""54    seen = {}55    def evaluate(value):56        if value not in seen:57            seen[value] = probe(value)58            if not math.isfinite(seen[value]) or seen[value] <= 0:59                raise ValueError("invalid callback ratio observation")60        return seen[value]61    lower, upper = settings["min_ratio"], settings["max_ratio"]62    low = high = 163    ratio = evaluate(1)64    while ratio < lower and high < settings["max_iterations"] and len(seen) < settings["max_candidates"]:65        low, high = high, min(2*high, settings["max_iterations"])66        ratio = evaluate(high)67    while not lower <= ratio <= upper and high-low > 1 and len(seen) < settings["max_candidates"]:68        middle = (low+high)//269        ratio = evaluate(middle)70        if ratio < lower:71            low = middle72        else:73            high = middle74    matching = [n for n, ratio in seen.items() if lower <= ratio <= upper]75    candidates = matching or list(seen)76    return min(candidates, key=lambda n: (abs(seen[n]-settings["target_ratio"]), n))777879def effective_config(base, changes):80    result = dict(base)81    for key, value in changes.items():82        if key == "cache_block_bytes":83            if isinstance(value,bool) or not isinstance(value,int) or value<1:84                raise ValueError("cache_block_bytes must be positive")85            for prefix in ("source","output"):86                block_key="source_block_size" if prefix=="source" else "output_block_size"87                old_blocks=base[prefix+"_pool_max"]88                budget=base[block_key]*old_blocks89                if base[prefix+"_pool_min"]!=old_blocks or budget%value:90                    raise ValueError("cache_block_bytes requires a fixed pool and an exactly divisible byte budget")91                result[block_key]=value92                result[prefix+"_pool_min"]=result[prefix+"_pool_max"]=budget//value93                result[prefix+"_queue_capacity"]=max(2,base[prefix+"_queue_capacity"]*(budget//value)//old_blocks)94            continue95        for target in ALIASES.get(key, (key,)):96            result[target] = value97    return result9899100def expected_round_count(config, result):101    """Validate natural publication boundaries before accepting a measurement."""102    record_bytes = result["source_record_bytes"]103    alignment = result["publication_alignment"]104    warmup = result["effective_warmup_messages"]105    if (config.get("fixed_round_boundaries", False) or106        not config.get("is_scan_current", True) or config["publish_count"] == 2**32-1):107        limit = config["round_storage_bytes"]//config["payload_bytes"]108        if (record_bytes != config["payload_bytes"] or alignment != 1 or limit < 1 or109            warmup != max(1, config["warmup_messages"]) or110            result["round_message_limit"] != limit or result["round_message_minimum"] != limit):111            raise ValueError("invalid completed-block publication metadata")112        return (config["messages"]+limit-1)//limit113    valid_alignments = {n - n % config["publish_count"] for n in PUBLISH_COUNTS if n >= config["publish_count"]}114    if record_bytes != config["payload_bytes"] or alignment not in valid_alignments:115        raise ValueError("invalid publication metadata")116    records = config["source_block_size"]//record_bytes117    requested = max(1, config["warmup_messages"])118    position = (requested-1)%records+1119    rounded = (position+alignment-1)//alignment*alignment120    expected_warmup = requested-position+(rounded if rounded<=records else records+alignment)121    limit = config["round_storage_bytes"]//config["payload_bytes"]122    minimum = limit-2*(alignment-1)123    if (config["publish_count"]>alignment or records<alignment or minimum<1 or124        warmup!=expected_warmup or result["round_message_limit"]!=limit or125        result["round_message_minimum"]!=minimum):126        raise ValueError("invalid publication boundary/warmup")127    sent, rounds = 0, 0128    while sent<config["messages"]:129        if config["messages"]-sent<=limit:130            end = config["messages"]131        else:132            desired = warmup+sent+limit133            position = (desired-1)%records+1134            block_begin = desired-position135            published = (block_begin+position//alignment*alignment if position>=alignment136                         else block_begin-records%alignment)137            end = published-warmup138        if end<=sent or end-sent>limit or (end<config["messages"] and end-sent<minimum):139            raise ValueError("invalid round progress")140        sent, rounds = end, rounds+1141    return rounds142143144def candidate_id(config):145    data = {k: v for k, v in config.items() if k not in {"mode", "messages", "random_seed", "verify_payload"}}146    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()[:12]147148149def validate_space(space, defaults, transport):150    assigned = set()151    for key, values in space.items():152        if key in FROZEN or key not in defaults and key not in ALIASES:153            raise ValueError("Not a tunable queue parameter: " + key)154        if transport != "two_buf_queue" and key not in SPSC_KEYS:155            raise ValueError("Unused SPSC tuning parameter: " + key)156        if transport == "two_buf_queue" and key.startswith("spsc_"):157            raise ValueError("Unused TwoBufQueue tuning parameter: " + key)158        if key == "rigtorp_write_prefetch" and transport != "rigtorp":159            raise ValueError("Unused rigtorp tuning parameter: " + key)160        if key == "readerwriterqueue_copy_after_zero" and transport != "readerwriterqueue":161            raise ValueError("Unused ReaderWriterQueue tuning parameter: " + key)162        if not isinstance(values, list) or not values:163            raise ValueError("Search values must be nonempty lists: " + key)164        targets = set(ALIASES.get(key, (key,)))165        if assigned & targets:166            raise ValueError("Overlapping search fields/aliases: " + key)167        assigned.update(targets)168169170def validate_length_reference(changes, defaults):171    # A reference configuration can specify block sizes and native queue capacity for each payload length.172    # Validate them separately to preserve ownership constraints and detect overlapping block aliases.173    validate_space({k: [v] for k, v in changes.items() if not k.startswith("spsc_") and k not in {"rigtorp_write_prefetch","readerwriterqueue_copy_after_zero"}},174                   defaults, "two_buf_queue")175    validate_space({k: [v] for k, v in changes.items() if k.startswith("spsc_")},176                   defaults, "readerwriterqueue")177    validate_space({k: [v] for k, v in changes.items() if k == "rigtorp_write_prefetch"},178                   defaults, "rigtorp")179    validate_space({k: [v] for k, v in changes.items() if k == "readerwriterqueue_copy_after_zero"},180                   defaults, "readerwriterqueue")181182183def initial_candidates(base, space, strategy, limit, rng):184    result, seen = [], set()185    def add(config):186        key = candidate_id(config)187        if key not in seen:188            seen.add(key)189            result.append(config)190    add(base)191    if strategy == "grid":192        count = math.prod(len(v) for v in space.values())193        if limit and count > limit:194            # Sample indices without materializing an enormous Cartesian product.195            indices = sorted(rng.sample(range(count), min(limit - 1, count)))196            for index in indices:197                changes = {}198                for key, values in reversed(list(space.items())):199                    index, remainder = divmod(index, len(values))200                    changes[key] = values[remainder]201                add(effective_config(base, changes))202        else:203            for values in itertools.product(*space.values()):204                add(effective_config(base, dict(zip(space, values))))205    else:206        # First visit each dimension around the same baseline, before spending207        # the remaining budget on combinations of metric-specific incumbents.208        for key, values in space.items():209            for value in values:210                add(effective_config(base, {key: value}))211    return result[:limit] if limit else result212213214def aggregate(rows):215    throughput = [r["result"] for r in rows if r["result"]["config"]["mode"] == "throughput"]216    latency = [r["result"] for r in rows if r["result"]["config"]["mode"] == "write_latency"]217    if not throughput or not latency:218        return None219    kinds = {r.get("latency_kind", "single_call") for r in latency}220    if len(kinds) != 1:221        raise ValueError("cannot mix single-call and batch-average samples")222    kind = kinds.pop()223    sample_key = "batch_average_write_ns" if kind == "batch_average" else "successful_write_ns"224    divisor = latency[0]["config"].get("latency_batch_size", 1) if kind == "batch_average" else 1225    if any((r["config"].get("latency_batch_size", 1) if kind == "batch_average" else 1) != divisor for r in latency):226        raise ValueError("cannot mix different latency batch sizes")227    values = [r["m_messages_per_second"] for r in throughput]228    result = {"throughput": statistics.fmean(values), "throughput_min": min(values), "throughput_max": max(values),229              "cpu_cores": statistics.fmean(r["cpu_cores"] for r in throughput),230              "peak_rss_mib": max(r["peak_rss_mib"] for r in throughput),231              "failed_write_calls": statistics.fmean(r["failed_write_calls"] for r in throughput),232              "actual_spsc_capacity": throughput[0]["actual_spsc_capacity"],233              "nominal_buffer_bytes": throughput[0]["nominal_buffer_bytes"],234              "latency_samples_min": min(r["latency_samples"] for r in latency),235              "timer_quantum_ns": max(r["timer_quantum_ns"] for r in latency),236              "latency_kind": kind, "latency_batch_size": divisor,237              "latency_quantum_ns": max(r.get("latency_quantum_ns", r["timer_quantum_ns"]) for r in latency),238              "latency_overhead_p50_ns": statistics.fmean(r["timer_overhead_ns"]["p50"]/divisor for r in latency),239              "timer_overhead_p50_ns": statistics.fmean(r["timer_overhead_ns"]["p50"] for r in latency),240              "resolution_limited": any(r["resolution_limited"] for r in latency),241              "p999_sample_count_warning": any(r["p999_sample_count_warning"] for r in latency)}242    if any(not r[sample_key] for r in latency):243        return None244    for key in OBJECTIVES[1:]:245        result[key] = statistics.fmean(r[sample_key][key] for r in latency)246        result[key+"_status"] = "unresolved" if any(247            r[sample_key][key] <= 0 or r[sample_key][key] <248            max(5*r.get("latency_quantum_ns", r["timer_quantum_ns"]),249                2*r["timer_overhead_ns"]["p50"]/divisor) for r in latency) else "observed"250    if all(r["config"].get("measure_callback_time", False) for r in throughput):251        for key in ("producer_seconds", "callback_processing_seconds", "callback_to_producer_ratio", "drain_seconds"):252            result[key] = statistics.fmean(r[key] for r in throughput)253        result["callback_ratio_min"] = min(r["callback_to_producer_ratio"] for r in throughput)254        result["callback_ratio_max"] = max(r["callback_to_producer_ratio"] for r in throughput)255    if all("backpressured_messages" in r for r in throughput):256        for key in ("backpressured_messages", "producer_retry_seconds", "producer_retry_max_seconds"):257            result[key] = statistics.fmean(r[key] for r in throughput)258        result["backpressured_message_fraction"] = statistics.fmean(r["backpressured_messages"]/r["config"]["messages"] for r in throughput)259        result["failed_attempt_fraction"] = statistics.fmean(r["failed_write_calls"]/(r["config"]["messages"]+r["failed_write_calls"]) for r in throughput)260        pressure = [r.get("library_backpressure") for r in throughput]261        result["library_backpressure"] = None if any(p is None for p in pressure) else {262            key: statistics.fmean(p[key] for p in pressure) for key in (263                "writer_pressure_events", "writer_rejected_attempts", "writer_blocked_events", "writer_blocked_ticks",264                "copy_memory_full_events", "copy_backpressured_frames")}265    return result266267268def latency_status(metrics, objective):269    # Conservative reporting guard, not a confidence interval or an overhead correction.270    limit = max(5*metrics.get("latency_quantum_ns", metrics.get("timer_quantum_ns", 0)),271                2*metrics.get("latency_overhead_p50_ns", metrics.get("timer_overhead_p50_ns", 0)))272    return metrics.get(objective+"_status", "unresolved" if metrics[objective] <= 0 or metrics[objective] < limit else "observed")273274275def select_best(candidates, objective):276    if objective not in OBJECTIVES:277        raise ValueError("Unknown objective")278    if not candidates:279        raise ValueError("No complete successful measurements")280    if objective == "throughput":281        return max(candidates, key=lambda c: c["metrics"][objective])282    unresolved = [c for c in candidates if latency_status(c["metrics"], objective) == "unresolved"]283    if unresolved:284        # Values below the measurement floor cannot establish a latency winner.285        # Continue the search with a throughput-selected representative of this group.286        return max(unresolved, key=lambda c: c["metrics"]["throughput"])287    return min(candidates, key=lambda c: (c["metrics"][objective], -c["metrics"]["throughput"]))288289290def objective_result(candidates, objective, length, iterations):291    best = select_best(candidates, objective)292    unresolved = objective != "throughput" and latency_status(best["metrics"], objective) == "unresolved"293    result = dict(best, objective=objective, payload_bytes=length, callback_iterations=iterations,294                  ranking_status="unresolved" if unresolved else "observed",295                  objective_value=None if unresolved else best["metrics"][objective],296                  selection_basis="throughput_representative" if unresolved else objective)297    if unresolved:298        result["resolution_limited_candidates"] = [c["candidate_id"] for c in candidates299            if latency_status(c["metrics"], objective) == "unresolved"]300    elif objective != "throughput":301        floor = best["metrics"][objective]302        result["within_one_timer_tick"] = [c["candidate_id"] for c in candidates303            if c["metrics"][objective] <= floor+max(c["metrics"]["latency_quantum_ns"], best["metrics"]["latency_quantum_ns"])]304    return result305306307class ParameterSearch:308    def __init__(self, executable, profile, output, allow_debug=False):309        self.executable = Path(executable).resolve()310        self.profile = profile311        if set(profile) - {"payload_bytes", "callback_iterations", "base", "search", "spaces", "transports", "callback_calibration", "base_by_payload"}:312            raise ValueError("unknown profile field")313        self.output = Path(output)314        if self.output.exists() and any(self.output.iterdir()):315            raise ValueError("output directory must be empty; existing measurements are preserved")316        self.output.mkdir(parents=True, exist_ok=True)317        self.defaults = json.loads(subprocess.check_output([str(self.executable), "--defaults"], text=True))318        self.build_info = json.loads(subprocess.check_output([str(self.executable), "--build-info"], text=True))319        if not allow_debug and (not self.build_info["optimized"] or self.build_info["sanitized"]):320            raise ValueError("Performance search requires an optimized non-sanitized binary")321        self.settings = {"strategy": "coordinate", "max_candidates_per_transport": 32, "passes": 2,322                         "screen_messages": 1000000, "final_messages": 20000000,323                         "screen_repetitions": 3, "final_repetitions": 3,324                         "screen_min_rounds": 0, "final_min_rounds": 0, "validation_messages": 10000,325                         "batch_reference": True}326        self.settings.update(profile.get("search", {}))327        if set(self.settings) - {"strategy", "max_candidates_per_transport", "passes", "screen_messages",328            "final_messages", "screen_repetitions", "final_repetitions", "validation_messages", "batch_reference",329            "screen_min_rounds", "final_min_rounds"}:330            raise ValueError("unknown search setting")331        if self.settings["strategy"] not in ("coordinate", "grid"):332            raise ValueError("strategy must be coordinate or grid")333        for key in ("screen_messages", "final_messages", "screen_repetitions", "final_repetitions", "validation_messages", "passes"):334            if not isinstance(self.settings[key], int) or self.settings[key] < 1:335                raise ValueError(key + " must be positive")336        for key in ("screen_min_rounds", "final_min_rounds"):337            if not isinstance(self.settings[key], int) or self.settings[key] < 0:338                raise ValueError(key + " must be nonnegative")339        if not isinstance(self.settings["max_candidates_per_transport"], int) or self.settings["max_candidates_per_transport"] < 0:340            raise ValueError("candidate limit must be nonnegative (0 means unlimited)")341        if not isinstance(self.settings["batch_reference"], bool):342            raise ValueError("batch_reference must be boolean")343        self.transports = profile.get("transports", list(TRANSPORTS))344        if not self.transports or len(set(self.transports)) != len(self.transports) or any(t not in TRANSPORTS for t in self.transports):345            raise ValueError("invalid transport list")346        self.base = dict(self.defaults)347        if set(profile.get("base", {})) - set(self.defaults):348            raise ValueError("unknown base parameter")349        self.base.update(profile.get("base", {}))350        self.base_by_payload = profile.get("base_by_payload", {})351        if not isinstance(self.base_by_payload,dict) or any(str(k) not in ("8","16","32","64") for k in self.base_by_payload):352            raise ValueError("invalid base_by_payload lengths")353        for changes in self.base_by_payload.values():354            if not isinstance(changes,dict): raise ValueError("base_by_payload entries must be mappings")355            validate_length_reference(changes,self.defaults)356        self.calibration = profile.get("callback_calibration")357        self.calibration_results = []358        if self.calibration is not None:359            defaults = dict(target_ratio=2.5, min_ratio=2.0, max_ratio=3.0, repetitions=3,360                            messages=1000000, min_rounds=2, max_iterations=1024,361                            max_candidates=10, reference_publish_count=16)362            if not isinstance(self.calibration, dict) or set(self.calibration) - set(defaults):363                raise ValueError("invalid callback calibration settings")364            defaults.update(self.calibration)365            self.calibration = defaults366            validate_calibration(defaults)367            if "callback_iterations" in profile or not self.base.get("measure_callback_time") or not self.base.get("round_storage_bytes"):368                raise ValueError("calibration requires callback timing/rounds and replaces callback_iterations")369        if any(k in profile.get("base", {}) for k in ("payload_bytes", "messages", "mode", "transport", "callback_iterations")):370            raise ValueError("put payload_bytes / callback_iterations at profile top level; counts belong in search")371        self.spaces = profile.get("spaces", {})372        if set(self.spaces) - set(TRANSPORTS) - {"spsc"}:373            raise ValueError("unknown search-space transport")374        for transport in self.transports:375            validate_space(self.space(transport), self.defaults, transport)376        self.rng = random.Random(self.base["random_seed"])377        self.rows, self.failures, self.checksums = [], [], {}378        self.raw = (self.output / "raw.jsonl").open("w", encoding="utf-8")379        self.errors = (self.output / "failures.jsonl").open("w", encoding="utf-8")380        self.runs = 0381        self.started = time.monotonic()382383    def length_base(self, length):384        return effective_config(self.base,self.base_by_payload.get(str(length),{}))385386    def space(self, transport):387        return self.spaces.get(transport, self.spaces.get("spsc", {}) if transport != "two_buf_queue" else {})388389    def run_one(self, config, stage, trial, identity):390        command = [str(self.executable)]391        for key, value in config.items():392            command += ["--" + key, str(value).lower() if isinstance(value, bool) else str(value)]393        try:394            process = subprocess.run(command, capture_output=True, text=True,395                                     timeout=config["timeout_seconds"] + 10, check=True)396            result = json.loads(process.stdout)397            if result["consumed"] != config["messages"]:398                raise ValueError("wrong consumed message count")399            if config.get("round_storage_bytes", 0):400                if (result["timing_scope"] != "producer_rounds" or401                    result["completed_rounds"] != expected_round_count(config, result) or402                    result["seconds"] != result["producer_seconds"]):403                    raise ValueError("invalid producer round timing/completion")404            if config.get("producer_qos") == "user_interactive" and not result["producer_qos_applied"]:405                raise ValueError("producer high QoS was not applied")406            if config.get("measure_callback_time") and (result["timed_callback_messages"] != config["messages"] or407                                                       result["callback_processing_seconds"] <= 0):408                raise ValueError("callback timing count/elapsed validation failed")409            signature = (stage, config["payload_bytes"], config["callback_iterations"], config["messages"],410                         result["effective_warmup_messages"])411            checksum = result["callback_work_checksum"]412            if signature in self.checksums and self.checksums[signature] != checksum:413                raise ValueError("callback work differs between configurations/libraries")414            self.checksums[signature] = checksum415            row = {"stage": stage, "trial": trial, "candidate_id": identity, "command": command, "result": result}416            self.rows.append(row)417            self.raw.write(json.dumps(row, ensure_ascii=False) + "\n")418            self.raw.flush()419            return row420        except (subprocess.SubprocessError, ValueError) as error:421            row = {"stage": stage, "trial": trial, "candidate_id": identity, "command": command,422                   "error": str(error), "stderr": (getattr(error, "stderr", "") or "")[-8192:]}423            self.failures.append(row)424            self.errors.write(json.dumps(row, ensure_ascii=False) + "\n")425            self.errors.flush()426            return None427        finally:428            self.runs += 1429            if self.runs % 20 == 0:430                print("{} runs, {:.1f}s elapsed".format(self.runs, time.monotonic()-self.started), file=sys.stderr, flush=True)431432    def measurement_count(self, config, stage):433        count = self.settings["screen_messages" if stage == "screen" else "final_messages"]434        if config.get("round_storage_bytes", 0):435            count = max(count, (config["round_storage_bytes"]//config["payload_bytes"]) *436                        self.settings.get(stage+"_min_rounds", 0))437        return count438439    def evaluate(self, configs, stage):440        count_key = "screen_messages" if stage == "screen" else "final_messages"441        repetition_key = "screen_repetitions" if stage == "screen" else "final_repetitions"442        valid, jobs = [], []443        if stage == "screen":444            validations = list(configs)445            self.rng.shuffle(validations)446            for config in validations:447                checked = dict(config, mode="throughput", verify_payload=True,448                               messages=self.settings["validation_messages"],449                               warmup_messages=min(config["warmup_messages"], self.settings["validation_messages"]))450                if self.run_one(checked, "validation", 0, candidate_id(config)):451                    valid.append(config)452        else:453            valid = list(configs)454        for trial in range(self.settings[repetition_key]):455            round_jobs = []456            for config in valid:457                for mode in ("throughput", "write_latency"):458                    checked = dict(config, mode=mode, messages=self.measurement_count(config, stage),459                                   random_seed=config["random_seed"]+trial)460                    round_jobs.append((checked, trial, candidate_id(config)))461            self.rng.shuffle(round_jobs)462            jobs.extend(round_jobs)463        start_index = len(self.rows)464        for checked, trial, identity in jobs:465            self.run_one(checked, stage, trial, identity)466        result = []467        for config in valid:468            identity = candidate_id(config)469            rows = [r for r in self.rows[start_index:] if r["candidate_id"] == identity and r["stage"] == stage]470            # A candidate must finish every scheduled run; do not select partial successes.471            if len(rows) != self.settings[repetition_key]*2:472                continue473            metrics = aggregate(rows)474            if metrics:475                result.append({"candidate_id": identity, "parameters": dict(config, messages=self.measurement_count(config, stage)),476                               "stage": stage, "metrics": metrics})477        return result478479    def calibrate_callback(self, length):480        settings = self.calibration481        observations = {}482        def probe(iterations):483            length_base = self.length_base(length)484            reference_publish = (length_base["publish_count"] if "publish_count" in self.base_by_payload.get(str(length),{})485                                 else settings["reference_publish_count"])486            config = dict(length_base, transport="two_buf_queue", payload_bytes=length,487                          callback_iterations=iterations, publish_count=reference_publish,488                          mode="throughput", verify_payload=False)489            config["messages"] = max(settings["messages"], settings["min_rounds"] *490                                     (config["round_storage_bytes"]//length))491            measured = []492            for trial in range(settings["repetitions"]):493                row = self.run_one(dict(config, random_seed=config["random_seed"]+trial),494                                   "callback_calibration", trial, candidate_id(config))495                if row is None:496                    raise RuntimeError("callback calibration run failed; raw failure retained")497                measured.append(row["result"])498            observation = dict(iterations=iterations,499                ratio=statistics.fmean(r["callback_to_producer_ratio"] for r in measured),500                producer_seconds=statistics.fmean(r["producer_seconds"] for r in measured),501                callback_processing_seconds=statistics.fmean(r["callback_processing_seconds"] for r in measured),502                ratios=[r["callback_to_producer_ratio"] for r in measured])503            observations[iterations] = observation504            print("callback calibration {} B: {} loops, {:.2f}x".format(length, iterations, observation["ratio"]),505                  file=sys.stderr, flush=True)506            return observation["ratio"]507        selected = choose_callback_iterations(probe, settings)508        record = dict(payload_bytes=length, reference_transport="two_buf_queue",509                      reference_publish_count=(self.length_base(length)["publish_count"] if "publish_count" in self.base_by_payload.get(str(length),{}) else settings["reference_publish_count"]),510                      reference_parameters=self.length_base(length),511                      selected_iterations=selected, selected_ratio=observations[selected]["ratio"],512                      target_met=settings["min_ratio"] <= observations[selected]["ratio"] <= settings["max_ratio"],513                      observations=list(observations.values()), settings=settings,514                      timing_scope="round first-to-last callback including dispatch and in-round waits; excluding between-round idle")515        self.calibration_results.append(record)516        (self.output/"callback-calibration.json").write_text(json.dumps(self.calibration_results, indent=2)+"\n")517        return selected518519    def run(self):520        lengths = self.profile.get("payload_bytes", [8, 16, 32, 64])521        loops = self.profile.get("callback_iterations", [0])522        if not lengths or any(n not in (8, 16, 32, 64) for n in lengths):523            raise ValueError("invalid payload sizes")524        if not loops or any(not isinstance(n, int) or n < 0 for n in loops):525            raise ValueError("invalid callback loop values")526        all_candidates, groups, winners = [], [], []527        limit = self.settings["max_candidates_per_transport"]528        workloads = ((length, self.calibrate_callback(length)) for length in lengths) if self.calibration is not None else itertools.product(lengths, loops)529        for length, iterations in workloads:530            configurations, initial, seen = {}, [], set()531            for transport in self.transports:532                base = dict(self.length_base(length), transport=transport, payload_bytes=length, callback_iterations=iterations)533                candidates = initial_candidates(base, self.space(transport), self.settings["strategy"], limit, self.rng)534                configurations[transport] = candidates535                initial.extend(candidates)536                seen.update(candidate_id(c) for c in candidates)537            screened = self.evaluate(initial, "screen")538            if self.settings["strategy"] == "coordinate":539                for unused_pass in range(self.settings["passes"]):540                    additions = []541                    for transport in self.transports:542                        current = [c for c in screened if c["parameters"]["transport"] == transport]543                        if not current:544                            continue545                        parents = {select_best(current, o)["candidate_id"]: select_best(current, o)["parameters"] for o in OBJECTIVES}546                        possibilities = []547                        for parent in parents.values():548                            for key, values in self.space(transport).items():549                                for value in values:550                                    candidate = effective_config(parent, {key: value})551                                    if candidate_id(candidate) not in seen:552                                        possibilities.append(candidate)553                        self.rng.shuffle(possibilities)554                        for candidate in possibilities:555                            if limit and len(configurations[transport]) >= limit:556                                break557                            identity = candidate_id(candidate)558                            if identity in seen:559                                continue560                            seen.add(identity)561                            configurations[transport].append(candidate)562                            additions.append(candidate)563                    if not additions:564                        break565                    screened.extend(self.evaluate(additions, "screen"))566            all_candidates.extend(screened)567            finalists = {}568            for transport in self.transports:569                candidates = [c for c in screened if c["parameters"]["transport"] == transport]570                if not candidates:571                    raise RuntimeError("No valid candidates for {} {} bytes".format(transport, length))572                for objective in OBJECTIVES:573                    best = select_best(candidates, objective)574                    finalists[best["candidate_id"]] = best["parameters"]575            confirmed = self.evaluate(list(finalists.values()), "final")576            all_candidates.extend(confirmed)577            selected_configs = {}578            for transport in self.transports:579                candidates = [c for c in confirmed if c["parameters"]["transport"] == transport]580                for objective in OBJECTIVES:581                    winner = objective_result(candidates, objective, length, iterations)582                    winners.append(winner)583                    selected_configs[winner["candidate_id"]] = winner["parameters"]584                groups.append({"payload_bytes": length, "callback_iterations": iterations, "transport": transport,585                               "tested_candidates": len(configurations[transport]),586                               "successful_screen_candidates": len([c for c in screened if c["parameters"]["transport"] == transport]),587                               "search_space_product": math.prod(len(v) for v in self.space(transport).values()),588                               "finalists": len(candidates)})589            if self.settings["batch_reference"]:590                for identity, config in selected_configs.items():591                    checked = dict(config, mode="call_batch", producer_interval_ns=0, messages=self.settings["final_messages"])592                    self.run_one(checked, "batch_reference", 0, identity)593            self.write_reports(all_candidates, groups, winners)594        self.raw.close()595        self.errors.close()596        return winners597598    def write_reports(self, candidates, groups, winners):599        ranked = [w for w in winners if w["ranking_status"] != "unresolved"]600        unresolved = [w for w in winners if w["ranking_status"] == "unresolved"]601        summary = {"scope": "best measured configurations within the declared search space and budget; not a global optimum",602                   "aggregation": "arithmetic mean of per-run throughput and per-run quantiles; no pooled quantiles",603                   "ranking_policy": "Latency below 5 effective ticks or twice scaled median empty-bracket overhead is unresolved; zero is never a latency winner. For batch averages both tick and overhead are divided by batch size. This guard is not a confidence interval.",604                   "settings": self.settings, "groups": groups, "winners": ranked,605                   "unresolved": unresolved, "failed_runs": len(self.failures)}606        (self.output/"best.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")607        self.write_csv(self.output/"candidates.csv", candidates)608        self.write_csv(self.output/"best.csv", ranked)609        self.write_csv(self.output/"unresolved.csv", unresolved)610        lines = ["# 可调性能测试结果", "", "最优仅指本次候选集合；各指标独立选择，再用独立正式轮次复测。",611                 "低于测量下限的延时指标显示为精度不足，不评选最优；对应行仅保留按吞吐选择的代表配置。",612                 "延时为每组平均写入耗时的分位数：组总耗时 / 组内成功写入次数，包含循环、重试与计时开销，不主动 flush。",613                 "这不是单次 write 调用的 p50/p99/p99.9；预生成负载在计时区之外。组大小见完整参数 latency_batch_size。",614                 "", "| 字节 | 回调循环 | 实现 | 优化目标 | 排名状态 | M msg/s | 组均值 p50 ns | p99 ns | p99.9 ns | CPU 核 | 配置 |",615                 "|---:|---:|---|---|---|---:|---:|---:|---:|---:|---|"]616        for w in winners:617            m = w["metrics"]618            shown = ["精度不足" if latency_status(m, key) == "unresolved" else "{:.3f}".format(m[key]) for key in OBJECTIVES[1:]]619            status = "无法可靠排名；代表配置" if w["ranking_status"] == "unresolved" else "按观测值选择"620            lines.append("| {} | {} | {} | {} | {} | {:.3f} | {} | {} | {} | {:.2f} | {} |".format(621                w["payload_bytes"], w["callback_iterations"], w["parameters"]["transport"], w["objective"],622                status, m["throughput"], *shown, m["cpu_cores"], w["candidate_id"]))623        lines += ["", "各指标为独立运行的算术平均；分位数先逐次计算，再平均。重复次数见 settings。"]624        pressure_winners = [w for w in winners if w["objective"] == "throughput" and625                            "backpressured_messages" in w["metrics"]]626        if pressure_winners:627            lines += ["", "## 吞吐配置的时间比与背压", "",628                      "下表各项均为吞吐正式轮次的算术平均。回调时间为各轮首条入口至末条业务完成的区间之和，含轮内分发、等待和调度；比例大于 1 本身不能证明写端背压。",629                      "写入失败后重试直到成功，故失败次数不等于最终丢失条数。重试计时仅在失败路径启用。", "",630                      "| 字节 | 实现 | 写线程 ms | 回调区间 ms | 回调/写线程 | 受压消息 | 失败尝试 | 重试 ms | Writer压力 | Writer阻塞 | 二层拒绝 |",631                      "|---:|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|"]632            for w in pressure_winners:633                m = w["metrics"]634                ms = lambda key: "{:.3f}".format(m[key]*1000) if key in m else "—"635                ratio = "{:.3f}×".format(m["callback_to_producer_ratio"]) if "callback_to_producer_ratio" in m else "—"636                p = m.get("library_backpressure")637                internal = ["{:.1f}".format(p[key]) if p is not None else "—" for key in638                            ("writer_pressure_events", "writer_blocked_events", "copy_memory_full_events")]639                lines.append("| {} | {} | {} | {} | {} | {:.1f} | {:.1f} | {} | {} | {} | {} |".format(640                    w["payload_bytes"], w["parameters"]["transport"], ms("producer_seconds"),641                    ms("callback_processing_seconds"), ratio, m["backpressured_messages"],642                    m["failed_write_calls"], ms("producer_retry_seconds"), *internal))643            lines += ["", "受压消息指至少一次 try 写入失败的消息；失败尝试含同一条消息的多次重试。TwoBufQueue Block 模式内部等待可能最终成功，不体现在 try 失败数中，另见 Writer阻塞计数及原始 blocked_ticks。",644                      "TwoBufQueue 内部统计扣除预热，覆盖正式发送、排空和停止；二层拒绝按复制片段尝试计数，不能与逐条写失败等同。第三方未提供同口径的内部计数，显示 —；其 try 失败和重试耗时仍可直接统计。"]645        if getattr(self, "calibration_results", None):646            lines += ["回调循环先以 TwoBufQueue 固定 reference_publish_count 校准，各库使用相同次数和计算逻辑。",647                      "回调计时为每轮首条到末条的处理区间，含分发/轮内等待，排除轮间等待；不是逐条纯函数耗时之和。",648                      "比例以 TwoBufQueue 为基准，其他库写入速度不同，不强行调整各自循环次数。校准观测见 callback-calibration.json。"]649        if any(w["parameters"].get("round_storage_bytes", 0) for w in winners):650            lines += ["吞吐按各轮写线程发送耗时之和计算；轮末排空回调和最终 finish 在计时外，不主动 flush。复制和回调仍会并发竞争资源。",651                      "每轮条数上限 = round_storage_bytes / payload_bytes 向下取整；按源块内自然发布位置略微向下调整。相同源配置的四种实现使用相同边界。",652                      "本机高 QoS 如启用仅是调度偏好；producer_cpu=-1 表示未硬绑核。实际状态见 raw.jsonl。"]653        lines += ["", "已排名的最优参数见 best.json 的 winners / best.csv；无法排名的代表配置见 unresolved / unresolved.csv。",654                  "所有候选及原始观测值见 candidates.csv，逐轮命令与数据见 raw.jsonl。",655                  "回调循环 0 表示无模拟业务计算；测试框架仍检查序号和计数。TwoBufQueue 包含第二层复制。",656                  "SPSC 使用原生轮询消费，TwoBufQueue 使用自身自旋/事件策略；CPU 和名义缓冲容量同时报告。"]657        if any(w["metrics"]["resolution_limited"] for w in winners):658            lines += ["", "**计时分辨率提示：部分 write 调用接近计时器粒度或空测量开销。**",659                      "使用保守阈值：任一轮的分位数低于 5 个 tick 或两倍空计时中位数时，不评选该项最优。阈值不是置信区间。",660                      "resolution_limited_candidates 列出无法分辨的候选；原始的 0 ns 观测仍保留供审计。",661                      "本机 p50/p99/p99.9 是组均值的原始观测值，不能解释为单次函数分位数；没有机械扣减计时开销。",662                      "batch_reference 给出批次均值参考，不能替代单次调用分位数。"]663        if any(w["metrics"]["p999_sample_count_warning"] for w in winners):664            lines += ["", "部分正式轮次少于 100,000 个有效样本，p99.9 尾部样本较少；应增加 messages 或减小 latency_batch_size 后复测（组大小改变了统计口径）。"]665        if self.settings["screen_repetitions"] < 2 or self.settings["final_repetitions"] < 3:666            lines += ["", "本次重复次数较少，只适合流程验证或初筛，不能据此宣布稳定性能排名。"]667        if self.failures:668            lines += ["", "失败的配置/轮次保留在 failures.jsonl，不参与最优选择。"]669        (self.output/"SUMMARY.md").write_text("\n".join(lines)+"\n", encoding="utf-8")670671    @staticmethod672    def write_csv(path, candidates):673        rows = []674        for c in candidates:675            row = {"stage": c["stage"], "candidate_id": c["candidate_id"], "objective": c.get("objective", ""),676                   "ranking_status": c.get("ranking_status", ""), "objective_value": c.get("objective_value", ""),677                   "selection_basis": c.get("selection_basis", "")}678            row.update(c["parameters"])679            row.update(c["metrics"])680            rows.append(row)681        if not rows:682            return683        with path.open("w", newline="", encoding="utf-8") as stream:684            writer = csv.DictWriter(stream, fieldnames=list(rows[0]))685            writer.writeheader()686            writer.writerows(rows)687688689def write_environment(search, profile_path):690    root = Path(__file__).resolve().parents[2]691    files = [p for folder in (root/"src", root/"include", root/"benchmarks/tunable") for p in folder.rglob("*") if p.is_file()]692    files += [root/"CMakeLists.txt", Path(__file__), root/"benchmarks/spsc/spsc_dependencies.json", profile_path]693    manifest = {str(p.relative_to(root)) if str(p).startswith(str(root)+os.sep) else str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in files}694    cache = search.executable.parent/"CMakeCache.txt"695    if not cache.exists():696        cache = search.executable.parent.parent/"CMakeCache.txt"697    build_flags = {}698    if cache.exists():699        for line in cache.read_text(encoding="utf-8").splitlines():700            if line.startswith(("CMAKE_CXX_COMPILER:", "CMAKE_CXX_FLAGS", "CMAKE_BUILD_TYPE:",701                                "CMAKE_EXE_LINKER_FLAGS", "CMAKE_OSX_DEPLOYMENT_TARGET:", "CMAKE_GENERATOR:")) and "=" in line:702                key, value = line.split("=", 1)703                build_flags[key.split(":", 1)[0]] = value704    cpu_model = platform.processor()705    if sys.platform == "darwin":706        try:707            cpu_model = subprocess.check_output(["/usr/sbin/sysctl", "-n", "machdep.cpu.brand_string"],708                                               text=True, stderr=subprocess.DEVNULL).strip()709        except (OSError, subprocess.SubprocessError):710            pass711    info = {"platform": platform.platform(), "machine": platform.machine(), "cpu_model": cpu_model,712            "logical_cpus": os.cpu_count(), "cmake_flags": build_flags,713            "python": sys.version, "build": search.build_info, "executable": str(search.executable),714            "executable_sha256": hashlib.sha256(search.executable.read_bytes()).hexdigest(), "profile": search.profile,715            "resolved_search": search.settings, "source_sha256": manifest,716            "dependencies": json.loads((root/"benchmarks/spsc/spsc_dependencies.json").read_text()),717            "note": "Record power policy, CPU model and background load when publishing results; no affinity claimed unless successful."}718    (search.output/"environment.json").write_text(json.dumps(info, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")719720721def main():722    parser = argparse.ArgumentParser(description=__doc__)723    parser.add_argument("executable", type=Path)724    parser.add_argument("--profile", type=Path, default=Path(__file__).resolve().parent.parent/"profiles"/"spsc_quick.json")725    parser.add_argument("--output", type=Path, required=True)726    parser.add_argument("--allow-debug", action="store_true", help="Contract testing only; never publish these as performance results")727    args = parser.parse_args()728    profile = json.loads(args.profile.read_text(encoding="utf-8"))729    if args.output.exists() and any(args.output.iterdir()):730        parser.error("output directory must be empty to preserve previous measurements")731    search = ParameterSearch(args.executable, profile, args.output, args.allow_debug)732    write_environment(search, args.profile.resolve())733    search.run()734    print("Results:", (args.output/"SUMMARY.md").resolve())735736737if __name__ == "__main__":738    main()