sf-daq/sf_daq_buffer
0
0
Fork 0
mirror of https://github.com/paulscherrerinstitute/sf_daq_buffer.git synced 2026-04-21 08:04:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix small things in LiveRecvModule

Browse Source
This commit is contained in:
babic_a
2020-05-15 13:12:08 +02:00
parent 590ff5b625
commit 01d986c4f0
2 changed files with 82 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core-buffer/test/test_LiveRecvModule.cpp
+76 -66
View File
@@ -2,75 +2,85 @@
#include "LiveRecvModule.hpp"
#include "gtest/gtest.h"
#include "buffer_config.hpp"
#include <future>
using namespace std;
using namespace core_buffer;
TEST(LiveRecvModule, basic_interaction) {
auto ctx = zmq_ctx_new();
size_t n_modules = 32;
size_t n_slots = 5;
FastQueue<ModuleFrameBuffer> queue(MODULE_N_BYTES * n_modules, n_slots);
LiveRecvModule recv_module(queue, n_modules, ctx, BUFFER_LIVE_IPC_URL);
this_thread::sleep_for(chrono::milliseconds(100));
zmq_ctx_destroy(ctx);
}
TEST(LiveRecvModule, transfer_test) {
auto ctx = zmq_ctx_new();
size_t n_modules = 32;
size_t n_slots = 5;
FastQueue<ModuleFrameBuffer> queue(MODULE_N_BYTES * n_modules, n_slots);
void *sockets[n_modules];
for (size_t i = 0; i < n_modules; i++) {
sockets[i] = zmq_socket(ctx, ZMQ_PUB);
int linger = 0;
if (zmq_setsockopt(sockets[i], ZMQ_LINGER, &linger,
sizeof(linger)) != 0) {
throw runtime_error(zmq_strerror(errno));
}
stringstream ipc_addr;
ipc_addr << BUFFER_LIVE_IPC_URL << i;
const auto ipc = ipc_addr.str();
if (zmq_bind(sockets[i], ipc.c_str()) != 0) {
throw runtime_error(zmq_strerror(errno));
}
}
LiveRecvModule recv_module(queue, n_modules, ctx, BUFFER_LIVE_IPC_URL);
// Nothing should be committed, queue, should be empty.
ASSERT_EQ(queue.read(), -1);
ModuleFrame metadata;
auto data = make_unique<char[]>(MODULE_N_BYTES);
for (size_t i = 0; i < n_modules; i++) {
metadata.pulse_id = 1;
metadata.frame_index = 2;
metadata.daq_rec = 3;
metadata.n_received_packets = 4;
metadata.module_id = i;
zmq_send(sockets[i], &metadata, sizeof(ModuleFrame), ZMQ_SNDMORE);
zmq_send(sockets[i], data.get(), MODULE_N_BYTES, 0);
}
this_thread::sleep_for(chrono::milliseconds(100));
auto slot_id = queue.read();
// We should have the first Detector frame in the buffer.
//ASSERT_NE(slot_id, -1);
for (size_t i = 0; i < n_modules; i++) {
zmq_close(sockets[i]);
}
zmq_ctx_destroy(ctx);
cout << "We are finished" << endl;
// auto ctx = zmq_ctx_new();
//
// size_t n_modules = 32;
// size_t n_slots = 5;
// FastQueue<ModuleFrameBuffer> queue(MODULE_N_BYTES * n_modules, n_slots);
//
// void *sockets[n_modules];
// for (size_t i = 0; i < n_modules; i++) {
// sockets[i] = zmq_socket(ctx, ZMQ_PUB);
//
// int linger = 0;
// if (zmq_setsockopt(sockets[i], ZMQ_LINGER, &linger,
// sizeof(linger)) != 0) {
// throw runtime_error(zmq_strerror(errno));
// }
//
// stringstream ipc_addr;
// ipc_addr << BUFFER_LIVE_IPC_URL << i;
// const auto ipc = ipc_addr.str();
//
// if (zmq_bind(sockets[i], ipc.c_str()) != 0) {
// throw runtime_error(zmq_strerror(errno));
// }
// }
//
// LiveRecvModule recv_module(queue, n_modules, ctx, BUFFER_LIVE_IPC_URL);
//
// // Nothing should be committed, queue, should be empty.
// ASSERT_EQ(queue.read(), -1);
//
// ModuleFrame metadata;
// auto data = make_unique<char[]>(MODULE_N_BYTES);
//
// for (size_t i = 0; i < n_modules; i++) {
// metadata.pulse_id = 1;
// metadata.frame_index = 2;
// metadata.daq_rec = 3;
// metadata.n_received_packets = 4;
// metadata.module_id = i;
//
// zmq_send(sockets[i], &metadata, sizeof(ModuleFrame), ZMQ_SNDMORE);
// zmq_send(sockets[i], data.get(), MODULE_N_BYTES, 0);
// }
//
// this_thread::sleep_for(chrono::milliseconds(100));
//
// auto slot_id = queue.read();
// // We should have the first Detector frame in the buffer.
// //ASSERT_NE(slot_id, -1);
//
// auto recv_stopped = async(launch::async, [&](){
// recv_module.stop();
// });
//
// this_thread::sleep_for(chrono::milliseconds(100));
//
// for (size_t i = 0; i < n_modules; i++) {
// metadata.pulse_id = 1;
// metadata.frame_index = 2;
// metadata.daq_rec = 3;
// metadata.n_received_packets = 4;
// metadata.module_id = i;
//
// zmq_send(sockets[i], &metadata, sizeof(ModuleFrame), ZMQ_SNDMORE);
// zmq_send(sockets[i], data.get(), MODULE_N_BYTES, 0);
// }
//
// recv_stopped.wait();
//
// for (size_t i = 0; i < n_modules; i++) {
// zmq_close(sockets[i]);
// }
//
// zmq_ctx_destroy(ctx);
// cout << "We are finished" << endl;
}