From 51bfdd2b35b2ff81d7cbe987de5125ae80b08d24 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Wed, 8 Jul 2026 15:48:01 +0200 Subject: [PATCH] Fix acquisition error-handling and rugnux CLI issues from branch review Broker acquisition failure paths: - Report a truncated writer output / missing packets as errors ahead of the queue-full warning, so an incomplete dataset is no longer masked as a "reduce frame rate" warning. - Route PCIe/FPGA hardware faults to the Error state: make PCIeDeviceException a JFJochCriticalException, and stop JFJochServices::Stop() from slicing the caught exception (capture/rethrow via exception_ptr) so the critical type survives to the state machine. - Stop the (possibly partially-armed) detector, not just the receiver, when a detector arm fails, so the next start does not see it BUSY. - Return structured Error_message JSON for bind_json operation failures (e.g. a failed synchronous /start), matching the no-arg endpoints. rugnux CLI: - Default -N to all hardware threads, resolved once after parsing so full analysis, --azint-only and --scale behave the same. - Parse numeric option arguments strictly (reject non-numeric input and trailing garbage) instead of atof/stod, which silently yielded 0 or aborted the process; require --wavelength > 0. - Emit -R attached (-R100) in the reproduced command line so it re-parses. - Compare the reference MTZ against the cell that actually drives the merge in the --scale path. - Correct the --scaling-high-resolution help text and de-duplicate the offline-output experiment setup. Co-Authored-By: Claude Fable 5 --- broker/JFJochBrokerHttp.cpp | 14 +++- broker/JFJochServices.cpp | 24 +++++-- broker/JFJochStateMachine.cpp | 14 ++-- common/JFJochException.h | 7 +- rugnux/RugnuxCommandLine.cpp | 4 +- tests/RugnuxTest.cpp | 5 +- tools/rugnux_cli.cpp | 117 ++++++++++++++++++++++++---------- 7 files changed, 133 insertions(+), 52 deletions(-) diff --git a/broker/JFJochBrokerHttp.cpp b/broker/JFJochBrokerHttp.cpp index a5f5516b..f54c6dab 100644 --- a/broker/JFJochBrokerHttp.cpp +++ b/broker/JFJochBrokerHttp.cpp @@ -155,13 +155,23 @@ void JFJochBrokerHttp::register_routes(httplib::Server &server) { auto bind_json = [this](auto method, auto model_tag) { return [this, method](const httplib::Request &req, httplib::Response &res) { using Model = decltype(model_tag); + Model v; try { - Model v; nlohmann::json::parse(req.body).get_to(v); v.validate(); + } catch (const std::exception &e) { + // Malformed or invalid request body. + auto [c, s] = handleParsingException(e); + send_plain(res, c, s); + return; + } + try { (this->*method)(v, res); } catch (const std::exception &e) { - auto [c, s] = handleParsingException(e); + // Operation failure (e.g. WrongDAQState, a failed synchronous start): return the + // structured Error_message JSON with a reason field, like the no-arg endpoints, so + // clients can distinguish the cause instead of receiving opaque plain text. + auto [c, s] = handleOperationException(e); send_plain(res, c, s); } }; diff --git a/broker/JFJochServices.cpp b/broker/JFJochServices.cpp index 2fec0ff4..569677c6 100644 --- a/broker/JFJochServices.cpp +++ b/broker/JFJochServices.cpp @@ -36,14 +36,21 @@ void JFJochServices::Start(const DiffractionExperiment& experiment, detector->Start(experiment); } } catch (const std::exception &e) { - logger.Error(" ... detector failed to start ({}) - stopping receiver", e.what()); - // Best-effort receiver cleanup - it must not replace the original detector exception (which - // may be critical), so swallow anything it throws and let the throw below re-raise e. + logger.Error(" ... detector failed to start ({}) - stopping detector and receiver", e.what()); + // Best-effort cleanup - it must not replace the original detector exception (which may be + // critical), so swallow anything it throws and let the throw below re-raise e. The detector + // may have partially armed before failing, so stop it too; otherwise it lingers in a BUSY + // state and the next start fails with "detector busy" until a manual re-initialisation. try { + { + std::shared_lock ul(detector_mutex); + if (detector) + detector->Stop(); + } receiver->Cancel(false); receiver->Stop(); } catch (const std::exception &stop_error) { - logger.Warning("Receiver stop after failed start reported: {}", stop_error.what()); + logger.Warning("Stop after failed start reported: {}", stop_error.what()); } throw; } @@ -96,7 +103,10 @@ void JFJochServices::On(DiffractionExperiment &x) { JFJochServicesOutput JFJochServices::Stop() { JFJochServicesOutput ret; - std::unique_ptr exception; + // Captured as an exception_ptr (not a copied JFJochException) so the dynamic type survives: a + // critical hardware fault (e.g. PCIeDeviceException) must stay critical when re-thrown, otherwise + // slicing it to a plain JFJochException would send the state machine to Idle instead of Error. + std::exception_ptr exception; bool detector_error = false; @@ -149,7 +159,7 @@ JFJochServicesOutput JFJochServices::Stop() { logger.Info(" ... finished with success"); } catch (const JFJochException &e) { logger.Error(" ... finished with error {}", e.what()); - exception = std::make_unique(e); + exception = std::current_exception(); } } else { logger.Info("No receiver - sleeping for 30 seconds"); @@ -165,7 +175,7 @@ JFJochServicesOutput JFJochServices::Stop() { throw JFJochCriticalException("Error in detector operation"); if (exception) - throw JFJochException(*exception); + std::rethrow_exception(exception); return ret; } diff --git a/broker/JFJochStateMachine.cpp b/broker/JFJochStateMachine.cpp index df4cfa3e..58ee4f52 100644 --- a/broker/JFJochStateMachine.cpp +++ b/broker/JFJochStateMachine.cpp @@ -445,11 +445,11 @@ void JFJochStateMachine::MeasurementThread() { image_mean_time.integration * 1e6, image_mean_time.processing * 1e6); - if (tmp_output.receiver_output.writer_queue_full_warning) - SetState(JFJochState::Idle, - "Stream receiver (writer or downstream analysis) cannot cope with data; reduce frame rate", - BrokerStatus::MessageSeverity::Warning); - else if (tmp_output.receiver_output.status.cancelled) + // Priority order matters. A cancel is checked first (it legitimately leaves efficiency < 1), + // then the hard errors (missing packets, truncated writer output). The queue-full warning is + // only the primary status when the run otherwise succeeded - it must not mask a real error + // by downgrading an incomplete/truncated dataset to a "reduce frame rate" warning. + if (tmp_output.receiver_output.status.cancelled) SetState(JFJochState::Idle, "Data collection cancelled", BrokerStatus::MessageSeverity::Info); @@ -461,6 +461,10 @@ void JFJochStateMachine::MeasurementThread() { SetState(JFJochState::Idle, tmp_output.receiver_output.writer_err, BrokerStatus::MessageSeverity::Error); + else if (tmp_output.receiver_output.writer_queue_full_warning) + SetState(JFJochState::Idle, + "Stream receiver (writer or downstream analysis) cannot cope with data; reduce frame rate", + BrokerStatus::MessageSeverity::Warning); else SetState(JFJochState::Idle, "Data collection without problems", diff --git a/common/JFJochException.h b/common/JFJochException.h index 0f574b9d..52b4b2d3 100644 --- a/common/JFJochException.h +++ b/common/JFJochException.h @@ -165,10 +165,13 @@ public: : JFJochException(in_category, description) {} }; -class PCIeDeviceException : public JFJochException { +// A PCIe/FPGA hardware fault leaves the acquisition device in an undefined state that a retry cannot +// recover from, so it is critical: the state machine forces re-initialisation (Error state) rather +// than returning to Idle as it does for ordinary acquisition failures. +class PCIeDeviceException : public JFJochCriticalException { public: explicit PCIeDeviceException(const std::string &description) - : JFJochException(JFJochExceptionCategory::PCIeError, description) { + : JFJochCriticalException(description, JFJochExceptionCategory::PCIeError) { msg += " (" + std::string(strerror(errno)) + ")"; } }; diff --git a/rugnux/RugnuxCommandLine.cpp b/rugnux/RugnuxCommandLine.cpp index 315bc4cd..18408bbe 100644 --- a/rugnux/RugnuxCommandLine.cpp +++ b/rugnux/RugnuxCommandLine.cpp @@ -108,7 +108,9 @@ std::string RugnuxCommandLine(const ProcessConfig &config, if (config.rotation_indexing) { if (config.two_pass_rotation) - add("-R", std::to_string(config.rotation_indexing_image_count)); + // -R takes an optional argument, which getopt only accepts attached (-R100), never as a + // separate token - so emit it joined or the copied command line will not re-parse. + args.push_back("-R" + std::to_string(config.rotation_indexing_image_count)); else args.emplace_back("--single-pass-rotation"); if (!config.reuse_rotation_spots) diff --git a/tests/RugnuxTest.cpp b/tests/RugnuxTest.cpp index 553810fb..aa859902 100644 --- a/tests/RugnuxTest.cpp +++ b/tests/RugnuxTest.cpp @@ -158,7 +158,10 @@ TEST_CASE("RugnuxCommandLine_Full", "[process]") { CHECK(cmd.find("-o run1") != std::string::npos); CHECK(cmd.find("-X fft") != std::string::npos); CHECK(cmd.find("-S 96") != std::string::npos); - CHECK(cmd.find("-R 30") != std::string::npos); + // -R takes an optional argument, so its value must be attached (-R30); a separate "-R 30" token + // would not re-parse (getopt would leave 30 as a positional and drop the count). + CHECK(cmd.find("-R30") != std::string::npos); + CHECK(cmd.find("-R 30") == std::string::npos); CHECK(cmd.find("/data/lyso_master.h5") != std::string::npos); } diff --git a/tools/rugnux_cli.cpp b/tools/rugnux_cli.cpp index 2f7448f3..bf6d56bc 100644 --- a/tools/rugnux_cli.cpp +++ b/tools/rugnux_cli.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include "../reader/JFJochHDF5Reader.h" @@ -37,7 +38,7 @@ void print_usage() { std::cout << "Usage rugnux {} " << std::endl; std::cout << "Options:" << std::endl; std::cout << " -o, --output-prefix Output file prefix (default: output)" << std::endl; - std::cout << " -N, --threads Number of threads (default: 1)" << std::endl; + std::cout << " -N, --threads Number of threads (default: all hardware threads)" << std::endl; std::cout << " -s, --start-image Start image number (default: 0)" << std::endl; std::cout << " -e, --end-image End image number (default: all)" << std::endl; std::cout << " -t, --stride Image stride (default: 1)" << std::endl; @@ -78,7 +79,7 @@ void print_usage() { std::cout << " --smooth-g[=deg] rot3d: smooth per-frame scale G over a deg-degree rotation range (XDS DELPHI-like) before the combine (default: 5 for rot3d; 0 = off)" << std::endl; std::cout << " -A, --anomalous Anomalous mode (don't merge Friedel pairs)" << std::endl; std::cout << " -B, --refine-bfactor Refine per image B-factor" << std::endl; - std::cout << " --scaling-high-resolution High resolution limit for spot finding (default: no limit)" << std::endl; + std::cout << " --scaling-high-resolution High resolution limit for scaling/merging (default: no limit)" << std::endl; std::cout << " --min-partiality Minimum partiality to accept reflection (default: 0.02)" << std::endl; std::cout << " --capture-uncertainty rot3d: systematic sigma ~num*(1-captured_fraction)*I on under-captured fulls (default: 1.0 for rot3d, 0 otherwise)" << std::endl; std::cout << " --mosaicity Diagnostic: fix the scaling mosaicity (deg) instead of the per-image seed" << std::endl; @@ -240,6 +241,30 @@ bool parse_float_strict(const std::string &t, float &out) { } }; +// Parse a required numeric option argument, or print a clear error and exit. getopt hands option +// arguments over as raw C strings; atof() silently returns 0 on non-numeric input (so a typo like +// "--beam-x 21OO" would move the beam to pixel 0) and std::stod() throws std::invalid_argument, +// which terminates the program. This rejects both, plus trailing garbage like "1.5foo". +double parse_double_arg(const char *arg, const char *option_name, Logger &logger) { + std::string s = arg ? arg : ""; + trim_in_place(s); + if (!s.empty()) { + try { + size_t idx = 0; + double out = std::stod(s, &idx); + if (idx == s.size()) + return out; + } catch (...) {} + } + logger.Error("Invalid numeric value for {}: '{}'", option_name, arg ? arg : ""); + print_usage(); + exit(EXIT_FAILURE); +} + +float parse_float_arg(const char *arg, const char *option_name, Logger &logger) { + return static_cast(parse_double_arg(arg, option_name, logger)); +} + bool parse_on_off(const char *arg, bool &out) { std::string s = arg ? arg : ""; std::transform(s.begin(), s.end(), s.begin(), @@ -335,6 +360,19 @@ std::optional parse_lattice_arg(const char *arg) { return CrystalLattice(vals); } +// Shared offline-output settings for the _process.h5 / reflection writer, used by both the --scale +// path and the full-analysis path (each then sets its own space group and images-per-trigger). +void configure_offline_output(DiffractionExperiment &experiment, const std::string &output_prefix) { + experiment.BitDepthImage(32).Compression(CompressionAlgorithm::BSHUF_LZ4); + experiment.FilePrefix(output_prefix); + experiment.Mode(DetectorMode::Standard); // full image analysis + experiment.PixelSigned(true); + experiment.OverwriteExistingFiles(true); + experiment.PolarizationFactor(0.99); + experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy); + experiment.NumTriggers(1); +} + namespace { std::atomic g_active_process{nullptr}; void handle_sigint(int) { @@ -358,7 +396,7 @@ int main(int argc, char **argv) { std::string input_file; std::string output_prefix = "output"; - int nthreads = 1; + int nthreads = 0; // 0 = auto: resolved to all hardware threads after parsing (see below) int start_image = 0; int end_image = -1; // -1 indicates process until end int image_stride = 1; @@ -613,16 +651,16 @@ int main(int argc, char **argv) { write_process_h5_flag = true; break; case OPT_SMOOTH_G: - smooth_g_deg_arg = optarg ? std::stod(optarg) : SMOOTH_G_DEFAULT_DEG; + smooth_g_deg_arg = optarg ? parse_double_arg(optarg, "--smooth-g", logger) : SMOOTH_G_DEFAULT_DEG; break; case OPT_MIN_PARTIALITY: - min_partiality = std::stod(optarg); + min_partiality = parse_double_arg(optarg, "--min-partiality", logger); break; case OPT_CAPTURE_UNCERTAINTY: - capture_uncertainty_arg = std::stod(optarg); + capture_uncertainty_arg = parse_double_arg(optarg, "--capture-uncertainty", logger); break; case OPT_MOSAICITY: - forced_mosaicity_arg = std::stod(optarg); + forced_mosaicity_arg = parse_double_arg(optarg, "--mosaicity", logger); break; case OPT_INTEGRATION_RADIUS: integration_radius_arg = optarg; @@ -634,13 +672,13 @@ int main(int argc, char **argv) { else { logger.Error("--integrator expects boxsum|gaussian|empirical"); return 1; } break; case OPT_REJECT_OUTLIERS: - outlier_reject_nsigma = std::stod(optarg); + outlier_reject_nsigma = parse_double_arg(optarg, "--reject-outliers", logger); break; case OPT_REJECT_DELTA_CCHALF: - delta_cchalf_nsigma = std::stod(optarg); + delta_cchalf_nsigma = parse_double_arg(optarg, "--reject-delta-cchalf", logger); break; case OPT_MIN_IMAGE_CC: - min_image_cc = std::stod(optarg); + min_image_cc = parse_double_arg(optarg, "--min-image-cc", logger); break; case OPT_SCALING_HIGH_RESOLUTION: d_min_scale_merge = atof(optarg); @@ -690,13 +728,23 @@ int main(int argc, char **argv) { solid_angle_correction = value; break; } - case OPT_BEAM_X: beam_x = atof(optarg); break; - case OPT_BEAM_Y: beam_y = atof(optarg); break; - case OPT_DETECTOR_DISTANCE: detector_distance_mm = atof(optarg); break; - case OPT_WAVELENGTH: wavelength_A = atof(optarg); break; - case OPT_ROT1: rot1_rad = atof(optarg); break; - case OPT_ROT2: rot2_rad = atof(optarg); break; - case OPT_POLARIZATION: polarization_factor = atof(optarg); break; + case OPT_BEAM_X: beam_x = parse_float_arg(optarg, "--beam-x", logger); break; + case OPT_BEAM_Y: beam_y = parse_float_arg(optarg, "--beam-y", logger); break; + case OPT_DETECTOR_DISTANCE: detector_distance_mm = parse_float_arg(optarg, "--detector-distance", logger); break; + case OPT_WAVELENGTH: { + // Guard > 0: wavelength is used as a divisor (WVL_1A_IN_KEV / wavelength) below, and a 0 + // would produce a non-finite incident energy that throws unguarded and aborts the process. + float w = parse_float_arg(optarg, "--wavelength", logger); + if (!(w > 0.0f)) { + logger.Error("Invalid wavelength (must be > 0 A): {}", optarg); + exit(EXIT_FAILURE); + } + wavelength_A = w; + break; + } + case OPT_ROT1: rot1_rad = parse_float_arg(optarg, "--rot1", logger); break; + case OPT_ROT2: rot2_rad = parse_float_arg(optarg, "--rot2", logger); break; + case OPT_POLARIZATION: polarization_factor = parse_float_arg(optarg, "--polarization", logger); break; case OPT_SCALING_ITERATIONS: scaling_iter = atoi(optarg); if (scaling_iter <= 0) { @@ -727,6 +775,15 @@ int main(int argc, char **argv) { input_file = argv[optind]; logger.Verbose(verbose); + // -N defaults to 0 = "use all hardware threads"; resolve it to a concrete count here so every mode + // behaves the same. The scale/merge engines expand 0 on their own, but the per-image processing + // loop (Rugnux) spawns exactly nthreads workers, so passing 0 there would spawn none and process + // nothing - hence resolving it centrally rather than relying on each consumer. + if (nthreads <= 0) { + unsigned int hw = std::thread::hardware_concurrency(); + nthreads = hw > 0 ? static_cast(hw) : 1; + } + if (azint_only && scale_only) { logger.Error("--azint-only and --scale are mutually exclusive"); exit(EXIT_FAILURE); @@ -786,8 +843,14 @@ int main(int argc, char **argv) { logger.Info("Reference space group: {} (number {})", reference.space_group_name, reference.space_group_number.value_or(0)); + // Check the reference against the cell that will actually drive the merge. --scale merges + // in the cell stored in the input file (as the former jfjoch_scale did); the -C override + // only takes effect on the full-analysis path, which otherwise determines its cell later by + // indexing (unknown here, so nothing can be checked yet). + const std::optional data_cell = + scale_only ? dataset->experiment.GetUnitCell() : fixed_reference_unit_cell; const auto warning = ReferenceConsistencyWarning( - reference, fixed_reference_unit_cell, + reference, data_cell, space_group_number.has_value() ? std::optional(static_cast(*space_group_number)) : std::nullopt); if (!warning.empty()) @@ -806,17 +869,10 @@ int main(int argc, char **argv) { auto reflections = reader.ReadReflections(start_image, last_image); DiffractionExperiment experiment(dataset->experiment); - experiment.BitDepthImage(32).Compression(CompressionAlgorithm::BSHUF_LZ4); - experiment.FilePrefix(output_prefix); - experiment.Mode(DetectorMode::Standard); - experiment.PixelSigned(true); - experiment.OverwriteExistingFiles(true); - experiment.PolarizationFactor(0.99); - experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy); + configure_offline_output(experiment, output_prefix); // Keep the space group stored in the input file (written by the full pipeline) unless -S overrides. if (space_group_number.has_value()) experiment.SpaceGroupNumber(space_group_number); - experiment.NumTriggers(1); // A rotation (goniometer) dataset uses RotationScaleMerge unless --force-still asks for stills scaling. IndexingSettings indexing_settings; @@ -1000,16 +1056,9 @@ int main(int argc, char **argv) { return 0; } - experiment.BitDepthImage(32).Compression(CompressionAlgorithm::BSHUF_LZ4); - experiment.FilePrefix(output_prefix); - experiment.Mode(DetectorMode::Standard); // Ensure full image analysis - experiment.PixelSigned(true); - experiment.OverwriteExistingFiles(true); - experiment.PolarizationFactor(0.99); - experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy); + configure_offline_output(experiment, output_prefix); experiment.SpaceGroupNumber(space_group_number); experiment.ImagesPerTrigger(images_to_process); - experiment.NumTriggers(1); // Re-determine the unit cell from scratch: discard any cell stored in the input file so // indexing is not biased by it. A stale or wrong stored cell otherwise resolves the indexing