pdmlabs.optimization.trial_sink

Cross-process best-trial spool for PdMLabs experiments.

Why this exists

Every optimizer backend that honours n_jobs > 1 evaluates the experiment’s optimization_objective in a separate OS process:

• mango – joblib.Parallel(n_jobs) (loky)

• gpyopt – joblib.Parallel(n_jobs=n_jobs) (loky)

• optuna – joblib.Parallel(backend="loky") (loky)

• smac – Scenario(n_workers=n_jobs) -> dask ``Client(processes=True,

  threads_per_worker=1)``

• hyperopt – sequential (main process)

Assignments the objective makes to self therefore land on a deserialized copy of the experiment that is discarded when the task ends. This module is a real IPC channel: workers spool their best trial into a shared directory and the main process reads the global best back after the optimizer returns. Nothing is ever re-evaluated to recover it.

Design invariants

1. TrialSink holds only str / bool. It is cheap to pickle into the objective closure and owns no OS handles, sockets or threads.

2. The per-process high-water mark (_LOCAL_BEST) is a module global of this importable module and is never pickled. cloudpickle serialises classes from importable modules by reference, so TrialSink.record running in a worker touches that worker’s own registry. A registry defined in __main__ or captured in a closure would instead be copied by value into every task: the high-water mark would never advance and – worse – several processes would share one file token and clobber each other. Corollary: never call cloudpickle.register_pickle_by_value("pdmlabs").

3. The registry is keyed by (spool_dir, os.getpid()) so a forked child cannot inherit, and then overwrite, its parent’s spool file.

4. Commit protocol: the pipeline blob is written to a versioned name that no process has written before, then the metadata file (which names that blob) is atomically replaced. A worker killed between the two steps leaves a metadata file pointing at the previous, complete blob – never a torn pair.

5. Two marks per process, in two file families. meta-*.json tracks the best score seen and is scalars-only; pmeta-*.json plus its blob tracks the best trial that actually carried a pipeline. A trial can have a score and a threshold but no pipeline – that is what a run served from the MLflow cache looks like – and separating the two means such a trial can win on score without costing us the best pipeline we hold. Because only the pipe family owns a blob, this costs no extra pipeline serialisations.

Correctness of the per-process high-water mark

The globally best trial ran in exactly one process, and within that process it was necessarily that process’s best too. Hence the globally best record is always among the spooled files. Non-finite scores are rejected outright: a NaN high-water mark would make every later comparison False and silently stop that process from spooling again.

What a recovered record guarantees

score/th/th_to_rul/params always describe one real trial: the best one recorded. pipeline is normally that same trial’s, and pipeline_is_winner says so. When the winner carried no pipeline it falls back to the best trial that did, and pipeline_score/pipeline_params/ pipeline_th describe that trial – so a fallback model is stamped with its own threshold rather than the winner’s.

Ties

Several parameterizations can reach the same best objective. The optimizer picks one of them as best_params while the sink independently picks one tied record, and they need not be the same configuration. TrialSink.best() therefore accepts prefer_params and, among tied records, prefers the one whose params match; failing that it takes the earliest tied trial (deterministic given the file set) and reports matched_best_params=False so the caller can say so out loud. No scheme can guarantee a match: a strict high-water mark keeps the first trial to reach a score, so if the optimizer’s pick was a later tie in the same process that record was never spooled.

Functions

normalise_params(params)

Canonicalise a trial's params for storage and display.

Classes

TrialRecord(score[, th, th_to_rul, ...])	The globally best trial recovered from a spool directory.
TrialSink(spool_dir[, maximize])	Spools each worker process's best trial into a shared directory.

class pdmlabs.optimization.trial_sink.TrialRecord(score: float, th: float = None, th_to_rul: float = None, pipeline: object = None, pipeline_error: str = None, params: dict = None, matched_best_params: bool = True, tied_count: int = 1, pipeline_is_winner: bool = True, pipeline_score: float = None, pipeline_params: dict = None, pipeline_th: float = None)

Bases: object

The globally best trial recovered from a spool directory.

score/th/th_to_rul/params always describe one real trial – the best one recorded. pipeline usually belongs to that same trial, but when the winner carried no pipeline (it came from the run cache, or failed to serialize) it falls back to the best trial that did produce one; pipeline_is_winner says which, and pipeline_score/pipeline_params identify the fallback.

matched_best_params: bool = True

params: dict = None

pipeline: object = None

pipeline_error: str = None

pipeline_is_winner: bool = True

pipeline_params: dict = None

pipeline_score: float = None

pipeline_th: float = None

score: float

th: float = None

th_to_rul: float = None

tied_count: int = 1

class pdmlabs.optimization.trial_sink.TrialSink(spool_dir, maximize=True)

Bases: object

Spools each worker process’s best trial into a shared directory.

Instances are created in the main process by create(), captured by the optimization_objective closure, and shipped to every worker. Workers call record(); the main process calls best() once, after _run_optimizer returns, then cleanup().

best(prefer_params=None)

Return the globally best spooled trial, or None if there is none.

Call only after the optimizer has returned (all workers finished). Reads every small metadata file but unpickles at most one blob.

prefer_params is the optimizer’s best_params. When several trials tie on the best score, the record whose params match it is preferred, so th/best_pipeline describe the same configuration the optimizer reported. See the module docstring’s Ties section.

cleanup()

Remove the spool directory. Idempotent; safe to call from finally.

classmethod create(maximize=True)

Create a fresh, private spool directory.

mkdtemp uses O_CREAT | O_EXCL, so the directory is unique against every process on the machine – concurrent experiments cannot collide. Set PDMLABS_TRIAL_SPOOL_DIR to place the spool somewhere other than $TMPDIR (useful when pipelines are large or /tmp is small).

is_better(candidate, incumbent)

Public direction-aware comparison (used by the experiment cross-check).

maximize

record(score, pipeline=None, th=None, th_to_rul=None, params=None)

Spool this trial if score beats this process’s high-water marks.

Two marks are kept per process. The any mark tracks the best score regardless of whether a pipeline came with it, and is written as a small scalars-only meta-*.json; it is authoritative for score/th. The pipe mark tracks the best trial that actually carried a pipeline and owns the expensive blob. Keeping them apart means a trial with no pipeline (served from the run cache) can still win on score without discarding the best pipeline we hold, and costs no extra blob writes.

Safe to call from any process, including the main one. Returns True if anything was committed. Never raises: failing to spool degrades the reported result but must not abort an optimization run.

spool_dir

pdmlabs.optimization.trial_sink.normalise_params(params)

Canonicalise a trial’s params for storage and display.

Values are kept as strings, matching the idiom _check_cached_run already uses against MLflow (current_run.loc['params.' + k] != str(v)). Exact equality on the raw values is not usable: an adapter may hand back a value with a different type than it received.