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)); +}