Thread fine-tuning

core-buffer/include/buffer_config.hpp

@@ -29,10 +29,10 @@ namespace core_buffer {
     const size_t BUFFER_INTERNAL_QUEUE_SIZE = 1000;
 
     // Time to sleep before retrying to read the queue.
-    const size_t BUFFER_QUEUE_RETRY_MS = 10;
+    const size_t BUFFER_QUEUE_RETRY_MS = 5;
 
     // Microseconds timeout for UDP recv.
-    const int BUFFER_UDP_US_TIMEOUT = 10 * 1000;
+    const int BUFFER_UDP_US_TIMEOUT = 5 * 1000;
 
     // Output queue length for buffer live stream.
     const int BUFFER_LIVE_SEND_HWM = 10;

core-buffer/src/UdpReceiver.cpp

@@ -51,16 +51,18 @@ void UdpReceiver::bind(const uint16_t port, const size_t usec_timeout)
         throw runtime_error("Cannot bind socket.");
     }
 
-//    struct timeval udp_socket_timeout;
-//    udp_socket_timeout.tv_sec = 0;
-//    udp_socket_timeout.tv_usec = usec_timeout;
-//
-//    setsockopt(
-//            socket_fd_,
-//            SOL_SOCKET,
-//            SO_RCVTIMEO,
-//            (const char*)&udp_socket_timeout,
-//            sizeof(struct timeval));
+    struct timeval udp_socket_timeout;
+    udp_socket_timeout.tv_sec = 0;
+    udp_socket_timeout.tv_usec = usec_timeout;
+
+    setsockopt(
+            socket_fd_,
+            SOL_SOCKET,
+            SO_RCVTIMEO,
+            (const char*)&udp_socket_timeout,
+            sizeof(struct timeval));
+
+    SO_RCVLOWAT
 }
 
 int UdpReceiver::receive_many(mmsghdr* msgs, const size_t n_msgs)
@@ -70,7 +72,7 @@ int UdpReceiver::receive_many(mmsghdr* msgs, const size_t n_msgs)
 
 bool UdpReceiver::receive(void* buffer, size_t buffer_n_bytes)
 {
-    auto data_len = recv(socket_fd_, buffer, buffer_n_bytes, MSG_DONTWAIT);
+    auto data_len = recv(socket_fd_, buffer, buffer_n_bytes, 0);
 
     if (data_len < 0) {
         return false;

core-buffer/src/UdpRecvModule.cpp

@@ -73,7 +73,6 @@ void UdpRecvModule::receive_thread(const uint16_t udp_port)
             if (!udp_receiver.receive(
                     &packet_buffer,
                     JUNGFRAU_BYTES_PER_PACKET)) {
-                this_thread::yield();
                 continue;
             }
 
@@ -104,6 +103,7 @@ void UdpRecvModule::receive_thread(const uint16_t udp_port)
             {
                 queue_.commit();
                 reserve_next_frame_buffers(frame_metadata, frame_buffer);
+                this_thread::yield();
             }
         }
 

sf-buffer/src/sf_buffer.cpp

@@ -114,7 +114,5 @@ int main (int argc, char *argv[]) {
             n_corrupted_frames = 0;
             n_missed_frames = 0;
         }
-
-        this_thread::yield();
     }
 }