From 67398de47fb614c74f2a199d753478661d16751c Mon Sep 17 00:00:00 2001 From: Andrej Babic Date: Thu, 4 Jun 2020 11:27:55 +0200 Subject: [PATCH] Remove pulse_id from module alignment --- sf-stream/include/LiveRecvModule.hpp | 2 +- sf-stream/src/LiveRecvModule.cpp | 66 +++++++++++++--------------- 2 files changed, 31 insertions(+), 37 deletions(-) diff --git a/sf-stream/include/LiveRecvModule.hpp b/sf-stream/include/LiveRecvModule.hpp index 3020573..d0820ef 100644 --- a/sf-stream/include/LiveRecvModule.hpp +++ b/sf-stream/include/LiveRecvModule.hpp @@ -19,7 +19,7 @@ class LiveRecvModule { void* connect_socket(size_t module_id); void receive_thread(); - void recv_single_module(void* socket, ModuleFrame* metadata, char* data); + void recv_single_module(void* socket, ModuleFrame* meta, char* data); uint64_t align_modules(const std::vector& sockets, ModuleFrameBuffer *metadata, char *data); diff --git a/sf-stream/src/LiveRecvModule.cpp b/sf-stream/src/LiveRecvModule.cpp index d18054b..a1f5b6f 100644 --- a/sf-stream/src/LiveRecvModule.cpp +++ b/sf-stream/src/LiveRecvModule.cpp @@ -13,11 +13,11 @@ LiveRecvModule::LiveRecvModule( const size_t n_modules, void* ctx_, const string& ipc_prefix) : - queue_(queue_), - n_modules_(n_modules), - ctx_(ctx_), - ipc_prefix_(ipc_prefix), - is_receiving_(true) + queue_(queue_), + n_modules_(n_modules), + ctx_(ctx_), + ipc_prefix_(ipc_prefix), + is_receiving_(true) { receiving_thread_ = thread(&LiveRecvModule::receive_thread, this); } @@ -35,17 +35,18 @@ void LiveRecvModule::stop() void* LiveRecvModule::connect_socket(size_t module_id) { - void* sock = zmq_socket(ctx_, ZMQ_SUB); - if (sock == nullptr) { + void* socket = zmq_socket(ctx_, ZMQ_SUB); + if (socket == nullptr) { throw runtime_error(zmq_strerror(errno)); } int rcvhwm = STREAM_RCVHWM; - if (zmq_setsockopt(sock, ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0) { + if (zmq_setsockopt(socket, ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0) { throw runtime_error(zmq_strerror(errno)); } + int linger = 0; - if (zmq_setsockopt(sock, ZMQ_LINGER, &linger, sizeof(linger)) != 0) { + if (zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)) != 0) { throw runtime_error(zmq_strerror(errno)); } @@ -53,45 +54,38 @@ void* LiveRecvModule::connect_socket(size_t module_id) ipc_addr << ipc_prefix_ << module_id; const auto ipc = ipc_addr.str(); - if (zmq_connect(sock, ipc.c_str()) != 0) { + if (zmq_connect(socket, ipc.c_str()) != 0) { throw runtime_error(zmq_strerror(errno)); } - if (zmq_setsockopt(sock, ZMQ_SUBSCRIBE, "", 0) != 0) { + if (zmq_setsockopt(socket, ZMQ_SUBSCRIBE, "", 0) != 0) { throw runtime_error(zmq_strerror(errno)); } - return sock; + return socket; } void LiveRecvModule::recv_single_module( - void* socket, ModuleFrame* metadata, char* data) + void* socket, ModuleFrame* meta, char* data) { - auto n_bytes_metadata = zmq_recv( - socket, - metadata, - sizeof(ModuleFrame), - 0); + auto n_bytes_meta = zmq_recv(socket, meta, sizeof(ModuleFrame), 0); - if (n_bytes_metadata == -1) { + if (n_bytes_meta == -1) { throw runtime_error(zmq_strerror(errno)); - }else if (n_bytes_metadata != sizeof(ModuleFrame)) { + } + if (n_bytes_meta != sizeof(ModuleFrame)) { throw runtime_error("Stream header of wrong size."); } - - if (metadata->pulse_id == 0) { + if (meta->pulse_id == 0) { throw runtime_error("Received invalid pulse_id=0."); } - auto n_bytes_image = zmq_recv( - socket, - data, - MODULE_N_BYTES, - 0); + auto n_bytes_frame = zmq_recv(socket, data, MODULE_N_BYTES, 0); - if (n_bytes_image == -1) { + if (n_bytes_frame == -1) { throw runtime_error(zmq_strerror(errno)); - } else if (n_bytes_image != MODULE_N_BYTES) { + } + if (n_bytes_frame != MODULE_N_BYTES) { throw runtime_error("Stream data of wrong size."); } } @@ -153,10 +147,9 @@ void LiveRecvModule::receive_thread() data + (MODULE_N_BYTES * i_module)); } - auto current_pulse_id = align_modules(sockets, metadata, data); + align_modules(sockets, metadata, data); queue_.commit(); - current_pulse_id++; while(is_receiving_.load(memory_order_relaxed)) { auto slot_id = queue_.reserve(); @@ -169,6 +162,7 @@ void LiveRecvModule::receive_thread() metadata = queue_.get_metadata_buffer(slot_id); data = queue_.get_data_buffer(slot_id); + uint64_t frame_pulse_id; bool sync_needed = false; for (size_t i_module = 0; i_module < n_modules_; i_module++) { auto& module_metadata = metadata->module[i_module]; @@ -178,7 +172,10 @@ void LiveRecvModule::receive_thread() &module_metadata, data + (MODULE_N_BYTES * i_module)); - if (module_metadata.pulse_id != current_pulse_id) { + if (i_module == 0) { + frame_pulse_id = module_metadata.pulse_id; + + } else if (frame_pulse_id != module_metadata.pulse_id) { sync_needed = true; } } @@ -186,9 +183,7 @@ void LiveRecvModule::receive_thread() if (sync_needed) { auto start_time = chrono::steady_clock::now(); - auto new_pulse_id = align_modules(sockets, metadata, data); - auto lost_pulses = new_pulse_id - current_pulse_id; - current_pulse_id = new_pulse_id; + auto lost_pulses = align_modules(sockets, metadata, data); auto end_time = chrono::steady_clock::now(); auto us_duration = chrono::duration_cast( @@ -200,7 +195,6 @@ void LiveRecvModule::receive_thread() } queue_.commit(); - current_pulse_id++; } for (size_t i = 0; i < n_modules_; i++) {