Add START_RECEIVING

trx_ecpri_dpdk.c

@@ -50,16 +50,18 @@
 //#define DEBUG // Enables / deactivates log_debug
 //#define DST_ADDR_SYNTAX // Depends on DPDK version
 //#define TRACE
-#define MONITOR
+//#define MONITOR
 
 #define RECV_STOP_THRESHOLD 3
 #define PPS_UPDATE_PERIOD INT64_C(1000000000)
 
 #include "utils.c"
 
-#define EFREQ 38400
+//#define EFREQ 38400
+#define EFREQ 100
 #define STAT_FRAME_INTERVAL INT64_C(EFREQ * 150)
-#define START_SENDING
+//#define START_SENDING
+#define START_RECEIVING
 
 #define RX_N_CHANNEL 1
 #define TX_N_CHANNEL 4
@@ -1015,9 +1017,10 @@ static void trx_ecpri_end(TRXState *s1)
   free(s);
 }
 
-int64_t prev_ts = 0;
-int64_t prev_count = 0;
+static int64_t prev_ts = 0;
+static int64_t prev_count = 0;
 #define M 32
+static volatile int64_t trxr_counter = 0;
 
 static void trx_ecpri_write(TRXState *s1, trx_timestamp_t timestamp, const void **__samples, int count, int tx_port_index, TRXWriteMetadata *md)
 {
@@ -1072,6 +1075,10 @@ static void trx_ecpri_write(TRXState *s1, trx_timestamp_t timestamp, const void
   update_counter(&write_counter, count / M);
 }
 
+static int64_t trxr_target_counter = 0;
+static int trxr_i = 0;
+static struct timespec trxr_next;
+
 static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__samples, int count, int rx_port_index, TRXReadMetadata *md)
 {
   (void) s1;
@@ -1080,6 +1087,24 @@ static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__sa
   int read_count = (count / M);
   int offset = 0;
 
+#ifdef START_RECEIVING
+  if(!sync_complete && rbuf_read_amount(&trxr_rbuf[0]) < read_count) {
+    if(trxr_i == 0)
+      clock_gettime(CLOCK_TAI, &trxr_next);
+    // Limit packets sent
+    if(trxr_counter > trxr_target_counter) {
+      int k = (trxr_counter - trxr_target_counter + EFREQ - 1) / EFREQ;
+      add_ns(&trxr_next, k * 1000 * 1000 * 10); // 10ms to send 38400 packets
+      clock_nanosleep(CLOCK_TAI, TIMER_ABSTIME, &trxr_next, NULL);
+      trxr_target_counter += k * EFREQ;
+    }
+    *ptimestamp = trxr_counter * M;
+    trxr_counter += read_count;
+    update_counter(&read_counter, read_count);
+    return count;
+  }
+#endif
+
   while(rbuf_read_amount(&trxr_rbuf[0]) < read_count);
 
   log_debug("TRX_ECPRI_READ", "count = %ld (%li)", read_count, read_counter.counter);
@@ -1098,7 +1123,7 @@ static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__sa
     offset += len;
   }
 
-  *ptimestamp = recv_counter.counter * M;
+  *ptimestamp = (recv_counter.counter) * M;
   update_counter(&read_counter, read_count);
 
   return count;