maki/openairinterface5g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Shared memory time domain bidirectional IQ channel

Browse Source 
This implements a server-client bidirectional time domain IQ channel via
shared memory.

The server creates the channel and should be aware of the client
receiver configuration (num rx/tx antennas).

Server shall be the time source and should call shm_iq_channel_produce_samples
at regular intervals
This commit is contained in:
Bartosz Podrygajlo
2025-03-26 14:24:47 +01:00
parent 1d479b2aa9
commit 1d1814f8b3
6 changed files with 615 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/CMakeLists.txt
+1
View File
@@ -21,3 +21,4 @@ if (ENABLE_TESTS)
endif()
add_subdirectory(barrier)
add_subdirectory(actor)
add_subdirectory(shm_iq_channel)
common/utils/shm_iq_channel/CMakeLists.txt
+6
View File
@@ -0,0 +1,6 @@
add_library(shm_td_iq_channel shm_td_iq_channel.c)
target_include_directories(shm_td_iq_channel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
if (ENABLE_TESTS)
add_subdirectory(tests)
endif()
common/utils/shm_iq_channel/shm_td_iq_channel.c
+286
View File
@@ -0,0 +1,286 @@
/*
* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The OpenAirInterface Software Alliance licenses this file to You under
* the OAI Public License, Version 1.1 (the "License"); you may not use this file
* except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.openairinterface.org/?page_id=698
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*-------------------------------------------------------------------------------
* For more information about the OpenAirInterface (OAI) Software Alliance:
* contact@openairinterface.org
*/
#include "shm_td_iq_channel.h"
#include <stdio.h>
#include <sys/mman.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <stdbool.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include "assertions.h"
#include "utils.h"
#include "common/utils/threadPool/pthread_utils.h"
#define CIRCULAR_BUFFER_SIZE (30720 * 14 * 20)
typedef struct {
int magic;
int num_antennas_tx;
int num_antennas_rx;
uint64_t timestamp;
bool is_connected;
pthread_mutex_t mutex;
pthread_cond_t cond;
} ShmTDIQChannelData;
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;
} ShmTDIQChannel;
ShmTDIQChannel *shm_td_iq_channel_create(const char *name, int num_tx_ant, int num_rx_ant)
{
// 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 total_size = sizeof(ShmTDIQChannelData) + tx_buffer_size + rx_buffer_size;
// Set the size of the shared memory segment
int res = ftruncate(fd, total_size);
AssertFatal(res != -1, "ftruncate failed: %s\n", strerror(errno));
// Map shared memory segment to address space
ShmTDIQChannelData *shm_ptr = mmap(NULL, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
AssertFatal(shm_ptr != MAP_FAILED, "mmap failed: %s\n", strerror(errno));
// Initialize shared memory
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);
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));
shm_ptr->magic = SHM_MAGIC_NUMBER;
close(fd);
return channel;
}
ShmTDIQChannel *shm_td_iq_channel_connect(const char *name, int timeout_in_seconds)
{
// Create shared memory segment
int fd = -1;
while (timeout_in_seconds > 0 && fd == -1) {
fd = shm_open(name, O_RDWR, S_IRUSR | S_IWUSR);
timeout_in_seconds--;
printf("Waiting for server to create shared memory segment\n");
sleep(1);
}
AssertFatal(fd != -1, "shm_open() failed: errno %d, %s", errno, strerror(errno));
struct stat buf;
fstat(fd, &buf);
size_t total_size = buf.st_size;
// Map shared memory segment to address space
ShmTDIQChannelData *shm_ptr = mmap(NULL, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
if (shm_ptr == MAP_FAILED) {
perror("mmap");
exit(1);
}
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->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;
}
IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
uint64_t timestamp,
uint64_t num_samples,
int antenna,
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) {
return CHANNEL_ERROR_TOO_LATE;
}
// timestamp is too far in the future
if (timestamp - current_time + num_samples >= CIRCULAR_BUFFER_SIZE) {
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;
}
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(base_ptr + first_sample, tx_iq_data, num_samples_first_copy * sizeof(sample_t));
memcpy(base_ptr, tx_iq_data + num_samples_first_copy, (num_samples - num_samples_first_copy) * sizeof(sample_t));
} else {
memcpy(base_ptr + first_sample, tx_iq_data, num_samples * sizeof(sample_t));
}
return CHANNEL_NO_ERROR;
}
IQChannelErrorType shm_td_iq_channel_rx(ShmTDIQChannel *channel,
uint64_t timestamp,
uint64_t num_samples,
int antenna,
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) {
return CHANNEL_ERROR_TOO_EARLY;
}
// timestamp is too far in the past
if (current_time - timestamp >= CIRCULAR_BUFFER_SIZE) {
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;
}
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));
} else {
memcpy(tx_iq_data, base_ptr + first_sample, num_samples * sizeof(sample_t));
}
return CHANNEL_NO_ERROR;
}
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)
{
ShmTDIQChannelData *data = channel->data;
if (data->is_connected == false) {
abort();
return;
}
size_t current_timestamp = data->timestamp;
if (current_timestamp >= timestamp) {
return;
}
mutexlock(data->mutex);
while (current_timestamp < timestamp) {
condwait(data->cond, data->mutex);
current_timestamp = data->timestamp;
}
mutexunlock(data->mutex);
return;
}
uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel)
{
ShmTDIQChannelData *data = channel->data;
return data->timestamp;
}
bool shm_td_iq_channel_is_connected(const ShmTDIQChannel *channel)
{
return channel->data->is_connected;
}
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) {
shm_unlink(channel->name);
}
free(channel);
}
common/utils/shm_iq_channel/shm_td_iq_channel.h
+148
View File
@@ -0,0 +1,148 @@
/*
* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The OpenAirInterface Software Alliance licenses this file to You under
* the OAI Public License, Version 1.1 (the "License"); you may not use this file
* except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.openairinterface.org/?page_id=698
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*-------------------------------------------------------------------------------
* For more information about the OpenAirInterface (OAI) Software Alliance:
* contact@openairinterface.org
*/
#ifndef SHM_IQ_CHANNEL_H
#define SHM_IQ_CHANNEL_H
#include "../threadPool/pthread_utils.h"
#include <stdint.h>
#include <stdbool.h>
#include <semaphore.h>
#define SHM_MAGIC_NUMBER 0x12345678
/**
* ShmTdIqChannel is a shared memory bidirectional time domain IQ channel with a single clock
* source. The server (clock source) shall create the channel while the client should connect to
* it.
*
* To write samples, send them via a pointer passed to shm_td_iq_channel_tx.
* To read samples, receive them via a pointer passed to shm_td_iq_channel_rx. You can either wait until
* the sample are available using shm_td_iq_channel_wait or poll using shm_td_iq_channel_rx until the return
* value is not CHANNEL_ERROR_TOO_EARLY.
* To indicate that samples are ready to be read by the client, call shm_td_iq_channel_produce_samples
* (server only).
*
* The timestamps used in the API are in absolute samples from the time the client/server connection started.
*/
typedef enum IQChannelType { IQ_CHANNEL_TYPE_SERVER, IQ_CHANNEL_TYPE_CLIENT } IQChannelType;
typedef uint32_t sample_t;
typedef enum IQChannelErrorType {
CHANNEL_NO_ERROR,
CHANNEL_ERROR_TOO_LATE,
CHANNEL_ERROR_TOO_EARLY,
CHANNEL_ERROR_NOT_CONNECTED
} IQChannelErrorType;
typedef struct ShmTDIQChannel_s ShmTDIQChannel;
/**
* @brief Creates a shared memory IQ channel.
*
* @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.
* @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);
/**
* @brief Connects to an existing shared memory IQ channel.
*
* @param name The name of the shared memory segment.
* @param timeout_in_seconds The timeout in seconds for the connection attempt.
* @return A pointer to the connected ShmTDIQChannel structure.
*/
ShmTDIQChannel *shm_td_iq_channel_connect(const char *name, int timeout_in_seconds);
/**
* @brief Transmit data in the channel
*
* @param channel The ShmTDIQChannel structure.
* @param timestamp The timestamp for which to get the TX IQ data slot.
* @param num_samples The number of samples to write.
* @param antenna The antenna index.
* @param tx_iq_data The TX IQ data to write.
*
* @return CHANNEL_NO_ERROR if successful, error type otherwise
*/
IQChannelErrorType shm_td_iq_channel_tx(ShmTDIQChannel *channel,
uint64_t timestamp,
uint64_t num_samples,
int antenna,
const sample_t *tx_iq_data);
/**
* @brief Receive iq data from the channel
*
* @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 tx_iq_data pointer to the RX IQ data slot.
* @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);
/**
* @brief Advances the time in the channel by specified number of samples
*
* @param channel The ShmTDIQChannel structure.
* @param num_symbols The number of samples to produce.
*/
void shm_td_iq_channel_produce_samples(ShmTDIQChannel *channel, uint64_t num_samples);
/**
* @brief Wait until sample at the specified timestamp is available
*
* @param channel The ShmTDIQChannel structure.
*/
void shm_td_iq_channel_wait(ShmTDIQChannel *channel, uint64_t timestamp);
/**
* @brief Checks if the IQ channel is connected.
*
* @param channel The ShmTDIQChannel structure.
* @return True if the channel is connected, false otherwise.
*/
bool shm_td_iq_channel_is_connected(const ShmTDIQChannel *channel);
/**
* @brief Destroys the shared memory IQ channel.
*
* @param channel The ShmTDIQChannel structure.
*/
void shm_td_iq_channel_destroy(ShmTDIQChannel *channel);
/**
* @brief Returns current sample
*
* @param channel The ShmTDIQChannel structure.
*
* @return Current time as sample count since beginning of transmission
*/
uint64_t shm_td_iq_channel_get_current_sample(const ShmTDIQChannel *channel);
#endif
common/utils/shm_iq_channel/tests/CMakeLists.txt
+5
View File
@@ -0,0 +1,5 @@
add_executable(test_shm_td_iq_channel test_shm_td_iq_channel.c)
target_link_libraries(test_shm_td_iq_channel shm_td_iq_channel minimal_lib pthread)
add_dependencies(tests test_shm_td_iq_channel)
# Commented out due to issues with ASAN in the unit test docker container
#add_test(test_shm_td_iq_channel ./test_shm_td_iq_channel)
common/utils/shm_iq_channel/tests/test_shm_td_iq_channel.c
+169
View File
@@ -0,0 +1,169 @@
/*
* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The OpenAirInterface Software Alliance licenses this file to You under
* the OAI Public License, Version 1.1 (the "License"); you may not use this file
* except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.openairinterface.org/?page_id=698
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*-------------------------------------------------------------------------------
* For more information about the OpenAirInterface (OAI) Software Alliance:
* contact@openairinterface.org
*/
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include "assertions.h"
#define SHM_CHANNEL_NAME "shm_iq_channel_test_file"
#include "shm_td_iq_channel.h"
void server(void);
int client(void);
enum { MODE_SERVER, MODE_CLIENT, MODE_FORK };
int main(int argc, char *argv[])
{
int mode = MODE_FORK;
if (argc == 2) {
if (strcmp(argv[1], "server") == 0) {
mode = MODE_SERVER;
} else if (strcmp(argv[1], "client") == 0) {
mode = MODE_CLIENT;
}
} else if (argc > 2) {
printf("Usage: %s [server|client]\n", argv[0]);
return 1;
}
switch (mode) {
case MODE_SERVER:
server();
break;
case MODE_CLIENT:
return client();
break;
case MODE_FORK: {
char file[256];
snprintf(file, sizeof(file), "/dev/shm/%s", SHM_CHANNEL_NAME);
remove(file);
int pid = fork();
if (pid == 0) {
server();
} else {
return client();
}
} break;
}
return 0;
}
const int num_updates = 50;
const int num_samples_per_update = 30720;
void *produce_symbols(void *arg)
{
ShmTDIQChannel *channel = (ShmTDIQChannel *)arg;
for (int i = 0; i < num_updates; i++) {
shm_td_iq_channel_produce_samples(channel, num_samples_per_update);
usleep(20000);
}
return 0;
}
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);
AssertFatal(ret == 0, "pthread_create() failed: errno %d, %s\n", errno, strerror(errno));
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);
uint64_t target_timestamp = timestamp + num_samples_per_update;
timestamp += num_samples_per_update;
uint32_t iq_data[num_samples_per_update];
for (int i = 0; i < num_samples_per_update; i++) {
iq_data[i] = iq_contents;
}
int result = shm_td_iq_channel_tx(channel, target_timestamp, num_samples_per_update, 0, iq_data);
AssertFatal(result == CHANNEL_NO_ERROR, "Failed to write data\n");
iq_contents++;
}
printf("Finished writing data\n");
ret = pthread_join(producer_thread, NULL);
AssertFatal(ret == 0, "pthread_join() failed: errno %d, %s\n", errno, strerror(errno));
shm_td_iq_channel_destroy(channel);
}
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);
// Server starts producing from second slot
if (timestamp > num_samples_per_update) {
uint64_t target_timestamp = timestamp - num_samples_per_update;
uint32_t iq_data[num_samples_per_update];
int result = shm_td_iq_channel_rx(channel, target_timestamp, num_samples_per_update, 0, iq_data);
AssertFatal(result == CHANNEL_NO_ERROR, "Failed to read data\n");
int num_errors = 0;
for (int i = 0; i < num_samples_per_update; i++) {
if (iq_data[i] != iq_contents) {
num_errors++;
}
}
if (num_errors) {
printf("Found %d errors, value = %d, reference = %d\n", num_errors, iq_data[0], iq_contents);
total_errors += num_errors;
}
iq_contents++;
}
timestamp += num_samples_per_update;
}
printf("Finished reading data\n");
shm_td_iq_channel_destroy(channel);
return total_errors;
}