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")