From 41983ca589194cb572061686f33cf4d9d844c7fe Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Thu, 27 Aug 2026 12:20:29 +0200 Subject: [PATCH] Broker: refuse a run the writer cannot write before arming the detector 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) Claude-Session: https://claude.ai/code/session_016WmryXe8ASbNi632sUMfsa --- broker/JFJochBrokerHttp.cpp | 35 +++-- broker/JFJochServices.cpp | 6 + broker/JFJochServices.h | 4 + broker/JFJochStateMachine.cpp | 5 + broker/jfjoch_api.yaml | 9 +- common/JfjochTCP.h | 8 +- docs/CHANGELOG.md | 4 + docs/IMAGE_STREAM.md | 23 ++- image_puller/TCPImagePuller.cpp | 4 +- image_pusher/HDF5FilePusher.cpp | 5 + image_pusher/HDF5FilePusher.h | 1 + image_pusher/ImagePusher.h | 5 + image_pusher/TCPStreamPusher.cpp | 67 +++++++- image_pusher/TCPStreamPusher.h | 7 +- receiver/JFJochReceiverService.cpp | 18 +++ receiver/JFJochReceiverService.h | 3 + tests/CMakeLists.txt | 1 + tests/HDF5WritingTest.cpp | 23 ++- tests/JFJochStateMachineTest.cpp | 195 +++++++++++++++++++++++ tests/PreflightTest.cpp | 242 +++++++++++++++++++++++++++++ writer/FileWriter.cpp | 56 +++++-- writer/FileWriter.h | 8 +- writer/StreamWriter.cpp | 38 +++++ writer/StreamWriter.h | 1 + 24 files changed, 723 insertions(+), 45 deletions(-) create mode 100644 tests/PreflightTest.cpp diff --git a/broker/JFJochBrokerHttp.cpp b/broker/JFJochBrokerHttp.cpp index 6117fe33..dbb9196b 100644 --- a/broker/JFJochBrokerHttp.cpp +++ b/broker/JFJochBrokerHttp.cpp @@ -74,6 +74,15 @@ namespace { return "application/octet-stream"; } + std::string error_message_json(const std::string &msg, const std::string &reason) { + Error_message m; + m.setMsg(msg); + m.setReason(reason); + nlohmann::json j; + to_json(j, m); + return j.dump(); + } + inline bool read_file_to_string(const std::string &path, std::string &out) { std::ifstream f(path, std::ios::binary); if (!f) @@ -120,19 +129,9 @@ std::pair JFJochBrokerHttp::handleOperationException(const std try { throw; } catch (const WrongDAQStateException &) { - Error_message msg; - msg.setMsg(ex.what()); - msg.setReason("WrongDAQState"); - nlohmann::json j; - to_json(j, msg); - return {500, j.dump()}; + return {500, error_message_json(ex.what(), "WrongDAQState")}; } catch (const std::exception &) { - Error_message msg; - msg.setMsg(ex.what()); - msg.setReason("Other"); - nlohmann::json j; - to_json(j, msg); - return {500, j.dump()}; + return {500, error_message_json(ex.what(), "Other")}; } } @@ -443,6 +442,18 @@ void JFJochBrokerHttp::wait_till_done_post(const std::optional &timeout switch (status.state) { case JFJochState::Idle: + // An operation that finished badly leaves the state Idle - the detector is still + // usable - and says so only through the message severity. Reporting that as a plain + // 200 makes a run that lost packets, or one whose writer could not write, look to a + // script exactly like a good one. A cancelled collection does not land here: the + // cancel is reported with Info severity, and MeasurementThread checks it before it + // looks at the completeness of the data, so the incomplete run a cancel leaves behind + // is never reported as an error. A Warning - a receiver that could not keep up - is + // still a 200: the data is there, and the run is done. + if (status.message_severity == BrokerStatus::MessageSeverity::Error) { + send_plain(response, 500, error_message_json(status.message.value_or("Unknown error"), "Other")); + return; + } response.status = 200; break; case JFJochState::Inactive: diff --git a/broker/JFJochServices.cpp b/broker/JFJochServices.cpp index 569677c6..baa81aaf 100644 --- a/broker/JFJochServices.cpp +++ b/broker/JFJochServices.cpp @@ -8,6 +8,12 @@ JFJochServices::JFJochServices(Logger &in_logger) : logger(in_logger) {} +void JFJochServices::Preflight(const DiffractionExperiment& experiment) { + if (receiver == nullptr) + return; + receiver->Preflight(experiment); +} + void JFJochServices::Start(const DiffractionExperiment& experiment, const PixelMask &pixel_mask, const JFCalibration *calibration) { diff --git a/broker/JFJochServices.h b/broker/JFJochServices.h index 88a6a8b8..99ddc23d 100644 --- a/broker/JFJochServices.h +++ b/broker/JFJochServices.h @@ -28,6 +28,10 @@ public: void On(DiffractionExperiment& experiment); void Off(); void ConfigureDetector(const DiffractionExperiment& experiment); + // Refuse a run the writer could not write, while refusing it is still free: this runs before + // the receiver is created and before the detector is armed, so a rejected run leaves nothing + // to tear down. Does nothing for transports without a back-channel (ZeroMQ). + void Preflight(const DiffractionExperiment& experiment); void Start(const DiffractionExperiment& experiment, const PixelMask &pixel_mask, const JFCalibration *calibration = nullptr); diff --git a/broker/JFJochStateMachine.cpp b/broker/JFJochStateMachine.cpp index a701fdc8..218f1434 100644 --- a/broker/JFJochStateMachine.cpp +++ b/broker/JFJochStateMachine.cpp @@ -434,6 +434,11 @@ PixelMaskStatistics JFJochStateMachine::GetPixelMaskStatistics() const { void JFJochStateMachine::MeasurementThread() { try { + // Before anything is started: ask the writer whether it could write this run. A run that + // would be refused for a file already there, or a directory that cannot be made, is + // refused now - with no receiver built and, above all, no detector armed, so there is + // nothing to tear down and no series left waiting for a trigger that will not come. + services.Preflight(experiment); services.SetSpotFindingSettings(GetSpotFindingSettings()); services.Start(experiment, pixel_mask, calibration.get()); { diff --git a/broker/jfjoch_api.yaml b/broker/jfjoch_api.yaml index f3533315..6130d725 100644 --- a/broker/jfjoch_api.yaml +++ b/broker/jfjoch_api.yaml @@ -2633,11 +2633,16 @@ paths: Extending timeout is possible, but requires to ensure safety that client will not close the connection and retry the connection. responses: "200": - description: Detector in `Idle` state, another data collection can start immediately + description: | + Detector in `Idle` state, another data collection can start immediately. + The operation that finished either succeeded, was cancelled, or ended with a warning. "400": description: Timeout parameter out of bounds "500": - description: Error within Jungfraujoch code - see output message. + description: | + Error within Jungfraujoch code, or the operation that finished ended in an error - + missing packets, or a writer that could not write - see output message. + A cancelled data collection is not an error. content: application/json: schema: diff --git a/common/JfjochTCP.h b/common/JfjochTCP.h index b74d4d79..267dd973 100644 --- a/common/JfjochTCP.h +++ b/common/JfjochTCP.h @@ -6,7 +6,7 @@ #include constexpr uint32_t JFJOCH_TCP_MAGIC = 0x4A464A54; // JFJT -constexpr uint32_t JFJOCH_TCP_VERSION = 3; +constexpr uint32_t JFJOCH_TCP_VERSION = 4; // Upper bound on a single frame's payload. A receiver rejects any header claiming more // before allocating for it: an uncapped payload_size would otherwise drive a huge @@ -26,6 +26,12 @@ enum class TCPFrameType : uint16_t { // waiting for a proper ACK rather than declaring the connection dead. Carries the // writer's current FIFO occupancy in the ack_fifo_* fields. BUSY = 8, + // Pusher -> writer dry run, sent before the detector is armed. Carries the same CBOR + // StartMessage a START would, but the writer only checks that it could write the output + // (path, directory, no file in the way) and answers with an ACK; it opens nothing and + // leaves its state untouched. A rejected PREFLIGHT is not fatal - the run was never + // started, so the connection stays usable for the next attempt. + PREFLIGHT = 9, }; enum class TCPAckCode : uint16_t { diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 3771378d..a3e5a511 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -17,6 +17,10 @@ This is an UNSTABLE release. It includes many experimental features, as well as * The PCIe driver DKMS package builds for the kernel it is being installed for instead of the running one, so a module built while a kernel update is being applied loads after the reboot. * The PCIe driver builds on RHEL 9.5 and later, and on their CentOS Stream, Rocky and AlmaLinux equivalents, where the `vm_flags` kernel interface was backported into the 5.14 kernel. * A data collection started with `async_start` that fails to start - a writer refusing to overwrite an existing file, for instance - is reported as an error by `/wait_until_running` and `/wait_till_done` instead of as a timeout and a successful collection respectively. The error message is the one the writer gave. +* `/start` asks the writer whether the run can be written before the detector is armed, so a run whose output file already exists, or whose output directory cannot be created, is refused up front with the writer's own message instead of failing once the detector is running. This needs the TCP image stream or the built-in HDF5 writer; the ZeroMQ stream has no way to answer and is unchanged. +* An output data file already in the way is refused when the collection starts, not when the file is renamed into place at the end of it. +* `/wait_till_done` answers 500 with the message when a collection ended in an error - packets missing, or a writer that could not write - instead of 200. A cancelled collection and a collection that only triggered a warning still answer 200. +* The TCP image stream protocol version is 4. `jfjoch_writer` and `jfjoch_broker` have to be of the same release, as before. * A calibration that is cancelled or that fails to collect its pedestals is no longer reported as a successful one. The broker goes to `Inactive` with an error message and has to be initialized again, instead of sitting in `Idle` looking ready to measure while holding partial pedestals - data collected in that state was silently mis-converted. * A failed `/initialize` is reported to `/wait_until_running` and `/wait_till_done` as soon as it happens, instead of when their timeout expires. * `space_group_number` accepts space groups up to 230 in the API schema, so cubic space groups can be recorded. The broker always accepted them; the generated clients rejected them before the request was sent. diff --git a/docs/IMAGE_STREAM.md b/docs/IMAGE_STREAM.md index f445eb92..02c4284a 100644 --- a/docs/IMAGE_STREAM.md +++ b/docs/IMAGE_STREAM.md @@ -60,19 +60,29 @@ For TCP/IP image stream, Jungfraujoch **listens** on a single TCP port and all w Using `*` as port number (e.g. `tcp://127.0.0.1:*`) is supported — the OS assigns a free port and the actual bound address can be queried via `GetAddress()`. -Payloads for `START`, `DATA`, `CALIBRATION` and `END` frames are CBOR messages, equivalent in content to the ZeroMQ image stream messages. -`ACK`, `CANCEL`, and `KEEPALIVE` are control frames (no CBOR payload). +Payloads for `PREFLIGHT`, `START`, `DATA`, `CALIBRATION` and `END` frames are CBOR messages, equivalent in content to the ZeroMQ image stream messages. +`ACK`, `CANCEL`, `KEEPALIVE` and `BUSY` are control frames (no CBOR payload). The data collection lifecycle on each connection follows: -`START` → `CALIBRATION` (socket 0 only) → `DATA` (repeated) → `END` +`PREFLIGHT` → `START` → `CALIBRATION` (socket 0 only) → `DATA` (repeated) → `END` If a `START` ACK fails on any connection, Jungfraujoch sends `CANCEL` to all already-started connections and rolls back. For each frame: 1. Read one `TcpFrameHeader` (fixed size, 64-byte aligned). -2. Validate `magic` (`0x4A464A54` / `"JFJT"`) and `version` (`2`). +2. Validate `magic` (`0x4A464A54` / `"JFJT"`) and `version` (`4`). Both ends reject a frame of any other version, so writer and broker must be of the same release. 3. Read `payload_size` bytes (if non-zero). +#### Pre-flight + +Before a data collection is started - before the detector is armed - Jungfraujoch sends a `PREFLIGHT` frame on every connection and waits for its ACK. It carries the same CBOR start message a `START` would, and asks the writer one question: could this run be written? The writer checks the output path, creates the output directory, and checks that no output file is in the way; it opens nothing, writes nothing, and does not change its state. A run that would fail on the first file it wrote is therefore refused while refusing it is free, with the writer's own message reported to the client by `/start` and `/wait_until_running`. + +`write_master_file` is assigned exactly as it is for `START` (connection index 0), and the writer that owns the master file is the one that answers for the output files - the master and every data file the run will write, its siblings' included. + +The `PREFLIGHT` payload describes the run but carries none of the per-pixel arrays a `START` does - no pixel mask, no azimuthal-integration map, no ROI map - so it stays small: about 1.4 kB on a JUNGFRAU 9M, against about 540 kB for the `START` that follows. + +A rejected `PREFLIGHT` is **not** fatal: nothing was started, so the connection stays usable and the next attempt (a different file prefix, or `overwrite` set) proceeds on it. The check cannot be exhaustive - a file created in the moment between the pre-flight and the start still fails at the start, and free space and quota are not inspected - so it lowers how often a run fails on its output, it does not remove the case. + When image stream is split into multiple connections: - `START` and `END` are sent on all connections, - `CALIBRATION` is sent only on connection 0, @@ -96,6 +106,7 @@ TCP/IP image stream is configured in the broker JSON configuration file under th #### ACK handling ACK handling is mandatory for correct operation: +- `PREFLIGHT` **must** be acknowledged (`ack_for=PREFLIGHT`) on each connection within 5 seconds, otherwise the collection is not started. A rejected pre-flight (`OK` clear) carries the reason as error text and does not break the connection. - `START` **must** be acknowledged (`ACK` with `ack_for=START`) on each connection within 5 seconds, otherwise collection start fails and a rollback is triggered. - `END` **must** be acknowledged (`ack_for=END`) on each connection within 10 seconds for successful completion. - `CANCEL` should be acknowledged during rollback paths (500ms timeout). @@ -122,13 +133,15 @@ On Linux, large payload transmission (`DATA` and `CALIBRATION` frames) can use k | 5 | `ACK` | Acknowledgement / error reporting | | 6 | `CANCEL` | Cancel run initialization/stream | | 7 | `KEEPALIVE` | Connection liveness probe/pong | +| 8 | `BUSY` | Writer alive but stalled; carries its FIFO occupancy | +| 9 | `PREFLIGHT` | Dry run before a collection starts: can this run be written? | #### TCP frame header (`TcpFrameHeader`) | Field | Type | Description | |--------------------------|---|----------------------------------------------------------| | `magic` | `uint32_t` | Protocol magic (`0x4A464A54`, `"JFJT"`) | -| `version` | `uint16_t` | Protocol version (`2`) | +| `version` | `uint16_t` | Protocol version (`4`) | | `type` | `uint16_t` | Frame type (see table above) | | `image_number` | `uint64_t` | Image index for `DATA` frames | | `payload_size` | `uint64_t` | Number of payload bytes after header | diff --git a/image_puller/TCPImagePuller.cpp b/image_puller/TCPImagePuller.cpp index 8b2ff11c..9fd20065 100644 --- a/image_puller/TCPImagePuller.cpp +++ b/image_puller/TCPImagePuller.cpp @@ -132,7 +132,9 @@ void TCPImagePuller::CBORThread() { } else { ret.cbor = CBORStream2Deserialize(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size()); outside_fifo.PutBlocking(ret); - if (repub_socket) { + // A PREFLIGHT carries a StartMessage payload but starts nothing, so republishing it + // would announce a run to downstream consumers that is not going to happen. + if (repub_socket && (type != TCPFrameType::PREFLIGHT)) { if ((ret.cbor->msg_type == CBORImageType::START) || (ret.cbor->msg_type == CBORImageType::END)) repub_fifo.PutBlocking(ret); diff --git a/image_pusher/HDF5FilePusher.cpp b/image_pusher/HDF5FilePusher.cpp index 31a1bc2a..d4bd146b 100644 --- a/image_pusher/HDF5FilePusher.cpp +++ b/image_pusher/HDF5FilePusher.cpp @@ -15,6 +15,11 @@ HDF5FilePusher::HDF5FilePusher(const std::string &repub_address, } } +void HDF5FilePusher::Preflight(StartMessage &message) { + // In-process writer, so the check is the same call the FileWriter constructor will make. + FileWriter::Preflight(message); +} + void HDF5FilePusher::StartDataCollection(StartMessage &message) { if (writer) throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher is already writing images"); diff --git a/image_pusher/HDF5FilePusher.h b/image_pusher/HDF5FilePusher.h index f7fcbdc3..34d078bc 100644 --- a/image_pusher/HDF5FilePusher.h +++ b/image_pusher/HDF5FilePusher.h @@ -31,6 +31,7 @@ public: const std::optional &repub_watermark = {}); // Thread safety: StartDataCollection, EndDataCollection and SendCalibration must run poorly in serial context // SendImage can be executed in parallel + void Preflight(StartMessage &message) override; void StartDataCollection(StartMessage &message) override; bool EndDataCollection(const EndMessage &message) override; bool SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override; diff --git a/image_pusher/ImagePusher.h b/image_pusher/ImagePusher.h index 065d4ef1..243b8003 100644 --- a/image_pusher/ImagePusher.h +++ b/image_pusher/ImagePusher.h @@ -38,6 +38,11 @@ protected: CBORStream2Serializer serializer; ImagePusher(); public: + // Dry run before the detector is armed: ask the writer whether it could write this run's + // output, and throw if it could not. Only transports with a back-channel can answer, so the + // default is to say nothing - the ZeroMQ pusher is fire-and-forget and keeps failing at the + // final rename instead. Called from the same serialized control thread as StartDataCollection. + virtual void Preflight(StartMessage& message) {} virtual void StartDataCollection(StartMessage& message) = 0; virtual bool EndDataCollection(const EndMessage& message) = 0; // Non-blocking virtual bool SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) = 0; diff --git a/image_pusher/TCPStreamPusher.cpp b/image_pusher/TCPStreamPusher.cpp index 9a082d3a..8d2fe829 100644 --- a/image_pusher/TCPStreamPusher.cpp +++ b/image_pusher/TCPStreamPusher.cpp @@ -497,7 +497,7 @@ void TCPStreamPusher::PersistentAckThread(Connection* c) { // Validate run number: discard stale ACKs from a previous run on a persistent connection if (h.run_number != run_number) { logger.Warning("Discarding ACK with stale run_number " + std::to_string(h.run_number) - + " (expected " + std::to_string(run_number) + ") on socket " + + " (expected " + std::to_string(run_number.load()) + ") on socket " + std::to_string(c->socket_number)); continue; } @@ -508,7 +508,12 @@ void TCPStreamPusher::PersistentAckThread(Connection* c) { if (!error_text.empty()) c->last_ack_error = error_text; - if (ack_for == TCPFrameType::START) { + if (ack_for == TCPFrameType::PREFLIGHT) { + c->preflight_ack_received = true; + c->preflight_ack_ok = ok; + if (!ok && error_text.empty()) + c->last_ack_error = "PREFLIGHT rejected"; + } else if (ack_for == TCPFrameType::START) { c->start_ack_received = true; c->start_ack_ok = ok; if (!ok && error_text.empty()) @@ -718,6 +723,8 @@ size_t TCPStreamPusher::GetConnectedWriters() const { void TCPStreamPusher::StartDataCollectionThreads(Connection& c) { { std::unique_lock ul(c.ack_mutex); + c.preflight_ack_received = false; + c.preflight_ack_ok = false; c.start_ack_received = false; c.start_ack_ok = false; c.end_ack_received = false; @@ -771,6 +778,8 @@ void TCPStreamPusher::StopDataCollectionThreads(Connection& c) { bool TCPStreamPusher::WaitForAck(Connection& c, TCPFrameType ack_for, std::chrono::milliseconds timeout, std::string* error_text) { std::unique_lock ul(c.ack_mutex); const bool ok = c.ack_cv.wait_for(ul, timeout, [&] { + if (ack_for == TCPFrameType::PREFLIGHT) + return c.preflight_ack_received || c.broken.load(); if (ack_for == TCPFrameType::START) return c.start_ack_received || c.broken.load(); if (ack_for == TCPFrameType::END) @@ -791,6 +800,7 @@ bool TCPStreamPusher::WaitForAck(Connection& c, TCPFrameType ack_for, std::chron } bool ack_ok = false; + if (ack_for == TCPFrameType::PREFLIGHT) ack_ok = c.preflight_ack_ok; if (ack_for == TCPFrameType::START) ack_ok = c.start_ack_ok; if (ack_for == TCPFrameType::END) ack_ok = c.end_ack_ok; if (ack_for == TCPFrameType::CANCEL) ack_ok = c.cancel_ack_ok; @@ -834,6 +844,59 @@ bool TCPStreamPusher::WaitForEndAck(Connection& c, std::chrono::milliseconds liv return c.end_ack_ok; } +void TCPStreamPusher::Preflight(StartMessage& message) { + std::vector> local_connections; + { + std::lock_guard lg(connections_mutex); + for (const auto& c : connections) { + if (c->connected && !c->broken) + local_connections.push_back(c); + } + } + + if (local_connections.empty()) + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "No writers connected to " + endpoint); + + // Set before the first frame goes out: PersistentAckThread discards any ACK whose run_number + // does not match, so an ACK for this run would be thrown away if run_number still named the + // previous one. StartDataCollection sets it again to the same value a moment later. + run_number = message.run_number; + + for (size_t i = 0; i < local_connections.size(); i++) { + auto& c = *local_connections[i]; + + // Same assignment StartDataCollection makes, and it has to be the same one: only the + // writer holding the master file checks the output files, so a different index 0 here + // than there would check the run on the wrong writer. + message.socket_number = static_cast(c.socket_number); + message.write_master_file = (i == 0); + + { + std::unique_lock ul(c.ack_mutex); + c.preflight_ack_received = false; + c.preflight_ack_ok = false; + c.last_ack_error.clear(); + } + + serializer.SerializeSequenceStart(message); + + { + std::unique_lock ul(c.send_mutex); + if (!SendFrame(c, serialization_buffer.data(), serializer.GetBufferSize(), TCPFrameType::PREFLIGHT, -1)) + throw JFJochException(JFJochExceptionCategory::FileWriteError, + "Timeout/failure sending PREFLIGHT on socket " + + std::to_string(c.socket_number)); + } + + std::string ack_err; + if (!WaitForAck(c, TCPFrameType::PREFLIGHT, std::chrono::seconds(5), &ack_err)) + throw JFJochException(JFJochExceptionCategory::FileWriteError, + "Pre-flight check failed on socket " + std::to_string(c.socket_number) + + ": " + ack_err); + } +} + void TCPStreamPusher::StartDataCollection(StartMessage& message) { if (message.images_per_file < 1) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Images per file cannot be zero or negative"); diff --git a/image_pusher/TCPStreamPusher.h b/image_pusher/TCPStreamPusher.h index db6dcc1c..dde301d8 100644 --- a/image_pusher/TCPStreamPusher.h +++ b/image_pusher/TCPStreamPusher.h @@ -63,6 +63,8 @@ class TCPStreamPusher : public ImagePusher { std::mutex ack_mutex; std::condition_variable ack_cv; + bool preflight_ack_received = false; + bool preflight_ack_ok = false; bool start_ack_received = false; bool start_ack_ok = false; bool end_ack_received = false; @@ -123,7 +125,9 @@ class TCPStreamPusher : public ImagePusher { std::chrono::milliseconds max_backpressure_timeout{60000}; int64_t images_per_file = 1; - uint64_t run_number = 0; + // Written by the control thread (Preflight/StartDataCollection), read by every + // PersistentAckThread to discard ACKs belonging to an earlier run. + std::atomic run_number{0}; std::string run_name; std::atomic transmission_error = false; std::atomic data_collection_active{false}; @@ -187,6 +191,7 @@ public: /// Returns the number of currently connected writers (can be called at any time) size_t GetConnectedWriters() const override; + void Preflight(StartMessage& message) override; void StartDataCollection(StartMessage& message) override; bool EndDataCollection(const EndMessage& message) override; bool SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override; diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index 1f810c25..d25a5996 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -44,6 +44,24 @@ std::optional JFJochReceiverService::GetStatus() { return receiver_status.GetStatus(); } +void JFJochReceiverService::Preflight(const DiffractionExperiment &experiment) { + // Same condition JFJochReceiver uses to decide whether it pushes to a writer at all: with no + // images or no file prefix nothing is written, so there is nothing to check. + if ((experiment.GetImageNum() <= 0) || experiment.GetFilePrefix().empty()) + return; + + // FillMessage alone, which is what makes this frame small: the per-pixel arrays that dominate a + // real start message - the pixel mask, the azimuthal-integration map, the ROI map - are added + // by JFJochReceiver::SendStartMessage and are not here. Measured on a JUNGFRAU 9M: 1.4 kB + // against 543 kB for the start message that follows. It is deliberately the whole of + // FillMessage rather than the handful of fields the check reads today, so that the writer sees + // the same description of the run it will be given at start - a hand-picked subset would + // answer for a different set of file names the day the check learns to read one more field. + StartMessage message{}; + experiment.FillMessage(message); + image_pusher.Preflight(message); +} + void JFJochReceiverService::Start(const DiffractionExperiment &experiment, const PixelMask &pixel_mask, const JFCalibration *calibration, diff --git a/receiver/JFJochReceiverService.h b/receiver/JFJochReceiverService.h index 5c4748fd..678417b3 100644 --- a/receiver/JFJochReceiverService.h +++ b/receiver/JFJochReceiverService.h @@ -59,6 +59,9 @@ public: void LoadInternalGeneratorImage(const DiffractionExperiment& experiment, const std::vector &raw_expected_image, uint64_t image_number); + // Ask the writer whether this run's output could be written, before anything is started. + // Throws if it could not. + void Preflight(const DiffractionExperiment &experiment); void Start(const DiffractionExperiment &experiment, const PixelMask &pixel_mask, const JFCalibration *calibration, diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 0757fd7c..7e93f539 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -84,6 +84,7 @@ ADD_EXECUTABLE(jfjoch_test TopPixelsTest.cpp HKLKeyTest.cpp TCPImagePusherTest.cpp + PreflightTest.cpp SearchSpaceGroupTest.cpp SearchSpaceGroupTwinTest.cpp SyntheticMergedReflections.h diff --git a/tests/HDF5WritingTest.cpp b/tests/HDF5WritingTest.cpp index 2f31ea4f..48616c49 100644 --- a/tests/HDF5WritingTest.cpp +++ b/tests/HDF5WritingTest.cpp @@ -1351,9 +1351,11 @@ TEST_CASE("HDF5Objects_VDS_reverse_strided", "[HDF5][Unit]") { remove("scratch_vds_reverse_strided.h5"); REQUIRE(H5Fget_obj_count(H5F_OBJ_ALL, H5F_OBJ_ALL) == 0); } -// Overwrite is detected up front for back-channel transports (default) by checking -// the master file only - never the staggered per-writer data files. The ZeroMQ path -// (no back-channel) must opt out and keep writing .tmp files instead. +// Overwrite is detected up front for back-channel transports (default): the writer that owns the +// master file checks the master file and every data file the run will write, its staggered +// siblings' included - they share a directory by construction, since the master links them by +// relative name. The ZeroMQ path (no back-channel) must opt out and keep writing .tmp files +// instead. TEST_CASE("FileWriter_overwrite_detected_at_start", "[HDF5][Overwrite]") { RegisterHDF5Filter(); @@ -1366,10 +1368,19 @@ TEST_CASE("FileWriter_overwrite_detected_at_start", "[HDF5][Overwrite]") { x.FillMessage(start_message); REQUIRE(start_message.write_master_file.value_or(false)); // this writer owns the master - // A stray data file (owned by another, staggered writer) must NOT trip the check - - // only the master file is inspected. The temporary writer cleans up its own tmp. + // A data file in the way is refused up front, even one another writer would have written. + // It used to pass here and fail only when the file was renamed into place at the end of the + // run - after the whole dataset had been collected. { std::ofstream(HDF5Metadata::DataFileName(start_message, 0)) << "blocker"; } - REQUIRE_NOTHROW(FileWriter(start_message)); + REQUIRE_THROWS_AS(FileWriter(start_message), JFJochException); + + // A writer that does not own the master file speaks for none of them: it would otherwise stat + // files its siblings are already creating. The temporary writer cleans up its own tmp. + { + StartMessage not_master = start_message; + not_master.write_master_file = false; + REQUIRE_NOTHROW(FileWriter(not_master)); + } remove(HDF5Metadata::DataFileName(start_message, 0).c_str()); // The master file, on the other hand, does collide. diff --git a/tests/JFJochStateMachineTest.cpp b/tests/JFJochStateMachineTest.cpp index 3c43e8ee..dd5f9ea5 100644 --- a/tests/JFJochStateMachineTest.cpp +++ b/tests/JFJochStateMachineTest.cpp @@ -355,3 +355,198 @@ TEST_CASE("JFJochStateMachine_CalibrationFailure") { 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 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(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(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 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(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); +} diff --git a/tests/PreflightTest.cpp b/tests/PreflightTest.cpp new file mode 100644 index 00000000..25ae8637 --- /dev/null +++ b/tests/PreflightTest.cpp @@ -0,0 +1,242 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include + +#include +#include +#include +#include + +#include "../image_pusher/HDF5FilePusher.h" +#include "../image_pusher/TCPStreamPusher.h" +#include "../image_pusher/ZMQStream2Pusher.h" +#include "../image_puller/TCPImagePuller.h" +#include "../writer/StreamWriter.h" +#include "../common/PixelMask.h" + +namespace { + const std::string preflight_dir = "preflight_test"; + + void PlaceFile(const std::string &name) { + std::filesystem::path path(name); + if (path.has_parent_path()) + std::filesystem::create_directories(path.parent_path()); + std::ofstream f(name); + f << "in the way"; + } + + // A run number that is neither zero nor the default: an ACK is only matched to its request if + // both sides put this number on it, so anything that forgets it shows up as an ACK timeout. + constexpr uint64_t preflight_run_number = 4711; + + StartMessage PreflightMessage(const std::string &prefix) { + StartMessage msg{}; + msg.file_prefix = prefix; + msg.run_number = preflight_run_number; + msg.run_name = "preflight"; + msg.images_per_file = 2; + msg.number_of_images = 5; + msg.file_format = FileWriterFormat::NXmxVDS; + msg.overwrite = false; + return msg; + } + + void WaitForWriters(const TCPStreamPusher &pusher, size_t expected) { + for (int i = 0; i < 200 && pusher.GetConnectedWriters() < expected; i++) + std::this_thread::sleep_for(std::chrono::milliseconds(25)); + } +} + +// The pre-flight is a dry run of the writer's own start-time checks, sent before anything is +// started. What it has to get right is that the answer comes back at all (an ACK carries the run +// number of the run being checked, and either side forgetting it looks like a timeout), that a +// refusal is reported with the writer's own message, and that a refusal costs nothing: the writer +// stays idle and the connection stays usable for the next attempt. +TEST_CASE("Preflight_TCP", "[Preflight][TCP]") { + Logger logger("Preflight_TCP"); + + std::filesystem::remove_all(preflight_dir); + + TCPStreamPusher pusher("tcp://127.0.0.1:*", 1); + TCPImagePuller puller(pusher.GetAddress()[0], 8 * 1024 * 1024); + StreamWriter writer(logger, puller); + + auto writer_future = std::async(std::launch::async, [&] { return writer.Run(); }); + + WaitForWriters(pusher, 1); + REQUIRE(pusher.GetConnectedWriters() == 1); + + // Nothing in the way: the check passes, and it creates the output directory on the way - the + // prefix names a subdirectory that does not exist yet. + auto clean = PreflightMessage(preflight_dir + "/sub/clean"); + REQUIRE_NOTHROW(pusher.Preflight(clean)); + CHECK(std::filesystem::is_directory(preflight_dir + "/sub")); + + // A dry run starts nothing: the writer is still waiting for a START. + CHECK(writer.GetStatistics().state == StreamWriterState::Idle); + + // A master file already there is refused, and the writer's own message comes back with it. + auto collision = PreflightMessage(preflight_dir + "/collision"); + PlaceFile(preflight_dir + "/collision_master.h5"); + REQUIRE_THROWS_WITH(pusher.Preflight(collision), + Catch::Matchers::ContainsSubstring("collision_master.h5")); + + // Refusing a run is not a transport failure - the connection carries the next attempt. + CHECK(pusher.GetConnectedWriters() == 1); + CHECK(writer.GetStatistics().state == StreamWriterState::Idle); + auto second = PreflightMessage(preflight_dir + "/second"); + REQUIRE_NOTHROW(pusher.Preflight(second)); + + // The same collision passes once overwriting is allowed. + auto overwriting = PreflightMessage(preflight_dir + "/collision"); + overwriting.overwrite = true; + REQUIRE_NOTHROW(pusher.Preflight(overwriting)); + + // A data file in the way is refused too, with no master file present. This is the case the + // start-time check used to let through: the run was written in full and only failed when the + // first data file was renamed into place at the end of it. + auto data_collision = PreflightMessage(preflight_dir + "/data_collision"); + PlaceFile(HDF5Metadata::DataFileName(data_collision, 1)); + REQUIRE_FALSE(std::filesystem::exists(preflight_dir + "/data_collision_master.h5")); + REQUIRE_THROWS_WITH(pusher.Preflight(data_collision), + Catch::Matchers::ContainsSubstring("Output file already exists")); + + // A prefix that escapes the output directory is refused by the same path guard the writer + // applies at start time. + auto escaping = PreflightMessage("../escape"); + REQUIRE_THROWS(pusher.Preflight(escaping)); + + // A run the pre-flight passed still starts: the run number the pre-flight left behind must not + // stop the START ACK from being matched. + auto started = PreflightMessage(preflight_dir + "/started"); + REQUIRE_NOTHROW(pusher.Preflight(started)); + REQUIRE_NOTHROW(pusher.StartDataCollection(started)); + CHECK(writer.GetStatistics().run_number == preflight_run_number); + EndMessage end{}; + end.run_number = preflight_run_number; + CHECK(pusher.EndDataCollection(end)); + + writer.Cancel(); + REQUIRE_NOTHROW(writer_future.get()); + + std::filesystem::remove_all(preflight_dir); +} + +// With more than one writer every connection is asked, but only the one holding the master file +// checks the output - the same assignment the start makes, so a pre-flight cannot pass a run the +// start would then refuse. +TEST_CASE("Preflight_TCP_TwoWriters", "[Preflight][TCP]") { + Logger logger("Preflight_TCP_TwoWriters"); + + std::filesystem::remove_all(preflight_dir); + + TCPStreamPusher pusher("tcp://127.0.0.1:*", 2); + + std::vector> pullers; + std::vector> writers; + std::vector> futures; + + for (int i = 0; i < 2; i++) { + pullers.push_back(std::make_unique(pusher.GetAddress()[0], 8 * 1024 * 1024)); + writers.push_back(std::make_unique(logger, *pullers.back())); + futures.push_back(std::async(std::launch::async, [w = writers.back().get()] { return w->Run(); })); + } + + WaitForWriters(pusher, 2); + REQUIRE(pusher.GetConnectedWriters() == 2); + + auto clean = PreflightMessage(preflight_dir + "/two_clean"); + REQUIRE_NOTHROW(pusher.Preflight(clean)); + + // Both connections were asked, and each was told which socket it is. + CHECK(writers[0]->GetStatistics().socket_number != writers[1]->GetStatistics().socket_number); + + auto collision = PreflightMessage(preflight_dir + "/two_collision"); + PlaceFile(preflight_dir + "/two_collision_master.h5"); + REQUIRE_THROWS_WITH(pusher.Preflight(collision), + Catch::Matchers::ContainsSubstring("two_collision_master.h5")); + + for (auto &w : writers) + w->Cancel(); + for (auto &f : futures) + REQUIRE_NOTHROW(f.get()); + + std::filesystem::remove_all(preflight_dir); +} + +TEST_CASE("Preflight_TCP_NoWriters", "[Preflight][TCP]") { + TCPStreamPusher pusher("tcp://127.0.0.1:*", 1); + + auto msg = PreflightMessage(preflight_dir + "/no_writers"); + REQUIRE_THROWS_WITH(pusher.Preflight(msg), + Catch::Matchers::ContainsSubstring("No writers connected")); +} + +// The in-process writer has no transport to ask, so its pre-flight is the check itself. +TEST_CASE("Preflight_HDF5FilePusher", "[Preflight]") { + std::filesystem::remove_all(preflight_dir); + + HDF5FilePusher pusher; + + auto clean = PreflightMessage(preflight_dir + "/hdf5_clean"); + clean.write_master_file = true; + REQUIRE_NOTHROW(pusher.Preflight(clean)); + CHECK(std::filesystem::is_directory(preflight_dir)); + + auto collision = PreflightMessage(preflight_dir + "/hdf5_collision"); + collision.write_master_file = true; + PlaceFile(preflight_dir + "/hdf5_collision_master.h5"); + REQUIRE_THROWS_WITH(pusher.Preflight(collision), + Catch::Matchers::ContainsSubstring("hdf5_collision_master.h5")); + + // A pusher that is only asked to write data files does not speak for the master file. + auto not_master = PreflightMessage(preflight_dir + "/hdf5_collision"); + not_master.write_master_file = false; + REQUIRE_NOTHROW(pusher.Preflight(not_master)); + + std::filesystem::remove_all(preflight_dir); +} + +// The pre-flight frame goes out once per run per writer, and it describes the run without carrying +// any of the per-pixel arrays a start message does. Keeping it that way is the point of building it +// from FillMessage alone: measured on a JUNGFRAU 9M it is 1.4 kB, against 359 kB once the pixel mask +// is added and 543 kB with the azimuthal and ROI maps on top. The bound below is loose - it is here +// to catch a per-pixel array finding its way into FillMessage, not to pin the exact size. +TEST_CASE("Preflight_MessageStaysSmall", "[Preflight]") { + DiffractionExperiment x(DetJF(18)); // JUNGFRAU 9M - the largest detector in the standard set + x.FilePrefix("size_probe").NumTriggers(1).ImagesPerTrigger(3600).ImagesPerFile(1000) + .UseInternalPacketGenerator(true).IncidentEnergy_keV(12.4); + + // Built exactly as JFJochReceiverService::Preflight builds it. + StartMessage message{}; + x.FillMessage(message); + + std::vector buffer(MESSAGE_SIZE_FOR_START_END); + CBORStream2Serializer serializer(buffer.data(), buffer.size()); + serializer.SerializeSequenceStart(message); + CHECK(serializer.GetBufferSize() < 64 * 1024); + + // For contrast: one per-pixel array is two orders of magnitude larger than the whole message. + PixelMask mask(x); + message.pixel_mask["default"] = mask.GetMask(x); + CBORStream2Serializer with_mask(buffer.data(), buffer.size()); + with_mask.SerializeSequenceStart(message); + CHECK(with_mask.GetBufferSize() > 100 * 1024); +} + +// The transport with no back-channel cannot ask anything, so its pre-flight has to be silent +// rather than guess - the ZeroMQ writer still fails at the final rename, as it always has. +TEST_CASE("Preflight_NoBackChannelIsSilent", "[Preflight]") { + std::filesystem::remove_all(preflight_dir); + + ZMQStream2Pusher pusher({"ipc://*"}); + + auto collision = PreflightMessage(preflight_dir + "/zmq_collision"); + collision.write_master_file = true; + PlaceFile(preflight_dir + "/zmq_collision_master.h5"); + REQUIRE_NOTHROW(pusher.Preflight(collision)); + + std::filesystem::remove_all(preflight_dir); +} + diff --git a/writer/FileWriter.cpp b/writer/FileWriter.cpp index e00f586e..a566cc01 100644 --- a/writer/FileWriter.cpp +++ b/writer/FileWriter.cpp @@ -24,7 +24,7 @@ FileWriter::FileWriter(const StartMessage &request, bool check_overwrite_at_star CheckPath(start_message.file_prefix); MakeDirectory(start_message.file_prefix); if (check_overwrite_at_start) - CheckOutputFilesAvailable(); + CheckOutputFilesAvailable(start_message, format); if (start_message.write_master_file && start_message.write_master_file.value()) { switch (format) { case FileWriterFormat::NXmxLegacy: @@ -218,25 +218,53 @@ void FileWriter::CreateHDF5MasterFile(const StartMessage &msg) { master_file = std::make_unique(msg); } -void FileWriter::CheckOutputFilesAvailable() const { - if (start_message.overwrite.value_or(false)) +void FileWriter::CheckOutputFilesAvailable(const StartMessage &msg, FileWriterFormat format) { + if (msg.overwrite.value_or(false)) return; - // Only the master file is checked, and only by the single writer that owns it - // (write_master_file - index 0 in a multi-writer TCP/ZMQ setup). Data files are - // staggered across writers by file number, so enumerating them here would make - // every writer stat files it never writes and race sibling writers that are - // already creating them; those conflicts are caught per-writer at finalize. + // Checked by the single writer that owns the master file (write_master_file - index 0 in a + // multi-writer TCP/ZMQ setup), and it checks the whole run, its siblings' data files included. + // Data files are staggered across writers by file number, so letting each writer check its own + // would race the siblings already creating them; one writer can speak for all of them because + // the master links the data files by relative name, so they share a directory by construction. + if (!msg.write_master_file.value_or(false)) + return; + + auto refuse = [](const std::string &name) { + if (std::filesystem::exists(name)) + throw JFJochException(JFJochExceptionCategory::FileWriteError, + "Output file already exists and overwrite is off: " + name); + }; + const bool nxmx = format == FileWriterFormat::NXmxLegacy || format == FileWriterFormat::NXmxVDS || format == FileWriterFormat::NXmxIntegrated; - if (nxmx && start_message.write_master_file.value_or(false)) { - const std::string name = HDF5Metadata::MasterFileName(start_message); - if (std::filesystem::exists(name)) - throw JFJochException(JFJochExceptionCategory::FileWriteError, - "Output file already exists and overwrite is off: " + name); - } + if (nxmx) + refuse(HDF5Metadata::MasterFileName(msg)); + + // NXmxIntegrated puts the images in the master file, so there are no data files to check. + if (format == FileWriterFormat::NXmxIntegrated || format == FileWriterFormat::NoFile) + return; + + const int64_t per_file = msg.images_per_file > 0 + ? msg.images_per_file + : static_cast(default_images_per_file); + const int64_t total_images = static_cast(msg.number_of_images); + for (int64_t file_number = 0; file_number * per_file < total_images; file_number++) + refuse(HDF5Metadata::DataFileName(msg, file_number)); +} + +void FileWriter::Preflight(const StartMessage &request, bool trusted_path) { + if (!trusted_path) + CheckPath(request.file_prefix); + MakeDirectory(request.file_prefix); + + FileWriterFormat format = FileWriterFormat::NXmxLegacy; + if (request.file_format) + format = request.file_format.value(); + + CheckOutputFilesAvailable(request, format); } void FileWriter::WriteHDF5(const CompressedImage &msg) { diff --git a/writer/FileWriter.h b/writer/FileWriter.h index 338b0bae..031cc6f1 100644 --- a/writer/FileWriter.h +++ b/writer/FileWriter.h @@ -24,7 +24,7 @@ class FileWriter { constexpr static uint64_t close_file_lag_images = 1000; constexpr static uint64_t default_images_per_file = 1000; void CreateHDF5MasterFile(const StartMessage& msg); - void CheckOutputFilesAvailable() const; + static void CheckOutputFilesAvailable(const StartMessage &msg, FileWriterFormat format); void AddStats(const std::optional& s); void CloseFile(uint64_t file_number); void CloseOldFiles(uint64_t current_image_number); @@ -42,6 +42,12 @@ public: // never set it, so their behaviour is unchanged. explicit FileWriter(const StartMessage &request, bool check_overwrite_at_start = true, bool trusted_path = false); + // Everything the constructor does before it opens anything: the path guard, creating the + // output directory and checking that no output file is in the way. Sent to the writer as a + // PREFLIGHT frame before the detector is armed, so a run that cannot be written is refused + // while refusing it is still free. Shares its code with the constructor, so a pre-flight that + // passes cannot be contradicted by the start that follows it. + static void Preflight(const StartMessage &request, bool trusted_path = false); void Write(const DataMessage& msg); void WriteHDF5(const DataMessage& msg); void WriteHDF5(const CompressedImage& msg); diff --git a/writer/StreamWriter.cpp b/writer/StreamWriter.cpp index 438de0e9..7303ec19 100644 --- a/writer/StreamWriter.cpp +++ b/writer/StreamWriter.cpp @@ -87,6 +87,28 @@ void StreamWriter::ProcessStartMessage() { } } +// Dry run for a run that has not started: check that the output could be written and answer, +// without opening anything and without touching state - the writer stays Idle and ready for the +// START that follows. +void StreamWriter::ProcessPreflight() { + const StartMessage &msg = image_puller_output.cbor->start_message.value(); + + // The pusher's ACK reader discards frames whose run_number does not match the run it is about + // to start, so the ACK has to carry the run number of the message being checked - it is not yet + // the one ProcessStartMessage will store. + run_number = msg.run_number; + socket_number = msg.socket_number.value_or(0); + + try { + FileWriter::Preflight(msg); + NotifyTcpAck(TCPFrameType::PREFLIGHT, true, false, TCPAckCode::None); + } catch (const JFJochException &e) { + logger.Warning("Pre-flight check failed: {}", e.what()); + // Not fatal: nothing was started, so the connection stays usable for the next attempt. + NotifyTcpAck(TCPFrameType::PREFLIGHT, false, false, TCPAckCode::StartFailed, e.what()); + } +} + void StreamWriter::ProcessCalibrationImage() { switch (state) { case StreamWriterState::Started: @@ -220,6 +242,11 @@ void StreamWriter::CollectImages() { bool run = true; while (run && state != StreamWriterState::Finalized) { run = WaitForImage(); + // No new frame: WaitForImage leaves the previous one in image_puller_output, and the body + // below would go on to process it a second time - writing the last image twice, or + // repeating the last check - before the loop condition noticed. + if (!run) + break; if (image_puller_output.tcp_msg && static_cast(image_puller_output.tcp_msg->header.type) == TCPFrameType::CANCEL) { @@ -240,6 +267,17 @@ void StreamWriter::CollectImages() { continue; } + // A PREFLIGHT carries a StartMessage payload, so it has to be told apart by frame type + // before the payload-driven dispatch below would take it for a real START. + if (image_puller_output.tcp_msg && + static_cast(image_puller_output.tcp_msg->header.type) == TCPFrameType::PREFLIGHT) { + if (image_puller_output.cbor->start_message) + ProcessPreflight(); + else + logger.Warning("PREFLIGHT frame without start message"); + continue; + } + if (image_puller_output.cbor->start_message) ProcessStartMessage(); else if (image_puller_output.cbor->calibration) diff --git a/writer/StreamWriter.h b/writer/StreamWriter.h index 95c18bed..00784b17 100644 --- a/writer/StreamWriter.h +++ b/writer/StreamWriter.h @@ -63,6 +63,7 @@ class StreamWriter { void NotifyReceiverOnFinalizedWrite(const std::string &detector_update_zmq_addr); void NotifyTcpAck(TCPFrameType ack_for, bool ok, bool fatal, TCPAckCode code, const std::string &error_text = ""); void ProcessStartMessage(); + void ProcessPreflight(); void ProcessEndMessage(); void ProcessDataImage(); void ProcessCalibrationImage();