WIP: One Way Delay Measure

trx_ecpri_dpdk.c

@@ -100,8 +100,10 @@ typedef struct {
   int flow_id;
   int start_receiving;
   int master;
-  int generic_data_sync;
   int trx_read_null;
+  int one_way_measure;
+  int one_way_period;
+  int tdd_frame_start;
 } TRXEcpriState;
 
 typedef struct {
@@ -141,6 +143,10 @@ static ring_buffer_t tx_rbuf; // Packets to send
 static ring_buffer_t trxw_rbuf[4]; // Uncompressed IQ samples
 static ring_buffer_t trxw_group_rbuf; // Group of IQ samples
 
+static ring_buffer_t one_way_rbuf; // One way delay measurements
+
+static volatile uint64_t one_way_timestamp;
+
 // Counters
 static volatile counter_stat_t recv_counter; // frames received from eRE
 static volatile counter_stat_t decode_counter; // decoded frames
@@ -704,17 +710,30 @@ static void *send_thread(void *p) {
   pthread_exit(EXIT_SUCCESS);
 }
 
-static void encode_empty_frames(int n, int tx_n_channel, int trx, int tdd_period) {
+static int one_way_delay_measure(uint8_t ** buf) {
+  struct timespec t;
+  clock_gettime(CLOCK_TAI, &t);
+  one_way_timestamp = ts_to_int(t);
+  *(            (*buf - 7)) = 3;                   // eCPRI Type
+  *(            (*buf - 4)) = 0;                   // Measurement ID
+  *(            (*buf - 3)) = 0;                   // Action Type
+  *((uint64_t *)(*buf - 2)) = one_way_timestamp;   // Timestamp
+  *((uint64_t *)(*buf + 7)) = 0;                   // Compensation
+  *buf += tx_rbuf.len;
+  return 1;
+}
+
+static void encode_empty_frames(int n, int trx, TRXEcpriState * s) {
   uint8_t * buf = RBUF_WRITE0(tx_rbuf, uint8_t) + 8;
   int n2 = n;
   for(int i = 0; i < n; i++) {
-    //if(trx && !((trx_encode_counter.counter + i) % tdd_period)) {
-    //  *(buf - 7) = 3;
-    //  *((uint16_t *)(buf - 2)) = htons(tx_seq_id++);
-    //  buf += tx_rbuf.len;
-    //  n2++;
-    //}
-    memset(buf, 0x00, 60 * tx_n_channel);
+    // ONE WAY DELAY MEASURE
+    if(s->one_way_measure && (encode_counter.counter + i) % s->one_way_period)
+      n2 += one_way_delay_measure(&buf);
+    // TDD FRAME START
+    if(s->tdd_frame_start)
+      *((uint8_t *)(buf + 60 * s->tx_n_channel)) = (trx && !((trx_encode_counter.counter + i) % s->tdd_period));
+    memset(buf, 0x00, 60 * s->tx_n_channel);
     *((uint16_t *)(buf - 2)) = htons(tx_seq_id++);
     buf += tx_rbuf.len;
   }
@@ -722,27 +741,27 @@ static void encode_empty_frames(int n, int tx_n_channel, int trx, int tdd_period
   if(trx)
     trxw_rbuf[0].read_index = (trxw_rbuf[0].read_index + n) % trxw_rbuf[0].buf_len;
 }
-static void encode_trx_frames(int n, int tx_n_channel, int tdd_period) {
+static void encode_trx_frames(int n, TRXEcpriState * s) {
   int nc;
   int nf = n;
   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;
     int nc2 = nc;
-    for(int j = 0; j < tx_n_channel; j++)
+    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++) {
-      //if(!((trx_encode_counter.counter + i) % tdd_period)) {
-      //  *(buf - 7) = 3;
-      //  *((uint16_t *)(buf - 2)) = htons(tx_seq_id++);
-      //  buf += tx_rbuf.len;
-      //  nc2++;
-      //}
-      for(int i = 0; i < tx_n_channel ; i++)
+      // ONE WAY DELAY MEASURE
+      if(s->one_way_measure && (encode_counter.counter + i) % s->one_way_period)
+        nc2 += one_way_delay_measure(&buf);
+      // TDD FRAME START
+      if(s->tdd_frame_start)
+        *((uint8_t *)(buf + 60 * s->tx_n_channel)) = (trx && !((trx_encode_counter.counter + i) % s->tdd_period));
+      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(tx_seq_id++);
-      for(int j = 0; j < tx_n_channel; j++)
+      for(int j = 0; j < s->tx_n_channel; j++)
         iq_samples[j] += trxw_rbuf[0].len;
       buf += tx_rbuf.len;
     }
@@ -781,9 +800,9 @@ int read_trx(int n_max, TRXEcpriState * s) {
     }
 
     if(g->zeroes)
-      encode_empty_frames(n_trx, s->tx_n_channel, 1, s->tdd_period);
+      encode_empty_frames(n_trx, 1, s);
     else
-      encode_trx_frames(n_trx, s->tx_n_channel, s->tdd_period);
+      encode_trx_frames(n_trx, s);
 
     if(!g->count) {
       rbuf_update_read_index(&trxw_group_rbuf);
@@ -830,7 +849,7 @@ static void *encode_thread(void *p) {
 
     n = rbuf_write_amount(&tx_rbuf);
     n_min = s->encode_burst;
-    n_min += s->generic_data_sync ? (s->encode_burst / s->tdd_period) + 1 : 0;
+    n_min += s->one_way_measure ? (s->encode_burst / s->one_way_period) + 1 : 0;
     if(s->master && n < n_min)
       log_exit("ENCODE_THREAD", "Not enough space in TX RBUF (%d < %d)\n", n, s->encode_burst);
 
@@ -841,7 +860,7 @@ static void *encode_thread(void *p) {
     if(s->master) {
       struct timespec current;
       n_empty = s->encode_burst - n_trx;
-      encode_empty_frames(n_empty, s->tx_n_channel, 0, s->tdd_period);
+      encode_empty_frames(n_empty, 0, s);
       target_counter += s->encode_burst;
       next = int_to_ts(initial_ts + target_counter * NSEC_PER_SEC / s->frame_frequency);
       do {
@@ -902,6 +921,7 @@ static void *decode_thread(void *p) {
     while((nc = rbuf_contiguous_copy(&rx_rbuf, &trxr_rbuf[0], n))) {
 
       uint8_t * buf = ((uint8_t *) rx_rbuf.buffer) + (rx_rbuf.read_index * rx_rbuf.len) + 22;
+      int one_way_count = 0;
   
       if(s->trace_rx) {
         if(received_pkts && ((decode_counter.counter + nc) >= (trxr_rbuf[0].buf_len + s->trace_offset))) {
@@ -918,6 +938,15 @@ static void *decode_thread(void *p) {
         iq_samples[i] = (((Complex *) trxr_rbuf[i].buffer) + (trxr_rbuf[0].write_index * trxr_rbuf[0].len));
 
       for(int i = 0; i < nc; i++) {
+
+        // ONE WAY MEASURE
+        if(s->one_way_measure && *((buf + i * rx_rbuf.len) - 7) == 3) {
+          uint64_t timestamp = (uint64_t) *(buf + 7);
+          *(RBUF_WRITE0(one_way_rbuf, int64_t)) = ((int64_t) timestamp) - ((int64_t) one_way_timestamp);
+          rbuf_update_write_index(&one_way_rbuf);
+          continue;
+        }
+
         for(int j = 0; j < s->rx_n_channel ; j++) {
           decode_s64_b60_2((float *) (iq_samples[j] + i * 32), buf + j * 60 + i * rx_rbuf.len);
         }
@@ -1168,8 +1197,8 @@ int startdpdk(TRXEcpriState * s) {
   init_counter(&sent_counter);
   init_counter(&empty_encode_counter);
 
-  RBUF_INIT(rx_rbuf, "RX ring buffer", s->txrx_buf_size, RX_MAX_PACKET_SIZE, uint8_t);
-  RBUF_INIT(tx_rbuf, "TX ring buffer", s->txrx_buf_size, TX_ECPRI_PACKET_SIZE, uint8_t);
+  RBUF_INIT(rx_rbuf, "RX ring buffer", s->txrx_buf_size, RX_MAX_PACKET_SIZE + s->one_way_measure, uint8_t);
+  RBUF_INIT(tx_rbuf, "TX ring buffer", s->txrx_buf_size, TX_ECPRI_PACKET_SIZE + s->one_way_measure, uint8_t);
   for(int i = 0; i < s->tx_n_channel; i++) {
     char name[256];
     sprintf(name, "TRXWrite Ring Buffer %d", i);
@@ -1182,6 +1211,8 @@ int startdpdk(TRXEcpriState * s) {
   }
   RBUF_INIT(trxw_group_rbuf, "TRXGroupWrite ring buffer", TRX_MAX_GROUP, 1, sample_group_t);
 
+  RBUF_INIT(one_way_rbuf, "One-way ring buffer", 1024, 1, int64_t);
+
   memset((uint8_t *) ecpri_message, 0, TX_ECPRI_PACKET_SIZE);
 
 #ifdef DPDK
@@ -1587,8 +1618,12 @@ int trx_driver_init(TRXState *s1)
   s->tx_n_channel = (int) val;
   trx_get_param_double(s1, &val, "master");
   s->master = (int) val;
-  trx_get_param_double(s1, &val, "generic_data_sync");
-  s->generic_data_sync = (int) val;
+  trx_get_param_double(s1, &val, "one_way_measure");
+  s->one_way_measure = (int) val;
+  trx_get_param_double(s1, &val, "one_way_period");
+  s->one_way_period = (int) val;
+  trx_get_param_double(s1, &val, "tdd_frame_start");
+  s->tdd_frame_start = (int) val;
   trx_get_param_double(s1, &val, "trx_read_null");
   s->trx_read_null = (int) val;
   trx_get_param_double(s1, &val, "tdd_period");