leonarski_fandClaude Opus 5
41983ca589
A writer refusing to overwrite an existing file was reported badly, and the
earlier fix only covered half of it. The commit message of a6a715703 states the
assumption that broke: "SendStartMessage precedes every std::async in the
receiver constructor". That is true of JFJochReceiverFPGA, where the start
message goes out from the constructor body, and false of JFJochReceiverLite,
where it goes out from MeasurementThread - launched by std::async from the
constructor, which waits only on data_analysis_started. On the DECTRIS path the
writer's refusal therefore cannot reach services.Start(); it arrives from
services.Stop(), i.e. the second try-block of the state machine's
MeasurementThread, which never touched start_exception. /wait_till_done answered
200 for a run that never wrote anything.
Two changes, in front of the problem and behind it.
In front: a PREFLIGHT frame (TCP protocol version 3 -> 4), sent to every writer
before the receiver is built and before the detector is armed. It carries the
same CBOR start message a START does and asks one question: could this run be
written? The writer runs FileWriter::Preflight - the first three statements of
the FileWriter constructor, stopping short of the only one that creates
anything - and answers with an ACK. Shared code, so a pre-flight that passes
cannot be contradicted by the start behind it. A refusal is not fatal: nothing
was started, the connection stays usable, and above all no series is left armed
waiting for a trigger that will not come. Both ends already reject a frame of
another protocol version, so the bump costs no compatibility.
The check now covers the data files, not only the master. A stale
_data_000001.h5 used to sail through and fail when the file was renamed into
place at the end of the run, after the whole dataset had been collected. The
writer holding the master file checks for all of them - they share a directory
by construction, the master linking them by relative name - so the siblings do
not race each other statting files the others are creating.
Behind: /wait_till_done maps an Idle state to 200 or 500 by the severity of the
end-of-run message, so a collection that lost packets, or whose writer could not
write, is no longer indistinguishable from a good one to a script. A cancelled
collection is not affected - MeasurementThread checks the cancel before it looks
at the completeness of the data, and reports it with Info severity - and a
warning is still a 200. start_exception is untouched.
The PREFLIGHT payload is built from FillMessage alone, which is what keeps it
small: the per-pixel arrays are added by SendStartMessage and are not in it.
Measured on a JUNGFRAU 9M, 1.4 kB against 543 kB for the start message. It is
deliberately the whole of FillMessage rather than the fields the check reads
today - a hand-picked subset would answer for a different set of file names the
day the check learns to read one more field, and would do it silently.
Three defects found on the way:
TCPImagePuller republished a PREFLIGHT as a real start message, announcing to
downstream consumers a run that was not going to happen.
TCPStreamPusher::run_number was a plain uint64_t written by the control thread
and read by every ACK thread; Preflight adds a second writer to it.
StreamWriter::CollectImages reprocessed the last frame whenever WaitForImage
returned false - run is only tested at the top of the loop - so an abort wrote
the last image a second time. Pre-existing; the new branch made it visible.
Verified against a running broker on the local configuration: a colliding master
file and a colliding data file are both refused by /start, /wait_until_running
and /wait_till_done with the writer's own message, and the log shows the receiver
was never started for either; a cancelled collection still answers 200.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016WmryXe8ASbNi632sUMfsa
553 lines
26 KiB
C++
553 lines
26 KiB
C++
// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#include <catch2/catch_all.hpp>
|
|
#include "../broker/JFJochStateMachine.h"
|
|
#include "../acquisition_device/HLSSimulatedDevice.h"
|
|
#include "../receiver/JFJochReceiverService.h"
|
|
|
|
using namespace std::literals::chrono_literals;
|
|
|
|
TEST_CASE("JFJochStateMachine_States") {
|
|
Logger logger("JFJochStateMachine_States");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF4M());
|
|
|
|
DatasetSettings setup;
|
|
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
|
|
|
|
REQUIRE_THROWS(state_machine.Start(setup));
|
|
REQUIRE_THROWS(state_machine.Pedestal());
|
|
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
|
|
REQUIRE_NOTHROW(state_machine.Deactivate());
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_State_Pedestal") {
|
|
Logger logger("JFJochStateMachine_State_Pedestal");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF4M());
|
|
|
|
DatasetSettings setup;
|
|
|
|
state_machine.DebugOnly_SetState(JFJochState::Calibration);
|
|
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Calibration);
|
|
|
|
REQUIRE_THROWS(state_machine.Start(setup));
|
|
REQUIRE_THROWS(state_machine.Pedestal());
|
|
REQUIRE_THROWS(state_machine.Initialize());
|
|
REQUIRE(state_machine.WaitTillMeasurementDone(std::chrono::milliseconds(1)).state == JFJochState::Calibration);
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_State_Measure") {
|
|
Logger logger("JFJochStateMachine_State_Measure");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF4M());
|
|
|
|
DatasetSettings setup;
|
|
|
|
state_machine.DebugOnly_SetState(JFJochState::Measuring);
|
|
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Measuring);
|
|
|
|
REQUIRE_THROWS(state_machine.Start(setup));
|
|
REQUIRE_THROWS(state_machine.Pedestal());
|
|
REQUIRE_THROWS(state_machine.Initialize());
|
|
REQUIRE(state_machine.WaitTillMeasurementDone(std::chrono::milliseconds(1)).state == JFJochState::Measuring);
|
|
|
|
DetectorSettings settings{};
|
|
REQUIRE_THROWS(state_machine.LoadDetectorSettings(settings));
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_State_Error") {
|
|
Logger logger("JFJochStateMachine_State_Error");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF4M());
|
|
|
|
DatasetSettings setup;
|
|
|
|
state_machine.DebugOnly_SetState(JFJochState::Error,
|
|
"msg1234",
|
|
BrokerStatus::MessageSeverity::Error);
|
|
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Error);
|
|
REQUIRE(state_machine.GetStatus().message.has_value());
|
|
REQUIRE(state_machine.GetStatus().message == "msg1234");
|
|
REQUIRE(state_machine.GetStatus().message_severity == BrokerStatus::MessageSeverity::Error);
|
|
|
|
state_machine.DebugOnly_SetState(JFJochState::Inactive, "msg3456", BrokerStatus::MessageSeverity::Info);
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
|
|
REQUIRE(state_machine.GetStatus().message.has_value());
|
|
REQUIRE(state_machine.GetStatus().message == "msg3456");
|
|
REQUIRE(state_machine.GetStatus().message_severity == BrokerStatus::MessageSeverity::Info);
|
|
|
|
state_machine.DebugOnly_SetState(JFJochState::Error);
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Error);
|
|
REQUIRE(!state_machine.GetStatus().message.has_value());
|
|
|
|
REQUIRE_THROWS(state_machine.Start(setup));
|
|
REQUIRE_THROWS(state_machine.Pedestal());
|
|
|
|
REQUIRE(state_machine.WaitTillMeasurementDone(std::chrono::milliseconds(1)).state == JFJochState::Error);
|
|
|
|
DetectorSettings settings;
|
|
REQUIRE_NOTHROW(state_machine.LoadDetectorSettings(settings));
|
|
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_NoDetectorSetup") {
|
|
Logger logger("JFJochStateMachine_NoDetectorSetup");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
REQUIRE_THROWS(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_AddDetectorSetup") {
|
|
Logger logger("JFJochStateMachine_AddDetectorSetup");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
DetectorSetup setup = DetJF4M();
|
|
state_machine.AddDetectorSetup(setup);
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_AddDetectorSetup_Gain") {
|
|
Logger logger("JFJochStateMachine_AddDetectorSetup_Gain");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
DetectorSetup setup = DetJF(4);
|
|
setup.LoadGain({"../../tests/test_data/gainMaps_M049.bin",
|
|
"../../tests/test_data/gainMaps_M049.bin",
|
|
"../../tests/test_data/gainMaps_M049.bin",
|
|
"../../tests/test_data/gainMaps_M049.bin"});
|
|
state_machine.AddDetectorSetup(setup);
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_AddDetectorSetup_Multiple") {
|
|
Logger logger("JFJochStateMachine_AddDetectorSetup_Multiple");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
REQUIRE_NOTHROW(state_machine.AddDetectorSetup(DetJF(4, 1, 0,0,false,"Det1", {"mx1", "mx2", "mx3", "mx4"})));
|
|
REQUIRE_NOTHROW(state_machine.AddDetectorSetup(DetEIGER(2, 1, 0, 0, false, "Det2", {"mx1", "mx2", "mx3", "mx4"})));
|
|
REQUIRE_NOTHROW(state_machine.AddDetectorSetup(DetDECTRIS(23,45, "Det3", {"mx5"})));
|
|
|
|
|
|
auto dl = state_machine.GetDetectorsList();
|
|
REQUIRE(dl.detector.size() == 3);
|
|
REQUIRE(dl.detector[0].description == "Det1");
|
|
REQUIRE(dl.detector[0].nmodules == 4);
|
|
REQUIRE(dl.detector[0].detector_type == DetectorType::JUNGFRAU);
|
|
REQUIRE(dl.detector[1].description == "Det2");
|
|
REQUIRE(dl.detector[1].nmodules == 2);
|
|
REQUIRE(dl.detector[1].detector_type == DetectorType::EIGER);
|
|
REQUIRE(dl.detector[2].description == "Det3");
|
|
REQUIRE(dl.detector[2].nmodules == 1);
|
|
REQUIRE(dl.detector[2].detector_type == DetectorType::DECTRIS);
|
|
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
REQUIRE(state_machine.Experiment().GetModulesNum() == 4);
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
|
|
REQUIRE_THROWS(state_machine.SelectDetector(7));
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
|
|
REQUIRE_NOTHROW(state_machine.SelectDetector(2));
|
|
REQUIRE(state_machine.Experiment().GetModulesNum() == 1);
|
|
REQUIRE(state_machine.Experiment().GetDetectorDescription() == "Det3");
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
|
|
}
|
|
|
|
TEST_CASE("JFJochStateMachine_LoadDetectorSettings_Error") {
|
|
Logger logger("JFJochStateMachine_LoadDetectorSettings_Error");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF4M());
|
|
|
|
DatasetSettings setup;
|
|
|
|
state_machine.DebugOnly_SetState(JFJochState::Idle);
|
|
|
|
DetectorSettings settings;
|
|
settings.FrameTime(std::chrono::microseconds(1));
|
|
|
|
REQUIRE_THROWS(state_machine.LoadDetectorSettings(settings));
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
}
|
|
|
|
namespace {
|
|
// Stands in for a writer that refuses to start the run - e.g. the output file exists and cannot be
|
|
// overwritten. The refusal reaches the broker as an ordinary exception thrown while the start
|
|
// message is being sent, which is what separates it from a critical detector fault.
|
|
class RefusingImagePusher : public ImagePusher {
|
|
public:
|
|
bool refuse = true;
|
|
void StartDataCollection(StartMessage &) override {
|
|
if (refuse)
|
|
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "writer_refused_1234");
|
|
}
|
|
bool EndDataCollection(const EndMessage &) override { return true; }
|
|
bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
|
|
bool SendCalibration(const CompressedImage &) override { return true; }
|
|
std::string PrintSetup() const override { return "RefusingImagePusher"; }
|
|
ImagePusherType GetType() const override { return ImagePusherType::Test; }
|
|
};
|
|
}
|
|
|
|
// An asynchronous start returns before the measurement thread runs, so a failure there has to be
|
|
// reported to whoever asks next. Without this the state is a plain Idle, indistinguishable from a
|
|
// timeout, and the writer's message is lost.
|
|
TEST_CASE("JFJochStateMachine_AsyncStartFailure") {
|
|
Logger logger("JFJochStateMachine_AsyncStartFailure");
|
|
|
|
DiffractionExperiment experiment(DetJF(2));
|
|
experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
|
|
.ImagesPerTrigger(4).IncidentEnergy_keV(12.4);
|
|
|
|
AcquisitionDeviceGroup aq_devices;
|
|
for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
|
|
aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));
|
|
|
|
RefusingImagePusher pusher;
|
|
JFJochReceiverService receiver_service(aq_devices, logger, pusher);
|
|
|
|
JFJochServices services(logger);
|
|
services.Receiver(&receiver_service);
|
|
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF(2));
|
|
state_machine.DebugOnly_SetState(JFJochState::Idle);
|
|
|
|
DatasetSettings setup;
|
|
setup.ImagesPerTrigger(4).NumTriggers(1);
|
|
|
|
// The asynchronous start itself succeeds - it only launches the measurement thread.
|
|
REQUIRE_NOTHROW(state_machine.Start(setup, true));
|
|
|
|
// Both wait functions report the failure, and repeatedly: the exception_ptr is not consumed.
|
|
REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(std::chrono::seconds(10)),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(std::chrono::seconds(10)),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
REQUIRE_THROWS_WITH(state_machine.WaitTillMeasurementDone(std::chrono::seconds(10)),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
|
|
// An ordinary failure leaves the detector usable, so the run can be retried from Idle. The
|
|
// status keeps the writer's message, and nothing is left of the run that never happened - a
|
|
// progress figure in particular would show up in the frontend as a stalled acquisition.
|
|
auto status = state_machine.GetStatus();
|
|
REQUIRE(status.state == JFJochState::Idle);
|
|
REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
|
|
REQUIRE_THAT(status.message.value_or(""), Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
REQUIRE_FALSE(status.progress.has_value());
|
|
|
|
auto statistics = state_machine.GetMeasurementStatistics();
|
|
REQUIRE(statistics.has_value());
|
|
REQUIRE(statistics->images_collected == 0);
|
|
REQUIRE(statistics->images_sent == 0);
|
|
REQUIRE_FALSE(statistics->collection_efficiency.has_value());
|
|
|
|
// A synchronous start reports the same failure directly, and leaves it visible to the wait
|
|
// functions afterwards.
|
|
REQUIRE_THROWS_WITH(state_machine.Start(setup),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(std::chrono::seconds(10)),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
|
|
// A start rejected before the measurement thread is launched must not leave the previous
|
|
// failure behind for the next wait call to report.
|
|
DatasetSettings rejected_setup;
|
|
rejected_setup.ImagesPerTrigger(4).NumTriggers(1)
|
|
.ImageTime(state_machine.Experiment().GetFrameTime() + std::chrono::nanoseconds(1));
|
|
REQUIRE_THROWS(state_machine.Start(rejected_setup, true));
|
|
REQUIRE_NOTHROW(state_machine.WaitTillNotBusy(std::chrono::seconds(10)));
|
|
|
|
// Any other operation started from Idle supersedes the pending failure - it must not be
|
|
// attributed to the pedestal, which is allowed from Idle and has nothing to do with it.
|
|
REQUIRE_THROWS_WITH(state_machine.Start(setup),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
REQUIRE_NOTHROW(state_machine.Pedestal());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
|
|
// Nothing is left half-started: once the writer accepts the run, the next start goes through
|
|
// and completes without any intervening re-initialisation.
|
|
pusher.refuse = false;
|
|
REQUIRE_NOTHROW(state_machine.Start(setup));
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Measuring);
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
REQUIRE(state_machine.GetMeasurementStatistics()->images_sent == 4);
|
|
|
|
// Re-initialising discards a pending failure too, so a detector brought back up does not report
|
|
// the failure of the run before it.
|
|
pusher.refuse = true;
|
|
REQUIRE_THROWS_WITH(state_machine.Start(setup),
|
|
Catch::Matchers::ContainsSubstring("writer_refused_1234"));
|
|
services.Receiver(nullptr);
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
}
|
|
|
|
// A calibration that does not complete must not be reported as a good one. Every step signals
|
|
// failure by calling SetState and returning normally, so nothing threw and the sequence used to
|
|
// finish with an unconditional "Calibration sequence done"/Success on top of it - leaving the
|
|
// broker Idle and apparently ready while holding partial pedestals.
|
|
TEST_CASE("JFJochStateMachine_CalibrationFailure") {
|
|
Logger logger("JFJochStateMachine_CalibrationFailure");
|
|
JFJochServices services(logger);
|
|
DiffractionExperiment experiment;
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF4M());
|
|
|
|
// Ask for a G0 pedestal. There is no receiver, so no frames come back and the import fails -
|
|
// the same outcome as a pedestal that collected nothing on real hardware.
|
|
DetectorSettings settings = state_machine.GetDetectorSettings();
|
|
settings.PedestalG0Frames(100);
|
|
REQUIRE_NOTHROW(state_machine.LoadDetectorSettings(settings));
|
|
|
|
REQUIRE_NOTHROW(state_machine.Initialize());
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
|
|
// Inactive, not Idle and not Error: the calibration is undefined, so the detector has to be
|
|
// initialized again rather than looking ready to measure.
|
|
auto status = state_machine.GetStatus();
|
|
REQUIRE(status.state == JFJochState::Inactive);
|
|
REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
|
|
REQUIRE(status.message == "Pedestal not collected properly");
|
|
|
|
// ... and a data collection is refused on it, instead of producing mis-converted images.
|
|
DatasetSettings setup;
|
|
REQUIRE_THROWS(state_machine.Start(setup));
|
|
|
|
// The success path is unchanged - with no pedestal to take, the sequence still reports success.
|
|
JFJochStateMachine calibrated(experiment, services, logger);
|
|
calibrated.AddDetectorSetup(DetJF4M());
|
|
REQUIRE_NOTHROW(calibrated.Initialize());
|
|
REQUIRE_NOTHROW(calibrated.WaitTillMeasurementDone());
|
|
REQUIRE(calibrated.GetStatus().state == JFJochState::Idle);
|
|
REQUIRE(calibrated.GetStatus().message_severity == BrokerStatus::MessageSeverity::Success);
|
|
}
|
|
|
|
namespace {
|
|
// Holds the run open until the test lets go, so a cancel lands while the collection is running
|
|
// instead of racing its end.
|
|
class GatedPusher : public ImagePusher {
|
|
std::atomic<bool> released{true};
|
|
public:
|
|
void Hold() { released = false; }
|
|
void Release() { released = true; }
|
|
void StartDataCollection(StartMessage &) override {}
|
|
bool EndDataCollection(const EndMessage &) override { return true; }
|
|
bool SendImage(const uint8_t *, size_t, int64_t) override {
|
|
while (!released)
|
|
std::this_thread::sleep_for(1ms);
|
|
return true;
|
|
}
|
|
bool SendCalibration(const CompressedImage &) override { return true; }
|
|
std::string PrintSetup() const override { return "GatedPusher"; }
|
|
ImagePusherType GetType() const override { return ImagePusherType::Test; }
|
|
};
|
|
|
|
// A writer that cannot finish the run - a rename that collides, a disk that filled up. It
|
|
// reaches the state machine from Stop(), not from Start(), which is the case that used to be
|
|
// reported as a run without problems.
|
|
class FinalizeFailingPusher : public ImagePusher {
|
|
public:
|
|
void StartDataCollection(StartMessage &) override {}
|
|
bool EndDataCollection(const EndMessage &) override { return true; }
|
|
bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
|
|
bool SendCalibration(const CompressedImage &) override { return true; }
|
|
std::string Finalize() override {
|
|
throw JFJochException(JFJochExceptionCategory::FileWriteError, "writer_finalize_failed_9876");
|
|
}
|
|
std::string PrintSetup() const override { return "FinalizeFailingPusher"; }
|
|
ImagePusherType GetType() const override { return ImagePusherType::Test; }
|
|
};
|
|
}
|
|
|
|
// /wait_till_done reads the severity of the end-of-run message to decide between 200 and 500, so
|
|
// what the state machine puts there is what a script sees. A cancelled collection must not be an
|
|
// error: it is incomplete by definition - the packets it never received would otherwise report
|
|
// themselves - and calling every cancel a failure would make the endpoint useless.
|
|
TEST_CASE("JFJochStateMachine_CancelledRunIsNotAnError") {
|
|
Logger logger("JFJochStateMachine_CancelledRunIsNotAnError");
|
|
|
|
DiffractionExperiment experiment(DetJF(2));
|
|
experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
|
|
.ImagesPerTrigger(8).IncidentEnergy_keV(12.4);
|
|
|
|
AcquisitionDeviceGroup aq_devices;
|
|
for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
|
|
aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));
|
|
|
|
GatedPusher pusher;
|
|
JFJochReceiverService receiver_service(aq_devices, logger, pusher);
|
|
|
|
JFJochServices services(logger);
|
|
services.Receiver(&receiver_service);
|
|
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF(2));
|
|
state_machine.DebugOnly_SetState(JFJochState::Idle);
|
|
|
|
DatasetSettings setup;
|
|
setup.ImagesPerTrigger(8).NumTriggers(1);
|
|
|
|
// A run that finished properly is a Success, so /wait_till_done answers 200.
|
|
REQUIRE_NOTHROW(state_machine.Start(setup));
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
|
|
REQUIRE(state_machine.GetStatus().message_severity == BrokerStatus::MessageSeverity::Success);
|
|
|
|
// The same run cancelled halfway. The gate keeps it running until the cancel is in.
|
|
pusher.Hold();
|
|
REQUIRE_NOTHROW(state_machine.Start(setup, true));
|
|
|
|
for (int i = 0; i < 400 && state_machine.GetStatus().state != JFJochState::Measuring; i++)
|
|
std::this_thread::sleep_for(25ms);
|
|
REQUIRE(state_machine.GetStatus().state == JFJochState::Measuring);
|
|
|
|
state_machine.Cancel();
|
|
pusher.Release();
|
|
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
auto status = state_machine.GetStatus();
|
|
REQUIRE(status.state == JFJochState::Idle);
|
|
// Info, not Error and not Warning: the collection did what it was told to do.
|
|
REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Info);
|
|
REQUIRE_THAT(status.message.value_or(""), Catch::Matchers::ContainsSubstring("cancelled"));
|
|
|
|
// The run really was short - what makes this test worth having is that the incompleteness did
|
|
// not get reported ahead of the cancel.
|
|
REQUIRE(state_machine.GetMeasurementStatistics()->cancelled);
|
|
}
|
|
|
|
// A writer that failed while the run was being finalized arrives from Stop(), which leaves the
|
|
// state Idle - the detector is still usable - and says so only in the severity. /wait_till_done
|
|
// turns that into a 500 with the writer's message rather than reporting a good collection.
|
|
TEST_CASE("JFJochStateMachine_WriterFailureAtEndIsAnError") {
|
|
Logger logger("JFJochStateMachine_WriterFailureAtEndIsAnError");
|
|
|
|
DiffractionExperiment experiment(DetJF(2));
|
|
experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
|
|
.ImagesPerTrigger(4).IncidentEnergy_keV(12.4);
|
|
|
|
AcquisitionDeviceGroup aq_devices;
|
|
for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
|
|
aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));
|
|
|
|
FinalizeFailingPusher pusher;
|
|
JFJochReceiverService receiver_service(aq_devices, logger, pusher);
|
|
|
|
JFJochServices services(logger);
|
|
services.Receiver(&receiver_service);
|
|
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF(2));
|
|
state_machine.DebugOnly_SetState(JFJochState::Idle);
|
|
|
|
DatasetSettings setup;
|
|
setup.ImagesPerTrigger(4).NumTriggers(1);
|
|
|
|
REQUIRE_NOTHROW(state_machine.Start(setup, true));
|
|
REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
|
|
|
|
auto status = state_machine.GetStatus();
|
|
REQUIRE(status.state == JFJochState::Idle);
|
|
REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
|
|
REQUIRE_THAT(status.message.value_or(""),
|
|
Catch::Matchers::ContainsSubstring("writer_finalize_failed_9876"));
|
|
}
|
|
|
|
namespace {
|
|
// Refuses the run before it starts, the way a writer refuses to overwrite a file that is
|
|
// already there. Records whether the collection was started anyway.
|
|
class PreflightRefusingPusher : public ImagePusher {
|
|
public:
|
|
std::atomic<bool> start_called{false};
|
|
void Preflight(StartMessage &) override {
|
|
throw JFJochException(JFJochExceptionCategory::FileWriteError, "preflight_refused_2468");
|
|
}
|
|
void StartDataCollection(StartMessage &) override { start_called = true; }
|
|
bool EndDataCollection(const EndMessage &) override { return true; }
|
|
bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
|
|
bool SendCalibration(const CompressedImage &) override { return true; }
|
|
std::string PrintSetup() const override { return "PreflightRefusingPusher"; }
|
|
ImagePusherType GetType() const override { return ImagePusherType::Test; }
|
|
};
|
|
}
|
|
|
|
// The pre-flight is the first thing a measurement does, so a run the writer will not accept is
|
|
// refused before the receiver is built and before the detector is armed - which is what makes the
|
|
// refusal free: there is no series left armed and waiting for a trigger to tear down.
|
|
TEST_CASE("JFJochStateMachine_PreflightRefusesBeforeStarting") {
|
|
Logger logger("JFJochStateMachine_PreflightRefusesBeforeStarting");
|
|
|
|
DiffractionExperiment experiment(DetJF(2));
|
|
experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
|
|
.ImagesPerTrigger(4).IncidentEnergy_keV(12.4).FilePrefix("preflight_refused");
|
|
|
|
AcquisitionDeviceGroup aq_devices;
|
|
for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
|
|
aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));
|
|
|
|
PreflightRefusingPusher pusher;
|
|
JFJochReceiverService receiver_service(aq_devices, logger, pusher);
|
|
|
|
JFJochServices services(logger);
|
|
services.Receiver(&receiver_service);
|
|
|
|
JFJochStateMachine state_machine(experiment, services, logger);
|
|
state_machine.AddDetectorSetup(DetJF(2));
|
|
state_machine.DebugOnly_SetState(JFJochState::Idle);
|
|
|
|
DatasetSettings setup;
|
|
setup.ImagesPerTrigger(4).NumTriggers(1).FilePrefix("preflight_refused");
|
|
|
|
// A synchronous start reports the writer's own message.
|
|
REQUIRE_THROWS_WITH(state_machine.Start(setup),
|
|
Catch::Matchers::ContainsSubstring("preflight_refused_2468"));
|
|
REQUIRE_FALSE(pusher.start_called);
|
|
|
|
// An ordinary failure, so the detector stays usable and the run can be retried from Idle.
|
|
auto status = state_machine.GetStatus();
|
|
REQUIRE(status.state == JFJochState::Idle);
|
|
REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
|
|
|
|
// The asynchronous start returns, and the wait functions report it.
|
|
REQUIRE_NOTHROW(state_machine.Start(setup, true));
|
|
REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(10s),
|
|
Catch::Matchers::ContainsSubstring("preflight_refused_2468"));
|
|
REQUIRE_THROWS_WITH(state_machine.WaitTillMeasurementDone(10s),
|
|
Catch::Matchers::ContainsSubstring("preflight_refused_2468"));
|
|
REQUIRE_FALSE(pusher.start_called);
|
|
}
|