From: jingleyang <yjyfrombnu@gmail.com>
Date: Mon, 24 Dec 2018 17:24:04 +0000 (+0800)
Subject: dpdk_read_with_timeout() non-blocking, blocking with timeout read
X-Git-Tag: libpcap-1.10-bp~689^2~2^2
X-Git-Url: https://git.tcpdump.org/libpcap/commitdiff_plain/ca09dfb4e8795ad6b37b3addb222957e1457d025

dpdk_read_with_timeout() non-blocking, blocking with timeout read
---

diff --git a/pcap-dpdk.c b/pcap-dpdk.c
index 285a63ab..824a5de0 100644
--- a/pcap-dpdk.c
+++ b/pcap-dpdk.c
@@ -124,7 +124,7 @@ static int is_dpdk_pre_inited=0;
 #define DPDK_DEV_NAME_MAX 32
 #define DPDK_DEV_DESC_MAX 512 
 #define DPDK_CFG_ENV_NAME "DPDK_CFG"
-#define DPDK_DEF_MIN_SLEEP_MS 100
+#define DPDK_DEF_MIN_SLEEP_MS 1
 static char dpdk_cfg_buf[DPDK_CFG_MAX_LEN];
 #define DPDK_MAC_ADDR_SIZE 32
 #define DPDK_DEF_MAC_ADDR "00:00:00:00:00:00"
@@ -215,11 +215,40 @@ static uint32_t dpdk_gather_data(unsigned char *data, int len, struct rte_mbuf *
 	return total_len;
 }
 
+
+static int dpdk_read_with_timeout(pcap_t *p, uint16_t portid, uint16_t queueid,struct rte_mbuf **pkts_burst, const uint16_t burst_cnt){
+	struct pcap_dpdk *pd = (struct pcap_dpdk*)(p->priv);
+	int nb_rx = 0;	
+	int timeout_ms = p->opt.timeout;
+	int sleep_ms = 0;
+	if (pd->nonblock){
+		// In non-blocking mode, just read once, no mater how many packets are captured.
+		nb_rx = (int)rte_eth_rx_burst(pd->portid, 0, pkts_burst, burst_cnt);
+	}else{
+		// In blocking mode, read many times until packets are captured or timeout or break_loop is setted.	
+		// if timeout_ms == 0, it may be blocked forever.
+		while (timeout_ms == 0 || sleep_ms < timeout_ms){
+			nb_rx = (int)rte_eth_rx_burst(pd->portid, 0, pkts_burst, burst_cnt);
+			if (nb_rx){ // got packets within timeout_ms
+				break;	
+			}else{ // no packet arrives at this round.
+				if (p->break_loop){
+					break;
+				}
+				// sleep for a very short while, but do not block CPU.
+				rte_delay_us_block(DPDK_DEF_MIN_SLEEP_MS*1000);
+				sleep_ms += DPDK_DEF_MIN_SLEEP_MS;
+			}
+		}
+	}
+	return nb_rx;
+}
+
 static int pcap_dpdk_dispatch(pcap_t *p, int max_cnt, pcap_handler cb, u_char *cb_arg)
 {
 	struct pcap_dpdk *pd = (struct pcap_dpdk*)(p->priv);
 	int burst_cnt = 0;
-	int nb_rx=0;
+	int nb_rx = 0;
 	struct rte_mbuf *pkts_burst[MAX_PKT_BURST];
 	struct rte_mbuf *m;
 	struct pcap_pkthdr pcap_header;
@@ -233,7 +262,8 @@ static int pcap_dpdk_dispatch(pcap_t *p, int max_cnt, pcap_handler cb, u_char *c
 	int pkt_cnt = 0;
 	int is_accepted=0;
 	u_char *large_buffer=NULL;
-	
+	int timeout_ms = p->opt.timeout;
+
 	pd->rx_pkts = 0;
 	if ( !PACKET_COUNT_IS_UNLIMITED(max_cnt) && max_cnt < MAX_PKT_BURST){
 		burst_cnt = max_cnt;
@@ -243,9 +273,27 @@ static int pcap_dpdk_dispatch(pcap_t *p, int max_cnt, pcap_handler cb, u_char *c
 
 	while( PACKET_COUNT_IS_UNLIMITED(max_cnt) || pkt_cnt < max_cnt){
 		if (p->break_loop){
-			break;
+			p->break_loop = 1;
+			return PCAP_ERROR_BREAK;
+		}
+		// read once in non-blocking mode, or try many times waiting for timeout_ms.
+		// if timeout_ms == 0, it will be blocked until one packet arrives or break_loop is setted.
+		nb_rx = dpdk_read_with_timeout(p, portid, 0, pkts_burst, burst_cnt);
+		if (nb_rx == 0){ 
+			if (pd->nonblock){
+				RTE_LOG(DEBUG, USER1, "dpdk: no packets available in non-blocking mode.\n");
+			}else{
+				if (p->break_loop){
+					RTE_LOG(DEBUG, USER1, "dpdk: no packets available and break_loop is setted in blocking mode.\n");
+					p->break_loop = 1;
+					return PCAP_ERROR_BREAK;
+
+				}
+				RTE_LOG(DEBUG, USER1, "dpdk: no packets available for timeout %d ms in blocking mode.\n", timeout_ms);
+			}
+			// break if dpdk reads 0 packet, no matter in blocking(timeout) or non-blocking mode.
+			break;	
 		}
-		nb_rx = (int)rte_eth_rx_burst(portid, 0, pkts_burst, burst_cnt);
 		pkt_cnt += nb_rx;
 		for ( i = 0; i < nb_rx; i++) {
 			m = pkts_burst[i];