perf(zmq): optimize TX path using ZMQ zero-copy message queue

Optimize the TX path by converting int16 complex samples to float
complex format directly into ZMQ message buffers using SIMD. Queue
these ZMQ messages (`zmq_msg_t`) directly, avoiding temporary internal
copying and enabling ZMQ's zero-copy interface during actual transmit
in the polling thread.

Signed-off-by: Bartosz Podrygajlo <bartosz.podrygajlo@openairinterface.org>

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;

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<cf_t> 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);
 

radio/zmq/zmq_radio.cpp

@@ -90,16 +90,15 @@ static void poll_thread(zmq_state_t *s)
       if (!reply_requested[i]) {
         continue;
       }
-      std::vector<cf_t> samples(1024);
+      zmq_msg_t msg;
-      size_t num_popped = chan->buffer_.pop_samples(samples.data(), 1024);
+      if (chan->pop_message(&msg)) {
-      if (num_popped == 0) {
+        int rc = zmq_msg_send(&msg, chan->socket_, 0);
-        continue;
+        if (rc < 0) {
+          LOG_E(HW, "[ZMQ] poll_thread zmq_msg_send for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno));
+        }
+        zmq_msg_close(&msg);
+        reply_requested[i] = false;
       }
-      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