Detect disconnecton & handle error states in vrtsim

Add handling for error states in vrtsim which allows to close the
softmodems with CTRL+C without hanging, improving user experience
 - Client detection of stale timer.
 - Abort state for client and server

common/utils/shm_iq_channel/shm_td_iq_channel.c

@@ -52,6 +52,7 @@ typedef struct ShmTDIQChannel_s {
   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)
@@ -237,10 +238,11 @@ void shm_td_iq_channel_produce_samples(ShmTDIQChannel *channel, size_t num_sampl
   mutexunlock(data->mutex);
 }
 
-void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp)
+void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS)
 {
   ShmTDIQChannelData *data = channel->data;
   if (data->is_connected == false) {
+    fprintf(stderr, "Error: Channel is not connected.\n");
     abort();
     return;
   }
@@ -249,13 +251,41 @@ void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp)
   if (current_timestamp >= timestamp) {
     return;
   }
-  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));
+      channel->abort = true;
+      return;
+    }
+    
+    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) {
+        channel->abort = true;
+        fprintf(stderr, "Error: Timed out waiting for samples. Aborting vrtsim\n");\
+        break;
+      } else if (ret != 0) {
+        channel->abort = true;
+        fprintf(stderr, "Error: pthread_cond_timedwait failed: %s\n", strerror(ret));
+        break;
+      } else {
+        current_timestamp = data->timestamp;
+      }
+    }
+    mutexunlock(data->mutex);
   }
-  mutexunlock(data->mutex);
-  return;
 }
 
 uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel)
@@ -269,15 +299,32 @@ bool shm_td_iq_channel_is_connected(const ShmTDIQChannel *channel)
   return channel->data->is_connected;
 }
 
+void shm_td_iq_channel_abort(ShmTDIQChannel *channel)
+{
+  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)
 {
   ShmTDIQChannelData *data = channel->data;
   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) {
+    data->is_connected = false;
+    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,8 +118,10 @@ 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.
  */
-void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp);
+void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp, uint64_t timeout_uS);
 
 /**
  * @brief Checks if the IQ channel is connected.
@@ -129,6 +131,21 @@ void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp);
  */
 bool shm_td_iq_channel_is_connected(const ShmTDIQChannel *channel);
 
+/**
+ * @brief Aborts the IQ channel causing the wait to return immediately
+ *
+ * @param channel The ShmTDIQChannel structure.
+ */
+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

@@ -104,7 +104,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];
@@ -141,7 +141,7 @@ int client(void)
   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;

radio/vrtsim/vrtsim.c

@@ -558,7 +558,15 @@ static int vrtsim_write(openair0_device *device, openair0_timestamp timestamp, v
 static int vrtsim_read(openair0_device *device, openair0_timestamp *ptimestamp, void **samplesVoid, int nsamps, int nbAnt)
 {
   vrtsim_state_t *vrtsim_state = (vrtsim_state_t *)device->priv;
-  shm_td_iq_channel_wait(vrtsim_state->channel, vrtsim_state->last_received_sample + nsamps);
+  if (shm_td_iq_channel_is_aborted(vrtsim_state->channel)) {
+    LOG_E(HW, "Channel is aborted, returning void samples\n");
+    for (int i = 0; i < nbAnt; i++) {
+      memset(samplesVoid[i], 0, sizeof(c16_t) * nsamps);
+    }
+    return nsamps;
+  }
+  uint64_t timeout_uS = 2 * 1000 * 1000 / vrtsim_state->timescale; // 2 seconds in uS
+  shm_td_iq_channel_wait(vrtsim_state->channel, vrtsim_state->last_received_sample + nsamps, timeout_uS);
   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;
@@ -599,8 +607,7 @@ static void vrtsim_end(openair0_device *device)
       taps_client_stop();
     }
   }
-  // produce 1 second of extra samples so threads can finish
-  shm_td_iq_channel_produce_samples(vrtsim_state->channel, vrtsim_state->sample_rate);
+  shm_td_iq_channel_abort(vrtsim_state->channel);
   sleep(1);
   shm_td_iq_channel_destroy(vrtsim_state->channel);