receive multimodule (#65)

* add config files for multimodule receiving

* read subfiles with unordered and missing frames

* save work debugging

* Revert "save work debugging"

This reverts commit e791992a05.

* Revert "read subfiles with unordered and missing frames"

This reverts commit 1177fd129d.

* throw when two frames have different frame numbers

* write single part RawFile (working beta)

* correct total number of frames in master file

* add new mythen file with syncd frames

* save work

* save work for receive multimodule
multimodule config results in too much packet loss. needs more debugging.

* setup Task Distributiosn/ parallelization programming model

* read frames with same frame number

* clang-tidy fixes, formatting, add tests

* added second receiver

* Synchronize between zmq streams and merge frames

* improve readability in loop

* fix failing tests

* add simple test for merge frames

---------

Co-authored-by: Bechir <bechir.brahem420@gmail.com>
Co-authored-by: Erik Frojdh <erik.frojdh@gmail.com>

network_io/CMakeLists.txt

@@ -17,6 +17,10 @@ add_library(network_io STATIC
     src/ZmqSocketSender.cpp
     src/ZmqSocket.cpp
     src/ZmqHeader.cpp
+    src/ZmqMultiReceiver.cpp
+    src/ZmqVentilator.cpp
+    src/ZmqWorker.cpp
+    src/ZmqSink.cpp
     )
 
 

network_io/include/aare/network_io/ZmqHeader.hpp

@@ -8,29 +8,28 @@
 #include <map>
 #include <string>
 namespace simdjson {
-/**
- * @brief cast a simdjson::ondemand::value to a std::array<int,4>
- * useful for writing rx_roi from json header
- */
-template <> simdjson_inline simdjson::simdjson_result<std::array<int, 4>> simdjson::ondemand::value::get() noexcept {
-    ondemand::array array;
-    auto error = get_array().get(array);
-    if (error) {
-        return error;
-    }
-    std::array<int, 4> arr{};
-    int i = 0;
-    for (auto v : array) {
-        int64_t val = 0;
-        error = v.get_int64().get(val);
+// template <typename T, int N> std::array
 
-        if (error) {
-            return error;
-        }
-        arr[i++] = static_cast<int>(val);
-    }
-    return arr;
-}
+// template <int N> simdjson_inline simdjson::simdjson_result<std::array<int, N>> simdjson::ondemand::value::get()
+// noexcept {
+//     ondemand::array array;
+//     auto error = get_array().get(array);
+//     if (error) {
+//         return error;
+//     }
+//     std::array<int, N> arr{};
+//     int i = 0;
+//     for (auto v : array) {
+//         int64_t val = 0;
+//         error = v.get_int64().get(val);
+
+//         if (error) {
+//             return error;
+//         }
+//         arr[i++] = static_cast<int>(val);
+//     }
+//     return arr;
+// }
 
 /**
  * @brief cast a simdjson::ondemand::value to a uint32_t
@@ -81,22 +80,17 @@ simdjson::ondemand::value::get() noexcept {
 } // namespace simdjson
 
 namespace aare {
-
 /** zmq header structure (from slsDetectorPackage)*/
 struct ZmqHeader {
     /** true if incoming data, false if end of acquisition */
     bool data{true};
     uint32_t jsonversion{0};
-    uint32_t dynamicRange{0};
+    uint32_t bitmode{0};
     uint64_t fileIndex{0};
-    /** number of detectors/port in x axis */
-    uint32_t ndetx{0};
-    /** number of detectors/port in y axis */
-    uint32_t ndety{0};
-    /** number of pixels/channels in x axis for this zmq socket */
-    uint32_t npixelsx{0};
-    /** number of pixels/channels in y axis for this zmq socket */
-    uint32_t npixelsy{0};
+    /** number of detectors/port*/
+    t_xy<uint32_t> detshape{0, 0};
+    /** number of pixels/channels for this zmq socket */
+    t_xy<uint32_t> shape{0, 0};
     /** number of bytes for an image in this socket */
     uint32_t size{0};
     /** frame number from detector */
@@ -138,5 +132,25 @@ struct ZmqHeader {
     // compare operator
     bool operator==(const ZmqHeader &other) const;
 };
+/**
+ * @brief cast a simdjson::ondemand::value to a std::array<int,4>
+ * useful for writing rx_roi from json header
+ */
+template <typename T, int N, typename SIMDJSON_VALUE> std::array<T, N> simd_convert_array(SIMDJSON_VALUE field) {
+    simdjson::ondemand::array simd_array;
+    auto err = field.value().get_array().get(simd_array);
+    if (err)
+        throw std::runtime_error("error converting simdjson::ondemand::value to simdjson::ondemend::array");
+    std::array<T, N> arr{};
+    int i = 0;
+    for (auto v : simd_array) {
+        int64_t tmp;
+        err = v.get(tmp);
+        if (err)
+            throw std::runtime_error("error converting simdjson::ondemand::value");
+        arr[i++] = tmp;
+    }
+    return arr;
+}
 
 } // namespace aare

network_io/include/aare/network_io/ZmqMultiReceiver.hpp

@@ -0,0 +1,27 @@
+#pragma once
+#include "aare/network_io/ZmqSocketReceiver.hpp"
+#include "aare/network_io/defs.hpp"
+#include <unordered_map>
+
+#include <string>
+#include <vector>
+using zmq_pollitem_t = struct zmq_pollitem_t;
+namespace aare {
+
+class ZmqMultiReceiver {
+  public:
+    explicit ZmqMultiReceiver(const std::vector<std::string> &endpoints, const xy &geometry = {1, 1});
+    int connect();
+    ZmqFrame receive_zmqframe();
+    std::vector<ZmqFrame> receive_n();
+    ~ZmqMultiReceiver();
+
+  private:
+    ZmqFrame receive_zmqframe_(std::unordered_map<uint64_t, std::vector<ZmqFrame>> &frames_map);
+    xy m_geometry;
+    std::vector<std::string> m_endpoints;
+    std::vector<ZmqSocketReceiver *> m_receivers;
+    zmq_pollitem_t *items{};
+};
+
+} // namespace aare

network_io/include/aare/network_io/ZmqSink.hpp

@@ -0,0 +1,17 @@
+#pragma once
+
+#include "aare/network_io/ZmqSocketReceiver.hpp"
+
+namespace aare {
+
+class ZmqSink {
+  public:
+    explicit ZmqSink(const std::string &sink_endpoint);
+    Task *pull();
+    ~ZmqSink();
+
+  private:
+    ZmqSocketReceiver *m_receiver;
+};
+
+} // namespace aare

network_io/include/aare/network_io/ZmqSocket.hpp

@@ -25,10 +25,12 @@ class ZmqSocket {
     void set_zmq_hwm(int hwm);
     void set_timeout_ms(int n);
     void set_potential_frame_size(size_t size);
+    void *get_socket();
 
   protected:
     void *m_context{nullptr};
     void *m_socket{nullptr};
+    int m_socket_type{};
     std::string m_endpoint;
     int m_zmq_hwm{1000};
     int m_timeout_ms{1000};

network_io/include/aare/network_io/ZmqSocketReceiver.hpp

@@ -19,14 +19,14 @@ namespace aare {
  */
 class ZmqSocketReceiver : public ZmqSocket {
   public:
-    explicit ZmqSocketReceiver(const std::string &endpoint);
+    explicit ZmqSocketReceiver(const std::string &endpoint, int socket_type = 2 /* ZMQ_SUB */);
     void connect();
+    void bind();
     std::vector<ZmqFrame> receive_n();
 
-  private:
-    int receive_data(std::byte *data, size_t size);
     ZmqFrame receive_zmqframe();
     ZmqHeader receive_header();
+    int receive_data(std::byte *data, size_t size);
 };
 
 } // namespace aare

network_io/include/aare/network_io/ZmqSocketSender.hpp

@@ -12,9 +12,11 @@ namespace aare {
  */
 class ZmqSocketSender : public ZmqSocket {
   public:
-    explicit ZmqSocketSender(const std::string &endpoint);
+    explicit ZmqSocketSender(const std::string &endpoint, int socket_type = 1 /* ZMQ_PUB */);
+    void connect();
     void bind();
-    size_t send(const ZmqHeader &header, const std::byte *data, size_t size);
+    size_t send(const void *data, size_t size);
+    size_t send(const ZmqHeader &header, const void *data, size_t size);
     size_t send(const ZmqFrame &zmq_frame);
     size_t send(const std::vector<ZmqFrame> &zmq_frames);
 };

network_io/include/aare/network_io/ZmqVentilator.hpp

@@ -0,0 +1,20 @@
+
+#pragma once
+#include "aare/core/Frame.hpp"
+#include "aare/network_io/ZmqHeader.hpp"
+#include "aare/network_io/ZmqSocket.hpp"
+#include "aare/network_io/ZmqSocketSender.hpp"
+#include "aare/network_io/defs.hpp"
+
+namespace aare {
+
+class ZmqVentilator {
+  public:
+    explicit ZmqVentilator(const std::string &endpoint);
+    size_t push(const Task *task);
+    ~ZmqVentilator();
+
+  private:
+    ZmqSocketSender *m_sender;
+};
+} // namespace aare

network_io/include/aare/network_io/ZmqWorker.hpp

@@ -0,0 +1,18 @@
+#include "aare/network_io/ZmqSocketReceiver.hpp"
+#include "aare/network_io/ZmqSocketSender.hpp"
+
+namespace aare {
+
+class ZmqWorker {
+  public:
+    explicit ZmqWorker(const std::string &ventilator_endpoint, const std::string &sink_endpoint = "");
+    Task *pull();
+    size_t push(const Task *task);
+    ~ZmqWorker();
+
+  private:
+    ZmqSocketReceiver *m_receiver;
+    ZmqSocketSender *m_sender;
+};
+
+} // namespace aare

network_io/include/aare/network_io/defs.hpp

@@ -14,8 +14,52 @@ namespace aare {
 struct ZmqFrame {
     ZmqHeader header;
     Frame frame;
+    std::string to_string() const {
+        return "ZmqFrame{header: " + header.to_string() + ", frame:\nrows: " + std::to_string(frame.rows()) +
+               ", cols: " + std::to_string(frame.cols()) + ", bitdepth: " + std::to_string(frame.bitdepth()) + "\n}";
+    }
+    size_t size() const { return frame.size() + header.size; }
 };
 
+struct Task {
+
+    size_t id{};
+    int opcode{}; // operation to perform on the data (what type should this be? char*? enum?)
+    size_t data_size{};
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wpedantic"
+    std::byte payload[];
+#pragma GCC diagnostic pop
+
+    static const size_t MAX_DATA_SIZE = 1024 * 1024; // 1MB
+    size_t size() const { return sizeof(Task) + data_size; }
+    static Task *init(std::byte *data, size_t data_size) {
+        Task *task = (Task *)new std::byte[sizeof(Task) + data_size];
+        task->data_size = data_size;
+        if (data_size > 0)
+            memcpy(task->payload, data, data_size);
+        return task;
+    }
+    static int destroy(Task *task) {
+        delete[] task;
+        return 0;
+    }
+
+    Task() = delete;
+    Task(Task &) = delete;
+    Task(Task &&) = default;
+
+    // common operations to perform
+    // users can still send custom operations
+    enum class Operation {
+        PEDESTAL,
+        PEDESTAL_AND_SAVE,
+        PEDESTAL_AND_CLUSTER,
+        PEDESTAL_AND_CLUSTER_AND_SAVE,
+        COUNT,
+    };
+} __attribute__((packed));
+
 namespace network_io {
 /**
  * @brief NetworkError exception class

network_io/src/ZmqHeader.cpp

@@ -12,12 +12,10 @@ std::string ZmqHeader::to_string() const {
     s += "{";
     write_digit(s, "data", data ? 1 : 0);
     write_digit(s, "jsonversion", jsonversion);
-    write_digit(s, "dynamicRange", dynamicRange);
+    write_digit(s, "bitmode", bitmode);
     write_digit(s, "fileIndex", fileIndex);
-    write_digit(s, "ndetx", ndetx);
-    write_digit(s, "ndety", ndety);
-    write_digit(s, "npixelsx", npixelsx);
-    write_digit(s, "npixelsy", npixelsy);
+    write_array<uint32_t, 2>(s, "detshape", std::array<uint32_t, 2>{detshape.row, detshape.col});
+    write_array<uint32_t, 2>(s, "shape", std::array<uint32_t, 2>{shape.row, shape.col});
     write_digit(s, "size", size);
     write_digit(s, "acqIndex", acqIndex);
     write_digit(s, "frameIndex", frameIndex);
@@ -40,7 +38,7 @@ std::string ZmqHeader::to_string() const {
     write_digit(s, "quad", quad);
     write_digit(s, "completeImage", completeImage ? 1 : 0);
     write_map(s, "addJsonHeader", addJsonHeader);
-    write_array(s, "rx_roi", rx_roi);
+    write_array<int, 4>(s, "rx_roi", rx_roi);
     // remove last comma
     s.pop_back();
     s.pop_back();
@@ -63,18 +61,18 @@ void ZmqHeader::from_string(std::string &s) { // NOLINT
             data = static_cast<uint64_t>(field.value()) != 0;
         } else if (key == "jsonversion") {
             jsonversion = static_cast<uint32_t>(field.value());
-        } else if (key == "dynamicRange") {
-            dynamicRange = static_cast<uint32_t>(field.value());
+        } else if (key == "bitmode") {
+            bitmode = static_cast<uint32_t>(field.value());
         } else if (key == "fileIndex") {
             fileIndex = static_cast<uint64_t>(field.value());
-        } else if (key == "ndetx") {
-            ndetx = static_cast<uint32_t>(field.value());
-        } else if (key == "ndety") {
-            ndety = static_cast<uint32_t>(field.value());
-        } else if (key == "npixelsx") {
-            npixelsx = static_cast<uint32_t>(field.value());
-        } else if (key == "npixelsy") {
-            npixelsy = static_cast<uint32_t>(field.value());
+        } else if (key == "detshape") {
+            std::array<uint32_t, 2> arr = simd_convert_array<uint32_t, 2>(field);
+            detshape.row = arr[0];
+            detshape.col = arr[1];
+        } else if (key == "shape") {
+            std::array<uint32_t, 2> arr = simd_convert_array<uint32_t, 2>(field);
+            shape.row = arr[0];
+            shape.col = arr[1];
         } else if (key == "size") {
             size = static_cast<uint32_t>(field.value());
         } else if (key == "acqIndex") {
@@ -121,21 +119,20 @@ void ZmqHeader::from_string(std::string &s) { // NOLINT
         } else if (key == "addJsonHeader") {
             addJsonHeader = static_cast<std::map<std::string, std::string>>(field.value());
         } else if (key == "rx_roi") {
-            rx_roi = static_cast<std::array<int, 4>>(field.value());
+            rx_roi = simd_convert_array<int, 4>(field);
         }
     }
 }
 bool ZmqHeader::operator==(const ZmqHeader &other) const {
-    return data == other.data && jsonversion == other.jsonversion && dynamicRange == other.dynamicRange &&
-           fileIndex == other.fileIndex && ndetx == other.ndetx && ndety == other.ndety && npixelsx == other.npixelsx &&
-           npixelsy == other.npixelsy && size == other.size && acqIndex == other.acqIndex &&
-           frameIndex == other.frameIndex && progress == other.progress && fname == other.fname &&
-           frameNumber == other.frameNumber && expLength == other.expLength && packetNumber == other.packetNumber &&
-           detSpec1 == other.detSpec1 && timestamp == other.timestamp && modId == other.modId && row == other.row &&
-           column == other.column && detSpec2 == other.detSpec2 && detSpec3 == other.detSpec3 &&
-           detSpec4 == other.detSpec4 && detType == other.detType && version == other.version &&
-           flipRows == other.flipRows && quad == other.quad && completeImage == other.completeImage &&
-           addJsonHeader == other.addJsonHeader && rx_roi == other.rx_roi;
+    return data == other.data && jsonversion == other.jsonversion && bitmode == other.bitmode &&
+           fileIndex == other.fileIndex && detshape == other.detshape && shape == other.shape && size == other.size &&
+           acqIndex == other.acqIndex && frameIndex == other.frameIndex && progress == other.progress &&
+           fname == other.fname && frameNumber == other.frameNumber && expLength == other.expLength &&
+           packetNumber == other.packetNumber && detSpec1 == other.detSpec1 && timestamp == other.timestamp &&
+           modId == other.modId && row == other.row && column == other.column && detSpec2 == other.detSpec2 &&
+           detSpec3 == other.detSpec3 && detSpec4 == other.detSpec4 && detType == other.detType &&
+           version == other.version && flipRows == other.flipRows && quad == other.quad &&
+           completeImage == other.completeImage && addJsonHeader == other.addJsonHeader && rx_roi == other.rx_roi;
 }
 
 } // namespace aare

network_io/src/ZmqMultiReceiver.cpp

@@ -0,0 +1,101 @@
+#include "aare/network_io/ZmqMultiReceiver.hpp"
+#include "aare/utils/merge_frames.hpp"
+#include <unordered_map>
+#include <zmq.h>
+namespace aare {
+ZmqMultiReceiver::ZmqMultiReceiver(const std::vector<std::string> &endpoints, const xy &geometry)
+    : m_geometry(geometry), m_endpoints(endpoints) {
+    assert(m_geometry.row * m_geometry.col == static_cast<uint32_t>(m_endpoints.size()));
+    for (const auto &endpoint : m_endpoints) {
+        m_receivers.push_back(new ZmqSocketReceiver(endpoint));
+    }
+}
+
+int ZmqMultiReceiver::connect() {
+    for (auto *receiver : m_receivers) {
+        receiver->connect();
+    }
+    items = new zmq_pollitem_t[m_receivers.size()];
+    for (size_t i = 0; i < m_receivers.size(); i++) {
+        items[i] = {m_receivers[i]->get_socket(), 0, ZMQ_POLLIN, 0};
+    }
+    return 0;
+}
+ZmqFrame ZmqMultiReceiver::receive_zmqframe() {
+    std::unordered_map<uint64_t, std::vector<ZmqFrame>> frames_map;
+    return receive_zmqframe_(frames_map);
+}
+std::vector<ZmqFrame> ZmqMultiReceiver::receive_n() {
+    std::vector<ZmqFrame> frames;
+    std::unordered_map<uint64_t, std::vector<ZmqFrame>> frames_map;
+    while (true) {
+        // receive header and frame
+        ZmqFrame const zmq_frame = receive_zmqframe_(frames_map);
+        if (!zmq_frame.header.data) {
+            break;
+        }
+        frames.push_back(zmq_frame);
+        frames_map.erase(zmq_frame.header.frameNumber);
+    }
+    return frames;
+}
+ZmqFrame ZmqMultiReceiver::receive_zmqframe_(std::unordered_map<uint64_t, std::vector<ZmqFrame>> &frames_map) {
+    // iterator to store the frame to return
+    std::unordered_map<uint64_t, std::vector<ZmqFrame>>::iterator ret_frames;
+    bool exit_loop = false;
+
+    while (true) {
+        zmq_poll(items, static_cast<int>(m_receivers.size()), -1);
+        aare::logger::debug("Received frame");
+        for (size_t i = 0; i < m_receivers.size() && !exit_loop; i++) {
+            if (items[i].revents & ZMQ_POLLIN) {
+                auto new_frame = m_receivers[i]->receive_zmqframe();
+                if (frames_map.find(new_frame.header.frameNumber) == frames_map.end()) {
+                    frames_map[new_frame.header.frameNumber] = {};
+                }
+
+                ret_frames = frames_map.find(new_frame.header.frameNumber);
+                ret_frames->second.push_back(new_frame);
+
+                exit_loop = ret_frames->second.size() == m_receivers.size();
+            }
+        }
+        if (exit_loop) {
+            break;
+        }
+    }
+    std::vector<ZmqFrame> &frames = ret_frames->second;
+    if (!frames[0].header.data) {
+        return ZmqFrame{frames[0].header, Frame(0, 0, 0)};
+    }
+    // check that all frames have the same shape
+    auto shape = frames[0].header.shape;
+    auto bitdepth = frames[0].header.bitmode;
+    auto part_size = shape.row * shape.col * (bitdepth / 8);
+    for (auto &frame : frames) {
+        assert(shape == frame.header.shape);
+        assert(bitdepth == frame.header.bitmode);
+        // TODO: find solution for numinterfaces=2
+        assert(m_geometry == frame.header.detshape);
+        assert(part_size == frame.header.size);
+    }
+    // merge frames
+    // prepare the input for merge_frames
+    std::vector<std::byte *> part_buffers;
+    part_buffers.reserve(frames.size());
+    for (auto &zmq_frame : frames) {
+        part_buffers.push_back(zmq_frame.frame.data());
+    }
+    Frame const f(shape.row, shape.col, bitdepth);
+    merge_frames(part_buffers, part_size, f.data(), m_geometry, shape.row, shape.col, bitdepth);
+    ZmqFrame zmq_frame = {std::move(frames[0].header), f};
+    return zmq_frame;
+}
+
+ZmqMultiReceiver::~ZmqMultiReceiver() {
+    delete[] items;
+    for (auto *receiver : m_receivers) {
+        delete receiver;
+    }
+}
+} // namespace aare

network_io/src/ZmqSink.cpp

@@ -0,0 +1,19 @@
+
+#include "aare/network_io/ZmqSink.hpp"
+#include "zmq.h"
+
+namespace aare {
+
+ZmqSink::ZmqSink(const std::string &sink_endpoint) : m_receiver(new ZmqSocketReceiver(sink_endpoint, ZMQ_PULL)) {
+    m_receiver->bind();
+};
+
+Task *ZmqSink::pull() {
+    Task *task = reinterpret_cast<Task *>(new std::byte[Task::MAX_DATA_SIZE + sizeof(Task)]);
+    m_receiver->receive_data(reinterpret_cast<std::byte *>(task), Task::MAX_DATA_SIZE);
+    return task;
+};
+
+ZmqSink::~ZmqSink() { delete m_receiver; };
+
+} // namespace aare

network_io/src/ZmqSocket.cpp

@@ -25,6 +25,12 @@ ZmqSocket::~ZmqSocket() {
     delete[] m_header_buffer;
 }
 
+/**
+ * @brief get the socket
+ * @return void*
+ */
+void *ZmqSocket::get_socket() { return m_socket; }
+
 void ZmqSocket::set_zmq_hwm(int hwm) { m_zmq_hwm = hwm; }
 
 void ZmqSocket::set_timeout_ms(int n) { m_timeout_ms = n; }

network_io/src/ZmqSocketReceiver.cpp

@@ -9,8 +9,9 @@ namespace aare {
 /**
  * @brief Construct a new ZmqSocketReceiver object
  */
-ZmqSocketReceiver::ZmqSocketReceiver(const std::string &endpoint) {
-    m_endpoint = endpoint;
+ZmqSocketReceiver::ZmqSocketReceiver(const std::string &endpoint, int socket_type) {
+    m_endpoint = (endpoint);
+    m_socket_type = (socket_type);
     memset(m_header_buffer, 0, m_max_header_size);
 }
 
@@ -20,22 +21,33 @@ ZmqSocketReceiver::ZmqSocketReceiver(const std::string &endpoint) {
  */
 void ZmqSocketReceiver::connect() {
     m_context = zmq_ctx_new();
-    m_socket = zmq_socket(m_context, ZMQ_SUB);
+    m_socket = zmq_socket(m_context, m_socket_type);
     fmt::print("Setting ZMQ_RCVHWM to {}\n", m_zmq_hwm);
     int rc = zmq_setsockopt(m_socket, ZMQ_RCVHWM, &m_zmq_hwm, sizeof(m_zmq_hwm)); // should be set before connect
     if (rc)
         throw network_io::NetworkError(fmt::format("Could not set ZMQ_RCVHWM: {}", zmq_strerror(errno)));
 
-    size_t bufsize = m_potential_frame_size * m_zmq_hwm;
+    int bufsize = static_cast<int>(m_potential_frame_size) * m_zmq_hwm;
     fmt::print("Setting ZMQ_RCVBUF to: {} MB\n", bufsize / (static_cast<size_t>(1024) * 1024));
     rc = zmq_setsockopt(m_socket, ZMQ_RCVBUF, &bufsize, sizeof(bufsize));
-    if (rc)
+    if (rc) {
+        perror("zmq_setsockopt");
         throw network_io::NetworkError(fmt::format("Could not set ZMQ_RCVBUF: {}", zmq_strerror(errno)));
-
+    }
     zmq_connect(m_socket, m_endpoint.c_str());
     zmq_setsockopt(m_socket, ZMQ_SUBSCRIBE, "", 0);
 }
 
+void ZmqSocketReceiver::bind() {
+    m_context = zmq_ctx_new();
+    m_socket = zmq_socket(m_context, m_socket_type);
+    size_t const rc = zmq_bind(m_socket, m_endpoint.c_str());
+    if (rc != 0) {
+        std::string const error = zmq_strerror(zmq_errno());
+        throw network_io::NetworkError("zmq_bind failed: " + error);
+    }
+}
+
 /**
  * @brief receive a ZmqHeader
  * @return ZmqHeader
@@ -43,15 +55,13 @@ void ZmqSocketReceiver::connect() {
 ZmqHeader ZmqSocketReceiver::receive_header() {
 
     // receive string ZmqHeader
-    aare::logger::debug("Receiving header");
     int const header_bytes_received = zmq_recv(m_socket, m_header_buffer, m_max_header_size, 0);
-    aare::logger::debug("Bytes: ", header_bytes_received);
+    aare::logger::debug("Header: ", m_header_buffer);
 
     m_header_buffer[header_bytes_received] = '\0'; // make sure we zero terminate
     if (header_bytes_received < 0) {
         throw network_io::NetworkError(LOCATION + "Error receiving header");
     }
-    aare::logger::debug("Bytes: ", header_bytes_received, ", Header: ", m_header_buffer);
 
     // parse header
     ZmqHeader header;
@@ -72,9 +82,12 @@ ZmqHeader ZmqSocketReceiver::receive_header() {
  */
 int ZmqSocketReceiver::receive_data(std::byte *data, size_t size) {
     int const data_bytes_received = zmq_recv(m_socket, data, size, 0);
-    if (data_bytes_received == -1)
-        throw network_io::NetworkError("Got half of a multipart msg!!!");
-    aare::logger::debug("Bytes: ", data_bytes_received);
+    if (data_bytes_received == -1) {
+        logger::error(zmq_strerror(zmq_errno()));
+        // TODO: refactor this error message
+        throw network_io::NetworkError(LOCATION + "Error receiving data");
+    }
+    // aare::logger::debug("Bytes: ", data_bytes_received);
 
     return data_bytes_received;
 }
@@ -93,7 +106,13 @@ ZmqFrame ZmqSocketReceiver::receive_zmqframe() {
     }
 
     // receive frame data
-    Frame frame(header.npixelsx, header.npixelsy, header.dynamicRange);
+    if (header.shape == t_xy<uint32_t>{0, 0} || header.bitmode == 0) {
+        logger::warn("Invalid header");
+    }
+    if (header.bitmode == 0) {
+        header.bitmode = 16;
+    }
+    Frame frame(header.shape.row, header.shape.col, header.bitmode);
     int bytes_received = receive_data(frame.data(), frame.size());
     if (bytes_received == -1) {
         throw network_io::NetworkError(LOCATION + "Error receiving frame");

network_io/src/ZmqSocketSender.cpp

@@ -8,14 +8,17 @@ namespace aare {
  * Constructor
  * @param endpoint ZMQ endpoint
  */
-ZmqSocketSender::ZmqSocketSender(const std::string &endpoint) { m_endpoint = endpoint; }
+ZmqSocketSender::ZmqSocketSender(const std::string &endpoint, int socket_type) {
+    m_socket_type = socket_type;
+    m_endpoint = endpoint;
+}
 
 /**
  * bind to the given port
  */
 void ZmqSocketSender::bind() {
     m_context = zmq_ctx_new();
-    m_socket = zmq_socket(m_context, ZMQ_PUB);
+    m_socket = zmq_socket(m_context, m_socket_type);
     size_t const rc = zmq_bind(m_socket, m_endpoint.c_str());
     if (rc != 0) {
         std::string const error = zmq_strerror(zmq_errno());
@@ -23,6 +26,31 @@ void ZmqSocketSender::bind() {
     }
 }
 
+void ZmqSocketSender::connect() {
+    m_context = zmq_ctx_new();
+    m_socket = zmq_socket(m_context, m_socket_type);
+    fmt::print("Setting ZMQ_RCVHWM to {}\n", m_zmq_hwm);
+    int rc = zmq_setsockopt(m_socket, ZMQ_RCVHWM, &m_zmq_hwm, sizeof(m_zmq_hwm)); // should be set before connect
+    if (rc)
+        throw network_io::NetworkError(fmt::format("Could not set ZMQ_RCVHWM: {}", zmq_strerror(errno)));
+
+    int bufsize = static_cast<int>(m_potential_frame_size) * m_zmq_hwm;
+    fmt::print("Setting ZMQ_RCVBUF to: {} MB\n", bufsize / (static_cast<size_t>(1024) * 1024));
+    rc = zmq_setsockopt(m_socket, ZMQ_RCVBUF, &bufsize, sizeof(bufsize));
+    if (rc) {
+        perror("zmq_setsockopt");
+        throw network_io::NetworkError(fmt::format("Could not set ZMQ_RCVBUF: {}", zmq_strerror(errno)));
+    }
+    zmq_connect(m_socket, m_endpoint.c_str());
+    zmq_setsockopt(m_socket, ZMQ_SUBSCRIBE, "", 0);
+}
+
+size_t ZmqSocketSender::send(const void *data, size_t size) {
+    size_t const rc2 = zmq_send(m_socket, data, size, 0);
+    assert(rc2 == size);
+    return rc2;
+}
+
 /**
  * send a header and data
  * @param header
@@ -30,7 +58,7 @@ void ZmqSocketSender::bind() {
  * @param size size of data
  * @return number of bytes sent
  */
-size_t ZmqSocketSender::send(const ZmqHeader &header, const std::byte *data, size_t size) {
+size_t ZmqSocketSender::send(const ZmqHeader &header, const void *data, size_t size) {
     size_t rc = 0;
     // if (serialize_header) {
     //     rc = zmq_send(m_socket, &header, sizeof(ZmqHeader), ZMQ_SNDMORE);

network_io/src/ZmqVentilator.cpp

@@ -0,0 +1,21 @@
+#include "aare/network_io/ZmqVentilator.hpp"
+#include <cassert>
+#include <zmq.h>
+
+namespace aare {
+
+ZmqVentilator::ZmqVentilator(const std::string &endpoint) : m_sender(new ZmqSocketSender(endpoint, ZMQ_PUSH)) {
+    m_sender->bind();
+}
+
+size_t ZmqVentilator::push(const Task *task) {
+    if (task->data_size > Task::MAX_DATA_SIZE) {
+        throw network_io::NetworkError("Data size exceeds maximum allowed size");
+    }
+    logger::debug("Pushing workers");
+    return m_sender->send(task, task->size());
+}
+
+ZmqVentilator::~ZmqVentilator() { delete m_sender; }
+
+} // namespace aare

network_io/src/ZmqWorker.cpp

@@ -0,0 +1,39 @@
+#include "aare/network_io/ZmqWorker.hpp"
+#include "aare/network_io/ZmqSocketReceiver.hpp"
+#include "aare/network_io/ZmqSocketSender.hpp"
+#include "aare/network_io/defs.hpp"
+#include "zmq.h"
+namespace aare {
+
+ZmqWorker::ZmqWorker(const std::string &ventilator_endpoint, const std::string &sink_endpoint)
+    : m_receiver(new ZmqSocketReceiver(ventilator_endpoint, ZMQ_PULL)), m_sender(nullptr) {
+    m_receiver->connect();
+    if (not sink_endpoint.empty()) {
+        m_sender = new ZmqSocketSender(sink_endpoint, ZMQ_PUSH);
+        m_sender->connect();
+    }
+}
+
+Task *ZmqWorker::pull() {
+    Task *task = reinterpret_cast<Task *>(new std::byte[Task::MAX_DATA_SIZE + sizeof(Task)]);
+    m_receiver->receive_data(reinterpret_cast<std::byte *>(task), Task::MAX_DATA_SIZE);
+    logger::debug("Received task", task->id, task->data_size);
+    return task;
+}
+
+size_t ZmqWorker::push(const Task *task) {
+    if (m_sender == nullptr) {
+        throw network_io::NetworkError("Worker not connected to sink: did you provide a sink endpoint?");
+    }
+    if (task->data_size > Task::MAX_DATA_SIZE) {
+        throw network_io::NetworkError("Data size exceeds maximum allowed size");
+    }
+    return m_sender->send(task, task->size());
+}
+
+ZmqWorker::~ZmqWorker() {
+    delete m_receiver;
+    delete m_sender;
+}
+
+} // namespace aare

network_io/test/ZmqHeader.test.cpp

@@ -5,14 +5,12 @@
 using namespace aare;
 TEST_CASE("Test ZmqHeader") {
     ZmqHeader header;
-    header.npixelsx = 10;
-    header.npixelsy = 15;
+    header.shape = {10, 15};
     header.data = 1;
-    header.jsonversion = 2;
-    header.dynamicRange = 32;
+    header.jsonversion = 5;
+    header.bitmode = 32;
     header.fileIndex = 4;
-    header.ndetx = 5;
-    header.ndety = 6;
+    header.detshape = {5, 6};
     header.size = 4800;
     header.acqIndex = 8;
     header.frameIndex = 9;
@@ -39,13 +37,11 @@ TEST_CASE("Test ZmqHeader") {
 
     std::string json_header = "{"
                               "\"data\": 1, "
-                              "\"jsonversion\": 2, "
-                              "\"dynamicRange\": 32, "
+                              "\"jsonversion\": 5, "
+                              "\"bitmode\": 32, "
                               "\"fileIndex\": 4, "
-                              "\"ndetx\": 5, "
-                              "\"ndety\": 6, "
-                              "\"npixelsx\": 10, "
-                              "\"npixelsy\": 15, "
+                              "\"detshape\": [5, 6], "
+                              "\"shape\": [10, 15], "
                               "\"size\": 4800, "
                               "\"acqIndex\": 8, "
                               "\"frameIndex\": 9, "