Re: [dpdk-dev] [PATCH] Fix librte_pmd_ring: connect primary and secondary ring with correct port.

[Masaru Oki <m-oki@stratosphere.co.jp>]

thank you for reply.

> From my reading, this is really two patches:
> 1. add in a port id to packets received from an eth_ring port
> 2. change the rte_eth_ring_create function to create two sets of rings.
> so perhaps this could be split up into two proposed patches.

yes.

> A ring or a ring-ethdev is a undirectional channel, and
> this is explicitly by design.

Hmm, I missed usage of ring-ethdev?
I want to connect bidirectional port in two DPDK program by ring-ethdev or
other pmd,

at first, I tried to connect port in 1 process.
run DPDK app with command line
--vdev eth_ring0,nodeaction=ring0:0:CREATE,nodeaction=ring0:0:ATTACH

port 0 and port 1 has created and connected (it looks like to me),
and rx and tx queue in both ports shares only one rte_ring in original code
with above command line.
received packet by ANY port if sent packet from ANY port.  so bad.

second, I tried with command line
--vdev eth_ring0,nodeaction=ring00:0:CREATE,nodeaction=ring01:0:CREATE,
nodeaction=ring01:0:ATTACH,nodeaction=ring00:0:ATTACH

It gots log by dpdk:
init (0) eth_ring0
PMD: Initializing pmd_ring for eth_ring0
PMD: Creating rings-backed ethdev on numa socket 0
EAL: memzone_reserve_aligned_thread_unsafe(): memzone
<RG_ETH_RXTX0_eth_ring0> already exists
RING: Cannot reserve memory
PMD: Creating rings-backed ethdev on numa socket 0
PMD: Creating rings-backed ethdev on numa socket 0

and created THREE ports.  also bad.

How to connected port 0 rx and port1 tx without patch?  (also port 0 tx and
port 1 rx)

by the way, I found wanted code in rte_pmd_ring.c
but it marked as deprecated (and no longer used).



2014-10-14 19:17 GMT+09:00 Bruce Richardson <bruce.richardson@intel.com>:

> On Tue, Oct 07, 2014 at 11:11:45PM +0900, Masaru OKI wrote:
> > librte_pmd_ring provides created port and attached port.
> > Packet is received from attached port if packet is sent to created port.
> > So, packet is received from created port if packet is sent to attached
> port.
> > It must be need two rings such as "create to attach" and "attach to
> create".
> > But current implementation uses only one ring for rx/tx.
> > It causes incorrect result.
> > Fixed:
> > - Make ring both rx and tx
> > - Connect created (primary) ring and attached (secondary) ring
> > - Correct m->port like librte_pmd_pcap
> >
> > Signed-off-by: Masaru OKI <m-oki@stratosphere.co.jp>
>
> From my reading, this is really two patches:
> 1. add in a port id to packets received from an eth_ring port
> 2. change the rte_eth_ring_create function to create two sets of rings.
> so perhaps this could be split up into two proposed patches.
>
> Unfortunately, while I have no issue with the first part, I disagree with
> the second part. A ring or a ring-ethdev is a undirectional channel, and
> this is explicitly by design. When you create or attach to a ring pmd
> instance (or ring ethdev), you get exactly that, an ethdev instance that
> has
> a ring, or set of rings internally. It's up to the application how to use
> those rings. If you need a bidirectional channel, you need to create two
> ring ethdevs, in exactly the same way that if you want bidirectional
> messaging between two cores using rte_rings, you need to create two rings,
> one for each direction.
>
> regards,
> /Bruce
>
> > ---
> >  lib/librte_pmd_ring/rte_eth_ring.c | 36
> +++++++++++++++++++++++++++---------
> >  1 file changed, 27 insertions(+), 9 deletions(-)
> >
> > diff --git a/lib/librte_pmd_ring/rte_eth_ring.c
> b/lib/librte_pmd_ring/rte_eth_ring.c
> > index 4f1b6ed..d926d00 100644
> > --- a/lib/librte_pmd_ring/rte_eth_ring.c
> > +++ b/lib/librte_pmd_ring/rte_eth_ring.c
> > @@ -54,6 +54,7 @@ struct ring_queue {
> >       rte_atomic64_t rx_pkts;
> >       rte_atomic64_t tx_pkts;
> >       rte_atomic64_t err_pkts;
> > +     uint8_t in_port;
> >  };
> >
> >  struct pmd_internals {
> > @@ -80,10 +81,14 @@ eth_ring_rx(void *q, struct rte_mbuf **bufs,
> uint16_t nb_bufs)
> >       struct ring_queue *r = q;
> >       const uint16_t nb_rx = (uint16_t)rte_ring_dequeue_burst(r->rng,
> >                       ptrs, nb_bufs);
> > +        uint16_t cnt;
> >       if (r->rng->flags & RING_F_SC_DEQ)
> >               r->rx_pkts.cnt += nb_rx;
> >       else
> >               rte_atomic64_add(&(r->rx_pkts), nb_rx);
> > +        for (cnt = 0; cnt < nb_rx; cnt++) {
> > +          bufs[cnt]->port = r->in_port;
> > +        }
> >       return nb_rx;
> >  }
> >
> > @@ -129,6 +134,8 @@ eth_rx_queue_setup(struct rte_eth_dev *dev,uint16_t
> rx_queue_id,
> >  {
> >       struct pmd_internals *internals = dev->data->dev_private;
> >       dev->data->rx_queues[rx_queue_id] =
> &internals->rx_ring_queues[rx_queue_id];
> > +     internals->rx_ring_queues[rx_queue_id].in_port =
> dev->data->port_id;
> > +
> >       return 0;
> >  }
> >
> > @@ -319,23 +326,34 @@ eth_dev_ring_create(const char *name, const
> unsigned numa_node,
> >       /* rx and tx are so-called from point of view of first port.
> >        * They are inverted from the point of view of second port
> >        */
> > -     struct rte_ring *rxtx[RTE_PMD_RING_MAX_RX_RINGS];
> > +     struct rte_ring *rx[RTE_PMD_RING_MAX_RX_RINGS];
> > +     struct rte_ring *tx[RTE_PMD_RING_MAX_TX_RINGS];
> >       unsigned i;
> > -     char rng_name[RTE_RING_NAMESIZE];
> > +     char rng_rxname[RTE_RING_NAMESIZE];
> > +     char rng_txname[RTE_RING_NAMESIZE];
> >       unsigned num_rings = RTE_MIN(RTE_PMD_RING_MAX_RX_RINGS,
> >                       RTE_PMD_RING_MAX_TX_RINGS);
> >
> >       for (i = 0; i < num_rings; i++) {
> > -             snprintf(rng_name, sizeof(rng_name), "ETH_RXTX%u_%s", i,
> name);
> > -             rxtx[i] = (action == DEV_CREATE) ?
> > -                             rte_ring_create(rng_name, 1024, numa_node,
> > -
>  RING_F_SP_ENQ|RING_F_SC_DEQ) :
> > -                             rte_ring_lookup(rng_name);
> > -             if (rxtx[i] == NULL)
> > +             snprintf(rng_rxname, sizeof(rng_rxname),
> > +                         "ETH_RX%u_%s", i, name);
> > +             snprintf(rng_txname, sizeof(rng_txname),
> > +                         "ETH_TX%u_%s", i, name);
> > +             rx[i] = (action == DEV_CREATE) ?
> > +                        rte_ring_create(rng_rxname, 1024, numa_node,
> > +                                        RING_F_SP_ENQ|RING_F_SC_DEQ) :
> > +                        rte_ring_lookup(rng_txname);
> > +             if (rx[i] == NULL)
> > +                     return -1;
> > +             tx[i] = (action == DEV_CREATE) ?
> > +                        rte_ring_create(rng_txname, 1024, numa_node,
> > +                                        RING_F_SP_ENQ|RING_F_SC_DEQ) :
> > +                        rte_ring_lookup(rng_rxname);
> > +             if (tx[i] == NULL)
> >                       return -1;
> >       }
> >
> > -     if (rte_eth_from_rings(name, rxtx, num_rings, rxtx, num_rings,
> numa_node))
> > +     if (rte_eth_from_rings(name, rx, num_rings, tx, num_rings,
> numa_node))
> >               return -1;
> >
> >       return 0;
> > --
> > 1.9.1
> >
>