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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

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<uint8_t> 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);

preview/ZMQMetadataSocket.h

@@ -23,12 +23,12 @@ class ZMQMetadataSocket {
 
     MetadataMessage metadata_message;
     ZMQSocket socket;
-    std::vector<uint8_t> buffer;
+    std::vector<uint8_t> 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);
 

tests/ZMQMetadataSocketTest.cpp

@@ -1,6 +1,8 @@
 // SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
 // SPDX-License-Identifier: GPL-3.0-only
 
+#include <algorithm>
+#include <thread>
 #include <catch2/catch_all.hpp>
 
 #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));
-}
+}
+// 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<std::thread> 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<int64_t> 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));
+}