Merge remote-tracking branch 'bpodrygajlo/zmq-radio-speedup' into integration_2026_w28

fix(zmq): speedup zmq radio (#149)

- Utilize ZMQ "zero-copy" inteface on TX
- Skip one memcpy in RX
- Use AVX-512 optimized cf_t <-> c16_t conversion loops in tx and rx
- Parallelize TX and RX polling threads for faster execution

Closes: #118
Reviewed-by: Robert Schmidt <robert.schmidt@openairinterface.org>

radio/zmq/ring_buffer.cpp

@@ -7,12 +7,14 @@
 #include <iostream>
 #include <algorithm>
 
-ring_buffer::ring_buffer(size_t max_size) : max_size_(max_size)
+template <typename T>
+ring_buffer<T>::ring_buffer(size_t max_size) : max_size_(max_size)
 {
-  buffer_ = std::make_unique<cf_t[]>(max_size);
+  buffer_ = std::make_unique<T[]>(max_size);
 }
 
-size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
+template <typename T>
+size_t ring_buffer<T>::push_samples(const T *samples, const size_t nsamps)
 {
   size_t overflow = 0;
   // if nsamps > max_size skip nsamps - max_size samples
@@ -31,12 +33,12 @@ size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
   }
 
   size_t first_chunk = std::min(nsamps_left, max_size_ - head_);
-  memcpy(&buffer_[head_], samples, first_chunk * sizeof(cf_t));
+  memcpy(&buffer_[head_], samples, first_chunk * sizeof(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));
+    memcpy(&buffer_[0], samples, nsamps_left * sizeof(T));
     head_ = nsamps_left;
   }
 
@@ -45,7 +47,8 @@ size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
   return overflow;
 }
 
-size_t ring_buffer::push_zeros(const size_t num_zeros)
+template <typename T>
+size_t ring_buffer<T>::push_zeros(const size_t num_zeros)
 {
   size_t overflow = 0;
   // if nsamps > max_size skip nsamps - max_size samples
@@ -63,11 +66,11 @@ size_t ring_buffer::push_zeros(const size_t num_zeros)
   }
 
   size_t first_chunk = std::min(nsamps_left, max_size_ - head_);
-  memset(&buffer_[head_], 0, first_chunk * sizeof(cf_t));
+  memset(&buffer_[head_], 0, first_chunk * sizeof(T));
   head_ = (head_ + first_chunk) % max_size_;
   nsamps_left -= first_chunk;
   if (nsamps_left > 0) {
-    memset(&buffer_[0], 0, nsamps_left * sizeof(cf_t));
+    memset(&buffer_[0], 0, nsamps_left * sizeof(T));
     head_ = nsamps_left;
   }
 
@@ -76,16 +79,17 @@ size_t ring_buffer::push_zeros(const size_t num_zeros)
   return overflow;
 }
 
-size_t ring_buffer::pop_samples(cf_t *samples, size_t num_samples)
+template <typename T>
+size_t ring_buffer<T>::pop_samples(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, &buffer_[tail_], first_chunk * sizeof(T));
-      memcpy(samples + first_chunk, &buffer_[0], (samples_to_pop - first_chunk) * sizeof(cf_t));
+      memcpy(samples + first_chunk, &buffer_[0], (samples_to_pop - first_chunk) * sizeof(T));
     } else {
-      memcpy(samples, &buffer_[tail_], samples_to_pop * sizeof(cf_t));
+      memcpy(samples, &buffer_[tail_], samples_to_pop * sizeof(T));
     }
     tail_ = (tail_ + samples_to_pop) % max_size_;
     size_ -= samples_to_pop;
@@ -94,24 +98,28 @@ size_t ring_buffer::pop_samples(cf_t *samples, size_t num_samples)
   return 0;
 }
 
-void ring_buffer::clear_samples()
+template <typename T>
+void ring_buffer<T>::clear_samples()
 {
   head_ = 0;
   tail_ = 0;
   size_ = 0;
 }
 
-void ring_buffer::reset()
+template <typename T>
+void ring_buffer<T>::reset()
 {
   clear_samples();
 }
 
-size_t ring_buffer::size() const
+template <typename T>
+size_t ring_buffer<T>::size() const
 {
   return size_;
 }
 
-size_t overflow_buffer::push_samples(const cf_t *samples, size_t nsamps)
+template <typename T>
+size_t overflow_buffer<T>::push_samples(const T *samples, size_t nsamps)
 {
   std::lock_guard<std::mutex> lock(mutex_);
   size_t overflow = buffer_.push_samples(samples, nsamps);
@@ -119,7 +127,8 @@ size_t overflow_buffer::push_samples(const cf_t *samples, size_t nsamps)
   return overflow;
 }
 
-size_t overflow_buffer::push_zeros(size_t num_zeros)
+template <typename T>
+size_t overflow_buffer<T>::push_zeros(size_t num_zeros)
 {
   std::lock_guard<std::mutex> lock(mutex_);
   size_t overflow = buffer_.push_zeros(num_zeros);
@@ -127,13 +136,14 @@ size_t overflow_buffer::push_zeros(size_t num_zeros)
   return overflow;
 }
 
-size_t overflow_buffer::pop_samples(cf_t *samples, size_t num_samples)
+template <typename T>
+size_t overflow_buffer<T>::pop_samples(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));
+    memset(samples, 0, num_zeros * sizeof(T));
     zeros_to_send_ -= num_zeros;
     samples += num_zeros;
     num_samples -= num_zeros;
@@ -146,22 +156,30 @@ size_t overflow_buffer::pop_samples(cf_t *samples, size_t num_samples)
   return samples_popped;
 }
 
-void overflow_buffer::reset()
+template <typename T>
+void overflow_buffer<T>::reset()
 {
   std::lock_guard<std::mutex> lock(mutex_);
   buffer_.reset();
   zeros_to_send_ = buffer_.size() / 2;
 }
 
-void overflow_buffer::clear_samples()
+template <typename T>
+void overflow_buffer<T>::clear_samples()
 {
   std::lock_guard<std::mutex> lock(mutex_);
   buffer_.clear_samples();
   zeros_to_send_ = 0;
 }
 
-size_t overflow_buffer::size()
+template <typename T>
+size_t overflow_buffer<T>::size()
 {
   std::lock_guard<std::mutex> lock(mutex_);
   return buffer_.size() + zeros_to_send_;
 }
+
+template class ring_buffer<cf_t>;
+template class ring_buffer<c16_t>;
+template class overflow_buffer<cf_t>;
+template class overflow_buffer<c16_t>;

radio/zmq/ring_buffer.h

@@ -8,10 +8,11 @@
 #include <memory>
 #include "common/platform_types.h"
 
-// A basic cirular sample buffer class
+// A basic circular sample buffer class
+template <typename T = cf_t>
 class ring_buffer {
  private:
-  std::unique_ptr<cf_t[]> buffer_;
+  std::unique_ptr<T[]> buffer_;
   size_t head_ = 0;
   size_t tail_ = 0;
   size_t size_ = 0;
@@ -19,17 +20,18 @@ class ring_buffer {
 
  public:
   ring_buffer(size_t max_size = 614400);
-  size_t push_samples(const cf_t *samples, size_t nsamps);
+  size_t push_samples(const T *samples, size_t nsamps);
   size_t push_zeros(size_t num_zeros);
-  size_t pop_samples(cf_t *samples, size_t num_samples);
+  size_t pop_samples(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
+template <typename T = cf_t>
 class overflow_buffer {
-  ring_buffer buffer_;
+  ring_buffer<T> buffer_;
   size_t zeros_to_send_;
   std::mutex mutex_;
 
@@ -37,9 +39,9 @@ class overflow_buffer {
   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_samples(const T *samples, size_t nsamps);
   size_t push_zeros(size_t num_zeros);
-  size_t pop_samples(cf_t *samples, size_t num_samples);
+  size_t pop_samples(T *samples, size_t num_samples);
   void reset();
   void clear_samples();
   size_t size();

radio/zmq/tests/test_zmq_radio.cpp

@@ -153,6 +153,81 @@ TEST_F(ZMQTest, TxRxSamples)
   t2.join();
 }
 
+TEST_F(ZMQTest, TxRxSamplesSIMD)
+{
+  const int size = 100;
+  std::thread t1([this, size]() {
+    c16_t rx_samples[size];
+    openair0_timestamp_t rx_timestamp;
+    void *samples[1] = {rx_samples};
+    ASSERT_EQ(device1.trx_read_func(&device1, &rx_timestamp, samples, size, 1), size);
+    for (int i = 0; i < size; i++) {
+      ASSERT_EQ(rx_samples[i].r, 0);
+      ASSERT_EQ(rx_samples[i].i, 0);
+    }
+    c16_t tx_samples[size];
+    openair0_timestamp_t tx_timestamp = rx_timestamp + size;
+    for (int i = 0; i < size; i++) {
+      tx_samples[i].r = (int16_t)i;
+      tx_samples[i].i = (int16_t)(i + 1);
+    }
+    samples[0] = tx_samples;
+    ASSERT_EQ(device1.trx_write_func(&device1, tx_timestamp, samples, size, 1, 0), size);
+  });
+  std::thread t2([this, size]() {
+    c16_t rx_samples[size];
+    openair0_timestamp_t rx_timestamp;
+    void *samples[1] = {rx_samples};
+    ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp, samples, size, 1), size);
+    for (int i = 0; i < size; 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, size, 1), size);
+    for (int i = 0; i < size; i++) {
+      ASSERT_EQ(rx_samples[i].r, (int16_t)i);
+      ASSERT_EQ(rx_samples[i].i, (int16_t)(i + 1));
+    }
+    ASSERT_EQ(rx_timestamp + size, rx_timestamp2);
+  });
+  t1.join();
+  t2.join();
+}
+
+TEST_F(ZMQTest, BenchmarkThroughput)
+{
+  int level = g_log->log_component[HW].level;
+  g_log->log_component[HW].level = OAILOG_ERR;
+  const size_t nsamps = 10000;
+  const size_t num_iters = 100000;
+  c16_t tx_samples[nsamps];
+  for (size_t i = 0; i < nsamps; i++) {
+    tx_samples[i].r = i;
+    tx_samples[i].i = i + 1;
+  }
+
+  void *samples[1] = {tx_samples};
+  openair0_timestamp_t tx_timestamp = 0;
+
+  auto start = std::chrono::high_resolution_clock::now();
+
+  for (size_t i = 0; i < num_iters; i++) {
+    ASSERT_EQ(device1.trx_write_func(&device1, tx_timestamp, samples, nsamps, 1, 0), nsamps);
+    tx_timestamp += nsamps;
+  }
+
+  auto end = std::chrono::high_resolution_clock::now();
+  std::chrono::duration<double> diff = end - start;
+
+  std::cout << "Benchmark:" << std::endl;
+  std::cout << "Time taken: " << diff.count() << " s\n";
+  std::cout << "Throughput: " << (nsamps * num_iters) / diff.count() / 1e6 << " MSamples/s\n";
+
+  // No need to drain messages as device shutdown handles cleanup
+  g_log->log_component[HW].level = level;
+}
+
 int main(int argc, char **argv)
 {
   logInit();

radio/zmq/zmq_imported.cpp

@@ -7,32 +7,57 @@
 #include "zmq_imported.h"
 #include "log.h"
 
-const float c16_t_to_cf_t_factor = std::numeric_limits<int16_t>::max();
+#include "zmq_simd.h"
+
 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;
+
+  size_t zeros_to_push = 0;
   if (timestamp > sample_count_) {
-    overflow += buffer_.push_zeros(timestamp - sample_count_);
+    zeros_to_push = timestamp - sample_count_;
     sample_count_ = timestamp;
   }
-  cf_t samples_float[nsamps];
+
-  for (size_t i = 0; i < nsamps; i++) {
+  size_t total_samples = zeros_to_push + nsamps;
-    samples_float[i].r = samples[i].r / c16_t_to_cf_t_factor;
+  if (total_samples > 0) {
-    samples_float[i].i = samples[i].i / c16_t_to_cf_t_factor;
+    zmq_msg_t msg;
+    zmq_msg_init_size(&msg, total_samples * sizeof(cf_t));
+    cf_t *msg_data = static_cast<cf_t *>(zmq_msg_data(&msg));
+
+    if (zeros_to_push > 0) {
+      memset(msg_data, 0, zeros_to_push * sizeof(cf_t));
+    }
+
+    convert_samples_avx512_tx(reinterpret_cast<float *>(&msg_data[zeros_to_push]),
+                              reinterpret_cast<const int16_t *>(samples),
+                              nsamps * 2,
+                              c16_t_to_cf_t_factor);
+
+    std::lock_guard<std::mutex> q_lock(queue_mutex_);
+    queue_.push(std::move(msg));
   }
-  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();
 }
 
+bool zmq_tx_channel::pop_message(zmq_msg_t *msg)
+{
+  std::lock_guard<std::mutex> lock(queue_mutex_);
+  if (queue_.empty()) {
+    return false;
+  }
+  *msg = std::move(queue_.front());
+  queue_.pop();
+  return true;
+}
+
 void zmq_tx_channel::start(uint64_t init_time)
 {
   sample_count_ = init_time;
@@ -54,7 +79,14 @@ bool zmq_tx_channel::align(uint64_t timestamp, std::chrono::milliseconds timeout
     is_tx_enabled_ = false;
   }
   if (sample_count_ < timestamp) {
-    buffer_.push_zeros(timestamp - sample_count_);
+    size_t zeros_to_push = timestamp - sample_count_;
+    zmq_msg_t msg;
+    zmq_msg_init_size(&msg, zeros_to_push * sizeof(cf_t));
+    memset(zmq_msg_data(&msg), 0, zeros_to_push * sizeof(cf_t));
+    {
+      std::lock_guard<std::mutex> q_lock(queue_mutex_);
+      queue_.push(std::move(msg));
+    }
     sample_count_ = timestamp;
   }
   return false;
@@ -63,18 +95,13 @@ bool zmq_tx_channel::align(uint64_t timestamp, std::chrono::milliseconds timeout
 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);
+    size_t popped_now = buffer_.pop_samples(samples + 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()
 {

radio/zmq/zmq_imported.h

@@ -13,21 +13,33 @@
 #include <atomic>
 #include <mutex>
 #include <vector>
+#include <queue>
 
 class zmq_tx_channel {
  public:
   void *socket_;
-  overflow_buffer buffer_;
+  std::queue<zmq_msg_t> queue_;
+  std::mutex queue_mutex_;
   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)
+  zmq_tx_channel(void *s, uint64_t buffer_size) : socket_(s)
   {
   }
 
+  ~zmq_tx_channel()
+  {
+    std::lock_guard<std::mutex> lock(queue_mutex_);
+    while (!queue_.empty()) {
+      zmq_msg_close(&queue_.front());
+      queue_.pop();
+    }
+  }
+
   void transmit(c16_t *samples, size_t nsamps, uint64_t timestamp);
+  bool pop_message(zmq_msg_t *msg);
 
   void start(uint64_t init_time);
 
@@ -36,11 +48,11 @@ class zmq_tx_channel {
 
 class zmq_rx_channel {
  public:
-  void *socket_;
+   void *socket_;
-  overflow_buffer buffer_;
+   overflow_buffer<c16_t> buffer_;
-  bool request_sent_;
+   bool request_sent_;
-  std::atomic<bool> stopped_;
+   std::atomic<bool> stopped_;
-  zmq_rx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size), stopped_(false)
+   zmq_rx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size), stopped_(false)
   {
   }
   void receive(c16_t *samples, size_t nsamps);

radio/zmq/zmq_radio.cpp

@@ -40,13 +40,14 @@
 #include <ring_buffer.h>
 #include <zmq.h>
 #include "zmq_imported.h"
+#include "zmq_simd.h"
 
 #define ZMQ_SECTION "zmq"
 #define ZMQ_TX_CHANNELS "tx_channels"
 #define ZMQ_RX_CHANNELS "rx_channels"
 
-#define ZMQ_PARAMS_DESC                                                                                                           \
+#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), \
   };
@@ -58,110 +59,110 @@ struct zmq_state_t {
   void *context;
   zmq_tx_stream tx_stream;
   zmq_rx_stream rx_stream;
-  std::thread poll_thread;
+  std::vector<std::thread> tx_poll_threads;
+  std::vector<std::thread> rx_poll_threads;
   std::atomic<bool> poll_thread_running;
   bool stopped = false;
   double sample_rate;
 };
 
-static void poll_thread(zmq_state_t *s)
+static void tx_poll_thread(zmq_tx_channel *chan, size_t i, std::atomic<bool> *poll_thread_running)
 {
-  s->poll_thread_running = true;
+  zmq_pollitem_t item = {chan->socket_, 0, ZMQ_POLLIN, 0};
-  unsigned char *rx_buffer = static_cast<unsigned char *>(malloc(rx_buffer_size));
+  bool reply_requested = false;
-  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 (*poll_thread_running) {
-  while (s->poll_thread_running) {
+    if (reply_requested) {
-    for (size_t i = 0; i < num_tx_channels; i++) {
+      zmq_msg_t msg;
-      auto chan = s->tx_stream.channels_[i];
+      if (chan->pop_message(&msg)) {
-      if (!reply_requested[i]) {
+        int rc = zmq_msg_send(&msg, chan->socket_, 0);
-        continue;
+        if (rc < 0) {
+          LOG_E(HW, "[ZMQ] tx_poll_thread zmq_msg_send for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+        }
+        zmq_msg_close(&msg);
+        reply_requested = false;
       }
-      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
+    int rc = zmq_poll(&item, 1, 10); // 10ms timeout
     if (rc < 0) {
       if (errno == EINTR)
         continue;
-      LOG_E(HW, "[ZMQ] poll_thread zmq_poll failed: %s\n", zmq_strerror(errno));
+      LOG_E(HW, "[ZMQ] tx_poll_thread zmq_poll failed for TX antenna %d: %s\n", (int)i, zmq_strerror(errno));
       break;
     }
     if (rc == 0) {
       continue; // timeout
     }
 
-    // --- TX Sockets (ZMQ_REP) ---
+    if (item.revents & ZMQ_POLLIN) {
-    for (size_t i = 0; i < num_tx_channels; i++) {
+      char dummy;
-      if (items[i].revents & ZMQ_POLLIN) {
+      rc = zmq_recv(chan->socket_, &dummy, 1, 0);
-        auto chan = s->tx_stream.channels_[i];
+      if (rc < 0) {
-        char dummy;
+        LOG_E(HW, "[ZMQ] tx_poll_thread zmq_recv for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
-        rc = zmq_recv(chan->socket_, &dummy, 1, 0);
+        continue;
-        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;
       }
+      if (reply_requested) {
+        LOG_E(HW, "[ZMQ] Error, unexpected REQ before REP on TX antenna %d\n", (int)i);
+      }
+      reply_requested = true;
+    }
+  }
+}
+
+static void rx_poll_thread(zmq_rx_channel *chan, size_t i, std::atomic<bool> *poll_thread_running)
+{
+  unsigned char *rx_buffer = static_cast<unsigned char *>(malloc(rx_buffer_size));
+  c16_t *rx_buffer_c16 = static_cast<c16_t *>(malloc(rx_buffer_size / sizeof(cf_t) * sizeof(c16_t)));
+  zmq_pollitem_t item = {chan->socket_, 0, ZMQ_POLLIN, 0};
+
+  while (*poll_thread_running) {
+    int rc = zmq_poll(&item, 1, 10); // 10ms timeout
+    if (rc < 0) {
+      if (errno == EINTR)
+        continue;
+      LOG_E(HW, "[ZMQ] rx_poll_thread zmq_poll failed for RX antenna %d: %s\n", (int)i, zmq_strerror(errno));
+      break;
+    }
+    if (rc == 0) {
+      continue; // timeout
     }
 
-    // --- RX Sockets (ZMQ_REQ) ---
+    if (item.revents & ZMQ_POLLIN) {
-    for (size_t i = 0; i < num_rx_channels; i++) {
+      rc = zmq_recv(chan->socket_, rx_buffer, rx_buffer_size, 0);
-      if (items[i + num_tx_channels].revents & ZMQ_POLLIN) {
+      if (rc < 0) {
-        auto chan = s->rx_stream.channels_[i];
+        LOG_E(HW, "[ZMQ] rx_poll_thread zmq_recv for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
-        rc = zmq_recv(chan->socket_, rx_buffer, rx_buffer_size, 0);
+      } else {
-        if (rc < 0) {
+        size_t received_bytes = rc;
-          LOG_E(HW, "[ZMQ] poll_thread zmq_recv for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+        if (rx_buffer_size < received_bytes) {
-        } else {
+          LOG_W(HW,
-          size_t received_bytes = rc;
+                "[ZMQ] the RX buffer is too small! The received message size is %lu while the buffer is %lu. Message truncated\n",
-          if (rx_buffer_size < received_bytes) {
+                received_bytes,
-            LOG_W(HW,
+                rx_buffer_size);
-                  "[ZMQ] the RX buffer is too small! The received message size is %lu while the buffer is %lu. Message truncated\n",
+        }
-                  received_bytes,
+        size_t num_samples_received = std::min(received_bytes, rx_buffer_size) / sizeof(cf_t);
-                  rx_buffer_size);
+        cf_t *samples = reinterpret_cast<cf_t *>(rx_buffer);
-          }
+        convert_samples_avx512_rx(reinterpret_cast<const float *>(samples),
-          size_t num_samples_received = std::min(received_bytes, rx_buffer_size) / sizeof(cf_t);
+                                  reinterpret_cast<int16_t *>(rx_buffer_c16),
-          cf_t *samples = reinterpret_cast<cf_t *>(rx_buffer);
+                                  num_samples_received * 2,
-          size_t overflow = chan->buffer_.push_samples(samples, num_samples_received);
+                                  c16_t_to_cf_t_factor);
-          if (rx_buffer_size < received_bytes) {
+        size_t overflow = chan->buffer_.push_samples(rx_buffer_c16, num_samples_received);
-            overflow += chan->buffer_.push_zeros((received_bytes - rx_buffer_size) / sizeof(cf_t));
+        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");
+        if (overflow) {
-          }
+          LOG_W(HW, "Overflow on receive\n");
-          // After receiving, send next request to keep the stream flowing
+        }
-          char dummy = 0;
+        // After receiving, send next request to keep the stream flowing
-          if (zmq_send(chan->socket_, &dummy, 1, 0) != 1) {
+        char dummy = 0;
-            LOG_E(HW, "[ZMQ] poll_thread zmq_send for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+        if (zmq_send(chan->socket_, &dummy, 1, 0) != 1) {
-          }
+          LOG_E(HW, "[ZMQ] rx_poll_thread zmq_send for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
         }
       }
     }
   }
   free(rx_buffer);
+  free(rx_buffer_c16);
 }
 
 static int zmq_write(openair0_device_t *device, openair0_timestamp_t timestamp, void **buff, int nsamps, int cc, int flags)
@@ -204,8 +205,15 @@ static void zmq_end(openair0_device_t *device)
   if (s) {
     if (s->poll_thread_running) {
       s->poll_thread_running = false;
-      if (s->poll_thread.joinable()) {
+      for (auto &t : s->tx_poll_threads) {
-        s->poll_thread.join();
+        if (t.joinable()) {
+          t.join();
+        }
+      }
+      for (auto &t : s->rx_poll_threads) {
+        if (t.joinable()) {
+          t.join();
+        }
       }
     }
     for (auto &chan : s->tx_stream.channels_) {
@@ -242,7 +250,13 @@ static int zmq_start(openair0_device_t *device)
       return -1;
     }
   }
-  s->poll_thread = std::thread(poll_thread, s);
+  s->poll_thread_running = true;
+  for (size_t i = 0; i < s->tx_stream.channels_.size(); ++i) {
+    s->tx_poll_threads.push_back(std::thread(tx_poll_thread, s->tx_stream.channels_[i], i, &s->poll_thread_running));
+  }
+  for (size_t i = 0; i < s->rx_stream.channels_.size(); ++i) {
+    s->rx_poll_threads.push_back(std::thread(rx_poll_thread, s->rx_stream.channels_[i], i, &s->poll_thread_running));
+  }
   return 0;
 }
 

radio/zmq/zmq_simd.h

@@ -0,0 +1,103 @@
+/*
+ * SPDX-License-Identifier: LicenseRef-CSSL-1.0
+ */
+
+#ifndef ZMQ_SIMD_H
+#define ZMQ_SIMD_H
+
+#include "simde/x86/avx512.h"
+#include "simde/x86/avx2.h"
+
+#include "common/platform_types.h"
+#include <limits>
+#include <algorithm>
+
+constexpr float c16_t_to_cf_t_factor = 32767.0f;
+
+static inline void convert_samples_avx512_tx(float *msg_data, const int16_t *samples, size_t nsamps, float factor)
+{
+  float r_factor = 1.0f / factor;
+  size_t total_elements = nsamps;
+  size_t i = 0;
+  float *output_ptr = msg_data;
+
+#if defined(__AVX512F__)
+  {
+    simde__m512 v_factor = simde_mm512_set1_ps(r_factor);
+    for (; i + 16 <= total_elements; i += 16) {
+      simde__m256i v_in16 = simde_mm256_loadu_si256((const simde__m256i *)&samples[i]);
+      simde__m512i v_in32 = simde_mm512_cvtepi16_epi32(v_in16);
+      simde__m512 v_float = simde_mm512_cvtepi32_ps(v_in32);
+      simde__m512 v_out = simde_mm512_mul_ps(v_float, v_factor);
+      simde_mm512_storeu_ps(&output_ptr[i], v_out);
+    }
+  }
+#endif
+
+#if defined(__AVX2__)
+  {
+    simde__m256 v_factor = simde_mm256_set1_ps(r_factor);
+    for (; i + 16 <= total_elements; i += 16) {
+      simde__m256i v_in16 = simde_mm256_loadu_si256((const simde__m256i *)&samples[i]);
+      simde__m128i v_in16_lo = simde_mm256_castsi256_si128(v_in16);
+      simde__m128i v_in16_hi = simde_mm256_extractf128_si256(v_in16, 1);
+      simde__m256i v_in32_lo = simde_mm256_cvtepi16_epi32(v_in16_lo);
+      simde__m256i v_in32_hi = simde_mm256_cvtepi16_epi32(v_in16_hi);
+      simde__m256 v_float_lo = simde_mm256_cvtepi32_ps(v_in32_lo);
+      simde__m256 v_float_hi = simde_mm256_cvtepi32_ps(v_in32_hi);
+      simde__m256 v_out_lo = simde_mm256_mul_ps(v_float_lo, v_factor);
+      simde__m256 v_out_hi = simde_mm256_mul_ps(v_float_hi, v_factor);
+      simde_mm256_storeu_ps(&output_ptr[i], v_out_lo);
+      simde_mm256_storeu_ps(&output_ptr[i + 8], v_out_hi);
+    }
+  }
+#endif
+
+  // Cleanup loop for remaining elements
+  for (; i < total_elements; i++) {
+    output_ptr[i] = (float)samples[i] * r_factor;
+  }
+}
+
+static inline void convert_samples_avx512_rx(const float *input_floats, int16_t *output_ints, size_t nsamps, float factor)
+{
+  size_t total_elements = nsamps;
+  size_t i = 0;
+
+#if defined(__AVX512F__)
+  {
+    simde__m512 v_factor = simde_mm512_set1_ps(factor);
+    for (; i + 16 <= total_elements; i += 16) {
+      simde__m512 v_float = simde_mm512_loadu_ps(&input_floats[i]);
+      simde__m512 v_mul = simde_mm512_mul_ps(v_float, v_factor);
+      simde__m512i v_int32 = simde_mm512_cvtps_epi32(v_mul);
+      simde__m256i v_int16 = simde_mm512_cvtsepi32_epi16(v_int32);
+      simde_mm256_storeu_si256((simde__m256i *)&output_ints[i], v_int16);
+    }
+  }
+#endif
+
+#if defined(__AVX2__)
+  {
+    simde__m256 v_factor = simde_mm256_set1_ps(factor);
+    for (; i + 16 <= total_elements; i += 16) {
+      simde__m256 v_float1 = simde_mm256_loadu_ps(&input_floats[i]);
+      simde__m256 v_float2 = simde_mm256_loadu_ps(&input_floats[i + 8]);
+      simde__m256 v_mul1 = simde_mm256_mul_ps(v_float1, v_factor);
+      simde__m256 v_mul2 = simde_mm256_mul_ps(v_float2, v_factor);
+      simde__m256i v_int32_1 = simde_mm256_cvtps_epi32(v_mul1);
+      simde__m256i v_int32_2 = simde_mm256_cvtps_epi32(v_mul2);
+      simde__m256i v_packed = simde_mm256_packs_epi32(v_int32_1, v_int32_2);
+      simde__m256i v_permuted = simde_mm256_permute4x64_epi64(v_packed, 0xD8);
+      simde_mm256_storeu_si256((simde__m256i *)&output_ints[i], v_permuted);
+    }
+  }
+#endif
+
+  // Cleanup loop for remaining elements
+  for (; i < total_elements; i++) {
+    output_ints[i] = (int16_t)(input_floats[i] * factor + 0.5f);
+  }
+}
+
+#endif