leonarski_fandClaude Opus 5.5
2981aedd3e
With a DECTRIS detector the broker starts the writers - and so learned the collection's start message - only when the detector's own start message arrived, after /start had returned. A reader started right after /start found no collection (404) and had to wait for, and race with, the detector. - /start registers the collection in LiveCollection (Expect) as it is accepted, so GET /live/events is accepted from then on. The image pusher fills in the start message when it starts the writers (Begin) - at once on the FPGA path, when the detector's start message arrives on the DECTRIS path. - The event stream's first event is now `start`, carrying that start message (CBOR, base64) with run number, file prefix, image count and images per file; it replaces `collection`. Until it is known the stream only keeps alive. file and end follow as before. - However the measurement thread ends, it ends a collection the pusher did not end (EndIfOpen), so a cancel or a failure before the detector streamed reaches the reader as `end` with an error. The wait is event-driven: it ends with the start message or with the measurement. - GET /live/start.cbor is removed: the stream carries the start message, one route less. - rugnux takes the start message from the stream (BrokerFeed::WaitForStart), so it can be started right after /start; files reported before it attached the reader are replayed to it. Tests: LiveCollection_ExpectedAtStart; BrokerFeed_StartFromEventStream (an event stream whose start message comes late, then a file and the end). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SVmAWnzCmRKAXVUCdc4iNi
75 lines
3.0 KiB
C++
75 lines
3.0 KiB
C++
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#pragma once
|
|
|
|
#include <chrono>
|
|
#include <condition_variable>
|
|
#include <cstdint>
|
|
#include <mutex>
|
|
#include <optional>
|
|
#include <string>
|
|
#include <vector>
|
|
|
|
#include "../common/JFJochMessages.h"
|
|
|
|
// What a reader following the current collection needs from the broker (GET /live/start.cbor and
|
|
// GET /live/events): the start message the writer of the master file received, then each data file
|
|
// as a writer reports it closed and renamed into place, then the end of the collection. Fed by the
|
|
// image pusher; read by the HTTP routes. One follower at a time.
|
|
class LiveCollection {
|
|
public:
|
|
struct DataFile {
|
|
uint64_t file_number = 0;
|
|
std::string file; // relative to the collection's root, as the writer named it
|
|
uint64_t first_image = 0;
|
|
uint64_t image_count = 0;
|
|
};
|
|
|
|
struct Snapshot {
|
|
uint64_t generation = 0; // 0: no collection since the broker started
|
|
bool started = false; // the start message is known
|
|
uint64_t run_number = 0;
|
|
std::string run_name;
|
|
std::string file_prefix;
|
|
uint64_t number_of_images = 0;
|
|
int64_t images_per_file = 0;
|
|
std::vector<DataFile> files; // in the order they were closed
|
|
bool ended = false;
|
|
uint64_t end_images = 0;
|
|
std::string end_error;
|
|
};
|
|
|
|
// A collection was accepted (/start); its start message comes with Begin, once the image pusher
|
|
// starts the writers - for a DECTRIS detector only when the detector's own start message arrives.
|
|
// A reader following it in between waits for it.
|
|
void Expect();
|
|
// start_cbor: the start message as the writer of the master file received it.
|
|
void Begin(const StartMessage &start, std::vector<uint8_t> start_cbor);
|
|
void FileClosed(uint64_t file_number, uint64_t image_count);
|
|
// After every writer has acknowledged END: the data files and the master are in place.
|
|
void End(uint64_t images, const std::string &error = "");
|
|
// The measurement is over: ends a collection the image pusher did not end, with this error.
|
|
void EndIfOpen(const std::string &error);
|
|
|
|
// The start message of the current collection; nothing before it is known.
|
|
std::optional<std::vector<uint8_t>> StartCBOR() const;
|
|
|
|
// Blocks until the start message arrives (when not `started_seen`), the collection has more than
|
|
// `files_seen` files, has ended, a new collection has begun, or `wait` passes; returns the state then.
|
|
Snapshot Wait(uint64_t generation, bool started_seen, size_t files_seen, std::chrono::milliseconds wait);
|
|
Snapshot Get() const;
|
|
|
|
// The single follower slot; false when another follower holds it.
|
|
bool AcquireFollower();
|
|
void ReleaseFollower();
|
|
|
|
private:
|
|
mutable std::mutex m;
|
|
std::condition_variable cv;
|
|
Snapshot state;
|
|
StartMessage start_message;
|
|
std::vector<uint8_t> start_cbor;
|
|
bool follower = false;
|
|
};
|