leonarski_fandClaude Fable 5.1
a5b4529f9c
A send that blocked for over a second drops that image and turns the socket non-blocking for the rest of the run, so every later image that finds the queue full is dropped too - without a word in the log. Log the moment it happens, once per run, with the image number. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
80 lines
2.9 KiB
C++
80 lines
2.9 KiB
C++
// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#include "ZMQStream2PusherSocket.h"
|
|
|
|
#include <spdlog/spdlog.h>
|
|
|
|
void zmq_socket_free(void *data, void *hint) {
|
|
auto z = (ZeroCopyReturnValue *) hint;
|
|
z->release();
|
|
}
|
|
|
|
ZMQStream2PusherSocket::ZMQStream2PusherSocket(const std::string &addr,
|
|
std::optional<int32_t> send_buffer_high_watermark,
|
|
std::optional<int32_t> 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);
|
|
active = true;
|
|
send_future = std::async(std::launch::async, &ZMQStream2PusherSocket::WriterThread, this);
|
|
}
|
|
|
|
void ZMQStream2PusherSocket::StopWriterThread() {
|
|
std::unique_lock ul(m);
|
|
if (!active)
|
|
return; // Not active so no need to stop anything
|
|
|
|
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)) {
|
|
if (!transmission_error)
|
|
spdlog::warn("ZeroMQ image socket {}: send blocked for over a second - image {} dropped, and "
|
|
"images that find the queue full are dropped for the rest of the run",
|
|
s.GetEndpointName(), e.z->GetImageNumber());
|
|
transmission_error = true;
|
|
// The slot is already released: on a failed send SendZeroCopy closes the message, and
|
|
// closing a zmq_msg_init_data message runs its free function (zmq_socket_free).
|
|
}
|
|
e = queue.GetBlocking();
|
|
}
|
|
}
|
|
|
|
void ZMQStream2PusherSocket::SendImage(ZeroCopyReturnValue &z) {
|
|
queue.PutBlocking(ImagePusherQueueElement{
|
|
.image_data = static_cast<uint8_t *>(z.GetImage()),
|
|
.z = &z,
|
|
.end = false
|
|
});
|
|
}
|