v1.0.0-rc.153 (#63)

This is an UNSTABLE release. It includes many experimental features, as well as many AI generated fixes. We recommend using rc.152 for production use.

* jfjoch_broker: Add EXPERIMENTAL pixelrefine mode for image processing
* jfjoch_broker: Allow to load user mask from 8-bit and 16-bit TIFF files
* jfjoch_broker: Add ROI calculation in non-FPGA workflow
* jfjoch_broker: Fixes to TCP image pusher
* jfjoch_broker: Remove NUMA bindings
* jfjoch_broker: Improvements to indexing
* jfjoch_broker: For PSI EIGER, trimming energies are taken from the detector configuration (now compulsory) instead of hardcoded values
* jfjoch_writer: Save ROI definitions and the per-pixel ROI bitmap in the master file; azimuthal ROIs support phi (angular) sectors
* jfjoch_viewer: Major redesign with dockable panels and saved layouts, plus on-canvas creation/move/resize of box, circle and azimuthal ROIs
* jfjoch_viewer: Run jfjoch_process reprocessing jobs from inside the GUI and overlay per-run results

Reviewed-on: #63

image_puller/TCPImagePuller.cpp

@@ -54,6 +54,7 @@ TCPImagePuller::TCPImagePuller(const std::string &tcp_addr,
 
     receiver_thread = std::thread(&TCPImagePuller::ReceiverThread, this);
     cbor_thread = std::thread(&TCPImagePuller::CBORThread, this);
+    heartbeat_thread = std::thread(&TCPImagePuller::HeartbeatThread, this);
 
     if (!repub_address.empty()) {
         repub_socket = std::make_unique<ZMQSocket>(ZMQSocketType::Push);
@@ -91,6 +92,8 @@ bool TCPImagePuller::SendAll(const void *buf, size_t len) {
 }
 
 bool TCPImagePuller::SendAck(const PullerAckMessage &ack) {
+    std::lock_guard lg(send_mutex);
+
     TcpFrameHeader h{};
     h.type = static_cast<uint16_t>(TCPFrameType::ACK);
     h.run_number = ack.run_number;
@@ -319,7 +322,12 @@ void TCPImagePuller::ReceiverThread() {
                 TcpFrameHeader pong{};
                 pong.type = static_cast<uint16_t>(TCPFrameType::KEEPALIVE);
                 pong.payload_size = 0;
-                if (!SendAll(&pong, sizeof(pong))) {
+                bool pong_ok;
+                {
+                    std::lock_guard lg(send_mutex);
+                    pong_ok = SendAll(&pong, sizeof(pong));
+                }
+                if (!pong_ok) {
                     logger.Info("Keepalive pong send failed, reconnecting to " + addr);
                     CloseSocket();
                     std::this_thread::sleep_for(std::chrono::milliseconds(20));
@@ -365,6 +373,37 @@ void TCPImagePuller::ReceiverThread() {
     cbor_fifo.PutBlocking(ImagePullerOutput{});
 }
 
+void TCPImagePuller::HeartbeatThread() {
+    // While connected, periodically tell the pusher we are alive even if the
+    // consuming pipeline is stalled (e.g. blocked on a slow filesystem). This lets
+    // the pusher distinguish a busy-but-healthy writer from a dead one and keep
+    // waiting through arbitrarily long backpressure instead of dropping the run.
+    while (!disconnect) {
+        // Sleep in small slices so shutdown stays prompt.
+        for (int i = 0; i < 5 && !disconnect; i++)
+            std::this_thread::sleep_for(HeartbeatInterval / 5);
+        if (disconnect)
+            break;
+
+        {
+            std::unique_lock ul(fd_mutex);
+            if (fd < 0)
+                continue; // Not connected; ReceiverThread is (re)establishing the link.
+        }
+
+        TcpFrameHeader h{};
+        h.type = static_cast<uint16_t>(TCPFrameType::BUSY);
+        h.payload_size = 0;
+        h.ack_fifo_occupancy = cbor_fifo.GetCurrentUtilization();
+        h.ack_fifo_max_occupancy = cbor_fifo.Size();
+
+        // Best effort: a failure here just means the socket is gone, which
+        // ReceiverThread will detect and reconnect on its own.
+        std::lock_guard lg(send_mutex);
+        SendAll(&h, sizeof(h));
+    }
+}
+
 void TCPImagePuller::Disconnect() {
     if (disconnect.exchange(true))
         return;
@@ -377,4 +416,6 @@ void TCPImagePuller::Disconnect() {
         cbor_thread.join();
     if (repub_thread.joinable())
         repub_thread.join();
+    if (heartbeat_thread.joinable())
+        heartbeat_thread.join();
 }