vrtsim: Channel modelling on reception

Removes static variables and return handle to differentiate
multiple taps_clients per process.

common/utils/shm_iq_channel/shm_td_iq_channel.c

@@ -20,8 +20,10 @@
  */
 
 #include "shm_td_iq_channel.h"
+#include <stdint.h>
 #include <stdio.h>
 #include <sys/mman.h>
+#include <sys/types.h>
 #include <unistd.h>
 #include <fcntl.h>
 #include <sys/stat.h>
@@ -34,14 +36,24 @@
 #include "common/utils/threadPool/pthread_utils.h"
 
 #define CIRCULAR_BUFFER_SIZE (30720 * 14 * 20)
+// Buffer prefix is a copy of the ending of the buffer to the beginning to
+// allow continuous read up to this size without wrapping when doing channel modelling
+#define BUFFER_PREFIX_SIZE (30720 * 10)
+
+typedef struct {
+  uint64_t timestamp;
+  pthread_mutex_t mutex;
+  pthread_cond_t cond;
+} sync_data_t;
 
 typedef struct {
   int magic;
   int num_antennas_tx;
   int num_antennas_rx;
-  uint64_t timestamp;
-  pthread_mutex_t mutex;
-  pthread_cond_t cond;
+  sync_data_t sync_data;
+
+  bool client_sync;
+  sync_data_t client_sync_data;
 } ShmTDIQChannelData;
 
 typedef struct ShmTDIQChannel_s {
@@ -50,16 +62,89 @@ typedef struct ShmTDIQChannel_s {
   char name[256];
   sample_t *tx_iq_data;
   sample_t *rx_iq_data;
+  int num_antennas_tx;
+  int num_antennas_rx;
   bool abort;
 } ShmTDIQChannel;
 
-ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int num_rx_ant)
+static void init_sync_data(sync_data_t *sync_data)
 {
+  sync_data->timestamp = 0;
+  pthread_mutexattr_t mutex_attr;
+  pthread_condattr_t cond_attr;
+  int ret = pthread_mutexattr_init(&mutex_attr);
+  AssertFatal(ret == 0, "pthread_mutexattr_init() failed: errno %d, %s\n", errno, strerror(errno));
+
+  ret = pthread_condattr_init(&cond_attr);
+  AssertFatal(ret == 0, "pthread_condattr_init() failed: errno %d, %s\n", errno, strerror(errno));
+
+  ret = pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED);
+  AssertFatal(ret == 0, "pthread_mutexattr_setpshared() failed: errno %d, %s\n", errno, strerror(errno));
+
+  ret = pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED);
+  AssertFatal(ret == 0, "pthread_condattr_setpshared() failed: errno %d, %s\n", errno, strerror(errno));
+
+  ret = pthread_mutex_init(&sync_data->mutex, &mutex_attr);
+  AssertFatal(ret == 0, "pthread_mutex_init() failed: errno %d, %s\n", errno, strerror(errno));
+
+  ret = pthread_cond_init(&sync_data->cond, &cond_attr);
+  AssertFatal(ret == 0, "pthread_cond_init() failed: errno %d, %s\n", errno, strerror(errno));
+}
+
+static void wait_for_sync(sync_data_t *sync_data, uint64_t timestamp, bool *should_abort)
+{
+  mutexlock(sync_data->mutex);
+  while (sync_data->timestamp < timestamp && !*should_abort) {
+    condwait(sync_data->cond, sync_data->mutex);
+  }
+  mutexunlock(sync_data->mutex);
+}
+
+static int wait_for_sync_with_timeout(sync_data_t *sync_data, uint64_t timestamp, uint64_t timeout_uS, bool *should_abort)
+{
+  struct timespec ts = {.tv_sec = 0, .tv_nsec = 0};
+  if (clock_gettime(CLOCK_REALTIME, &ts) != 0) {
+    fprintf(stderr, "Error: clock_gettime failed: %s\n", strerror(errno));
+    return 1;
+  }
+
+  ts.tv_sec += timeout_uS / 1000000; // Convert microseconds to seconds
+  ts.tv_nsec += (timeout_uS % 1000000) * 1000; // Convert remaining microseconds to nanoseconds
+
+  mutexlock(sync_data->mutex);
+  while (sync_data->timestamp < timestamp && !*should_abort) {
+    int ret = pthread_cond_timedwait(&sync_data->cond, &sync_data->mutex, &ts);
+    if (ret == ETIMEDOUT) {
+      fprintf(stderr, "Error: Timed out waiting for samples.\n");
+      mutexunlock(sync_data->mutex);
+      return 1;
+    } else if (ret != 0) {
+      fprintf(stderr, "Error: pthread_cond_timedwait failed: %s\n", strerror(ret));
+      mutexunlock(sync_data->mutex);
+      return 1;
+    }
+  }
+  mutexunlock(sync_data->mutex);
+  return 0;
+}
+
+static void update_timestamp(sync_data_t *sync_data, uint64_t timestamp)
+{
+  mutexlock(sync_data->mutex);
+  sync_data->timestamp = timestamp;
+  condbroadcast(sync_data->cond);
+  mutexunlock(sync_data->mutex);
+}
+
+ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int num_rx_ant, bool client_sync)
+{
+  AssertFatal(num_tx_ant > 0, "Number of TX antennas must be greater than 0\n");
+  AssertFatal(num_rx_ant > 0, "Number of RX antennas must be greater than 0\n");
   // Create shared memory segment
   int fd = shm_open(name, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
   AssertFatal(fd != -1, "shm_open failed: %s\n", strerror(errno));
-  size_t tx_buffer_size = CIRCULAR_BUFFER_SIZE * sizeof(sample_t) * num_tx_ant;
-  size_t rx_buffer_size = CIRCULAR_BUFFER_SIZE * sizeof(sample_t) * num_rx_ant;
+  size_t tx_buffer_size = (CIRCULAR_BUFFER_SIZE + BUFFER_PREFIX_SIZE) * sizeof(sample_t) * num_tx_ant;
+  size_t rx_buffer_size = (CIRCULAR_BUFFER_SIZE + BUFFER_PREFIX_SIZE) * sizeof(sample_t) * num_rx_ant;
   size_t total_size = sizeof(ShmTDIQChannelData) + tx_buffer_size + rx_buffer_size;
 
   // Set the size of the shared memory segment
@@ -78,29 +163,19 @@ ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int n
   strncpy(channel->name, name, sizeof(channel->name) - 1);
   channel->tx_iq_data = (sample_t *)(shm_ptr + 1);
   channel->rx_iq_data = channel->tx_iq_data + tx_buffer_size / sizeof(sample_t);
+  channel->num_antennas_tx = num_tx_ant;
+  channel->num_antennas_rx = num_rx_ant;
   channel->data = shm_ptr;
   channel->type = IQ_CHANNEL_TYPE_SERVER;
-  pthread_mutexattr_t mutex_attr;
-  pthread_condattr_t cond_attr;
-  int ret = pthread_mutexattr_init(&mutex_attr);
-  AssertFatal(ret == 0, "pthread_mutexattr_init() failed: errno %d, %s\n", errno, strerror(errno));
 
-  ret = pthread_condattr_init(&cond_attr);
-  AssertFatal(ret == 0, "pthread_condattr_init() failed: errno %d, %s\n", errno, strerror(errno));
-
-  ret = pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED);
-  AssertFatal(ret == 0, "pthread_mutexattr_setpshared() failed: errno %d, %s\n", errno, strerror(errno));
-
-  ret = pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED);
-  AssertFatal(ret == 0, "pthread_condattr_setpshared() failed: errno %d, %s\n", errno, strerror(errno));
-
-  ret = pthread_mutex_init(&shm_ptr->mutex, &mutex_attr);
-  AssertFatal(ret == 0, "pthread_mutex_init() failed: errno %d, %s\n", errno, strerror(errno));
-
-  ret = pthread_cond_init(&shm_ptr->cond, &cond_attr);
-  AssertFatal(ret == 0, "pthread_cond_init() failed: errno %d, %s\n", errno, strerror(errno));
+  init_sync_data(&shm_ptr->sync_data);
 
   shm_ptr->magic = SHM_MAGIC_NUMBER;
+
+  if (client_sync) {
+    shm_ptr->client_sync = true;
+    init_sync_data(&shm_ptr->client_sync_data);
+  }
   close(fd);
   return channel;
 }
@@ -130,10 +205,12 @@ ShmTDIQChannel *shm_td_iq_channel_connect(const char *name, int timeout_in_secon
 
   ShmTDIQChannel *channel = calloc_or_fail(1, sizeof(ShmTDIQChannel));
   channel->data = shm_ptr;
-  channel->tx_iq_data = (sample_t *)(shm_ptr + 1);
-  size_t tx_buffer_size = CIRCULAR_BUFFER_SIZE * sizeof(sample_t) * channel->data->num_antennas_tx;
-  channel->rx_iq_data = channel->tx_iq_data + tx_buffer_size / sizeof(sample_t);
+  channel->rx_iq_data = (sample_t *)(shm_ptr + 1);
+  size_t tx_buffer_size = (CIRCULAR_BUFFER_SIZE + BUFFER_PREFIX_SIZE) * sizeof(sample_t) * channel->data->num_antennas_tx;
+  channel->tx_iq_data = channel->rx_iq_data + tx_buffer_size / sizeof(sample_t);
   channel->type = IQ_CHANNEL_TYPE_CLIENT;
+  channel->num_antennas_rx = shm_ptr->num_antennas_tx;
+  channel->num_antennas_tx = shm_ptr->num_antennas_rx;
   while (shm_ptr->magic != SHM_MAGIC_NUMBER) {
     printf("Waiting for server to initialize shared memory\n");
     sleep(1);
@@ -142,15 +219,39 @@ ShmTDIQChannel *shm_td_iq_channel_connect(const char *name, int timeout_in_secon
   return channel;
 }
 
+int shm_td_iq_channel_get_nb_antennas_tx(ShmTDIQChannel *channel)
+{
+  return channel->num_antennas_tx;
+}
+
+int shm_td_iq_channel_get_nb_antennas_rx(ShmTDIQChannel *channel)
+{
+  return channel->num_antennas_rx;
+}
+
+static sample_t *get_prefix_buffer_ptr(sample_t *base_ptr, int antenna)
+{
+  return base_ptr + antenna * (CIRCULAR_BUFFER_SIZE + BUFFER_PREFIX_SIZE);
+}
+
+static sample_t *get_main_buffer_ptr(sample_t *base_ptr, int antenna)
+{
+  return get_prefix_buffer_ptr(base_ptr, antenna) + BUFFER_PREFIX_SIZE;
+}
+
 IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
                                         uint64_t timestamp,
                                         uint64_t num_samples,
                                         int antenna,
                                         const sample_t *tx_iq_data)
 {
+  AssertFatal(antenna < channel->num_antennas_tx,
+              "Antenna index %d out of range (num antennas: %d)\n",
+              antenna,
+              channel->num_antennas_tx);
   ShmTDIQChannelData *data = channel->data;
   // timestamp in the past
-  uint64_t current_time = data->timestamp;
+  uint64_t current_time = data->sync_data.timestamp;
   if (timestamp < current_time) {
     return CHANNEL_ERROR_TOO_LATE;
   }
@@ -159,13 +260,7 @@ IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
     return CHANNEL_ERROR_TOO_EARLY;
   }
 
-  sample_t *base_ptr;
-  if (channel->type == IQ_CHANNEL_TYPE_CLIENT) {
-    base_ptr = channel->rx_iq_data + antenna * CIRCULAR_BUFFER_SIZE;
-  } else {
-    base_ptr = channel->tx_iq_data + antenna * CIRCULAR_BUFFER_SIZE;
-  }
-
+  sample_t *base_ptr = get_main_buffer_ptr(channel->tx_iq_data, antenna);
   uint64_t first_sample = timestamp % CIRCULAR_BUFFER_SIZE;
   uint64_t last_sample = first_sample + num_samples - 1;
   if (last_sample >= CIRCULAR_BUFFER_SIZE) {
@@ -175,6 +270,23 @@ IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
   } else {
     memcpy(base_ptr + first_sample, tx_iq_data, num_samples * sizeof(sample_t));
   }
+
+  // Mirror end of buffer to prefix buffer for continuous zero copy reading
+  int64_t mirror_start = CIRCULAR_BUFFER_SIZE - BUFFER_PREFIX_SIZE;
+  int64_t mirror_end = CIRCULAR_BUFFER_SIZE;
+  int64_t tx_end = (first_sample + num_samples);
+  int64_t tx_start = first_sample;
+  int64_t overlap_start = (tx_start > mirror_start) ? tx_start : mirror_start;
+  int64_t overlap_end = (tx_end < mirror_end) ? tx_end : mirror_end;
+  if (overlap_end - overlap_start > 0) {
+    sample_t *prefix_ptr = get_prefix_buffer_ptr(channel->tx_iq_data, antenna);
+    memcpy(prefix_ptr + (overlap_start - mirror_start), base_ptr + overlap_start, (overlap_end - overlap_start) * sizeof(sample_t));
+  }
+
+  if (channel->type == IQ_CHANNEL_TYPE_CLIENT && data->client_sync) {
+    update_timestamp(&data->client_sync_data, timestamp + num_samples);
+  }
+
   return CHANNEL_NO_ERROR;
 }
 
@@ -182,11 +294,15 @@ IQChannelErrorType shm_td_iq_channel_rx(ShmTDIQChannel *channel,
                                         uint64_t timestamp,
                                         uint64_t num_samples,
                                         int antenna,
-                                        sample_t *tx_iq_data)
+                                        sample_t *rx_iq_data)
 {
+  AssertFatal(antenna < channel->num_antennas_rx,
+              "Antenna index %d out of range (num antennas: %d)\n",
+              antenna,
+              channel->num_antennas_rx);
   ShmTDIQChannelData *data = channel->data;
   // timestamp in the future
-  uint64_t current_time = data->timestamp;
+  uint64_t current_time = data->sync_data.timestamp;
   if (timestamp > current_time) {
     return CHANNEL_ERROR_TOO_EARLY;
   }
@@ -195,91 +311,123 @@ IQChannelErrorType shm_td_iq_channel_rx(ShmTDIQChannel *channel,
     return CHANNEL_ERROR_TOO_LATE;
   }
 
-  sample_t *base_ptr;
-  if (channel->type == IQ_CHANNEL_TYPE_CLIENT) {
-    base_ptr = channel->tx_iq_data + antenna * CIRCULAR_BUFFER_SIZE;
-  } else {
-    base_ptr = channel->rx_iq_data + antenna * CIRCULAR_BUFFER_SIZE;
-  }
+  sample_t *base_ptr = get_main_buffer_ptr(channel->rx_iq_data, antenna);
 
   uint64_t first_sample = timestamp % CIRCULAR_BUFFER_SIZE;
   uint64_t last_sample = first_sample + num_samples - 1;
   if (last_sample >= CIRCULAR_BUFFER_SIZE) {
     size_t num_samples_first_copy = CIRCULAR_BUFFER_SIZE - first_sample;
-    memcpy(tx_iq_data, base_ptr + first_sample, num_samples_first_copy * sizeof(sample_t));
-    memcpy(tx_iq_data + num_samples_first_copy, base_ptr, (num_samples - num_samples_first_copy) * sizeof(sample_t));
+    memcpy(rx_iq_data, base_ptr + first_sample, num_samples_first_copy * sizeof(sample_t));
+    memcpy(rx_iq_data + num_samples_first_copy, base_ptr, (num_samples - num_samples_first_copy) * sizeof(sample_t));
   } else {
-    memcpy(tx_iq_data, base_ptr + first_sample, num_samples * sizeof(sample_t));
+    memcpy(rx_iq_data, base_ptr + first_sample, num_samples * sizeof(sample_t));
   }
   return CHANNEL_NO_ERROR;
 }
 
+IQChannelErrorType shm_td_iq_channel_zc_rx(ShmTDIQChannel *channel,
+                                           uint64_t timestamp,
+                                           uint64_t num_samples,
+                                           int antenna,
+                                           sample_t **rx_iq_data)
+{
+  AssertFatal(num_samples <= BUFFER_PREFIX_SIZE,
+              "Number of samples %lu exceeds buffer prefix size %d for zero-copy RX\n",
+              num_samples,
+              BUFFER_PREFIX_SIZE);
+  AssertFatal(antenna < channel->num_antennas_rx,
+              "Antenna index %d out of range (num antennas: %d)\n",
+              antenna,
+              channel->num_antennas_rx);
+  ShmTDIQChannelData *data = channel->data;
+  // timestamp in the future
+  uint64_t current_time = data->client_sync_data.timestamp;
+  if (timestamp > current_time) {
+    *rx_iq_data = NULL;
+    return CHANNEL_ERROR_TOO_EARLY;
+  }
+  // timestamp is too far in the past
+  if (current_time - timestamp >= CIRCULAR_BUFFER_SIZE) {
+    *rx_iq_data = NULL;
+    return CHANNEL_ERROR_TOO_LATE;
+  }
+
+  sample_t *base_ptr = get_prefix_buffer_ptr(channel->rx_iq_data, antenna);
+
+  uint64_t first_sample_index = timestamp % CIRCULAR_BUFFER_SIZE;
+  if (first_sample_index + num_samples > CIRCULAR_BUFFER_SIZE) {
+    uint64_t num_samples_in_prefix_buffer = CIRCULAR_BUFFER_SIZE - first_sample_index;
+    // Handle wrap around inside the prefix buffer
+    *rx_iq_data = base_ptr + (BUFFER_PREFIX_SIZE - num_samples_in_prefix_buffer);
+  } else {
+    *rx_iq_data = base_ptr + BUFFER_PREFIX_SIZE + first_sample_index;
+  }
+  AssertFatal(*rx_iq_data + num_samples < base_ptr + BUFFER_PREFIX_SIZE + CIRCULAR_BUFFER_SIZE,
+              "Requested samples exceed buffer limits\n");
+  return CHANNEL_NO_ERROR;
+}
+
 void shm_td_iq_channel_produce_samples(ShmTDIQChannel *channel, size_t num_samples)
 {
   ShmTDIQChannelData *data = channel->data;
-  mutexlock(data->mutex);
-  data->timestamp += num_samples;
-  condbroadcast(data->cond);
-  mutexunlock(data->mutex);
+  update_timestamp(&data->sync_data, data->sync_data.timestamp + num_samples);
 }
 
 int shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS)
 {
   ShmTDIQChannelData *data = channel->data;
-  size_t current_timestamp = data->timestamp;
+  size_t current_timestamp = data->sync_data.timestamp;
   if (current_timestamp >= timestamp) {
     return 0;
   }
   if (timeout_uS == 0) {
-    mutexlock(data->mutex);
-    while (current_timestamp < timestamp && !channel->abort) {
-      condwait(data->cond, data->mutex);
-      current_timestamp = data->timestamp;
-    }
-    mutexunlock(data->mutex);
+    wait_for_sync(&data->sync_data, timestamp, &channel->abort);
+    return 0;
   } else {
-    struct timespec ts = {.tv_sec = 0, .tv_nsec = 0};
-    if (clock_gettime(CLOCK_REALTIME, &ts) != 0) {
-      fprintf(stderr, "Error: clock_gettime failed: %s\n", strerror(errno));
-      return 1;
-    }
-    
-    ts.tv_sec += timeout_uS / 1000000; // Convert microseconds to seconds
-    ts.tv_nsec += (timeout_uS % 1000000) * 1000; // Convert remaining microseconds to nanoseconds
-
-    mutexlock(data->mutex);
-    while (current_timestamp < timestamp && !channel->abort) {
-      int ret = pthread_cond_timedwait(&data->cond, &data->mutex, &ts);
-      if (ret == ETIMEDOUT) {
-        fprintf(stderr, "Error: Timed out waiting for samples.\n");
-        mutexunlock(data->mutex);
-        return 1;
-      } else if (ret != 0) {
-        fprintf(stderr, "Error: pthread_cond_timedwait failed: %s\n", strerror(ret));
-        mutexunlock(data->mutex);
-        return 1;
-      } else {
-        current_timestamp = data->timestamp;
-      }
-    }
-    mutexunlock(data->mutex);
+    return wait_for_sync_with_timeout(&data->sync_data, timestamp, timeout_uS, &channel->abort);
+  }
+}
+
+int shm_td_iq_channel_wait_for_client(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS)
+{
+  ShmTDIQChannelData *data = channel->data;
+  AssertFatal(data->client_sync, "Client sync not enabled on this channel\n");
+  size_t current_timestamp = data->client_sync_data.timestamp;
+  if (current_timestamp >= timestamp) {
+    return 0;
+  }
+  if (timeout_uS == 0) {
+    wait_for_sync(&data->client_sync_data, timestamp, &channel->abort);
+    return 0;
+  } else {
+    return wait_for_sync_with_timeout(&data->client_sync_data, timestamp, timeout_uS, &channel->abort);
   }
-  return 0;
 }
 
 uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel)
 {
   ShmTDIQChannelData *data = channel->data;
-  return data->timestamp;
+  return data->sync_data.timestamp;
+}
+
+uint64_t shm_td_iq_channel_get_current_client_sample(const ShmTDIQChannel *channel)
+{
+  ShmTDIQChannelData *data = channel->data;
+  return data->client_sync_data.timestamp;
 }
 
 void shm_td_iq_channel_abort(ShmTDIQChannel *channel)
 {
   ShmTDIQChannelData *data = channel->data;
-  mutexlock(data->mutex);
+  mutexlock(data->sync_data.mutex);
   channel->abort = true;
-  condbroadcast(data->cond);
-  mutexunlock(data->mutex);
+  condbroadcast(data->sync_data.cond);
+  mutexunlock(data->sync_data.mutex);
+  if (data->client_sync) {
+    mutexlock(data->client_sync_data.mutex);
+    condbroadcast(data->client_sync_data.cond);
+    mutexunlock(data->client_sync_data.mutex);
+  }
 }
 
 bool shm_td_iq_channel_is_aborted(const ShmTDIQChannel *channel)

common/utils/shm_iq_channel/shm_td_iq_channel.h

@@ -60,9 +60,10 @@ typedef struct ShmTDIQChannel_s ShmTDIQChannel;
  * @param name The name of the shared memory segment.
  * @param num_tx_ant The number of TX antennas.
  * @param num_rx_ant The number of RX antennas.
+ * @param client_sync Whether to enable client synchronization.
  * @return A pointer to the created ShmTDIQChannel structure.
  */
-ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int num_rx_ant);
+ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int num_rx_ant, bool client_sync);
 
 /**
  * @brief Connects to an existing shared memory IQ channel.
@@ -97,14 +98,30 @@ IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
  * @param timestamp The timestamp for which to get the RX IQ data slot.
  * @param num_samples The number of samples to read.
  * @param antenna The antenna index.
- * @param tx_iq_data pointer to the RX IQ data slot.
+ * @param rx_iq_data pointer to the RX IQ data.
  * @return CHANNEL_NO_ERROR if successful, error type otherwise
  */
 IQChannelErrorType shm_td_iq_channel_rx(ShmTDIQChannel *channel,
                                         uint64_t timestamp,
                                         uint64_t num_samples,
                                         int antenna,
-                                        sample_t *tx_iq_data);
+                                        sample_t *rx_iq_data);
+
+/**
+ * @brief Receive iq data from the channel, zero-copy interface
+ *
+ * @param channel The ShmTDIQChannel structure.
+ * @param timestamp The timestamp for which to get the RX IQ data slot.
+ * @param num_samples The number of samples to read.
+ * @param antenna The antenna index.
+ * @param rx_iq_data pointer to the RX IQ data slot.
+ * @return CHANNEL_NO_ERROR if successful, error type otherwise
+ */
+IQChannelErrorType shm_td_iq_channel_zc_rx(ShmTDIQChannel *channel,
+                                           uint64_t timestamp,
+                                           uint64_t num_samples,
+                                           int antenna,
+                                           sample_t **rx_iq_data);
 
 /**
  * @brief Advances the time in the channel by specified number of samples
@@ -125,6 +142,16 @@ void shm_td_iq_channel_produce_samples(ShmTDIQChannel *channel, uint64_t num_sam
  */
 int shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS);
 
+/**
+ * @brief Wait until sample at the specified timestamp is transmitted by the client
+ *
+ * @param channel The ShmTDIQChannel structure.
+ * @param timestamp The timestamp for which to wait.
+ * @param timeout_uS The timeout in microseconds to wait for the sample. 0 means wait indefinitely.
+ *
+ * @return 0 if the sample is available, 1 if timed out
+ */
+int shm_td_iq_channel_wait_for_client(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS);
 /**
  * @brief Aborts the IQ channel causing the wait to return immediately
  *
@@ -156,4 +183,16 @@ void shm_td_iq_channel_destroy(ShmTDIQChannel *channel);
  */
 uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel);
 
+/**
+ * @brief Returns current sample written by the client
+ *
+ * @param channel The ShmTDIQChannel structure.
+ *
+ * @return Current time as sample count since beginning of transmission written by the client
+ */
+uint64_t shm_td_iq_channel_get_current_client_sample(const ShmTDIQChannel *channel);
+
+int shm_td_iq_channel_get_nb_antennas_tx(ShmTDIQChannel *channel);
+int shm_td_iq_channel_get_nb_antennas_rx(ShmTDIQChannel *channel);
+
 #endif

common/utils/shm_iq_channel/tests/test_shm_td_iq_channel.c

@@ -86,7 +86,7 @@ void server(void)
 {
   int num_ant_tx = 1;
   int num_ant_rx = 1;
-  ShmTDIQChannel *channel = shm_td_iq_channel_create(SHM_CHANNEL_NAME, num_ant_tx, num_ant_rx);
+  ShmTDIQChannel *channel = shm_td_iq_channel_create(SHM_CHANNEL_NAME, num_ant_tx, num_ant_rx, false);
 
   pthread_t producer_thread;
   int ret = pthread_create(&producer_thread, NULL, produce_symbols, channel);

openair1/SIMULATION/TOOLS/random_channel.c

@@ -2371,7 +2371,72 @@ int load_channellist(uint8_t nb_tx, uint8_t nb_rx, double sampling_rate, uint64_
   return channel_list.numelt;
 } /* load_channelist */
 
-int get_noise_power_dBFS(void) {
+channel_desc_t *load_channel(uint8_t nb_tx,
+                             uint8_t nb_rx,
+                             double sampling_rate,
+                             uint64_t center_freq,
+                             double channel_bandwidth,
+                             const char *name)
+{
+  paramdef_t achannel_params[] = CHANNELMOD_MODEL_PARAMS_DESC;
+  paramlist_def_t channel_list;
+  memset(&channel_list, 0, sizeof(paramlist_def_t));
+  memcpy(channel_list.listname, modellist_name, sizeof(channel_list.listname) - 1);
+  int numparams = sizeofArray(achannel_params);
+  config_getlist(config_get_if(), &channel_list, achannel_params, numparams, CHANNELMOD_SECTION);
+  AssertFatal(channel_list.numelt > 0, "List %s.%s not found in config file\n", CHANNELMOD_SECTION, channel_list.listname);
+  int pindex_NAME = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_NAME_PNAME);
+  int pindex_DT = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_DT_PNAME);
+  int pindex_FF = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_FF_PNAME);
+  int pindex_CO = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_CO_PNAME);
+  int pindex_PL = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_PL_PNAME);
+  int pindex_NP = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_NP_PNAME);
+  int pindex_TYPE = config_paramidx_fromname(achannel_params, numparams, CHANNELMOD_MODEL_TYPE_PNAME);
+
+  for (int i = 0; i < channel_list.numelt; i++) {
+    int modid = modelid_fromstrtype(*(channel_list.paramarray[i][pindex_TYPE].strptr));
+
+    if (modid < 0) {
+      LOG_E(OCM, "Valid channel model types:\n");
+
+      for (int m = 0; channelmod_names[i].name != NULL; m++) {
+        printf(" %s ", map_int_to_str(channelmod_names, m));
+      }
+
+      AssertFatal(0, "\n  Choose a valid model type\n");
+    }
+    if (strncmp(*channel_list.paramarray[i][pindex_NAME].strptr, name, strlen(name)) == 0) {
+      channel_desc_t *channeldesc_p = new_channel_desc_scm(nb_tx,
+                                                           nb_rx,
+                                                           modid,
+                                                           sampling_rate,
+                                                           center_freq,
+                                                           channel_bandwidth,
+                                                           *(channel_list.paramarray[i][pindex_DT].dblptr),
+                                                           0.0,
+                                                           CORR_LEVEL_LOW,
+                                                           *(channel_list.paramarray[i][pindex_FF].dblptr),
+                                                           *(channel_list.paramarray[i][pindex_CO].iptr),
+                                                           *(channel_list.paramarray[i][pindex_PL].dblptr),
+                                                           *(channel_list.paramarray[i][pindex_NP].dblptr));
+      AssertFatal((channeldesc_p != NULL),
+                  "Could not allocate channel %s type %s \n",
+                  *(channel_list.paramarray[i][pindex_NAME].strptr),
+                  *(channel_list.paramarray[i][pindex_TYPE].strptr));
+      channeldesc_p->model_name = strdup(*(channel_list.paramarray[i][pindex_NAME].strptr));
+      LOG_I(OCM,
+            "Model %s type %s allocated from config file, list %s\n",
+            *(channel_list.paramarray[i][pindex_NAME].strptr),
+            *(channel_list.paramarray[i][pindex_TYPE].strptr),
+            modellist_name);
+      return channeldesc_p;
+    } /* for loop on channel_list */
+  }
+  return NULL;
+}
+
+int get_noise_power_dBFS(void)
+{
   return noise_power_dBFS;
 }
 

openair1/SIMULATION/TOOLS/sim.h

@@ -552,6 +552,12 @@ double channelmod_get_snr_dB(void);
 double channelmod_get_sinr_dB(void);
 void init_channelmod(void) ;
 int load_channellist(uint8_t nb_tx, uint8_t nb_rx, double sampling_rate, uint64_t center_freq, double channel_bandwidth) ;
+channel_desc_t *load_channel(uint8_t nb_tx,
+                             uint8_t nb_rx,
+                             double sampling_rate,
+                             uint64_t center_freq,
+                             double channel_bandwidth,
+                             const char *name);
 double N_RB2sampling_rate(uint16_t N_RB);
 double N_RB2channel_bandwidth(uint16_t N_RB);
 

radio/vrtsim/CMakeLists.txt

@@ -26,6 +26,6 @@ if (OAI_VRTSIM_TAPS_CLIENT)
 
   add_library(taps_client taps_client.cpp ${TAPS_API_HEADER})
   target_include_directories(taps_client PUBLIC .)
-  find_package(Flatbuffers REQUIRED)
+  find_package(FlatBuffers REQUIRED)
   target_link_libraries(taps_client PUBLIC flatbuffers taps_api SIMU nanomsg)
 endif()

radio/vrtsim/taps_client.cpp

@@ -38,16 +38,6 @@ extern "C" {
 #define MAX_TAPS_LEN 100
 #define MAX_TAPS_MSG_SIZE (sizeof(struct complexf) * MAX_TAPS_LEN * 4 * 4 + 20)
 
-static pthread_t client_thread;
-static bool should_run = true;
-typedef struct {
-  int id;
-  int sock;
-  uint32_t num_tx_antennas;
-  uint32_t num_rx_antennas;
-  channel_desc_t **channel_desc;
-} client_thread_args_t;
-
 typedef struct {
   void *taps_msg;
   channel_desc_t *channel_desc;
@@ -58,6 +48,21 @@ typedef struct {
   int current_buffer;
 } taps_storage_t;
 
+typedef struct {
+  int id;
+  int sock;
+  uint32_t num_tx_antennas;
+  uint32_t num_rx_antennas;
+  channel_desc_t **channel_desc;
+  taps_storage_t storage;
+  bool should_run;
+} client_thread_args_t;
+
+typedef struct {
+  client_thread_args_t args;
+  pthread_t thread;
+} taps_client_handle_t;
+
 void ascii_line_plot(const float *data, size_t size, char *buffer)
 {
   const char levels[] = "_.-=#"; // ASCII characters for different levels
@@ -100,14 +105,12 @@ static void init_taps_storage(taps_storage_t *storage, int num_tx_antennas, int
   storage->current_buffer = 0;
 }
 
-static taps_storage_t taps_storage;
-
 void *client_thread_func(void *args)
 {
   client_thread_args_t *client_thread_args = (client_thread_args_t *)args;
-  while (should_run) {
-    int next_buffer = (taps_storage.current_buffer + 1) % NUM_TAPS_BUFFERS;
-    taps_buffer_t *taps_buffer = &taps_storage.taps_buffers[next_buffer];
+  while (client_thread_args->should_run) {
+    int next_buffer = (client_thread_args->storage.current_buffer + 1) % NUM_TAPS_BUFFERS;
+    taps_buffer_t *taps_buffer = &client_thread_args->storage.taps_buffers[next_buffer];
     int ret = nn_recv(client_thread_args->sock, taps_buffer->taps_msg, MAX_TAPS_MSG_SIZE, NN_DONTWAIT);
     if (ret < 0) {
       if (errno == EAGAIN) {
@@ -143,7 +146,7 @@ void *client_thread_func(void *args)
     channel_desc->path_loss_dB = 0;
     channel_desc->channel_length = taps_len;
     *client_thread_args->channel_desc = channel_desc;
-    taps_storage.current_buffer = next_buffer;
+    client_thread_args->storage.current_buffer = next_buffer;
     LOG_A(HW, "Receved new taps message, channel_length %d, buffer %d\n", channel_desc->channel_length, next_buffer);
     for (unsigned int aarx = 0; aarx < client_thread_args->num_rx_antennas; aarx++) {
       for (unsigned int aatx = 0; aatx < client_thread_args->num_tx_antennas; aatx++) {
@@ -179,25 +182,28 @@ extern "C" void taps_client_connect(int id,
   ret = nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, "", 0);
   AssertFatal(ret == 0, "nn_setsockopt() failed, errno %d, %s\n", errno, strerror(errno));
 
-  init_taps_storage(&taps_storage, num_tx_antennas, num_rx_antennas);
 
-  client_thread_args_t *client_thread_args = static_cast<client_thread_args_t *>(malloc(sizeof(client_thread_args_t)));
+  taps_client_handle_t *client_handle = static_cast<taps_client_handle_t *>(malloc(sizeof(taps_client_handle_t)));
+  client_thread_args_t *client_thread_args = &client_handle->args;
+  init_taps_storage(&client_thread_args->storage, num_tx_antennas, num_rx_antennas);
+  client_thread_args->should_run = true;
   client_thread_args->id = id;
   client_thread_args->sock = sock;
   client_thread_args->num_rx_antennas = num_rx_antennas;
   client_thread_args->num_tx_antennas = num_tx_antennas;
   client_thread_args->channel_desc = channel_desc;
-  ret = pthread_create(&client_thread, NULL, client_thread_func, client_thread_args);
+  ret = pthread_create(&client_handle->thread, NULL, client_thread_func, client_thread_args);
   AssertFatal(ret == 0, "pthread_create() failed: errno: %d, %s\n", errno, strerror(errno));
 }
 
-extern "C" void taps_client_stop()
+extern "C" void taps_client_stop(taps_client_handle_t *handle)
 {
-  should_run = false;
-  pthread_join(client_thread, NULL);
+  handle->args.should_run = false;
+  pthread_join(handle->thread, NULL);
   for (int i = 0; i < NUM_TAPS_BUFFERS; i++) {
-    free(taps_storage.taps_buffers[i].taps_msg);
-    free(taps_storage.taps_buffers[i].channel_desc->ch_ps);
-    free(taps_storage.taps_buffers[i].channel_desc);
+    free(handle->args.storage.taps_buffers[i].taps_msg);
+    free(handle->args.storage.taps_buffers[i].channel_desc->ch_ps);
+    free(handle->args.storage.taps_buffers[i].channel_desc);
   }
+  free(handle);
 }

radio/vrtsim/taps_client.h

@@ -28,8 +28,8 @@ extern "C" {
 
 #include "sim.h"
 
-void taps_client_connect(int id, const char *socket_path, int num_tx_antennas, int num_rx_antennas, channel_desc_t **channel_desc);
-void taps_client_stop();
+void* taps_client_connect(int id, const char *socket_path, int num_tx_antennas, int num_rx_antennas, channel_desc_t **channel_desc);
+void taps_client_stop(void* handle);
 
 #ifdef __cplusplus
 }

radio/vrtsim/vrtsim.c

@@ -54,7 +54,9 @@
 typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
 
 #define MAX_NUM_ANTENNAS_TX 4
+#define MAX_NUM_ANTENNAS_RX 4
 #define MAX_CHANNEL_LENGTH (1 << 20)
+#define RX_SAMPLE_BUFFER_SIZE (1 << 20)
 
 #define ROLE_CLIENT_STRING "client"
 #define ROLE_SERVER_STRING "server"
@@ -64,6 +66,7 @@ typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
   "sample time scale. 1.0 means realtime. Values > 1 mean faster than realtime. Values < 1 mean slower than realtime\n"
 #define TAPS_SOCKET_HLP "Socket to connect to the channel emulation server\n"
 #define CLIENT_NUM_RX_HLP "Number of RX antennas of the client, specified on the server\n"
+#define CLIENT_NUM_TX_HLP "Number of TX antennas of the client, specified on the server\n"
 #define CONNECTION_DESCRIPTOR_HLP "Path to the file written by the server that the client can use to connect."
 #define DEFAULT_CHANNEL_NAME "vrtsim_channel"
 #define DEFAULT_DESCRIPTOR "/tmp/vrtsim_connection"
@@ -76,10 +79,24 @@ typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
      {"timescale",              TIME_SCALE_HLP,              0, .dblptr = &vrtsim_state->timescale,              .defdblval = 1.0,                TYPE_DOUBLE, 0}, \
      {"chanmod",                "Enable channel modelling",  0, .iptr = &vrtsim_state->chanmod,                  .defintval = 0,                  TYPE_INT,    0}, \
      {"taps-socket",            TAPS_SOCKET_HLP,             0, .strptr = &vrtsim_state->taps_socket,            .defstrval = NULL,               TYPE_STRING, 0}, \
+     {"peer-taps-socket",       TAPS_SOCKET_HLP,             0, .strptr = &vrtsim_state->peer_taps_socket,       .defstrval = NULL,               TYPE_STRING, 0}, \
      {"client-num-rx-antennas", CLIENT_NUM_RX_HLP,           0, .iptr = &vrtsim_state->client_num_rx_antennas,   .defintval = 1,                  TYPE_INT,    0}, \
-  };
+     {"client-num-tx-antennas", CLIENT_NUM_TX_HLP,           0, .iptr = &vrtsim_state->client_num_tx_antennas,   .defintval = 1,                  TYPE_INT,    0}, \
+  }
 // clang-format on
 
+enum direction {
+  RX = 0,
+  TX = 1,
+  MAX_DIRECTIONS
+};
+
+enum chanmod {
+  CHANMOD_OFF = 0,
+  CHANMOD_TX = 1,
+  CHANMOD_TXRX = 2,
+};
+
 typedef struct histogram_s {
   uint64_t diff[30];
   int num_samples;
@@ -90,15 +107,23 @@ typedef struct histogram_s {
 // Information about the peer
 typedef struct peer_info_s {
   int num_rx_antennas;
+  int num_tx_antennas;
 } peer_info_t;
 
 typedef struct tx_timing_s {
-  uint64_t tx_samples_late;
-  uint64_t tx_early;
-  uint64_t tx_samples_total;
-  double average_tx_budget;
-  histogram_t tx_histogram;
-} tx_timing_t;
+  uint64_t samples_late;
+  uint64_t early;
+  uint64_t samples_total;
+  double average_budget;
+  histogram_t histogram;
+} vrtsim_timing_t;
+
+typedef struct {
+  Actor_t *actors;
+  channel_desc_t *channel_desc;
+  char *taps_socket;
+  void* taps_client;
+} channel_modelling_t;
 
 typedef struct {
   int role;
@@ -107,26 +132,46 @@ typedef struct {
   uint64_t last_received_sample;
   pthread_t timing_thread;
   bool run_timing_thread;
+  bool run_rx_listener_thread;
   double timescale;
   double sample_rate;
   uint64_t rx_samples_late;
   uint64_t rx_early;
   uint64_t rx_samples_total;
-  tx_timing_t *tx_timing;
+  vrtsim_timing_t *tx_timing;
+  vrtsim_timing_t *rx_timing;
   peer_info_t peer_info;
   int chanmod;
   double rx_freq;
   double tx_bw;
   int tx_num_channels;
   int rx_num_channels;
-  channel_desc_t *channel_desc;
-  Actor_t *channel_modelling_actors;
-  char *taps_socket;
   int client_num_rx_antennas;
+  int client_num_tx_antennas;
+  int client_chanmod_on_tx;
+
+  char *taps_socket;
+  char *peer_taps_socket;
+  channel_modelling_t *channel_modelling[MAX_DIRECTIONS];
+  pthread_t rx_listener_thread;
 } vrtsim_state_t;
 
+typedef struct {
+  vrtsim_state_t *vrtsim_state;
+  openair0_timestamp timestamp;
+  c16_t *samples[MAX_NUM_ANTENNAS_TX];
+  int nsamps;
+  int nbAnt;
+  int flags;
+  int aarx;
+} channel_modelling_args_t;
+
 // Sample history for channel impulse response
 static c16_t saved_samples[MAX_NUM_ANTENNAS_TX][MAX_CHANNEL_LENGTH] __attribute__((aligned(32))) = {0};
+static c16_t rx_samples[MAX_NUM_ANTENNAS_RX][RX_SAMPLE_BUFFER_SIZE] __attribute__((aligned(32))) = {0};
+
+static void perform_channel_modelling(void *arg);
+static void perform_channel_modelling_rx(void *arg);
 
 static void histogram_add(histogram_t *histogram, double diff)
 {
@@ -156,27 +201,46 @@ static void histogram_merge(histogram_t *dest, histogram_t *src)
   dest->num_samples += src->num_samples;
 }
 
-static void load_channel_model(vrtsim_state_t *vrtsim_state)
+static void *rx_listener_thread(void *arg)
 {
-  load_channellist(vrtsim_state->tx_num_channels,
-                   vrtsim_state->peer_info.num_rx_antennas,
-                   vrtsim_state->sample_rate,
-                   vrtsim_state->rx_freq,
-                   vrtsim_state->tx_bw);
-  char *model_name = vrtsim_state->role == ROLE_CLIENT ? "client_tx_channel_model" : "server_tx_channel_model";
-  vrtsim_state->channel_desc = find_channel_desc_fromname(model_name);
-  AssertFatal(vrtsim_state->channel_desc != NULL,
-              "Could not find model name %s. Make sure it is present in the config file",
-              model_name);
-  LOG_I(HW,
-        "Channel model %s parameters: path_loss_dB=%.2f, nb_tx=%d, nb_rx=%d, channel_length=%d\n",
-        model_name,
-        vrtsim_state->channel_desc->path_loss_dB,
-        vrtsim_state->channel_desc->nb_tx,
-        vrtsim_state->channel_desc->nb_rx,
-        vrtsim_state->channel_desc->channel_length);
-  random_channel(vrtsim_state->channel_desc, 0);
-  AssertFatal(vrtsim_state->channel_desc != NULL, "Could not find channel model %s\n", model_name);
+  vrtsim_state_t *vrtsim_state = (vrtsim_state_t *)arg;
+  ShmTDIQChannel *channel = vrtsim_state->channel;
+  shm_td_iq_channel_wait_for_client(channel, 1, 0);
+  uint64_t current_sample = shm_td_iq_channel_get_current_client_sample(channel);
+  while (vrtsim_state->run_rx_listener_thread) {
+    shm_td_iq_channel_wait_for_client(channel, current_sample + 1, 1000000);
+    uint64_t new_sample = shm_td_iq_channel_get_current_client_sample(channel);
+    if (new_sample > current_sample) {
+      uint64_t diff = new_sample - current_sample;
+      uint64_t timestamp = diff > vrtsim_state->sample_rate / 1000 ? new_sample - vrtsim_state->sample_rate / 1000 : current_sample;
+      diff = new_sample - timestamp;
+      for (int aarx = 0; aarx < vrtsim_state->rx_num_channels; aarx++) {
+        notifiedFIFO_elt_t *task = newNotifiedFIFO_elt(sizeof(channel_modelling_args_t), 0, NULL, perform_channel_modelling_rx);
+        channel_modelling_args_t *args = (channel_modelling_args_t *)NotifiedFifoData(task);
+        args->vrtsim_state = vrtsim_state;
+        args->timestamp = timestamp;
+        args->nsamps = diff;
+        args->nbAnt = vrtsim_state->peer_info.num_tx_antennas;
+        args->flags = 0;
+        args->aarx = aarx;
+        IQChannelErrorType error = CHANNEL_NO_ERROR;
+        for (int i = 0; i < vrtsim_state->peer_info.num_tx_antennas; i++) {
+          error = shm_td_iq_channel_zc_rx(channel, timestamp, diff, i, (sample_t **)&args->samples[i]);
+          if (error != CHANNEL_NO_ERROR) {
+            LOG_W(HW, "VRTSIM: Error getting RX samples for antenna %d at timestamp %lu: %d\n", i, timestamp, error);
+            break;
+          }
+        }
+        if (error != CHANNEL_NO_ERROR) {
+          free(task);
+          continue;
+        }
+        pushNotifiedFIFO(&vrtsim_state->channel_modelling[RX]->actors[aarx].fifo, task);
+      }
+      current_sample = new_sample;
+    }
+  }
+  return NULL;
 }
 
 static void vrtsim_readconfig(vrtsim_state_t *vrtsim_state)
@@ -285,13 +349,27 @@ static int vrtsim_connect(openair0_device *device)
   // Setup a shared memory channel
   if (vrtsim_state->role == ROLE_SERVER) {
     vrtsim_state->peer_info.num_rx_antennas = vrtsim_state->client_num_rx_antennas;
+    vrtsim_state->peer_info.num_tx_antennas = vrtsim_state->client_num_tx_antennas;
+    int nb_rx;
+    int nb_tx;
+    if (vrtsim_state->chanmod == CHANMOD_OFF) {
+      nb_rx = device->openair0_cfg[0].rx_num_channels;
+      nb_tx = device->openair0_cfg[0].tx_num_channels;
+    } else if (vrtsim_state->chanmod == CHANMOD_TX) {
+      nb_rx = device->openair0_cfg[0].rx_num_channels;
+      nb_tx = vrtsim_state->client_num_tx_antennas;
+    } else {
+      nb_rx = vrtsim_state->client_num_rx_antennas;
+      nb_tx = vrtsim_state->client_num_tx_antennas;
+    }
     vrtsim_state->channel = shm_td_iq_channel_create(DEFAULT_CHANNEL_NAME,
-                                                     vrtsim_state->peer_info.num_rx_antennas,
-                                                     device->openair0_cfg[0].rx_num_channels);
+                                                     nb_tx,
+                                                     nb_rx,
+                                                     true);
     // Exchange peer info
     client_info_t client_info = {
         .server_num_rx_antennas = device->openair0_cfg[0].rx_num_channels,
-        .client_num_rx_antennas = vrtsim_state->client_num_rx_antennas,
+        .client_num_rx_antennas = vrtsim_state->peer_info.num_rx_antennas,
     };
     server_publish_client_info(client_info, vrtsim_state->connection_descriptor);
 
@@ -307,75 +385,122 @@ static int vrtsim_connect(openair0_device *device)
         client_info.client_num_rx_antennas,
         device->openair0_cfg[0].rx_num_channels);
     vrtsim_state->channel = shm_td_iq_channel_connect(DEFAULT_CHANNEL_NAME, 10);
-    vrtsim_state->peer_info.num_rx_antennas = client_info.server_num_rx_antennas;
     vrtsim_state->last_received_sample = shm_td_iq_channel_get_current_sample(vrtsim_state->channel);
   }
 
   // Handle channel modelling after number of RX antennas are known
-  int num_tx_stats = 1;
-  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
-    vrtsim_state->channel_modelling_actors = calloc_or_fail(vrtsim_state->peer_info.num_rx_antennas, sizeof(Actor_t));
+  if (vrtsim_state->chanmod == CHANMOD_TX || vrtsim_state->chanmod == CHANMOD_TXRX) {
+    vrtsim_state->channel_modelling[TX] = calloc_or_fail(1, sizeof(channel_modelling_t));
+    channel_modelling_t *chanmod_tx = vrtsim_state->channel_modelling[TX];
+    chanmod_tx->actors = calloc_or_fail(vrtsim_state->peer_info.num_rx_antennas, sizeof(Actor_t));
     for (int i = 0; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
-      init_actor(&vrtsim_state->channel_modelling_actors[i], "chanmod", -1);
+      init_actor(&chanmod_tx->actors[i], "chanmod_tx", -1);
     }
+    int nb_tx = device->openair0_cfg[0].tx_num_channels;
+    int nb_rx = vrtsim_state->peer_info.num_rx_antennas;
     if (vrtsim_state->taps_socket) {
-      taps_client_connect(0,
-                          vrtsim_state->taps_socket,
-                          device->openair0_cfg[0].tx_num_channels,
-                          vrtsim_state->peer_info.num_rx_antennas,
-                          &vrtsim_state->channel_desc);
+      chanmod_tx->taps_client = taps_client_connect(0, vrtsim_state->taps_socket, nb_tx, nb_rx, &chanmod_tx->channel_desc);
     } else {
-      load_channel_model(vrtsim_state);
+      const char *model_name = vrtsim_state->role == ROLE_SERVER ? "server_tx_channel_model" : "client_tx_channel_model";
+      chanmod_tx->channel_desc =
+          load_channel(nb_tx, nb_rx, vrtsim_state->sample_rate, vrtsim_state->rx_freq, vrtsim_state->tx_bw, model_name);
+      AssertFatal(chanmod_tx->channel_desc != NULL, "Failed to load server channel model\n");
+      random_channel(chanmod_tx->channel_desc, 0);
+    }
+
+    if (vrtsim_state->chanmod == CHANMOD_TXRX) {
+      vrtsim_state->channel_modelling[RX] = calloc_or_fail(1, sizeof(channel_modelling_t));
+      channel_modelling_t *chanmod_rx = vrtsim_state->channel_modelling[RX];
+      nb_tx = vrtsim_state->peer_info.num_tx_antennas;
+      nb_rx = device->openair0_cfg[0].rx_num_channels;
+      chanmod_rx->actors = calloc_or_fail(nb_rx, sizeof(Actor_t));
+      for (int i = 0; i < nb_rx; i++) {
+        init_actor(&chanmod_rx->actors[i], "chanmod_rx", -1);
+      }
+      if (vrtsim_state->peer_taps_socket) {
+        chanmod_rx->taps_client = taps_client_connect(0, vrtsim_state->peer_taps_socket, nb_tx, nb_rx, &chanmod_rx->channel_desc);
+      } else {
+        const char *model_name = vrtsim_state->role == ROLE_SERVER ? "server_rx_channel_model" : "client_rx_channel_model";
+        chanmod_rx->channel_desc = load_channel(nb_tx,
+                                                nb_rx,
+                                                vrtsim_state->sample_rate,
+                                                vrtsim_state->rx_freq,
+                                                vrtsim_state->tx_bw,
+                                                model_name);
+        AssertFatal(chanmod_rx->channel_desc != NULL, "Failed to load server channel model\n");
+        random_channel(chanmod_rx->channel_desc, 0);
+      }
+      vrtsim_state->run_rx_listener_thread = true;
+      pthread_create(&vrtsim_state->rx_listener_thread, NULL, rx_listener_thread, vrtsim_state);
     }
-    num_tx_stats = vrtsim_state->peer_info.num_rx_antennas;
   }
-  vrtsim_state->tx_timing = calloc_or_fail(num_tx_stats, sizeof(tx_timing_t));
+
+  int num_tx_stats = vrtsim_state->chanmod == CHANMOD_OFF ? device->openair0_cfg[0].tx_num_channels : vrtsim_state->peer_info.num_rx_antennas;
+  vrtsim_state->tx_timing = calloc_or_fail(num_tx_stats, sizeof(vrtsim_timing_t));
   for (int i = 0; i < num_tx_stats; i++) {
-    vrtsim_state->tx_timing[i].tx_histogram.min_samples = 100;
+    vrtsim_state->tx_timing[i].histogram.min_samples = 100;
     // Set the histogram range to 3000uS. Anything above that is not interesting
-    vrtsim_state->tx_timing[i].tx_histogram.range = 3000.0;
+    vrtsim_state->tx_timing[i].histogram.range = 3000.0;
+  }
+
+  int num_rx_stats = vrtsim_state->chanmod == CHANMOD_TXRX ? device->openair0_cfg[0].rx_num_channels : 0;
+  vrtsim_state->rx_timing = calloc_or_fail(num_rx_stats, sizeof(vrtsim_timing_t));
+  for (int i = 0; i < num_rx_stats; i++) {
+    vrtsim_state->rx_timing[i].histogram.min_samples = 100;
+    // Set the histogram range to 3000uS. Anything above that is not interesting
+    vrtsim_state->rx_timing[i].histogram.range = 3000.0;
   }
 
   return 0;
 }
 
-static int vrtsim_write_internal(vrtsim_state_t *vrtsim_state,
+static void vrtsim_write_internal(vrtsim_state_t *vrtsim_state,
                                  openair0_timestamp timestamp,
                                  c16_t *samples,
                                  int nsamps,
-                                 int aarx,
+                                 int aatx,
                                  int flags,
                                  int stats_index)
 {
-  tx_timing_t *tx_timing = &vrtsim_state->tx_timing[stats_index];
+  vrtsim_timing_t *tx_timing = &vrtsim_state->tx_timing[stats_index];
 
   uint64_t sample = shm_td_iq_channel_get_current_sample(vrtsim_state->channel);
   int64_t diff = timestamp - sample;
   double budget = diff / (vrtsim_state->sample_rate / 1e6);
-  tx_timing->average_tx_budget = .05 * budget + .95 * tx_timing->average_tx_budget;
-  histogram_add(&tx_timing->tx_histogram, budget);
+  tx_timing->average_budget = .05 * budget + .95 * tx_timing->average_budget;
+  histogram_add(&tx_timing->histogram, budget);
 
-  int ret = shm_td_iq_channel_tx(vrtsim_state->channel, timestamp, nsamps, aarx, (sample_t *)samples);
+  int ret = shm_td_iq_channel_tx(vrtsim_state->channel, timestamp, nsamps, aatx, (sample_t *)samples);
 
   if (ret == CHANNEL_ERROR_TOO_LATE) {
-    tx_timing->tx_samples_late += nsamps;
+    tx_timing->samples_late += nsamps;
   } else if (ret == CHANNEL_ERROR_TOO_EARLY) {
-    tx_timing->tx_early += 1;
+    tx_timing->early += 1;
   }
-  tx_timing->tx_samples_total += nsamps;
-
-  return nsamps;
+  tx_timing->samples_total += nsamps;
 }
 
-typedef struct {
-  vrtsim_state_t *vrtsim_state;
-  openair0_timestamp timestamp;
-  c16_t *samples[MAX_NUM_ANTENNAS_TX];
-  int nsamps;
-  int nbAnt;
-  int flags;
-  int aarx;
-} channel_modelling_args_t;
+static void cf_to_c16(const cf_t *in, c16_t *out, int nsamps)
+{
+#if defined(__AVX512F__)
+  for (int i = 0; i < nsamps / 8; i++) {
+    simde__m512 *in512 = (simde__m512 *)&in[i * 8];
+    simde__m256i *out512 = (simde__m256i *)&out[i * 8];
+    *out512 = simde_mm512_cvtsepi32_epi16(simde_mm512_cvtps_epi32(*in512));
+  }
+#elif defined(__AVX2__)
+  for (int i = 0; i < nsamps / 4; i++) {
+    simde__m256 *in256 = (simde__m256 *)&in[i * 4];
+    simde__m128i *out128 = (simde__m128i *)&out[i * 4];
+    *out128 = simde_mm256_cvtsepi32_epi16(simde_mm256_cvtps_epi32(*in256));
+  }
+#else
+  for (int i = 0; i < nsamps; i++) {
+    out[i].r = lroundf(in[i].r);
+    out[i].i = lroundf(in[i].i);
+  }
+#endif
+}
 
 static void perform_channel_modelling(void *arg)
 {
@@ -391,7 +516,7 @@ static void perform_channel_modelling(void *arg)
   // Apply noise from global settings
   get_noise_vector((float *)samples, nsamps * 2);
 
-  channel_desc_t *channel_desc = vrtsim_state->channel_desc;
+  channel_desc_t *channel_desc = vrtsim_state->channel_modelling[TX]->channel_desc;
 
   if (channel_desc == NULL) {
     return;
@@ -434,24 +559,8 @@ static void perform_channel_modelling(void *arg)
 
   // Convert to c16_t
   c16_t samples_out[aligned_nsamps] __attribute__((aligned(64)));
-#if defined(__AVX512F__)
-  for (int i = 0; i < aligned_nsamps / 8; i++) {
-    simde__m512 *in = (simde__m512 *)&samples[i * 8];
-    simde__m256i *out = (simde__m256i *)&samples_out[i * 8];
-    *out = simde_mm512_cvtsepi32_epi16(simde_mm512_cvtps_epi32(*in));
-  }
-#elif defined(__AVX2__)
-  for (int i = 0; i < aligned_nsamps / 4; i++) {
-    simde__m256 *in = (simde__m256 *)&samples[i * 4];
-    simde__m128i *out = (simde__m128i *)&samples_out[i * 4];
-    *out = simde_mm256_cvtsepi32_epi16(simde_mm256_cvtps_epi32(*in));
-  }
-#else
-  for (int i = 0; i < nsamps; i++) {
-    samples_out[i].r = lroundf(samples[i].r);
-    samples_out[i].i = lroundf(samples[i].i);
-  }
-#endif
+  cf_to_c16(samples, samples_out, aligned_nsamps);
+
 
   vrtsim_write_internal(channel_modelling_args->vrtsim_state,
                         channel_modelling_args->timestamp,
@@ -462,6 +571,83 @@ static void perform_channel_modelling(void *arg)
                         aarx);
 }
 
+static void perform_channel_modelling_rx(void *arg)
+{
+  channel_modelling_args_t *channel_modelling_args = arg;
+  vrtsim_state_t *vrtsim_state = channel_modelling_args->vrtsim_state;
+  int nsamps = channel_modelling_args->nsamps;
+  int aarx = channel_modelling_args->aarx;
+  int nb_tx_ant = channel_modelling_args->nbAnt;
+  c16_t **input_samples = (c16_t **)channel_modelling_args->samples;
+
+  int aligned_nsamps = ceil_mod(nsamps, (512 / 8) / sizeof(cf_t));
+  cf_t samples[aligned_nsamps] __attribute__((aligned(64)));
+  // Apply noise from global settings
+  get_noise_vector((float *)samples, nsamps * 2);
+
+  channel_desc_t *channel_desc = vrtsim_state->channel_modelling[RX]->channel_desc;
+
+  if (channel_desc == NULL) {
+    return;
+  }
+
+  cf_t channel_impulse_response[nb_tx_ant][channel_desc->channel_length];
+  cf_t *channel_impulse_response_p[nb_tx_ant];
+  if (!vrtsim_state->peer_taps_socket) {
+    const float pathloss_linear = powf(10, channel_desc->path_loss_dB / 20.0);
+    // Convert channel impulse response to float + apply pathloss
+    for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+      const struct complexd *channelModel = channel_desc->ch[aarx + (aatx * channel_desc->nb_rx)];
+      for (int i = 0; i < channel_desc->channel_length; i++) {
+        channel_impulse_response[aatx][i].r = channelModel[i].r * pathloss_linear;
+        channel_impulse_response[aatx][i].i = channelModel[i].i * pathloss_linear;
+      }
+      channel_impulse_response_p[aatx] = channel_impulse_response[aatx];
+    }
+  } else {
+    for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+      struct complexf *channelModel = channel_desc->ch_ps[aarx + (aatx * channel_desc->nb_rx)];
+      channel_impulse_response_p[aatx] = channelModel;
+    }
+  }
+
+  for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+    for (int i = 0; i < nsamps + channel_desc->channel_length - 1; i++) {
+      cf_t *impulse_response = channel_impulse_response_p[aatx];
+      cf_t sample_out = {0, 0};
+      for (int l = 0; l < channel_desc->channel_length; l++) {
+        int index = i - l;
+        if (index < 0) {
+          continue;
+        }
+        if (index >= nsamps) {
+          continue;
+        }
+        c16_t tx_input = input_samples[aatx][index];
+        sample_out.r += tx_input.r * impulse_response[l].r - tx_input.i * impulse_response[l].i;
+        sample_out.i += tx_input.i * impulse_response[l].r + tx_input.r * impulse_response[l].i;
+      }
+      c16_t *rx_output = &rx_samples[aarx][(channel_modelling_args->timestamp + i) % RX_SAMPLE_BUFFER_SIZE];
+      rx_output->r += sample_out.r;
+      rx_output->i += sample_out.i;
+    }
+  }
+  vrtsim_timing_t *rx_timing = &vrtsim_state->rx_timing[aarx];
+
+  uint64_t sample = shm_td_iq_channel_get_current_sample(vrtsim_state->channel);
+  int64_t diff = channel_modelling_args->timestamp - sample;
+  double budget = diff / (vrtsim_state->sample_rate / 1e6);
+  rx_timing->average_budget = .05 * budget + .95 * rx_timing->average_budget;
+  histogram_add(&rx_timing->histogram, budget);
+
+  if (sample >= channel_modelling_args->timestamp) {
+    rx_timing->samples_late += nsamps;
+  } else if (channel_modelling_args->timestamp - sample >= RX_SAMPLE_BUFFER_SIZE) {
+    rx_timing->early += 1;
+  }
+  rx_timing->samples_total += nsamps;
+}
+
 static int vrtsim_write_with_chanmod(vrtsim_state_t *vrtsim_state,
                                      openair0_timestamp timestamp,
                                      void **samplesVoid,
@@ -482,7 +668,7 @@ static int vrtsim_write_with_chanmod(vrtsim_state_t *vrtsim_state,
     for (int i = 0; i < nbAnt; i++) {
       args->samples[i] = samplesVoid[i];
     }
-    pushNotifiedFIFO(&vrtsim_state->channel_modelling_actors[aarx].fifo, task);
+    pushNotifiedFIFO(&vrtsim_state->channel_modelling[TX]->actors[aarx].fifo, task);
   }
   int start_index = timestamp % MAX_CHANNEL_LENGTH;
   int end_index = min(start_index + nsamps, MAX_CHANNEL_LENGTH);
@@ -514,9 +700,34 @@ static int vrtsim_write(openair0_device *device, openair0_timestamp timestamp, v
   AssertFatal(timestamp >= 0, "Timestamp must be non-negative, got %ld\n", timestamp);
   timestamp -= device->openair0_cfg->command_line_sample_advance;
   vrtsim_state_t *vrtsim_state = (vrtsim_state_t *)device->priv;
-  bool channel_modelling = vrtsim_state->chanmod || vrtsim_state->taps_socket;
-  return channel_modelling ? vrtsim_write_with_chanmod(vrtsim_state, timestamp, samplesVoid, nsamps, nbAnt, flags)
-                           : vrtsim_write_internal(vrtsim_state, timestamp, (c16_t *)samplesVoid[0], nsamps, 0, flags, 0);
+
+  if (vrtsim_state->chanmod > CHANMOD_OFF) {
+    return vrtsim_write_with_chanmod(vrtsim_state, timestamp, samplesVoid, nsamps, nbAnt, flags);
+  } else {
+    int nb_ant_to_write = min(nbAnt, shm_td_iq_channel_get_nb_antennas_tx(vrtsim_state->channel));
+    int aatx;
+    for (aatx = 0; aatx < nb_ant_to_write; aatx++) {
+      vrtsim_write_internal(vrtsim_state,
+                           timestamp,
+                           (c16_t *)samplesVoid[aatx],
+                           nsamps,
+                           aatx,
+                           flags,
+                           0);
+    }
+    c16_t zero_samples[nsamps] __attribute__((aligned(32)));
+    memset(zero_samples, 0, sizeof(c16_t) * nsamps);
+    for (; aatx < shm_td_iq_channel_get_nb_antennas_tx(vrtsim_state->channel); aatx++) {
+      vrtsim_write_internal(vrtsim_state,
+                           timestamp,
+                           zero_samples,
+                           nsamps,
+                           aatx,
+                           flags,
+                           0);
+    }
+    return nsamps;
+  }
 }
 
 static int vrtsim_read(openair0_device *device, openair0_timestamp *ptimestamp, void **samplesVoid, int nsamps, int nbAnt)
@@ -532,7 +743,7 @@ static int vrtsim_read(openair0_device *device, openair0_timestamp *ptimestamp,
     uint64_t start_sample = shm_td_iq_channel_get_current_sample(vrtsim_state->channel);
     uint64_t timeout_uS = 2 * 1000 * 1000; // 2 seconds timeout waiting for sample number to change
     //
-    while (shm_td_iq_channel_wait(vrtsim_state->channel, vrtsim_state->last_received_sample + nsamps, timeout_uS) == 1) {
+    while (shm_td_iq_channel_wait(vrtsim_state->channel, vrtsim_state->last_received_sample + nsamps, timeout_uS) != 0) {
       uint64_t sample = shm_td_iq_channel_get_current_sample(vrtsim_state->channel);
       if (sample == start_sample) {
         LOG_E(HW,
@@ -546,12 +757,42 @@ static int vrtsim_read(openair0_device *device, openair0_timestamp *ptimestamp,
     }
   }
 
-  int ret = shm_td_iq_channel_rx(vrtsim_state->channel, vrtsim_state->last_received_sample, nsamps, 0, samplesVoid[0]);
-  if (ret == CHANNEL_ERROR_TOO_LATE) {
-    vrtsim_state->rx_samples_late += nsamps;
-  } else if (ret == CHANNEL_ERROR_TOO_EARLY) {
-    vrtsim_state->rx_early += 1;
+  if (vrtsim_state->chanmod == CHANMOD_TXRX) {
+    if (vrtsim_state->last_received_sample + nsamps > RX_SAMPLE_BUFFER_SIZE) {
+      for (int aarx = 0; aarx < nbAnt; aarx++) {
+        int first_cp_nsamps = min(RX_SAMPLE_BUFFER_SIZE - (vrtsim_state->last_received_sample % RX_SAMPLE_BUFFER_SIZE), nsamps);
+        memcpy(samplesVoid[aarx],
+               &rx_samples[aarx][vrtsim_state->last_received_sample % RX_SAMPLE_BUFFER_SIZE],
+               sizeof(c16_t) * first_cp_nsamps);
+        memset(&rx_samples[aarx][vrtsim_state->last_received_sample % RX_SAMPLE_BUFFER_SIZE], 0, sizeof(c16_t) * first_cp_nsamps);
+        memcpy(&((c16_t *)samplesVoid[aarx])[first_cp_nsamps], &rx_samples[aarx][0], sizeof(c16_t) * (nsamps - first_cp_nsamps));
+        memset(&rx_samples[aarx][0], 0, sizeof(c16_t) * (nsamps - first_cp_nsamps));
+      }
+    } else {
+      for (int aarx = 0; aarx < nbAnt; aarx++) {
+        memcpy(samplesVoid[aarx],
+               &rx_samples[aarx][vrtsim_state->last_received_sample % RX_SAMPLE_BUFFER_SIZE],
+               sizeof(c16_t) * nsamps);
+        memset(&rx_samples[aarx][vrtsim_state->last_received_sample % RX_SAMPLE_BUFFER_SIZE], 0, sizeof(c16_t) * nsamps);
+      }
+    }
+  } else {
+    int nb_ant_to_read = min(nbAnt, shm_td_iq_channel_get_nb_antennas_rx(vrtsim_state->channel));
+    int aarx;
+    for (aarx = 0; aarx < nb_ant_to_read; aarx++) {
+      int ret = shm_td_iq_channel_rx(vrtsim_state->channel, vrtsim_state->last_received_sample, nsamps, aarx, samplesVoid[aarx]);
+      if (ret == CHANNEL_ERROR_TOO_LATE) {
+        vrtsim_state->rx_samples_late += nsamps;
+      } else if (ret == CHANNEL_ERROR_TOO_EARLY) {
+        vrtsim_state->rx_early += 1;
+      }
+    }
+    for (; aarx < nbAnt; aarx++) {
+      // Fill remaining antennas with zeros
+      memset(samplesVoid[aarx], 0, sizeof(c16_t) * nsamps);
+    }
   }
+
   vrtsim_state->rx_samples_total += nsamps;
   *ptimestamp = vrtsim_state->last_received_sample;
   vrtsim_state->last_received_sample += nsamps;
@@ -567,40 +808,95 @@ static void vrtsim_end(openair0_device *device)
     AssertFatal(ret == 0, "pthread_join() failed: errno: %d, %s\n", errno, strerror(errno));
   }
 
-  tx_timing_t *tx_timing = vrtsim_state->tx_timing;
-  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
+  vrtsim_timing_t *tx_timing = vrtsim_state->tx_timing;
+  vrtsim_timing_t *rx_timing = vrtsim_state->rx_timing;
+  if (vrtsim_state->chanmod != CHANMOD_OFF) {
     for (int i = 0; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
-      shutdown_actor(&vrtsim_state->channel_modelling_actors[i]);
+      shutdown_actor(&vrtsim_state->channel_modelling[TX]->actors[i]);
     }
-    free(vrtsim_state->channel_modelling_actors);
     for (int i = 1; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
-      histogram_merge(&tx_timing->tx_histogram, &tx_timing[i].tx_histogram);
-      tx_timing->tx_early += tx_timing[i].tx_early;
-      tx_timing->tx_samples_late += tx_timing[i].tx_samples_late;
-      tx_timing->average_tx_budget += tx_timing[i].average_tx_budget;
-      tx_timing->tx_samples_total += tx_timing[i].tx_samples_total;
+      histogram_merge(&tx_timing->histogram, &tx_timing[i].histogram);
+      tx_timing->early += tx_timing[i].early;
+      tx_timing->samples_late += tx_timing[i].samples_late;
+      tx_timing->average_budget += tx_timing[i].average_budget;
+      tx_timing->samples_total += tx_timing[i].samples_total;
     }
-    tx_timing->average_tx_budget /= vrtsim_state->peer_info.num_rx_antennas;
-    free_noise_device();
-    if (vrtsim_state->taps_socket) {
-      taps_client_stop();
+    tx_timing->average_budget /= vrtsim_state->peer_info.num_rx_antennas;
+
+    if (vrtsim_state->chanmod == CHANMOD_TXRX) {
+      for (int i = 0; i < vrtsim_state->peer_info.num_tx_antennas; i++) {
+        shutdown_actor(&vrtsim_state->channel_modelling[RX]->actors[i]);
+      }
+
+      for (int i = 1; i < vrtsim_state->rx_num_channels; i++) {
+        histogram_merge(&rx_timing->histogram, &rx_timing[i].histogram);
+        rx_timing->early += rx_timing[i].early;
+        rx_timing->samples_late += rx_timing[i].samples_late;
+        rx_timing->average_budget += rx_timing[i].average_budget;
+        rx_timing->samples_total += rx_timing[i].samples_total;
+      }
+      tx_timing->average_budget /= vrtsim_state->peer_info.num_rx_antennas;
     }
   }
-  shm_td_iq_channel_abort(vrtsim_state->channel);
-  sleep(1);
-  shm_td_iq_channel_destroy(vrtsim_state->channel);
 
   LOG_I(HW,
         "VRTSIM: Realtime issues: TX %.2f%%, RX %.2f%%\n",
-        tx_timing->tx_samples_late / (float)tx_timing->tx_samples_total * 100,
+        tx_timing->samples_late / (float)tx_timing->samples_total * 100,
         vrtsim_state->rx_samples_late / (float)vrtsim_state->rx_samples_total * 100);
   LOG_I(HW,
         "VRTSIM: Read/write too early (suspected radio implementaton error) TX: %lu, RX: %lu\n",
-        tx_timing->tx_early,
+        tx_timing->early,
         vrtsim_state->rx_early);
-  LOG_I(HW, "VRTSIM: Average TX budget %.3lf uS (more is better)\n", tx_timing->average_tx_budget);
-  histogram_print(&tx_timing->tx_histogram);
+  LOG_I(HW, "VRTSIM: Average TX budget %.3lf uS (more is better)\n", tx_timing->average_budget);
+  histogram_print(&tx_timing->histogram);
+  if (vrtsim_state->rx_timing) {
+    vrtsim_timing_t *rx_timing = vrtsim_state->rx_timing;
+    for (int i = 1; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
+      histogram_merge(&rx_timing->histogram, &rx_timing[i].histogram);
+      rx_timing->early += rx_timing[i].early;
+      rx_timing->samples_late += rx_timing[i].samples_late;
+      rx_timing->average_budget += rx_timing[i].average_budget;
+      rx_timing->samples_total += rx_timing[i].samples_total;
+    }
+  }
+
+  if (rx_timing) {
+    LOG_I(HW, "VRTSIM: Average RX budget %.3lf uS (more is better)\n", rx_timing->average_budget);
+    histogram_print(&rx_timing->histogram);
+    vrtsim_timing_t *rx_timing = vrtsim_state->rx_timing;
+    for (int i = 1; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
+      histogram_merge(&rx_timing->histogram, &rx_timing[i].histogram);
+      rx_timing->early += rx_timing[i].early;
+      rx_timing->samples_late += rx_timing[i].samples_late;
+      rx_timing->average_budget += rx_timing[i].average_budget;
+      rx_timing->samples_total += rx_timing[i].samples_total;
+    }
+  }
+
+  if (vrtsim_state->run_rx_listener_thread) {
+    vrtsim_state->run_rx_listener_thread = false;
+    int ret = pthread_join(vrtsim_state->rx_listener_thread, NULL);
+    AssertFatal(ret == 0, "pthread_join() failed: errno: %d, %s\n", errno, strerror(errno));
+  }
+
+  shm_td_iq_channel_abort(vrtsim_state->channel);
+  sleep(1);
+  shm_td_iq_channel_destroy(vrtsim_state->channel);
+  free_noise_device();
+
   free(vrtsim_state->tx_timing);
+  if (vrtsim_state->rx_timing) {
+    free(vrtsim_state->rx_timing);
+  }
+  for (int dir = TX; dir <= RX; dir++) {
+    if (vrtsim_state->channel_modelling[dir]) {
+      if (vrtsim_state->channel_modelling[dir]->taps_client) {
+        taps_client_stop(vrtsim_state->channel_modelling[dir]->taps_client);
+      }
+      free(vrtsim_state->channel_modelling[dir]->actors);
+      free(vrtsim_state->channel_modelling[dir]);
+    }
+  }
 
   if (vrtsim_state->role == ROLE_SERVER) {
     int ret = remove(vrtsim_state->connection_descriptor);