Optimize reception and fix PPS stats

ecpri-tests/test-dpdk-ecpri.c

@@ -95,37 +95,59 @@ int main(int argc, char * argv[]) {
   s = malloc(sizeof(TRXEcpriState));
   memset(s, 0, sizeof(*s));
 
-  // Tiogapass001 mellanox
-  //s->rec_mac = "b8:59:9f:07:7d:da";
-  //s->re_mac = "04:09:a5:0f:9f:4c";
-  //s->rec_if = "ens9f0np0";
-
-  // Loopback
-  //s->rec_mac = "b8:ce:f6:4b:00:22";
-  //s->re_mac = "b8:ce:f6:4b:00:23";
-  //s->rec_if = "ens5f0np0";
-
-  //s->rec_mac = "b4:96:91:a7:1c:f4";
-  //s->re_mac = "04:09:a5:0f:9f:4c";
-  //s->rec_if = "ens5f0";
-
-  // tiogapass003 mellanox
-  //s->rec_mac = "b8:59:9f:07:7e:2a";
-  //s->re_mac = "b8:59:9f:07:86:42";
-  //s->rec_if = "ens9f0";
-  //s->dpdk_options = " -l 28 -b 0000:04:00.0 -b 0000:5e:00.1 ";
-
-  // HFR tiogapass mellanox
+#if 0
+  // tiogapass-003 MT27710
+  s->rec_mac = "b8:59:9f:07:7e:2a";
+  //s->re_mac = "04:09:a5:0f:9f:4c"; // Lille M6424 Switch
+  s->re_mac = "b8:59:9f:07:86:42"; // tiogapass-004 MT27710
+  //s->re_mac = "b4:96:91:a7:1c:f4"; // tiogapass-004 XXV710DA2T port0
+  s->rec_if = "ens9f0";
+  s->dpdk_options = "-l 10,20 -b 0000:04:00.0 -b 0000:5e:00.1 ";
+#endif
+
+#if 0
+  // tiogapass-003 MT27710 port1
+  s->rec_mac = "b8:59:9f:07:7e:2b";
+  s->re_mac = "04:09:a5:0f:9f:4c"; // Lille M6424 Switch
+  //s->re_mac = "b8:59:9f:07:86:42"; // tiogapass-004 MT27710
+  //s->re_mac = "b4:96:91:a7:1c:f5"; // tiogapass-004 XXV710DA2T port1
+  s->rec_if = "ens9f1";
+  s->dpdk_options = "-l 10,20 -b 0000:04:00.0 -b 0000:5e:00.0 ";
+#endif
+
+#if 0
+  // hfr-tiogapass-001 MT27710
   s->rec_mac = "b8:59:9f:07:82:ca";
-  s->re_mac = "04:09:a5:0f:76:1c";
+  s->re_mac = "04:09:a5:0f:76:1c"; // HFR M6424 switch
   s->rec_if = "ens9f0";
-  s->dpdk_options = " -l 28 -b 0000:04:00.0 -b 0000:5e:00.1 ";
-
-  // tiogapass004 mellanox
-  //s->rec_mac = "b8:59:9f:07:86:42";
-  //s->re_mac = "b8:59:9f:07:7e:2a";
-  //s->rec_if = "ens9f0";
-  //s->dpdk_options = " -l 28 -b 0000:04:00.0 -b 0000:3b:00.0 -b 0000:3b:00.1 -b 0000:5e:00.1 ";
+  s->dpdk_options = "-l 10,20 -b 0000:04:00.0 -b 0000:18:00.0 -b 0000:18:00.1 -b 0000:5e:00.1 ";
+#endif
+
+#if 0
+  // hfr-tiogapass-001 XXV710DA2T
+  s->rec_mac = "b4:96:91:a7:1b:28";
+  s->re_mac = "04:09:a5:0f:76:1c"; // HFR M6424 switch
+  s->rec_if = "ens1f0";
+  s->dpdk_options = "-l 10,20 -b 0000:04:00.0 -b 0000:18:00.1 -b 0000:5e:00.0 -b 0000:5e:00.1 ";
+#endif
+
+#if 1
+  // tiogapass-004 MT27710
+  s->rec_mac = "b8:59:9f:07:86:42";
+  //s->re_mac = "04:09:a5:0f:9f:4c"; // Lille M6424 Switch
+  s->re_mac = "b8:59:9f:07:7e:2a"; // tiogapass-003 MT27710
+  s->rec_if = "ens9f0";
+  s->dpdk_options = "-l 10,20 -b 0000:04:00.0 -b 0000:3b:00.0 -b 0000:3b:00.1 -b 0000:5e:00.1 ";
+#endif
+
+#if 0
+  // tiogapass-004 XXV710DA2T
+  s->rec_mac = "b4:96:91:a7:1c:f4";
+  s->re_mac = "04:09:a5:0f:9f:4c"; // Lille M6424 Switch
+  s->re_mac = "b8:59:9f:07:7e:2a"; // tiogapass-003 MT27710
+  s->rec_if = "ens5f0";
+  s->dpdk_options = "-l 10,20 -b 0000:04:00.0 -b 0000:3b:00.1 -b 0000:5e:00.0 -b 0000:5e:00.1 ";
+#endif
 
   s->recv_affinity = 39;
   s->send_affinity = 38;

ecpri-tests/test-lte.sh

+#!/bin/bash
+
+cd ..;
+make;
+cd ecpri-tests;
+make all;
+export LD_LIBRARY_PATH="/root/enb"
+~/enb/lteenb ~/enb/config/enb.cfg

trx_ecpri_dpdk.c

@@ -46,7 +46,7 @@
 #define DEBUG
 #define SSE4 /* define if CPU supports SSE4.1 */
 // Tiogapass004
-//#define DST_ADDR_SYNTAX
+#define DST_ADDR_SYNTAX
 
 #include "private.c"
 
@@ -157,8 +157,6 @@ void set_latency_target(void) {
   printf("# /dev/cpu_dma_latency set to %dus\n", latency_target_value);
 }
 
-
-
 typedef struct {
   volatile void * buffer;
   char name[64];
@@ -168,6 +166,13 @@ typedef struct {
   volatile int read_index;
 } ring_buffer_t;
 
+typedef struct {
+  int64_t counter;
+  int64_t pps_counter;
+  int64_t pps_ts;
+  int64_t pps;
+} counter_stat_t;
+
 typedef struct {
   const char * re_mac;
   const char * rec_mac;
@@ -203,10 +208,13 @@ pthread_mutex_t tx_ready_mutex;
 pthread_cond_t tx_ready_cond;
 sem_t trx_read_sem;
 // Counters
-static volatile int64_t prepared_frame_count; // compressed samples
-static volatile int64_t read_frame_count; // frames passed to amarisoft stack
-static volatile int64_t sent_frame_count; // frames sent to eRE
-static volatile int64_t recv_frame_count; // frames received from eRE
+static volatile counter_stat_t prepared_counter; // compressed samples
+static volatile counter_stat_t read_counter; // frames passed to amarisoft stack
+static volatile counter_stat_t sent_counter; // frames sent to eRE
+static volatile counter_stat_t recv_counter; // frames received from eRE
+
+#define STAT_FRAME_INTERVAL INT64_C(3800000)
+
 // Computed values
 static int rxtx_buf_size;
 static int ecpri_period_mult;
@@ -380,13 +388,15 @@ static void send_packets(int port) {
 // TODO store received packets' data in buffer
 static int recv_packets(int port) {
   struct rte_mbuf * pkt[1024];
+  uint8_t * buf;
+  uint8_t * rtebuf;
   while(1) {
     const int nb_rx = rte_eth_rx_burst(port, 0, pkt, 1024);
 
     for(int i = 0; i < nb_rx; i++) {
-      for(int j = 0; j < PACKET_SIZE; j++) {
-        *(RBUF_WRITE(rx_rbuf, uint8_t) + j) = *rte_pktmbuf_mtod_offset(pkt[i], uint8_t *, j);
-      }
+      buf = ((uint8_t *) rx_rbuf.buffer) + (rx_rbuf.write_index * rx_rbuf.len);
+      rtebuf = (uint8_t *) (pkt[i])->buf_addr + (pkt[i])->data_off;
+      memcpy(buf, rtebuf, PACKET_SIZE);
       rbuf_update_write_index(&rx_rbuf);
       rte_pktmbuf_free(pkt[i]);
     }
@@ -397,6 +407,27 @@ static int recv_packets(int port) {
 }
 /* DPDK */
 
+static void init_counter(volatile counter_stat_t * c) {
+  c->counter = 0;
+  c->pps_counter = 0;
+  c->pps_ts = 0;
+  c->pps = 0;
+}
+static void update_counter(volatile counter_stat_t * c, int64_t v) {
+  struct timespec _ts;
+  int64_t ts;
+  c->counter += v;
+  c->pps_counter += v;
+  if(c->pps_counter >= STAT_FRAME_INTERVAL) {
+    clock_gettime(CLOCK_TAI, &_ts);
+    ts = ts_to_int(_ts);
+    if(c->pps_ts)
+      c->pps = (c->pps_counter * NSEC_PER_SEC) / (ts - c->pps_ts);
+    c->pps_counter = 0;
+    c->pps_ts = ts;
+  }
+}
+
 static void *recv_thread(void *p) {
 
   cpu_set_t mask;
@@ -412,7 +443,7 @@ static void *recv_thread(void *p) {
 
   for(;;) {
 
-    recv_frame_count += recv_packets(0);
+    update_counter(&recv_counter, recv_packets(0));
 
     //for(int j = 0; j < ecpri_period_mult; j++) {
       // TODO write rx_buf
@@ -450,7 +481,7 @@ static void *send_thread(void *p) {
       int64_t d;
       clock_gettime(CLOCK_TAI, &next);
       d = calcdiff_ns(next, initial);
-      log_debug("SEND_THREAD", "Packets sent: %" PRIi64, sent_frame_count);
+      log_debug("SEND_THREAD", "Packets sent: %" PRIi64, sent_counter.counter);
       log_debug("SEND_THREAD", "Duration: %" PRIi64, d);
       log_debug("SEND_THREAD", "ecpri_period_mult: %" PRIi64, ecpri_period_mult);
       log_debug("SEND_THREAD", "FRAME_FREQ: %" PRIi64, FRAME_FREQ);
@@ -468,7 +499,7 @@ static void *send_thread(void *p) {
         rbuf_update_read_index(&tx_rbuf);
       }
       send_packets(0);
-      sent_frame_count += BURST_SIZE;
+      update_counter(&sent_counter, BURST_SIZE);
     }
 
     pthread_mutex_lock(&tx_mutex);
@@ -502,13 +533,13 @@ static void *prepare_thread(void *p) {
       // 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;
+        int empty_frames_ahead = ts - prepared_counter.counter;
         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++;
+            update_counter(&prepared_counter, 1);
           }
         }
         else if (empty_frames_ahead == 0) {
@@ -529,7 +560,7 @@ static void *prepare_thread(void *p) {
 
             rbuf_update_write_index(&tx_rbuf);
             rbuf_update_read_index(&trx_write_rbuf);
-            prepared_frame_count++;
+            update_counter(&prepared_counter, 1);
           }
           if(m == 0)
             trx_wb_part_read_index = (trx_wb_part_read_index + 1) % TRX_WB_MAX_PARTS;
@@ -541,7 +572,7 @@ static void *prepare_thread(void *p) {
       else {
         *((uint16_t *) (RBUF_WRITE(tx_rbuf, uint8_t) + 6)) = htons(seq_id++);
         rbuf_update_write_index(&tx_rbuf);
-        prepared_frame_count++;
+        update_counter(&prepared_counter, 1);
       }
     }
     else {
@@ -631,16 +662,11 @@ static void *decompress_thread(void *p) {
   pthread_exit(EXIT_SUCCESS);
 }
 static void *statistic_thread(void *p) {
-  struct timespec next, initial, current;
+  struct timespec next, initial;
   cpu_set_t mask;
   int64_t duration_ns;
   TRXEcpriState * s = (TRXEcpriState *) p;
 
-  const int div_a[] = {1,2,4,5,8,10,16,20,25,32,40,50,64,80,100,125,128,160,200,250,256,320,400,500,512,625,640,800,1000,1250,1280,1600,2000,2500,2560,3125,3200,4000,5000,6250,6400,8000,10000,12500,12800,15625,16000,20000,25000,31250};
-  const int div_b[] = {1000000000,500000000,250000000,200000000,125000000,100000000,62500000,50000000,40000000,31250000,25000000,20000000,15625000,12500000,10000000,8000000,7812500,6250000,5000000,4000000,3906250,3125000,2500000,2000000,1953125,1600000,1562500,1250000,1000000,800000,781250,625000,500000,400000,390625,320000,312500,250000,200000,160000,156250,125000,100000,80000,78125,64000,62500,50000,40000,32000};
-  int div_index = 0;
-  int div_indexr = 0;
-
   log_info("STATISTIC_THREAD", "Thread init");
 
   // Set thread CPU affinity
@@ -659,23 +685,14 @@ static void *statistic_thread(void *p) {
         "pps",
         "ppsr");
   for(;;) {
-    int64_t pps, ppsr;
     add_ns(&next, STATISTIC_REFRESH_RATE);
-    clock_gettime(CLOCK_TAI, &current);
-    duration_ns = calcdiff_ns(current, initial);
-    if(sent_frame_count >= INT64_MAX / div_b[div_index])
-      div_index++;
-    if(recv_frame_count >= INT64_MAX / div_b[div_indexr])
-      div_indexr++;
-    pps = ((sent_frame_count * div_b[div_index]) / duration_ns) * div_a[div_index];
-    ppsr = ((recv_frame_count * div_b[div_indexr]) / duration_ns) * div_a[div_indexr];
     log_info("STATS", "%14" PRIi64 " - %14" PRIi64 " - %14" PRIi64 " - %14" PRIi64 " - %14" PRIi64 "pps %14" PRIi64 "pps",
-        prepared_frame_count,
-        read_frame_count,
-        sent_frame_count,
-        recv_frame_count,
-        pps,
-        ppsr);
+        prepared_counter.counter,
+        read_counter.counter,
+        sent_counter.counter,
+        recv_counter.counter,
+        sent_counter.pps,
+        recv_counter.pps);
     clock_nanosleep(CLOCK_TAI, TIMER_ABSTIME, &next, NULL);
   }
   pthread_exit(EXIT_SUCCESS);
@@ -781,23 +798,24 @@ static int start_threads(TRXEcpriState * s) {
 int startdpdk(TRXEcpriState * s) {
   uint8_t ecpri_message[DATA_SIZE];
 
-  int argc = 0;
-  int k = 0;
-  int prev_space = 0;
+  int argc = 1;
+  int k = 1;
+  int prev_space = -1;
   char ** argv;
 
-  for(int i = 0;; i++)
+  for(int i = 0;; i++) {
     if(s->dpdk_options[i] == ' ')
       argc++;
     else if(s->dpdk_options[i] == '\0')
       break;
+  }
   argv = (char **) malloc(sizeof(char *) * argc);
 
   for(int i = 0;; i++) {
     if(s->dpdk_options[i] == ' ') {
-      argv[k] = (char *) malloc(i - prev_space + 1);
-      strncpy(argv[k], s->dpdk_options, i - prev_space);
-      argv[k][i - prev_space] = '\0';
+      argv[k] = (char *) malloc(i - prev_space);
+      strncpy(argv[k], s->dpdk_options + prev_space + 1, i - prev_space -1);
+      argv[k][i - prev_space-1] = '\0';
       prev_space = i;
       k++;
     }
@@ -807,15 +825,15 @@ int startdpdk(TRXEcpriState * s) {
   }
 
   init_dpdk(argc, argv);
-
   log_debug("TRX_ECPRI", "start");
 
   //set_latency_target();
 
   seq_id = 0;
-  read_frame_count = 0;
-  sent_frame_count = 0;
-  prepared_frame_count = 0;
+  init_counter(&read_counter);
+  init_counter(&sent_counter);
+  init_counter(&prepared_counter);
+  init_counter(&recv_counter);
 
   ecpri_period_mult = (s->ecpri_period * FRAME_FREQ) / 1000000;
   rxtx_buf_size = (3 * ecpri_period_mult);
@@ -910,8 +928,8 @@ static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__sa
         _samples[i][j + (k << 6)] = trx_samples[i * 64 + j];
     rbuf_update_read_index(&trx_read_rbuf);
   }
-  *ptimestamp = read_frame_count << 5;
-  read_frame_count += read_count;
+  *ptimestamp = read_counter.counter << 5;
+  update_counter(&read_counter, read_count);
 
   return count;
 }