ZMQ radio

This commit introduces ZMQ-based radio library. Each pair of RX/TX antennas is
represented by a ZMQ REQ/REP socket pair which streams continuous IQ samples
from radio start until stop.

Usage:
Simplest configuration is to connect OAI NR UE to OAI GNB with the same number
of antennas - by inverting the RX and TX channels in ZMQ radio configuration the
gNBs TX is mapped to UEs RX antennas and vice versa.

NOTICE

@@ -84,3 +84,11 @@ Nuand: https://github.com/Nuand/bladeRF/tree/master?tab=License-1-ov-file
 
 Credits for https://github.com/pothosware/SoapySDR
 Pothosware: BSL 1.0 License
+
+Credits for https://github.com/zeromq/libzmq
+ZeroMQ authors: Mozilla Public License Version 2.0
+
+Credits for source code:
+  - radio/zmq/zmq_imported.cpp
+  - radio/zmq/zmq_imported.h
+Software Radio Systems Limited: BSD-3-Clause-Open-MPI

ci-scripts/docker/Dockerfile.unittest.ubuntu

@@ -13,11 +13,12 @@ RUN apt-get update && \
     DEBIAN_FRONTEND=noninteractive apt-get upgrade --yes && \
     DEBIAN_FRONTEND=noninteractive apt-get install --yes \
         libgtest-dev \
-        libyaml-cpp-dev
+        libyaml-cpp-dev \
+        libzmq3-dev
 
 RUN rm -Rf /oai-ran
 WORKDIR /oai-ran
 COPY . .
 
 WORKDIR /oai-ran/build
-RUN cmake -GNinja -DENABLE_TESTS=ON -DCMAKE_BUILD_TYPE=Debug -DSANITIZE_ADDRESS=True .. && ninja tests
+RUN cmake -GNinja -DENABLE_TESTS=ON -DOAI_ZMQ=ON -DCMAKE_BUILD_TYPE=Debug -DSANITIZE_ADDRESS=True .. && ninja tests

common/config/config_load_configmodule.c

@@ -282,12 +282,9 @@ configmodule_interface_t *load_configmodule(int argc,
       cfgmode = strdup(CONFIG_LIBCONFIGFILE);
     }
   }
-  static configmodule_interface_t *cfgptr = NULL;
-  if (cfgptr)
-    fprintf(stderr, "ERROR: Call load_configmodule more than one time\n");
 
   // The macros are not thread safe print_params and similar
-  cfgptr = calloc(sizeof(configmodule_interface_t), 1);
+  configmodule_interface_t *cfgptr = calloc(sizeof(configmodule_interface_t), 1);
   if (!cfgptr) {
     fprintf(stderr, "ERROR: cannot allocate a memory for configuration\n");
     return NULL;

common/config/config_load_configmodule.h

@@ -10,6 +10,9 @@
 #ifndef INCLUDE_CONFIG_LOADCONFIGMODULE_H
 #define INCLUDE_CONFIG_LOADCONFIGMODULE_H
 
+#ifdef __cplusplus
+extern "C" {
+#endif
 
 #include <string.h>
 #include <stdlib.h>
@@ -117,5 +120,8 @@ void write_parsedcfg(configmodule_interface_t *cfg);
 extern void free_configmodule(void);
 #define CONFIG_PRINTF_ERROR(f, x... ) if (isLogInitDone ()) { LOG_E(ENB_APP,f,x);} else {printf(f,x);}; if ( !CONFIG_ISFLAGSET(CONFIG_NOABORTONCHKF) ) exit_fun("exit because configuration failed\n");
 
+#ifdef __cplusplus
+}
+#endif
 
 #endif  /* INCLUDE_CONFIG_LOADCONFIGMODULE_H */

common/config/config_paramdesc.h

@@ -181,6 +181,11 @@ typedef struct paramdef {
   {                                                                                                                          \
     OPTNAME(name), HELPSTR(help), PARAMFLAG(flags), .u64ptr = ptr, .defuintval = defval, PARAMTYPE(TYPE_UINT64), .numelt = 0 \
   }
+#define STRINGLISTPARAM(name, help, flags, ptr, defval)                                                                 \
+  {                                                                                                                     \
+    OPTNAME(name), HELPSTR(help), PARAMFLAG(flags), .strptr = ptr, .defstrlistval = defval, PARAMTYPE(TYPE_STRINGLIST), \
+        .numelt = 0                                                                                                     \
+  }
 
 typedef struct paramlist_def {
   char listname[MAX_OPTNAME_SIZE];

doc/README.md

@@ -108,6 +108,7 @@ Some directories under `radio` contain READMEs:
 - [fhi_72](../radio/fhi_72/README.md)
 - [vrtsim](../radio/vrtsim/README.md)
 - [rf_emulator](../radio/emulator/README.md)
+- [zmq](../radio/zmq/README.md)
 
 The other SDRs (AW2S, LimeSDR, ...) have no READMEs.
 

radio/CMakeLists.txt

@@ -51,3 +51,8 @@ add_boolean_option(OAI_RF_EMULATOR ON "Activate OAI's RF emulator" OFF)
 if(OAI_RF_EMULATOR)
   add_subdirectory(emulator)
 endif()
+
+add_boolean_option(OAI_ZMQ OFF "Activate OAI's ZMQ radio" OFF)
+if(OAI_ZMQ)
+  add_subdirectory(zmq)
+endif()

radio/zmq/CMakeLists.txt

@@ -0,0 +1,21 @@
+# SPDX-License-Identifier: LicenseRef-CSSL-1.0
+
+pkg_check_modules(zmq REQUIRED libzmq)
+add_library(ring_buffer ring_buffer.cpp)
+add_library(zmq_imported zmq_imported.cpp)
+target_include_directories(zmq_imported PUBLIC ./)
+target_link_libraries(zmq_imported PUBLIC ${zmq_LIBRARIES} log_headers)
+
+target_include_directories(ring_buffer PUBLIC ./)
+add_library(oai_zmqdevif SHARED zmq_radio.cpp)
+target_link_libraries(oai_zmqdevif PUBLIC ${zmq_LIBRARIES} ring_buffer log_headers zmq_imported)
+target_include_directories(oai_zmqdevif PUBLIC ${zmq_INCLUDE_DIRS})
+set_target_properties(oai_zmqdevif PROPERTIES LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
+
+add_library(zmq_radio_static STATIC zmq_radio.cpp)
+target_link_libraries(zmq_radio_static PUBLIC ${zmq_LIBRARIES} ring_buffer log_headers zmq_imported)
+target_include_directories(zmq_radio_static PUBLIC ${zmq_INCLUDE_DIRS})
+
+if (ENABLE_TESTS)
+  add_subdirectory(tests)
+endif()

radio/zmq/README.md

@@ -0,0 +1,40 @@
+<!-- SPDX-License-Identifier: CC-BY-4.0 -->
+
+# Overview
+
+This library implements ZMQ-based radio driver
+
+# Architecture
+
+The radio simulates RX/TX pairs between different processes over network or on the same machine
+as ZMQ REQ/REP socket pairs. All the antennas have to be configured and connected before the
+simulation can start.
+
+# Limitations
+
+You cannot reconnect a device once the simulation has started. The simulation has to be restarted
+from scratch in order to add devices or restart a process.
+
+# Requirements
+
+Depends on zmq library being installed in the system. On ubuntu: `libzmq3-dev`, on RHEL/Fedora: `zeromq-devel`
+
+# Usage
+
+## Simple 1-to-1 antenna mapping
+
+Add `--device.name oai_zmqdevif` to load the library in UE / gNB process. Add `--zmq.[0].tx_channels <channels>` and
+`--zmq.[0].rx_channels <channels>` to define ZMQ REQ/REP pairs.
+
+On the opposite side, load the library like specified above but invert the `rx` and `tx` channels. This way all
+antennas of UE will be directly mapped to all antennas of the gNB.
+
+### Example:
+
+```
+sudo ./nr-uesoftmodem -r 106 --numerology 1 --band 78 -C 3619200000 --device.name oai_zmqdevif --zmq.[0].tx_channels tcp://127.0.0.1:4557 --zmq.[0].rx_channels tcp://127.0.0.1:4556 --ssb 516
+```
+
+```
+sudo ./nr-softmodem -O ../targets/PROJECTS/GENERIC-NR-5GC/CONF/gnb.sa.band78.fr1.106PRB.usrpb210.conf --gNBs.[0].min_rxtxtime 6 --device.name oai_zmqdevif --zmq.[0].tx_channels tcp://127.0.0.1:4556 --zmq.[0].rx_channels tcp://127.0.0.1:4557
+```

radio/zmq/ring_buffer.cpp

@@ -0,0 +1,167 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#include "ring_buffer.h"
+#include <cstring>
+#include <iostream>
+#include <algorithm>
+
+ring_buffer::ring_buffer(size_t max_size) : max_size_(max_size)
+{
+  buffer_ = std::make_unique<cf_t[]>(max_size);
+}
+
+size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
+{
+  size_t overflow = 0;
+  // if nsamps > max_size skip nsamps - max_size samples
+  size_t nsamps_left = nsamps;
+  if (nsamps > max_size_) {
+    samples += nsamps - max_size_;
+    nsamps_left = max_size_;
+    overflow += nsamps - max_size_;
+  }
+
+  // Detect overflow
+  if (size_ + nsamps_left > max_size_) {
+    size_t newtail__pos = (head_ + nsamps_left) % max_size_;
+    overflow += (size_ + nsamps_left) - max_size_;
+    tail_ = newtail__pos;
+  }
+
+  size_t first_chunk = std::min(nsamps_left, max_size_ - head_);
+  memcpy(&buffer_[head_], samples, first_chunk * sizeof(cf_t));
+  head_ = (head_ + first_chunk) % max_size_;
+  samples += first_chunk;
+  nsamps_left -= first_chunk;
+  if (nsamps_left > 0) {
+    memcpy(&buffer_[0], samples, nsamps_left * sizeof(cf_t));
+    head_ = nsamps_left;
+  }
+
+  size_ = std::min(size_ + nsamps, max_size_);
+
+  return overflow;
+}
+
+size_t ring_buffer::push_zeros(const size_t num_zeros)
+{
+  size_t overflow = 0;
+  // if nsamps > max_size skip nsamps - max_size samples
+  size_t nsamps_left = num_zeros;
+  if (num_zeros > max_size_) {
+    nsamps_left = max_size_;
+    overflow += num_zeros - max_size_;
+  }
+
+  // Detect overflow
+  if (size_ + nsamps_left > max_size_) {
+    size_t new_tail_pos = (head_ + nsamps_left) % max_size_;
+    overflow += (size_ + nsamps_left) - max_size_;
+    tail_ = new_tail_pos;
+  }
+
+  size_t first_chunk = std::min(nsamps_left, max_size_ - head_);
+  memset(&buffer_[head_], 0, first_chunk * sizeof(cf_t));
+  head_ = (head_ + first_chunk) % max_size_;
+  nsamps_left -= first_chunk;
+  if (nsamps_left > 0) {
+    memset(&buffer_[0], 0, nsamps_left * sizeof(cf_t));
+    head_ = nsamps_left;
+  }
+
+  size_ = std::min(size_ + num_zeros, max_size_);
+
+  return overflow;
+}
+
+size_t ring_buffer::pop_samples(cf_t *samples, size_t num_samples)
+{
+  size_t samples_to_pop = std::min(size_, num_samples);
+  if (samples_to_pop > 0) {
+    if (tail_ + samples_to_pop > max_size_) {
+      size_t first_chunk = max_size_ - tail_;
+      memcpy(samples, &buffer_[tail_], first_chunk * sizeof(cf_t));
+      memcpy(samples + first_chunk, &buffer_[0], (samples_to_pop - first_chunk) * sizeof(cf_t));
+    } else {
+      memcpy(samples, &buffer_[tail_], samples_to_pop * sizeof(cf_t));
+    }
+    tail_ = (tail_ + samples_to_pop) % max_size_;
+    size_ -= samples_to_pop;
+    return samples_to_pop;
+  }
+  return 0;
+}
+
+void ring_buffer::clear_samples()
+{
+  head_ = 0;
+  tail_ = 0;
+  size_ = 0;
+}
+
+void ring_buffer::reset()
+{
+  clear_samples();
+}
+
+size_t ring_buffer::size() const
+{
+  return size_;
+}
+
+size_t overflow_buffer::push_samples(const cf_t *samples, size_t nsamps)
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  size_t overflow = buffer_.push_samples(samples, nsamps);
+  zeros_to_send_ += overflow;
+  return overflow;
+}
+
+size_t overflow_buffer::push_zeros(size_t num_zeros)
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  size_t overflow = buffer_.push_zeros(num_zeros);
+  zeros_to_send_ += overflow;
+  return overflow;
+}
+
+size_t overflow_buffer::pop_samples(cf_t *samples, size_t num_samples)
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  size_t samples_popped = 0;
+  if (zeros_to_send_ > 0) {
+    size_t num_zeros = std::min(zeros_to_send_, num_samples);
+    memset(samples, 0, num_zeros * sizeof(cf_t));
+    zeros_to_send_ -= num_zeros;
+    samples += num_zeros;
+    num_samples -= num_zeros;
+    samples_popped += num_zeros;
+  }
+
+  if (num_samples > 0) {
+    samples_popped += buffer_.pop_samples(samples, num_samples);
+  }
+  return samples_popped;
+}
+
+void overflow_buffer::reset()
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  buffer_.reset();
+  zeros_to_send_ = buffer_.size() / 2;
+}
+
+void overflow_buffer::clear_samples()
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  buffer_.clear_samples();
+  zeros_to_send_ = 0;
+}
+
+size_t overflow_buffer::size()
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  return buffer_.size() + zeros_to_send_;
+}

radio/zmq/ring_buffer.h

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#pragma once
+#include <vector>
+#include <mutex>
+#include <memory>
+#include "common/platform_types.h"
+
+// A basic cirular sample buffer class
+class ring_buffer {
+ private:
+  std::unique_ptr<cf_t[]> buffer_;
+  size_t head_ = 0;
+  size_t tail_ = 0;
+  size_t size_ = 0;
+  size_t max_size_;
+
+ public:
+  ring_buffer(size_t max_size = 614400);
+  size_t push_samples(const cf_t *samples, size_t nsamps);
+  size_t push_zeros(size_t num_zeros);
+  size_t pop_samples(cf_t *samples, size_t num_samples);
+  void reset();
+  void clear_samples();
+  size_t size() const;
+};
+
+// A thread-safe wrapper around ring_buffer that counts overflows
+class overflow_buffer {
+  ring_buffer buffer_;
+  size_t zeros_to_send_;
+  std::mutex mutex_;
+
+ public:
+  overflow_buffer(size_t max_size = 614400) : buffer_(max_size), zeros_to_send_(0)
+  {
+  }
+  size_t push_samples(const cf_t *samples, size_t nsamps);
+  size_t push_zeros(size_t num_zeros);
+  size_t pop_samples(cf_t *samples, size_t num_samples);
+  void reset();
+  void clear_samples();
+  size_t size();
+};

radio/zmq/tests/CMakeLists.txt

@@ -0,0 +1,13 @@
+# SPDX-License-Identifier: LicenseRef-CSSL-1.0
+
+add_executable(test_ring_buffer test_ring_buffer.cpp)
+target_link_libraries(test_ring_buffer PRIVATE ring_buffer GTest::gtest)
+add_dependencies(tests test_ring_buffer)
+add_test(NAME test_ring_buffer
+  COMMAND ./test_ring_buffer)
+
+add_executable(test_zmq_radio test_zmq_radio.cpp)
+target_link_libraries(test_zmq_radio PRIVATE zmq_radio_static UTIL CONFIG_LIB GTest::gtest zmq)
+add_dependencies(tests test_zmq_radio)
+add_test(NAME test_zmq_radio
+    COMMAND ./test_zmq_radio)

radio/zmq/tests/test_ring_buffer.cpp

@@ -0,0 +1,221 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#include <gtest/gtest.h>
+#include "ring_buffer.h"
+
+TEST(CircularBuffer, simplePushPop)
+{
+  ring_buffer cb(10);
+  cf_t data[10];
+  for (int i = 0; i < 10; i++) {
+    data[i] = { (float)i, (float)(i + 1)};
+  }
+  ASSERT_EQ(cb.push_samples(data, 10), 0);
+
+  cf_t read_data[10];
+  ASSERT_EQ(cb.pop_samples(read_data, 10), 10);
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(read_data[i].r, data[i].r);
+    ASSERT_EQ(read_data[i].i, data[i].i);
+  }
+
+  ASSERT_EQ(cb.size(), 0);
+  ASSERT_EQ(cb.pop_samples(read_data, 10), 0);
+}
+
+TEST(CircularBuffer, overflow)
+{
+  ring_buffer cb(10);
+  cf_t data[15];
+  for (int i = 0; i < 15; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  ASSERT_EQ(cb.push_samples(data, 15), 5);
+  cf_t read_data[10];
+  ASSERT_EQ(cb.pop_samples(read_data, 10), 10);
+  // The first 5 samples are lost due to overflow
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(read_data[i].r, i + 5);
+    ASSERT_EQ(read_data[i].i, i + 6);
+  }
+}
+
+TEST(CircularBuffer, partialPop)
+{
+  ring_buffer cb(10);
+  cf_t data[10];
+  for (int i = 0; i < 10; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  ASSERT_EQ(cb.push_samples(data, 10), 0);
+  cf_t read_data[5];
+  ASSERT_EQ(cb.pop_samples(read_data, 5), 5);
+  for (int i = 0; i < 5; i++) {
+    ASSERT_EQ(read_data[i].r, i);
+    ASSERT_EQ(read_data[i].i, i + 1);
+  }
+  ASSERT_EQ(cb.size(), 5);
+  ASSERT_EQ(cb.pop_samples(read_data, 5), 5);
+  for (int i = 0; i < 5; i++) {
+    ASSERT_EQ(read_data[i].r, i + 5);
+    ASSERT_EQ(read_data[i].i, i + 6);
+  }
+}
+
+TEST(CircularBuffer, popMore)
+{
+  ring_buffer cb(10);
+  cf_t data[10];
+  for (int i = 0; i < 10; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  cb.push_samples(data, 10);
+  cf_t read_data[15];
+  ASSERT_EQ(cb.pop_samples(read_data, 15), 10);
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(read_data[i].r, i);
+    ASSERT_EQ(read_data[i].i, i + 1);
+  }
+}
+
+TEST(CircularBuffer, newOverflow)
+{
+  ring_buffer cb(10);
+  cf_t data1[8];
+  for (int i = 0; i < 8; i++) {
+    data1[i] = {(float)i, (float)(i + 1)};
+  }
+  ASSERT_EQ(cb.push_samples(data1, 8), 0);
+
+  cf_t data2[4];
+  for (int i = 0; i < 4; i++) {
+    data2[i] = {(float)(i + 8), (float)(i + 9)};
+  }
+  ASSERT_EQ(cb.push_samples(data2, 4), 2);
+
+  cf_t read_data[10];
+  ASSERT_EQ(cb.pop_samples(read_data, 10), 10);
+  // The first 2 samples are lost due to overflow
+  for (int i = 0; i < 8; i++) {
+    ASSERT_EQ(read_data[i].r, i + 2);
+    ASSERT_EQ(read_data[i].i, i + 3);
+  }
+  ASSERT_EQ(read_data[8].r, 10);
+  ASSERT_EQ(read_data[8].i, 11);
+  ASSERT_EQ(read_data[9].r, 11);
+  ASSERT_EQ(read_data[9].i, 12);
+}
+
+TEST(overflow_buffer, simplePushPop)
+{
+  overflow_buffer pb(10);
+  cf_t data[10];
+  for (int i = 0; i < 10; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  pb.push_samples(data, 10);
+  ASSERT_EQ(pb.size(), 10);
+  cf_t read_data[10];
+  ASSERT_EQ(pb.pop_samples(read_data, 10), 10);
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(read_data[i].r, i);
+    ASSERT_EQ(read_data[i].i, i + 1);
+  }
+}
+
+TEST(overflow_buffer, zeros_to_send)
+{
+  overflow_buffer pb(10);
+  pb.push_zeros(10);
+  cf_t data[5];
+  for (int i = 0; i < 5; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  pb.push_samples(data, 5);
+  ASSERT_EQ(pb.size(), 15);
+  cf_t read_data[10];
+  ASSERT_EQ(pb.pop_samples(read_data, 10), 10);
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(read_data[i].r, 0);
+    ASSERT_EQ(read_data[i].i, 0);
+  }
+  ASSERT_EQ(pb.pop_samples(read_data, 10), 5);
+  for (int i = 0; i < 5; i++) {
+    ASSERT_EQ(read_data[i].r, i);
+    ASSERT_EQ(read_data[i].i, i + 1);
+  }
+}
+
+TEST(overflow_buffer, no_zeros_to_send)
+{
+  overflow_buffer pb(10);
+  cf_t data[10];
+  for (int i = 0; i < 10; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  pb.push_samples(data, 10);
+  ASSERT_EQ(pb.size(), 10);
+  cf_t read_data[10];
+  ASSERT_EQ(pb.pop_samples(read_data, 10), 10);
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(read_data[i].r, i);
+    ASSERT_EQ(read_data[i].i, i + 1);
+  }
+}
+
+TEST(overflow_buffer, overflow)
+{
+  overflow_buffer pb(10);
+  cf_t data[15];
+  for (int i = 0; i < 15; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  pb.push_samples(data, 15);
+  ASSERT_EQ(pb.size(), 15);
+  cf_t read_data[15];
+  ASSERT_EQ(pb.pop_samples(read_data, 15), 15);
+  // The first 5 samples were lost due to overflow.
+  int i = 0;
+  for (; i < 5; i++) {
+    ASSERT_EQ(read_data[i].r, 0);
+    ASSERT_EQ(read_data[i].i, 0);
+  }
+  for (; i < 15; i++) {
+    ASSERT_EQ(read_data[i].r, i);
+    ASSERT_EQ(read_data[i].i, i + 1);
+  }
+  ASSERT_EQ(pb.size(), 0);
+}
+
+TEST(overflow_buffer, push_zeros)
+{
+  overflow_buffer pb(10);
+  pb.push_zeros(5);
+  ASSERT_EQ(pb.size(), 5);
+  cf_t data[5];
+  for (int i = 0; i < 5; i++) {
+    data[i] = {(float)i, (float)(i + 1)};
+  }
+  pb.push_samples(data, 5);
+  ASSERT_EQ(pb.size(), 10);
+  cf_t read_data[10];
+  ASSERT_EQ(pb.pop_samples(read_data, 10), 10);
+  for (int i = 0; i < 10; i++) {
+    if (i < 5) {
+      ASSERT_EQ(read_data[i].r, 0);
+      ASSERT_EQ(read_data[i].i, 0);
+    } else {
+      ASSERT_EQ(read_data[i].r, data[i - 5].r);
+      ASSERT_EQ(read_data[i].i, data[i - 5].i);
+    }
+  }
+  ASSERT_EQ(pb.size(), 0);
+}
+
+int main(int argc, char **argv)
+{
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

radio/zmq/tests/test_zmq_radio.cpp

@@ -0,0 +1,162 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#include "common_lib.h"
+#include <gtest/gtest.h>
+#include "common/config/config_userapi.h"
+#include <zmq.h>
+extern "C" {
+#include "common/config/config_userapi.h"
+#include "openair1/SIMULATION/TOOLS/sim.h"
+extern int device_init(openair0_device_t *device, openair0_config_t *openair0_cfg);
+static softmodem_params_t softmodem_params;
+softmodem_params_t *get_softmodem_params(void)
+{
+  return &softmodem_params;
+}
+}
+#include "common/platform_types.h"
+#include <thread>
+
+configmodule_interface_t *uniqCfg = NULL;
+
+extern "C" void exit_function(const char *file, const char *function, const int line, const char *s, const int assert)
+{
+  fprintf(stderr, "FATAL: %s at %s:%s:%d\n", s, file, function, line);
+  exit(EXIT_FAILURE);
+}
+
+class ZMQTest : public ::testing::Test {
+ protected:
+  configmodule_interface_t *cfg1 = nullptr;
+  configmodule_interface_t *cfg2 = nullptr;
+  openair0_device_t device1 = {0};
+  openair0_device_t device2 = {0};
+  openair0_config_t config1 = {0};
+  openair0_config_t config2 = {0};
+  char *argv[4];
+
+  void SetUp() override
+  {
+    argv[0] = strdup("--zmq.[0].tx_channels");
+    argv[1] = strdup("tcp://127.0.0.1:5555");
+    argv[2] = strdup("--zmq.[0].rx_channels");
+    argv[3] = strdup("tcp://127.0.0.1:5556");
+
+    cfg1 = load_configmodule(sizeofArray(argv), argv, CONFIG_ENABLECMDLINEONLY);
+    uniqCfg = cfg1;
+    // 2. Initialize the ZMQ device
+    config1.tx_num_channels = 1;
+    config1.rx_num_channels = 1;
+    config1.sample_rate = 30;
+    ASSERT_EQ(device_init(&device1, &config1), 0);
+    ASSERT_EQ(device1.trx_start_func(&device1), 0);
+
+    // Swap the RX with TX for second device
+    char *tmp = argv[0];
+    argv[0] = argv[2];
+    argv[2] = tmp;
+    cfg2 = load_configmodule(sizeofArray(argv), argv, CONFIG_ENABLECMDLINEONLY);
+    uniqCfg = cfg2;
+    config2.tx_num_channels = 1;
+    config2.rx_num_channels = 1;
+    config2.sample_rate = 1500;
+    ASSERT_EQ(device_init(&device2, &config2), 0);
+    ASSERT_EQ(device2.trx_start_func(&device2), 0);
+  }
+
+  void TearDown() override
+  {
+    if (device1.trx_end_func) {
+      device1.trx_end_func(&device1);
+    }
+    if (device2.trx_end_func) {
+      device2.trx_end_func(&device2);
+    }
+    if (cfg1) {
+      end_configmodule(cfg1);
+    }
+    if (cfg2) {
+      end_configmodule(cfg2);
+    }
+    for (auto i = 0U; i < sizeofArray(argv); i++) {
+      free(argv[i]);
+    }
+  }
+};
+
+TEST_F(ZMQTest, RXSamples)
+{
+  std::thread t1([this]() {
+    c16_t rx_samples[10];
+    openair0_timestamp_t rx_timestamp;
+    void *samples[1] = {rx_samples};
+    ASSERT_EQ(device1.trx_read_func(&device1, &rx_timestamp, samples, 10, 1), 10);
+    for (int i = 0; i < 10; i++) {
+      ASSERT_EQ(rx_samples[i].r, 0);
+      ASSERT_EQ(rx_samples[i].i, 0);
+    }
+  });
+  std::thread t2([this]() {
+    c16_t rx_samples[10];
+    openair0_timestamp_t rx_timestamp;
+    void *samples[1] = {rx_samples};
+    ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp, samples, 10, 1), 10);
+    for (int i = 0; i < 10; i++) {
+      ASSERT_EQ(rx_samples[i].r, 0);
+      ASSERT_EQ(rx_samples[i].i, 0);
+    }
+  });
+  t1.join();
+  t2.join();
+}
+
+TEST_F(ZMQTest, TxRxSamples)
+{
+  std::thread t1([this]() {
+    c16_t rx_samples[10];
+    openair0_timestamp_t rx_timestamp;
+    void *samples[1] = {rx_samples};
+    ASSERT_EQ(device1.trx_read_func(&device1, &rx_timestamp, samples, 10, 1), 10);
+    for (int i = 0; i < 10; i++) {
+      ASSERT_EQ(rx_samples[i].r, 0);
+      ASSERT_EQ(rx_samples[i].i, 0);
+    }
+    c16_t tx_samples[10];
+    openair0_timestamp_t tx_timestamp = rx_timestamp + 10;
+    for (int i = 0; i < 10; i++) {
+      tx_samples[i].r = i;
+      tx_samples[i].i = i + 1;
+    }
+    samples[0] = tx_samples;
+    ASSERT_EQ(device1.trx_write_func(&device1, tx_timestamp, samples, 10, 1, 0), 10);
+  });
+  std::thread t2([this]() {
+    c16_t rx_samples[10];
+    openair0_timestamp_t rx_timestamp;
+    void *samples[1] = {rx_samples};
+    ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp, samples, 10, 1), 10);
+    for (int i = 0; i < 10; i++) {
+      ASSERT_EQ(rx_samples[i].r, 0);
+      ASSERT_EQ(rx_samples[i].i, 0);
+    }
+    openair0_timestamp_t rx_timestamp2;
+    ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp2, samples, 10, 1), 10);
+    for (int i = 0; i < 10; i++) {
+      ASSERT_EQ(rx_samples[i].r, i);
+      ASSERT_EQ(rx_samples[i].i, i + 1);
+    }
+    ASSERT_EQ(rx_timestamp + 10, rx_timestamp2);
+  });
+  t1.join();
+  t2.join();
+}
+
+int main(int argc, char **argv)
+{
+  logInit();
+  g_log->log_component[HW].level = OAILOG_DEBUG;
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

radio/zmq/zmq_imported.cpp

@@ -0,0 +1,131 @@
+/*
+ * SPDX-License-Identifier: BSD-3-Clause-Open-MPI
+ * Based on zmq library in OCUDU project: ocudu/lib/radio/zmq
+ * Refer to https://gitlab.com/ocudu/ocudu/-/raw/dev/LICENSE?ref_type=heads
+ */
+
+#include "zmq_imported.h"
+#include "log.h"
+
+const float c16_t_to_cf_t_factor = std::numeric_limits<int16_t>::max();
+static constexpr std::chrono::milliseconds TRANSMIT_TS_ALIGN_TIMEOUT = std::chrono::milliseconds(0);
+static constexpr std::chrono::milliseconds RECEIVE_TS_ALIGN_TIMEOUT = std::chrono::milliseconds(100);
+
+void zmq_tx_channel::transmit(c16_t *samples, size_t nsamps, uint64_t timestamp)
+{
+  std::scoped_lock lock(transmit_alignment_mutex_);
+  size_t overflow = 0;
+  if (timestamp > sample_count_) {
+    overflow += buffer_.push_zeros(timestamp - sample_count_);
+    sample_count_ = timestamp;
+  }
+  cf_t samples_float[nsamps];
+  for (size_t i = 0; i < nsamps; i++) {
+    samples_float[i].r = samples[i].r / c16_t_to_cf_t_factor;
+    samples_float[i].i = samples[i].i / c16_t_to_cf_t_factor;
+  }
+  overflow += buffer_.push_samples(samples_float, nsamps);
+  sample_count_ += nsamps;
+  if (overflow) {
+    LOG_W(HW, "Overflow on ZMQ channel by %lu samples\n", overflow);
+  }
+  is_tx_enabled_ = true;
+  transmit_alignment_cvar_.notify_all();
+}
+
+void zmq_tx_channel::start(uint64_t init_time)
+{
+  sample_count_ = init_time;
+}
+
+bool zmq_tx_channel::align(uint64_t timestamp, std::chrono::milliseconds timeout)
+{
+  if (sample_count_ >= timestamp) {
+    return sample_count_ > timestamp;
+  }
+  std::unique_lock<std::mutex> lock(transmit_alignment_mutex_);
+  if (is_tx_enabled_ && (timeout.count() != 0)) {
+    bool is_not_timeout =
+        transmit_alignment_cvar_.wait_for(lock, timeout, [this, timestamp]() { return sample_count_ >= timestamp; });
+    if (is_not_timeout) {
+      return sample_count_ > timestamp;
+    }
+    LOG_W(HW, "Timeout waiting for TX path to align samples\n");
+    is_tx_enabled_ = false;
+  }
+  if (sample_count_ < timestamp) {
+    buffer_.push_zeros(timestamp - sample_count_);
+    sample_count_ = timestamp;
+  }
+  return false;
+}
+
+void zmq_rx_channel::receive(c16_t *samples, size_t nsamps)
+{
+  size_t samples_popped = 0;
+  cf_t samples_float[nsamps];
+  while (samples_popped < (size_t)nsamps && !stopped_) {
+    size_t popped_now = buffer_.pop_samples(samples_float + samples_popped, nsamps - samples_popped);
+    samples_popped += popped_now;
+    if (popped_now == 0) {
+      usleep(100); // wait for more samples to arrive
+    }
+  }
+  for (size_t i = 0; i < nsamps; i++) {
+    samples[i].r = samples_float[i].r * c16_t_to_cf_t_factor + 0.5;
+    samples[i].i = samples_float[i].i * c16_t_to_cf_t_factor + 0.5;
+  }
+}
+void zmq_rx_channel::stop()
+{
+  stopped_ = true;
+}
+
+void zmq_tx_stream::start(uint64_t init_time)
+{
+  for (auto &chan : channels_) {
+    chan->start(init_time);
+  }
+}
+bool zmq_tx_stream::align(uint64_t timestamp, std::chrono::milliseconds timeout)
+{
+  bool timestamp_passed = false;
+  for (auto &chan : channels_) {
+    timestamp_passed = timestamp_passed || chan->align(timestamp, timeout);
+  }
+  return timestamp_passed;
+}
+void zmq_tx_stream::transmit(c16_t **samples, size_t nsamps, uint64_t timestamp)
+{
+  bool timestamp_passed = align(timestamp, TRANSMIT_TS_ALIGN_TIMEOUT);
+  if (timestamp_passed) {
+    LOG_W(HW, "Error, channel timeout\n");
+    return;
+  }
+  int i = 0;
+  for (auto chan : channels_) {
+    chan->transmit(samples[i++], nsamps, timestamp);
+  }
+}
+
+void zmq_rx_stream::start(uint64_t init_time)
+{
+  sample_count_ = init_time;
+}
+void zmq_rx_stream::stop()
+{
+  for (auto &chan : channels_) {
+    chan->stop();
+  }
+}
+void zmq_rx_stream::receive(c16_t **samples, size_t nsamps, uint64_t *timestamp)
+{
+  *timestamp = sample_count_;
+  uint64_t passed_timestamp = sample_count_ + nsamps;
+  tx_stream_->align(passed_timestamp, RECEIVE_TS_ALIGN_TIMEOUT);
+  int i = 0;
+  for (auto chan : channels_) {
+    chan->receive(samples[i++], nsamps);
+  }
+  sample_count_ += nsamps;
+}

radio/zmq/zmq_imported.h

@@ -0,0 +1,71 @@
+/*
+ * SPDX-License-Identifier: BSD-3-Clause-Open-MPI
+ * Based on zmq library in OCUDU project: ocudu/lib/radio/zmq
+ * Refer to https://gitlab.com/ocudu/ocudu/-/raw/dev/LICENSE?ref_type=heads
+ */
+
+#ifndef ZMQ_IMPORTED_H
+#define ZMQ_IMPORTED_H
+
+#include <zmq.h>
+#include "ring_buffer.h"
+#include <condition_variable>
+#include <atomic>
+#include <mutex>
+#include <vector>
+
+class zmq_tx_channel {
+ public:
+  void *socket_;
+  overflow_buffer buffer_;
+  std::atomic<uint64_t> sample_count_ = 0;
+  std::atomic<bool> is_tx_enabled_ = false;
+  std::mutex transmit_alignment_mutex_;
+  std::condition_variable transmit_alignment_cvar_;
+
+  zmq_tx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size)
+  {
+  }
+
+  void transmit(c16_t *samples, size_t nsamps, uint64_t timestamp);
+
+  void start(uint64_t init_time);
+
+  bool align(uint64_t timestamp, std::chrono::milliseconds timeout);
+};
+
+class zmq_rx_channel {
+ public:
+  void *socket_;
+  overflow_buffer buffer_;
+  bool request_sent_;
+  std::atomic<bool> stopped_;
+  zmq_rx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size), stopped_(false)
+  {
+  }
+  void receive(c16_t *samples, size_t nsamps);
+  void stop();
+};
+
+class zmq_tx_stream {
+ public:
+  std::vector<zmq_tx_channel *> channels_;
+  void start(uint64_t init_time);
+  bool align(uint64_t timestamp, std::chrono::milliseconds timeout);
+  void transmit(c16_t **samples, size_t nsamps, uint64_t timestamp);
+};
+
+class zmq_rx_stream {
+ public:
+  std::vector<zmq_rx_channel *> channels_;
+  zmq_tx_stream *tx_stream_;
+  uint64_t sample_count_ = 0;
+  zmq_rx_stream() : sample_count_(0)
+  {
+  }
+  void start(uint64_t init_time);
+  void stop();
+  void receive(c16_t **samples, size_t nsamps, uint64_t *timestamp);
+};
+
+#endif

radio/zmq/zmq_radio.cpp

@@ -0,0 +1,345 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#include "PHY/TOOLS/tools_defs.h"
+#include "PHY/defs_common.h"
+#include "common/platform_types.h"
+#include "softmodem-common.h"
+#include "utils.h"
+#include <chrono>
+#include <cstdint>
+#include <limits>
+#include <stddef.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+#include <stdbool.h>
+#include <errno.h>
+#include <sys/epoll.h>
+#include <netdb.h>
+
+#include <common/utils/assertions.h>
+#include <common/utils/LOG/log.h>
+#include <common/config/config_userapi.h>
+#include "common_lib.h"
+#include <queue>
+#include <mutex>
+#include <vector>
+#include <sstream>
+#include <algorithm>
+#include <numeric>
+#include <thread>
+#include <atomic>
+#include <condition_variable>
+#include <ring_buffer.h>
+#include <zmq.h>
+#include "zmq_imported.h"
+
+#define ZMQ_SECTION "zmq"
+#define ZMQ_TX_CHANNELS "tx_channels"
+#define ZMQ_RX_CHANNELS "rx_channels"
+
+#define ZMQ_PARAMS_DESC                                                                                                           \
+  {                                                                                                                               \
+      STRINGLISTPARAM(ZMQ_TX_CHANNELS, "list of zmq addresses represeting tx channels_\n", PARAMFLAG_MANDATORY, nullptr, nullptr), \
+      STRINGLISTPARAM(ZMQ_RX_CHANNELS, "list of zmq addresses represeting rx channels_\n", PARAMFLAG_MANDATORY, nullptr, nullptr), \
+  };
+
+const size_t sample_size = sizeof(cf_t);
+const size_t rx_buffer_size = sample_size * 300000;
+
+typedef struct {
+  void *context;
+  zmq_tx_stream tx_stream;
+  zmq_rx_stream rx_stream;
+  std::thread poll_thread;
+  std::atomic<bool> poll_thread_running;
+  bool stopped = false;
+  double sample_rate;
+} zmq_state_t;
+
+static void poll_thread(zmq_state_t *s)
+{
+  s->poll_thread_running = true;
+  unsigned char *rx_buffer = static_cast<unsigned char *>(malloc(rx_buffer_size));
+  const auto num_tx_channels = s->tx_stream.channels_.size();
+  const auto num_rx_channels = s->rx_stream.channels_.size();
+  std::vector<zmq_pollitem_t> items(num_tx_channels + num_rx_channels);
+  std::vector<bool> reply_requested(num_tx_channels);
+  for (size_t i = 0; i < num_tx_channels; ++i) {
+    items[i] = {s->tx_stream.channels_[i]->socket_, 0, ZMQ_POLLIN, 0};
+    // wait for REQ
+    reply_requested[i] = false;
+  }
+  for (size_t i = 0; i < num_rx_channels; i++) {
+    items[i + num_tx_channels] = {s->rx_stream.channels_[i]->socket_, 0, ZMQ_POLLIN, 0};
+  }
+
+  const auto num_channels = num_tx_channels + num_rx_channels;
+  while (s->poll_thread_running) {
+    for (size_t i = 0; i < num_tx_channels; i++) {
+      auto chan = s->tx_stream.channels_[i];
+      if (!reply_requested[i]) {
+        continue;
+      }
+      std::vector<cf_t> samples(1024);
+      size_t num_popped = chan->buffer_.pop_samples(samples.data(), 1024);
+      if (num_popped == 0) {
+        continue;
+      }
+      int rc = zmq_send(chan->socket_, samples.data(), num_popped * sizeof(cf_t), 0);
+      if (rc < 0) {
+        LOG_E(HW, "[ZMQ] poll_thread zmq_send for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+      }
+      reply_requested[i] = false;
+    }
+
+    int rc = zmq_poll(items.data(), num_channels, 10); // 10ms timeout
+    if (rc < 0) {
+      if (errno == EINTR)
+        continue;
+      LOG_E(HW, "[ZMQ] poll_thread zmq_poll failed: %s\n", zmq_strerror(errno));
+      break;
+    }
+    if (rc == 0) {
+      continue; // timeout
+    }
+
+    // --- TX Sockets (ZMQ_REP) ---
+    for (size_t i = 0; i < num_tx_channels; i++) {
+      if (items[i].revents & ZMQ_POLLIN) {
+        auto chan = s->tx_stream.channels_[i];
+        char dummy;
+        rc = zmq_recv(chan->socket_, &dummy, 1, 0);
+        if (rc < 0) {
+          LOG_E(HW, "[ZMQ] poll_thread zmq_recv for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+          continue;
+        }
+        if (reply_requested[i]) {
+          LOG_E(HW, "[ZMQ] Error, unexpected REQ before REP on TX antenna %d\n", (int)i);
+        }
+        reply_requested[i] = true;
+      }
+    }
+
+    // --- RX Sockets (ZMQ_REQ) ---
+    for (size_t i = 0; i < num_rx_channels; i++) {
+      if (items[i + num_tx_channels].revents & ZMQ_POLLIN) {
+        auto chan = s->rx_stream.channels_[i];
+        rc = zmq_recv(chan->socket_, rx_buffer, rx_buffer_size, 0);
+        if (rc < 0) {
+          LOG_E(HW, "[ZMQ] poll_thread zmq_recv for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+        } else {
+          size_t received_bytes = rc;
+          if (rx_buffer_size < received_bytes) {
+            LOG_W(HW,
+                  "[ZMQ] the RX buffer is too small! The received message size is %lu while the buffer is %lu. Message truncated\n",
+                  received_bytes,
+                  rx_buffer_size);
+          }
+          size_t num_samples_received = std::min(received_bytes, rx_buffer_size) / sizeof(cf_t);
+          cf_t *samples = reinterpret_cast<cf_t *>(rx_buffer);
+          size_t overflow = chan->buffer_.push_samples(samples, num_samples_received);
+          if (rx_buffer_size < received_bytes) {
+            overflow += chan->buffer_.push_zeros((received_bytes - rx_buffer_size) / sizeof(cf_t));
+          }
+          if (overflow) {
+            LOG_W(HW, "Overflow on receive\n");
+          }
+          // After receiving, send next request to keep the stream flowing
+          char dummy = 0;
+          if (zmq_send(chan->socket_, &dummy, 1, 0) != 1) {
+            LOG_E(HW, "[ZMQ] poll_thread zmq_send for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+          }
+        }
+      }
+    }
+  }
+  free(rx_buffer);
+}
+
+static int zmq_write(openair0_device_t *device, openair0_timestamp_t timestamp, void **buff, int nsamps, int cc, int flags)
+{
+  zmq_state_t *s = static_cast<zmq_state_t *>(device->priv);
+  AssertFatal((uint)cc == s->tx_stream.channels_.size(),
+              "Request to write on more antennas (%d) than configured (%d)",
+              cc,
+              (int)s->tx_stream.channels_.size());
+
+  s->tx_stream.transmit((c16_t **)buff, nsamps, timestamp);
+
+  return nsamps;
+}
+
+static int zmq_read(openair0_device_t *device, openair0_timestamp_t *ptimestamp, void **samplesVoid, int nsamps, int nbAnt)
+{
+  zmq_state_t *s = static_cast<zmq_state_t *>(device->priv);
+  AssertFatal((uint)nbAnt == s->rx_stream.channels_.size(),
+              "Request to read on more antennas (%d) than configured (%d)",
+              nbAnt,
+              (int)s->rx_stream.channels_.size());
+  uint64_t timestamp;
+  s->rx_stream.receive((c16_t **)samplesVoid, nsamps, &timestamp);
+  *ptimestamp = timestamp;
+  return nsamps;
+}
+
+static int zmq_get_stats(openair0_device_t *device)
+{
+  return 0;
+}
+static int zmq_reset_stats(openair0_device_t *device)
+{
+  return 0;
+}
+static void zmq_end(openair0_device_t *device)
+{
+  zmq_state_t *s = static_cast<zmq_state_t *>(device->priv);
+  if (s) {
+    if (s->poll_thread_running) {
+      s->poll_thread_running = false;
+      if (s->poll_thread.joinable()) {
+        s->poll_thread.join();
+      }
+    }
+    for (auto &chan : s->tx_stream.channels_) {
+      if (chan->socket_)
+        zmq_close(chan->socket_);
+      delete chan;
+    }
+    s->tx_stream.channels_.clear();
+
+    for (auto &chan : s->rx_stream.channels_) {
+      if (chan->socket_)
+        zmq_close(chan->socket_);
+      delete chan;
+    }
+    s->rx_stream.channels_.clear();
+
+    if (s->context)
+      zmq_ctx_destroy(s->context);
+    delete s;
+  }
+}
+
+static int zmq_start(openair0_device_t *device)
+{
+  zmq_state_t *s = static_cast<zmq_state_t *>(device->priv);
+  s->rx_stream.start(s->sample_rate / 100);
+  s->tx_stream.start(s->sample_rate / 100);
+  for (size_t i = 0; i < s->rx_stream.channels_.size(); i++) {
+    auto channel = s->rx_stream.channels_[i];
+    // Send initial request to start data flow
+    char dummy = 0;
+    if (zmq_send(channel->socket_, &dummy, 1, 0) != 1) {
+      LOG_E(HW, "[ZMQ] zmq_send for initial RX request failed for antenna %lu: %s\n", i, zmq_strerror(errno));
+      return -1;
+    }
+  }
+  s->poll_thread = std::thread(poll_thread, s);
+  return 0;
+}
+
+static int zmq_stop(openair0_device_t *device)
+{
+  zmq_state_t *s = static_cast<zmq_state_t *>(device->priv);
+  s->rx_stream.stop();
+  return 0;
+}
+
+static int zmq_set_freq(openair0_device_t *device, openair0_config_t *openair0_cfg)
+{
+  return 0;
+}
+static int zmq_set_gains(openair0_device_t *device, openair0_config_t *openair0_cfg)
+{
+  return 0;
+}
+static int zmq_write_init(openair0_device_t *device)
+{
+  return 0;
+}
+
+extern "C" __attribute__((__visibility__("default"))) int device_init(openair0_device_t *device, openair0_config_t *openair0_cfg)
+{
+  auto *zmq_state = new zmq_state_t();
+  zmq_state->context = zmq_ctx_new();
+  AssertFatal(zmq_state->context != NULL, "zmq_ctx_new failed");
+
+  LOG_I(HW, "[ZMQ] tx_antennas: %d, rx_antennas: %d\n", openair0_cfg->tx_num_channels, openair0_cfg->rx_num_channels);
+  configmodule_interface_t *cfg = config_get_if();
+  paramdef_t param_desc[] = ZMQ_PARAMS_DESC;
+  std::string zmq_section = std::string(ZMQ_SECTION);
+  int ru_id = openair0_cfg->ru_id;
+  std::string zmq_array_section = std::string(ZMQ_SECTION) + ".[" + std::to_string(ru_id) + "]";
+  int ret = config_get(cfg, param_desc, sizeofArray(param_desc), zmq_array_section.c_str());
+  AssertFatal(ret >= 0, "configuration couldn't be performed\n");
+  int num_configured_tx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_TX_CHANNELS)->numelt;
+  AssertFatal(num_configured_tx_channels == openair0_cfg->tx_num_channels,
+              "Incorrect configuration: Number of zmq tx channels (%d) != number of configured tx channels (%d)\n",
+              num_configured_tx_channels,
+              openair0_cfg->tx_num_channels);
+  int num_configured_rx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_RX_CHANNELS)->numelt;
+  AssertFatal(num_configured_rx_channels == openair0_cfg->rx_num_channels,
+              "Incorrect configuration: Number of zmq rx channels (%d) != number of configured rx channels (%d)\n",
+              num_configured_rx_channels,
+              openair0_cfg->rx_num_channels);
+  char **tx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_TX_CHANNELS)->strlistptr;
+  char **rx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_RX_CHANNELS)->strlistptr;
+
+  // Setup TX sockets (one per antenna)
+  if (openair0_cfg->tx_num_channels > 0) {
+    zmq_state->tx_stream.channels_.resize(openair0_cfg->tx_num_channels);
+    for (int i = 0; i < openair0_cfg->tx_num_channels; i++) {
+      void *socket = zmq_socket(zmq_state->context, ZMQ_REP);
+      AssertFatal(socket != NULL, "zmq_socket(ZMQ_REP) for TX antenna %d failed", i);
+      int linger = 0;
+      zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger));
+      AssertFatal(zmq_bind(socket, tx_channels[i]) == 0, "zmq_bind for TX antenna %d failed on %s", i, tx_channels[i]);
+      auto channel = new zmq_tx_channel(socket, openair0_cfg->sample_rate);
+      LOG_I(HW, "[ZMQ] TX socket for antenna %d bound to %s\n", i, tx_channels[i]);
+      zmq_state->tx_stream.channels_[i] = channel;
+    }
+  }
+  zmq_state->sample_rate = openair0_cfg->sample_rate;
+
+  // Setup RX sockets (one per antenna)
+  if (openair0_cfg->rx_num_channels > 0) {
+    zmq_state->rx_stream.channels_.resize(openair0_cfg->rx_num_channels);
+    for (int i = 0; i < openair0_cfg->rx_num_channels; i++) {
+      void *socket = zmq_socket(zmq_state->context, ZMQ_REQ);
+      AssertFatal(socket != NULL, "zmq_socket(ZMQ_REQ) for RX antenna %d failed", i);
+      int linger = 0;
+      zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger));
+      AssertFatal(zmq_connect(socket, rx_channels[i]) == 0, "zmq_connect for RX antenna %d failed on %s", i, rx_channels[i]);
+      auto channel = new zmq_rx_channel(socket, openair0_cfg->sample_rate);
+      LOG_I(HW, "[ZMQ] RX socket for antenna %d connected to %s\n", i, rx_channels[i]);
+      zmq_state->rx_stream.channels_[i] = channel;
+    }
+    zmq_state->rx_stream.tx_stream_ = &zmq_state->tx_stream;
+  }
+
+  device->trx_start_func = zmq_start;
+  device->trx_get_stats_func = zmq_get_stats;
+  device->trx_reset_stats_func = zmq_reset_stats;
+  device->trx_end_func = zmq_end;
+  device->trx_stop_func = zmq_stop;
+  device->trx_set_freq_func = zmq_set_freq;
+  device->trx_set_gains_func = zmq_set_gains;
+  device->trx_write_func = zmq_write;
+  device->trx_read_func = zmq_read;
+  device->type = RFSIMULATOR;
+  IS_SOFTMODEM_RFSIM = 1U;
+  openair0_cfg->rx_gain[0] = 0;
+  device->openair0_cfg = openair0_cfg;
+  device->priv = zmq_state;
+  device->trx_write_init = zmq_write_init;
+
+  return 0;
+}