wip: rx_burst

trx_ecpri.c

@@ -369,7 +369,7 @@ static void update_counter(counter_stat_t * c, int64_t v) {
   c->counter += v;
 }
 
-#define RX_BURST_SIZE 4000
+#define MAX_RX_BURST 32768
 static void *recv_thread(void *p) {
 #ifdef DISABLE_RECV
   pthread_exit(EXIT_SUCCESS);
@@ -388,26 +388,26 @@ static void *recv_thread(void *p) {
 
   for(;;) {
 
-    struct mmsghdr msgh[RX_BURST_SIZE];
-    struct iovec msgv[RX_BURST_SIZE];
+    struct mmsghdr msgh[MAX_RX_BURST];
+    struct iovec msgv[MAX_RX_BURST];
 
     memset(msgv, 0, sizeof(msgv));
     memset(msgh, 0, sizeof(msgh));
 
-    if(rbuf_write_amount(&rx_rbuf) < RX_BURST_SIZE * PACKET_SIZE)
+    if(rbuf_write_amount(&rx_rbuf) < s->rx_burst * PACKET_SIZE)
       log_error("RECV_THREAD",
                 "Not enough space in RX buffer (%d < %d)\n",
-                rbuf_write_amount(&rx_rbuf), RX_BURST_SIZE * PACKET_SIZE);
-    for(int j = 0; j < RX_BURST_SIZE; j++) {
+                rbuf_write_amount(&rx_rbuf), s->rx_burst * PACKET_SIZE);
+    for(int j = 0; j < s->rx_burst; j++) {
       msgv[j].iov_base = rx_rbuf.buffer + (rx_rbuf.write_index + j * PACKET_SIZE) % rx_rbuf.buf_len;
       msgv[j].iov_len = PACKET_SIZE;
       msgh[j].msg_hdr.msg_iov = &msgv[j];
       msgh[j].msg_hdr.msg_iovlen = 1;
     }
     
-    for(int j = 0; j < RX_BURST_SIZE;) {
+    for(int j = 0; j < s->rx_burst;) {
       uint8_t * data;
-      int ret = recvmmsg(recv_sockfd, msgh + j, RX_BURST_SIZE - j, 0, NULL);
+      int ret = recvmmsg(recv_sockfd, msgh + j, s->rx_burst - j, 0, NULL);
       if(ret <= -1)
         error(EXIT_FAILURE, errno, "recvmmsg error");
       for(int l = 0; l < ret; l++) {
@@ -425,7 +425,7 @@ static void *recv_thread(void *p) {
       j += ret;
       update_counter(&recv_counter, ret);
     }
-    rbuf_increment_write(&rx_rbuf, RX_BURST_SIZE * PACKET_SIZE);
+    rbuf_increment_write(&rx_rbuf, s->rx_burst * PACKET_SIZE);
   }
   pthread_exit(EXIT_SUCCESS);
 }