Merge remote-tracking branch 'origin/vrtsim-taps-client' into integration_2025_w34 (!3407)

Taps client for vrtsim

The MR allows vrtsim to connect to the Raytracing Channel Emulator (CE later).
https://gitlab.eurecom.fr/oai/raytracing-channel-emulator

Additional Ubuntu packages required:

- libnanomsg-dev
- flatbuffers-compiler
- libflatbuffers-dev

Everything was tested on the same host. Follow the steps below to
successfully replicate:

1. Build nr-softmodem dependencies
2. Build nr-softmodem for (gNB, nrUE) with OAI_VRTSIM_TAPS_CLIENTset to ON.
3. Start the core network (Optional) follow this tutorial
4. Start the gNB

       sudo ./nr-softmodem -O ../../../targets/PROJECTS/GENERIC-NR-5GC/CONF/gnb.sa.band78.fr1.106PRB.usrpb210.conf --gNBs.[0].min_rxtxtime 6 --device.name vrtsim --vrtsim.role server --vrtsim.timescale 0.5 --vrtsim.taps-socket ipc:///tmp/ru_socket_0

5. Start the nrUE

       sudo ./nr-uesoftmodem -C 3619200000 -r 106 --numerology 1 --ssb 516  --band 78 --device.name vrtsim --vrtsim.taps-socket ipc:///tmp/ue_socket_0

6. Start the CE Server, once the UE and gNB are running properly, so the
   server can connect to IPC sockets
7. Start CE client; the binaries can be downloaded from a pipeline run
   or from this link
8. Move the UE in the CE client to generate a new taps array and
   populate the taps in softmodems

common/utils/shm_iq_channel/shm_td_iq_channel.c

@@ -40,7 +40,6 @@ typedef struct {
   int num_antennas_tx;
   int num_antennas_rx;
   uint64_t timestamp;
-  bool is_connected;
   pthread_mutex_t mutex;
   pthread_cond_t cond;
 } ShmTDIQChannelData;
@@ -48,10 +47,10 @@ typedef struct {
 typedef struct ShmTDIQChannel_s {
   IQChannelType type;
   ShmTDIQChannelData *data;
-  uint64_t last_timestamp;
   char name[256];
   sample_t *tx_iq_data;
   sample_t *rx_iq_data;
+  bool abort;
 } ShmTDIQChannel;
 
 ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int num_rx_ant)
@@ -75,14 +74,12 @@ ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int n
   memset(shm_ptr, 0, total_size);
   shm_ptr->num_antennas_tx = num_tx_ant;
   shm_ptr->num_antennas_rx = num_rx_ant;
-  shm_ptr->is_connected = false;
   ShmTDIQChannel *channel = calloc_or_fail(1, sizeof(ShmTDIQChannel));
   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->data = shm_ptr;
   channel->type = IQ_CHANNEL_TYPE_SERVER;
-  channel->last_timestamp = 0;
   pthread_mutexattr_t mutex_attr;
   pthread_condattr_t cond_attr;
   int ret = pthread_mutexattr_init(&mutex_attr);
@@ -137,12 +134,10 @@ ShmTDIQChannel *shm_td_iq_channel_connect(const char *name, int timeout_in_secon
   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->type = IQ_CHANNEL_TYPE_CLIENT;
-  channel->last_timestamp = 0;
   while (shm_ptr->magic != SHM_MAGIC_NUMBER) {
     printf("Waiting for server to initialize shared memory\n");
     sleep(1);
   }
-  shm_ptr->is_connected = true;
   close(fd);
   return channel;
 }
@@ -154,9 +149,6 @@ IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
                                         const sample_t *tx_iq_data)
 {
   ShmTDIQChannelData *data = channel->data;
-  if (data->is_connected == false) {
-    return CHANNEL_ERROR_NOT_CONNECTED;
-  }
   // timestamp in the past
   uint64_t current_time = data->timestamp;
   if (timestamp < current_time) {
@@ -193,9 +185,6 @@ IQChannelErrorType shm_td_iq_channel_rx(ShmTDIQChannel *channel,
                                         sample_t *tx_iq_data)
 {
   ShmTDIQChannelData *data = channel->data;
-  if (data->is_connected == false) {
-    return CHANNEL_ERROR_NOT_CONNECTED;
-  }
   // timestamp in the future
   uint64_t current_time = data->timestamp;
   if (timestamp > current_time) {
@@ -228,37 +217,54 @@ IQChannelErrorType shm_td_iq_channel_rx(ShmTDIQChannel *channel,
 void shm_td_iq_channel_produce_samples(ShmTDIQChannel *channel, size_t num_samples)
 {
   ShmTDIQChannelData *data = channel->data;
-  if (channel->type != IQ_CHANNEL_TYPE_SERVER) {
-    return;
-  }
-  if (data->is_connected == false) {
-    return;
-  }
   mutexlock(data->mutex);
   data->timestamp += num_samples;
   condbroadcast(data->cond);
   mutexunlock(data->mutex);
 }
 
-void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp)
+int shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS)
 {
   ShmTDIQChannelData *data = channel->data;
-  if (data->is_connected == false) {
-    abort();
-    return;
-  }
-
   size_t current_timestamp = data->timestamp;
   if (current_timestamp >= timestamp) {
-    return;
+    return 0;
   }
-  mutexlock(data->mutex);
-  while (current_timestamp < timestamp) {
-    condwait(data->cond, data->mutex);
-    current_timestamp = data->timestamp;
+  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);
+  } 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);
   }
-  mutexunlock(data->mutex);
-  return;
+  return 0;
 }
 
 uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel)
@@ -267,9 +273,18 @@ uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel)
   return data->timestamp;
 }
 
-bool shm_td_iq_channel_is_connected(const ShmTDIQChannel *channel)
+void shm_td_iq_channel_abort(ShmTDIQChannel *channel)
 {
-  return channel->data->is_connected;
+  ShmTDIQChannelData *data = channel->data;
+  mutexlock(data->mutex);
+  channel->abort = true;
+  condbroadcast(data->cond);
+  mutexunlock(data->mutex);
+}
+
+bool shm_td_iq_channel_is_aborted(const ShmTDIQChannel *channel)
+{
+  return channel->abort;
 }
 
 void shm_td_iq_channel_destroy(ShmTDIQChannel *channel)
@@ -278,9 +293,11 @@ void shm_td_iq_channel_destroy(ShmTDIQChannel *channel)
   size_t tx_buffer_size = CIRCULAR_BUFFER_SIZE * sizeof(sample_t) * data->num_antennas_tx;
   size_t rx_buffer_size = CIRCULAR_BUFFER_SIZE * sizeof(sample_t) * data->num_antennas_rx;
   size_t total_size = sizeof(ShmTDIQChannelData) + tx_buffer_size + rx_buffer_size;
-  munmap(data, total_size);
   if (channel->type == IQ_CHANNEL_TYPE_SERVER) {
+    munmap(data, total_size);
     shm_unlink(channel->name);
+  } else {
+    munmap(data, total_size);
   }
   free(channel);
 }

common/utils/shm_iq_channel/shm_td_iq_channel.h

@@ -118,16 +118,27 @@ void shm_td_iq_channel_produce_samples(ShmTDIQChannel *channel, uint64_t num_sam
  * @brief Wait until sample at the specified timestamp is available
  *
  * @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
  */
-void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp);
+int shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS);
 
 /**
- * @brief Checks if the IQ channel is connected.
+ * @brief Aborts the IQ channel causing the wait to return immediately
  *
  * @param channel The ShmTDIQChannel structure.
- * @return True if the channel is connected, false otherwise.
  */
-bool shm_td_iq_channel_is_connected(const ShmTDIQChannel *channel);
+void shm_td_iq_channel_abort(ShmTDIQChannel *channel);
+
+/**
+ * @brief Checks if the IQ channel is aborted.
+ *
+ * @param channel The ShmTDIQChannel structure.
+ * @return True if the channel is aborted, false otherwise.
+ */
+bool shm_td_iq_channel_is_aborted(const ShmTDIQChannel *channel);
 
 /**
  * @brief Destroys the shared memory IQ channel.

common/utils/shm_iq_channel/tests/test_shm_td_iq_channel.c

@@ -87,15 +87,6 @@ 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);
-  for (int i = 0; i < 10; i++) {
-    if (shm_td_iq_channel_is_connected(channel)) {
-      printf("Server connected\n");
-      break;
-    }
-    printf("Waiting for client\n");
-    sleep(1);
-  }
-  AssertFatal(shm_td_iq_channel_is_connected(channel), "Server failed to connect\n");
 
   pthread_t producer_thread;
   int ret = pthread_create(&producer_thread, NULL, produce_symbols, channel);
@@ -104,7 +95,7 @@ void server(void)
   uint64_t timestamp = 0;
   int iq_contents = 0;
   while (timestamp < num_samples_per_update * num_updates) {
-    shm_td_iq_channel_wait(channel, timestamp + num_samples_per_update);
+    shm_td_iq_channel_wait(channel, timestamp + num_samples_per_update, 0);
     uint64_t target_timestamp = timestamp + num_samples_per_update;
     timestamp += num_samples_per_update;
     uint32_t iq_data[num_samples_per_update];
@@ -127,21 +118,12 @@ int client(void)
 {
   int total_errors = 0;
   ShmTDIQChannel *channel = shm_td_iq_channel_connect(SHM_CHANNEL_NAME, 10);
-  for (int i = 0; i < 10; i++) {
-    if (shm_td_iq_channel_is_connected(channel)) {
-      printf("Client connected\n");
-      break;
-    }
-    printf("Waiting for server\n");
-    sleep(1);
-  }
-  AssertFatal(shm_td_iq_channel_is_connected(channel), "Client failed to connect\n");
 
   uint64_t timestamp = 0;
   int iq_contents = 0;
 
   while (timestamp < num_samples_per_update * num_updates) {
-    shm_td_iq_channel_wait(channel, timestamp + num_samples_per_update);
+    shm_td_iq_channel_wait(channel, timestamp + num_samples_per_update, 0);
     // Server starts producing from second slot
     if (timestamp > num_samples_per_update) {
       uint64_t target_timestamp = timestamp - num_samples_per_update;