lb runs analyze --experiment ID (or --folder for every run of the output
folder, written to data_exports/_folders/<folder>/) writes the tables to
data_exports/<experiment>/, after printing a
summary; --host and --workload filter them. Without either option, an
interactive terminal opens a picker with a preview, then the summary with
Unify, Filters and Cancel. The result ends with the equivalent command.
lb runs list → Analyze opens the same flow on the run's experiment.
lb run --analyze unifies the run's experiment when the run ends; an
interactive lb run asks instead (default No), and --no-analyze never
asks.
The GUI Analytics view lists the experiments, prepares the same summary and
unifies in the background.
All of them use UnificationService from lb_app.api.
List experiments, preview their unification, write it.
Source code in lb_app/services/unification_service.py
78798081828384858687
def__init__(self,catalog:RunCatalogService,export_root:Path,config_path:Path|None=None,)->None:self._catalog=catalog# Absolute, so the printed output folder means the same from anywhere.self._export_root=export_root.resolve()self._config_path=config_path
The named experiment, or the whole folder for None.
Source code in lb_app/services/unification_service.py
100101102103104105106107108109110111112113114
deffind(self,experiment_id:str|None)->ExperimentInfo:"""The named experiment, or the whole folder for ``None``."""ifexperiment_idisNone:folder=self._catalog.folder_experiment()ifnotfolder.runs:raiseUnificationError(f"No runs in {self.output_dir}")returnfolderfound=self._catalog.get_experiment(experiment_id)iffoundisNone:recent=[e.idforeinself._catalog.list_experiments()[:5]]hint=f"; recent experiments: {', '.join(recent)}"ifrecentelse""raiseUnificationError(f"Experiment {experiment_id!r} not found in {self.output_dir}{hint}")returnfound
Experiments newest first, then the whole folder; [] without runs.
Source code in lb_app/services/unification_service.py
939495969798
deflist_targets(self)->list[ExperimentInfo]:"""Experiments newest first, then the whole folder; [] without runs."""experiments=self._catalog.list_experiments()ifnotexperiments:return[]return[*experiments,self._catalog.folder_experiment()]
defprepare(self,experiment:ExperimentInfo,hosts:Sequence[str]=(),workloads:Sequence[str]=(),)->UnificationPreview:"""Load and filter the experiment; nothing is written yet."""ifnot_has_pyarrow():raiseUnificationError("Writing Parquet needs pyarrow: ""pip install 'linux-benchmark-lib[controller]'")_check_names("host",hosts,experiment.hosts)_check_names("workload",workloads,experiment.workloads)host_filter=_effective(hosts,experiment.hosts)workload_filter=_effective(workloads,experiment.workloads)data=load_experiment(experiment.runs).filter(host_filter,workload_filter)returnUnificationPreview(experiment=experiment,data=data,hosts=host_filter,workloads=workload_filter,out_dir=self._out_dir(experiment),command=self._command(experiment,host_filter,workload_filter),)
Write the previewed tables, replacing an earlier unification.
Source code in lb_app/services/unification_service.py
142143144145146147148149150151
defwrite(self,preview:UnificationPreview)->list[Path]:"""Write the previewed tables, replacing an earlier unification."""ifpreview.is_empty:raiseUnificationError(f"Nothing to unify for {preview.experiment.id}: {preview.empty_reason}")try:returnpreview.data.to_parquet(preview.out_dir)exceptExceptionasexc:# disk, permissions, pyarrow: say why, no tracebackraiseUnificationError(f"Could not write {preview.out_dir}: {exc}")fromexc
defsummary_rows(self)->list[tuple[str,str]]:"""Label/value pairs every UI shows before the user confirms."""report=self.data.reporthosts=self.hostsorself.experiment.hostsworkloads=self.workloadsorself.experiment.workloadsrows=[("Experiment",self.experiment.id),("Kind",self.experiment.kind),("Runs",str(len(self.experiment.runs))),("Hosts",", ".join(hosts)or"-"),("Workloads",", ".join(workloads)or"-"),]rows+=[(f"Rows: {name}",str(n))forname,ninself.row_counts.items()]rows+=[("Failed repetitions",str(len(report.failed_repetitions))),("Load errors",str(len(report.errors))),("Load warnings",str(len(report.warnings))),]shown=report.errors[:MAX_SUMMARY_ERRORS]rows+=[(f"Error: {e['source']}",e["message"])foreinshown]hidden=len(report.errors)-len(shown)ifhidden:rows.append(("More errors",f"{hidden} more in load_report.json"))ifself.is_empty:rows.append(("Nothing to unify",self.empty_reason))rows.append(("Output",str(self.out_dir)))returnrows
load_experiment(runs) reads every datasets.json manifest of the given runs
and returns runs, host_info, repetitions, results and samples as pandas
DataFrames plus a LoadReport; ExperimentData.to_parquet(dir) writes them.
defload_experiment(runs:Sequence[RunInfo])->ExperimentData:"""Load every declared dataset of ``runs`` into unified tables."""report=LoadReport()frames:dict[str,list[pd.DataFrame]]={"results":[],"samples":[],"host_info":[],}run_rows:list[dict[str,Any]]=[]repetition_rows:list[dict[str,Any]]=[]forruninruns:root=run.output_rootrun_rows.append(_run_row(run,report))formanifest_dirin_manifest_dirs(root):try:manifest=read_manifest(manifest_dir)exceptExceptionasexc:report.error(str(manifest_dir),f"invalid {MANIFEST_FILENAME}: {exc}")continuectx=SourceContext(run.run_id,_host_of(root,manifest_dir,manifest),manifest.workload,manifest.plugin,)_load_manifest(manifest,manifest_dir,ctx,frames,report)repetition_rows.extend(_repetitions(manifest,manifest_dir,ctx,report))_report_unmanifested(root,report)returnExperimentData(runs=_runs_frame(run_rows),host_info=_concat(frames["host_info"],HOST_INFO_COLUMNS),repetitions=_repetitions_frame(repetition_rows),results=_concat(frames["results"],RESULTS_COLUMNS),samples=_concat(frames["samples"],SAMPLES_COLUMNS),report=report,)
Keep the given hosts and workloads; an empty filter keeps everything.
A filter applies to every table that has its column, so runs is
never filtered and host_info only by host.
Source code in lb_analytics/unify/experiment.py
3435363738394041424344454647484950
deffilter(self,hosts:Sequence[str]=(),workloads:Sequence[str]=())->ExperimentData:"""Keep the given hosts and workloads; an empty filter keeps everything. A filter applies to every table that has its column, so ``runs`` is never filtered and ``host_info`` only by host. """tables:dict[str,pd.DataFrame]={}fornameinTABLES:frame=getattr(self,name)ifhostsand"host"inframe.columns:frame=frame[frame["host"].isin(hosts)]ifworkloadsand"workload"inframe.columns:frame=frame[frame["workload"].isin(workloads)]tables[name]=frame.reset_index(drop=True)returnExperimentData(**tables,report=self.report)
Write one Parquet file per table plus load_report.json.
Source code in lb_analytics/unify/experiment.py
525354555657585960616263
defto_parquet(self,out_dir:Path)->list[Path]:"""Write one Parquet file per table plus ``load_report.json``."""out_dir.mkdir(parents=True,exist_ok=True)paths=[]fornameinTABLES:path=out_dir/f"{name}.parquet"getattr(self,name).to_parquet(path,index=False)paths.append(path)report_path=out_dir/"load_report.json"report_path.write_text(json.dumps(self.report.to_dict(),indent=2))paths.append(report_path)returnpaths
Runs are grouped into experiments by RunCatalogService: list_experiments(),
get_experiment(id) and folder_experiment() return ExperimentInfo objects
whose runs feed load_experiment directly:
lb_analytics.api can answer, per benchmark metric, how much of a machine's
performance its specifications explain, and predict a machine that was not
benchmarked. It is a library for notebooks; there is no command.
fromlb_analytics.apiimportevaluate,fit,load_experiment,machines,predict,targetsfromlb_app.apiimportRunCatalogServicecatalog=RunCatalogService(output_dir)data=[load_experiment(catalog.get_experiment(e).runs)foreinexperiment_ids]m,t=machines(*data),targets(*data)report=evaluate(m,t,["physical_cpus"],targets_filter="stress_ng")target="stress_ng/stress_ng/stress_ng_plugin/bogo_ops_per_s_real[stressor=cpu]"model=fit(m,t,target,["physical_cpus"])new=machines(load_experiment(catalog.get_experiment("exp-new-host").runs))predict(model,new.iloc[0])# Prediction(value, low, high, unit, extrapolated)
Pass the same experiments to machines and targets: the two tables are
joined on the machine name.
machines has one row per machine: a host with one set of features
(cores, MHz, caches, memory, disk). The same host re-created with another
size is another machine (host#<hash>); small drifts (a few KiB of
MemTotal, bogomips, swap) do not split a host.
targets has one row per machine and target, with the median over the
successful repetitions, the IQR and the count. A run whose host_info is
missing counts for its host's machine when the host has only one; otherwise
its results are left out with a warning.
fit is a log-linear regression: a coefficient of 0.9 on physical_cpus
means doubling the cores multiplies the metric by about 1.87.
evaluate leaves each machine out in turn and compares the model with the
mean of the other machines and with the nearest one on the chosen features.
A model is worth using only where beats_baselines is true.
predict gives a range: the worst error seen leaving machines out, plus the
run-to-run noise. With one repetition per machine the noise is unknown, and
predict warns that the range covers only the validation error. A feature
outside the training range, or a combination of features unlike any training
machine (16 cores with the smallest memory), warns that the prediction is an
extrapolation.
Limits: the machines must differ in the chosen features (identical VMs explain
nothing), and features that move together (cores and memory on VM sizes that
scale both) are refused as collinear; use one to three features with fewer than
15 machines; the kernel
reports virtio disks as rotational, so disk_rotational is unreliable on VMs.
The features of a machine come from any run on it, so a short baseline run is
enough to predict it.
One row per machine: a host with one set of identifying features.
A host seen with one size is named after the host. A host seen with
several sizes (a VM re-created) gives one machine per size, named
host#<hash of its features> so the name does not depend on which
experiments are passed. Feature values come from the machine's first run.
defmachines(*data:ExperimentData)->pd.DataFrame:"""One row per machine: a host with one set of identifying features. A host seen with one size is named after the host. A host seen with several sizes (a VM re-created) gives one machine per size, named ``host#<hash of its features>`` so the name does not depend on which experiments are passed. Feature values come from the machine's first run. """info=pd.concat([d.host_infofordindata],ignore_index=True)ifinfo.empty:returnpd.DataFrame(columns=MACHINE_COLUMNS)runs=pd.concat([d.runsfordindata],ignore_index=True)started=runs.drop_duplicates("run_id").set_index("run_id")["created_at"]rows=[]for(run_id,host),groupininfo.groupby(["run_id","host"],sort=False):values={(category,name):valueforcategory,name,valueingroup[["category","name","value"]].itertuples(index=False)ifnotpd.isna(value)}rows.append({"host":host,"run_id":run_id,"started":started.get(run_id),**_features(values),})frame=pd.DataFrame(rows).sort_values(["started","run_id"],na_position="last",kind="stable")frame["mem_gib"]=(frame["mem_bytes"]/_GIB).round()frame["disk_gib"]=(frame["disk_bytes"]/_GIB).round()keys=["host",*_IDENTITY,"mem_gib","disk_gib"]groups=list(frame.groupby(keys,dropna=False,sort=False))sizes=Counter(key[0]forkey,_ingroups)out=[]forkey,groupingroups:host=str(key[0])name=hostifsizes[host]==1elsef"{host}#{_digest(key[1:])}"first=group.iloc[0].to_dict()out.append({**first,"machine":name,"runs":list(group["run_id"])})returnpd.DataFrame(out,columns=MACHINE_COLUMNS)
Median, IQR and count of every target on every machine.
Failed repetitions are left out. A run whose host_info is missing
counts for its host's machine when the host has only one; otherwise (a
host never described, or one with several sizes) its results are left out
with a warning, since they have no features to model.
deftargets(*data:ExperimentData)->pd.DataFrame:"""Median, IQR and count of every target on every machine. Failed repetitions are left out. A run whose ``host_info`` is missing counts for its host's machine when the host has only one; otherwise (a host never described, or one with several sizes) its results are left out with a warning, since they have no features to model. """found=machines(*data)owner={(run_id,host):machineformachine,host,run_idsinzip(found["machine"],found["host"],found["runs"],strict=True)forrun_idinrun_ids}results=pd.concat([d.resultsfordindata],ignore_index=True)reps=pd.concat([d.repetitionsfordindata],ignore_index=True)keys=["run_id","host","workload","repetition"]failed_rows=reps.loc[reps["success"].eq(False),keys]failed=set(failed_rows.itertuples(index=False,name=None))keep=[keynotinfailedforkeyinresults[keys].itertuples(index=False,name=None)]results=results[keep].copy()by_host=found.groupby("host")["machine"].unique()results["machine"]=[owner.get((run_id,host))or(by_host[host][0]ifhostinby_hostandlen(by_host[host])==1elseNone)forrun_id,hostinzip(results["run_id"],results["host"],strict=True)]unowned=results.loc[results["machine"].isna(),"host"].unique().tolist()ifunowned:warnings.warn(f"Results of {', '.join(map(str,unowned))} are left out: no ""host_info tells which machine produced them.",stacklevel=2,)results["value"]=pd.to_numeric(results["value"],errors="coerce")results=results.dropna(subset=["machine","value"])ifresults.empty:returnpd.DataFrame(columns=TARGET_COLUMNS)results["target"]=_target_names(results)grouped=results.groupby(["machine","target"],sort=True)out=grouped.agg(workload=("workload","first"),metric=("metric","first"),unit=("unit","first"),median=("value","median"),n=("value","size"),)out["iqr"]=grouped["value"].quantile(0.75)-grouped["value"].quantile(0.25)returnout.reset_index()[TARGET_COLUMNS]
Fit log(median) = b0 + Σ bᵢ·log(featureᵢ) for one target.
disk_rotational enters as 0/1 instead of its log. The returned model
carries its leave-one-machine-out error, which sets the range of
predict, and the run-to-run noise of machines measured at least twice
(None when none was).
deffit(machines:pd.DataFrame,targets:pd.DataFrame,target:str,features:Sequence[str],)->Model:"""Fit ``log(median) = b0 + Σ bᵢ·log(featureᵢ)`` for one target. ``disk_rotational`` enters as 0/1 instead of its log. The returned model carries its leave-one-machine-out error, which sets the range of ``predict``, and the run-to-run noise of machines measured at least twice (``None`` when none was). """features=list(features)table=_table(machines,targets,target,features)design=_design(table,features)coef=_solve(design,np.log(table["median"].to_numpy(float)))# _table refused collinear features, so the inverse exists.xtx_inv=np.linalg.inv(design.T@design)held_out=_loo(table,features)held_out=held_out[held_out["method"]=="model"]ratios=held_out["predicted"].to_numpy(float)/held_out["actual"].to_numpy(float)log_errors=np.abs(np.log(ratios))repeated=table["n"]>=2rel_iqr=(table["iqr"]/table["median"])[repeated]returnModel(target=target,unit=_unit(table),features=tuple(features),intercept=float(coef[0]),coefficients={f:float(c)forf,cinzip(features,coef[1:],strict=True)},machines=tuple(table["machine"]),feature_ranges={f:(float(table[f].min()),float(table[f].max()))forfinfeatures},max_log_error=float(log_errors.max()),rel_iqr=float(rel_iqr.median())ifrepeated.any()elseNone,xtx_inv=tuple(tuple(float(v)forvinrow)forrowinxtx_inv),max_leverage=float(_leverage(design,xtx_inv).max()),)
defevaluate(machines:pd.DataFrame,targets:pd.DataFrame,features:Sequence[str],targets_filter:str|None=None,)->pd.DataFrame:"""Leave-one-machine-out errors of the model and both baselines. One row per target and method. Targets that cannot be fitted get one ``skipped`` row with the reason in ``note``. """features=list(features)_check_features(features)names=sorted(targets["target"].unique())iftargets_filter:names=[namefornameinnamesiftargets_filterinname]rows:list[dict[str,Any]]=[]fornameinnames:count=int((targets["target"]==name).sum())try:table=_table(machines,targets,name,features)exceptPredictionErrorasexc:rows.append({"target":name,"method":"skipped","machines":count,"note":str(exc),})continueerrors=_loo(table,features).groupby("method")["pct_error"]medians=errors.apply(lambdae:e.abs().median())maxima=errors.apply(lambdae:e.abs().max())wins=bool(medians["model"]<min(medians["mean"],medians["nearest"]))rows.extend({"target":name,"method":method,"machines":count,"median_abs_pct_error":float(medians[method]),"max_abs_pct_error":float(maxima[method]),"beats_baselines":winsifmethod=="model"elseNone,"note":"",}formethodinMETHODS)returnpd.DataFrame(rows,columns=EVALUATE_COLUMNS)
Every held-out prediction behind evaluate for one target.
One row per machine and method: actual, predicted, pct_error.
Source code in lb_analytics/predict/model.py
166167168169170171172173174175176177
defloo_predictions(machines:pd.DataFrame,targets:pd.DataFrame,target:str,features:Sequence[str],)->pd.DataFrame:"""Every held-out prediction behind ``evaluate`` for one target. One row per machine and method: ``actual``, ``predicted``, ``pct_error``. """features=list(features)return_loo(_table(machines,targets,target,features),features)
Predict model.target for one machine, with the model's range.
The range is the worst leave-one-machine-out error plus the target's
run-to-run noise. A feature outside the training range, or a combination
of features unlike any training machine, warns and sets extrapolated.
defpredict(model:Model,machine:Mapping[str,Any]|pd.Series)->Prediction:"""Predict ``model.target`` for one machine, with the model's range. The range is the worst leave-one-machine-out error plus the target's run-to-run noise. A feature outside the training range, or a combination of features unlike any training machine, warns and sets ``extrapolated``. """log_value=model.interceptextrapolated=Falsepoint=[1.0]forfeatureinmodel.features:x=to_number(machine.get(feature))ifmath.isnan(x)or(x<=0andfeaturenotinBINARY_FEATURES):raisePredictionError(f"Feature '{feature}' is missing or not positive "f"({machine.get(feature)!r}).")low,high=model.feature_ranges[feature]ifnotlow<=x<=high:extrapolated=Truewarnings.warn(f"{feature}={x:g} is outside the training range "f"[{low:g}, {high:g}]: the prediction is an extrapolation.",stacklevel=2,)term=xiffeatureinBINARY_FEATURESelsemath.log(x)point.append(term)log_value+=model.coefficients[feature]*termrow=np.array([point])leverage=float(_leverage(row,np.array(model.xtx_inv))[0])ifnotextrapolatedandleverage>model.max_leverage*(1+1e-9):extrapolated=Truewarnings.warn("This combination of features is unlike any training machine, ""although each one is in range: the prediction is an extrapolation.",stacklevel=2,)noise=model.rel_iqrifnoiseisNone:warnings.warn("The run-to-run noise is unknown (one repetition per machine): ""the range covers only the validation error.",stacklevel=2,)noise=0.0value=math.exp(log_value)spread=math.exp(model.max_log_error+math.log1p(noise))returnPrediction(value,value/spread,value*spread,model.unit,extrapolated)