diff --git a/common/ZMQWrappers.cpp b/common/ZMQWrappers.cpp index a969838c..adc1886a 100644 --- a/common/ZMQWrappers.cpp +++ b/common/ZMQWrappers.cpp @@ -97,7 +97,7 @@ bool ZMQSocket::SendZeroCopy(void *data, size_t size,void (*callback)(void *, vo auto result = zmq_msg_send(&msg, socket, (blocking ? 0 : ZMQ_DONTWAIT)); zmq_msg_close(&msg); - if (result == 0) + if (result == size) return true; if ((errno == EAGAIN) || (errno == EINTR)) diff --git a/image_pusher/CMakeLists.txt b/image_pusher/CMakeLists.txt index 19c04df2..f5cb4ccd 100644 --- a/image_pusher/CMakeLists.txt +++ b/image_pusher/CMakeLists.txt @@ -7,6 +7,8 @@ ADD_LIBRARY(ImagePusher STATIC HDF5FilePusher.cpp HDF5FilePusher.h NonePusher.cpp NonePusher.h + ZMQStream2PusherSocket.cpp + ZMQStream2PusherSocket.h ) TARGET_LINK_LIBRARIES(ImagePusher JFJochZMQ CBORStream2FrameSerialize JFJochCommon Compression JFJochWriter) \ No newline at end of file diff --git a/image_pusher/HDF5FilePusher.cpp b/image_pusher/HDF5FilePusher.cpp index 9b7a591c..7af63e68 100644 --- a/image_pusher/HDF5FilePusher.cpp +++ b/image_pusher/HDF5FilePusher.cpp @@ -42,10 +42,8 @@ bool HDF5FilePusher::SendImage(const uint8_t *image_data, size_t image_size, int } void HDF5FilePusher::SendImage(ZeroCopyReturnValue &z) { - writer_queue.PutBlocking(HDF5FilePusherQueueElement{ + writer_queue.PutBlocking(ImagePusherQueueElement{ .image_data = (uint8_t *) z.GetImage(), - .image_size = z.GetImageSize(), - .image_number = z.GetImageNumber(), .z = &z, .end = false }); @@ -60,9 +58,9 @@ bool HDF5FilePusher::SendCalibration(const CompressedImage &message) { } void HDF5FilePusher::WriterThread() { - HDF5FilePusherQueueElement e = writer_queue.GetBlocking(); + ImagePusherQueueElement e = writer_queue.GetBlocking(); while (!e.end) { - SendImage(e.image_data, e.image_size, e.image_number); + SendImage(e.image_data, e.z->GetImageSize(), e.z->GetImageNumber()); e.z->release(); e = writer_queue.GetBlocking(); } diff --git a/image_pusher/HDF5FilePusher.h b/image_pusher/HDF5FilePusher.h index 3a161fa9..4a1c0940 100644 --- a/image_pusher/HDF5FilePusher.h +++ b/image_pusher/HDF5FilePusher.h @@ -11,19 +11,12 @@ #include "../writer/FileWriter.h" #include "../common/ThreadSafeFIFO.h" -struct HDF5FilePusherQueueElement { - const uint8_t *image_data; - size_t image_size; - int64_t image_number; - ZeroCopyReturnValue *z; - bool end; -}; - class HDF5FilePusher : public ImagePusher { std::unique_ptr writer; std::mutex m; std::future writer_future; - ThreadSafeFIFO writer_queue; + ThreadSafeFIFO writer_queue; + void WriterThread(); public: // Thread safety: StartDataCollection, EndDataCollection and SendCalibration must run poorly in serial context // SendImage can be executed in parallel @@ -33,7 +26,7 @@ public: void SendImage(ZeroCopyReturnValue &z) override; bool SendCalibration(const CompressedImage &message) override; - void WriterThread(); + std::string PrintSetup() const override; }; diff --git a/image_pusher/ImagePusher.h b/image_pusher/ImagePusher.h index afcb2a3d..be030171 100644 --- a/image_pusher/ImagePusher.h +++ b/image_pusher/ImagePusher.h @@ -14,6 +14,12 @@ #include "../common/ZeroCopyReturnValue.h" #include "../common/Logger.h" +struct ImagePusherQueueElement { + const uint8_t *image_data; + ZeroCopyReturnValue *z; + bool end; +}; + void PrepareCBORImage(DataMessage& message, const DiffractionExperiment &experiment, void *image, size_t image_size); diff --git a/image_pusher/ZMQStream2Pusher.cpp b/image_pusher/ZMQStream2Pusher.cpp index 79425950..94ab1b00 100644 --- a/image_pusher/ZMQStream2Pusher.cpp +++ b/image_pusher/ZMQStream2Pusher.cpp @@ -14,13 +14,7 @@ ZMQStream2Pusher::ZMQStream2Pusher(const std::vector &addr, throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "No writer ZMQ address provided"); for (const auto &a : addr) { - auto s = std::make_unique(ZMQSocketType::Push); - if (send_buffer_size) - s->SendBufferSize(send_buffer_size.value()); - if (send_buffer_high_watermark) - s->SendWaterMark(send_buffer_high_watermark.value()); - s->SendTimeout(std::chrono::seconds(5)); // 5 seconds should be more than enough to flush buffers and to still give fast response - s->Bind(a); + auto s = std::make_unique(a, send_buffer_high_watermark, send_buffer_size); socket.emplace_back(std::move(s)); } } @@ -28,20 +22,16 @@ ZMQStream2Pusher::ZMQStream2Pusher(const std::vector &addr, bool ZMQStream2Pusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) { if (!socket.empty()) { auto socket_number = (image_number / images_per_file) % socket.size(); - return socket[socket_number]->Send(image_data, image_size, false); - } else - return false; + return socket[socket_number]->Send(image_data, image_size); + } + return false; } -void zmq_socket_free(void *data, void *hint) { - auto z = (ZeroCopyReturnValue *) hint; - z->release(); -} void ZMQStream2Pusher::SendImage(ZeroCopyReturnValue &z) { if (!socket.empty()) { auto socket_number = (z.GetImageNumber() / images_per_file) % socket.size(); - socket[socket_number]->SendZeroCopy(z.GetImage(), z.GetImageSize(), zmq_socket_free, &z); + socket[socket_number]->SendImage(z); } else z.release(); } @@ -59,10 +49,13 @@ void ZMQStream2Pusher::StartDataCollection(StartMessage& message) { if (i > 0) message.write_master_file = false; // Only writer on first socket is asked to write master file serializer.SerializeSequenceStart(message); - if (!socket[i]->Send(serialization_buffer.data(), serializer.GetBufferSize(), true)) + if (!socket[i]->Send(serialization_buffer.data(), serializer.GetBufferSize())) throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Timeout on pushing start message on addr " + socket[i]->GetEndpointName()); } + + for (const auto & i : socket) + i->StartWriterThread(); } bool ZMQStream2Pusher::SendCalibration(const CompressedImage &message) { @@ -71,15 +64,17 @@ bool ZMQStream2Pusher::SendCalibration(const CompressedImage &message) { serializer.SerializeCalibration(message); - return socket[0]->Send(serialization_buffer.data(), serializer.GetBufferSize(), true); + return socket[0]->Send(serialization_buffer.data(), serializer.GetBufferSize()); } bool ZMQStream2Pusher::EndDataCollection(const EndMessage& message) { serializer.SerializeSequenceEnd(message); bool ret = true; + for (auto &s: socket) { - if (!s->Send(serialization_buffer.data(), serializer.GetBufferSize(), true)) + s->StopWriterThread(); + if (!s->Send(serialization_buffer.data(), serializer.GetBufferSize())) ret = false; } return ret; @@ -125,4 +120,4 @@ std::string ZMQStream2Pusher::PrintSetup() const { for (const auto &s: socket) output += s->GetEndpointName() + " "; return output; -} \ No newline at end of file +} diff --git a/image_pusher/ZMQStream2Pusher.h b/image_pusher/ZMQStream2Pusher.h index 23c26b93..ea50110b 100644 --- a/image_pusher/ZMQStream2Pusher.h +++ b/image_pusher/ZMQStream2Pusher.h @@ -4,18 +4,17 @@ #ifndef JUNGFRAUJOCH_ZMQSTREAM2PUSHER_H #define JUNGFRAUJOCH_ZMQSTREAM2PUSHER_H -#include - #include "ImagePusher.h" #include "../common/ZMQWrappers.h" #include "../preview/PreviewCounter.h" #include "ZMQWriterNotificationPuller.h" +#include "ZMQStream2PusherSocket.h" class ZMQStream2Pusher : public ImagePusher { std::vector serialization_buffer; CBORStream2Serializer serializer; - std::vector> socket; + std::vector> socket; std::unique_ptr writer_notification_socket; diff --git a/image_pusher/ZMQStream2PusherSocket.cpp b/image_pusher/ZMQStream2PusherSocket.cpp new file mode 100644 index 00000000..8b3580b9 --- /dev/null +++ b/image_pusher/ZMQStream2PusherSocket.cpp @@ -0,0 +1,71 @@ +// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include "ZMQStream2PusherSocket.h" + +void zmq_socket_free(void *data, void *hint) { + auto z = (ZeroCopyReturnValue *) hint; + z->release(); +} + +ZMQStream2PusherSocket::ZMQStream2PusherSocket(const std::string &addr, + std::optional send_buffer_high_watermark, + std::optional send_buffer_size) +: s(ZMQSocketType::Push) { + if (send_buffer_size) + s.SendBufferSize(send_buffer_size.value()); + if (send_buffer_high_watermark) + s.SendWaterMark(send_buffer_high_watermark.value()); + s.SendTimeout(std::chrono::seconds(1)); // 1 second should be more than enough to flush buffers and to still give fast response + s.Bind(addr); +} + +std::string ZMQStream2PusherSocket::GetEndpointName() { + return s.GetEndpointName(); +} + +bool ZMQStream2PusherSocket::Send(const uint8_t *data, size_t size) { + std::unique_lock ul(m); + return s.Send(data, size, true); +} + +void ZMQStream2PusherSocket::StartWriterThread() { + std::unique_lock ul(m); + + send_future = std::async(std::launch::async, &ZMQStream2PusherSocket::WriterThread, this); + active = true; +} + +void ZMQStream2PusherSocket::StopWriterThread() { + std::unique_lock ul(m); + if (!active) + throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher not ready for sending"); + + active = false; + + if (send_future.valid()) { + queue.PutBlocking({.end = true}); + send_future.get(); + } else + throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher wasn't sending images"); +} + +void ZMQStream2PusherSocket::WriterThread() { + bool transmission_error = false; + ImagePusherQueueElement e = queue.GetBlocking(); + while (!e.end) { + // When hitting timeout on SendZeroCopy consider it a transmission error and switch to non-blocking communication + bool blocking_send = !transmission_error; + if (!s.SendZeroCopy(e.z->GetImage(), e.z->GetImageSize(), zmq_socket_free, e.z, blocking_send)) + transmission_error = true; + e = queue.GetBlocking(); + } +} + +void ZMQStream2PusherSocket::SendImage(ZeroCopyReturnValue &z) { + queue.PutBlocking(ImagePusherQueueElement{ + .image_data = static_cast(z.GetImage()), + .z = &z, + .end = false + }); +} diff --git a/image_pusher/ZMQStream2PusherSocket.h b/image_pusher/ZMQStream2PusherSocket.h new file mode 100644 index 00000000..6efc0d9b --- /dev/null +++ b/image_pusher/ZMQStream2PusherSocket.h @@ -0,0 +1,42 @@ +// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#ifndef JFJOCH_ZMQSTREAM2PUSHERSOCKET_H +#define JFJOCH_ZMQSTREAM2PUSHERSOCKET_H + +#include +#include +#include +#include + +#include "ImagePusher.h" +#include "../common/ZMQWrappers.h" +#include "../common/ThreadSafeFIFO.h" +#include "../common/ZeroCopyReturnValue.h" + +class ZMQStream2PusherSocket { + std::mutex m; + std::atomic active = false; + std::future send_future; + + ThreadSafeFIFO queue; + ZMQSocket s; + void WriterThread(); +public: + explicit ZMQStream2PusherSocket(const std::string& addr, + std::optional send_buffer_high_watermark, + std::optional send_buffer_size); + + std::string GetEndpointName(); + + // Strictly serial, as order of these is important + bool Send(const uint8_t *data, size_t size); + + void StartWriterThread(); + void StopWriterThread(); + + // Thread-safe + void SendImage(ZeroCopyReturnValue &z); +}; + +#endif //JFJOCH_ZMQSTREAM2PUSHERSOCKET_H \ No newline at end of file