Survey Driver

One driver runs the survey for every backend. It owns the checkpoint cadence, the halt rule and the monitor, and a backend owns the search.

The next chunk of work goes to the device before the monitor runs, so host work in a monitor overlaps device work. The map read and the tree read come first, because the next chunk changes the archive.

Every backend halts on the trailing window of gate_window checkpoints that windowed_rates measures, so gate_window, coverage_eps and precision_eps carry one meaning on every engine.

run_survey

run_survey(backend, ctx, params, *, monitor=None, task=None) -> ArchiveResult

Run one survey on one backend and return the result.

Parameters:

Name	Type	Description	Default
backend		An object that obeys the :class:~hifuku.backend.Backend protocol.	required
ctx	RunContext	The trees, alignment, model and chart geometry for this survey.	required
params	ArchiveParams	The resolved run parameters.	required
monitor	(callable, None or False)	A checkpoint callback. None selects the default reporter and False turns reporting off. See :mod:hifuku.monitor.	None
task	SurveyTask	Passed to the monitor so that a caller can tell surveys apart.	None

Raises:

Type	Description
ValueError	The monitor names an unknown value, or a value this backend cannot serve. The check runs before any survey work starts.

Source code in src/hifuku/driver.py

def run_survey(backend, ctx, params, *, monitor=None, task=None) -> ArchiveResult:
    """Run one survey on one backend and return the result.

    Parameters
    ----------
    backend
        An object that obeys the :class:`~hifuku.backend.Backend` protocol.
    ctx : RunContext
        The trees, alignment, model and chart geometry for this survey.
    params : ArchiveParams
        The resolved run parameters.
    monitor : callable, None or False
        A checkpoint callback.  None selects the default reporter and False
        turns reporting off.  See :mod:`hifuku.monitor`.
    task : SurveyTask, optional
        Passed to the monitor so that a caller can tell surveys apart.

    Raises
    ------
    ValueError
        The monitor names an unknown value, or a value this backend cannot
        serve.  The check runs before any survey work starts.
    """
    per_checkpoint = params.passes_per_checkpoint * params.variations_per_pass
    n_checkpoints = max(1, params.n_iters // per_checkpoint)

    # The halt reads a trailing window of gate_window checkpoints, so a budget
    # that holds fewer can never fire it.  Say so before the run, because the
    # caller asked for a halt and would otherwise wait for one that cannot come.
    if (params.converge and not params.force_full_run
            and n_checkpoints < params.gate_window):
        warnings.warn(
            f"this budget gives {n_checkpoints} checkpoints and the halt needs "
            f"a window of {params.gate_window}, so the survey cannot halt on "
            f"saturation and will run the full budget.  Raise n_iters, lower "
            f"checkpoint_every, or lower gate_window.",
            RuntimeWarning, stacklevel=2)

    # Validate the monitor before the run starts.  A typo must not surface
    # twenty minutes into a survey.
    bound = bind_monitor(monitor, caps=backend.caps, n_checkpoints=n_checkpoints,
                         gate_window=params.gate_window)

    # The gate scales with the alignment, so the site count is read once here
    # rather than at every checkpoint.
    n_sites = _site_count(ctx)

    handle = backend.prepare(ctx, params)
    backend.seed(handle)
    backend.sync()

    stats = backend.read_stats(handle)
    keys = backend.read_keys(handle, 0)
    n_copied = stats.n_keys

    history = [(0, stats.n_keys, stats.best, float("nan"), float("nan"))]
    cp_new: list[int] = []
    cp_gain: list[float] = []
    prev_keys, prev_gain = stats.n_keys, stats.gain
    converged, halt_iter, total = False, -1, 0
    started = time.perf_counter()

    if keys.size:
        backend.submit_batch(handle, keys, per_checkpoint, _floor(params, stats.best, n_sites))

        for cp in range(n_checkpoints):
            backend.sync()
            stats = backend.read_stats(handle)
            total += per_checkpoint

            cp_new.append(stats.n_keys - prev_keys)
            cp_gain.append(stats.gain - prev_gain)
            prev_keys, prev_gain = stats.n_keys, stats.gain

            coverage, precision = windowed_rates(
                cp_new, cp_gain, stats.elite_range or 1.0, params.gate_window,
                stats.n_keys)
            history.append((total, stats.n_keys, stats.best, coverage, precision))

            halting = (params.converge and not params.force_full_run
                       and survey_converged(coverage, precision,
                                            params.coverage_eps,
                                            params.precision_eps))
            if halting:
                converged, halt_iter = True, total

            if stats.n_keys > n_copied:
                fresh = backend.read_keys(handle, n_copied)
                keys = np.concatenate([keys, fresh]) if keys.size else fresh
                n_copied = stats.n_keys

            # The next chunk changes the archive, so both reads come first.
            logL_map = backend.read_map(handle) if bound and bound.wants_map else None
            trees = backend.read_trees(handle) if bound and bound.wants_trees else None

            more = (cp + 1 < n_checkpoints) and not halting and keys.size > 0
            if more:
                backend.submit_batch(handle, keys, per_checkpoint,
                                     _floor(params, stats.best, n_sites))

            if bound is not None:
                bound(Checkpoint(
                    task=task, iters=total, n_filled=stats.n_keys,
                    best_logL=stats.best, coverage=coverage, precision=precision,
                    converged=converged, final=not more,
                    elapsed=time.perf_counter() - started,
                    logL_map=logL_map, trees=trees))

            if not more:
                break

    return ArchiveResult(
        archive=backend.finalize(handle),
        history=np.array(history, dtype=np.float64),
        converged=converged,
        halt_iter=halt_iter,
    )

ArchiveResult dataclass

A completed survey: the archive, its history, and the halt state.

history has one row per checkpoint of (iteration, n_filled, best_logL, coverage_rate, precision_rate), after a first row that records the seeded archive. The two rates are scale invariant. See :func:windowed_rates. They are NaN until a full window of checkpoints has run.

Source code in src/hifuku/driver.py

@dataclass
class ArchiveResult:
    """A completed survey: the archive, its history, and the halt state.

    ``history`` has one row per checkpoint of
    ``(iteration, n_filled, best_logL, coverage_rate, precision_rate)``, after a
    first row that records the seeded archive.  The two rates are scale
    invariant.  See :func:`windowed_rates`.  They are NaN until a full window of
    checkpoints has run.
    """

    archive: object
    history: np.ndarray
    converged: bool = False
    halt_iter: int = -1

windowed_rates

windowed_rates(cp_new, cp_gain, elite_range, gate_window, n_filled)

Trailing-window coverage and precision rates for the halt.

Both rates are scale invariant in the size of the domain. coverage is the growth of the filled set per checkpoint, as a fraction of the filled count. precision is the mean per-niche elite gain per checkpoint, as a fraction of the elite range.

The division by n_filled makes both rates settle toward zero whether the domain holds tens of niches or tens of thousands. One pair of thresholds therefore fits any cell size and any gate margin.

Both rates are NaN until gate_window checkpoints exist. A NaN compares False against a threshold, so a short run does not halt by accident.

Source code in src/hifuku/driver.py

def windowed_rates(cp_new, cp_gain, elite_range, gate_window, n_filled):
    """Trailing-window coverage and precision rates for the halt.

    Both rates are scale invariant in the size of the domain.  ``coverage`` is
    the growth of the filled set per checkpoint, as a fraction of the filled
    count.  ``precision`` is the mean per-niche elite gain per checkpoint, as a
    fraction of the elite range.

    The division by ``n_filled`` makes both rates settle toward zero whether the
    domain holds tens of niches or tens of thousands.  One pair of thresholds
    therefore fits any cell size and any gate margin.

    Both rates are NaN until ``gate_window`` checkpoints exist.  A NaN compares
    False against a threshold, so a short run does not halt by accident.
    """
    if len(cp_new) < gate_window or n_filled == 0:
        return float("nan"), float("nan")
    denom = gate_window * n_filled
    coverage_rate = sum(cp_new[-gate_window:]) / denom
    precision_rate = sum(cp_gain[-gate_window:]) / (denom * elite_range)
    return coverage_rate, precision_rate

survey_converged

survey_converged(coverage_rate, precision_rate, coverage_eps, precision_eps)

Discovery saturation: both windowed rates at or below their thresholds.

Coverage is the crisp signal. Precision uses a loose threshold because refinement inside a niche never fully stops.

Source code in src/hifuku/driver.py

def survey_converged(coverage_rate, precision_rate, coverage_eps, precision_eps):
    """Discovery saturation: both windowed rates at or below their thresholds.

    Coverage is the crisp signal.  Precision uses a loose threshold because
    refinement inside a niche never fully stops.
    """
    return coverage_rate <= coverage_eps and precision_rate <= precision_eps