Add std-udp-sync project

std-udp-sync/src/SyncStats.cpp

@@ -0,0 +1,55 @@
+#include "SyncStats.hpp"
+
+#include <iostream>
+
+using namespace std;
+using namespace chrono;
+
+SyncStats::SyncStats(
+        const std::string &detector_name,
+        const size_t stats_modulo) :
+            detector_name_(detector_name),
+            stats_modulo_(stats_modulo)
+{
+    reset_counters();
+}
+
+void SyncStats::reset_counters()
+{
+    image_counter_ = 0;
+    n_sync_lost_images_ = 0;
+    stats_interval_start_ = steady_clock::now();
+}
+
+void SyncStats::record_stats(const uint32_t n_lost_pulses)
+{
+    image_counter_++;
+    n_sync_lost_images_ += n_lost_pulses;
+
+    if (image_counter_ == stats_modulo_) {
+        print_stats();
+        reset_counters();
+    }
+}
+
+void SyncStats::print_stats()
+{
+    auto interval_ms_duration = duration_cast<milliseconds>(
+            steady_clock::now()-stats_interval_start_).count();
+    // * 1000 because milliseconds, + 250 because of truncation.
+    int rep_rate = ((image_counter_ * 1000) + 250) / interval_ms_duration;
+    uint64_t timestamp = time_point_cast<nanoseconds>(
+            system_clock::now()).time_since_epoch().count();
+
+    // Output in InfluxDB line protocol
+    cout << "std_udp_sync";
+    cout << ",detector_name=" << detector_name_;
+    cout << " ";
+    cout << "n_processed_images=" << image_counter_ << "i";
+    cout << ",n_sync_lost_images=" << n_sync_lost_images_ << "i";
+    cout << ",repetition_rate=" << rep_rate << "i";
+    cout << " ";
+    cout << timestamp;
+    cout << endl;
+}
+

std-udp-sync/src/ZmqPulseSyncReceiver.cpp

@@ -0,0 +1,127 @@
+#include "ZmqPulseSyncReceiver.hpp"
+#include "BufferUtils.hpp"
+
+#include <zmq.h>
+#include <stdexcept>
+#include <sstream>
+#include <chrono>
+#include <iostream>
+#include "date.h"
+
+#include "sync_config.hpp"
+
+using namespace std;
+using namespace chrono;
+using namespace sync_config;
+
+
+ZmqPulseSyncReceiver::ZmqPulseSyncReceiver(
+        void * ctx,
+        const string& detector_name,
+        const int n_modules) :
+            ctx_(ctx),
+            n_modules_(n_modules)
+{
+    sockets_.reserve(n_modules_);
+
+    for (int i=0; i<n_modules_; i++) {
+        sockets_.push_back(
+                BufferUtils::connect_socket(ctx_, detector_name, to_string(i)));
+    }
+}
+
+ZmqPulseSyncReceiver::~ZmqPulseSyncReceiver()
+{
+    for (auto& socket:sockets_) {
+        zmq_close(socket);
+    }
+}
+
+PulseAndSync ZmqPulseSyncReceiver::get_next_pulse_id() const
+{
+    uint64_t pulses[n_modules_];
+
+    bool modules_in_sync = true;
+    for (int i = 0; i < n_modules_; i++) {
+        zmq_recv(sockets_[i], &pulses[i], sizeof(uint64_t), 0);
+        if (pulses[0] != pulses[i]) {
+            modules_in_sync = false;
+        }
+    }
+
+    if (modules_in_sync) {
+        #ifdef DEBUG_OUTPUT
+            using namespace date;
+            cout << " [" << std::chrono::system_clock::now();
+            cout << "] [ZmqPulseSyncReceiver::get_next_pulse_id]  ";
+            cout << "] (modules_in_sync) Frame index:" << pulses[0];
+            cout << endl;
+        #endif
+        return {pulses[0], 0};
+    }
+
+    // How many pulses we lost in total to get the next pulse_id.
+    uint32_t n_lost_pulses = 0;
+    for (int i_sync=0; i_sync < SYNC_RETRY_LIMIT; i_sync++) {
+        uint64_t min_pulse_id = numeric_limits<uint64_t>::max();;
+        uint64_t max_pulse_id = 0;
+
+        for (int i = 0; i < n_modules_; i++) {
+            min_pulse_id = min(min_pulse_id, pulses[i]);
+            max_pulse_id = max(max_pulse_id, pulses[i]);
+        }
+
+        auto max_diff = max_pulse_id - min_pulse_id;
+        if (max_diff > PULSE_OFFSET_LIMIT) {
+            stringstream err_msg;
+            err_msg << "[ZmqPulseSyncReceiver::get_next_pulse_id]";
+            err_msg << " PULSE_OFFSET_LIMIT exceeded.";
+            err_msg << " max_diff=" << max_diff << " pulses.";
+
+            for (int i = 0; i < n_modules_; i++) {
+                err_msg << " (module " << i << ", ";
+                err_msg << pulses[i] << "),";
+            }
+            err_msg << endl;
+
+            throw runtime_error(err_msg.str());
+        }
+
+        modules_in_sync = true;
+        // Max pulses we lost in this sync attempt.
+        uint32_t i_sync_lost_pulses = 0;
+        for (int i = 0; i < n_modules_; i++) {
+            // How many pulses we lost for this specific module.
+            uint32_t i_module_lost_pulses = 0;
+            while (pulses[i] < max_pulse_id) {
+                zmq_recv(sockets_[i], &pulses[i], sizeof(uint64_t), 0);
+                i_module_lost_pulses++;
+            }
+
+            i_sync_lost_pulses = max(i_sync_lost_pulses, i_module_lost_pulses);
+
+            if (pulses[i] != max_pulse_id) {
+                modules_in_sync = false;
+            }
+        }
+        n_lost_pulses += i_sync_lost_pulses;
+
+        if (modules_in_sync) {
+            #ifdef DEBUG_OUTPUT
+                using namespace date;
+                cout << " [" << std::chrono::system_clock::now();
+                cout << "] [ZmqPulseSyncReceiver::get_next_pulse_id]";
+                cout << " modules_in_sync false";
+                cout << endl;
+            #endif
+            return {pulses[0], n_lost_pulses};
+        }
+    }
+
+    stringstream err_msg;
+    err_msg << "[ZmqLiveReceiver::get_next_pulse_id]";
+    err_msg << " SYNC_RETRY_LIMIT exceeded.";
+    err_msg << endl;
+
+    throw runtime_error(err_msg.str());
+}

std-udp-sync/src/main.cpp

@@ -0,0 +1,69 @@
+#include <iostream>
+#include <string>
+#include <zmq.h>
+#include <RamBuffer.hpp>
+#include <BufferUtils.hpp>
+#include <SyncStats.hpp>
+
+#include "date.h"
+#include <chrono>
+#include "sync_config.hpp"
+#include "ZmqPulseSyncReceiver.hpp"
+#include "UdpSyncConfig.hpp"
+ 
+
+using namespace std;
+using namespace sync_config;
+
+#ifdef USE_EIGER
+    #include "eiger.hpp"
+#else
+    #include "jungfrau.hpp"
+#endif
+
+int main (int argc, char *argv[])
+{
+    if (argc != 3) {
+        cout << endl;
+        cout << "Usage: std_udp_sync [detector_json_filename] [bit_depth]" << endl;
+        cout << "\tdetector_json_filename: detector config file path." << endl;
+        cout << "\tbit_depth: bit depth of the incoming udp packets." << endl;
+        cout << endl;
+
+        exit(-1);
+    }
+
+    const auto config = UdpSyncConfig::from_json_file(string(argv[1]));
+    const int bit_depth = atoi(argv[2]);
+
+    const size_t FRAME_N_BYTES = MODULE_N_PIXELS * bit_depth / 8;
+
+    auto ctx = zmq_ctx_new();
+    zmq_ctx_set(ctx, ZMQ_IO_THREADS, 1);
+
+    auto sender = BufferUtils::bind_socket(ctx, config.detector_name, "sync");
+
+    #ifdef DEBUG_OUTPUT
+        using namespace date;
+        cout << " [" << std::chrono::system_clock::now();
+        cout << "] [Assembler] :";
+        cout << " Details of Assembler:";
+        cout << " detector_name: " << config.detector_name;
+        cout << " || n_modules: " << config.n_modules;
+        cout << endl;
+    #endif
+
+    RamBuffer frame_buffer(config.detector_name, sizeof(ModuleFrame),
+                           FRAME_N_BYTES, config.n_modules);
+
+    ZmqPulseSyncReceiver receiver(ctx, config.detector_name, config.n_modules);
+    SyncStats stats(config.detector_name, SYNC_STATS_MODULO);
+
+    while (true) {
+        auto meta = receiver.get_next_pulse_id();
+
+        zmq_send(sender, &meta.image_id, sizeof(meta.image_id), 0);
+
+        stats.record_stats(meta.n_lost_pulses);
+    }
+}