mx/Jungfraujoch
3
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 81 Wiki Activity
Files
8fa5b14bc03f97fa61f7d47b5cc7315cd412445f
Jungfraujoch/tests/ZMQImagePusherTest.cpp
T
leonarski_fandClaude Fable 5.1 1be6093429 Writer: received images live in a byte-bounded ring buffer until written and republished 
Every stage between the socket and the HDF5 write was bounded by a message count and never by
bytes: ZeroMQ receive watermark 100, cbor_fifo 200, outside_fifo 1000, repub_fifo 200 - up to
some 1500 images in flight per writer process, each a whole compressed image, so 6-12 GB for a
JUNGFRAU 4M and 30-60 GB for a 9M, and the writer's memory was a quantity nobody had chosen.

ReceiveRingBuffer is a fixed ring of bytes owned by the ImagePuller base class, so the ZeroMQ
and TCP pullers share it. A message is received straight into it (TCP reads the payload into
the ring; ZeroMQ copies the message in) and travels as a shared_ptr region in ImagePullerOutput;
the deserialised image points into that region, republish sends it as is, and the bytes come
back when the last holder lets go. Space is reclaimed in arrival order: a region released out
of turn - republish finishes with messages in its own order - is marked and waits for the ones
in front of it. Because of that, what republish may hold is capped at a quarter of the ring,
and an image over the cap is not republished, exactly as one that finds the republish queue
full; otherwise a stalled republish consumer would pin the ring and slow the writer to its
rate. A message larger than the whole ring (a START or END of a very long run) goes on the heap,
so it cannot wedge the receiver.

A full ring stalls the receive thread, and what that does depends on the stream. On TCP the
pusher waits behind the writer's BUSY heartbeats and gives up only after a minute in which no
image at all was written. On ZeroMQ the pusher's send blocks for one second, then drops the
image and sends without waiting for the rest of the run - so on ZeroMQ a ring that is smaller
than the writer's stalls costs images, where the old 1500-message queues absorbed them. That
is the trade: the budget is now a number, and it has to be chosen. jfjoch_writer sets it with
-m/--rcv_ring_mb, default 2048 MB per writer process; the ZeroMQ receive watermark default goes
from 100 to 10 messages, since it now only counts what sits beyond the ring. The ring's peak
use is in the end-of-run log line, and docs/JFJOCH_WRITER.md has a Memory section with the
sizing rule.

The puller's own default is no ring at all, which keeps every other user as it was: the
broker's DECTRIS stream2 puller (JFJochServices, JFJochReceiverService) stays zero-copy with
the ZeroMQ message as the buffer, and the broker's direct HDF5 writer (HDF5FilePusher ->
FileWriter) never goes through a puller. The puller threads now also catch std::exception, as
the writer does since the previous commit: deserialising the END message of a million-image
run allocates a vector per per-image array, and a bad_alloc leaving the thread ended the process.

Tests: ReceiveRingBufferTest (reuse, wrap-around, full ring, out-of-order release, heap
fallback, zero size, stop, a region outliving its ring); ZMQImageCommTest_SmallRingBuffer (64
one-megabyte images through a three-image ring, nothing lost, peak within the bound); one of
the two writers in TCPImageCommTest_2Writers_WithAck runs on a three-image ring.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-13 12:51:34 +02:00

495 lines
20 KiB
C++
Raw Blame History

// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#include <random>
#include <catch2/catch_all.hpp>
#include "../image_puller/ZMQImagePuller.h"
#include "../image_pusher/ZMQStream2Pusher.h"
#include "../common/DiffractionSpot.h"
void test_puller(ZMQImagePuller *puller,
const DiffractionExperiment& x,
const std::vector<uint16_t> &image1,
int64_t nwriter,
int64_t writer_id,
std::vector<size_t> &diff_split,
std::vector<size_t> &diff_size,
std::vector<size_t> &diff_content,
std::vector<size_t> &nimages) {
auto timeout = std::chrono::minutes(3);
auto img = puller->PollImage(timeout);
if (!img || !img->cbor || !img->cbor->start_message) {
diff_content[writer_id]++;
return;
}
if ((!img->cbor->start_message->write_master_file) || (img->cbor->start_message->write_master_file.value() != (writer_id == 0)))
diff_content[writer_id]++;
img = puller->PollImage(timeout);
while (img && img->cbor && !img->cbor->end_message) {
if (img->cbor->data_message) {
if ((nwriter > 1) && ((img->cbor->data_message->number / 16) % nwriter != writer_id))
diff_split[writer_id]++;
if (img->cbor->data_message->image.GetCompressedSize() != x.GetPixelsNum() * sizeof(uint16_t))
diff_size[writer_id]++;
else if (memcmp(img->cbor->data_message->image.GetCompressed(), image1.data() + img->cbor->data_message->number * x.GetPixelsNum(),
x.GetPixelsNum() * sizeof(uint16_t)) != 0)
diff_content[writer_id]++;
if (img->cbor->data_message->image.GetWidth() != RAW_MODULE_COLS)
diff_content[writer_id]++;
if (img->cbor->data_message->image.GetHeight() != RAW_MODULE_LINES)
diff_content[writer_id]++;
if (img->cbor->data_message->image.GetByteDepth() != 2)
diff_content[writer_id]++;
if (img->cbor->data_message->image.GetCompressionAlgorithm() != CompressionAlgorithm::NO_COMPRESSION)
diff_content[writer_id]++;
nimages[writer_id]++;
}
img = puller->PollImage(timeout);
}
}
TEST_CASE("ZMQImageCommTest_1Writer","[ZeroMQ]") {
const size_t nframes = 256;
Logger logger(Catch::getResultCapture().getCurrentTestName());
DiffractionExperiment x(DetJF(1));
x.Raw();
x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
.ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
std::vector<DiffractionSpot> empty_spot_vector;
std::vector<float> empty_rad_int_profile;
REQUIRE(x.GetImageNum() == nframes);
std::mt19937 g1(1387);
std::uniform_int_distribution<uint16_t> dist;
std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
for (auto &i: image1) i = dist(g1);
// Puller needs to be declared first, but both objects need to exist till communication finished
// TODO: ImageSender should not allow if there are still completions to be done
ZMQStream2Pusher pusher({"ipc://*"});
std::vector<size_t> diff_size(1), diff_content(1), diff_split(1), nimages(1);
auto pusher_addr = pusher.GetAddress();
ZMQImagePuller puller(pusher_addr[0]);
std::thread sender_thread = std::thread([&] {
std::vector<uint8_t> serialization_buffer(16*1024*1024);
CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
StartMessage message {
.images_per_file = 16,
.write_master_file = true
};
EndMessage end_message{};
pusher.StartDataCollection(message);
for (int i = 0; i < nframes; i++) {
DataMessage data_message;
data_message.number = i;
data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
x.GetPixelsNum() * sizeof(uint16_t),
x.GetXPixelsNum(),
x.GetYPixelsNum(),
x.GetImageMode(),
x.GetCompressionAlgorithm());
serializer.SerializeImage(data_message);
pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
});
std::thread puller_thread(test_puller, &puller, std::cref(x), std::cref(image1), 1, 0,
std::ref(diff_split), std::ref(diff_size), std::ref(diff_content),
std::ref(nimages));
sender_thread.join();
puller_thread.join();
puller.Disconnect();
REQUIRE(nimages[0] == nframes);
REQUIRE(diff_size[0] == 0);
REQUIRE(diff_content[0] == 0);
}
TEST_CASE("ZMQImageCommTest_2Writers","[ZeroMQ]") {
const size_t nframes = 256;
Logger logger(Catch::getResultCapture().getCurrentTestName());
DiffractionExperiment x(DetJF(1));
x.Raw();
x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
.ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
REQUIRE(x.GetImageNum() == nframes);
std::mt19937 g1(1387);
std::uniform_int_distribution<uint16_t> dist;
std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
for (auto &i: image1) i = dist(g1);
std::vector<DiffractionSpot> empty_spot_vector;
std::vector<float> empty_rad_int_profile;
std::vector<std::string> zmq_addr;
int64_t npullers = 2;
for (int i = 0; i < npullers; i++)
zmq_addr.push_back("ipc://*");
ZMQStream2Pusher pusher(zmq_addr);
// Puller needs to be declared first, but both objects need to exist till communication finished
// TODO: ImageSender should not allow if there are still completions to be done
std::vector<std::unique_ptr<ZMQImagePuller> > puller;
auto pusher_addr = pusher.GetAddress();
REQUIRE(pusher_addr.size() == 2);
for (int i = 0; i < npullers; i++) {
puller.push_back(std::make_unique<ZMQImagePuller>(pusher_addr[i]));
}
std::vector<size_t> diff_size(npullers), diff_content(npullers), diff_split(npullers), nimages(npullers);
std::thread sender_thread = std::thread([&] {
std::vector<uint8_t> serialization_buffer(16*1024*1024);
CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
StartMessage message {
.images_per_file = 16,
.write_master_file = true
};
EndMessage end_message{};
pusher.StartDataCollection(message);
for (int i = 0; i < nframes; i++) {
DataMessage data_message;
data_message.number = i;
data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
x.GetPixelsNum() * sizeof(uint16_t),
x.GetXPixelsNum(),
x.GetYPixelsNum(),
x.GetImageMode(),
x.GetCompressionAlgorithm());
serializer.SerializeImage(data_message);
pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
});
std::vector<std::thread> puller_threads;
for (int i = 0; i < npullers; i++)
puller_threads.emplace_back(test_puller, puller[i].get(), std::cref(x),
std::cref(image1), npullers, i,
std::ref(diff_split), std::ref(diff_size), std::ref(diff_content), std::ref(nimages));
for (int i = 0; i < npullers; i++)
puller_threads[i].join();
sender_thread.join();
REQUIRE_NOTHROW(puller[0]->Disconnect());
REQUIRE_NOTHROW(puller[1]->Disconnect());
REQUIRE(nimages[0] == nframes / 2);
REQUIRE(nimages[1] == nframes / 2);
REQUIRE(diff_size[0] == 0);
REQUIRE(diff_content[0] == 0);
REQUIRE(diff_size[1] == 0);
REQUIRE(diff_content[1] == 0);
REQUIRE(diff_split[0] == 0);
REQUIRE(diff_split[1] == 0);
}
TEST_CASE("ZMQImageCommTest_4Writers","[ZeroMQ]") {
const size_t nframes = 255;
Logger logger(Catch::getResultCapture().getCurrentTestName());
DiffractionExperiment x(DetJF(1));
x.Raw();
x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
.ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
REQUIRE(x.GetImageNum() == nframes);
std::mt19937 g1(1387);
std::uniform_int_distribution<uint16_t> dist;
std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
for (auto &i: image1) i = dist(g1);
std::vector<DiffractionSpot> empty_spot_vector;
std::vector<float> empty_rad_int_profile;
std::vector<std::string> zmq_addr;
int64_t npullers = 4;
for (int i = 0; i < npullers; i++)
zmq_addr.push_back("ipc://*");
ZMQStream2Pusher pusher(zmq_addr);
auto pusher_addr = pusher.GetAddress();
REQUIRE(pusher_addr.size() == npullers);
// Puller needs to be declared first, but both objects need to exist till communication finished
// TODO: ImageSender should not allow if there are still completions to be done
std::vector<std::unique_ptr<ZMQImagePuller> > puller;
for (int i = 0; i < npullers; i++) {
puller.push_back(std::make_unique<ZMQImagePuller>(pusher_addr[i]));
}
std::vector<size_t> diff_size(npullers), diff_content(npullers), diff_split(npullers), nimages(npullers);
std::thread sender_thread = std::thread([&] {
std::vector<uint8_t> serialization_buffer(16*1024*1024);
CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
StartMessage message {
.images_per_file = 16,
.write_master_file = true
};
EndMessage end_message{};
pusher.StartDataCollection(message);
for (int i = 0; i < nframes; i++) {
DataMessage data_message;
data_message.number = i;
data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
x.GetPixelsNum() * sizeof(uint16_t),
x.GetXPixelsNum(),
x.GetYPixelsNum(),
x.GetImageMode(),
x.GetCompressionAlgorithm());
serializer.SerializeImage(data_message);
pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
});
std::vector<std::thread> puller_threads;
for (int i = 0; i < npullers; i++)
puller_threads.emplace_back(test_puller, puller[i].get(), std::cref(x),
std::cref(image1), npullers, i,
std::ref(diff_split), std::ref(diff_size), std::ref(diff_content), std::ref(nimages));
for (int i = 0; i < npullers; i++)
puller_threads[i].join();
sender_thread.join();
REQUIRE_NOTHROW(puller[0]->Disconnect());
REQUIRE_NOTHROW(puller[1]->Disconnect());
REQUIRE_NOTHROW(puller[2]->Disconnect());
REQUIRE_NOTHROW(puller[3]->Disconnect());
REQUIRE(nimages[0] == 64);
REQUIRE(nimages[1] == 64);
REQUIRE(nimages[2] == 64);
REQUIRE(nimages[3] == 63);
for (int i = 0; i < npullers; i++) {
REQUIRE(diff_size[i] == 0);
REQUIRE(diff_content[i] == 0);
REQUIRE(diff_split[i] == 0);
}
}
TEST_CASE("ZMQImageCommTest_NoWriter","[ZeroMQ]") {
Logger logger(Catch::getResultCapture().getCurrentTestName());
ZMQStream2Pusher pusher({"ipc://*"});
StartMessage msg{};
REQUIRE_THROWS(pusher.StartDataCollection(msg));
std::vector<uint8_t> test(512*1024, 11);
CompressedImage image(test, 1024, 512);
DataMessage data_message{
.number = 1,
.image = image
};
std::vector<uint8_t> v(16*1024*1024);
CBORStream2Serializer serializer(v.data(), v.size());
serializer.SerializeImage(data_message);
REQUIRE(!pusher.SendImage(v.data(), serializer.GetBufferSize(), 1));
EndMessage end_message{};
REQUIRE(!pusher.EndDataCollection(end_message));
}
// A puller whose queue nobody drains must still shut down. Before this was fixed, the CBOR thread
// parked in PutBlocking on the full outside_fifo, the puller thread backed up behind it in
// cbor_fifo, and Disconnect() hung for ever on the joins - which froze the whole broker, because
// the receiver service destroys the previous run's puller from Start(), holding its state mutex,
// which in a calibration sequence is itself held under the state machine's mutex.
TEST_CASE("ZMQImagePuller_DisconnectWithFullQueue","[ZeroMQ]") {
Logger logger(Catch::getResultCapture().getCurrentTestName());
ZMQSocket sender(ZMQSocketType::Push);
sender.SendWaterMark(4 * ImagePuller::DefaultQueueSize);
sender.Bind("ipc://*");
auto puller = std::make_unique<ZMQImagePuller>(sender.GetEndpointName());
std::vector<uint8_t> pixels(1024, 11);
DataMessage data_message{
.number = 1,
.image = CompressedImage(pixels, 32, 32)
};
std::vector<uint8_t> v(64 * 1024);
CBORStream2Serializer serializer(v.data(), v.size());
serializer.SerializeImage(data_message);
// Twice the queue size, so the CBOR thread is certainly blocked on a full queue by the end
for (int i = 0; i < 2 * ImagePuller::DefaultQueueSize; i++)
sender.Send(v.data(), serializer.GetBufferSize());
while (puller->GetCurrentFifoUtilization() < ImagePuller::DefaultQueueSize)
std::this_thread::sleep_for(std::chrono::milliseconds(10));
auto shutdown = std::async(std::launch::async, [&] { puller.reset(); });
REQUIRE(shutdown.wait_for(std::chrono::seconds(30)) == std::future_status::ready);
}
// A run that ends while the detector is still streaming leaves outside_fifo full and the CBOR
// thread blocked in PutBlocking on it. Suspend() cannot release it - the flag is only tested
// before the put - so the next run's ResumeAndClear has to. Nothing else ever would: a Get on an
// empty queue notifies no producer, so a queue emptied under a waiting producer leaves it asleep
// for good, and every collection after that one silently receives nothing at all - no images and
// no start message - while the puller looks perfectly healthy.
TEST_CASE("ZMQImagePuller_RecoversFromFullQueue","[ZeroMQ]") {
Logger logger(Catch::getResultCapture().getCurrentTestName());
ZMQSocket sender(ZMQSocketType::Push);
sender.SendWaterMark(8 * ImagePuller::DefaultQueueSize);
sender.Bind("ipc://*");
ZMQImagePuller puller(sender.GetEndpointName());
std::vector<uint8_t> pixels(1024, 11);
auto send_image = [&](int64_t number) {
DataMessage data_message{
.number = number,
.image = CompressedImage(pixels, 32, 32)
};
std::vector<uint8_t> v(64 * 1024);
CBORStream2Serializer serializer(v.data(), v.size());
serializer.SerializeImage(data_message);
sender.Send(v.data(), serializer.GetBufferSize());
};
// Images still arriving after the last run stopped consuming them
for (int i = 0; i < 2 * ImagePuller::DefaultQueueSize; i++)
send_image(1);
while (puller.GetCurrentFifoUtilization() < ImagePuller::DefaultQueueSize)
std::this_thread::sleep_for(std::chrono::milliseconds(10));
puller.Suspend(); // what JFJochServices::Stop does when the run ends
puller.ResumeAndClear(); // ... and what JFJochServices::Start does for the run after it
// The run after it must still be able to receive
send_image(4242);
bool received = false;
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(30);
while (!received && (std::chrono::steady_clock::now() < deadline)) {
auto msg = puller.PollImage(std::chrono::milliseconds(10));
if (msg && msg->cbor && msg->cbor->data_message && (msg->cbor->data_message->number == 4242))
received = true;
}
REQUIRE(received);
puller.Disconnect();
}
// The whole run goes through a ring that holds three images: the puller has to wait for space, the
// socket queue fills, the pusher blocks, and nothing is lost or reordered. The peak ring use proves
// the bound held.
TEST_CASE("ZMQImageCommTest_SmallRingBuffer","[ZeroMQ]") {
const size_t nframes = 64;
Logger logger(Catch::getResultCapture().getCurrentTestName());
DiffractionExperiment x(DetJF(1));
x.Raw();
x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
.ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
REQUIRE(x.GetImageNum() == nframes);
std::mt19937 g1(1387);
std::uniform_int_distribution<uint16_t> dist;
std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
for (auto &i: image1) i = dist(g1);
const size_t image_bytes = x.GetPixelsNum() * sizeof(uint16_t);
const size_t ring_size = 3 * image_bytes + 3 * 65536; // three images with their CBOR framing
ZMQStream2Pusher pusher({"ipc://*"});
std::vector<size_t> diff_size(1), diff_content(1), diff_split(1), nimages(1);
auto pusher_addr = pusher.GetAddress();
ZMQImagePuller puller(pusher_addr[0], "", {}, {}, ring_size);
REQUIRE(puller.GetRingBufferCapacity() == ring_size);
std::thread sender_thread = std::thread([&] {
std::vector<uint8_t> serialization_buffer(16*1024*1024);
CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
StartMessage message {
.images_per_file = 16,
.write_master_file = true
};
EndMessage end_message{};
pusher.StartDataCollection(message);
for (int i = 0; i < nframes; i++) {
DataMessage data_message;
data_message.number = i;
data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
image_bytes,
x.GetXPixelsNum(),
x.GetYPixelsNum(),
x.GetImageMode(),
x.GetCompressionAlgorithm());
serializer.SerializeImage(data_message);
pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
});
std::thread puller_thread(test_puller, &puller, std::cref(x), std::cref(image1), 1, 0,
std::ref(diff_split), std::ref(diff_size), std::ref(diff_content),
std::ref(nimages));
sender_thread.join();
puller_thread.join();
puller.Disconnect();
REQUIRE(nimages[0] == nframes);
REQUIRE(diff_size[0] == 0);
REQUIRE(diff_content[0] == 0);
CHECK(puller.GetMaxRingBufferUtilization() <= ring_size);
CHECK(puller.GetMaxRingBufferUtilization() >= image_bytes);
}
Reference in New Issue View Git Blame Copy Permalink