Shared memory frequency domain bidirectional IQ channel

This implements a server-client bidirectional frequency domain IQ
channel via shared memory.

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

Server shall be the time source and should call shm_iq_channel_produce_symbols
at slot intervals.

Future work:
 - Allow to read/write at offsets other than slot boundary
 - Add time domain interface
 - align memory to 512 bits for AVX, possibly add strides to the data in
   symbol dimension depending on application needs.

common/utils/CMakeLists.txt

@@ -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

@@ -0,0 +1,6 @@
+add_library(shm_iq_channel shm_iq_channel.c)
+target_include_directories(shm_iq_channel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
+
+if (ENABLE_TESTS)
+  add_subdirectory(tests)
+endif()

common/utils/shm_iq_channel/shm_iq_channel.c

@@ -0,0 +1,259 @@
+/*
+ * 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_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 "assertions.h"
+
+#define SYMBOLS_IN_SLOT 14
+
+static size_t calculate_buffer_size(int num_sc, int num_slots, int num_ant)
+{
+  size_t buffer_size = (num_sc * num_slots * SYMBOLS_IN_SLOT * num_ant) * sizeof(int32_t);
+  return buffer_size;
+}
+
+static size_t calculate_total_size(ShmIQChannelData *data)
+{
+  return calculate_buffer_size(data->num_sc, data->number_of_slots_in_buffer, data->num_antennas_rx)
+         + calculate_buffer_size(data->num_sc, data->number_of_slots_in_buffer, data->num_antennas_tx) + sizeof(ShmIQChannelData);
+}
+
+ShmIQChannel *shm_iq_channel_create(const char *name, int num_sc, int number_of_slots_in_buffer, int num_tx_ant, int num_rx_ant)
+{
+  ShmIQChannel *channel = malloc(sizeof(ShmIQChannel));
+  strncpy(channel->name, name, sizeof(channel->name) - 1);
+  // Create shared memory segment
+  int fd = shm_open(name, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
+  if (fd == -1) {
+    perror("shm_open");
+    exit(1);
+  }
+  size_t tx_buffer_size = calculate_buffer_size(num_sc, number_of_slots_in_buffer, num_rx_ant);
+  size_t total_size =
+      tx_buffer_size + calculate_buffer_size(num_sc, number_of_slots_in_buffer, num_tx_ant) + sizeof(ShmIQChannelData);
+
+  // Set the size of the shared memory segment
+  int res = ftruncate(fd, total_size);
+  if (res == -1) {
+    perror("ftruncate");
+    exit(1);
+  }
+
+  // Map shared memory segment to address space
+  ShmIQChannelData *shm_ptr = mmap(0, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+  if (shm_ptr == MAP_FAILED) {
+    perror("mmap");
+    exit(1);
+  }
+
+  // Initialize shared memory
+  memset(shm_ptr, 0, total_size);
+  shm_ptr->num_sc = num_sc;
+  shm_ptr->number_of_slots_in_buffer = number_of_slots_in_buffer;
+  shm_ptr->num_antennas_tx = num_tx_ant;
+  shm_ptr->num_antennas_rx = num_rx_ant;
+  shm_ptr->current_symbol_index = 0;
+  shm_ptr->is_connected = false;
+  channel->tx_iq_data = (int32_t *)(shm_ptr + 1);
+  channel->rx_iq_data = channel->tx_iq_data + tx_buffer_size / sizeof(int32_t);
+  channel->data = shm_ptr;
+  channel->type = IQ_CHANNEL_TYPE_SERVER;
+  channel->last_read_symbol_index = 0;
+  pthread_mutexattr_t mutex_attr;
+  pthread_condattr_t cond_attr;
+  pthread_mutexattr_init(&mutex_attr);
+  pthread_condattr_init(&cond_attr);
+  pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED);
+  pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED);
+  pthread_mutex_init(&shm_ptr->mutex, &mutex_attr);
+  pthread_cond_init(&shm_ptr->cond, &cond_attr);
+  shm_ptr->magic = SHM_MAGIC_NUMBER;
+  return channel;
+}
+
+ShmIQChannel *shm_iq_channel_connect(const char *name, int timeout_in_seconds)
+{
+  ShmIQChannel *channel = malloc(sizeof(ShmIQChannel));
+  // 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);
+  }
+  if (fd == -1) {
+    perror("shm_open");
+    exit(1);
+  }
+  struct stat buf;
+  fstat(fd, &buf);
+  size_t total_size = buf.st_size;
+
+  // Map shared memory segment to address space
+  ShmIQChannelData *shm_ptr = mmap(0, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+  if (shm_ptr == MAP_FAILED) {
+    perror("mmap");
+    exit(1);
+  }
+
+  channel->data = shm_ptr;
+  channel->tx_iq_data = (int32_t *)(shm_ptr + 1);
+  size_t tx_buffer_size = calculate_buffer_size(shm_ptr->num_sc, shm_ptr->number_of_slots_in_buffer, shm_ptr->num_antennas_rx);
+  channel->rx_iq_data = channel->tx_iq_data + tx_buffer_size / sizeof(int32_t);
+  channel->type = IQ_CHANNEL_TYPE_CLIENT;
+  channel->last_read_symbol_index = 0;
+  while (shm_ptr->magic != SHM_MAGIC_NUMBER) {
+    printf("Waiting for server to initialize shared memory\n");
+    sleep(1);
+  }
+  shm_ptr->is_connected = true;
+  return channel;
+}
+
+int32_t *shm_iq_channel_get_slot_tx_ptr(ShmIQChannel *channel, uint64_t timestamp, int antenna)
+{
+  ShmIQChannelData *data = channel->data;
+  if (data->is_connected == false) {
+    return NULL;
+  }
+  // timestamp in the past
+  uint64_t current_symbol_index = data->current_symbol_index;
+  if (timestamp < current_symbol_index) {
+    return NULL;
+  }
+  // Not reading at slot boundary
+  if (timestamp % SYMBOLS_IN_SLOT != 0) {
+    return NULL;
+  }
+  // timestamp is too far in the future
+  if (timestamp - current_symbol_index >= data->number_of_slots_in_buffer * SYMBOLS_IN_SLOT) {
+    return NULL;
+  }
+
+  // Data layout: SC x 14 symbols x Antennas x num_slots_in_buffer.
+  // This means contiguous slot buffer for each antenna
+  int slot_index = timestamp / SYMBOLS_IN_SLOT;
+  int slot_size = data->num_sc * SYMBOLS_IN_SLOT;
+  int slot_index_in_buffer = slot_index % data->number_of_slots_in_buffer;
+
+  int antbuf_size;
+  int32_t *base_ptr;
+  if (channel->type == IQ_CHANNEL_TYPE_CLIENT) {
+    antbuf_size = slot_size * data->num_antennas_tx;
+    base_ptr = channel->rx_iq_data;
+  } else {
+    antbuf_size = slot_size * data->num_antennas_rx;
+    base_ptr = channel->tx_iq_data;
+  }
+
+  int offset = slot_index_in_buffer * antbuf_size + antenna * slot_size;
+
+  return base_ptr + offset;
+}
+
+const int32_t *shm_iq_channel_get_slot_rx_ptr(ShmIQChannel *channel, uint64_t timestamp, int antenna)
+{
+  ShmIQChannelData *data = channel->data;
+  if (data->is_connected == false) {
+    return NULL;
+  }
+  // timestamp in the future
+  uint64_t current_symbol_index = data->current_symbol_index;
+  if (timestamp > current_symbol_index) {
+    return NULL;
+  }
+  // Not reading at slot boundary
+  if (timestamp % SYMBOLS_IN_SLOT != 0) {
+    return NULL;
+  }
+  // timestamp is too far in the past
+  if (current_symbol_index - timestamp >= data->number_of_slots_in_buffer * SYMBOLS_IN_SLOT) {
+    return NULL;
+  }
+
+  // Data layout: SC x 14 symbols x Antennas x num_slots_in_buffer.
+  // This means contiguous slot buffer for each antenna
+  int slot_index = timestamp / SYMBOLS_IN_SLOT;
+  int slot_size = data->num_sc * SYMBOLS_IN_SLOT;
+  int slot_index_in_buffer = slot_index % data->number_of_slots_in_buffer;
+
+  int antbuf_size;
+  int32_t *base_ptr;
+  if (channel->type == IQ_CHANNEL_TYPE_CLIENT) {
+    antbuf_size = slot_size * data->num_antennas_rx;
+    base_ptr = channel->tx_iq_data;
+  } else {
+    antbuf_size = slot_size * data->num_antennas_tx;
+    base_ptr = channel->rx_iq_data;
+  }
+
+  int offset = slot_index_in_buffer * antbuf_size + antenna * slot_size;
+
+  return base_ptr + offset;
+}
+
+void shm_iq_channel_produce_symbols(ShmIQChannel *channel, int num_symbols)
+{
+  ShmIQChannelData *data = channel->data;
+  if (channel->type != IQ_CHANNEL_TYPE_SERVER) {
+    return;
+  }
+  if (data->is_connected == false) {
+    return;
+  }
+  data->current_symbol_index += num_symbols;
+}
+
+size_t shm_iq_channel_symbols_ready(ShmIQChannel *channel)
+{
+  ShmIQChannelData *data = channel->data;
+  if (data->is_connected == false) {
+    return 0;
+  }
+  uint64_t current_symbol_index = data->current_symbol_index;
+  uint64_t diff = current_symbol_index - channel->last_read_symbol_index;
+  channel->last_read_symbol_index = current_symbol_index;
+  return diff;
+}
+
+bool shm_iq_channel_is_connected(ShmIQChannel *channel)
+{
+  return channel->data->is_connected;
+}
+
+void shm_iq_channel_destroy(ShmIQChannel *channel)
+{
+  ShmIQChannelData *data = channel->data;
+  munmap(data, calculate_total_size(data));
+  if (channel->type == IQ_CHANNEL_TYPE_SERVER) {
+    shm_unlink(channel->name);
+  }
+  free(channel);
+}

common/utils/shm_iq_channel/shm_iq_channel.h

@@ -0,0 +1,139 @@
+/*
+ * 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
+
+/**
+ * ShmIqChannel is a shared memory bidirectional 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, simply write to the pointer returned by shm_iq_channel_get_slot_tx_ptr
+ * To read samples, read IQ data from the pointer returned by shm_iq_channel_get_slot_rx_ptr
+ * To indicate that samples are ready to be read by the client, call shm_iq_channel_produce_symbols
+ * (server only)
+ *
+ * The timestamps used in the API are in symbols from the start of the channel
+ */
+
+typedef enum IQChannelType { IQ_CHANNEL_TYPE_SERVER, IQ_CHANNEL_TYPE_CLIENT } IQChannelType;
+
+typedef struct {
+  int magic;
+  int num_sc;
+  int number_of_slots_in_buffer;
+  int num_antennas_tx;
+  int num_antennas_rx;
+  uint64_t current_symbol_index;
+  bool is_connected;
+  pthread_mutex_t mutex;
+  pthread_cond_t cond;
+} ShmIQChannelData;
+
+typedef struct {
+  IQChannelType type;
+  ShmIQChannelData *data;
+  uint64_t last_read_symbol_index;
+  char name[256];
+  int32_t *tx_iq_data;
+  int32_t *rx_iq_data;
+} ShmIQChannel;
+
+/**
+ * @brief Creates a shared memory IQ channel.
+ *
+ * @param name The name of the shared memory segment.
+ * @param num_sc The number of subcarriers.
+ * @param number_of_slots_in_buffer The number of slots in the buffer.
+ * @param num_tx_ant The number of TX antennas.
+ * @param num_rx_ant The number of RX antennas.
+ * @return A pointer to the created ShmIQChannel structure.
+ */
+ShmIQChannel *shm_iq_channel_create(const char *name, int num_sc, int number_of_slots_in_buffer, 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 ShmIQChannel structure.
+ */
+ShmIQChannel *shm_iq_channel_connect(const char *name, int timeout_in_seconds);
+
+/**
+ * @brief Gets the pointer to the TX IQ data slot for a given timestamp and antenna.
+ *
+ * @param channel The ShmIQChannel structure.
+ * @param timestamp The timestamp for which to get the TX IQ data slot.
+ * @param antenna The antenna index.
+ * @return A pointer to the TX IQ data slot.
+ */
+int32_t *shm_iq_channel_get_slot_tx_ptr(ShmIQChannel *channel, uint64_t timestamp, int antenna);
+
+/**
+ * @brief Gets the pointer to the RX IQ data slot for a given timestamp and antenna.
+ *
+ * @param channel The ShmIQChannel structure.
+ * @param timestamp The timestamp for which to get the RX IQ data slot.
+ * @param antenna The antenna index.
+ * @return A pointer to the RX IQ data slot.
+ */
+const int32_t *shm_iq_channel_get_slot_rx_ptr(ShmIQChannel *channel, uint64_t timestamp, int antenna);
+
+/**
+ * @brief Produces a specified number of symbols in the IQ channel.
+ *
+ * @param channel The ShmIQChannel structure.
+ * @param num_symbols The number of symbols to produce.
+ */
+void shm_iq_channel_produce_symbols(ShmIQChannel *channel, int num_symbols);
+
+/**
+ * @brief Gets the number of symbols ready in the IQ channel.
+ *
+ * @param channel The ShmIQChannel structure.
+ * @return The number of symbols ready.
+ */
+size_t shm_iq_channel_symbols_ready(ShmIQChannel *channel);
+
+/**
+ * @brief Checks if the IQ channel is connected.
+ *
+ * @param channel The ShmIQChannel structure.
+ * @return True if the channel is connected, false otherwise.
+ */
+bool shm_iq_channel_is_connected(ShmIQChannel *channel);
+
+/**
+ * @brief Destroys the shared memory IQ channel.
+ *
+ * @param channel The ShmIQChannel structure.
+ */
+void shm_iq_channel_destroy(ShmIQChannel *channel);
+
+#endif

common/utils/shm_iq_channel/tests/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_executable(test_shm_iq_channel test_shm_iq_channel.c)
+target_link_libraries(test_shm_iq_channel shm_iq_channel minimal_lib)
+add_dependencies(tests test_shm_iq_channel)
+add_test(test_shm_iq_channel ./test_shm_iq_channel)

common/utils/shm_iq_channel/tests/test_shm_iq_channel.c

@@ -0,0 +1,191 @@
+/*
+ * 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 <threads.h>
+
+#define SHM_CHANNEL_NAME "shm_iq_channel_test_file"
+#include "shm_iq_channel.h"
+
+int server(void);
+void echo_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;
+    }
+  }
+  switch (mode) {
+    case MODE_SERVER:
+      return server();
+      break;
+    case MODE_CLIENT:
+      echo_client();
+      break;
+    case MODE_FORK: {
+      int pid = fork();
+      if (pid == 0) {
+        echo_client();
+      } else {
+        return server();
+      }
+    } break;
+  }
+
+  return 0;
+}
+
+// Test for 50 slots to check wrap around
+const int num_slots = 50;
+int produce_symbols(void *arg)
+{
+  ShmIQChannel *channel = (ShmIQChannel *)arg;
+  for (int i = 0; i < num_slots; i++) {
+    shm_iq_channel_produce_symbols(channel, 14);
+    usleep(20000);
+  }
+  return 0;
+}
+
+int server(void)
+{
+  int number_of_slots_per_frame = 20;
+  int num_ant_tx = 1;
+  int num_ant_rx = 1;
+  int num_sc = 12;
+  ShmIQChannel *channel = shm_iq_channel_create(SHM_CHANNEL_NAME, num_sc, number_of_slots_per_frame, num_ant_tx, num_ant_rx);
+  while (true) {
+    if (shm_iq_channel_is_connected(channel)) {
+      printf("Server connected\n");
+      break;
+    }
+    printf("Waiting for client\n");
+    sleep(1);
+  }
+
+  thrd_t producer_thread;
+  if (thrd_create(&producer_thread, produce_symbols, channel) != thrd_success) {
+    fprintf(stderr, "Failed to create producer thread\n");
+    exit(1);
+  }
+
+  int num_total_errors = 0;
+  uint64_t timestamp = 0;
+  int iq_contents = 0;
+  while (timestamp < num_slots * 14) {
+    uint64_t new_symbols = shm_iq_channel_symbols_ready(channel);
+    if (new_symbols > 0) {
+      timestamp += new_symbols;
+      uint64_t target_timestamp = timestamp + 14;
+      int32_t *data = shm_iq_channel_get_slot_tx_ptr(channel, target_timestamp, 0);
+      size_t write_size = num_sc * 14;
+      for (int i = 0; i < write_size; i++) {
+        data[i] = iq_contents;
+      }
+
+      if (timestamp > 14 * 3) {
+        // Read back from client
+        int reference = iq_contents - 3;
+        uint64_t read_timestamp = timestamp - 14;
+        const int32_t *iq_rx = shm_iq_channel_get_slot_rx_ptr(channel, read_timestamp, 0);
+        int num_errors = 0;
+        for (int i = 0; i < channel->data->num_sc * 14; i++) {
+          if (iq_rx[i] != reference) {
+            num_errors++;
+          }
+        }
+        if (num_errors) {
+          printf("Found errors = %d, value = %d, reference = %d\n", num_errors, iq_rx[0], reference);
+        }
+        num_total_errors += num_errors;
+      }
+
+      iq_contents++;
+    }
+    usleep(1000);
+  }
+
+  printf("Finished writing data\n");
+
+  shm_iq_channel_destroy(channel);
+  if (thrd_join(producer_thread, NULL) != thrd_success) {
+    fprintf(stderr, "Failed to join producer thread\n");
+    exit(1);
+  }
+  return num_total_errors;
+}
+
+void echo_client(void)
+{
+  ShmIQChannel *channel = shm_iq_channel_connect(SHM_CHANNEL_NAME, 10);
+  while (true) {
+    if (shm_iq_channel_is_connected(channel)) {
+      printf("Echo client connected\n");
+      break;
+    }
+    printf("Waiting for server\n");
+    sleep(1);
+  }
+
+  uint64_t timestamp = 0;
+  int iq_contents = 0;
+  while (timestamp < num_slots * 14) {
+    uint64_t new_symbols = shm_iq_channel_symbols_ready(channel);
+    if (new_symbols > 0) {
+      timestamp += new_symbols;
+      // Server starts producing from second slot
+      if (timestamp > 28) {
+        uint64_t target_timestamp = timestamp - 14;
+        const int32_t *iq = shm_iq_channel_get_slot_rx_ptr(channel, target_timestamp, 0);
+        int num_errors = 0;
+        for (int i = 0; i < channel->data->num_sc * 14; i++) {
+          if (iq[i] != iq_contents) {
+            num_errors++;
+          }
+        }
+        if (num_errors) {
+          printf("Found %d errors, value = %d, reference = %d\n", num_errors, iq[0], iq_contents);
+        }
+        iq_contents++;
+
+        // Write back to server
+        uint64_t write_timestamp = timestamp + 14;
+        int32_t *iq_tx = shm_iq_channel_get_slot_tx_ptr(channel, write_timestamp, 0);
+        for (int i = 0; i < channel->data->num_sc * 14; i++) {
+          iq_tx[i] = iq_contents;
+        }
+      }
+    }
+    usleep(1000);
+  }
+
+  printf("Finished reading data\n");
+  shm_iq_channel_destroy(channel);
+}