mirror of
https://github.com/paulscherrerinstitute/sf_daq_buffer.git
synced 2026-05-03 12:14:13 +02:00
Revert back to single threaded assembly
This commit is contained in:
+72
-71
@@ -33,93 +33,89 @@ int main (int argc, char *argv[])
|
||||
uint64_t start_pulse_id = (uint64_t) atoll(argv[2]);
|
||||
uint64_t stop_pulse_id = (uint64_t) atoll(argv[3]);
|
||||
|
||||
// H5Writer writer(output_file);
|
||||
// writer.create_file();
|
||||
|
||||
size_t n_modules = 32;
|
||||
|
||||
auto metadata_buffer = make_unique<ModuleFrame>();
|
||||
auto image_buffer = make_unique<uint16_t[]>(n_modules*512*1024);
|
||||
|
||||
uint32_t all_modules_ready = 0;
|
||||
for (size_t i=0; i<n_modules; i++) {
|
||||
all_modules_ready |= 1 << i;
|
||||
}
|
||||
|
||||
atomic_uint32_t modules_ready = all_modules_ready;
|
||||
H5Writer writer(output_file);
|
||||
writer.create_file();
|
||||
|
||||
auto ctx = zmq_ctx_new();
|
||||
zmq_ctx_set (ctx, ZMQ_IO_THREADS, 16);
|
||||
|
||||
vector<thread> threads(n_modules);
|
||||
|
||||
auto read_thread = [&](int module_id) {
|
||||
auto socket = zmq_socket(ctx, ZMQ_PULL);
|
||||
|
||||
int rcvhwm = 100;
|
||||
if (zmq_setsockopt(socket, ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
int linger = 0;
|
||||
if (zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
auto ipc_address = "ipc://sf-replay-" + to_string(module_id);
|
||||
if (zmq_bind(socket, ipc_address.c_str()) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
uint32_t module_mask = 0;
|
||||
module_mask |= 1 << module_id;
|
||||
|
||||
while (true) {
|
||||
if ((modules_ready & module_mask) == 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
auto n_bytes_metadata = zmq_recv(
|
||||
socket,
|
||||
metadata_buffer.get(),
|
||||
sizeof(ModuleFrame),
|
||||
0);
|
||||
|
||||
if (n_bytes_metadata != sizeof(ModuleFrame))
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
auto image_buffer_offset = 512*1024*module_id;
|
||||
auto n_bytes_image = zmq_recv(
|
||||
socket,
|
||||
image_buffer.get() + image_buffer_offset,
|
||||
512 * 1024 * 2,
|
||||
0);
|
||||
|
||||
if (n_bytes_image != 512 * 1024 * 2) {
|
||||
throw runtime_error(strerror (errno));
|
||||
}
|
||||
|
||||
modules_ready ^= 1 << module_id;
|
||||
}
|
||||
};
|
||||
size_t n_modules = 32;
|
||||
void* sockets[n_modules];
|
||||
|
||||
for (size_t i=0; i<n_modules; i++) {
|
||||
threads.emplace_back(read_thread, i);
|
||||
sockets[i] = zmq_socket(ctx, ZMQ_PULL);
|
||||
int rcvhwm = 100;
|
||||
if (zmq_setsockopt(sockets[i], ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0) {
|
||||
throw runtime_error(strerror (errno));
|
||||
}
|
||||
int linger = 0;
|
||||
if (zmq_setsockopt(sockets[i], ZMQ_LINGER, &linger, sizeof(linger)) != 0) {
|
||||
throw runtime_error(strerror (errno));
|
||||
}
|
||||
|
||||
stringstream ipc_addr;
|
||||
ipc_addr << "ipc://sf-replay-" << i;
|
||||
auto ipc = ipc_addr.str();
|
||||
|
||||
if (zmq_bind(sockets[i], ipc.c_str()) != 0) {
|
||||
throw runtime_error(strerror (errno));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
auto metadata_buffer = make_unique<ModuleFrame>();
|
||||
auto image_buffer = make_unique<uint16_t[]>(32*512*1024);
|
||||
|
||||
int i_write = 0;
|
||||
size_t total_ms = 0;
|
||||
size_t max_ms = 0;
|
||||
|
||||
auto start_time = chrono::steady_clock::now();
|
||||
|
||||
while (true) {
|
||||
uint64_t pulse_id = 0;
|
||||
|
||||
if (modules_ready > 0) {
|
||||
continue;
|
||||
auto start_time = chrono::steady_clock::now();
|
||||
|
||||
for (size_t i=0; i<n_modules; i++) {
|
||||
auto n_bytes_metadata = zmq_recv(
|
||||
sockets[i],
|
||||
metadata_buffer.get(),
|
||||
sizeof(ModuleFrame),
|
||||
0);
|
||||
|
||||
if (n_bytes_metadata != sizeof(ModuleFrame)) {
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
}
|
||||
|
||||
if (i == 0) {
|
||||
pulse_id = metadata_buffer->pulse_id;
|
||||
}
|
||||
|
||||
if (pulse_id != metadata_buffer->pulse_id) {
|
||||
cout << "Module " << i << " pulse " << metadata_buffer->pulse_id;
|
||||
cout << " instead of " << pulse_id << endl;
|
||||
}
|
||||
|
||||
auto n_bytes_image = zmq_recv(
|
||||
sockets[i],
|
||||
(image_buffer.get() + (512*1024*i)),
|
||||
512 * 1024 * 2,
|
||||
0);
|
||||
|
||||
if (n_bytes_image != 512 * 1024 * 2) {
|
||||
cout << "n_bytes_image " << n_bytes_image << endl;
|
||||
throw runtime_error("Unexpected number of bytes in image.");
|
||||
}
|
||||
}
|
||||
|
||||
// writer.write_data("image", i_write, (char*) (image_buffer.get()),
|
||||
// {32*512, 1024}, 32*512*1024*2, "uint16", "little");
|
||||
i_write++;
|
||||
|
||||
auto end_time = chrono::steady_clock::now();
|
||||
|
||||
// TODO: Some poor statistics.
|
||||
|
||||
auto ms_duration = chrono::duration_cast<chrono::milliseconds>(end_time-start_time).count();
|
||||
total_ms += ms_duration;
|
||||
if (ms_duration > max_ms) {
|
||||
@@ -127,16 +123,21 @@ int main (int argc, char *argv[])
|
||||
}
|
||||
|
||||
if (i_write==100) {
|
||||
cout << "avg_ms " << total_ms / 100;
|
||||
cout << "assembly_ms " << total_ms / 100;
|
||||
cout << " max_ms " << max_ms << endl;
|
||||
i_write = 0;
|
||||
total_ms = 0;
|
||||
max_ms = 0;
|
||||
}
|
||||
start_time = chrono::steady_clock::now();
|
||||
|
||||
modules_ready = all_modules_ready;
|
||||
}
|
||||
|
||||
writer.close_file();
|
||||
|
||||
for (size_t i=0; i<n_modules; i++) {
|
||||
zmq_close(sockets[i]);
|
||||
}
|
||||
|
||||
zmq_ctx_destroy(ctx);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user