RU comments

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)
@@ -228,9 +229,6 @@ 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;
   }
@@ -240,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;
   }
@@ -252,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)
@@ -272,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;

executables/nr-ru.c

@@ -1252,6 +1252,7 @@ void *ru_thread(void *param)
     // synchronization on input FH interface, acquire signals/data and block
     LOG_D(PHY,"[RU_thread] read data: frame_rx = %d, tti_rx = %d\n", frame, slot);
 
+    /// BPODRYGAJLO: RU read IQ data. Get frame and slot from here every slot
     if (ru->fh_south_in) ru->fh_south_in(ru,&frame,&slot);
     else AssertFatal(1==0, "No fronthaul interface at south port");
 
@@ -1291,6 +1292,7 @@ void *ru_thread(void *param)
       if (!wait_free_rx_tti(&gNB->L1_rx_out, rx_tti_busy, proc->frame_rx, proc->tti_rx))
         break; // nothing to wait for: we have to stop
       if (ru->feprx) {
+        /// BPODRYGAJLO: Front end processing (most likely dft)
         ru->feprx(ru,proc->tti_rx);
         LOG_D(NR_PHY, "Setting %d.%d (%d) to busy\n", proc->frame_rx, proc->tti_rx, proc->tti_rx % RU_RX_SLOT_DEPTH);
         //LOG_M("rxdata.m","rxs",ru->common.rxdata[0],1228800,1,1);
@@ -1303,7 +1305,7 @@ void *ru_thread(void *param)
                      gNB->frame_parms.samples_per_slot_wCP,
                      proc->tti_rx * gNB->frame_parms.samples_per_slot_wCP);
 
-        // Do PRACH RU processing
+        // Do PRACH RU processing BPODRYGAJLO: ONLY IF CONFIGURED
         int prach_id = find_nr_prach_ru(ru, proc->frame_rx, proc->tti_rx, SEARCH_EXIST);
         if (prach_id >= 0) {
           VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_RU_PRACH_RX, 1 );
@@ -1337,6 +1339,7 @@ void *ru_thread(void *param)
       } // end if (ru->feprx)
     } // end if (slot_type == NR_UPLINK_SLOT || slot_type == NR_MIXED_SLOT) {
 
+    // BPODRYGAJLO: This is supposed to trigger TX eventually
     notifiedFIFO_elt_t *resTx = newNotifiedFIFO_elt(sizeof(processingData_L1tx_t), 0, &gNB->L1_tx_out, NULL);
     processingData_L1tx_t *syncMsgTx = NotifiedFifoData(resTx);
     syncMsgTx->gNB = gNB;

openair1/SCHED_NR/nr_ru_procedures.c

@@ -385,6 +385,8 @@ void nr_fep_tp(RU_t *ru, int slot) {
         feprx_cmd->endSymbol = ru->nr_frame_parms->symbols_per_slot - 1;
 
         task_t t = {.func = nr_fep, .args = feprx_cmd};
+
+        // start thread BPODRYGAJLO: This is the thread pool for FEP RX
         pushTpool(ru->threadPool, t);
 
         nbfeprx++;

openair1/SIMULATION/TOOLS/sim.h

@@ -77,6 +77,8 @@ typedef struct {
   struct complexd **a;
   ///interpolated (sample-spaced) channel impulse response. size(ch) = (n_tx * n_rx) * channel_length. ATTENTION: the dimensions of ch are the transposed ones of a. This is to allow the use of BLAS when applying the correlation matrices to the state.
   struct complexd **ch;
+  ///Same as above but single precision
+  struct complexf **ch_ps;
   ///Sampled frequency response (90 kHz resolution)
   struct complexd **chF;
   ///Maximum path delay in mus.

radio/vrtsim/CMakeLists.txt

@@ -1,4 +1,17 @@
 add_library(vrtsim MODULE vrtsim.c noise_device.c)
 set_target_properties(vrtsim PROPERTIES LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
-target_link_libraries(vrtsim PRIVATE SIMU shm_td_iq_channel actor)
+target_link_libraries(vrtsim PRIVATE SIMU shm_td_iq_channel actor taps_client)
 add_dependencies(vrtsim generate_T)
+set(TAPS_API_HEADER taps_generated.h)
+
+add_custom_command(OUTPUT ${TAPS_API_HEADER}
+  COMMAND flatc -c ${CMAKE_CURRENT_SOURCE_DIR}/taps.fbs
+  DEPENDS taps.fbs
+  COMMENT "Generating flatbuffers API from ${TAPS_IF_PATH}"
+)
+add_library(taps_api INTERFACE)
+target_include_directories(taps_api INTERFACE ${CMAKE_CURRENT_BINARY_DIR})
+
+add_library(taps_client taps_client.cpp ${TAPS_API_HEADER})
+target_include_directories(taps_client PUBLIC .)
+target_link_libraries(taps_client PUBLIC flatbuffers taps_api SIMU nanomsg)

radio/vrtsim/noise_device.c

@@ -18,6 +18,7 @@
  * For more information about the OpenAirInterface (OAI) Software Alliance:
  *      contact@openairinterface.org
  */
+#include "assertions.h"
 #include <pthread.h>
 #include <errno.h>
 #include "noise_device.h"
@@ -74,7 +75,8 @@ void init_noise_device(float noise_power)
 void free_noise_device(void)
 {
   state.running = false;
-  pthread_join(state.noise_device_thread, NULL);
+  int ret = pthread_join(state.noise_device_thread, NULL);
+  AssertFatal(ret == 0, "pthread_join failed: %d, errno %d (%s)\n", ret, errno, strerror(errno));
 }
 
 void get_noise_vector(float *noise_vector, int length)

radio/vrtsim/taps.fbs

@@ -0,0 +1,14 @@
+// This file defines the structure of the Taps table used in the PHY layer.
+namespace Phy;
+
+// Taps form the discrete complex baseband-equivalent channel impulse response and can be used
+// to model the channel characteristics in link layer simulations.
+table Taps {
+    id:uint;
+    num_tx_antennas:uint;
+    num_rx_antennas:uint;
+    taps_len:uint;
+    taps:[float];
+}
+
+root_type Taps;

radio/vrtsim/taps_client.cpp

@@ -0,0 +1,194 @@
+/*
+ * 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 <nanomsg/nn.h>
+#include <nanomsg/pubsub.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+#include "pthread.h"
+#include "taps_generated.h"
+#include "SIMULATION/TOOLS/sim.h"
+extern "C" {
+#include "assertions.h"
+#include "common/utils/LOG/log.h"
+#include "sim.h"
+#include "utils.h"
+}
+#include <cfloat>
+
+#define NUM_TAPS_BUFFERS 4
+#define MAX_TAPS_LEN 100
+#define MAX_TAPS_MSG_SIZE (sizeof(struct complexf) * MAX_TAPS_LEN * 4 * 4 + 20)
+
+static pthread_t client_thread;
+static bool should_run = true;
+typedef struct {
+  int id;
+  int sock;
+  uint32_t num_tx_antennas;
+  uint32_t num_rx_antennas;
+  channel_desc_t **channel_desc;
+} client_thread_args_t;
+
+typedef struct {
+  void *taps_msg;
+  channel_desc_t *channel_desc;
+} taps_buffer_t;
+
+typedef struct {
+  taps_buffer_t taps_buffers[NUM_TAPS_BUFFERS];
+  int current_buffer;
+} taps_storage_t;
+
+void ascii_line_plot(const float *data, size_t size, char* buffer) {
+  const char *levels = "_.-=#"; // ASCII characters for different levels
+  size_t num_levels = strlen(levels) - 1; // Number of levels (excluding the null terminator)
+
+  // Calculate min and max values from the data
+  float min_val = FLT_MAX;
+  float max_val = FLT_MIN;
+  for (size_t i = 0; i < size; i++) {
+      if (data[i] < min_val) min_val = data[i];
+      if (data[i] > max_val) max_val = data[i];
+  }
+
+  // Handle edge case where all values are the same
+  if (min_val == max_val) {
+      min_val -= 1.0f;
+      max_val += 1.0f;
+  }
+
+  // Normalize and map data to levels
+  for (size_t i = 0; i < size; i++) {
+      float normalized = (data[i] - min_val) / (max_val - min_val); // Normalize to [0, 1]
+      size_t level_index = (size_t)(normalized * num_levels);       // Map to level index
+      if (level_index > num_levels) level_index = num_levels;       // Clamp to valid range
+      snprintf(buffer + i, 2, "%c", levels[level_index]);
+  }
+}
+
+static void init_taps_storage(taps_storage_t *storage, int num_tx_antennas, int num_rx_antennas)
+{
+  for (int i = 0; i < NUM_TAPS_BUFFERS; i++) {
+    storage->taps_buffers[i].taps_msg = calloc_or_fail(1, MAX_TAPS_MSG_SIZE);
+    storage->taps_buffers[i].channel_desc = (channel_desc_t *)calloc_or_fail(1, sizeof(channel_desc_t));
+    storage->taps_buffers[i].channel_desc->ch_ps =
+        (struct complexf **)calloc_or_fail(num_rx_antennas * num_tx_antennas, sizeof(struct complexf *));
+  }
+  storage->current_buffer = 0;
+}
+
+static taps_storage_t taps_storage;
+
+void *client_thread_func(void *args)
+{
+  client_thread_args_t *client_thread_args = (client_thread_args_t *)args;
+  while (should_run) {
+    int next_buffer = (taps_storage.current_buffer + 1) % NUM_TAPS_BUFFERS;
+    taps_buffer_t *taps_buffer = &taps_storage.taps_buffers[next_buffer];
+    int ret = nn_recv(client_thread_args->sock, taps_buffer->taps_msg, MAX_TAPS_MSG_SIZE, 0);
+    if (ret < 0) {
+      LOG_E(HW, "nn_recv() failed: errno: %d, %s\n", errno, strerror(errno));
+      continue;
+    }
+    auto taps_message = Phy::GetTaps(taps_buffer->taps_msg);
+    if (taps_message->num_rx_antennas() != client_thread_args->num_rx_antennas
+        || taps_message->num_tx_antennas() != client_thread_args->num_tx_antennas) {
+      LOG_E(HW,
+            "Number of antennas mismatch: expected %d x %d, got %d x %d\n",
+            client_thread_args->num_rx_antennas,
+            client_thread_args->num_tx_antennas,
+            taps_message->num_rx_antennas(),
+            taps_message->num_tx_antennas());
+      continue;
+    }
+    channel_desc_t *channel_desc = taps_buffer->channel_desc;
+    const flatbuffers::Vector<float> *taps = taps_message->taps();
+    struct complexf *base_pointer = (struct complexf *)taps->data();
+    int taps_len = taps_message->taps_len();
+    AssertFatal(taps_len < MAX_TAPS_LEN, "Too many samples in the taps array, taps_len = %d\n", taps_len);
+    for (unsigned int aarx = 0U; aarx < client_thread_args->num_rx_antennas; aarx++) {
+      for (unsigned int aatx = 0U; aatx < client_thread_args->num_tx_antennas; aatx++) {
+        channel_desc->ch_ps[aarx + (client_thread_args->num_rx_antennas * aatx)] =
+            &base_pointer[(aarx + (client_thread_args->num_rx_antennas * aatx)) * taps_len];
+      }
+    }
+    channel_desc->path_loss_dB = 0;
+    channel_desc->channel_length = taps_len;
+    *client_thread_args->channel_desc = channel_desc;
+    taps_storage.current_buffer = next_buffer;
+    LOG_A(HW, "Receved new taps message, channel_length %d, buffer %d\n", channel_desc->channel_length, next_buffer);
+    for (unsigned int aarx = 0; aarx < client_thread_args->num_rx_antennas; aarx++) {
+      for (unsigned int aatx = 0; aatx < client_thread_args->num_tx_antennas; aatx++) {
+        char buffer[MAX_TAPS_LEN + 1];
+        memset(buffer, 0, sizeof(buffer));
+        float magnitudes[MAX_TAPS_LEN];
+        cf_t *channel = channel_desc->ch_ps[aarx + (client_thread_args->num_rx_antennas * aatx)];
+        for (int i = 0; i < channel_desc->channel_length; i++) {
+          magnitudes[i] = sqrtf(powf(channel[i].r, 2) + powf(channel[i].i, 2));
+        }
+        ascii_line_plot(magnitudes, channel_desc->channel_length, buffer);
+        LOG_A(HW, "Taps message %d, channel %d x %d: %s\n", client_thread_args->id, aarx, aatx, buffer);
+      }
+    }
+  }
+  return NULL;
+}
+
+extern "C" void taps_client_connect(int id,
+                                    const char *socket_path,
+                                    int num_tx_antennas,
+                                    int num_rx_antennas,
+                                    channel_desc_t **channel_desc)
+{
+  // Create a socket
+  int sock = nn_socket(AF_SP, NN_SUB);
+  AssertFatal(sock >= 0, "nn_socket() failed: errno: %d, %s\n", errno, strerror(errno));
+
+  int ret = nn_connect(sock, socket_path);
+  AssertFatal(ret >= 0, "nn_connect() failed: errno: %d, %s\n", errno, strerror(errno));
+
+  // Subscribe to all messages
+  ret = nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, "", 0);
+  AssertFatal(ret == 0, "nn_setsockopt() failed, errno %d, %s\n", errno, strerror(errno));
+
+  init_taps_storage(&taps_storage, num_tx_antennas, num_rx_antennas);
+
+  client_thread_args_t *client_thread_args = static_cast<client_thread_args_t *>(malloc(sizeof(client_thread_args_t)));
+  client_thread_args->id = id;
+  client_thread_args->sock = sock;
+  client_thread_args->num_rx_antennas = num_rx_antennas;
+  client_thread_args->num_tx_antennas = num_tx_antennas;
+  client_thread_args->channel_desc = channel_desc;
+  ret = pthread_create(&client_thread, NULL, client_thread_func, client_thread_args);
+  AssertFatal(ret == 0, "pthread_create() failed: errno: %d, %s\n", errno, strerror(errno));
+}
+
+extern "C" void taps_client_stop()
+{
+  should_run = false;
+  pthread_join(client_thread, NULL);
+  for (int i = 0; i < NUM_TAPS_BUFFERS; i++) {
+    free(taps_storage.taps_buffers[i].taps_msg);
+    free(taps_storage.taps_buffers[i].channel_desc->ch_ps);
+    free(taps_storage.taps_buffers[i].channel_desc);
+  }
+}

radio/vrtsim/taps_client.h

@@ -0,0 +1,38 @@
+/*
+ * 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 TAPS_CLIENT_H
+#define TAPS_CLIENT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "sim.h"
+
+void taps_client_connect(int id, const char *socket_path, int num_tx_antennas, int num_rx_antennas, channel_desc_t **channel_desc);
+void taps_client_stop();
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

radio/vrtsim/vrtsim.c

@@ -19,6 +19,7 @@
  *      contact@openairinterface.org
  */
 
+#include "PHY/TOOLS/tools_defs.h"
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <netinet/in.h>
@@ -46,11 +47,13 @@
 #include "actor.h"
 #include "noise_device.h"
 #include "simde/x86/avx512.h"
+#include "taps_client.h"
 
 // Simulator role
 typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
 
-#define NUM_CHANMOD_THREADS 1
+#define MAX_NUM_ANTENNAS_TX 4
+#define MAX_CHANNEL_LENGTH (1 << 20)
 
 #define ROLE_CLIENT_STRING "client"
 #define ROLE_SERVER_STRING "server"
@@ -58,6 +61,8 @@ typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
 #define VRTSIM_SECTION "vrtsim"
 #define TIME_SCALE_HLP \
   "sample time scale. 1.0 means realtime. Values > 1 mean faster than realtime. Values < 1 mean slower than realtime\n"
+#define TAPS_SOCKET_HLP \
+  "Socket to connect to the channel emulation server. If not set channel emulation would be done inline\n"
 
 // clang-format off
 #define VRTSIM_PARAMS_DESC \
@@ -66,6 +71,7 @@ typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
      {"role",         "either client or server\n",     0, .strptr = &role,                       .defstrval = ROLE_CLIENT_STRING,  TYPE_STRING, 0}, \
      {"timescale",    TIME_SCALE_HLP,                  0, .dblptr = &vrtsim_state->timescale, .defdblval = 1.0,                 TYPE_DOUBLE, 0}, \
      {"chanmod",      "Enable channel modelling",      0, .iptr = &vrtsim_state->chanmod,     .defintval = 0,                   TYPE_INT,    0}, \
+     {"taps-socket",  TAPS_SOCKET_HLP,                 0, .strptr = &vrtsim_state->taps_socket, .defstrval = NULL, TYPE_STRING, 0}, \
   };
 // clang-format on
 
@@ -110,8 +116,12 @@ typedef struct {
   int rx_num_channels;
   channel_desc_t *channel_desc;
   Actor_t *channel_modelling_actors;
+  char *taps_socket;
 } vrtsim_state_t;
 
+// Sample history for channel impulse response
+static c16_t saved_samples[MAX_NUM_ANTENNAS_TX][MAX_CHANNEL_LENGTH] __attribute__((aligned(32))) = {0};
+
 static void histogram_add(histogram_t *histogram, double diff)
 {
   histogram->num_samples++;
@@ -149,6 +159,16 @@ static void load_channel_model(vrtsim_state_t *vrtsim_state)
                    vrtsim_state->tx_bw);
   char *model_name = vrtsim_state->role == ROLE_CLIENT ? "client_tx_channel_model" : "server_tx_channel_model";
   vrtsim_state->channel_desc = find_channel_desc_fromname(model_name);
+  AssertFatal(vrtsim_state->channel_desc != NULL,
+              "Could not find model name %s. Make sure it is present in the config file",
+              model_name);
+  LOG_I(HW,
+        "Channel model %s parameters: path_loss_dB=%.2f, nb_tx=%d, nb_rx=%d, channel_length=%d\n",
+        model_name,
+        vrtsim_state->channel_desc->path_loss_dB,
+        vrtsim_state->channel_desc->nb_tx,
+        vrtsim_state->channel_desc->nb_rx,
+        vrtsim_state->channel_desc->channel_length);
   random_channel(vrtsim_state->channel_desc, 0);
   AssertFatal(vrtsim_state->channel_desc != NULL, "Could not find channel model %s\n", model_name);
 }
@@ -168,6 +188,9 @@ static void vrtsim_readconfig(vrtsim_state_t *vrtsim_state)
   } else {
     AssertFatal(false, "Invalid role configuration\n");
   }
+  if (vrtsim_state->taps_socket) {
+    LOG_A(HW, "VRTSIM: will use taps socket %s\n", vrtsim_state->taps_socket);
+  }
 }
 
 static void *vrtsim_timing_job(void *arg)
@@ -311,12 +334,20 @@ static int vrtsim_connect(openair0_device *device)
 
   // Handle channel modelling after number of RX antennas are known
   int num_tx_stats = 1;
-  if (vrtsim_state->chanmod) {
+  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
     vrtsim_state->channel_modelling_actors = calloc_or_fail(vrtsim_state->peer_info.num_rx_antennas, sizeof(Actor_t));
     for (int i = 0; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
       init_actor(&vrtsim_state->channel_modelling_actors[i], "chanmod", -1);
     }
-    load_channel_model(vrtsim_state);
+    if (vrtsim_state->taps_socket) {
+      taps_client_connect(0,
+                          vrtsim_state->taps_socket,
+                          device->openair0_cfg[0].tx_num_channels,
+                          vrtsim_state->peer_info.num_rx_antennas,
+                          &vrtsim_state->channel_desc);
+    } else {
+      load_channel_model(vrtsim_state);
+    }
     num_tx_stats = vrtsim_state->peer_info.num_rx_antennas;
   }
   vrtsim_state->tx_timing = calloc_or_fail(num_tx_stats, sizeof(tx_timing_t));
@@ -376,7 +407,7 @@ static int vrtsim_write_internal(vrtsim_state_t *vrtsim_state,
 typedef struct {
   vrtsim_state_t *vrtsim_state;
   openair0_timestamp timestamp;
-  c16_t *samples[4];
+  c16_t *samples[MAX_NUM_ANTENNAS_TX];
   int nsamps;
   int nbAnt;
   int flags;
@@ -386,6 +417,7 @@ typedef struct {
 static void perform_channel_modelling(void *arg)
 {
   channel_modelling_args_t *channel_modelling_args = arg;
+  vrtsim_state_t *vrtsim_state = channel_modelling_args->vrtsim_state;
   int nsamps = channel_modelling_args->nsamps;
   int aarx = channel_modelling_args->aarx;
   int nb_tx_ant = channel_modelling_args->nbAnt;
@@ -396,31 +428,43 @@ static void perform_channel_modelling(void *arg)
   // Apply noise from global settings
   get_noise_vector((float *)samples, nsamps * 2);
 
-  channel_desc_t *channel_desc = channel_modelling_args->vrtsim_state->channel_desc;
-  const float pathloss_linear = powf(10, channel_desc->path_loss_dB / 20.0);
+  channel_desc_t *channel_desc = vrtsim_state->channel_desc;
+
+  if (channel_desc == NULL) {
+    return;
+  }
 
-  // Convert channel impulse response to float + apply pathloss
   cf_t channel_impulse_response[nb_tx_ant][channel_desc->channel_length];
-  for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
-    const struct complexd *channelModel = channel_desc->ch[aarx + (aatx * channel_desc->nb_rx)];
-    for (int i = 0; i < channel_desc->channel_length; i++) {
-      channel_impulse_response[aatx][i].r = channelModel[i].r * pathloss_linear;
-      channel_impulse_response[aatx][i].i = channelModel[i].i * pathloss_linear;
+  cf_t *channel_impulse_response_p[nb_tx_ant];
+  if (!vrtsim_state->taps_socket) {
+    const float pathloss_linear = powf(10, channel_desc->path_loss_dB / 20.0);
+    // Convert channel impulse response to float + apply pathloss
+    for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+      const struct complexd *channelModel = channel_desc->ch[aarx + (aatx * channel_desc->nb_rx)];
+      for (int i = 0; i < channel_desc->channel_length; i++) {
+        channel_impulse_response[aatx][i].r = channelModel[i].r * pathloss_linear;
+        channel_impulse_response[aatx][i].i = channelModel[i].i * pathloss_linear;
+      }
+      channel_impulse_response_p[aatx] = channel_impulse_response[aatx];
+    }
+  } else {
+    for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+      struct complexf *channelModel = channel_desc->ch_ps[aarx + (aatx * channel_desc->nb_rx)];
+      channel_impulse_response_p[aatx] = channelModel;
     }
   }
 
   for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+    c16_t *previous_samples = saved_samples[aatx];
     for (int i = 0; i < nsamps; i++) {
-      cf_t *impulse_response = channel_impulse_response[aatx];
+      cf_t *impulse_response = channel_impulse_response_p[aatx];
       for (int l = 0; l < channel_desc->channel_length; l++) {
         int idx = i - l;
-        // TODO: Use AVX512 for this
-        // TODO: What are the previously sent samples (for impulse response)
-        if (idx >= 0) {
-          c16_t tx_input = input_samples[aatx][idx];
-          samples[i].r += tx_input.r * impulse_response[l].r - tx_input.i * impulse_response[l].i;
-          samples[i].i += tx_input.i * impulse_response[l].r + tx_input.r * impulse_response[l].i;
-        }
+        // TODO: Use AVX2 for this
+        c16_t tx_input = idx >= 0 ? input_samples[aatx][idx]
+                                  : previous_samples[(channel_modelling_args->timestamp + i + idx) % MAX_CHANNEL_LENGTH];
+        samples[i].r += tx_input.r * impulse_response[l].r - tx_input.i * impulse_response[l].i;
+        samples[i].i += tx_input.i * impulse_response[l].r + tx_input.r * impulse_response[l].i;
       }
     }
   }
@@ -462,6 +506,7 @@ static int vrtsim_write_with_chanmod(vrtsim_state_t *vrtsim_state,
                                      int nbAnt,
                                      int flags)
 {
+  AssertFatal(nbAnt < MAX_NUM_ANTENNAS_TX, "Number of antennas %d exceeds maximum %d\n", nbAnt, MAX_NUM_ANTENNAS_TX);
   for (int aarx = 0; aarx < vrtsim_state->peer_info.num_rx_antennas; aarx++) {
     notifiedFIFO_elt_t *task = newNotifiedFIFO_elt(sizeof(channel_modelling_args_t), 0, NULL, perform_channel_modelling);
     channel_modelling_args_t *args = (channel_modelling_args_t *)NotifiedFifoData(task);
@@ -476,6 +521,23 @@ static int vrtsim_write_with_chanmod(vrtsim_state_t *vrtsim_state,
     }
     pushNotifiedFIFO(&vrtsim_state->channel_modelling_actors[aarx].fifo, task);
   }
+  int start_index = timestamp % MAX_CHANNEL_LENGTH;
+  if (start_index + nsamps > MAX_CHANNEL_LENGTH) {
+    // Requires two copies
+    for (int aatx = 0; aatx < nbAnt; aatx++) {
+      int first_copy_size = MAX_CHANNEL_LENGTH - start_index;
+      c16_t *samples = (c16_t *)samplesVoid[aatx];
+      memcpy(&saved_samples[aatx][start_index], &samples[0], sizeof(c16_t) * first_copy_size);
+      int second_copy_size = nsamps - first_copy_size;
+      memcpy(&saved_samples[aatx][0], &samples[first_copy_size], sizeof(c16_t) * second_copy_size);
+    }
+  } else {
+    // Single copy
+    for (int aatx = 0; aatx < nbAnt; aatx++) {
+      c16_t *samples = (c16_t *)samplesVoid[aatx];
+      memcpy(&saved_samples[aatx][start_index], &samples[0], sizeof(c16_t) * nsamps);
+    }
+  }
   return nsamps;
 }
 
@@ -483,14 +545,23 @@ static int vrtsim_write(openair0_device *device, openair0_timestamp timestamp, v
 {
   timestamp -= device->openair0_cfg->command_line_sample_advance;
   vrtsim_state_t *vrtsim_state = (vrtsim_state_t *)device->priv;
-  return vrtsim_state->chanmod ? vrtsim_write_with_chanmod(vrtsim_state, timestamp, samplesVoid, nsamps, nbAnt, flags)
-                               : vrtsim_write_internal(vrtsim_state, timestamp, (c16_t *)samplesVoid[0], nsamps, 0, flags, 0);
+  bool channel_modelling = vrtsim_state->chanmod || vrtsim_state->taps_socket;
+  return channel_modelling ? vrtsim_write_with_chanmod(vrtsim_state, timestamp, samplesVoid, nsamps, nbAnt, flags)
+                           : vrtsim_write_internal(vrtsim_state, timestamp, (c16_t *)samplesVoid[0], nsamps, 0, flags, 0);
 }
 
 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;
@@ -506,14 +577,14 @@ static int vrtsim_read(openair0_device *device, openair0_timestamp *ptimestamp,
 static void vrtsim_end(openair0_device *device)
 {
   vrtsim_state_t *vrtsim_state = (vrtsim_state_t *)device->priv;
-  if (vrtsim_state->role == ROLE_SERVER) {
+  if (vrtsim_state->role == ROLE_SERVER && vrtsim_state->run_timing_thread) {
     vrtsim_state->run_timing_thread = false;
     int ret = pthread_join(vrtsim_state->timing_thread, NULL);
     AssertFatal(ret == 0, "pthread_join() failed: errno: %d, %s\n", errno, strerror(errno));
   }
 
   tx_timing_t *tx_timing = vrtsim_state->tx_timing;
-  if (vrtsim_state->chanmod) {
+  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
     for (int i = 0; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
       shutdown_actor(&vrtsim_state->channel_modelling_actors[i]);
     }
@@ -526,9 +597,12 @@ static void vrtsim_end(openair0_device *device)
       tx_timing->tx_samples_total += tx_timing[i].tx_samples_total;
     }
     tx_timing->average_tx_budget /= vrtsim_state->peer_info.num_rx_antennas;
+    free_noise_device();
+    if (vrtsim_state->taps_socket) {
+      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);
 
@@ -588,10 +662,11 @@ __attribute__((__visibility__("default"))) int device_init(openair0_device *devi
   vrtsim_state->tx_num_channels = openair0_cfg->tx_num_channels;
   vrtsim_state->rx_num_channels = openair0_cfg->rx_num_channels;
 
-  if (vrtsim_state->chanmod) {
+  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
     init_channelmod();
     int noise_power_dBFS = get_noise_power_dBFS();
     int16_t noise_power = noise_power_dBFS == INVALID_DBFS_VALUE ? 0 : (int16_t)(32767.0 / powf(10.0, .05 * -noise_power_dBFS));
+    LOG_A(HW, "VRTSIM: Noise power %d sample value\n", noise_power);
     init_noise_device(noise_power);
   }
   return 0;