Add jf-assembler executable

This service reconstructs the various modules and sends out a
image metadata stream for further consumers.

jf-assembler/src/AssemblerStats.cpp

@@ -0,0 +1,62 @@
+#include "AssemblerStats.hpp"
+
+#include <iostream>
+
+using namespace std;
+using namespace chrono;
+
+AssemblerStats::AssemblerStats(
+        const std::string &detector_name,
+        const size_t stats_modulo) :
+            detector_name_(detector_name),
+            stats_modulo_(stats_modulo)
+{
+    reset_counters();
+}
+
+void AssemblerStats::reset_counters()
+{
+    image_counter_ = 0;
+    n_sync_lost_images_ = 0;
+    n_corrupted_images_ = 0;
+    stats_interval_start_ = steady_clock::now();
+}
+
+void AssemblerStats::record_stats(
+        const ImageMetadata &meta, const uint32_t n_lost_pulses)
+{
+    image_counter_++;
+    n_sync_lost_images_ += n_lost_pulses;
+
+    if (!meta.is_good_image) {
+        n_corrupted_images_++;
+    }
+
+    if (image_counter_ == stats_modulo_) {
+        print_stats();
+        reset_counters();
+    }
+}
+
+void AssemblerStats::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 << "jf_assembler";
+    cout << ",detector_name=" << detector_name_;
+    cout << " ";
+    cout << "n_processed_images=" << image_counter_ << "i";
+    cout << ",n_corrupted_images=" << n_corrupted_images_ << "i";
+    cout << ",n_sync_lost_images=" << n_sync_lost_images_ << "i";
+    cout << ",repetition_rate=" << rep_rate << "i";
+    cout << " ";
+    cout << timestamp;
+    cout << endl;
+}
+

jf-assembler/src/ZmqPulseSyncReceiver.cpp

@@ -0,0 +1,115 @@
+#include "ZmqPulseSyncReceiver.hpp"
+#include "BufferUtils.hpp"
+
+#include <zmq.h>
+#include <stdexcept>
+#include <sstream>
+#include <chrono>
+#include <algorithm>
+#include <iostream>
+
+#include "assembler_config.hpp"
+
+using namespace std;
+using namespace chrono;
+using namespace buffer_config;
+using namespace assembler_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) {
+        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) {
+            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());
+}

jf-assembler/src/main.cpp

@@ -0,0 +1,50 @@
+#include <iostream>
+#include <string>
+#include <zmq.h>
+#include <RamBuffer.hpp>
+#include <BufferUtils.hpp>
+#include <AssemblerStats.hpp>
+
+#include "buffer_config.hpp"
+#include "assembler_config.hpp"
+#include "ZmqPulseSyncReceiver.hpp"
+
+using namespace std;
+using namespace buffer_config;
+using namespace assembler_config;
+
+int main (int argc, char *argv[])
+{
+    if (argc != 2) {
+        cout << endl;
+        cout << "Usage: jf_assembler [detector_json_filename]" << endl;
+        cout << "\tdetector_json_filename: detector config file path." << endl;
+        cout << endl;
+
+        exit(-1);
+    }
+
+    auto config = BufferUtils::read_json_config(string(argv[1]));
+    auto const stream_name = "assembler";
+
+    string RECV_IPC_URL = BUFFER_LIVE_IPC_URL + config.detector_name + "-";
+    auto ctx = zmq_ctx_new();
+    zmq_ctx_set(ctx, ZMQ_IO_THREADS, ASSEMBLER_ZMQ_IO_THREADS);
+
+    ZmqPulseSyncReceiver receiver(ctx, config.detector_name, config.n_modules);
+    RamBuffer ram_buffer(config.detector_name, config.n_modules);
+    AssemblerStats stats(config.detector_name, ASSEMBLER_STATS_MODULO);
+
+    auto sender = BufferUtils::bind_socket(
+            ctx, config.detector_name, stream_name);
+
+    ImageMetadata meta;
+    while (true) {
+        auto pulse_and_sync = receiver.get_next_pulse_id();
+        ram_buffer.assemble_image(pulse_and_sync.pulse_id, meta);
+
+        zmq_send(sender, &meta, sizeof(meta), 0);
+
+        stats.record_stats(meta, pulse_and_sync.n_lost_pulses);
+    }
+}