From 68e567bd02bbccc3be716d450363d60a6d840e56 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Mon, 26 Oct 2026 22:20:24 +0100 Subject: [PATCH] Tail: the molecular replacement of a prediction or --mr starts beside the analyses; a pass's integrated observations are handed back on a thread of their own The search (or the carry of the early search) reads of the merge only its French-Wilson amplitudes, which the analyses make last from a tensor fitted to the merge alone, so - as the first validation of an experimental model already does - it starts before the analyses on a copy given the same amplitudes, where no twin hypothesis' merge is searched beside it and no reference supplies the free set. The pass's integrated observations (ReflectionArena mappings, gigabytes) are freed on a thread of their own as the pass returns, joined when the run goes, instead of before the outputs are written and before the broker publishes the result. Output unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015CeEGgpKdfkCTJb3YNMJcu --- rugnux/Rugnux.cpp | 12 ++++++ rugnux/Rugnux.h | 4 ++ rugnux/RugnuxScaleMerge.cpp | 77 +++++++++++++++++++++++++------------ 3 files changed, 69 insertions(+), 24 deletions(-) diff --git a/rugnux/Rugnux.cpp b/rugnux/Rugnux.cpp index 806de40e0..6eb5ee2f4 100644 --- a/rugnux/Rugnux.cpp +++ b/rugnux/Rugnux.cpp @@ -1132,6 +1132,16 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b std::atomic total_uncompressed_bytes = 0; EndMessage end_msg; std::future unmerged_mtz; + // The pass's integrated observations - gigabytes, in mappings of their own (ReflectionArena) - are handed + // back on a thread of their own as the pass returns: nothing after it reads them, and unmapping them was most + // of the time the pass took to end. The unmerged table made from them beside the refinement goes first. + const auto retire_outcomes = [&] { + retired_outcomes_ = std::async(std::launch::async, + [i = std::move(indexer), u = std::move(unmerged_mtz)]() mutable { + u = {}; + i.reset(); + }).share(); + }; PipelineLocals p{ .logger = logger, .result = result, .observer = observer, .geometry_prepass = geometry_prepass, .full = full, .calibration = calibration, @@ -1218,6 +1228,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b Note(note); } if (finished) { + retire_outcomes(); teardown_mark.emplace(logger, "pass teardown"); return result; } @@ -1363,6 +1374,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b TimingMark mark(logger, "leftover-lattice census wait"); result.leftover_lattices = census_future.get(); } + retire_outcomes(); teardown_mark.emplace(logger, "pass teardown"); return result; } diff --git a/rugnux/Rugnux.h b/rugnux/Rugnux.h index 2389bc1ac..f670373ff 100644 --- a/rugnux/Rugnux.h +++ b/rugnux/Rugnux.h @@ -1623,4 +1623,8 @@ public: // pass cut short gives way to the last complete one. void Cancel() { cancelled_ = true; } bool IsCancelled() const { return cancelled_; } + +private: + // The last pass's integrated observations being handed back (RunPipeline); last, so it is waited for first. + std::shared_future retired_outcomes_; }; diff --git a/rugnux/RugnuxScaleMerge.cpp b/rugnux/RugnuxScaleMerge.cpp index af29b845f..e3021e5dc 100644 --- a/rugnux/RugnuxScaleMerge.cpp +++ b/rugnux/RugnuxScaleMerge.cpp @@ -3840,6 +3840,47 @@ bool Rugnux::ScaleMergeAndSymmetry(PipelineLocals &p) { report_shell_d_min, first_schedule); }); } + // The molecular replacement (below): the search started on the geometry pre-pass's merge, where there is one, + // taken over where it holds on this one - not where the twin hypothesis' merge is searched too, as that search + // is this merge's own - and this merge searched itself where it does not. + const auto search_model = [&](const std::vector &merges, Logger &log) { + if (early_mr_ && merges.size() == 1) { + TimingMark early_mark(log, "early molecular replacement wait"); + try { + MolecularReplacementResult early = early_mr_->run.get(); + early_mark.Stop(); + early_mr_->log.ReplayInto(log); + if (auto carried = CarryMolecularReplacement( + early, {"geometry pre-pass's merge", &early_mr_->merged, early_mr_->cell, early_mr_->space_group}, + merges[0], log, static_cast(config_.nthreads), merge_on_card)) + return std::move(*carried); + } catch (const std::exception &e) { + log.Warning("Molecular replacement on the geometry pre-pass's merge failed ({}) - searched again", + e.what()); + } + } + return RunMolecularReplacement(*config_.model, merges, log, static_cast(config_.nthreads), + merge_on_card); + }; + // A prediction, or --mr, has no validation to start here, but the search reads of the merge only its amplitudes + // too. Where there is no twin hypothesis' merge to search beside it and no reference to take the free set from, + // nothing up to the search changes the merge or the group, so it starts here the same way, beside the analyses, + // on a copy given the same amplitudes. Its log is held, and printed where it is taken. + Logger model_search_log = Logger::Buffered(); + std::future model_search_run; + if (model_run && (model_predicted || config_.force_mr) && !twin_merge && !model_twin_merge + && config_.reference_data.empty()) { + model_merged = sm.merged; + model_search_run = std::async(std::launch::async, + [&, device = get_gpu(), sg = experiment_.GetSpaceGroupOrP1()] { + set_gpu(device); + FrenchWilsonOptions fw_opts; + fw_opts.num_threads = config_.nthreads; + fw_opts.anisotropy_hkl = AnisotropyPriorHKL(model_merged, gemmi::UnitCell(*result.consensus_cell), twin_sg); + ApplyFrenchWilson(model_merged, sg, fw_opts); + return search_model({{"MERGE", &model_merged, *result.consensus_cell, sg}}, model_search_log); + }); + } // Not on the geometry pre-pass, nor on a superseded one: the analysis goes into that pass's // statistics text and its written reflections, and neither survives the run. The promotion flag // below is a different thing - it is what the SEARCH did, the second pass reads it, and it is @@ -4692,31 +4733,19 @@ bool Rugnux::ScaleMergeAndSymmetry(PipelineLocals &p) { if (!ledger_ahead.valid()) rsm.reset(); const size_t merge_on_card = rsm && rsm->OnGPU() ? rsm->DeviceBytes() : 0; - std::vector merges{{"MERGE", &sm.merged, *result.consensus_cell, experiment_.GetSpaceGroupOrP1()}}; - if (model_twin) - merges.push_back({"TWIN_SUBGROUP_MERGE", &model_twin->merged, model_twin->cell, model_twin->space_group, - model_twin->law}); - // The search started on the geometry pre-pass's merge, where there is one, taken over where it holds on - // this one. Not where the twin hypothesis' merge is searched too: that search is this merge's own. - std::optional carried; - if (early_mr_ && !model_twin) { - TimingMark early_mark(logger, "early molecular replacement wait"); - try { - MolecularReplacementResult early = early_mr_->run.get(); - early_mark.Stop(); - early_mr_->log.ReplayInto(logger); - carried = CarryMolecularReplacement( - early, {"geometry pre-pass's merge", &early_mr_->merged, early_mr_->cell, early_mr_->space_group}, - merges[0], logger, static_cast(config_.nthreads), merge_on_card); - } catch (const std::exception &e) { - logger.Warning("Molecular replacement on the geometry pre-pass's merge failed ({}) - searched again", - e.what()); - } + MolecularReplacementResult mr; + if (model_search_run.valid()) { + TimingMark wait_mark(logger, "molecular replacement (beside the analyses) wait"); + mr = model_search_run.get(); + wait_mark.Stop(); + model_search_log.ReplayInto(logger); + } else { + std::vector merges{{"MERGE", &sm.merged, *result.consensus_cell, experiment_.GetSpaceGroupOrP1()}}; + if (model_twin) + merges.push_back({"TWIN_SUBGROUP_MERGE", &model_twin->merged, model_twin->cell, + model_twin->space_group, model_twin->law}); + mr = search_model(merges, logger); } - MolecularReplacementResult mr = carried ? std::move(*carried) - : RunMolecularReplacement(*config_.model, merges, logger, - static_cast(config_.nthreads), - merge_on_card); if (mr.chosen >= 0) { const MRCandidate &c = mr.candidates[mr.chosen]; if (mr.decision == "ENANTIOMORPH")