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

Trying ipc for transport protocol

Browse Source
This commit is contained in:
babic_a
2020-04-22 12:48:48 +02:00
parent 09a46daea0
commit 76bde34d97
2 changed files with 15 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sf-buffer/src/sf_replay.cpp
+6 -17
View File
@@ -47,20 +47,10 @@ int main (int argc, char *argv[]) {
auto ctx = zmq_ctx_new();
auto socket = zmq_socket(ctx, ZMQ_PUSH);
auto more_socket = zmq_socket(ctx, ZMQ_SUB);
//TODO: Use ipc?
if (zmq_connect(socket, "tcp://localhost:50000") != 0) {
throw runtime_error(strerror (errno));
}
if (zmq_connect(more_socket, "tcp://localhost:50001") != 0) {
throw runtime_error(strerror (errno));
}
int status = 0;
int sndhwm = 1000;
int sndhwm = 100;
status += zmq_setsockopt(socket, ZMQ_SNDHWM, &sndhwm, sizeof(sndhwm));
int linger_ms = 0;
status += zmq_setsockopt(socket, ZMQ_LINGER, &linger_ms, sizeof(linger_ms));
@@ -71,6 +61,11 @@ int main (int argc, char *argv[]) {
throw runtime_error(strerror (errno));
}
//TODO: Use ipc?
if (zmq_connect(socket, "ipc://writer") != 0) {
throw runtime_error(strerror (errno));
}
for (const auto& suffix:path_suffixes) {
metadata_buffer->start_pulse_id = suffix.start_pulse_id;
metadata_buffer->stop_pulse_id = suffix.stop_pulse_id;
@@ -126,12 +121,6 @@ int main (int argc, char *argv[]) {
(char*) (image_buffer.get() + (i_frame * 512 * 1024)),
512 * 1024 * 2,
0);
if ((i_frame>0) && (i_frame % 100 == 0)) {
// Waiting to send more.
uint64_t test = 0;
zmq_recv(more_socket, &test, sizeof(test), 0);
}
}
}
sf-writer/sf_h5_writer.cpp
+9 -17
View File
@@ -35,18 +35,8 @@ int main (int argc, char *argv[])
zmq_ctx_set (ctx, ZMQ_IO_THREADS, 16);
auto socket = zmq_socket(ctx, ZMQ_PULL);
auto more_socket = zmq_socket(ctx, ZMQ_PUB);
//TODO: Use ipc?
if (zmq_bind(socket, "tcp://127.0.0.1:50000") != 0) {
throw runtime_error(strerror (errno));
}
if (zmq_bind(more_socket, "tcp://127.0.0.1:50001") != 0) {
throw runtime_error(strerror (errno));
}
int rcvhwm = 10000;
int rcvhwm = 1000;
if (zmq_setsockopt(socket, ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0) {
throw runtime_error(strerror (errno));
}
@@ -56,6 +46,14 @@ int main (int argc, char *argv[])
throw runtime_error(strerror (errno));
}
//TODO: Use ipc?
if (zmq_bind(socket, "ipc://writer") != 0) {
throw runtime_error(strerror (errno));
}
auto metadata_buffer = make_unique<ModuleFrame>();
auto image_buffer = make_unique<uint16_t[]>(512 * 1024);
@@ -96,12 +94,6 @@ int main (int argc, char *argv[])
size_t n_in_progress_frames = received_counter.size();
cout << "n frames in progress " << n_in_progress_frames << endl;
if (n_in_progress_frames == 0) {
uint64_t test = 0;
zmq_send(more_socket, &test, sizeof(test), 0);
cout << "SENT!!!" << endl;
}
}
zmq_close(socket);