leonarski_fandClaude Opus 5
c40e0eb11d
A re-initialisation left the previous run's ZMQImagePuller connected: JFJochServices::On replaces its own shared_ptr, but JFJochReceiverService keeps a second reference to the puller of the last run. Two PULL sockets on one PUSH peer share messages round-robin, so the orphaned puller silently took half of the DECTRIS stream - the start message included - and the receiver then waited for a start message that had already been discarded. Once such a run was cancelled, nothing drained the orphaned puller's outside_fifo any more. Its CBOR thread parked in PutBlocking on the full queue (suspend is only tested before the put), the puller thread backed up behind it in cbor_fifo, and neither could reach the disconnect flag again. The next JFJochReceiverService::Start dropped the last reference to that puller, so ~ZMQImagePuller joined two threads that could never exit - while holding state_mutex, inside a calibration sequence that itself holds the state machine's mutex. The whole control plane froze with no way to cancel; only a restart got out of it. * ThreadSafeFIFO gains Stop(), which releases every waiter and makes further blocking operations return at once. Clear() now notifies c_full as well: clearing a full queue used to leave the blocked producer asleep, since the next Get on an empty queue notifies no one. * ZMQImagePuller::Disconnect and TCPImagePuller::Disconnect stop their queues before joining, so a puller whose consumer is gone can always shut down. * JFJochServices::On and ::Off disconnect the previous puller explicitly rather than relying on the shared_ptr going away, so two readers never share the detector stream. ZMQImagePuller_DisconnectWithFullQueue covers the shutdown; it hangs on the previous code. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012bew392LTGP2fkhfRJsMcB
371 lines
14 KiB
C++
371 lines
14 KiB
C++
// 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);
|
|
}
|