wip

trx_ecpri.c

@@ -33,6 +33,8 @@
 #include "private/trx_driver.h"
 
 #define DEBUG
+//#define DISABLE_SEND
+#define DISABLE_RECV
 
 #include "utils.c"
 #include "ring_buffer.c"
@@ -295,6 +297,9 @@ static void update_counter(counter_stat_t * c, int64_t v) {
 
 #define RX_BURST_SIZE 4000
 static void *recv_thread(void *p) {
+#ifdef DISABLE_RECV
+  pthread_exit(EXIT_SUCCESS);
+#endif
 
   cpu_set_t mask;
   TRXEcpriState * s = (TRXEcpriState *) p;
@@ -338,6 +343,9 @@ static void *recv_thread(void *p) {
 #define TX_BURST_SIZE 4000
 // Send as soon as packets are encoded
 static void *send_thread(void *p) {
+#ifdef DISABLE_SEND
+  pthread_exit(EXIT_SUCCESS);
+#endif
 
   cpu_set_t mask;
   struct timespec initial;
@@ -362,26 +370,28 @@ static void *send_thread(void *p) {
     msgh[j].msg_hdr.msg_iovlen = 1;
   }
   clock_gettime(CLOCK_TAI, &initial);
+  log_info("SEND_THREAD", "Starting loop");
 
   for(int64_t i = 1;; i++) {
 
     int burst_size;
     int to_send = rbuf_read_amount(&tx_rbuf);
+    if(to_send > TX_BURST_SIZE)
+      to_send = TX_BURST_SIZE;
 
-    for(burst_size = 0 ; to_send ; burst_size++) {
-      msgv[burst_size].iov_base = rbuf_read(&tx_rbuf);
-      msgv[burst_size].iov_len = tx_rbuf.block_len;
+    for(int j = 0 ; j < to_send ; j++) {
+      msgv[j].iov_base = rbuf_read(&tx_rbuf);
+      msgv[j].iov_len = tx_rbuf.block_len;
       rbuf_increment_read(&tx_rbuf, tx_rbuf.block_len);
-      if(burst_size > TX_BURST_SIZE)
+      if(j > TX_BURST_SIZE)
         log_exit("SEND_THREAD", "Too many burst packets");
-      to_send = rbuf_read_amount(&tx_rbuf);
     }
 
-    for(int msg_sent = 0; msg_sent < burst_size;) {
-      int ret = sendmmsg(send_sockfd, msgh + msg_sent, (burst_size - msg_sent), 0);
+    for(int j = 0; j < to_send;) {
+      int ret = sendmmsg(send_sockfd, msgh + j, (burst_size - j), 0);
       if(ret <= 0)
         error(EXIT_FAILURE, errno, "sendmmsg error (returned %d)", ret);
-      msg_sent += ret;
+      j += ret;
       update_counter(&sent_counter, ret);
     }
   }
@@ -389,11 +399,16 @@ static void *send_thread(void *p) {
 }
 
 static void *encode_thread(void *p) {
+#ifdef DISABLE_SEND
+  pthread_exit(EXIT_SUCCESS);
+#endif
 
   cpu_set_t mask;
   TRXEcpriState * s = (TRXEcpriState *) p;
   uint8_t * data;
 
+  log_info("ENCODE_THREAD", "Thread init");
+
   // Set thread CPU affinity
   CPU_ZERO(&mask);
   CPU_SET(s->encode_affinity, &mask);
@@ -464,6 +479,9 @@ static void *encode_thread(void *p) {
 }
 
 static void *decode_thread(void *p) {
+#ifdef DISABLE_RECV
+  pthread_exit(EXIT_SUCCESS);
+#endif
 
   cpu_set_t mask;
   uint8_t * data;