From 4147934e6fc1c149a202ab9f08e461aee679af96 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sun, 8 Nov 2026 08:31:14 +0100 Subject: [PATCH] ZMQMetadataSocket: a batch is serialised and sent outside the lock the image threads take Every image thread took the socket's mutex to add its message, and the thread that cut a batch then serialised and sent the whole period's images while holding it, so all image threads waited for the encode. Now the message is copied outside the lock without spots, reflections and image (the metadata message never carries them), the lock only appends and swaps the batch out, and the batch is serialised into its own buffer and sent with the lock released. Two batches cut close together can leave in either order, which the documentation already allows; END stays last. Two latent faults go with it: a disabled socket (no period) kept every image of the run and sent them all at END - now it keeps nothing; and the batch buffer was a fixed 20 kB per image, too small for a 2D azimuthal integration profile - it is now sized from the profile and ADU histogram lengths. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- preview/ZMQMetadataSocket.cpp | 71 ++++++++++++++++++----------- preview/ZMQMetadataSocket.h | 4 +- tests/ZMQMetadataSocketTest.cpp | 80 ++++++++++++++++++++++++++++++++- 3 files changed, 126 insertions(+), 29 deletions(-) diff --git a/preview/ZMQMetadataSocket.cpp b/preview/ZMQMetadataSocket.cpp index dd62e20eb..77c31f085 100644 --- a/preview/ZMQMetadataSocket.cpp +++ b/preview/ZMQMetadataSocket.cpp @@ -20,6 +20,7 @@ ZMQMetadataSocket::ZMQMetadataSocket(const std::string &addr) void ZMQMetadataSocket::StartDataCollection(const StartMessage &message) { std::unique_lock ul(m); + metadata_message = MetadataMessage{}; if (buffer.size() < MESSAGE_SIZE_FOR_START_END) buffer.resize(MESSAGE_SIZE_FOR_START_END); CBORStream2Serializer serializer(buffer.data(), buffer.size()); @@ -28,43 +29,61 @@ void ZMQMetadataSocket::StartDataCollection(const StartMessage &message) { first_image = true; } -void ZMQMetadataSocket::SendAll() { - if (buffer.size() < metadata_message.images.size() * MESSAGE_SIZE_FOR_IMAGE) - buffer.resize(metadata_message.images.size() * MESSAGE_SIZE_FOR_IMAGE); +// Called without m: serialising a batch must not hold up the image threads +void ZMQMetadataSocket::Send(const MetadataMessage &batch) { + size_t size = 0; + for (const auto &i: batch.images) + size += MESSAGE_SIZE_FOR_IMAGE + 16 * i.az_int_profile.size() + 9 * i.adu_histogram.size(); - // Serialize - CBORStream2Serializer serializer(buffer.data(), buffer.size()); - serializer.SerializeMetadata(metadata_message); - - // Stream out - socket.Send(buffer.data(), serializer.GetBufferSize(), false); - - metadata_message = MetadataMessage{}; + std::vector batch_buffer(size); + CBORStream2Serializer serializer(batch_buffer.data(), batch_buffer.size()); + serializer.SerializeMetadata(batch); + socket.Send(batch_buffer.data(), serializer.GetBufferSize(), false); } void ZMQMetadataSocket::AddDataMessage(const DataMessage &msg) { - std::unique_lock ul(m); + // Disabled: nothing is kept + if (!counter.GetPeriod()) + return; - if (first_image) { - // GeneratePreview() is designed to fire on the first try. This would result however in situation where - // first metadata packet always contains only 1 image, and this is not desired behavior. - // Therefore, on first use there is one extra "artificial" check of generate preview, that starts the clock - counter.GeneratePreview(); + // Spots, reflections and the image are not part of the metadata message + DataMessage copy = msg; + copy.spots = {}; + copy.reflections = {}; + copy.image = CompressedImage(); - metadata_message.run_name = msg.run_name; - metadata_message.run_number = msg.run_number; - first_image = false; + MetadataMessage batch; + { + std::unique_lock ul(m); + + if (first_image) { + // GeneratePreview() is designed to fire on the first try. This would result however in situation where + // first metadata packet always contains only 1 image, and this is not desired behavior. + // Therefore, on first use there is one extra "artificial" check of generate preview, that starts the clock + counter.GeneratePreview(); + + metadata_message.run_name = msg.run_name; + metadata_message.run_number = msg.run_number; + first_image = false; + } + + metadata_message.images.emplace_back(std::move(copy)); + if (!counter.GeneratePreview()) + return; + + batch.run_name = metadata_message.run_name; + batch.run_number = metadata_message.run_number; + std::swap(batch.images, metadata_message.images); } - - metadata_message.images.emplace_back(msg); - if (counter.GeneratePreview()) - SendAll(); + Send(batch); } void ZMQMetadataSocket::EndDataCollection(const EndMessage &message) { std::unique_lock ul(m); - if (!metadata_message.images.empty()) - SendAll(); + if (!metadata_message.images.empty()) { + Send(metadata_message); + metadata_message.images.clear(); + } if (buffer.size() < MESSAGE_SIZE_FOR_START_END) buffer.resize(MESSAGE_SIZE_FOR_START_END); diff --git a/preview/ZMQMetadataSocket.h b/preview/ZMQMetadataSocket.h index a83fbaa71..9870471d3 100644 --- a/preview/ZMQMetadataSocket.h +++ b/preview/ZMQMetadataSocket.h @@ -23,12 +23,12 @@ class ZMQMetadataSocket { MetadataMessage metadata_message; ZMQSocket socket; - std::vector buffer; + std::vector buffer; // START and END only, under m PreviewCounter counter; bool first_image; - void SendAll(); + void Send(const MetadataMessage &batch); public: explicit ZMQMetadataSocket(const std::string &addr); diff --git a/tests/ZMQMetadataSocketTest.cpp b/tests/ZMQMetadataSocketTest.cpp index 67263be36..93f5f8c7e 100644 --- a/tests/ZMQMetadataSocketTest.cpp +++ b/tests/ZMQMetadataSocketTest.cpp @@ -1,6 +1,8 @@ // SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute // SPDX-License-Identifier: GPL-3.0-only +#include +#include #include #include "../preview/ZMQMetadataSocket.h" @@ -53,4 +55,80 @@ TEST_CASE("ZMQMetadataSocket","[ZeroMQ]") { REQUIRE(msg_out); REQUIRE(msg_out->msg_type == CBORImageType::END); REQUIRE(!sub_socket.Receive(msg1, true)); -} \ No newline at end of file +} +// Image threads add at the same time; batches are cut and sent outside the lock, a batch larger than +// 20 kB per image (2D azimuthal integration) still fits, and every image arrives exactly once before END. +TEST_CASE("ZMQMetadataSocket_Concurrent","[ZeroMQ]") { + ZMQMetadataSocket metadata_socket("tcp://0.0.0.0:*"); + metadata_socket.Period(std::chrono::milliseconds(50)); + ZMQSocket sub_socket(ZMQSocketType::Sub); + sub_socket.Connect(metadata_socket.GetAddress()); + sub_socket.SubscribeAll(); + sub_socket.ReceiveTimeout(std::chrono::seconds(2)); + std::this_thread::sleep_for(std::chrono::seconds(2)); + + constexpr int nthreads = 8; + constexpr int nimages = 500; + + metadata_socket.StartDataCollection(StartMessage{}); + std::vector threads; + for (int t = 0; t < nthreads; t++) + threads.emplace_back([&, t] { + for (int i = 0; i < nimages; i++) { + DataMessage msg{.number = t * nimages + i}; + msg.az_int_profile.resize(36 * 300, 1.0f); + msg.spots.resize(1000); + metadata_socket.AddDataMessage(msg); + } + }); + for (auto &t: threads) + t.join(); + metadata_socket.EndDataCollection(EndMessage{}); + + ZMQMessage msg; + REQUIRE(sub_socket.Receive(msg, true)); + auto msg_out = CBORStream2Deserialize(msg.data(), msg.size()); + REQUIRE(msg_out->msg_type == CBORImageType::START); + + std::vector numbers; + while (true) { + REQUIRE(sub_socket.Receive(msg, true)); + msg_out = CBORStream2Deserialize(msg.data(), msg.size()); + REQUIRE(msg_out); + if (msg_out->msg_type == CBORImageType::END) + break; + REQUIRE(msg_out->msg_type == CBORImageType::METADATA); + for (const auto &i: msg_out->metadata->images) { + CHECK(i.az_int_profile.size() == 36 * 300); + numbers.push_back(i.number); + } + } + std::sort(numbers.begin(), numbers.end()); + REQUIRE(numbers.size() == nthreads * nimages); + for (int i = 0; i < nthreads * nimages; i++) + REQUIRE(numbers[i] == i); + REQUIRE(!sub_socket.Receive(msg, true)); +} + +// Disabled socket: START and END are still published, no image is kept or sent +TEST_CASE("ZMQMetadataSocket_Disabled","[ZeroMQ]") { + ZMQMetadataSocket metadata_socket("tcp://0.0.0.0:*"); + metadata_socket.Period(std::nullopt); + ZMQSocket sub_socket(ZMQSocketType::Sub); + sub_socket.Connect(metadata_socket.GetAddress()); + sub_socket.SubscribeAll(); + sub_socket.ReceiveTimeout(std::chrono::seconds(2)); + std::this_thread::sleep_for(std::chrono::seconds(2)); + + metadata_socket.StartDataCollection(StartMessage{}); + for (int i = 0; i < 100; i++) + metadata_socket.AddDataMessage(DataMessage{.number = i}); + metadata_socket.EndDataCollection(EndMessage{}); + + ZMQMessage msg; + REQUIRE(sub_socket.Receive(msg, true)); + REQUIRE(CBORStream2Deserialize(msg.data(), msg.size())->msg_type == CBORImageType::START); + REQUIRE(sub_socket.Receive(msg, true)); + REQUIRE(CBORStream2Deserialize(msg.data(), msg.size())->msg_type == CBORImageType::END); + REQUIRE(!sub_socket.Receive(msg, true)); +}