diff --git a/common/utils/shm_iq_channel/shm_td_iq_channel.c b/common/utils/shm_iq_channel/shm_td_iq_channel.c index 3920fe027f..1086979a75 100644 --- a/common/utils/shm_iq_channel/shm_td_iq_channel.c +++ b/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); } diff --git a/common/utils/shm_iq_channel/shm_td_iq_channel.h b/common/utils/shm_iq_channel/shm_td_iq_channel.h index 7c5a730914..30c7cd73f1 100644 --- a/common/utils/shm_iq_channel/shm_td_iq_channel.h +++ b/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. * diff --git a/common/utils/shm_iq_channel/tests/test_shm_td_iq_channel.c b/common/utils/shm_iq_channel/tests/test_shm_td_iq_channel.c index f5170b8144..351bdd678d 100644 --- a/common/utils/shm_iq_channel/tests/test_shm_td_iq_channel.c +++ b/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; diff --git a/radio/vrtsim/vrtsim.c b/radio/vrtsim/vrtsim.c index 9d498d450b..7fb85589ea 100644 --- a/radio/vrtsim/vrtsim.c +++ b/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);