wip

trx_ecpri.c

@@ -44,6 +44,7 @@
 #define PPS_UPDATE_PERIOD INT64_C(1000000000)
 #define STAT_INT_LEN "9"
 
+#define MIN_PACKET_SIZE 64
 #define MAX_CHANNELS 4
 
 /* eCPRI frame structure
@@ -90,16 +91,21 @@ typedef struct {
   int encode_affinity;
   int decode_affinity;
   int statistic_affinity;
+
   int ecpri_period;
   int flow_id;
   int frame_frequency;
+
   int trx_buf_size;
   int txrx_buf_size;
+
   int trace_rx;
   int trace_tx;
   int trace_offset;
+
   int monitor_pps;
   int monitor_trigger_duration;
+
   int start_sending;
   int start_receiving;
   int rx_n_channel;
@@ -226,11 +232,18 @@ static void log_exit(const char * section, const char * msg, ...) {
 
   // Dump useful information
   print_stats(stderr, 1);
-  fprintf(stderr, "TX RBUF: ri %d wi %d ra %d wa %d\n", tx_rbuf.read_index, tx_rbuf.write_index, rbuf_read_amount(&tx_rbuf), rbuf_write_amount(&tx_rbuf));
-  fprintf(stderr, "RX RBUF: ri %d wi %d ra %d wa %d\n", rx_rbuf.read_index, rx_rbuf.write_index, rbuf_read_amount(&rx_rbuf), rbuf_write_amount(&rx_rbuf));
-  fprintf(stderr, "TRXW RBUF: ri %d wi %d ra %d wa %d\n", trxw_rbuf[0].read_index, trxw_rbuf[0].write_index, rbuf_read_amount(&trxw_rbuf[0]), rbuf_write_amount(&trxw_rbuf[0]));
-  fprintf(stderr, "TRXR RBUF: ri %d wi %d ra %d wa %d\n", trxr_rbuf[0].read_index, trxr_rbuf[0].write_index, rbuf_read_amount(&trxr_rbuf[0]), rbuf_write_amount(&trxr_rbuf[0]));
-  fprintf(stderr, "TRXW GROUP RBUF: ri %d wi %d ra %d wa %d\n", trxw_group_rbuf.read_index, trxw_group_rbuf.write_index, rbuf_read_amount(&trxw_group_rbuf), rbuf_write_amount(&trxw_group_rbuf));
+  fprintf(stderr, "TX RBUF: ri %d wi %d ra %d wa %d\n",
+    tx_rbuf.read_index, tx_rbuf.write_index, rbuf_read_amount(&tx_rbuf),
+    rbuf_write_amount(&tx_rbuf));
+  fprintf(stderr, "RX RBUF: ri %d wi %d ra %d wa %d\n",
+    rx_rbuf.read_index, rx_rbuf.write_index, rbuf_read_amount(&rx_rbuf),
+    rbuf_write_amount(&rx_rbuf));
+  fprintf(stderr, "TRXW RBUF: ri %d wi %d ra %d wa %d\n",
+    trxw_rbuf[0].read_index, trxw_rbuf[0].write_index, rbuf_read_amount(&trxw_rbuf[0]),
+    rbuf_write_amount(&trxw_rbuf[0]));
+  fprintf(stderr, "TRXR RBUF: ri %d wi %d ra %d wa %d\n",
+    trxr_rbuf[0].read_index, trxr_rbuf[0].write_index, rbuf_read_amount(&trxr_rbuf[0]),
+    rbuf_write_amount(&trxr_rbuf[0]));
 
   fflush(stdout);
   fflush(stderr);
@@ -383,84 +396,125 @@ static void *send_thread(void *p) {
   pthread_exit(EXIT_SUCCESS);
 }
 
-static void *prepare_thread(void *p) {
+/* If sync has happenned (=we have received frames):
+     Prepare as soon as TRX has packets to write
+     Signal
+   Else:
+     Prepare as soon as there is space in tx buffer */
+#define TX_SYNC_BURST_SIZE 512
+static void *encode_thread(void *p) {
 
-	cpu_set_t mask;
+  cpu_set_t mask;
   TRXEcpriState * s = (TRXEcpriState *) p;
-  int tx_ready_buffer_full = 0;
-  log_info("PREPARE_THREAD", "Thread init");
+  int64_t target_counter = 0;
+  struct timespec next;
+  int reset_encode_counter = 1;
 
-	// Set thread CPU affinity
-	CPU_ZERO(&mask);
-	CPU_SET(s->prepare_affinity, &mask);
-	if (sched_setaffinity(0, sizeof(mask), &mask))
-		error(EXIT_FAILURE, errno, "Could not set CPU affinity to CPU %d\n", s->prepare_affinity);
+  // Set thread CPU affinity
+  CPU_ZERO(&mask);
+  CPU_SET(s->encode_affinity, &mask);
+  if (sched_setaffinity(0, sizeof(mask), &mask))
+    error(EXIT_FAILURE, errno, "Could not set CPU affinity to CPU %d\n", s->encode_affinity);
 
   for(int64_t i = 0;; i++) {
-    int16_t samples_int[256];
-
-    // If we have frames to prepare
-    int n = rbuf_write_amount(&tx_rbuf);
-    if((i == 0) || n) {
-      // If there are frames from trx_write callback to prepare
-      if(rbuf_read_amount(&trx_write_rbuf)) {
-        int64_t ts = trx_wb_ts[trx_wb_part_read_index];
-        int empty_frames_ahead = ts - prepared_frame_count;
-        empty_frames_ahead = empty_frames_ahead < n ? empty_frames_ahead : n;
-        if(empty_frames_ahead > 0) {
-          for(int j = 0; j < empty_frames_ahead; j++) {
-            *((uint16_t *) (RBUF_WRITE(tx_rbuf, uint8_t) + 20)) = htons(seq_id++);
-            rbuf_update_write_index(&tx_rbuf);
-            prepared_frame_count++;
-          }
+    int n;
+
+    n = rbuf_write_amount(&tx_rbuf);
+
+    // Send empty frames until we receive something
+    if(s->start_sending) {
+      if(!sync_complete) {
+        if(i == 0)
+          clock_gettime(CLOCK_TAI, &next);
+        // Limit packets sent
+        if(encode_counter.counter > target_counter) {
+          int k = (encode_counter.counter - target_counter + (s->frame_frequency / 100) - 1) / (s->frame_frequency / 100);
+          add_ns(&next, k * 1000 * 1000 * 10); // 10ms to send 38400 packets
+          clock_nanosleep(CLOCK_TAI, TIMER_ABSTIME, &next, NULL);
+          target_counter += k * s->frame_frequency / 100;
         }
-        else if (empty_frames_ahead == 0) {
-          int m = trx_wb_part[(trx_wb_part_read_index + 1) % TRX_WB_MAX_PARTS] - trx_write_rbuf.read_index;
-          m = m < n ? m : n;
-          for(int j = 0; j < m; j++) {
-            float * const trx_samples = RBUF_READ(trx_write_rbuf, float);
-            uint8_t * const tx_frame = RBUF_WRITE(tx_rbuf, uint8_t);
-            memset(samples_int, 0, 512);
-
-            float_to_int16(samples_int, trx_samples, 256, 32767);
-            encode_bf1(tx_frame + 22      , samples_int);
-            encode_bf1(tx_frame + 22 + 60 , samples_int + 64);
-            encode_bf1(tx_frame + 22 + 120, samples_int + 128);
-            encode_bf1(tx_frame + 22 + 180, samples_int + 192);
-
-            *((uint16_t *)(tx_frame + 20)) = htons(seq_id++);
-
-            rbuf_update_write_index(&tx_rbuf);
-            rbuf_update_read_index(&trx_write_rbuf);
-            prepared_frame_count++;
-          }
-          if(m == 0)
-            trx_wb_part_read_index = (trx_wb_part_read_index + 1) % TRX_WB_MAX_PARTS;
-        }
-        else {
-          log_error("PREPARE_THREAD", "missed trx_write timestamp");
+        n = (n > TX_SYNC_BURST_SIZE) ? n : TX_SYNC_BURST_SIZE;
+        n = (n < (s->frame_frequency / 100)) ? n : (s->frame_frequency / 100);
+        for(int j = 0; j < n; j++) {
+          *((uint16_t *) (RBUF_WRITE0(tx_rbuf, uint8_t) + 6)) = htons(seq_id++);
+          rbuf_update_write_index(&tx_rbuf);
         }
+        update_counter(&encode_counter, n);
       }
-      else {
-        *((uint16_t *) (RBUF_WRITE(tx_rbuf, uint8_t) + 20)) = htons(seq_id++);
-        rbuf_update_write_index(&tx_rbuf);
-        prepared_frame_count++;
+      else if (reset_encode_counter) {
+        if(s->trace_tx)
+          encode_counter_prev = encode_counter.counter;
+        encode_counter.counter = 0;
+        reset_encode_counter = 0;
+        seq_id = 0;
       }
     }
-    else {
-      if (!tx_ready_buffer_full) {
-        tx_ready_buffer_full = 1;
-        pthread_mutex_lock(&tx_ready_mutex);
-        pthread_cond_signal(&tx_ready_cond);
-        pthread_mutex_unlock(&tx_ready_mutex);
+
+    // If we have frames to encode (is there space in TX buffer)
+    // If there are frames from trx_write callback to encode
+    if(n && rbuf_read_amount(&trxw_rbuf[0])) {
+      sample_group_t * g; int nb_frames;
+
+      g = RBUF_READ0(trxw_group_rbuf, sample_group_t);
+      if(g->wait) {
+        g->wait = 0;
+        g->count -= encode_counter.counter;
+        g->zeroes = 1;
+      }
+      nb_frames = g->count > n ? n : g->count;
+      g->count -= nb_frames;
+      if(s->trace_tx) {
+        if(sync_complete && (encode_counter.counter + nb_frames) >= (tx_rbuf.buf_len + s->trace_offset)) {
+          tx_trace_ready = 1;
+          log_info("ENCODE_THREAD", "TX Trace ready");
+          pthread_exit(EXIT_SUCCESS);
+        } else if (tx_trace_ready) {
+          pthread_exit(EXIT_SUCCESS);
+        }
+      }
+      if(g->zeroes) {
+        for(int j = 0; j < nb_frames; j++) {
+          memset(RBUF_WRITE0(tx_rbuf, uint8_t) + 8, 0x00, 240);
+          *((uint16_t *) (RBUF_WRITE0(tx_rbuf, uint8_t) + 6)) = htons(seq_id++);
+          rbuf_update_write_index(&tx_rbuf);
+        }
+        trxw_rbuf[0].read_index = (trxw_rbuf[0].read_index + nb_frames) % trxw_rbuf[0].buf_len;
+      } else {
+
+        int nc;
+        int nf = nb_frames;
+        while((nc = rbuf_contiguous_copy(&trxw_rbuf[0], &tx_rbuf, nf))) {
+          Complex * iq_samples[4];
+          uint8_t * buf = RBUF_WRITE0(tx_rbuf, uint8_t) + 8;
+          for(int j = 0; j < s->tx_n_channel; j++)
+            iq_samples[j] = ((Complex *) trxw_rbuf[j].buffer) + (trxw_rbuf[0].read_index * trxw_rbuf[0].len);
+
+          for(int i = 0; i < nc; i++) {
+            for(int i = 0; i < s->tx_n_channel ; i++)
+              encode_s64_b60_2(buf + i * 60, (float *) iq_samples[i]);
+            *((uint16_t *)(buf - 2)) = htons(seq_id++);
+            for(int j = 0; j < s->tx_n_channel; j++)
+              iq_samples[j] += trxw_rbuf[0].len;
+            buf += tx_rbuf.len;
+          }
+          tx_rbuf.write_index = (tx_rbuf.write_index + nc) % tx_rbuf.buf_len;
+          trxw_rbuf[0].read_index = (trxw_rbuf[0].read_index + nc) % trxw_rbuf[0].buf_len;
+          nf -= nc;
+        }
+        if(nf)
+          exit(EXIT_FAILURE);
+
+      }
+      update_counter(&encode_counter, nb_frames);
+
+      if(!g->count) {
+        rbuf_update_read_index(&trxw_group_rbuf);
       }
-      pthread_mutex_lock(&tx_mutex);
-      pthread_cond_wait(&tx_cond, &tx_mutex);
-      pthread_mutex_unlock(&tx_mutex);
     }
   }
   pthread_exit(EXIT_SUCCESS);
 }
+
 static void *decompress_thread(void *p) {
 
 	cpu_set_t mask;
@@ -507,19 +561,101 @@ static void *decompress_thread(void *p) {
   pthread_exit(EXIT_SUCCESS);
 }
 
+static void *statistic_thread(void *p) {
+  struct timespec next, initial;
+  int64_t recv_stop = 0;
+  cpu_set_t mask;
+  TRXEcpriState * s = (TRXEcpriState *) p;
+  FILE * stats_file_desc;
+
+  log_info("STATISTIC_THREAD", "Thread init");
+
+  char stats_file_name[256];
+  memset(stats_file_name, '\0', 256);
+  sprintf(stats_file_name, "%s/ecpri.stats", s->log_directory);
+  stats_file_desc = fopen(stats_file_name, "w+");
+
+  if(!stats_file_desc)
+    error(EXIT_FAILURE, errno, "Couldn't open %s\n", stats_file_name);
+
+  // Set thread CPU affinity
+  CPU_ZERO(&mask);
+  CPU_SET(s->statistic_affinity, &mask);
+  if (sched_setaffinity(0, sizeof(mask), &mask))
+    error(EXIT_FAILURE, errno, "Could not set CPU affinity to CPU %d\n", s->statistic_affinity);
+
+  clock_gettime(CLOCK_TAI, &initial);
+  next = initial;
+  for(int64_t i = 0;; i++) {
+    add_ns(&next, s->statistics_refresh_rate_ns);
+    if(s->trace_rx || s->trace_tx)
+      trace_handler(initial, s);
+    print_stats(stats_file_desc, (i % 50) == 0);
+#ifdef DEBUG
+    fprintf(stats_file_desc, 
+        "%d %d %d %d %d %d %d %d\n", 
+        rx_rbuf.write_index,
+        rx_rbuf.read_index,
+        trxr_rbuf[0].write_index,
+        trxr_rbuf[0].read_index,
+        trxw_rbuf[0].write_index,
+        trxw_rbuf[0].read_index,
+        tx_rbuf.write_index,
+        tx_rbuf.read_index);
+    fprintf(stats_file_desc, "TRXW RBUF: ri %d wi %d ra %d wa %d\n", trxw_rbuf[0].read_index, trxw_rbuf[0].write_index, rbuf_read_amount(&trxw_rbuf[0]), rbuf_write_amount(&trxw_rbuf[0]));
+#endif
+    fflush(stats_file_desc);
+
+    update_counter_pps(&rx_drop_counter);
+    update_counter_pps(&tx_drop_counter);
+    update_counter_pps(&recv_counter);
+    update_counter_pps(&decode_counter);
+    update_counter_pps(&read_counter);
+    update_counter_pps(&write_counter);
+    update_counter_pps(&encode_counter);
+    update_counter_pps(&sent_counter);
+    if(s->monitor_pps) {
+      if(recv_counter.pps > recv_pps_threshold) {
+        recv_pps_threshold_hit = 1;
+      }
+      if(recv_pps_threshold_hit && recv_counter.pps < recv_pps_threshold) {
+        struct timespec _ts;
+        int64_t ts;
+        clock_gettime(CLOCK_MONOTONIC, &_ts);
+        ts = ts_to_int(_ts);
+        if((recv_stop && ((ts - recv_stop) > (s->monitor_trigger_duration) * INT64_C(1000000000)))) {
+          if(s->monitor_pps == 1)
+            log_exit("MONITOR", "Stopped recieving packets, restarting...");
+          log_info("MONITOR", "Stopped recieving packets, sending again...");
+          sync_complete = 0;
+          recv_stop = 0;
+        }
+        if(!recv_stop)
+          recv_stop = ts;
+      }
+    }
+
+    clock_nanosleep(CLOCK_TAI, TIMER_ABSTIME, &next, NULL);
+  }
+  pthread_exit(EXIT_SUCCESS);
+}
+
 static int start_threads(TRXEcpriState * s) {
   pthread_t recv_pthread;
   pthread_t send_pthread;
-  pthread_t prepare_pthread;
-  pthread_t decompress_pthread;
+  pthread_t encode_pthread;
+  pthread_t decode_pthread;
+  pthread_t statistic_pthread;
   struct sched_param recv_param;
   struct sched_param send_param;
-  struct sched_param prepare_param;
-  struct sched_param decompress_param;
+  struct sched_param encode_param;
+  struct sched_param decode_param;
+  struct sched_param statistic_param;
   pthread_attr_t recv_attr;
   pthread_attr_t send_attr;
-  pthread_attr_t prepare_attr;
-  pthread_attr_t decompress_attr;
+  pthread_attr_t encode_attr;
+  pthread_attr_t decode_attr;
+  pthread_attr_t statistic_attr;
 
   log_info("TRX_ECPRI", "Starting threads");
 
@@ -551,38 +687,56 @@ static int start_threads(TRXEcpriState * s) {
   if (pthread_attr_setinheritsched(&send_attr, PTHREAD_EXPLICIT_SCHED))
     log_error("TRX_ECPRI", "pthread setinheritsched failed\n");
 
-  if (pthread_attr_init(&prepare_attr))
+  if (pthread_attr_init(&encode_attr))
     log_error("TRX_ECPRI", "init pthread attributes failed\n");
-  if (pthread_attr_setstacksize(&prepare_attr, PTHREAD_STACK_MIN))
+  if (pthread_attr_setstacksize(&encode_attr, PTHREAD_STACK_MIN))
     log_error("TRX_ECPRI", "pthread setstacksize failed\n");
-  if (pthread_attr_setschedpolicy(&prepare_attr, SCHED_FIFO))
+  if (pthread_attr_setschedpolicy(&encode_attr, SCHED_FIFO))
     log_error("TRX_ECPRI", "pthread setschedpolicy failed\n");
-  prepare_param.sched_priority = 97;
-  if (pthread_attr_setschedparam(&prepare_attr, &prepare_param))
+  encode_param.sched_priority = 97;
+  if (pthread_attr_setschedparam(&encode_attr, &encode_param))
     log_error("TRX_ECPRI", "pthread setschedparam failed\n");
-  if (pthread_attr_setinheritsched(&prepare_attr, PTHREAD_EXPLICIT_SCHED))
+  if (pthread_attr_setinheritsched(&encode_attr, PTHREAD_EXPLICIT_SCHED))
     log_error("TRX_ECPRI", "pthread setinheritsched failed\n");
 
-  if (pthread_attr_init(&decompress_attr))
+  if (pthread_attr_init(&decode_attr))
     log_error("TRX_ECPRI", "init pthread attributes failed\n");
-  if (pthread_attr_setstacksize(&decompress_attr, PTHREAD_STACK_MIN))
+  if (pthread_attr_setstacksize(&decode_attr, PTHREAD_STACK_MIN))
     log_error("TRX_ECPRI", "pthread setstacksize failed\n");
-  if (pthread_attr_setschedpolicy(&decompress_attr, SCHED_FIFO))
+  if (pthread_attr_setschedpolicy(&decode_attr, SCHED_FIFO))
     log_error("TRX_ECPRI", "pthread setschedpolicy failed\n");
-  decompress_param.sched_priority = 97;
-  if (pthread_attr_setschedparam(&decompress_attr, &decompress_param))
+  decode_param.sched_priority = 97;
+  if (pthread_attr_setschedparam(&decode_attr, &decode_param))
     log_error("TRX_ECPRI", "pthread setschedparam failed\n");
-  if (pthread_attr_setinheritsched(&decompress_attr, PTHREAD_EXPLICIT_SCHED))
+  if (pthread_attr_setinheritsched(&decode_attr, PTHREAD_EXPLICIT_SCHED))
     log_error("TRX_ECPRI", "pthread setinheritsched failed\n");
 
-  if (pthread_create(&recv_pthread, NULL, recv_thread, s))
-    error(EXIT_FAILURE, errno, "Couldn't create recv thread");
+  if (pthread_attr_init(&statistic_attr))
+    log_error("TRX_ECPRI", "init pthread attributes failed\n");
+  if (pthread_attr_setstacksize(&statistic_attr, PTHREAD_STACK_MIN))
+    log_error("TRX_ECPRI", "pthread setstacksize failed\n");
+  if (pthread_attr_setschedpolicy(&statistic_attr, SCHED_FIFO))
+    log_error("TRX_ECPRI", "pthread setschedpolicy failed\n");
+  statistic_param.sched_priority = 97;
+  if (pthread_attr_setschedparam(&statistic_attr, &statistic_param))
+    log_error("TRX_ECPRI", "pthread setschedparam failed\n");
+  if (pthread_attr_setinheritsched(&statistic_attr, PTHREAD_EXPLICIT_SCHED))
+    log_error("TRX_ECPRI", "pthread setinheritsched failed\n");
+
+  if (pthread_create(&statistic_pthread, NULL, statistic_thread, s))
+    error(EXIT_FAILURE, errno, "Couldn't create statistic thread");
+  usleep(1000 * 20);
+  if (pthread_create(&encode_pthread, NULL, encode_thread, s))
+    error(EXIT_FAILURE, errno, "Couldn't create encode thread");
+  usleep(1000 * 20);
+  if (pthread_create(&decode_pthread, NULL, decode_thread, s))
+    error(EXIT_FAILURE, errno, "Couldn't create decode thread");
+  usleep(1000 * 20);
   if (pthread_create(&send_pthread, NULL, send_thread, s))
     error(EXIT_FAILURE, errno, "Couldn't create send thread");
-  if (pthread_create(&prepare_pthread, NULL, prepare_thread, s))
-    error(EXIT_FAILURE, errno, "Couldn't create prepare thread");
-  if (pthread_create(&decompress_pthread, NULL, decompress_thread, s))
-    error(EXIT_FAILURE, errno, "Couldn't create decompress thread");
+  usleep(1000 * 500);
+  if (pthread_create(&recv_pthread, NULL, recv_thread, s))
+    error(EXIT_FAILURE, errno, "Couldn't create recv thread");
 
   return 0;
 }
@@ -608,20 +762,19 @@ int start(TRXEcpriState * s) {
   init_counter(&encode_counter);
   init_counter(&sent_counter);
 
-  RBUF_INIT(rx_rbuf, "RX ring buffer", TXRX_BUF_MAX_SIZE, RX_MAX_PACKET_SIZE, uint8_t);
-  RBUF_INIT(tx_rbuf, "TX ring buffer", TXRX_BUF_MAX_SIZE, TX_ECPRI_PACKET_SIZE, uint8_t);
-  for(int i = 0; i < TX_N_CHANNEL; i++) {
-    char s[256];
-    sprintf(s, "TRXWrite Ring Buffer %d", i);
-    RBUF_INIT(trxw_rbuf[i], s, TRX_BUF_MAX_SIZE, N_SAMPLES, Complex);
+  init_rbuf(&rx_rbuf, "RX ring buffer", s->txrx_buf_size, MIN_PACKET_SIZE);
+  init_rbuf(&tx_rbuf, "TX ring buffer", s->txrx_buf_size, MIN_PACKET_SIZE);
+
+  for(int i = 0; i < s->tx_n_channel; i++) {
+    char name[256];
+    sprintf(name, "TRXWrite Ring Buffer %d", i);
+    init_rbuf(trxw_rbuf[i], name, s->trx_buf_size, Complex);
   }
-  for(int i = 0; i < RX_N_CHANNEL; i++) {
-    char s[256];
-    sprintf(s, "TRXRead Ring Buffer %d", i);
-    RBUF_INIT(trxr_rbuf[i], s, TRX_BUF_MAX_SIZE, N_SAMPLES, Complex);
+  for(int i = 0; i < s->rx_n_channel; i++) {
+    char name[256];
+    sprintf(name, "TRXRead Ring Buffer %d", i);
+    init_rbuf(trxr_rbuf[i], name, s->trx_buf_size, Complex);
   }
-  RBUF_INIT(trxw_group_rbuf, "TRXGroupWrite ring buffer", TRX_MAX_GROUP, 1, sample_group_t);
-
 
   trx_wb_part_read_index = 0;
   trx_wb_part_write_index = 0;
@@ -638,14 +791,14 @@ int start(TRXEcpriState * s) {
 
   memset((uint8_t *) ecpri_packet, 0, PACKET_SIZE);
 
-  if (!(if_index = if_nametoindex(s->rec_if))) {
+  if (!(if_index = if_nametoindex(s->bbu_if))) {
     perror("if_nametoindex");
     return 1;
   }
 
-  if(sscanf(s->re_mac, "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx%*c", &dst_mac[0], &dst_mac[1], &dst_mac[2], &dst_mac[3], &dst_mac[4], &dst_mac[5]) != 6)
+  if(sscanf(s->rrh_mac, "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx%*c", &dst_mac[0], &dst_mac[1], &dst_mac[2], &dst_mac[3], &dst_mac[4], &dst_mac[5]) != 6)
     fprintf(stderr, "Invalid RRH MAC address\n");
-  if(sscanf(s->rec_mac, "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx%*c", &src_mac[0], &src_mac[1], &src_mac[2], &src_mac[3], &src_mac[4], &src_mac[5]) != 6)
+  if(sscanf(s->bbu_mac, "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx%*c", &src_mac[0], &src_mac[1], &src_mac[2], &src_mac[3], &src_mac[4], &src_mac[5]) != 6)
     fprintf(stderr, "Invalid RRH MAC address\n");
   
   if ((send_sockfd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL))) == -1) {
@@ -695,41 +848,98 @@ static void trx_ecpri_end(TRXState *s1)
   free(s);
 }
 
+static int64_t prev_ts = 0;
+static int64_t prev_count = 0;
+#define M 32
+
 static void trx_ecpri_write(TRXState *s1, trx_timestamp_t timestamp, const void **__samples, int count, int tx_port_index, TRXWriteMetadata *md)
 {
-  (void) s1;
+  int write_count; int64_t ts; sample_group_t * g; int nc; int nk = 0;
   float ** _samples = (float **) __samples;
-  int write_count = count >> 5;
-  int64_t ts = timestamp >> 5;
-
-  trx_wb_part[trx_wb_part_write_index] = trx_write_rbuf.write_index;
-  trx_wb_ts[trx_wb_part_write_index] = ts;
-  for(int k = 0; k < write_count; k++) {
-    for(int i = 0; i < 4; i++)
-      for(int j = 0; j < 64; j++)
-        RBUF_WRITE(trx_write_rbuf, float)[i * 64 + j] = _samples[i][j + (k << 6)];
-    rbuf_update_write_index(&trx_write_rbuf);
+  TRXEcpriState *s = s1->opaque;
+  write_count = count / M;
+  ts = timestamp / M;
+
+  log_debug("TRX_ECPRI_WRITE", "trx_ecpri_write, count = %ld", count);
+  
+  if(prev_count && ((ts - prev_ts) != prev_count)) {
+    log_exit("TRX_ECPRI_WRITE", "Gap between timestamps: prev_ts %li ts %li prev_count %li count %li diff_ts %li", prev_ts, ts, prev_count, count, (ts - prev_ts));
+  }
+  prev_ts = ts; prev_count = write_count;
+
+  if(write_count > rbuf_write_amount(&trxw_rbuf[0])) {
+    //log_exit("TRX_ECPRI_WRITE", "Not enough space to write in trxw_rbuf (write count = %d)", write_count);
+    update_counter(&tx_drop_counter, write_count);
+    return;
+  }
+
+  if(s->trace_tx) {
+    if((write_counter.counter + write_count) >= (trxw_rbuf[0].buf_len + s->trace_offset)) {
+      tx_trace_ready = 1;
+      log_info("TRX_ECPRI_WRITE", "TX Trace ready");
+      pthread_exit(EXIT_SUCCESS);
+    } else if (tx_trace_ready) {
+      pthread_exit(EXIT_SUCCESS);
+    }
+  }
+
+  if(first_trx_write) {
+    sample_group_t * g2 = RBUF_WRITE0(trxw_group_rbuf, sample_group_t);
+    g2->count = ts;
+    g2->wait = 1;
+    g2->zeroes = 1;
+    rbuf_update_write_index(&trxw_group_rbuf);
+  }
+  g = RBUF_WRITE0(trxw_group_rbuf, sample_group_t);
+  g->zeroes = __samples ? 0 : 1;
+  g->wait = 0;
+  g->count = write_count;
+
+  while((nc = rbuf_contiguous_copy(NULL, &trxw_rbuf[0], write_count))) {
+    int len = nc * trxr_rbuf[0].len * sizeof(Complex);
+    if(__samples)
+      for(int i = 0; i < s->tx_n_channel; i++) {
+        uint8_t * src = ((uint8_t*) _samples[i]) + (nk * trxr_rbuf[0].len * sizeof(Complex));
+        uint8_t * dst = ((uint8_t *) trxw_rbuf[i].buffer) + trxw_rbuf[0].write_index * trxw_rbuf[0].len * sizeof(Complex);
+        memcpy(dst, src, len);
+      }
+    trxw_rbuf[0].write_index = (trxw_rbuf[0].write_index + nc) % trxw_rbuf[0].buf_len;
+    write_count -= nc;
+    nk += nc;
   }
-  trx_wb_part_write_index = (trx_wb_part_write_index + 1) % TRX_WB_MAX_PARTS;
-  trx_wb_part[trx_wb_part_write_index] = trx_write_rbuf.write_index + write_count;
+
+  rbuf_update_write_index(&trxw_group_rbuf);
+  update_counter(&write_counter, count / M);
 }
 
 static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__samples, int count, int rx_port_index, TRXReadMetadata *md)
 {
-  (void) s1;
+  int nc; int n;
   float ** _samples = (float **) __samples;
   int read_count = (count / M);
+  int offset = 0;
+  TRXEcpriState *s = s1->opaque;
 
-  for(int k = 0; k < read_count; k++) {
-    float * trx_samples;
-    sem_wait(&trx_read_sem);
-    trx_samples = RBUF_READ(trx_read_rbuf, float);
-    for(int i = 0; i < 4; i++)
-      for(int j = 0; j < 64; j++)
-        _samples[i][j + (k << 6)] = trx_samples[i * 64 + j];
-    rbuf_update_read_index(&trx_read_rbuf);
+  log_debug("TRX_ECPRI_READ", "count = %ld (%li)", read_count, read_counter.counter);
+
+  while(rbuf_read_amount(&trxr_rbuf[0]) < read_count);
+  
+  sync_complete = 1;
+
+  n = read_count;
+  while((nc = rbuf_contiguous_copy(&trxr_rbuf[0], NULL, n))) {
+    int len = nc * trxr_rbuf[0].len * sizeof(Complex);
+    for(int i = 0; i < s->rx_n_channel; i++ ) {
+      uint8_t * dst = (uint8_t*) (_samples[i] + offset);
+      uint8_t * src = ((uint8_t *) trxr_rbuf[i].buffer) + trxr_rbuf[0].read_index * trxr_rbuf[0].len * sizeof(Complex);
+      memcpy(dst, src, len);
+    }
+    trxr_rbuf[0].read_index = (trxr_rbuf[0].read_index + nc) % trxr_rbuf[0].buf_len;
+    n -= nc;
+    offset += len;
   }
-  *ptimestamp = recv_counter.counter * M;
+
+  *ptimestamp = (read_counter.counter) * M;
   update_counter(&read_counter, read_count);
 
   return count;
@@ -739,7 +949,7 @@ static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__sa
    rate. Here we don't implement it, so the user has to force a given
    sample rate with the "sample_rate" configuration option */
 static int trx_ecpri_get_sample_rate(TRXState *s1, TRXFraction *psample_rate,
-                                     int *psample_rate_num, int sample_rate_min)
+    int *psample_rate_num, int sample_rate_min)
 {
   return -1;
 }
@@ -752,7 +962,7 @@ static int trx_ecpri_start(TRXState *s1, const TRXDriverParams *params)
 
   log_info("TRX_ECPRI_START", "trx_api_version: %d", s1->trx_api_version);
   log_info("TRX_ECPRI_START", "config file: %s", s1->path);
-  log_info("TEST-DPDK-ECPRI", "rec-mac: %s, re-mac: %s, rec-if: %s", s->rec_mac, s->re_mac, s->rec_if);
+  log_info("TEST-DPDK-ECPRI", "bbu-mac: %s, rrh-mac: %s, bbu-if: %s", s->bbu_mac, s->rrh_mac, s->bbu_if);
 
   s->sample_rate = params->sample_rate[0].num / params->sample_rate[0].den;
 
@@ -768,13 +978,183 @@ void dummy_enb_init(TRXState *s1, TRXEcpriState *s) {
   start(s);
 }
 
+/* Called to start the tranceiver. Return 0 if OK, < 0 if  */
+int trx_start_func(TRXState *s, const TRXDriverParams *p) {
+  log_info("DEBUG", "trx_start_func");
+  return 0;
+}
+
+/* Deprecated, use trx_write_func2 instead.
+   Write 'count' samples on each channel of the TX port
+   'tx_port_index'. samples[0] is the array for the first
+   channel. timestamp is the time (in samples) at which the first
+   sample must be sent. When the TRX_WRITE_FLAG_PADDING flag is
+   set, samples is set to NULL. It indicates that no data should
+   be sent (TDD receive time). TRX_WRITE_FLAG_END_OF_BURST is set
+   to indicate in advance that the next write call will have the
+   TRX_WRITE_FLAG_PADDING flag set. Note:
+   TRX_WRITE_FLAG_END_OF_BURST and TRX_WRITE_FLAG_PADDING are
+   never set simultaneously.
+ */
+void trx_write_func(TRXState *s, trx_timestamp_t timestamp, const void **samples, int count, int flags, int tx_port_index) {
+  log_info("DEBUG", "**");
+}
+
+/* Deprecated, use trx_read_func2 instead.
+   Read 'count' samples from each channel. samples[0] is the array
+   for the first channel. *ptimestamp is the time at which the
+   first samples was received. Return the number of sample read
+   (=count).
+
+   Note: It is explicitely allowed that the application calls
+   trx_write_func, trx_read_func, trx_set_tx_gain_func and
+   trx_set_rx_gain_func from different threads.
+*/
+int trx_read_func(TRXState *s, trx_timestamp_t *ptimestamp, void **samples, int count, int rx_port_index) {
+  log_info("DEBUG", "**");
+  return 0;
+}
+
+/* Dynamic set the transmit gain (in dB). The origin and range are
+   driver dependent.
+
+   Note: this function is only used for user supplied dynamic
+   adjustements.
+*/
+void trx_set_tx_gain_func(TRXState *s, double gain, int channel_num) {
+  log_info("DEBUG", "trx_set_tx_gain_func");
+}
+
+/* Dynamic set the receive gain (in dB). The origin and range are
+   driver dependent.
+
+   Note: this function is only used for user supplied dynamic
+   adjustements.
+*/
+void trx_set_rx_gain_func(TRXState *s, double gain, int channel_num) {
+  log_info("DEBUG", "trx_set_rx_gain_func");
+}
+
+/* Return the maximum number of samples per TX packet. Called by
+ * the application after trx_start_func.
+ * Optional
+ */
+int trx_get_tx_samples_per_packet_func(TRXState *s) {
+  log_info("DEBUG", "trx_get_tx_samples_per_packet_func");
+  return 0;
+}
+
+/* Return some statistics. Return 0 if OK, < 0 if not available. */
+int trx_get_stats(TRXState *s, TRXStatistics *m) {
+  log_info("DEBUG", "trx_get_stats");
+  return 0;
+}
+
+/* Callback must allocate info buffer that will be displayed */
+void trx_dump_info(TRXState *s, trx_printf_cb cb, void *opaque) {
+  log_info("DEBUG", "*opaque");
+}
+
+/* Return the absolute TX power in dBm for the TX channel
+   'channel_num' assuming a square signal of maximum
+   amplitude. This function can be called from any thread and
+   needs to be fast. Return 0 if OK, -1 if the result is not
+   available. */
+int trx_get_abs_tx_power_func(TRXState *s,
+                                 float *presult, int channel_num) {
+  log_info("DEBUG", "trx_get_abs_tx_power_func");
+  return 0;
+}
+
+/* Return the absolute RX power in dBm for the RX channel
+   'channel_num' assuming a square signal of maximum
+   amplitude. This function can be called from any thread and
+   needs to be fast. Return 0 if OK, -1 if the result is not
+   available. */
+int trx_get_abs_rx_power_func(TRXState *s,
+                                 float *presult, int channel_num) {
+  log_info("DEBUG", "trx_get_abs_rx_power_func");
+  return 0;
+}
+
+/* Remote API communication
+ * Available since API v14
+ * trx_msg_recv_func: called for each trx received messages
+ * trx_msg_send_func: call it to send trx messages (They must be registered by client)
+ * For each message, a call to send API must be done
+ */
+void trx_msg_recv_func(TRXState *s, TRXMsg *msg) {
+  log_info("DEBUG", "trx_msg_recv_func");
+}
+
+TRXMsg* trx_msg_send_func(TRXState *s) {
+  log_info("DEBUG", "trx_msg_send_func");
+  return NULL;
+}
+
+/* Return actual transmit gain (in dB). The origin and range are
+   driver dependent.
+*/
+void trx_get_tx_gain_func(TRXState *s, double *gain, int channel_num) {
+  log_info("DEBUG", "trx_get_tx_gain_func");
+}
+
+/* Returns actual receive gain (in dB). The origin and range are
+   driver dependent.
+*/
+void trx_get_rx_gain_func(TRXState *s, double *gain, int channel_num) {
+  log_info("DEBUG", "trx_get_rx_gain_func");
+}
+
+/* Stop operation of the transceiver - to be called after trx_start.
+   resources allocated in init are not released, so trx_call can be called again */
+void trx_stop_func(TRXState *s) {
+  log_info("DEBUG", "trx_stop_func");
+}
+
+/* OFDM mode: experimental 7.2 API */
+
+/* read the current timestamp (only used in OFDM mode). Return 0
+   if OK, < 0 if not supported by device. */
+int trx_read_timestamp(TRXState *s, trx_timestamp_t *ptimestamp,
+                          int port_index) {
+  log_info("DEBUG", "trx_read_timestamp");
+  return 0;
+}
+
+void trx_set_tx_streams(TRXState *s, int rf_port_index,
+                           int n_streams, const TRXOFDMStreamInfo *streams) {
+  log_info("DEBUG", "trx_set_tx_streams");
+}
+void trx_set_rx_streams(TRXState *s, int rf_port_index,
+                           int n_streams, const TRXOFDMStreamInfo *streams) {
+  log_info("DEBUG", "trx_set_rx_streams");
+}
+/* schedule the reading of OFDM symbols. Return 0 if OK, < 0 if
+   error. */
+int trx_schedule_read(TRXState *s, int rf_port_index,
+                         const TRXScheduledSymbol *symbols,
+                         int n_symbols) {
+  log_info("DEBUG", "trx_schedule_read");
+  return 0;
+}
+
+/* AGC functions */
+int trx_set_agc_func(TRXState *s, const TRXAGCParams *p, int channel) {
+  log_info("DEBUG", "trx_set_agc_func");
+  return 0;
+}
+int trx_get_agc_func(TRXState *s, TRXAGCParams *p, int channel) {
+  log_info("DEBUG", "trx_get_agc_func");
+  return 0;
+}
+
 int trx_driver_init(TRXState *s1)
 {
   TRXEcpriState *s;
   double val;
 
-  // Lock all current and future pages from preventing of being paged to
-  // swap
+  // Lock all current and future pages from preventing of being paged to swap
   if (mlockall(MCL_CURRENT | MCL_FUTURE)) {
     log_error("TRX_ECPRI", "mlockall failed");
   }
@@ -782,9 +1162,9 @@ int trx_driver_init(TRXState *s1)
   log_info("TRX_ECPRI", "Init");
 
   if (s1->trx_api_version != TRX_API_VERSION) {
-      fprintf(stderr, "ABI compatibility mismatch between LTEENB and TRX driver (LTEENB ABI version=%d, TRX driver ABI version=%d)\n",
-              s1->trx_api_version, TRX_API_VERSION);
-      return -1;
+    fprintf(stderr, "ABI compatibility mismatch between LTEENB and TRX driver (LTEENB ABI version=%d, TRX driver ABI version=%d)\n",
+        s1->trx_api_version, TRX_API_VERSION);
+    return -1;
   }
 
   s = malloc(sizeof(TRXEcpriState));
@@ -794,29 +1174,85 @@ int trx_driver_init(TRXState *s1)
   s->recv_affinity = (int) val;
   trx_get_param_double(s1, &val, "send_affinity");
   s->send_affinity = (int) val;
-  trx_get_param_double(s1, &val, "prepare_affinity");
-  s->send_affinity = (int) val;
-  trx_get_param_double(s1, &val, "decompress_affinity");
-  s->send_affinity = (int) val;
+  trx_get_param_double(s1, &val, "encode_affinity");
+  s->encode_affinity = (int) val;
+  trx_get_param_double(s1, &val, "decode_affinity");
+  s->decode_affinity = (int) val;
+  trx_get_param_double(s1, &val, "statistic_affinity");
+  s->statistic_affinity = (int) val;
   trx_get_param_double(s1, &val, "flow_id");
   s->flow_id = (int) val;
+  trx_get_param_double(s1, &val, "frame_frequency");
+  s->frame_frequency = (int) val;
+  trx_get_param_double(s1, &val, "trx_buf_size");
+  s->trx_buf_size = (int) val;
+  trx_get_param_double(s1, &val, "txrx_buf_size");
+  s->txrx_buf_size = (int) val;
+  trx_get_param_double(s1, &val, "trace_rx");
+  s->trace_rx = (int) val;
+  trx_get_param_double(s1, &val, "trace_tx");
+  s->trace_tx = (int) val;
+  trx_get_param_double(s1, &val, "trace_offset");
+  s->trace_offset = (int) val;
+  trx_get_param_double(s1, &val, "monitor_pps");
+  s->monitor_pps = (int) val;
+  trx_get_param_double(s1, &val, "monitor_trigger_duration");
+  s->monitor_trigger_duration = (int) val;
+  trx_get_param_double(s1, &val, "start_sending");
+  s->start_sending = (int) val;
+  trx_get_param_double(s1, &val, "start_receiving");
+  s->start_receiving = (int) val;
+  trx_get_param_double(s1, &val, "rx_n_channel");
+  s->rx_n_channel = (int) val;
+  trx_get_param_double(s1, &val, "tx_n_channel");
+  s->tx_n_channel = (int) val;
+  trx_get_param_double(s1, &val, "statistics_refresh_rate_ns");
+  s->statistics_refresh_rate_ns = (int) val;
   trx_get_param_double(s1, &val, "ecpri_period");
-  if(((int) val) == 0) {
-    fprintf(stderr, "ecpri_period parameter can't be null\n");
-    return -1;
-  }
   s->ecpri_period = (int) val;
 
-  s->re_mac = trx_get_param_string(s1, "re_mac");
-  s->rec_mac = trx_get_param_string(s1, "rec_mac");
-  s->rec_if = trx_get_param_string(s1, "rec_if");
+  if(s->ecpri_period == 0)
+    log_exit("TRX_ECPRI", "ecpri_period parameter can't be null\n");
+  if(s->rx_n_channel == 0)
+    log_exit("TRX_ECPRI", "rx_n_channel parameter can't be null\n");
+  if(s->tx_n_channel == 0)
+    log_exit("TRX_ECPRI", "tx_n_channel parameter can't be null\n");
+
+  s->rrh_mac = (uint8_t *) trx_get_param_string(s1, "rrh_mac");
+  s->bbu_mac = (uint8_t *) trx_get_param_string(s1, "bbu_mac");
+  s->bbu_if = (uint8_t *) trx_get_param_string(s1, "bbu_if");
+  s->dpdk_options = trx_get_param_string(s1, "dpdk_options");
   s->log_directory = (uint8_t *) trx_get_param_string(s1, "log_directory");
 
+  recv_pps_threshold = (s->frame_frequency * 9 / 10);
+
   s1->opaque = s;
   s1->trx_end_func = trx_ecpri_end;
   s1->trx_write_func2 = trx_ecpri_write;
   s1->trx_read_func2 = trx_ecpri_read;
   s1->trx_start_func = trx_ecpri_start;
   s1->trx_get_sample_rate_func = trx_ecpri_get_sample_rate;
+
+  //s1->trx_write_func = trx_write_func;
+  //s1->trx_read_func = trx_read_func;
+  //s1->trx_set_tx_gain_func = trx_set_tx_gain_func;
+  //s1->trx_set_rx_gain_func = trx_set_rx_gain_func;
+  //s1->trx_get_tx_samples_per_packet_func = trx_get_tx_samples_per_packet_func;
+  //s1->trx_get_stats = trx_get_stats;
+  //s1->trx_dump_info = trx_dump_info;
+  //s1->trx_get_abs_tx_power_func = trx_get_abs_tx_power_func;
+  //s1->trx_get_abs_rx_power_func = trx_get_abs_rx_power_func;
+  //s1->trx_msg_recv_func = trx_msg_recv_func;
+  //s1->trx_get_tx_gain_func = trx_get_tx_gain_func;
+  //s1->trx_get_rx_gain_func = trx_get_rx_gain_func;
+  //s1->trx_stop_func = trx_stop_func;
+  //s1->trx_read_timestamp = trx_read_timestamp;
+  //s1->trx_set_tx_streams = trx_set_tx_streams;
+  //s1->trx_set_rx_streams = trx_set_rx_streams;
+  //s1->trx_schedule_read = trx_schedule_read;
+  //s1->trx_set_agc_func = trx_set_agc_func;
+  //s1->trx_get_agc_func = trx_get_agc_func;
+  //s1->trx_msg_send_func = trx_msg_send_func;
+
   return 0;
 }