leonarski_fandClaude Opus 5.5
48c14b6d1b
The live rugnux run wrote its whole processing log into the broker's journal - every
[Rugnux] line of the library (tables included) plus a "rugnux:" line per step and note -
about 300 lines per collection, all of them results the user already gets through
/rugnux/status, the end message and the written files.
- Logger::ServiceLevel(service, level): every Logger of that name made afterwards keeps only
that level and above. The library makes its own Logger("Rugnux") (and
Logger("ScaleOnTheFly")) wherever it logs, so this is the one switch: the broker sets them
to error, or to debug with the existing `verbose` setting. The per-Logger "Version" line goes
with it. The rugnux CLI routes every Logger to its own sinks (RouteAll) and never calls
this, so its console and _log.txt are unchanged.
- The live observer feeds steps and notes to the live status only.
- The run's outcome is one line at the end: state, START -> first image and frame rate,
done after the last image, last image -> all written. The probe result, the warm-up time
and the cut reason go to debug (the cut is in the outcome line).
- Per-collection step chatter goes to debug: receiver init/teardown, thread start/stop,
"Wait for receiver done" and friends, the per-image mean processing time. The JFJochBroker
logger now follows `verbose` too, so these are available again with it.
- Kept: Start request, measurement start, receiving started/finished, receiver efficiency
(and the per-module packet counts on loss), calibration sequence, all warnings and errors.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PXc1zD1yxHqaR1DGdCTHL4
584 lines
29 KiB
C++
584 lines
29 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"
|
|
|
|
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";
|
|
}
|
|
}
|
|
|
|
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);
|
|
ProcessResult result;
|
|
{
|
|
Rugnux process(reader, setup.experiment, *dataset->pixel_mask, setup.config);
|
|
Reachable reachable(rugnux_m, rugnux, process, cut, cancelled);
|
|
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);
|
|
live_status.Result(doc.verdict, doc.verdict_text, doc.summary_rows, warnings);
|
|
live_status.Finish(result.cancelled ? RugnuxLiveState::Cancelled : RugnuxLiveState::Done);
|
|
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.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());
|
|
|
|
// 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 placed model 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();
|
|
}
|
|
} 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();
|
|
}
|
|
}
|
|
|
|
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());
|
|
}
|