linux-gen: dpdk: improved zero-copy implementation

Improved zero-copy DPDK pktio implementation which better adheres to DPDK
APIs. The new implementation reduces overhead by moving mbuf
initialization to ODP pool create and by using offsets instead of saved
pointers to do ODP packet / DPDK mbuf conversion.

Signed-off-by: Matias Elo <matias.elo@nokia.com>
Reviewed-by: Bill Fischofer <bill.fischofer@linaro.org>
Signed-off-by: Maxim Uvarov <maxim.uvarov@linaro.org>

5 files changed, 208 insertions, 117 deletions

diff --git a/platform/linux-generic/include/odp_packet_dpdk.h b/platform/linux-generic/include/odp_packet_dpdk.h
index 0dac296ac..1b660babf 100644
--- a/platform/linux-generic/include/odp_packet_dpdk.h
+++ b/platform/linux-generic/include/odp_packet_dpdk.h
@@ -7,11 +7,25 @@
 #ifndef ODP_PACKET_DPDK_H
 #define ODP_PACKET_DPDK_H
 
+#include <stdint.h>
+
 #include <odp/api/packet_io.h>
 
+#include <odp_packet_internal.h>
+#include <odp_pool_internal.h>
+
 struct rte_mbuf;
 
-/** Cache for storing packets */
+/**
+ * Calculate size of zero-copy DPDK packet pool object
+ */
+uint32_t _odp_dpdk_pool_obj_size(pool_t *pool, uint32_t block_size);
+
+/**
+ * Create zero-copy DPDK packet pool
+ */
+int _odp_dpdk_pool_create(pool_t *pool);
+
 /** Packet parser using DPDK interface */
 int _odp_dpdk_packet_parse_common(packet_parser_t *pkt_hdr,
 				  const uint8_t *ptr,
diff --git a/platform/linux-generic/include/odp_packet_internal.h b/platform/linux-generic/include/odp_packet_internal.h
index 0dccc97da..d17a61a06 100644
--- a/platform/linux-generic/include/odp_packet_internal.h
+++ b/platform/linux-generic/include/odp_packet_internal.h
@@ -95,9 +95,6 @@ typedef struct {
 	 * Members below are not initialized by packet_init()
 	 */
 
-	/* Type of extra data */
-	uint8_t extra_type;
-
 	/* Flow hash value */
 	uint32_t flow_hash;
 
@@ -113,11 +110,6 @@ typedef struct {
 	/* Context for IPsec */
 	odp_ipsec_packet_result_t ipsec_ctx;
 
-#ifdef ODP_PKTIO_DPDK
-	/* Extra space for packet descriptors. E.g. DPDK mbuf. Keep as the last
-	 * member before data. */
-	uint8_t ODP_ALIGNED_CACHE extra[PKT_EXTRA_LEN];
-#endif
 	/* Packet data storage */
 	uint8_t data[0];
 } odp_packet_hdr_t;
diff --git a/platform/linux-generic/include/odp_pool_internal.h b/platform/linux-generic/include/odp_pool_internal.h
index e8a046142..df02fdafa 100644
--- a/platform/linux-generic/include/odp_pool_internal.h
+++ b/platform/linux-generic/include/odp_pool_internal.h
@@ -67,11 +67,15 @@ typedef struct pool_t {
 	uint32_t         max_len;
 	uint32_t         uarea_size;
 	uint32_t         block_size;
+	uint32_t         block_offset;
 	uint8_t         *base_addr;
 	uint8_t         *uarea_base_addr;
 
 	/* Used by DPDK zero-copy pktio */
-	uint8_t		mem_from_huge_pages;
+	uint32_t         dpdk_elt_size;
+	uint32_t         skipped_blocks;
+	uint8_t          pool_in_use;
+	uint8_t          mem_from_huge_pages;
 	pool_destroy_cb_fn ext_destroy;
 	void            *ext_desc;
 
@@ -110,7 +114,8 @@ static inline odp_buffer_hdr_t *buf_hdr_from_index(pool_t *pool,
 	uint64_t block_offset;
 	odp_buffer_hdr_t *buf_hdr;
 
-	block_offset = buffer_idx * (uint64_t)pool->block_size;
+	block_offset = (buffer_idx * (uint64_t)pool->block_size) +
+			pool->block_offset;
 
 	/* clang requires cast to uintptr_t */
 	buf_hdr = (odp_buffer_hdr_t *)(uintptr_t)&pool->base_addr[block_offset];
diff --git a/platform/linux-generic/odp_pool.c b/platform/linux-generic/odp_pool.c
index 8ae0e4e30..5bb5bc630 100644
--- a/platform/linux-generic/odp_pool.c
+++ b/platform/linux-generic/odp_pool.c
@@ -16,6 +16,7 @@
 #include <odp_pool_internal.h>
 #include <odp_init_internal.h>
 #include <odp_packet_internal.h>
+#include <odp_packet_dpdk.h>
 #include <odp_config_internal.h>
 #include <odp_debug_internal.h>
 #include <odp_ring_internal.h>
@@ -248,7 +249,10 @@ static void init_buffers(pool_t *pool)
 	type = pool->params.type;
 
 	for (i = 0; i < pool->num + skipped_blocks ; i++) {
-		addr    = &pool->base_addr[i * pool->block_size];
+		int skip = 0;
+
+		addr    = &pool->base_addr[(i * pool->block_size) +
+					   pool->block_offset];
 		buf_hdr = addr;
 		pkt_hdr = addr;
 		/* Skip packet buffers which cross huge page boundaries. Some
@@ -265,7 +269,7 @@ static void init_buffers(pool_t *pool)
 					~(page_size - 1));
 			if (last_page != first_page) {
 				skipped_blocks++;
-				continue;
+				skip = 1;
 			}
 		}
 		if (pool->uarea_size)
@@ -310,8 +314,10 @@ static void init_buffers(pool_t *pool)
 				     pool->tailroom];
 
 		/* Store buffer index into the global pool */
-		ring_enq(ring, mask, i);
+		if (!skip)
+			ring_enq(ring, mask, i);
 	}
+	pool->skipped_blocks = skipped_blocks;
 }
 
 static bool shm_is_from_huge_pages(odp_shm_t shm)
@@ -445,6 +451,17 @@ static odp_pool_t pool_create(const char *name, odp_pool_param_t *params,
 	block_size = ROUNDUP_CACHE_LINE(hdr_size + align + headroom + seg_len +
 					tailroom);
 
+	/* Calculate extra space required for storing DPDK objects and mbuf
+	 * headers. NOP if zero-copy is disabled. */
+	pool->block_offset = 0;
+	if (params->type == ODP_POOL_PACKET) {
+		block_size = _odp_dpdk_pool_obj_size(pool, block_size);
+		if (!block_size) {
+			ODP_ERR("Calculating DPDK mempool obj size failed\n");
+			return ODP_POOL_INVALID;
+		}
+	}
+
 	/* Allocate extra memory for skipping packet buffers which cross huge
 	 * page boundaries. */
 	if (params->type == ODP_POOL_PACKET) {
@@ -506,6 +523,12 @@ static odp_pool_t pool_create(const char *name, odp_pool_param_t *params,
 	ring_init(&pool->ring->hdr);
 	init_buffers(pool);
 
+	/* Create zero-copy DPDK memory pool. NOP if zero-copy is disabled. */
+	if (params->type == ODP_POOL_PACKET && _odp_dpdk_pool_create(pool)) {
+		ODP_ERR("Creating DPDK packet pool failed\n");
+		goto error;
+	}
+
 	return pool->pool_hdl;
 
 error:
diff --git a/platform/linux-generic/pktio/dpdk.c b/platform/linux-generic/pktio/dpdk.c
index 34c234403..23f39445e 100644
--- a/platform/linux-generic/pktio/dpdk.c
+++ b/platform/linux-generic/pktio/dpdk.c
@@ -68,8 +68,6 @@
 #if ODP_DPDK_ZERO_COPY
 ODP_STATIC_ASSERT(CONFIG_PACKET_HEADROOM == RTE_PKTMBUF_HEADROOM,
 		  "ODP and DPDK headroom sizes not matching!");
-ODP_STATIC_ASSERT(PKT_EXTRA_LEN >= sizeof(struct rte_mbuf),
-		  "DPDK rte_mbuf won't fit in odp_packet_hdr_t.extra!");
 #endif
 
 /* DPDK poll mode drivers requiring minimum RX burst size DPDK_MIN_RX_BURST */
@@ -81,6 +79,8 @@ ODP_STATIC_ASSERT(PKT_EXTRA_LEN >= sizeof(struct rte_mbuf),
 #define DPDK_MBUF_BUF_SIZE RTE_MBUF_DEFAULT_BUF_SIZE
 #define DPDK_MEMPOOL_CACHE_SIZE 64
 
+#define MBUF_OFFSET (ROUNDUP_CACHE_LINE(sizeof(struct rte_mbuf)))
+
 ODP_STATIC_ASSERT((DPDK_NB_MBUF % DPDK_MEMPOOL_CACHE_SIZE == 0) &&
 		  (DPDK_MEMPOOL_CACHE_SIZE <= RTE_MEMPOOL_CACHE_MAX_SIZE) &&
 		  (DPDK_MEMPOOL_CACHE_SIZE <= DPDK_MBUF_BUF_SIZE * 10 / 15)
@@ -165,6 +165,8 @@ void refer_constructors(void)
 }
 #endif
 
+static int dpdk_pktio_init(void);
+
 static int pool_alloc(struct rte_mempool *mp);
 
 static int lookup_opt(const char *opt_name, const char *drv_name, int *val)
@@ -272,34 +274,39 @@ static inline void mbuf_update(struct rte_mbuf *mbuf, odp_packet_hdr_t *pkt_hdr,
 }
 
 /**
- * Initialize mbuf
- *
- * Called once per ODP packet.
+ * Initialize packet mbuf. Modified version of standard rte_pktmbuf_init()
+ * function.
  */
-static void mbuf_init(struct rte_mempool *mp, struct rte_mbuf *mbuf,
-		      odp_packet_hdr_t *pkt_hdr)
+static void pktmbuf_init(struct rte_mempool *mp, void *opaque_arg ODP_UNUSED,
+			 void *_m, unsigned i ODP_UNUSED)
 {
-	void *buf_addr = pkt_hdr->buf_hdr.base_data - RTE_PKTMBUF_HEADROOM;
-
-	memset(mbuf, 0, sizeof(struct rte_mbuf));
-
-	mbuf->priv_size = 0;
-	mbuf->buf_addr = buf_addr;
-	mbuf->buf_physaddr = rte_mem_virt2phy(buf_addr);
-	if (odp_unlikely(mbuf->buf_physaddr == RTE_BAD_PHYS_ADDR ||
-			 mbuf->buf_physaddr == 0))
-		ODP_ABORT("Failed to map virt addr to phy");
-
-	mbuf->buf_len = (uint16_t)rte_pktmbuf_data_room_size(mp);
-	mbuf->data_off = RTE_PKTMBUF_HEADROOM;
-	mbuf->pool = mp;
-	mbuf->refcnt = 1;
-	mbuf->nb_segs = 1;
-	mbuf->port = 0xff;
-
-	/* Store ODP packet handle inside rte_mbuf */
-	mbuf->userdata = packet_handle(pkt_hdr);
-	pkt_hdr->extra_type = PKT_EXTRA_TYPE_DPDK;
+	struct rte_mbuf *m = _m;
+	uint32_t mbuf_size, buf_len;
+	odp_packet_hdr_t *pkt_hdr;
+	void *buf_addr;
+
+	pkt_hdr = (odp_packet_hdr_t *)(uintptr_t)((uint8_t *)m + MBUF_OFFSET);
+	buf_addr = pkt_hdr->buf_hdr.base_data - RTE_PKTMBUF_HEADROOM;
+
+	mbuf_size = sizeof(struct rte_mbuf);
+	buf_len = rte_pktmbuf_data_room_size(mp);
+
+	memset(m, 0, mbuf_size);
+	m->priv_size = 0;
+	m->buf_addr = buf_addr;
+	m->buf_iova = rte_mem_virt2iova(buf_addr);
+	m->buf_len = (uint16_t)buf_len;
+	m->data_off = RTE_PKTMBUF_HEADROOM;
+
+	if (odp_unlikely(m->buf_iova == RTE_BAD_IOVA || m->buf_iova == 0))
+		ODP_ABORT("Failed to map virt addr to iova\n");
+
+	/* Init some constant fields */
+	m->pool = mp;
+	m->nb_segs = 1;
+	m->port = MBUF_INVALID_PORT;
+	rte_mbuf_refcnt_set(m, 1);
+	m->next = NULL;
 }
 
 /**
@@ -308,60 +315,87 @@ static void mbuf_init(struct rte_mempool *mp, struct rte_mbuf *mbuf,
 static struct rte_mempool *mbuf_pool_create(const char *name,
 					    pool_t *pool_entry)
 {
-	struct rte_mempool *mp;
+	struct rte_mempool *mp = NULL;
 	struct rte_pktmbuf_pool_private mbp_priv;
-	unsigned elt_size;
-	unsigned num;
-	uint16_t data_room_size;
+	struct rte_mempool_objsz sz;
+	unsigned int elt_size = pool_entry->dpdk_elt_size;
+	unsigned int num = pool_entry->num;
+	uint32_t total_size;
 
 	if (!(pool_entry->mem_from_huge_pages)) {
 		ODP_ERR("DPDK requires memory is allocated from huge pages\n");
-		return NULL;
+		goto fail;
 	}
 
-	num = pool_entry->num;
-	data_room_size = pool_entry->seg_len + CONFIG_PACKET_HEADROOM;
-	elt_size = sizeof(struct rte_mbuf) + (unsigned)data_room_size;
-	mbp_priv.mbuf_data_room_size = data_room_size;
-	mbp_priv.mbuf_priv_size = 0;
+	total_size = rte_mempool_calc_obj_size(elt_size, MEMPOOL_F_NO_SPREAD,
+					       &sz);
+	if (total_size != pool_entry->block_size) {
+		ODP_ERR("DPDK pool block size not matching to ODP pool: "
+			"%" PRIu32 "/%" PRIu32 "\n", total_size,
+			pool_entry->block_size);
+		goto fail;
+	}
 
-	mp = rte_mempool_create_empty(name, num, elt_size, cache_size(num),
+	/* Skipped buffers have to be taken into account to populate pool
+	 * properly. */
+	mp = rte_mempool_create_empty(name, num + pool_entry->skipped_blocks,
+				      elt_size, cache_size(num),
 				      sizeof(struct rte_pktmbuf_pool_private),
-				      rte_socket_id(), 0);
+				      rte_socket_id(), MEMPOOL_F_NO_SPREAD);
 	if (mp == NULL) {
 		ODP_ERR("Failed to create empty DPDK packet pool\n");
-		return NULL;
+		goto fail;
 	}
 
 	if (rte_mempool_set_ops_byname(mp, "odp_pool", pool_entry)) {
 		ODP_ERR("Failed setting mempool operations\n");
-		return NULL;
+		goto fail;
 	}
 
+	mbp_priv.mbuf_data_room_size = pool_entry->seg_len;
+	mbp_priv.mbuf_priv_size = 0;
 	rte_pktmbuf_pool_init(mp, &mbp_priv);
 
 	if (pool_alloc(mp)) {
 		ODP_ERR("Failed allocating mempool\n");
-		return NULL;
+		goto fail;
+	}
+
+	num = rte_mempool_populate_iova(mp, (char *)pool_entry->base_addr,
+					RTE_BAD_IOVA, pool_entry->shm_size,
+					NULL, NULL);
+	if (num <= 0) {
+		ODP_ERR("Failed to populate mempool: %d\n", num);
+		goto fail;
 	}
 
+	rte_mempool_obj_iter(mp, pktmbuf_init, NULL);
+
 	return mp;
+
+fail:
+	if (mp)
+		rte_mempool_free(mp);
+	return NULL;
 }
 
 /* DPDK external memory pool operations */
 
-static int pool_enqueue(struct rte_mempool *mp ODP_UNUSED,
+static int pool_enqueue(struct rte_mempool *mp,
 			void * const *obj_table, unsigned num)
 {
 	odp_packet_t pkt_tbl[num];
+	pool_t *pool_entry = (pool_t *)mp->pool_config;
 	unsigned i;
 
-	if (odp_unlikely(num == 0))
+	if (odp_unlikely(num == 0 || !pool_entry->pool_in_use))
 		return 0;
 
-	for (i = 0; i < num; i++)
-		pkt_tbl[i] = (odp_packet_t)((struct rte_mbuf *)
-				obj_table[i])->userdata;
+	for (i = 0; i < num; i++) {
+		odp_packet_hdr_t *pkt_hdr = (odp_packet_hdr_t *)(uintptr_t)
+				((uint8_t *)obj_table[i] + MBUF_OFFSET);
+		pkt_tbl[i] = packet_handle(pkt_hdr);
+	}
 
 	odp_packet_free_multi(pkt_tbl, num);
 
@@ -387,13 +421,10 @@ static int pool_dequeue_bulk(struct rte_mempool *mp, void **obj_table,
 	}
 
 	for (i = 0; i < pkts; i++) {
-		odp_packet_t pkt = packet_tbl[i];
-		odp_packet_hdr_t *pkt_hdr = packet_hdr(pkt);
-		struct rte_mbuf *mbuf = (struct rte_mbuf *)
-					(uintptr_t)pkt_hdr->extra;
-		if (pkt_hdr->extra_type != PKT_EXTRA_TYPE_DPDK)
-			mbuf_init(mp, mbuf, pkt_hdr);
-		obj_table[i] = mbuf;
+		odp_packet_hdr_t *pkt_hdr = packet_hdr(packet_tbl[i]);
+
+		obj_table[i] = (struct rte_mbuf *)(uintptr_t)
+				((uint8_t *)pkt_hdr - MBUF_OFFSET);
 	}
 
 	return 0;
@@ -438,21 +469,23 @@ static void pool_destroy(void *pool)
 {
 	struct rte_mempool *mp = (struct rte_mempool *)pool;
 
-	if (mp != NULL)
+	if (mp != NULL) {
+		pool_t *pool_entry = (pool_t *)mp->pool_config;
+
+		pool_entry->pool_in_use = 0;
 		rte_mempool_free(mp);
+	}
 }
 
-static struct rte_mempool *pool_create(pool_t *pool)
+int _odp_dpdk_pool_create(pool_t *pool)
 {
 	struct rte_mempool *pkt_pool;
 	char pool_name[RTE_MEMPOOL_NAMESIZE];
 
-	odp_ticketlock_lock(&pool->lock);
+	if (!ODP_DPDK_ZERO_COPY)
+		return 0;
 
-	if (pool->ext_desc != NULL) {
-		odp_ticketlock_unlock(&pool->lock);
-		return (struct rte_mempool *)pool->ext_desc;
-	}
+	pool->pool_in_use = 0;
 
 	snprintf(pool_name, sizeof(pool_name),
 		 "dpdk_pktpool_%" PRIu32 "_%" PRIu32 "", odp_global_ro.main_pid,
@@ -460,17 +493,41 @@ static struct rte_mempool *pool_create(pool_t *pool)
 	pkt_pool = mbuf_pool_create(pool_name, pool);
 
 	if (pkt_pool == NULL) {
-		odp_ticketlock_unlock(&pool->lock);
 		ODP_ERR("Creating external DPDK pool failed\n");
-		return NULL;
+		return -1;
 	}
 
 	pool->ext_desc = pkt_pool;
 	pool->ext_destroy = pool_destroy;
+	pool->pool_in_use = 1;
+
+	return 0;
+}
+
+uint32_t _odp_dpdk_pool_obj_size(pool_t *pool, uint32_t block_size)
+{
+	struct rte_mempool_objsz sz;
+	uint32_t total_size;
+
+	if (!ODP_DPDK_ZERO_COPY)
+		return block_size;
 
-	odp_ticketlock_unlock(&pool->lock);
+	if (odp_global_rw->dpdk_initialized == 0) {
+		if (dpdk_pktio_init()) {
+			ODP_ERR("Initializing DPDK failed\n");
+			return 0;
+		}
+		odp_global_rw->dpdk_initialized = 1;
+	}
+
+	block_size += MBUF_OFFSET;
+	total_size = rte_mempool_calc_obj_size(block_size, MEMPOOL_F_NO_SPREAD,
+					       &sz);
 
-	return pkt_pool;
+	pool->dpdk_elt_size = sz.elt_size;
+	pool->block_offset = sz.header_size + MBUF_OFFSET;
+
+	return total_size;
 }
 
 static struct rte_mempool_ops ops_stack = {
@@ -748,7 +805,8 @@ fail:
 
 static inline void prefetch_pkt(struct rte_mbuf *mbuf)
 {
-	odp_packet_hdr_t *pkt_hdr = mbuf->userdata;
+	odp_packet_hdr_t *pkt_hdr = (odp_packet_hdr_t *)(uintptr_t)
+			((uint8_t *)mbuf + MBUF_OFFSET);
 	void *data = rte_pktmbuf_mtod(mbuf, char *);
 
 	odp_prefetch(pkt_hdr);
@@ -761,7 +819,6 @@ static inline int mbuf_to_pkt_zero(pktio_entry_t *pktio_entry,
 				   struct rte_mbuf *mbuf_table[],
 				   uint16_t mbuf_num, odp_time_t *ts)
 {
-	odp_packet_t pkt;
 	odp_packet_hdr_t *pkt_hdr;
 	uint16_t pkt_len;
 	struct rte_mbuf *mbuf;
@@ -799,8 +856,8 @@ static inline int mbuf_to_pkt_zero(pktio_entry_t *pktio_entry,
 		data = rte_pktmbuf_mtod(mbuf, char *);
 		pkt_len = rte_pktmbuf_pkt_len(mbuf);
 
-		pkt = (odp_packet_t)mbuf->userdata;
-		pkt_hdr = packet_hdr(pkt);
+		pkt_hdr = (odp_packet_hdr_t *)(uintptr_t)((uint8_t *)mbuf +
+				MBUF_OFFSET);
 
 		if (pktio_cls_enabled(pktio_entry)) {
 			packet_parse_reset(&parsed_hdr);
@@ -845,7 +902,7 @@ static inline int mbuf_to_pkt_zero(pktio_entry_t *pktio_entry,
 
 		packet_set_ts(pkt_hdr, ts);
 
-		pkt_table[nb_pkts++] = pkt;
+		pkt_table[nb_pkts++] = packet_handle(pkt_hdr);
 	}
 
 	return nb_pkts;
@@ -866,45 +923,25 @@ static inline int pkt_to_mbuf_zero(pktio_entry_t *pktio_entry,
 	for (i = 0; i < num; i++) {
 		odp_packet_t pkt = pkt_table[i];
 		odp_packet_hdr_t *pkt_hdr = packet_hdr(pkt);
-		struct rte_mbuf *mbuf = (struct rte_mbuf *)
-					(uintptr_t)pkt_hdr->extra;
+		struct rte_mbuf *mbuf = (struct rte_mbuf *)(uintptr_t)
+				((uint8_t *)pkt_hdr - MBUF_OFFSET);
 		uint16_t pkt_len = odp_packet_len(pkt);
 
 		if (odp_unlikely(pkt_len > pkt_dpdk->mtu))
 			goto fail;
 
-		if (odp_likely(pkt_hdr->buf_hdr.segcount == 1 &&
-			       pkt_hdr->extra_type == PKT_EXTRA_TYPE_DPDK)) {
+		if (odp_likely(pkt_hdr->buf_hdr.segcount == 1)) {
 			mbuf_update(mbuf, pkt_hdr, pkt_len);
 
 			if (odp_unlikely(pktio_entry->s.chksum_insert_ena))
 				pkt_set_ol_tx(pktout_cfg, pktout_capa, pkt_hdr,
 					      mbuf, odp_packet_data(pkt));
 		} else {
-			pool_t *pool_entry = pkt_hdr->buf_hdr.pool_ptr;
-
-			if (odp_unlikely(pool_entry->ext_desc == NULL)) {
-				if (pool_create(pool_entry) == NULL)
-					ODP_ABORT("Creating DPDK pool failed");
-			}
-
-			if (pkt_hdr->buf_hdr.segcount != 1 ||
-			    !pool_entry->mem_from_huge_pages) {
-				/* Fall back to packet copy */
-				if (odp_unlikely(pkt_to_mbuf(pktio_entry, &mbuf,
-							     &pkt, 1) != 1))
-					goto fail;
-				(*copy_count)++;
-
-			} else {
-				mbuf_init((struct rte_mempool *)
-					  pool_entry->ext_desc, mbuf, pkt_hdr);
-				mbuf_update(mbuf, pkt_hdr, pkt_len);
-				if (pktio_entry->s.chksum_insert_ena)
-					pkt_set_ol_tx(pktout_cfg, pktout_capa,
-						      pkt_hdr, mbuf,
-						      odp_packet_data(pkt));
-			}
+			/* Fall back to packet copy */
+			if (odp_unlikely(pkt_to_mbuf(pktio_entry, &mbuf,
+						     &pkt, 1) != 1))
+				goto fail;
+			(*copy_count)++;
 		}
 		mbuf_table[i] = mbuf;
 	}
@@ -1486,10 +1523,7 @@ static int dpdk_open(odp_pktio_t id ODP_UNUSED,
 		pkt_dpdk->min_rx_burst = 0;
 
 	if (ODP_DPDK_ZERO_COPY) {
-		if (pool_entry->ext_desc != NULL)
-			pkt_pool = (struct rte_mempool *)pool_entry->ext_desc;
-		else
-			pkt_pool = pool_create(pool_entry);
+		pkt_pool = (struct rte_mempool *)pool_entry->ext_desc;
 	} else {
 		snprintf(pool_name, sizeof(pool_name), "pktpool_%s", netdev);
 		/* Check if the pool exists already */
@@ -1879,4 +1913,27 @@ const pktio_if_ops_t dpdk_pktio_ops = {
 	.output_queues_config = dpdk_output_queues_config
 };
 
+#else
+
+#include <stdint.h>
+
+#include <odp/api/hints.h>
+
+#include <odp_packet_dpdk.h>
+#include <odp_pool_internal.h>
+
+/*
+ * Dummy functions for pool_create()
+ */
+
+uint32_t _odp_dpdk_pool_obj_size(pool_t *pool ODP_UNUSED, uint32_t block_size)
+{
+	return block_size;
+}
+
+int _odp_dpdk_pool_create(pool_t *pool ODP_UNUSED)
+{
+	return 0;
+}
+
 #endif /* ODP_PKTIO_DPDK */