[NETMAP,10/12] linux-netmap: pktio: refactor packet I/O using existing APIs

diff --git a/platform/linux-netmap/include/odp_packet_netmap.h b/platform/linux-netmap/include/odp_packet_netmap.h
index 89d1a27..fe376d1 100644
--- a/platform/linux-netmap/include/odp_packet_netmap.h
+++ b/platform/linux-netmap/include/odp_packet_netmap.h
@@ -38,10 +38,6 @@  typedef struct {
 	size_t frame_offset;		/* offset from start of pkt buf */
 	size_t buf_size;		/* size of buffer payload in 'pool' */
 	struct nm_desc *desc;		/* netmap meta-data for the device */
-	uint32_t begin;
-	uint32_t end;
-	struct netmap_ring *rxring;
-	struct netmap_ring *txring;
 	odp_queue_t tx_access;		/* exclusive access to send packets */
 	uint32_t if_flags;
 	netmap_dev_t *nm_dev;
diff --git a/platform/linux-netmap/odp_packet_netmap.c b/platform/linux-netmap/odp_packet_netmap.c
index 9558440..9b52bd6 100644
--- a/platform/linux-netmap/odp_packet_netmap.c
+++ b/platform/linux-netmap/odp_packet_netmap.c
@@ -64,6 +64,12 @@  static odp_spinlock_t nm_global_lock;
 static netmap_dev_t netmap_devs[MAX_DEVS];
 static int nm_dev_cnt;
 
+struct dispatch_args {
+	odp_packet_t *pkt_table;
+	unsigned nb_rx;
+	pkt_netmap_t *pkt_nm;
+};
+
 int odp_netmap_init_global(void)
 {
 	odp_spinlock_init(&nm_global_lock);
@@ -198,11 +204,6 @@  int setup_pkt_netmap(pkt_netmap_t * const pkt_nm, const char *netdev,
 		odp_thread_id(),
 		pkt_nm->desc->mem);
 
-	pkt_nm->begin  = 0;
-	pkt_nm->end    = pkt_nm->desc->req.nr_rx_rings;
-	pkt_nm->rxring = NETMAP_RXRING(pkt_nm->desc->nifp, 0);
-	pkt_nm->txring = NETMAP_TXRING(pkt_nm->desc->nifp, 0);
-
 	ret = nm_do_ioctl(pkt_nm, SIOCGIFFLAGS, 0);
 	if (ret)
 		return ret;
@@ -256,173 +257,96 @@  int close_pkt_netmap(pkt_netmap_t * const pkt_nm)
 	return 0;
 }
 
-int recv_pkt_netmap(pkt_netmap_t * const pkt_nm, odp_packet_t pkt_table[],
-		    unsigned len)
+static void nm_recv_cb(uint8_t *arg, const struct nm_pkthdr *hdr,
+		       const uint8_t *buf)
 {
-	struct netmap_ring *rxring = pkt_nm->rxring;
-	int fd;
-	unsigned nb_rx = 0;
-	uint32_t limit, rx;
-	uint32_t ringid = pkt_nm->begin;
-	odp_packet_t pkt = ODP_PACKET_INVALID;
-	struct pollfd fds[1];
-	int ret;
-
-	fd = pkt_nm->desc->fd;
-	fds[0].fd = fd;
-	fds[0].events = POLLIN;
-
-	while (nb_rx < len) {
-		ret = poll(fds, 1, POLL_TMO);
-		if (ret <= 0 || (fds[0].revents & POLLERR))
-			break;
-
-		/* Find first ring not empty */
-		while (nm_ring_empty(rxring)) {
-			ringid++;
+	struct dispatch_args *args = (struct dispatch_args *)arg;
+	odp_packet_t pkt;
+	uint8_t *pkt_buf, *l2_hdr;
+	size_t frame_len;
 
-			/* Return to scheduler if no more data to meet the
-			   requested amount (len) */
-			if (ringid == pkt_nm->end) {
-				ODP_DBG("No more data on the wire\n");
-				break;
-			}
+	frame_len = (size_t)hdr->len;
 
-			rxring = NETMAP_RXRING(pkt_nm->desc->nifp, ringid);
-		}
+	if (frame_len > args->pkt_nm->max_frame_len) {
+		ODP_ERR("[%04d] RX: frame too big %u %lu!\n",
+			odp_thread_id(),
+			(unsigned)frame_len, args->pkt_nm->max_frame_len);
+		return;
+	}
 
-		if (ringid == pkt_nm->end)
-			break;
+	if (odp_unlikely(frame_len < ODPH_ETH_LEN_MIN)) {
+		ODP_ERR("[%04d] RX: Frame truncated: %u\n",
+			odp_thread_id(),
+			(unsigned)frame_len);
+		return;
+	}
 
-		limit = len - nb_rx;
-		if (nm_ring_space(rxring) < limit)
-			limit = nm_ring_space(rxring);
-
-		for (rx = 0; rx < limit; rx++) {
-			struct netmap_slot *rslot;
-			char *p;
-			uint16_t frame_len;
-			uint8_t *pkt_buf;
-			uint8_t *l2_hdr;
-			uint32_t cur;
-
-			if (odp_likely(pkt == ODP_PACKET_INVALID)) {
-				pkt = odph_packet_alloc(pkt_nm->pool);
-				if (odp_unlikely(pkt == ODP_PACKET_INVALID))
-					break;
-			}
-
-			cur = rxring->cur;
-			rslot = &rxring->slot[cur];
-			p = NETMAP_BUF(rxring, rslot->buf_idx);
-			frame_len = rslot->len;
-
-			rxring->head = nm_ring_next(rxring, cur);
-			rxring->cur = rxring->head;
-
-			pkt_buf = odp_packet_addr(pkt);
-			l2_hdr = pkt_buf + pkt_nm->frame_offset;
-
-			if (frame_len > pkt_nm->max_frame_len) {
-				ODP_ERR("RX: frame too big %u %lu!\n",
-					frame_len, pkt_nm->max_frame_len);
-				/* drop the frame, reuse pkt next interation */
-				continue;
-			}
-			if (odp_unlikely(frame_len < ODPH_ETH_LEN_MIN)) {
-				ODP_ERR("RX: Frame truncated: %u\n",
-					(unsigned)frame_len);
-				continue;
-			}
-
-			/* For now copy the data in the mbuf,
-			   worry about zero-copy later */
-			memcpy(l2_hdr, p, frame_len);
-
-			/* Initialize, parse and set packet header data */
-			odp_packet_init(pkt);
-			odp_packet_parse(pkt, frame_len, pkt_nm->frame_offset);
-
-			pkt_table[nb_rx] = pkt;
-			pkt = ODP_PACKET_INVALID;
-			nb_rx++;
-		}
+	pkt = odph_packet_alloc(args->pkt_nm->pool);
 
-		if (odp_unlikely(pkt == ODP_PACKET_INVALID))
-			break;
+	if (odp_unlikely(pkt == ODP_PACKET_INVALID)) {
+		ODP_DBG("[%04d] Could not allocate packet\n", odp_thread_id());
+		return;
 	}
+	pkt_buf = odp_packet_addr(pkt);
+	l2_hdr = pkt_buf + args->pkt_nm->frame_offset;
+	/* For now copy the data in the mbuf,
+	   worry about zero-copy later */
+	memcpy(l2_hdr, buf, frame_len);
 
-	if (odp_unlikely(pkt != ODP_PACKET_INVALID))
-		odp_buffer_free((odp_buffer_t) pkt);
+	/* Initialize, parse and set packet header data */
+	odp_packet_init(pkt);
+	odp_packet_parse(pkt, frame_len, args->pkt_nm->frame_offset);
 
-	return nb_rx;
+	args->pkt_table[args->nb_rx++] = pkt;
 }
 
-int send_pkt_netmap(pkt_netmap_t * const pkt_nm, odp_packet_t pkt_table[],
+int recv_pkt_netmap(pkt_netmap_t * const pkt_nm, odp_packet_t pkt_table[],
 		    unsigned len)
 {
-	struct netmap_ring *txring = pkt_nm->txring;
-	int                 fd;
-	unsigned            nb_tx = 0;
-	uint32_t            limit, tx;
-	uint32_t            ringid = pkt_nm->begin;
-	odp_packet_t        pkt;
+	int fd = pkt_nm->desc->fd;
 	struct pollfd fds[1];
 	int ret;
+	struct dispatch_args args;
 
-	fd = pkt_nm->desc->fd;
 	fds[0].fd = fd;
-	fds[0].events = POLLOUT;
+	fds[0].events = POLLIN;
 
-	while (nb_tx < len) {
-		ret = poll(fds, 1, POLL_TMO);
-		if (ret <= 0 || (fds[0].revents & POLLERR))
-			break;
+	ret = poll(fds, 1, POLL_TMO);
+	if (ret <= 0 || (fds[0].revents & POLLERR))
+		return 0;
 
-		/* Find first ring not empty */
-		while (nm_ring_empty(txring)) {
-			ringid++;
+	args.pkt_table = pkt_table;
+	args.nb_rx = 0;
+	args.pkt_nm = pkt_nm;
+	nm_dispatch(pkt_nm->desc, len, nm_recv_cb, (uint8_t *)&args);
 
-			/* Return to scheduler if no more space to meet the
-			   requested amount (len) */
-			if (ringid == pkt_nm->end) {
-				ODP_DBG("No more space in TX rings\n");
-				break;
-			}
+	return args.nb_rx;
+}
 
-			txring = NETMAP_TXRING(pkt_nm->desc->nifp, ringid);
-		}
+int send_pkt_netmap(pkt_netmap_t * const pkt_nm, odp_packet_t pkt_table[],
+		    unsigned len)
+{
+	int fd = pkt_nm->desc->fd;
+	unsigned tx, nb_tx;
+	struct pollfd fds[2];
+	int ret;
 
-		if (ringid == pkt_nm->end)
-			break;
+	fds[0].fd = fd;
+	fds[0].events = POLLOUT;
 
-		limit = len - nb_tx;
-		if (nm_ring_space(txring) < limit)
-			limit = nm_ring_space(txring);
-
-		for (tx = 0; tx < limit; tx++) {
-			struct netmap_slot *tslot;
-			size_t frame_len;
-			uint32_t cur;
-			uint8_t *frame;
-			void *txbuf;
-
-			cur = txring->cur;
-			tslot = &txring->slot[cur];
-			txbuf = NETMAP_BUF(txring, tslot->buf_idx);
-
-			pkt = pkt_table[nb_tx];
-			frame = odp_packet_l2(pkt);
-			frame_len = odp_packet_get_len(pkt);
-
-			memcpy(txbuf, frame, frame_len);
-			tslot->len = frame_len;
-			txring->head = nm_ring_next(txring, cur);
-			txring->cur = txring->head;
-			nb_tx++;
-		}
+	ret = poll(fds, 1, POLL_TMO);
+	if (ret <= 0 || (fds[0].revents & POLLERR))
+		goto out;
+
+	for (nb_tx = 0; nb_tx < len; nb_tx++) {
+		odp_packet_t pkt = pkt_table[nb_tx];
+		uint8_t *frame = odp_packet_l2(pkt);
+		size_t frame_len = odp_packet_get_len(pkt);
+		if (nm_inject(pkt_nm->desc, frame, frame_len) == 0)
+			break;
 	}
 
+out:
 	for (tx = 0; tx < len; tx++)
 		odph_packet_free(pkt_table[tx]);