diff --git a/broker/JFJochServices.cpp b/broker/JFJochServices.cpp index b5913a81..9f82919b 100644 --- a/broker/JFJochServices.cpp +++ b/broker/JFJochServices.cpp @@ -15,18 +15,35 @@ void JFJochServices::Start(const DiffractionExperiment& experiment, cannot_stop_detector = false; - if (receiver != nullptr) { - logger.Info(" ... receiver start"); - if (image_puller) - image_puller->ResumeAndClear(); - receiver->Start(experiment, pixel_mask, calibration, image_puller); - { - std::shared_lock ul(detector_mutex); - if (detector && !experiment.IsUsingInternalPacketGen()) { - logger.Info(" ... detector start"); - detector->Start(experiment); - } + if (receiver == nullptr) { + logger.Info(" Done!"); + return; + } + + logger.Info(" ... receiver start"); + if (image_puller) + image_puller->ResumeAndClear(); + receiver->Start(experiment, pixel_mask, calibration, image_puller); + + // From here the receiver is running asynchronously. If starting the detector fails, stop the + // receiver again so the service returns to idle and the run can be retried without + // re-initialising. The detector failure is then propagated (a critical detector error stays + // critical, so the broker still goes to the Error state). + try { + std::shared_lock ul(detector_mutex); + if (detector && !experiment.IsUsingInternalPacketGen()) { + logger.Info(" ... detector start"); + detector->Start(experiment); } + } catch (const std::exception &e) { + logger.Error(" ... detector failed to start ({}) - stopping receiver", e.what()); + receiver->Cancel(false); + try { + receiver->Stop(); + } catch (const std::exception &stop_error) { + logger.Warning("Receiver stop after failed start reported: {}", stop_error.what()); + } + throw; } logger.Info(" Done!"); @@ -123,12 +140,9 @@ JFJochServicesOutput JFJochServices::Stop() { } } - // A writer that broke mid-run (e.g. a lost connection) leaves a truncated file. - // Surface that as a failed acquisition instead of silently reporting success. - if (!ret.receiver_output.writer_err.empty()) - throw JFJochException(JFJochExceptionCategory::FileWriteError, - "Writer error, written data may be incomplete: " - + ret.receiver_output.writer_err); + // A writer that broke mid-run (e.g. a lost connection) leaves a truncated file. This is + // reported to the caller via receiver_output.writer_err (the state machine surfaces it as + // an error message) - it is not a detector fault, so it does not need re-initialisation. logger.Info(" ... finished with success"); } catch (const JFJochException &e) { @@ -146,7 +160,7 @@ JFJochServicesOutput JFJochServices::Stop() { throw JFJochException(*exception); if (detector_error) - throw JFJochException(JFJochExceptionCategory::Detector, "Error in detector operation"); + throw JFJochCriticalException("Error in detector operation"); return ret; } diff --git a/broker/JFJochStateMachine.cpp b/broker/JFJochStateMachine.cpp index 1343bbfb..df4cfa3e 100644 --- a/broker/JFJochStateMachine.cpp +++ b/broker/JFJochStateMachine.cpp @@ -7,6 +7,7 @@ #include "../preview/JFJochTIFF.h" #include "../common/CUDAWrapper.h" #include "../common/GitInfo.h" +#include "../common/JFJochException.h" JFJochStateMachine::JFJochStateMachine(const DiffractionExperiment& in_experiment, JFJochServices &in_services, @@ -361,10 +362,19 @@ void JFJochStateMachine::Start(const DatasetSettings &settings, bool async) { experiment.IncrementRunNumber(); + start_exception = nullptr; SetState(JFJochState::Busy, "Preparing measurement", BrokerStatus::MessageSeverity::Info); measurement = std::async(std::launch::async, &JFJochStateMachine::MeasurementThread, this); - if (!async) + if (!async) { c.wait(ul, [&]() { return state != JFJochState::Busy; }); + // A synchronous start propagates the failure to the caller. The state has already been set + // by MeasurementThread (Idle for an ordinary failure, Error for a critical detector fault). + if (start_exception) { + auto e = start_exception; + start_exception = nullptr; + std::rethrow_exception(e); + } + } } BrokerStatus JFJochStateMachine::WaitTillNotBusy(std::chrono::milliseconds timeout) { @@ -392,12 +402,26 @@ void JFJochStateMachine::MeasurementThread() { SetState(JFJochState::Measuring, "Measuring ...", BrokerStatus::MessageSeverity::Info); } c.notify_all(); - } catch (std::exception &e) { + } catch (const JFJochCriticalException &e) { + // Detector left in an undefined state - force re-initialisation via the Error state. + logger.Error("Critical error starting measurement: {}", e.what()); { std::unique_lock ul(m); SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error); + start_exception = std::current_exception(); + } + c.notify_all(); + return; + } catch (const std::exception &e) { + // Ordinary acquisition failure - the detector is still configured/calibrated, so return to + // Idle and let the user retry without re-initialising. services.Start has already stopped the + // receiver it launched. + logger.Error("Error starting measurement: {}", e.what()); + { + std::unique_lock ul(m); + SetState(JFJochState::Idle, e.what(), BrokerStatus::MessageSeverity::Error); + start_exception = std::current_exception(); } - services.Cancel(); c.notify_all(); return; } @@ -442,9 +466,17 @@ void JFJochStateMachine::MeasurementThread() { "Data collection without problems", BrokerStatus::MessageSeverity::Success); } - } catch (const std::exception &e) { + } catch (const JFJochCriticalException &e) { + // Detector faulted during the run - it needs re-initialisation, so go to the Error state. + logger.Error("Critical error finishing measurement: {}", e.what()); std::unique_lock ul(m); SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error); + } catch (const std::exception &e) { + // Receiver/writer problem - the data may be incomplete, but the detector is still usable, so + // return to Idle rather than forcing re-initialisation. + logger.Error("Error finishing measurement: {}", e.what()); + std::unique_lock ul(m); + SetState(JFJochState::Idle, e.what(), BrokerStatus::MessageSeverity::Error); } c.notify_all(); } diff --git a/broker/JFJochStateMachine.h b/broker/JFJochStateMachine.h index 97c681c1..40f49e5e 100644 --- a/broker/JFJochStateMachine.h +++ b/broker/JFJochStateMachine.h @@ -7,6 +7,7 @@ #include #include #include +#include #include "../common/DiffractionExperiment.h" #include "../jungfrau/JFCalibration.h" @@ -99,6 +100,8 @@ class JFJochStateMachine { PixelMask pixel_mask; int64_t current_detector_setup; // Lock only on change std::optional scan_result; + // Set by MeasurementThread when a synchronous Start fails, so Start() can rethrow to the caller + std::exception_ptr start_exception; mutable std::mutex calibration_statistics_mutex; std::vector calibration_statistics; diff --git a/common/JFJochException.h b/common/JFJochException.h index 4c132a5a..0f574b9d 100644 --- a/common/JFJochException.h +++ b/common/JFJochException.h @@ -155,6 +155,16 @@ public: : JFJochException(JFJochExceptionCategory::WrongDAQState, description) {} }; +// Thrown when an operation leaves a subsystem (typically the detector) in an undefined state that a +// simple retry cannot recover from - the broker must re-initialise. The state machine catches this +// separately and goes to the Error state, whereas ordinary acquisition failures return to Idle. +class JFJochCriticalException : public JFJochException { +public: + explicit JFJochCriticalException(const std::string &description, + JFJochExceptionCategory in_category = JFJochExceptionCategory::Detector) + : JFJochException(in_category, description) {} +}; + class PCIeDeviceException : public JFJochException { public: explicit PCIeDeviceException(const std::string &description) diff --git a/detector_control/DectrisDetectorWrapper.cpp b/detector_control/DectrisDetectorWrapper.cpp index 136fab3c..dff0cb18 100644 --- a/detector_control/DectrisDetectorWrapper.cpp +++ b/detector_control/DectrisDetectorWrapper.cpp @@ -48,8 +48,7 @@ void DectrisDetectorWrapper::CheckBusy() { void DectrisDetectorWrapper::CheckBusyOrError() { switch(GetState()) { case DetectorState::ERROR: - throw JFJochException(JFJochExceptionCategory::Detector, - "Detector in error state"); + throw JFJochCriticalException("Detector in error state"); case DetectorState::BUSY: case DetectorState::WAITING: throw JFJochException(JFJochExceptionCategory::Detector, diff --git a/detector_control/SLSDetectorWrapper.cpp b/detector_control/SLSDetectorWrapper.cpp index c8ad5c95..3d8e0bd5 100644 --- a/detector_control/SLSDetectorWrapper.cpp +++ b/detector_control/SLSDetectorWrapper.cpp @@ -217,6 +217,8 @@ void SLSDetectorWrapper::Start(const DiffractionExperiment& experiment) { det.setNumberOfTriggers(experiment.GetNumTriggers()); det.startDetector(); + } catch (const JFJochException &) { + throw; // already a categorised JFJoch exception (incl. critical detector error) - keep its type } catch (std::exception &e) { logger.ErrorException(e); throw JFJochException(JFJochExceptionCategory::Detector, e.what()); @@ -229,7 +231,7 @@ void SLSDetectorWrapper::InternalStop() { // Assume it is executed in try-catch! auto state = GetState(); if (state == DetectorState::ERROR) - throw JFJochException(JFJochExceptionCategory::Detector, "Detector in error state"); + throw JFJochCriticalException("Detector in error state"); else if ((state == DetectorState::BUSY) || (state == DetectorState::WAITING)) { try { det.stopDetector();