[PATCHv10,7/8] linux-generic: add ipc pktio support

Signed-off-by: Maxim Uvarov <maxim.uvarov@linaro.org>
---
 platform/linux-generic/Makefile.am                 |   2 +
 .../linux-generic/include/odp_buffer_internal.h    |   3 +
 .../linux-generic/include/odp_packet_io_internal.h |  47 ++
 .../include/odp_packet_io_ipc_internal.h           |  51 ++
 platform/linux-generic/include/odp_shm_internal.h  |  21 +
 platform/linux-generic/odp_packet_io.c             |   1 +
 platform/linux-generic/odp_pool.c                  |  11 +-
 platform/linux-generic/odp_shared_memory.c         |  12 +-
 platform/linux-generic/pktio/io_ops.c              |   1 +
 platform/linux-generic/pktio/ipc.c                 | 703 +++++++++++++++++++++
 platform/linux-generic/pktio/ring.c                |   1 +
 11 files changed, 849 insertions(+), 4 deletions(-)
 create mode 100644 platform/linux-generic/include/odp_packet_io_ipc_internal.h
 create mode 100644 platform/linux-generic/include/odp_shm_internal.h
 create mode 100644 platform/linux-generic/pktio/ipc.c
 create mode 120000 platform/linux-generic/pktio/ring.c

\ No newline at end of file

On Mon, Oct 19, 2015 at 07:08:02PM +0300, Maxim Uvarov wrote:
> Signed-off-by: Maxim Uvarov <maxim.uvarov@linaro.org>
> ---
>  platform/linux-generic/Makefile.am                 |   2 +
>  .../linux-generic/include/odp_buffer_internal.h    |   3 +
>  .../linux-generic/include/odp_packet_io_internal.h |  47 ++
>  .../include/odp_packet_io_ipc_internal.h           |  51 ++
>  platform/linux-generic/include/odp_shm_internal.h  |  21 +
>  platform/linux-generic/odp_packet_io.c             |   1 +
>  platform/linux-generic/odp_pool.c                  |  11 +-
>  platform/linux-generic/odp_shared_memory.c         |  12 +-
>  platform/linux-generic/pktio/io_ops.c              |   1 +
>  platform/linux-generic/pktio/ipc.c                 | 703 +++++++++++++++++++++
>  platform/linux-generic/pktio/ring.c                |   1 +
>  11 files changed, 849 insertions(+), 4 deletions(-)
>  create mode 100644 platform/linux-generic/include/odp_packet_io_ipc_internal.h
>  create mode 100644 platform/linux-generic/include/odp_shm_internal.h
>  create mode 100644 platform/linux-generic/pktio/ipc.c
>  create mode 120000 platform/linux-generic/pktio/ring.c
> 
> diff --git a/platform/linux-generic/Makefile.am b/platform/linux-generic/Makefile.am
> index b9ed3b0..71353dd 100644
> --- a/platform/linux-generic/Makefile.am
> +++ b/platform/linux-generic/Makefile.am
> @@ -151,9 +151,11 @@ __LIB__libodp_la_SOURCES = \
>  			   odp_packet_flags.c \
>  			   odp_packet_io.c \
>  			   pktio/io_ops.c \
> +			   pktio/ipc.c \
>  			   pktio/loop.c \
>  			   pktio/socket.c \
>  			   pktio/socket_mmap.c \
> +			   pktio/ring.c \
>  			   odp_pool.c \
>  			   odp_queue.c \
>  			   odp_rwlock.c \
> diff --git a/platform/linux-generic/include/odp_buffer_internal.h b/platform/linux-generic/include/odp_buffer_internal.h
> index 4cacca1..a078e52 100644
> --- a/platform/linux-generic/include/odp_buffer_internal.h
> +++ b/platform/linux-generic/include/odp_buffer_internal.h
> @@ -132,6 +132,9 @@ struct odp_buffer_hdr_t {
>  	uint32_t                 uarea_size; /* size of user area */
>  	uint32_t                 segcount;   /* segment count */
>  	uint32_t                 segsize;    /* segment size */
> +	/* ipc mapped process can not walk over pointers,
> +	 * offset has to be used */
> +	uint64_t		 ipc_addr_offset[ODP_BUFFER_MAX_SEG];
>  	void                    *addr[ODP_BUFFER_MAX_SEG]; /* block addrs */
>  	uint64_t                 order;      /* sequence for ordered queues */
>  	queue_entry_t           *origin_qe;  /* ordered queue origin */
> diff --git a/platform/linux-generic/include/odp_packet_io_internal.h b/platform/linux-generic/include/odp_packet_io_internal.h
> index 6b03051..9fe1554 100644
> --- a/platform/linux-generic/include/odp_packet_io_internal.h
> +++ b/platform/linux-generic/include/odp_packet_io_internal.h
> @@ -23,6 +23,7 @@ extern "C" {
>  #include <odp_classification_datamodel.h>
>  #include <odp_align_internal.h>
>  #include <odp_debug_internal.h>
> +#include <odp/helper/ring.h>
>  
>  #include <odp/config.h>
>  #include <odp/hints.h>
> @@ -36,6 +37,50 @@ typedef struct {
>  	odp_bool_t promisc;		/**< promiscuous mode state */
>  } pkt_loop_t;
>  
> +typedef	struct {
> +	/* TX */
> +	struct  {
> +		odph_ring_t	*prod; /**< ODP ring for IPC msg packets
> +					  indexes transmitted to shared
> +					  memory */
> +		odph_ring_t	*cons; /**< ODP ring for IPC msg packets
> +					    indexes already processed by remote
> +					    process */
> +	} m; /* master */
> +	/* RX */
> +	struct {
> +		odph_ring_t	*prod; /**< ODP ring for IPC msg packets
> +					    indexes received from shared
> +					     memory (from remote process) */
> +		odph_ring_t	*cons; /**< ODP ring for IPC msg packets
> +					    indexes already processed by
> +					    current process */
> +	} s; /* slave */
> +	struct {
> +		odph_ring_t *r;	  /**< ring to receive from */
> +		odph_ring_t *r_p; /**< after recv is done place packet descr to
> +				       that produced ring */
> +	} recv; /* remapped above rings for easy usage in recv() */
> +	struct {
> +		odph_ring_t *r; /**< ring to transmit packets */
> +		odph_ring_t *r_p; /**< ring with already transmitted packets */
> +	} tx; /* remapped above rings for easy usage in send() */
> +	void		*pool_base;		/**< Remote pool base addr */
> +	void		*pool_mdata_base;	/**< Remote pool mdata base addr */
> +	uint64_t	pkt_size;		/**< Packet size in remote pool */
> +	odp_pool_t	pool;			/**< Pool of main process */
> +	odp_shm_t	pool_shm;		/**< Shm memory for remote pool */
> +	enum {
> +		PKTIO_TYPE_IPC_MASTER = 0, /**< Master is the process which
> +						creates shm */
> +		PKTIO_TYPE_IPC_SLAVE	   /**< Slave is the process which
> +						connects to shm */
> +	} type; /**< define if it's master or slave process */
> +	odp_atomic_u32_t ready; /**< 1 - pktio is ready and can recv/send
> +				     packet, 0 - not yet ready */
> +	void *pinfo;
> +} _ipc_pktio_t;
> +
>  struct pktio_entry {
>  	const struct pktio_if_ops *ops; /**< Implementation specific methods */
>  	odp_spinlock_t lock;		/**< entry spinlock */
> @@ -49,6 +94,7 @@ struct pktio_entry {
>  		pkt_sock_t pkt_sock;		/**< using socket API for IO */
>  		pkt_sock_mmap_t pkt_sock_mmap;	/**< using socket mmap
>  						 *   API for IO */
> +		_ipc_pktio_t ipc;		/**< IPC pktio data */
>  	};
>  	enum {
>  		STATE_START = 0,
> @@ -124,6 +170,7 @@ int pktin_poll(pktio_entry_t *entry);
>  extern const pktio_if_ops_t sock_mmsg_pktio_ops;
>  extern const pktio_if_ops_t sock_mmap_pktio_ops;
>  extern const pktio_if_ops_t loopback_pktio_ops;
> +extern const pktio_if_ops_t ipc_pktio_ops;
>  extern const pktio_if_ops_t * const pktio_if_ops[];
>  
>  #ifdef __cplusplus
> diff --git a/platform/linux-generic/include/odp_packet_io_ipc_internal.h b/platform/linux-generic/include/odp_packet_io_ipc_internal.h
> new file mode 100644
> index 0000000..4bf0550
> --- /dev/null
> +++ b/platform/linux-generic/include/odp_packet_io_ipc_internal.h
> @@ -0,0 +1,51 @@
> +/* Copyright (c) 2015, Linaro Limited
> + * All rights reserved.
> + *
> + * SPDX-License-Identifier:     BSD-3-Clause
> + */
> +
> +#include <odp/packet_io.h>
> +#include <odp_packet_io_internal.h>
> +#include <odp/packet.h>
> +#include <odp_packet_internal.h>
> +#include <odp_internal.h>
> +#include <odp/shared_memory.h>
> +
> +#include <string.h>
> +#include <unistd.h>
> +#include <stdlib.h>
> +
> +/* IPC packet I/O over shared memory ring */
> +#include <odp/helper/ring.h>
> +
> +#define PKTIO_IPC_ENTRIES     4096 /**< number of odp buffers in
> +					odp ring queue */
> +
> +/* that struct is exported to shared memory, so that 2 processes can find
> + * each other.
> + */
> +struct pktio_info {
> +	char remote_pool_name[ODP_POOL_NAME_LEN];
> +	int shm_pool_bufs_num;	/*< number of buffer in remote pool */
> +	size_t shm_pkt_pool_size; /*< size of remote pool */
> +	uint32_t shm_pkt_size; /*< size of packet/segment in remote pool */
> +	odp_shm_t shm;	/*< current structure stored in this shm */
> +	size_t mdata_offset; /*< offset from shared memory block start
> +			      *to pool_mdata_addr
> +			      * (linux-generic pool specific) */
> +	struct {
> +		size_t mdata_offset; /*< offset from shared memory block start
> +				      * to pool_mdata_addr in remote process.
> +				      * (linux-generic pool specific) */
> +		char   pool_name[ODP_POOL_NAME_LEN];
> +	} slave;
> +} ODP_PACKED;
> +
> +int ipc_pktio_init(pktio_entry_t *pktio_entry, const char *dev,
> +		   odp_pool_t pool);
> +
> +int ipc_pktio_recv(pktio_entry_t *pktio_entry, odp_packet_t pkt_table[],
> +		   unsigned len);
> +
> +int ipc_pktio_send(pktio_entry_t *pktio_entry, odp_packet_t pkt_table[],
> +		   unsigned len);
> diff --git a/platform/linux-generic/include/odp_shm_internal.h b/platform/linux-generic/include/odp_shm_internal.h
> new file mode 100644
> index 0000000..1fd7a3c
> --- /dev/null
> +++ b/platform/linux-generic/include/odp_shm_internal.h
> @@ -0,0 +1,21 @@
> +/* Copyright (c) 2013, Linaro Limited
> + * All rights reserved.
> + *
> + * SPDX-License-Identifier:     BSD-3-Clause
> + */
> +
> +#ifndef ODP_SHM_INTERNAL_H_
> +#define ODP_SHM_INTERNAL_H_
> +
> +#ifdef __cplusplus
> +extern "C" {
> +#endif
> +
> +#define _ODP_SHM_PROC_NOCREAT 0x4  /**< Do not create shm if not exist */
> +#define _ODP_SHM_O_EXCL	      0x8  /**< Do not create shm if exist */
> +
> +#ifdef __cplusplus
> +}
> +#endif
> +
> +#endif
> diff --git a/platform/linux-generic/odp_packet_io.c b/platform/linux-generic/odp_packet_io.c
> index 2e7b199..bedf221 100644
> --- a/platform/linux-generic/odp_packet_io.c
> +++ b/platform/linux-generic/odp_packet_io.c
> @@ -18,6 +18,7 @@
>  #include <odp_schedule_internal.h>
>  #include <odp_classification_internal.h>
>  #include <odp_debug_internal.h>
> +#include <odp_packet_io_ipc_internal.h>
>  
>  #include <string.h>
>  #include <sys/ioctl.h>
> diff --git a/platform/linux-generic/odp_pool.c b/platform/linux-generic/odp_pool.c
> index 2036c2a..e757235 100644
> --- a/platform/linux-generic/odp_pool.c
> +++ b/platform/linux-generic/odp_pool.c
> @@ -219,8 +219,11 @@ odp_pool_t _pool_create(const char *name,
>  			ODP_ALIGN_ROUNDUP(params->pkt.len, seg_len);
>  
>  		/* Reject create if pkt.len needs too many segments */
> -		if (blk_size / seg_len > ODP_BUFFER_MAX_SEG)
> +		if (blk_size / seg_len > ODP_BUFFER_MAX_SEG) {
> +			ODP_ERR("ODP_BUFFER_MAX_SEG exceed %d(%d)\n",
> +				blk_size / seg_len, ODP_BUFFER_MAX_SEG);
>  			return ODP_POOL_INVALID;
> +		}
>  
>  		p_udata_size = params->pkt.uarea_size;
>  		udata_stride = ODP_ALIGN_ROUNDUP(p_udata_size,
> @@ -241,8 +244,12 @@ odp_pool_t _pool_create(const char *name,
>  
>  	/* Validate requested number of buffers against addressable limits */
>  	if (buf_num >
> -	    (ODP_BUFFER_MAX_BUFFERS / (buf_stride / ODP_CACHE_LINE_SIZE)))
> +	    (ODP_BUFFER_MAX_BUFFERS / (buf_stride / ODP_CACHE_LINE_SIZE))) {
> +		ODP_ERR("buf_num %d > then expected %d\n",
> +			buf_num, ODP_BUFFER_MAX_BUFFERS /
> +			(buf_stride / ODP_CACHE_LINE_SIZE));
>  		return ODP_POOL_INVALID;
> +	}
>  
>  	/* Find an unused buffer pool slot and iniitalize it as requested */
>  	for (i = 0; i < ODP_CONFIG_POOLS; i++) {
> diff --git a/platform/linux-generic/odp_shared_memory.c b/platform/linux-generic/odp_shared_memory.c
> index ab48dda..62ce0a1 100644
> --- a/platform/linux-generic/odp_shared_memory.c
> +++ b/platform/linux-generic/odp_shared_memory.c
> @@ -15,6 +15,7 @@
>  #include <odp/debug.h>
>  #include <odp_debug_internal.h>
>  #include <odp_align_internal.h>
> +#include <odp_shm_internal.h>
>  #include <odp/config.h>
>  
>  #include <unistd.h>
> @@ -189,7 +190,7 @@ odp_shm_t odp_shm_reserve(const char *name, uint64_t size, uint64_t align,
>  	int fd = -1;
>  	int map_flag = MAP_SHARED;
>  	/* If already exists: O_EXCL: error, O_TRUNC: truncate to zero */
> -	int oflag = O_RDWR | O_CREAT | O_TRUNC;
> +	int oflag = O_RDWR;
>  	uint64_t alloc_size;
>  	uint64_t page_sz, huge_sz;
>  #ifdef MAP_HUGETLB
> @@ -207,7 +208,14 @@ odp_shm_t odp_shm_reserve(const char *name, uint64_t size, uint64_t align,
>  	alloc_hp_size = (size + align + (huge_sz - 1)) & (-huge_sz);
>  #endif
>  
> -	if (flags & ODP_SHM_PROC) {
> +	if (flags & ODP_SHM_PROC)
> +		oflag |= O_CREAT | O_TRUNC;
> +	if (flags & _ODP_SHM_O_EXCL)
> +		oflag |= O_EXCL;
> +
> +	if (flags & (ODP_SHM_PROC | _ODP_SHM_PROC_NOCREAT)) {
> +		need_huge_page = 0;
> +
>  		/* Creates a file to /dev/shm */
>  		fd = shm_open(name, oflag,
>  			      S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
> diff --git a/platform/linux-generic/pktio/io_ops.c b/platform/linux-generic/pktio/io_ops.c
> index 1d47e74..5d8d4a5 100644
> --- a/platform/linux-generic/pktio/io_ops.c
> +++ b/platform/linux-generic/pktio/io_ops.c
> @@ -12,6 +12,7 @@
>   * Array must be NULL terminated */
>  const pktio_if_ops_t * const pktio_if_ops[]  = {
>  	&loopback_pktio_ops,
> +	&ipc_pktio_ops,
>  	&sock_mmap_pktio_ops,
>  	&sock_mmsg_pktio_ops,
>  	NULL
> diff --git a/platform/linux-generic/pktio/ipc.c b/platform/linux-generic/pktio/ipc.c
> new file mode 100644
> index 0000000..835de17
> --- /dev/null
> +++ b/platform/linux-generic/pktio/ipc.c
> @@ -0,0 +1,703 @@
> +/* Copyright (c) 2015, Linaro Limited
> + * All rights reserved.
> + *
> + * SPDX-License-Identifier:     BSD-3-Clause
> + */
> +
> +#include <odp_packet_io_ipc_internal.h>
> +#include <odp_debug_internal.h>
> +#include <odp_packet_io_internal.h>
> +#include <odp_spin_internal.h>
> +#include <odp/system_info.h>
> +#include <odp_shm_internal.h>
> +
> +#include <sys/mman.h>
> +#include <sys/stat.h>
> +#include <fcntl.h>
> +
> +/* MAC address for the "ipc" interface */
> +static const char pktio_ipc_mac[] = {0x12, 0x12, 0x12, 0x12, 0x12, 0x12};
> +
> +static void *_ipc_map_remote_pool(const char *name, size_t size);
> +
> +static const char *_ipc_odp_buffer_pool_shm_name(odp_pool_t pool_hdl)
> +{
> +	pool_entry_t *pool;
> +	uint32_t pool_id;
> +	odp_shm_t shm;
> +	odp_shm_info_t info;
> +
> +	pool_id = pool_handle_to_index(pool_hdl);
> +	pool    = get_pool_entry(pool_id);
> +	shm = pool->s.pool_shm;
> +
> +	odp_shm_info(shm, &info);
> +
> +	return info.name;
> +}
> +
> +/**
> +* Look up for shared memory object.
> +*
> +* @param name   name of shm object
> +*
> +* @return 0 on success, otherwise non-zero
> +*/
> +static int _ipc_shm_lookup(const char *name)
> +{
> +	int shm;
> +
> +	shm = shm_open(name, O_RDWR, S_IRUSR | S_IWUSR);
> +	if (shm == -1) {
> +		if (errno == ENOENT)
> +			return -1;
> +		ODP_ABORT("shm_open for %s err %s\n",
> +			  name, strerror(errno));
> +	}
> +	close(shm);
> +	return 0;
> +}
> +
> +static int _ipc_map_pktio_info(pktio_entry_t *pktio_entry,
> +			       const char *dev,
> +			       int *slave)
> +{
> +	struct pktio_info *pinfo;
> +	char name[ODP_POOL_NAME_LEN + sizeof("_info")];
> +	uint32_t flags;
> +
> +	/* Create info about remote pktio */
> +	snprintf(name, sizeof(name), "%s_info", dev);
> +
> +	flags = ODP_SHM_PROC | _ODP_SHM_O_EXCL;
> +
> +	odp_shm_t shm = odp_shm_reserve(name, sizeof(struct pktio_info),
> +			ODP_CACHE_LINE_SIZE,
> +			flags);
> +	if (ODP_SHM_INVALID != shm) {
> +		pinfo = odp_shm_addr(shm);
> +		pinfo->remote_pool_name[0] = 0;
> +		*slave = 0;
> +	} else {
> +		flags = _ODP_SHM_PROC_NOCREAT | _ODP_SHM_O_EXCL;
> +		odp_shm_t shm = odp_shm_reserve(name, sizeof(struct pktio_info),
> +				ODP_CACHE_LINE_SIZE,
> +				flags);
> +		if (ODP_SHM_INVALID == shm)
> +			ODP_ABORT("can not connect to shm\n");
> +
> +		pinfo = odp_shm_addr(shm);
> +		*slave = 1;
> +	}
> +
> +	pktio_entry->s.ipc.pinfo = pinfo;
> +	pktio_entry->s.ipc.pool_shm = shm;
> +
> +	return 0;
> +}
> +
> +static int _ipc_master_post_init(pktio_entry_t *pktio_entry)
> +{
> +	struct pktio_info *pinfo = pktio_entry->s.ipc.pinfo;
> +	int ret;
> +	void *ipc_pool_base;
> +
> +	if (pinfo->slave.mdata_offset == 0)
> +		return -1;
> +
> +	ret = _ipc_shm_lookup(pinfo->slave.pool_name);
> +	if (ret) {
> +		ODP_DBG("no pool file %s\n", pinfo->slave.pool_name);
> +		return -1;
> +	}
> +
> +	ipc_pool_base = _ipc_map_remote_pool(pinfo->slave.pool_name,
> +					     pinfo->shm_pkt_pool_size);
> +	pktio_entry->s.ipc.pool_mdata_base = (char *)ipc_pool_base +
> +					     pinfo->slave.mdata_offset;
> +
> +	/* recv() rings */
> +	pktio_entry->s.ipc.recv.r = pktio_entry->s.ipc.s.prod;
> +	pktio_entry->s.ipc.recv.r_p = pktio_entry->s.ipc.s.cons;
> +	/* tx() rings */
> +	pktio_entry->s.ipc.tx.r = pktio_entry->s.ipc.m.prod;
> +	pktio_entry->s.ipc.tx.r_p = pktio_entry->s.ipc.m.cons;
> +
> +	odp_atomic_store_u32(&pktio_entry->s.ipc.ready, 1);
> +
> +	ODP_DBG("Post init... DONE.\n");
> +	return 0;
> +}
> +
> +static int _ipc_init_master(pktio_entry_t *pktio_entry,
> +			    const char *dev,
> +			    odp_pool_t pool)
> +{
> +	char ipc_shm_name[ODP_POOL_NAME_LEN + sizeof("_slave_r")];
> +	pool_entry_t *pool_entry;
> +	uint32_t pool_id;
> +	struct pktio_info *pinfo;
> +	const char *pool_name;
> +	odp_shm_t shm;
> +
> +	pool_id = pool_handle_to_index(pool);
> +	pool_entry    = get_pool_entry(pool_id);
> +
> +	if (strlen(dev) > (ODP_POOL_NAME_LEN - sizeof("_slave_r"))) {
> +		ODP_DBG("too big ipc name\n");
> +		return -1;
> +	}
> +
> +	/* generate name in shm like ipc_pktio_r for
> +	 * to be processed packets ring.
> +	 */
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_prod", dev);
> +	pktio_entry->s.ipc.m.prod = odph_ring_create(ipc_shm_name,
> +			PKTIO_IPC_ENTRIES,
> +			ODPH_RING_SHM_PROC | ODPH_RING_NO_LIST);
> +	if (!pktio_entry->s.ipc.m.prod) {
> +		ODP_DBG("pid %d unable to create ipc ring %s name\n",
> +			getpid(), ipc_shm_name);
> +		return -1;
> +	}
> +	ODP_DBG("Created IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.m.prod),
> +		odph_ring_free_count(pktio_entry->s.ipc.m.prod));
> +
> +	/* generate name in shm like ipc_pktio_p for
> +	 * already processed packets
> +	 */
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_cons", dev);
> +	pktio_entry->s.ipc.m.cons = odph_ring_create(ipc_shm_name,
> +			PKTIO_IPC_ENTRIES,
> +			ODPH_RING_SHM_PROC | ODPH_RING_NO_LIST);
> +	if (!pktio_entry->s.ipc.m.cons) {
> +		ODP_DBG("pid %d unable to create ipc ring %s name\n",
> +			getpid(), ipc_shm_name);
> +		goto free_m_prod;
> +	}
> +	ODP_DBG("Created IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.m.cons),
> +		odph_ring_free_count(pktio_entry->s.ipc.m.cons));
> +
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_s_prod", dev);
> +	pktio_entry->s.ipc.s.prod = odph_ring_create(ipc_shm_name,
> +			PKTIO_IPC_ENTRIES,
> +			ODPH_RING_SHM_PROC | ODPH_RING_NO_LIST);
> +	if (!pktio_entry->s.ipc.s.prod) {
> +		ODP_DBG("pid %d unable to create ipc ring %s name\n",
> +			getpid(), ipc_shm_name);
> +		goto free_m_cons;
> +	}
> +	ODP_DBG("Created IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.s.prod),
> +		odph_ring_free_count(pktio_entry->s.ipc.s.prod));
> +
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_s_cons", dev);
> +	pktio_entry->s.ipc.s.cons = odph_ring_create(ipc_shm_name,
> +			PKTIO_IPC_ENTRIES,
> +			ODPH_RING_SHM_PROC | ODPH_RING_NO_LIST);
> +	if (!pktio_entry->s.ipc.s.cons) {
> +		ODP_DBG("pid %d unable to create ipc ring %s name\n",
> +			getpid(), ipc_shm_name);
> +		goto free_s_prod;
> +	}
> +	ODP_DBG("Created IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.s.cons),
> +		odph_ring_free_count(pktio_entry->s.ipc.s.cons));
> +
> +	/* Set up pool name for remote info */
> +	pinfo = pktio_entry->s.ipc.pinfo;
> +	pool_name = _ipc_odp_buffer_pool_shm_name(pool);
> +	memcpy(pinfo->remote_pool_name, pool_name, strlen(pool_name));
> +	pinfo->shm_pkt_pool_size = pool_entry->s.pool_size;
> +	pinfo->shm_pool_bufs_num = pool_entry->s.buf_num;
> +	pinfo->shm_pkt_size = pool_entry->s.seg_size;
> +	pinfo->mdata_offset =  pool_entry->s.pool_mdata_addr -
> +			       pool_entry->s.pool_base_addr;
> +	pinfo->slave.mdata_offset = 0;
> +
> +	pktio_entry->s.ipc.pool = pool;
> +
> +	ODP_DBG("Pre init... DONE.\n");
> +
> +	_ipc_master_post_init(pktio_entry);
> +
> +	return 0;
> +
> +free_s_prod:
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_s_prod", dev);
> +	shm = odp_shm_lookup(ipc_shm_name);
> +	odp_shm_free(shm);
> +free_m_cons:
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_cons", dev);
> +	shm = odp_shm_lookup(ipc_shm_name);
> +	odp_shm_free(shm);
> +free_m_prod:
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_prod", dev);
> +	shm = odp_shm_lookup(ipc_shm_name);
> +	odp_shm_free(shm);
> +	return -1;
> +}
> +
> +static void _odp_ipc_export_pool(struct pktio_info *pinfo,
> +				 odp_pool_t pool)
> +{
> +	pool_entry_t *pool_entry;
> +
> +	pool_entry = odp_pool_to_entry(pool);
> +	if (pool_entry->s.blk_size != pinfo->shm_pkt_size)
> +		ODP_ABORT("pktio for same name should have the same pool size\n");
> +	if (pool_entry->s.buf_num != (unsigned)pinfo->shm_pool_bufs_num)
> +		ODP_ABORT("pktio for same name should have the same pool size\n");
> +
> +	snprintf(pinfo->slave.pool_name, ODP_POOL_NAME_LEN, "%s",
> +		 pool_entry->s.name);
> +	pinfo->slave.mdata_offset = pool_entry->s.pool_mdata_addr -
> +				    pool_entry->s.pool_base_addr;
> +}
> +
> +static void *_ipc_map_remote_pool(const char *name, size_t size)
> +{
> +	odp_shm_t shm;
> +	void *addr;
> +
> +	ODP_DBG("Mapping remote pool %s, size %ld\n", name, size);
> +	shm = odp_shm_reserve(name,
> +			      size,
> +			      ODP_CACHE_LINE_SIZE,
> +			      _ODP_SHM_PROC_NOCREAT);
> +	if (shm == ODP_SHM_INVALID)
> +		ODP_ABORT("unable map %s\n", name);
> +
> +	addr = odp_shm_addr(shm);
> +	ODP_DBG("MAP master: %p - %p size %ld, pool %s\n",
> +		addr, (char *)addr + size, size, name);
> +	return addr;
> +}
> +
> +static void *_ipc_shm_map(char *name, size_t size)
> +{
> +	odp_shm_t shm;
> +	int ret;
> +
> +	ret = _ipc_shm_lookup(name);
> +	if (ret == -1)
> +		return NULL;
> +
> +	shm = odp_shm_reserve(name, size,
> +			      ODP_CACHE_LINE_SIZE,
> +			      _ODP_SHM_PROC_NOCREAT);
> +	if (ODP_SHM_INVALID == shm)
> +		ODP_ABORT("unable to map: %s\n", name);
> +
> +	return odp_shm_addr(shm);
> +}
> +
> +static int _ipc_init_slave(const char *dev,
> +			   pktio_entry_t *pktio_entry,
> +			   odp_pool_t pool)
> +{
> +	if (strlen(dev) > (ODP_POOL_NAME_LEN - sizeof("_slave_r")))
> +		ODP_ABORT("too big ipc name\n");
> +
> +	pktio_entry->s.ipc.pool = pool;
> +	return 0;
> +}
> +
> +static int _ipc_slave_post_init(pktio_entry_t *pktio_entry)
> +{
> +	char ipc_shm_name[ODP_POOL_NAME_LEN + sizeof("_slave_r")];
> +	size_t ring_size = PKTIO_IPC_ENTRIES * sizeof(void *) +
> +			   sizeof(odph_ring_t);
> +	struct pktio_info *pinfo;
> +	void *ipc_pool_base;
> +	odp_shm_t shm;
> +	const char *dev = pktio_entry->s.name;
> +
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_prod", dev);
> +	pktio_entry->s.ipc.m.prod  = _ipc_shm_map(ipc_shm_name, ring_size);
> +	if (!pktio_entry->s.ipc.m.prod) {
> +		ODP_DBG("pid %d unable to find ipc ring %s name\n",
> +			getpid(), dev);
> +		sleep(1);
> +		return -1;
> +	}
> +	ODP_DBG("Connected IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.m.prod),
> +		odph_ring_free_count(pktio_entry->s.ipc.m.prod));
> +
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_cons", dev);
> +	pktio_entry->s.ipc.m.cons = _ipc_shm_map(ipc_shm_name, ring_size);
> +	if (!pktio_entry->s.ipc.m.cons) {
> +		ODP_DBG("pid %d unable to find ipc ring %s name\n",
> +			getpid(), dev);
> +		goto free_m_prod;
> +	}
> +	ODP_DBG("Connected IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.m.cons),
> +		odph_ring_free_count(pktio_entry->s.ipc.m.cons));
> +
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_s_prod", dev);
> +	pktio_entry->s.ipc.s.prod = _ipc_shm_map(ipc_shm_name, ring_size);
> +	if (!pktio_entry->s.ipc.s.prod) {
> +		ODP_DBG("pid %d unable to find ipc ring %s name\n",
> +			getpid(), dev);
> +		goto free_m_cons;
> +	}
> +	ODP_DBG("Connected IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.s.prod),
> +		odph_ring_free_count(pktio_entry->s.ipc.s.prod));
> +
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_s_cons", dev);
> +	pktio_entry->s.ipc.s.cons = _ipc_shm_map(ipc_shm_name, ring_size);
> +	if (!pktio_entry->s.ipc.s.cons) {
> +		ODP_DBG("pid %d unable to find ipc ring %s name\n",
> +			getpid(), dev);
> +		goto free_s_prod;
> +	}
> +	ODP_DBG("Connected IPC ring: %s, count %d, free %d\n",
> +		ipc_shm_name, odph_ring_count(pktio_entry->s.ipc.s.cons),
> +		odph_ring_free_count(pktio_entry->s.ipc.s.cons));
> +
> +	/* Get info about remote pool */
> +	pinfo = pktio_entry->s.ipc.pinfo;
> +	ipc_pool_base = _ipc_map_remote_pool(pinfo->remote_pool_name,
> +					     pinfo->shm_pkt_pool_size);
> +	pktio_entry->s.ipc.pool_mdata_base = (char *)ipc_pool_base +
> +					     pinfo->mdata_offset;
> +	pktio_entry->s.ipc.pkt_size = pinfo->shm_pkt_size;
> +
> +	/* @todo: to simplify in linux-generic implementation we create pool for
> +	 * packets from IPC queue. On receive implementation copies packets to
> +	 * that pool. Later we can try to reuse original pool without packets
> +	 * copying. (pkt refcounts needs to be implemented).
> +	 */
> +	_odp_ipc_export_pool(pinfo, pktio_entry->s.ipc.pool);
> +
> +	/* recv() rings */
> +	pktio_entry->s.ipc.recv.r = pktio_entry->s.ipc.m.prod;
> +	pktio_entry->s.ipc.recv.r_p = pktio_entry->s.ipc.m.cons;
> +	/* tx() rings */
> +	pktio_entry->s.ipc.tx.r = pktio_entry->s.ipc.s.prod;
> +	pktio_entry->s.ipc.tx.r_p = pktio_entry->s.ipc.s.cons;
> +

This isn't exactly what I had in mind, can't you just use these names
directly in the first place in the code above? Is there any reason to
retain two copies of the pointers?

> +	odp_atomic_store_u32(&pktio_entry->s.ipc.ready, 1);
> +
> +	ODP_DBG("Post init... DONE.\n");
> +	return 0;
> +
> +free_s_prod:
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_s_prod", dev);
> +	shm = odp_shm_lookup(ipc_shm_name);
> +	odp_shm_free(shm);
> +free_m_cons:
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_cons", dev);
> +	shm = odp_shm_lookup(ipc_shm_name);
> +	odp_shm_free(shm);
> +free_m_prod:
> +	snprintf(ipc_shm_name, sizeof(ipc_shm_name), "%s_m_prod", dev);
> +	shm = odp_shm_lookup(ipc_shm_name);
> +	odp_shm_free(shm);
> +	return -1;
> +}
> +
> +static int ipc_pktio_open(odp_pktio_t id ODP_UNUSED,
> +			  pktio_entry_t *pktio_entry,
> +			  const char *dev,
> +			  odp_pool_t pool)
> +{
> +	int ret = -1;
> +	int slave;
> +
> +	_ODP_STATIC_ASSERT(ODP_POOL_NAME_LEN == ODPH_RING_NAMESIZE,
> +			   "mismatch pool and ring name arrays");
> +
> +	if (strncmp(dev, "ipc", 3))
> +		return -1;
> +
> +	odp_atomic_init_u32(&pktio_entry->s.ipc.ready, 0);
> +
> +	_ipc_map_pktio_info(pktio_entry, dev, &slave);
> +	pktio_entry->s.ipc.type = (slave == 0) ? PKTIO_TYPE_IPC_MASTER :
> +						 PKTIO_TYPE_IPC_SLAVE;
> +
> +	switch (pktio_entry->s.ipc.type) {
> +	case PKTIO_TYPE_IPC_MASTER:
> +		ODP_DBG("process %d is master\n", getpid());
> +		ret = _ipc_init_master(pktio_entry, dev, pool);
> +		break;
> +	case PKTIO_TYPE_IPC_SLAVE:
> +		ODP_DBG("process %d is slave\n", getpid());
> +		ret = _ipc_init_slave(dev, pktio_entry, pool);
> +		break;
> +	default:
> +		ODP_ABORT("");
> +	}

I commented on the previous version about changing this to an if/else
and removing the abort.

> +
> +	return ret;
> +}

There's no odp_pktio_close() implementation, what happens when the
"master" process exits? I guess it doesn't do anything to unlink the
shmems and they just hang around?

I'm wondering about different process start/exit sequences:

  p1      p2

start
        start
        exit
exit

I suppose this works, but what if p1 (master) exits first?

What if p1 exits then p3 starts and connects to p2, are p2 and p3 now
both slaves? would this work?

At which point do the shmems actually get freed?

Ah, I see now that cleanup is being done in
platform/linux-generic/test/pktio_ipc/pktio_ipc_run - that's a bit of a
hack, I presume the tests fail without that rm?

> +
> +static inline void *_ipc_buffer_map(odp_buffer_hdr_t *buf,
> +				    uint32_t offset,
> +				    uint32_t *seglen,
> +				    uint32_t limit)
> +{
> +	int seg_index  = offset / buf->segsize;
> +	int seg_offset = offset % buf->segsize;
> +	void *addr = (char *)buf - buf->ipc_addr_offset[seg_index];
> +
> +	if (seglen) {
> +		uint32_t buf_left = limit - offset;
> +		*seglen = seg_offset + buf_left <= buf->segsize ?
> +			buf_left : buf->segsize - seg_offset;
> +	}
> +
> +	return (void *)(seg_offset + (uint8_t *)addr);
> +}
> +
> +static inline void *_ipc_packet_map(odp_packet_hdr_t *pkt_hdr,
> +				    uint32_t offset, uint32_t *seglen)
> +{
> +	if (offset > pkt_hdr->frame_len)
> +		return NULL;
> +
> +	return _ipc_buffer_map(&pkt_hdr->buf_hdr,
> +			  pkt_hdr->headroom + offset, seglen,
> +			  pkt_hdr->headroom + pkt_hdr->frame_len);
> +}
> +
> +int ipc_pktio_recv(pktio_entry_t *pktio_entry,
> +		   odp_packet_t pkt_table[], unsigned len)
> +{
> +	int pkts = 0;
> +	int i;
> +	odph_ring_t *r;
> +	odph_ring_t *r_p;
> +
> +	odp_packet_t remote_pkts[PKTIO_IPC_ENTRIES];
> +	void **ipcbufs_p = (void *)&remote_pkts;
> +	uint32_t ready = odp_atomic_load_u32(&pktio_entry->s.ipc.ready);
> +
> +	if (odp_unlikely(!ready)) {
> +		if (pktio_entry->s.ipc.type == PKTIO_TYPE_IPC_MASTER) {
> +			if (_ipc_master_post_init(pktio_entry))
> +				return 0;
> +		} else if (pktio_entry->s.ipc.type == PKTIO_TYPE_IPC_SLAVE) {
> +			if (_ipc_slave_post_init(pktio_entry))
> +				return 0;
> +		}

I thought you were going to get rid of these post_init calls, why are
they needed here?

> +	}
> +
> +	/* Free already processed packets, for send */
> +	while (1) {
> +		odp_packet_t r_p_pkts[PKTIO_IPC_ENTRIES];
> +		int ret;
> +		void **rbuf_p;
> +		odph_ring_t *tx_r_p;
> +
> +		rbuf_p = (void *)&r_p_pkts;
> +		tx_r_p = pktio_entry->s.ipc.tx.r_p;
> +		ret = odph_ring_mc_dequeue_burst(tx_r_p, rbuf_p,
> +						 PKTIO_IPC_ENTRIES);
> +		if (0 == ret)
> +			break;
> +		for (i = 0; i < ret; i++) {
> +			if (r_p_pkts[i] != ODP_PACKET_INVALID)
> +				odp_packet_free(r_p_pkts[i]);
> +		}
> +	}

Put the above loop into a helper? (and use it on send too)

> +
> +	r = pktio_entry->s.ipc.recv.r;
> +	pkts = odph_ring_mc_dequeue_burst(r, ipcbufs_p, len);
> +	if (odp_unlikely(pkts < 0))
> +		ODP_ABORT("error to dequeue no packets\n");
> +
> +	/* fast path */
> +	if (odp_likely(0 == pkts))
> +		return 0;
> +
> +	for (i = 0; i < pkts; i++) {
> +		odp_pool_t pool;
> +		odp_packet_t pkt;
> +		odp_packet_hdr_t phdr;
> +		void *ptr;
> +		odp_buffer_bits_t handle;
> +		int idx; /* Remote packet has coded pool and index.
> +			  * We need only index.*/
> +		void *pkt_data;
> +		void *remote_pkt_data;
> +
> +		if (remote_pkts[i] == ODP_PACKET_INVALID)
> +			continue;
> +
> +		handle.handle = _odp_packet_to_buffer(remote_pkts[i]);
> +		idx = handle.index;
> +
> +		/* Link to packed data. To this line we have Zero-Copy between
> +		 * processes, to simplify use packet copy in that version which
> +		 * can be removed later with more advance buffer management
> +		 * (ref counters).
> +		 */
> +		/* reverse odp_buf_to_hdr() */
> +		ptr = (char *)pktio_entry->s.ipc.pool_mdata_base +
> +		      (idx * ODP_CACHE_LINE_SIZE);
> +		memcpy(&phdr, ptr, sizeof(odp_packet_hdr_t));
> +
> +		/* Allocate new packet. Select*/
> +		pool = pktio_entry->s.ipc.pool;
> +		if (odp_unlikely(pool == ODP_POOL_INVALID))
> +			ODP_ABORT("invalid pool");
> +
> +		pkt = odp_packet_alloc(pool, phdr.frame_len);
> +		if (odp_unlikely(pkt == ODP_PACKET_INVALID)) {
> +			/* Original pool might be smaller then
> +			*  PKTIO_IPC_ENTRIES. If packet can not be
> +			 * allocated from pool at this time,
> +			 * simple get in on next recv() call.
> +			 */
> +			if (i == 0)
> +				return 0;
> +			break;
> +		}
> +
> +		/* Copy packet data. */
> +		pkt_data = odp_packet_data(pkt);
> +		if (odp_unlikely(!pkt_data))
> +			ODP_ABORT("unable to map pkt_data ipc_slave %d\n",
> +				  (PKTIO_TYPE_IPC_SLAVE ==
> +					pktio_entry->s.ipc.type));
> +
> +		remote_pkt_data = _ipc_packet_map(ptr, 0, NULL);
> +		if (odp_unlikely(!remote_pkt_data))
> +			ODP_ABORT("unable to map remote_pkt_data, ipc_slave %d\n",
> +				  (PKTIO_TYPE_IPC_SLAVE ==
> +					pktio_entry->s.ipc.type));
> +
> +		/* @todo fix copy packet!!! */
> +		memcpy(pkt_data, remote_pkt_data, phdr.frame_len);
> +
> +		/* Copy packets L2, L3 parsed offsets and size */
> +		copy_packet_parser_metadata(&phdr, odp_packet_hdr(pkt));
> +
> +		odp_packet_hdr(pkt)->frame_len = phdr.frame_len;
> +		odp_packet_hdr(pkt)->headroom = phdr.headroom;
> +		odp_packet_hdr(pkt)->tailroom = phdr.tailroom;
> +		pkt_table[i] = pkt;
> +	}
> +
> +	/* Now tell other process that we no longer need that buffers.*/
> +	r_p = pktio_entry->s.ipc.recv.r_p;
> +	pkts = odph_ring_mp_enqueue_burst(r_p, ipcbufs_p, i);
> +	if (odp_unlikely(pkts < 0))
> +		ODP_ABORT("ipc: odp_ring_mp_enqueue_bulk r_p fail\n");
> +
> +	return pkts;
> +}
> +
> +int ipc_pktio_send(pktio_entry_t *pktio_entry, odp_packet_t pkt_table[],
> +		   unsigned len)
> +{
> +	odph_ring_t *r;
> +	void **rbuf_p;
> +	int ret;
> +	unsigned i;
> +	uint32_t ready = odp_atomic_load_u32(&pktio_entry->s.ipc.ready);
> +
> +	if (odp_unlikely(!ready))
> +		return 0;
> +
> +	/* Free already processed packets, if any */
> +	while (1) {
> +		odp_packet_t r_p_pkts[PKTIO_IPC_ENTRIES];
> +		odph_ring_t *r_p;
> +
> +		rbuf_p = (void *)&r_p_pkts;
> +		r_p = pktio_entry->s.ipc.tx.r_p;
> +		ret = odph_ring_mc_dequeue_burst(r_p, rbuf_p,
> +						 PKTIO_IPC_ENTRIES);
> +		if (0 == ret)
> +			break;
> +		for (i = 0; i < (unsigned)ret; i++) {
> +			if (r_p_pkts[i] != ODP_PACKET_INVALID)
> +				odp_packet_free(r_p_pkts[i]);
> +		}
> +	}
> +
> +	/* Prepare packets: calculate offset from address. */
> +	for (i = 0; i < len; i++) {
> +		int j;
> +		odp_packet_t pkt =  pkt_table[i];
> +		odp_packet_hdr_t *pkt_hdr = odp_packet_hdr(pkt);
> +		odp_buffer_bits_t handle;
> +		uint32_t cur_mapped_pool_id =
> +			 pool_handle_to_index(pktio_entry->s.ipc.pool);
> +		uint32_t pool_id;
> +
> +		/* do copy if packet was allocated from not mapped pool */
> +		handle.handle = _odp_packet_to_buffer(pkt);
> +		pool_id = handle.pool_id;
> +		if (pool_id != cur_mapped_pool_id) {
> +			odp_packet_t newpkt;
> +
> +			newpkt = odp_packet_copy(pkt, pktio_entry->s.ipc.pool);
> +			if (newpkt == ODP_PACKET_INVALID)
> +				ODP_ABORT("Unable to copy packet\n");
> +
> +			odp_packet_free(pkt);
> +			pkt_table[i] = newpkt;
> +		}
> +
> +		rbuf_p = (void *)&pkt;
> +
> +		/* buf_hdr.addr can not be used directly in remote process,
> +		 * convert it to offset
> +		 */
> +		for (j = 0; j < ODP_BUFFER_MAX_SEG; j++) {
> +			pkt_hdr->buf_hdr.ipc_addr_offset[j] = (char *)pkt_hdr -
> +				(char *)pkt_hdr->buf_hdr.addr[j];
> +		}
> +	}
> +
> +	/* Put packets to ring to be processed in other process. */
> +	rbuf_p = (void *)&pkt_table[0];
> +	r = pktio_entry->s.ipc.tx.r;
> +	ret = odph_ring_mp_enqueue_burst(r, rbuf_p, len);
> +	if (odp_unlikely(ret < 0)) {
> +		ODP_ERR("pid %d odp_ring_mp_enqueue_bulk fail, ipc_slave %d, ret %d\n",
> +			getpid(),
> +			(PKTIO_TYPE_IPC_SLAVE == pktio_entry->s.ipc.type),
> +			ret);
> +		ODP_ERR("odp_ring_full: %d, odp_ring_count %d, odph_ring_free_count %d\n",
> +			odph_ring_full(r), odph_ring_count(r),
> +			odph_ring_free_count(r));
> +	}
> +
> +	return ret;
> +}
> +
> +static int ipc_mtu_get(pktio_entry_t *pktio_entry ODP_UNUSED)
> +{
> +	/* mtu not limited, pool settings are used. */
> +	return (9 * 1024);
> +}
> +
> +static int ipc_mac_addr_get(pktio_entry_t *pktio_entry ODP_UNUSED,
> +			    void *mac_addr)
> +{
> +	memcpy(mac_addr, pktio_ipc_mac, ETH_ALEN);
> +	return ETH_ALEN;
> +}
> +
> +const pktio_if_ops_t ipc_pktio_ops = {
> +	.init = NULL,
> +	.term = NULL,
> +	.open = ipc_pktio_open,
> +	.close = NULL,
> +	.recv =  ipc_pktio_recv,
> +	.send = ipc_pktio_send,
> +	.mtu_get = ipc_mtu_get,
> +	.promisc_mode_set = NULL,
> +	.promisc_mode_get = NULL,
> +	.mac_get = ipc_mac_addr_get
> +};
> diff --git a/platform/linux-generic/pktio/ring.c b/platform/linux-generic/pktio/ring.c
> new file mode 120000
> index 0000000..d35c589
> --- /dev/null
> +++ b/platform/linux-generic/pktio/ring.c
> @@ -0,0 +1 @@
> +../../../helper/ring.c
> \ No newline at end of file
> -- 
> 1.9.1
>

slave:

On 10/20/2015 15:41, Stuart Haslam wrote:
>> +	/* recv() rings */
>> >+	pktio_entry->s.ipc.recv.r = pktio_entry->s.ipc.m.prod;
>> >+	pktio_entry->s.ipc.recv.r_p = pktio_entry->s.ipc.m.cons;
>> >+	/* tx() rings */
>> >+	pktio_entry->s.ipc.tx.r = pktio_entry->s.ipc.s.prod;
>> >+	pktio_entry->s.ipc.tx.r_p = pktio_entry->s.ipc.s.cons;
>> >+
> This isn't exactly what I had in mind, can't you just use these names
> directly in the first place in the code above? Is there any reason to
> retain two copies of the pointers?
>
master:

+	/* recv() rings */
+	pktio_entry->s.ipc.recv.r = pktio_entry->s.ipc.s.prod;
+	pktio_entry->s.ipc.recv.r_p = pktio_entry->s.ipc.s.cons;
+	/* tx() rings */
+	pktio_entry->s.ipc.tx.r = pktio_entry->s.ipc.m.prod;
+	pktio_entry->s.ipc.tx.r_p = pktio_entry->s.ipc.m.cons;
+


They are crossed. Instead of implementing 2 function slave_recv() 
master_recv()
I add added one and this links.

But I can rename rings something like:

pktio_entry->s.ipc.local.prod
pktio_entry->s.ipc.local.cons
pktio_entry->s.ipc.remote.prod
pktio_entry->s.ipc.remote.cons

local - will reflect local pool/ring mappings;
remote - for remote process poll/ring mappings;

Maxim.

On 10/20/2015 15:41, Stuart Haslam wrote:
>> +int ipc_pktio_recv(pktio_entry_t *pktio_entry,
>> >+		   odp_packet_t pkt_table[], unsigned len)
>> >+{
>> >+	int pkts = 0;
>> >+	int i;
>> >+	odph_ring_t *r;
>> >+	odph_ring_t *r_p;
>> >+
>> >+	odp_packet_t remote_pkts[PKTIO_IPC_ENTRIES];
>> >+	void **ipcbufs_p = (void *)&remote_pkts;
>> >+	uint32_t ready = odp_atomic_load_u32(&pktio_entry->s.ipc.ready);
>> >+
>> >+	if (odp_unlikely(!ready)) {
>> >+		if (pktio_entry->s.ipc.type == PKTIO_TYPE_IPC_MASTER) {
>> >+			if (_ipc_master_post_init(pktio_entry))
>> >+				return 0;
>> >+		} else if (pktio_entry->s.ipc.type == PKTIO_TYPE_IPC_SLAVE) {
>> >+			if (_ipc_slave_post_init(pktio_entry))
>> >+				return 0;
>> >+		}
> I thought you were going to get rid of these post_init calls, why are
> they needed here?
>

I don't know when second process will be connected to shm. After it's 
connected I need to map pool of second process to first one.
That can go to odp_pktio_start(). I.e. first process can call start() 
until second process is not connected and than do post init.

Maxim.