leonarski_f
2e296e6bed
In broker live mode the refined model, its map coefficients and the difference-map blobs stayed in the broker for the viewer routes and were lost at the next run. They now go to the writer in the END message as the placed model and maps already do - EndMessage::refined_model (PlacedModelMessage), refined_map_coefficients (MtzTable) and blobs (DensityBlobsMessage, one vector per field) - and the writer writes <prefix>_refined_model.cif (+ .pdb), <prefix>_refined_maps.mtz and <prefix>_blobs.json, the names the CLI gives them. The blobs' JSON is made by one function (DensityBlobsJson) that the CLI now writes through as well, so the two files are byte-identical. Optional CBOR fields, no version bump. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015CeEGgpKdfkCTJb3YNMJcu (cherry picked from commit c65220d96a96d2b1897f3c6632c94f9379b670a3)
719 lines
35 KiB
C++
719 lines
35 KiB
C++
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#include "JFJochReceiverRugnux.h"
|
|
|
|
#include <cmath>
|
|
|
|
#include "spdlog/fmt/fmt.h"
|
|
|
|
#include "JFJochReceiverLite.h"
|
|
#include "../common/HostMemory.h"
|
|
#include "../frame_serialize/CBORStream2Deserializer.h"
|
|
#include "../frame_serialize/CBORStream2Serializer.h"
|
|
#include "../image_analysis/indexing/IndexerThreadPool.h"
|
|
#include "../rugnux/RugnuxLive.h"
|
|
#include "../rugnux/ResultReport.h"
|
|
#include "../writer/RugnuxResultBlocks.h"
|
|
#include "RugnuxStructure.h"
|
|
|
|
using namespace std::chrono_literals;
|
|
|
|
namespace {
|
|
// Kept for the rest of the broker - the receiver, the pushers, the preview - when the host memory
|
|
// rugnux plans against is decided.
|
|
constexpr size_t RUGNUX_HOST_MEMORY_MARGIN = 16ULL * 1024 * 1024 * 1024;
|
|
|
|
// The host memory rugnux plans against (HostMemory.h) is the machine's budget less what the broker
|
|
// already holds: the image buffer, which has to keep the whole collection and so is most of the
|
|
// machine, and a margin. Never 0, which would take the whole machine's budget: at worst what could
|
|
// run side by side runs one part after the other.
|
|
size_t RugnuxHostBudget(const ImageBuffer &buffer) {
|
|
const size_t budget = DefaultHostMemoryBudget();
|
|
const size_t taken = buffer.GetBufferSize() + RUGNUX_HOST_MEMORY_MARGIN;
|
|
return budget > taken ? budget - taken : 1;
|
|
}
|
|
|
|
double SecondsSince(std::chrono::steady_clock::time_point t) {
|
|
return std::chrono::duration<double>(std::chrono::steady_clock::now() - t).count();
|
|
}
|
|
|
|
// A run reachable for Cut while it exists: registered in `slot` (under `m`, cancelled at once where the
|
|
// cut came first) and taken off again before it is destroyed, an exception included.
|
|
class Reachable {
|
|
std::mutex &m;
|
|
Rugnux *&slot;
|
|
public:
|
|
Reachable(std::mutex &m, Rugnux *&slot, Rugnux &run, const bool &cut, const std::atomic<bool> &cancelled)
|
|
: m(m), slot(slot) {
|
|
std::unique_lock ul(m);
|
|
slot = &run;
|
|
if (cut || cancelled)
|
|
run.Cancel();
|
|
}
|
|
~Reachable() {
|
|
std::unique_lock ul(m);
|
|
slot = nullptr;
|
|
}
|
|
};
|
|
|
|
// What rugnux reports as it goes: the per-image results of its canonical pass into the receiver's
|
|
// plots and scan result - the plots accumulate, so an image reported by the pre-pass as well would be
|
|
// counted twice - and its steps into the live status (not the log: they are processing results).
|
|
class LiveObserver : public RugnuxObserver {
|
|
JFJochReceiverPlots &plots;
|
|
ScanResultGenerator &scan_result;
|
|
const AzimuthalIntegrationMapping &mapping;
|
|
RugnuxLiveStatus &status;
|
|
std::atomic<bool> canonical{false};
|
|
public:
|
|
LiveObserver(JFJochReceiverPlots &plots, ScanResultGenerator &scan_result,
|
|
const AzimuthalIntegrationMapping &mapping, RugnuxLiveStatus &status)
|
|
: plots(plots), scan_result(scan_result), mapping(mapping), status(status) {}
|
|
|
|
void OnImagePass(bool c) override { canonical = c; }
|
|
void OnImageProcessed(const DataMessage &msg) override {
|
|
if (!canonical)
|
|
return;
|
|
plots.Add(msg, AzimuthalIntegrationProfile(mapping));
|
|
scan_result.Add(msg);
|
|
}
|
|
void OnProgress(uint64_t done, uint64_t total) override { status.Progress(done, total); }
|
|
void OnStep(const std::string &title) override { status.Step(title); }
|
|
void OnNote(const std::string &text) override { status.Note(text); }
|
|
};
|
|
|
|
std::string StateName(RugnuxLiveState state) {
|
|
switch (state) {
|
|
case RugnuxLiveState::Idle: return "idle";
|
|
case RugnuxLiveState::Running: return "running";
|
|
case RugnuxLiveState::Done: return "done";
|
|
case RugnuxLiveState::Failed: return "failed";
|
|
case RugnuxLiveState::Cancelled: return "cancelled";
|
|
case RugnuxLiveState::Skipped: return "skipped";
|
|
}
|
|
return "unknown";
|
|
}
|
|
}
|
|
|
|
namespace {
|
|
// A score the status carries only where it is a number: JSON has no NaN or infinity (nlohmann writes null)
|
|
void SetIfFinite(std::optional<double> &out, double v) {
|
|
if (std::isfinite(v))
|
|
out = v;
|
|
}
|
|
}
|
|
|
|
std::optional<RugnuxLiveModel> LiveModel(const ProcessResult &result) {
|
|
if (!result.model_validation && !result.model_prior)
|
|
return std::nullopt;
|
|
RugnuxLiveModel m;
|
|
m.placement = result.model_placement;
|
|
if (result.model_prior) {
|
|
const MolecularReplacementResult &mr = *result.model_prior;
|
|
m.model_name = mr.model_name;
|
|
m.predicted = mr.predicted_model;
|
|
m.mr_decision = mr.decision;
|
|
if (!mr.ok)
|
|
m.reason = mr.failure_reason;
|
|
if (mr.chosen >= 0) {
|
|
const MRCandidate &c = mr.candidates[mr.chosen];
|
|
m.mr_space_group = c.space_group.xhm();
|
|
m.copies = c.copies;
|
|
SetIfFinite(m.tf_z, c.tf_z);
|
|
SetIfFinite(m.rf_z, c.rf_z);
|
|
SetIfFinite(m.llg_z, c.llg_z);
|
|
SetIfFinite(m.fit_sigma, c.null_sigma);
|
|
SetIfFinite(m.mr_d_min, mr.d_min_search);
|
|
}
|
|
}
|
|
if (result.model_validation) {
|
|
const ModelValidationResult &v = *result.model_validation;
|
|
if (!v.model_name.empty())
|
|
m.model_name = v.model_name;
|
|
if (!v.ok) {
|
|
if (m.reason.empty())
|
|
m.reason = v.failure_reason;
|
|
} else {
|
|
SetIfFinite(m.r_work, v.r_work);
|
|
SetIfFinite(m.r_free, v.r_free);
|
|
SetIfFinite(m.cc_model, v.cc_model_overall);
|
|
if (v.fit_tested)
|
|
SetIfFinite(m.fit_sigma, v.r_work_sigma);
|
|
SetIfFinite(m.map_atom_sigma, v.mean_atom_density_sigma);
|
|
if (v.rigid_body_applied) {
|
|
SetIfFinite(m.rigid_body_angle_deg, v.rigid_body_angle_deg);
|
|
SetIfFinite(m.rigid_body_shift_A, v.rigid_body_shift_A);
|
|
}
|
|
if (!v.anomalous_sites.empty()) {
|
|
m.anomalous_site = v.anomalous_sites.front().label;
|
|
SetIfFinite(m.anomalous_site_sigma, v.anomalous_sites.front().sigma);
|
|
}
|
|
}
|
|
}
|
|
if (result.refinement) {
|
|
const RefinementResult &rf = *result.refinement;
|
|
RugnuxLiveRefinement r;
|
|
r.performed = rf.ok;
|
|
r.cancelled = rf.cancelled;
|
|
r.reason = rf.failure_reason;
|
|
r.on_gpu = rf.on_gpu;
|
|
r.seconds = rf.seconds;
|
|
if (rf.ok) {
|
|
SetIfFinite(r.r_work, rf.restrained ? rf.r_work : rf.r_work_placement);
|
|
SetIfFinite(r.r_free, rf.restrained ? rf.r_free : rf.r_free_placement);
|
|
if (rf.restrained) {
|
|
SetIfFinite(r.rmsd_bonds_A, rf.geometry.rmsd_bonds);
|
|
SetIfFinite(r.rmsd_angles_deg, rf.geometry.rmsd_angles);
|
|
}
|
|
}
|
|
r.blob_count = static_cast<int64_t>(rf.blobs.size());
|
|
for (size_t i = 0; i < std::min(rf.blobs.size(), RugnuxLiveRefinement::MAX_BLOBS); i++) {
|
|
const DensityBlob &b = rf.blobs[i];
|
|
RugnuxLiveBlob blob{DensityBlobKindName(b.kind), b.centroid.x, b.centroid.y, b.centroid.z, b.peak_sigma,
|
|
b.volume, b.nearest_atom, std::nullopt};
|
|
SetIfFinite(blob.nearest_atom_distance_A, b.nearest_atom_distance); // infinite: no atom near
|
|
r.blobs.push_back(blob);
|
|
}
|
|
m.refinement = std::move(r);
|
|
}
|
|
return m;
|
|
}
|
|
|
|
JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_experiment,
|
|
const PixelMask &in_pixel_mask,
|
|
ImagePuller &in_image_puller,
|
|
ImagePusher &in_image_pusher,
|
|
Logger &in_logger,
|
|
int64_t in_forward_nthreads,
|
|
int in_rugnux_nthreads,
|
|
std::shared_ptr<const RugnuxModel> in_model,
|
|
double in_deadline_s,
|
|
RugnuxLiveStatus &in_live_status,
|
|
const SpotFindingSettings &in_spot_finding_settings,
|
|
JFJochReceiverCurrentStatus &in_current_status,
|
|
JFJochReceiverPlots &in_plots,
|
|
ImageBuffer &in_image_buffer,
|
|
ZMQPreviewSocket *in_zmq_preview_socket,
|
|
ZMQMetadataSocket *in_zmq_metadata_socket)
|
|
: JFJochReceiver(in_experiment, in_image_buffer, in_image_pusher, in_current_status, in_plots,
|
|
in_spot_finding_settings, in_logger, in_pixel_mask, in_zmq_preview_socket,
|
|
in_zmq_metadata_socket, nullptr),
|
|
image_puller(in_image_puller),
|
|
reader(in_image_buffer),
|
|
forward_nthreads(JFJochReceiverLite::NumberOfDataAnalysisThreads(in_forward_nthreads, in_experiment)),
|
|
rugnux_nthreads(in_rugnux_nthreads),
|
|
host_budget(RugnuxHostBudget(in_image_buffer)),
|
|
model(std::move(in_model)),
|
|
deadline_s(in_deadline_s),
|
|
live_status(in_live_status),
|
|
measurement_started(1),
|
|
forward_started(forward_nthreads) {
|
|
// GPU FFT indexers and cuFFT plans are kept for reuse for the life of the broker, not of one run as
|
|
// in `rugnux` (FFTIndexerReuse, FFTPlanReuse): every pass of every collection builds the same pools,
|
|
// and making a plan takes a process-wide lock and took over a second on a card busy with the online
|
|
// analysis. Never destroyed - at exit the CUDA runtime may already be gone.
|
|
[[maybe_unused]] static const FFTIndexerReuse *fft_indexers = new FFTIndexerReuse;
|
|
[[maybe_unused]] static const FFTPlanReuse *fft_plans = new FFTPlanReuse;
|
|
logger.Info("rugnux live: starting {} forward threads; rugnux runs on {} threads while the images arrive, "
|
|
"planning against {:.1f} GB of host memory", forward_nthreads, rugnux_nthreads,
|
|
static_cast<double>(host_budget) / 1e9);
|
|
for (int i = 0; i < forward_nthreads; i++)
|
|
forward_futures.emplace_back(std::async(std::launch::async, &JFJochReceiverRugnux::ForwardThread, this, i));
|
|
measurement = std::async(std::launch::async, &JFJochReceiverRugnux::MeasurementThread, this);
|
|
forward_started.wait();
|
|
}
|
|
|
|
JFJochReceiverRugnux::~JFJochReceiverRugnux() {
|
|
cancelled = true;
|
|
reader.Ended();
|
|
{
|
|
std::unique_lock ul(rugnux_m);
|
|
if (rugnux != nullptr)
|
|
rugnux->Cancel();
|
|
if (probe != nullptr)
|
|
probe->Cancel();
|
|
}
|
|
try {
|
|
if (measurement.valid())
|
|
measurement.get();
|
|
} catch (const std::exception &e) {
|
|
logger.Error("ReceiverRugnux teardown failed: {}", e.what());
|
|
}
|
|
}
|
|
|
|
void JFJochReceiverRugnux::MeasurementThread() {
|
|
std::future<void> rugnux_done, probe_done;
|
|
try {
|
|
// Waiting for the start message, as JFJochReceiverLite::MeasurementThread.
|
|
std::optional<std::chrono::steady_clock::time_point> give_up_at;
|
|
auto msg = image_puller.PollImage();
|
|
while (!cancelled && (!msg.has_value() || !msg->cbor || !msg->cbor->start_message)) {
|
|
if (detector_finished && !give_up_at)
|
|
give_up_at = std::chrono::steady_clock::now() + StartMessageGrace;
|
|
if (give_up_at && (std::chrono::steady_clock::now() > *give_up_at)) {
|
|
logger.Error("Detector finished, but no start message arrived - abandoning measurement");
|
|
cancelled = true;
|
|
break;
|
|
}
|
|
msg = image_puller.PollImage();
|
|
}
|
|
if (cancelled) {
|
|
current_status.SetProgress({});
|
|
current_status.SetStatus(GetStatus());
|
|
measurement_started.count_down();
|
|
return;
|
|
}
|
|
|
|
JFJochReceiverLite::ApplyStartMessage(experiment, msg->cbor->start_message.value());
|
|
image_buffer.StartMeasurement(experiment);
|
|
start_time = std::chrono::system_clock::now();
|
|
SendStartMessage();
|
|
|
|
// Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken
|
|
// again would hand it a later image. Without that the collection is only forwarded.
|
|
const auto slots = image_buffer.GetStatus().total_slots;
|
|
// The image time as the detector produces images, summation included
|
|
const double image_time_s = std::chrono::duration<double>(experiment.GetImageTime()).count();
|
|
std::optional<double> expected_collection_s;
|
|
if (image_time_s > 0)
|
|
expected_collection_s = static_cast<double>(experiment.GetImageNum()) * image_time_s;
|
|
live_status.Start(experiment.GetImageNum(), expected_collection_s, msg->received);
|
|
if (slots < experiment.GetImageNum()) {
|
|
skipped_warning = fmt::format("no rugnux analysis: the image buffer holds {} images, the collection "
|
|
"has {}", slots, experiment.GetImageNum());
|
|
logger.Error("{}", skipped_warning);
|
|
live_status.Finish(RugnuxLiveState::Skipped, "the image buffer cannot hold the whole collection");
|
|
} else
|
|
StartRugnux(rugnux_done, probe_done);
|
|
measurement_started.count_down();
|
|
} catch (const JFJochException &e) {
|
|
logger.ErrorException(e);
|
|
Cancel(e);
|
|
measurement_started.count_down();
|
|
throw;
|
|
} catch (const std::exception &e) {
|
|
logger.Error("Cancelling data collection due to exception: {}", e.what());
|
|
cancelled = true;
|
|
measurement_started.count_down();
|
|
throw;
|
|
}
|
|
|
|
logger.Info("Receiving started");
|
|
for (auto &f: forward_futures)
|
|
f.get();
|
|
reader.Ended();
|
|
images_done = true;
|
|
// The latency of the demonstration runs from the last image in (as the puller took it off the socket)
|
|
// to every result in place.
|
|
const auto last_image = live_status.LastImage();
|
|
logger.Info("Receiving finished, {} images", images_collected.load());
|
|
|
|
std::optional<double> done_s;
|
|
if (rugnux_done.valid()) {
|
|
// The fixed cutoff: the analysis gets deadline_s after the last image, then gives what it has.
|
|
if (deadline_s > 0 && rugnux_done.wait_for(std::chrono::duration<double>(deadline_s))
|
|
== std::future_status::timeout)
|
|
Cut("deadline");
|
|
rugnux_done.wait();
|
|
if (last_image)
|
|
done_s = SecondsSince(last_image.value());
|
|
}
|
|
if (probe_done.valid())
|
|
probe_done.wait();
|
|
|
|
current_status.SetProgress(1.0);
|
|
current_status.SetStatus(GetStatus());
|
|
|
|
SendEndMessage();
|
|
const double end_sent_s = last_image ? SecondsSince(last_image.value()) : 0.0;
|
|
|
|
if (!image_buffer.CheckIfBufferReturned(SendFinalizeTimeout)) {
|
|
logger.Error("Send commands not finalized in {} seconds", SendFinalizeTimeout.count());
|
|
throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Send commands not finalized in "
|
|
+ std::to_string(SendFinalizeTimeout.count()) + " seconds");
|
|
}
|
|
if (push_images_to_writer)
|
|
writer_error = image_pusher.Finalize();
|
|
if (rugnux_done.valid()) {
|
|
// The run's outcome, in one line: its results are in the live status, the end message and the files.
|
|
const auto live = live_status.Get();
|
|
std::string outcome = "rugnux live: " + StateName(live.state);
|
|
if (!live.message.empty())
|
|
outcome += " (" + live.message + ")";
|
|
if (!live.cut.empty())
|
|
outcome += " (cut: " + live.cut + ")";
|
|
if (live.start_to_first_image_s && live.last_image_s && live.images_received > 1)
|
|
outcome += fmt::format("; START -> first image {:.2f} s, {} images in {:.2f} s ({:.1f} Hz)",
|
|
live.start_to_first_image_s.value(), live.images_received,
|
|
live.last_image_s.value(),
|
|
static_cast<double>(live.images_received - 1) / live.last_image_s.value());
|
|
if (done_s) {
|
|
const double written_s = SecondsSince(last_image.value());
|
|
live_status.Latency(written_s);
|
|
outcome += fmt::format("; done {:.2f} s after the last image, last image -> all written {:.2f} s "
|
|
"(end message sent at {:.2f} s)", done_s.value(), written_s, end_sent_s);
|
|
}
|
|
logger.Info("{}", outcome);
|
|
}
|
|
|
|
end_time = std::chrono::system_clock::now();
|
|
current_status.SetProgress({});
|
|
current_status.SetStatus(GetStatus());
|
|
}
|
|
|
|
void JFJochReceiverRugnux::StartRugnux(std::future<void> &rugnux_done, std::future<void> &probe_done) {
|
|
// A failure here costs the analysis only: the images go on to the writer as they would without it.
|
|
try {
|
|
// The start message as the writer gets it, so rugnux sees the dataset an HDF5 master states.
|
|
std::vector<uint8_t> start_cbor;
|
|
image_buffer.GetStartMessage(start_cbor);
|
|
reader.Open(CBORStream2Deserialize(start_cbor)->start_message.value());
|
|
rugnux_done = std::async(std::launch::async, &JFJochReceiverRugnux::RugnuxThread, this);
|
|
probe_done = std::async(std::launch::async, &JFJochReceiverRugnux::ProbeThread, this);
|
|
} catch (const std::exception &e) {
|
|
logger.Error("rugnux live: cannot start - {}", e.what());
|
|
live_status.Finish(RugnuxLiveState::Failed, e.what());
|
|
}
|
|
}
|
|
|
|
void JFJochReceiverRugnux::RugnuxThread() {
|
|
try {
|
|
const auto dataset = reader.GetDataset();
|
|
RugnuxLiveSetup setup = MakeRugnuxLiveSetup(*dataset, rugnux_nthreads, model);
|
|
setup.config.host_memory_budget_bytes = host_budget;
|
|
// The cards are idle until the first images arrive: the indexers and plans the passes will
|
|
// need are built then, beside the run, which meanwhile waits for its first frames. Waited for
|
|
// when the run ends.
|
|
auto gpu_warm_up = std::async(std::launch::async, [this, setup] {
|
|
try {
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
WarmUpRugnuxLiveGPU(setup);
|
|
logger.Debug("rugnux live: FFT indexers and plans ready in {:.2f} s", std::chrono::duration<double>(
|
|
std::chrono::steady_clock::now() - t0).count());
|
|
} catch (const std::exception &e) {
|
|
logger.Warning("rugnux live: building the FFT indexers ahead failed, they are built when used - {}",
|
|
e.what());
|
|
}
|
|
});
|
|
LiveObserver observer(plots, scan_result, *az_int_mapping, live_status);
|
|
// The run goes - with the pinned buffers its engines kept - once its results are out (below).
|
|
std::optional<Rugnux> process;
|
|
process.emplace(reader, setup.experiment, *dataset->pixel_mask, setup.config);
|
|
std::optional<Reachable> reachable;
|
|
reachable.emplace(rugnux_m, rugnux, *process, cut, cancelled);
|
|
ProcessResult result = process->Run(&observer);
|
|
// The verdict and summary of the report a `rugnux` run writes, without writing it.
|
|
const ReportDocument doc = BuildReportDocument(experiment.GetFilePrefix(), experiment.GetFilePrefix(),
|
|
setup.experiment, result);
|
|
std::vector<std::string> warnings;
|
|
for (const auto &w: doc.warnings)
|
|
if (!w.developer_only)
|
|
warnings.push_back(w.text);
|
|
const double cc_anom = result.merge_statistics.overall.cc_anom;
|
|
live_status.Result(doc.verdict, doc.verdict_text, doc.summary_rows, warnings,
|
|
std::isfinite(cc_anom) && cc_anom > CC_ANOM_SIGNAL);
|
|
if (const auto m = LiveModel(result))
|
|
live_status.Model(*m);
|
|
// The numbers the summary's Space group, Resolution and Signal rows give
|
|
std::optional<double> d_min, isa;
|
|
if (result.has_merge_statistics)
|
|
SetIfFinite(d_min, result.merge_statistics.overall.d_min);
|
|
if (result.error_model_isa_resolved && result.error_model_isa > 0.0)
|
|
isa = result.error_model_isa;
|
|
live_status.Headline(result.space_group ? result.space_group->xhm() : "", d_min, isa);
|
|
// The placed model and a copy of its map coefficients (the end message takes the originals), and the
|
|
// refined model with its own, for the broker's viewer routes.
|
|
if (result.outputs && result.outputs->placed_model) {
|
|
const RugnuxOutputs &o = *result.outputs;
|
|
const PlacedModelOutput &p = *o.placed_model;
|
|
std::optional<MtzTable> coefficients;
|
|
if (o.maps)
|
|
coefficients = o.maps->coefficients;
|
|
std::shared_ptr<const RugnuxStructure> refined;
|
|
if (o.refined_model) {
|
|
std::optional<MtzTable> refined_coefficients;
|
|
if (o.refined_maps)
|
|
refined_coefficients = o.refined_maps->coefficients;
|
|
refined = std::make_shared<RugnuxStructure>(o.refined_model->model, o.refined_model->cell,
|
|
o.refined_model->space_group,
|
|
std::move(refined_coefficients));
|
|
}
|
|
live_status.Structure(std::make_shared<RugnuxStructure>(p.model, p.cell, p.space_group,
|
|
std::move(coefficients)),
|
|
std::move(refined));
|
|
}
|
|
live_status.Finish(result.cancelled ? RugnuxLiveState::Cancelled : RugnuxLiveState::Done);
|
|
reachable.reset();
|
|
process.reset();
|
|
Rugnux::ReleasePinnedPool();
|
|
std::unique_lock ul(rugnux_m);
|
|
rugnux_result = std::move(result);
|
|
} catch (const std::exception &e) {
|
|
logger.Error("rugnux live: {}", e.what());
|
|
live_status.Finish(RugnuxLiveState::Failed, e.what());
|
|
}
|
|
}
|
|
|
|
void JFJochReceiverRugnux::ProbeThread() {
|
|
// Rugnux waits for half the sweep before it decides on a lattice. In the meantime the machine is
|
|
// idle, so a small run of its own indexes the first 60 degrees and says at once whether there is
|
|
// a crystal - quick feedback for the person at the beamline, above all where there is none. It
|
|
// decides nothing for the run, which goes on as it would without it.
|
|
constexpr double PROBE_DEG = 60.0;
|
|
try {
|
|
const auto dataset = reader.GetDataset();
|
|
const auto gonio = dataset->experiment.GetGoniometer();
|
|
if (!gonio || gonio->GetIncrement_deg() == 0.0f)
|
|
return;
|
|
const int64_t frames = std::lround(PROBE_DEG / std::abs(gonio->GetIncrement_deg()));
|
|
if (2 * frames > static_cast<int64_t>(reader.GetNumberOfImages()))
|
|
return; // a sweep this short is decided on soon enough by the run itself
|
|
// A quarter of the threads: it runs beside the main run, and each of its workers holds GPU memory too.
|
|
RugnuxLiveSetup setup = MakeRugnuxLiveSetup(*dataset, std::max(2, rugnux_nthreads / 4), nullptr);
|
|
setup.config.host_memory_budget_bytes = host_budget; // the same as the main run's: one pool for both
|
|
setup.config.end_image = static_cast<int>(frames);
|
|
setup.config.rotation_postrefine_geometry = false;
|
|
setup.config.run_scaling = false;
|
|
// The file's centre only: the check captures the centre on the GPUs and runs a second first pass,
|
|
// both beside the main run's pre-scan, for a centre the probe does not report.
|
|
setup.config.beam_center_check = false;
|
|
ProcessResult r;
|
|
{
|
|
Rugnux probe_run(reader, setup.experiment, *dataset->pixel_mask, setup.config);
|
|
Reachable reachable(rugnux_m, probe, probe_run, cut, cancelled);
|
|
r = probe_run.Run(nullptr);
|
|
}
|
|
if (r.consensus_cell && r.rotation_lattice_type) {
|
|
const auto &c = *r.consensus_cell;
|
|
live_status.Probe(fmt::format("lattice on the first {:.0f} deg: {}-centred {}, {:.1f} {:.1f} {:.1f} "
|
|
"{:.1f} {:.1f} {:.1f}, {:.0f}% of the images indexed", PROBE_DEG,
|
|
r.rotation_lattice_type->centering,
|
|
gemmi::crystal_system_str(r.rotation_lattice_type->crystal_system),
|
|
c.a, c.b, c.c, c.alpha, c.beta, c.gamma,
|
|
100.0 * r.indexing_rate.value_or(0.0f)));
|
|
} else
|
|
live_status.Probe(fmt::format("no lattice on the first {:.0f} deg", PROBE_DEG));
|
|
} catch (const NoLatticeFound &e) {
|
|
live_status.Probe(fmt::format("no lattice on the first {:.0f} deg: {}", PROBE_DEG, e.reason));
|
|
} catch (const std::exception &e) {
|
|
logger.Warning("rugnux live: the lattice probe failed: {}", e.what());
|
|
}
|
|
logger.Debug("rugnux live: lattice probe - {}", live_status.Get().probe);
|
|
}
|
|
|
|
void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) {
|
|
// A failure here costs the end message the analysis only: it goes out as the receiver made it.
|
|
try {
|
|
std::unique_lock ul(rugnux_m);
|
|
if (!rugnux_result)
|
|
return;
|
|
const EndMessage &r = rugnux_result->end_message;
|
|
message.indexing_rate = r.indexing_rate;
|
|
message.bkg_estimate = r.bkg_estimate;
|
|
message.spindle_blind_fraction = r.spindle_blind_fraction;
|
|
message.ice_ring_score_mean = r.ice_ring_score_mean;
|
|
message.az_int_result = r.az_int_result;
|
|
message.rotation_lattice = r.rotation_lattice;
|
|
message.rotation_lattice_type = r.rotation_lattice_type;
|
|
message.rotation_extra_lattices = r.rotation_extra_lattices;
|
|
message.unit_cell = r.unit_cell;
|
|
message.unit_cell_esd = r.unit_cell_esd;
|
|
message.space_group_name = r.space_group_name;
|
|
message.space_group_number = r.space_group_number;
|
|
message.reindex_matrix = r.reindex_matrix;
|
|
message.refined_beam_center_x = r.refined_beam_center_x;
|
|
message.refined_beam_center_y = r.refined_beam_center_y;
|
|
message.refined_poni_rot1 = r.refined_poni_rot1;
|
|
message.refined_poni_rot2 = r.refined_poni_rot2;
|
|
message.refined_poni_rot3 = r.refined_poni_rot3;
|
|
message.refined_rotation_axis = r.refined_rotation_axis;
|
|
// Merged data, map coefficients and placed model, for the writer to write beside the master. Moved:
|
|
// the result is not read again.
|
|
if (!rugnux_result->outputs)
|
|
return;
|
|
RugnuxOutputs &o = *rugnux_result->outputs;
|
|
if (o.merged)
|
|
message.merged = std::move(o.merged->mtz);
|
|
if (o.maps)
|
|
message.map_coefficients = std::move(o.maps->coefficients);
|
|
if (o.placed_model)
|
|
message.placed_model = PlacedModelFromStructure(*o.placed_model->model, o.placed_model->cell,
|
|
o.placed_model->space_group.xhm());
|
|
// And the refinement's, as the CLI writes them (_refined_model.cif, _refined_maps.mtz, _blobs.json).
|
|
if (o.refined_model)
|
|
message.refined_model = PlacedModelFromStructure(*o.refined_model->model, o.refined_model->cell,
|
|
o.refined_model->space_group.xhm());
|
|
if (o.refined_maps)
|
|
message.refined_map_coefficients = std::move(o.refined_maps->coefficients);
|
|
if (o.blobs)
|
|
message.blobs = DensityBlobsToMessage(*o.blobs);
|
|
|
|
// Every pusher serialises END into a buffer of MESSAGE_SIZE_FOR_START_END, and an END that does
|
|
// not fit there fails the collection. So it is serialised here first, into a buffer of the same
|
|
// size (uninitialised: only the bytes written are touched); one that does not fit goes without
|
|
// the blocks.
|
|
const auto scratch = std::make_unique_for_overwrite<uint8_t[]>(MESSAGE_SIZE_FOR_START_END);
|
|
CBORStream2Serializer serializer(scratch.get(), MESSAGE_SIZE_FOR_START_END);
|
|
try {
|
|
serializer.SerializeSequenceEnd(message);
|
|
} catch (const JFJochException &e) {
|
|
const std::string text = fmt::format("the merged data, map coefficients and models do not "
|
|
"fit in the end message ({} MiB) and are not written",
|
|
MESSAGE_SIZE_FOR_START_END / (1024 * 1024));
|
|
logger.Error("rugnux live: {} - {}", text, e.what());
|
|
live_status.Warning(text);
|
|
message.merged.reset();
|
|
message.map_coefficients.reset();
|
|
message.placed_model.reset();
|
|
message.refined_model.reset();
|
|
message.refined_map_coefficients.reset();
|
|
message.blobs.reset();
|
|
}
|
|
} catch (const std::exception &e) {
|
|
logger.Error("rugnux live: the end message goes without the analysis results - {}", e.what());
|
|
message.merged.reset();
|
|
message.map_coefficients.reset();
|
|
message.placed_model.reset();
|
|
message.refined_model.reset();
|
|
message.refined_map_coefficients.reset();
|
|
message.blobs.reset();
|
|
}
|
|
}
|
|
|
|
void JFJochReceiverRugnux::ForwardThread(uint32_t id) {
|
|
forward_started.count_down();
|
|
measurement_started.wait();
|
|
|
|
while (!cancelled && !end_message_received) {
|
|
try {
|
|
auto msg = image_puller.PollImage(std::chrono::milliseconds(5));
|
|
if (!msg.has_value() || !msg->cbor)
|
|
continue;
|
|
if (msg->cbor->end_message) {
|
|
end_message_received = true;
|
|
} else if (msg->cbor->calibration) {
|
|
if (push_images_to_writer)
|
|
image_pusher.SendCalibration(msg->cbor->calibration.value());
|
|
} else if (msg->cbor->data_message) {
|
|
live_status.ImageReceived(msg->received);
|
|
DataMessage data_msg = msg->cbor->data_message.value();
|
|
++images_collected;
|
|
|
|
const auto compressed_size_img = data_msg.image.GetCompressedSize();
|
|
const auto uncompressed_size_img = data_msg.image.GetUncompressedSize();
|
|
compressed_size += compressed_size_img;
|
|
uncompressed_size += uncompressed_size_img;
|
|
UpdateMaxImageReceived(data_msg.number);
|
|
|
|
data_msg.original_number = data_msg.number;
|
|
data_msg.user_data = experiment.GetImageAppendix();
|
|
data_msg.run_number = experiment.GetRunNumber();
|
|
data_msg.run_name = experiment.GetRunName();
|
|
data_msg.receiver_buf_available = image_buffer.GetAvailSlots();
|
|
data_msg.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization();
|
|
if (compressed_size_img > 0)
|
|
data_msg.compression_ratio = static_cast<float>(uncompressed_size_img)
|
|
/ static_cast<float>(compressed_size_img);
|
|
|
|
auto loc = image_buffer.GetImageSlot();
|
|
if (loc == nullptr)
|
|
writer_queue_full = true;
|
|
else {
|
|
auto writer_buffer = (uint8_t *) loc->GetImage();
|
|
CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
|
|
if (JFJochReceiverLite::SerializeImageOrDrop(serializer, data_msg, *loc,
|
|
experiment.GetImageBufferLocationSize(), logger)) {
|
|
loc->SetImageNumber(data_msg.number);
|
|
loc->SetImageSize(serializer.GetBufferSize());
|
|
loc->SetIndexed(false);
|
|
loc->ReadyToSend();
|
|
|
|
if (zmq_preview_socket != nullptr)
|
|
zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());
|
|
if (zmq_metadata_socket != nullptr)
|
|
zmq_metadata_socket->AddDataMessage(data_msg);
|
|
if (push_images_to_writer) {
|
|
if (image_pusher.SendImage(*loc))
|
|
++images_sent;
|
|
} else
|
|
loc->release();
|
|
UpdateMaxImageSent(data_msg.number);
|
|
}
|
|
}
|
|
}
|
|
UpdateMaxDelay(image_puller.GetCurrentFifoUtilization());
|
|
current_status.SetProgress(GetProgress());
|
|
current_status.SetStatus(GetStatus());
|
|
} catch (const JFJochException &e) {
|
|
logger.ErrorException(e);
|
|
Cancel(e);
|
|
} catch (const std::exception &e) {
|
|
logger.Error("Cancelling data collection due to exception: {}", e.what());
|
|
cancelled = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
void JFJochReceiverRugnux::Cancel(bool silent) {
|
|
detector_finished = true;
|
|
// The silent cancel is the detector reading idle at the end of every collection: rugnux goes on.
|
|
if (silent)
|
|
return;
|
|
// /cancel is the one way in for the beamline, which cannot tell the collection from its analysis: it
|
|
// ends the collection while the images still arrive, as JFJochReceiverLite does, and in every case
|
|
// cuts the analysis, so the end message goes out at once with what rugnux has established.
|
|
if (!images_done)
|
|
JFJochReceiver::Cancel(silent);
|
|
Cut("cancel");
|
|
}
|
|
|
|
void JFJochReceiverRugnux::Cut(const std::string &reason) {
|
|
std::unique_lock ul(rugnux_m);
|
|
if (cut || rugnux_result)
|
|
return;
|
|
cut = true;
|
|
reader.Ended(); // no image is waited for any more
|
|
if (rugnux != nullptr)
|
|
rugnux->Cancel();
|
|
if (probe != nullptr)
|
|
probe->Cancel();
|
|
live_status.Cut(reason);
|
|
logger.Debug("rugnux live: analysis cut ({}) - the end message carries what is established", reason);
|
|
}
|
|
|
|
void JFJochReceiverRugnux::StopReceiver() {
|
|
if (measurement.valid()) {
|
|
measurement.get();
|
|
logger.Debug("Receiver stopped");
|
|
}
|
|
}
|
|
|
|
float JFJochReceiverRugnux::GetEfficiency() const {
|
|
if (experiment.GetFrameNum() == 0)
|
|
return 0;
|
|
return static_cast<float>(images_collected) / static_cast<float>(experiment.GetFrameNum());
|
|
}
|
|
|
|
JFJochReceiverOutput JFJochReceiverRugnux::GetFinalStatistics() const {
|
|
JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics();
|
|
ret.analysis_warning = skipped_warning;
|
|
// The receiver's own rotation indexer sees no image here: the lattice is rugnux's.
|
|
std::unique_lock ul(rugnux_m);
|
|
if (rugnux_result) {
|
|
const EndMessage &r = rugnux_result->end_message;
|
|
ret.scan_result.rotation_lattice = r.rotation_lattice;
|
|
if (r.rotation_lattice_type) {
|
|
ret.scan_result.rotation_crystal_system = r.rotation_lattice_type->crystal_system;
|
|
ret.scan_result.rotation_centering = r.rotation_lattice_type->centering;
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
float JFJochReceiverRugnux::GetProgress() const {
|
|
if (experiment.GetFrameNum() == 0)
|
|
return 0.0;
|
|
return static_cast<float>(max_image_number_received) / static_cast<float>(experiment.GetFrameNum());
|
|
}
|