leonarski_fandClaude Opus 5.5
509a54ef92
So that rugnux can process a collection while it is written, the broker reports each data file once
a writer has closed it and renamed it into place:
- Writer: FileWriter takes a callback, called for every closed data file. StreamWriter uses it on TCP
to send an acknowledgement with ack_for = FILE_CLOSED (new TCPFrameType 10): file number and the
images the file holds. TCP protocol version 5. Files closed at END are reported before the END
acknowledgement.
- LiveCollection (image_pusher): the start message the writer of the master file received, the
closed files, the end. Fed by the TCP pusher (FILE_CLOSED acknowledgements; end once every writer
acknowledged END) and by the in-process HDF5 pusher (the FileWriter callback). The ZeroMQ pusher
has no back channel and offers none.
- Two routes, protected by the dataset's bearer tokens like the other dataset routes:
- GET /live/start.cbor: that start message;
- GET /live/events: server-sent events (collection, file, end, superseded) with the history first
and a keepalive every 10 s. One follower at a time; a second gets 409.
Documented in jfjoch_api.yaml; they add no schema, so the generated C++ model is unchanged.
- The HTTP server's thread pool is set explicitly (16, up to 64): an event stream holds a thread for
the whole collection.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SVmAWnzCmRKAXVUCdc4iNi
85 lines
2.4 KiB
C++
85 lines
2.4 KiB
C++
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#include "LiveCollection.h"
|
|
|
|
#include "../writer/HDF5NXmx.h"
|
|
|
|
void LiveCollection::Begin(const StartMessage &start, std::vector<uint8_t> cbor) {
|
|
{
|
|
std::unique_lock ul(m);
|
|
const uint64_t generation = state.generation + 1;
|
|
state = Snapshot{};
|
|
state.generation = generation;
|
|
state.run_number = start.run_number;
|
|
state.run_name = start.run_name;
|
|
state.file_prefix = start.file_prefix;
|
|
state.number_of_images = start.number_of_images;
|
|
state.images_per_file = start.images_per_file;
|
|
start_message = start;
|
|
start_cbor = std::move(cbor);
|
|
}
|
|
cv.notify_all();
|
|
}
|
|
|
|
void LiveCollection::FileClosed(uint64_t file_number, uint64_t image_count) {
|
|
{
|
|
std::unique_lock ul(m);
|
|
if (state.generation == 0 || state.ended)
|
|
return;
|
|
state.files.push_back(DataFile{
|
|
.file_number = file_number,
|
|
.file = HDF5Metadata::DataFileName(start_message, static_cast<int64_t>(file_number)),
|
|
.first_image = file_number * static_cast<uint64_t>(state.images_per_file),
|
|
.image_count = image_count
|
|
});
|
|
}
|
|
cv.notify_all();
|
|
}
|
|
|
|
void LiveCollection::End(uint64_t images, const std::string &error) {
|
|
{
|
|
std::unique_lock ul(m);
|
|
if (state.generation == 0 || state.ended)
|
|
return;
|
|
state.ended = true;
|
|
state.end_images = images;
|
|
state.end_error = error;
|
|
}
|
|
cv.notify_all();
|
|
}
|
|
|
|
std::optional<std::vector<uint8_t>> LiveCollection::StartCBOR() const {
|
|
std::unique_lock ul(m);
|
|
if (state.generation == 0)
|
|
return {};
|
|
return start_cbor;
|
|
}
|
|
|
|
LiveCollection::Snapshot LiveCollection::Wait(uint64_t generation, size_t files_seen,
|
|
std::chrono::milliseconds wait) {
|
|
std::unique_lock ul(m);
|
|
cv.wait_for(ul, wait, [&] {
|
|
return state.generation != generation || state.ended || state.files.size() > files_seen;
|
|
});
|
|
return state;
|
|
}
|
|
|
|
LiveCollection::Snapshot LiveCollection::Get() const {
|
|
std::unique_lock ul(m);
|
|
return state;
|
|
}
|
|
|
|
bool LiveCollection::AcquireFollower() {
|
|
std::unique_lock ul(m);
|
|
if (follower)
|
|
return false;
|
|
follower = true;
|
|
return true;
|
|
}
|
|
|
|
void LiveCollection::ReleaseFollower() {
|
|
std::unique_lock ul(m);
|
|
follower = false;
|
|
}
|