From: Bruce Richardson Date: Wed, 29 Mar 2017 15:21:25 +0000 (+0100) Subject: ring: return remaining entry count when dequeuing X-Git-Tag: spdx-start~4366 X-Git-Url: http://git.droids-corp.org/?a=commitdiff_plain;h=ecaed092b677;p=dpdk.git ring: return remaining entry count when dequeuing Add an extra parameter to the ring dequeue burst/bulk functions so that those functions can optionally return the amount of remaining objs in the ring. This information can be used by applications in a number of ways, for instance, with single-consumer queues, it provides a max dequeue size which is guaranteed to work. Signed-off-by: Bruce Richardson Reviewed-by: Yuanhan Liu Acked-by: Olivier Matz --- diff --git a/app/pdump/main.c b/app/pdump/main.c index b88090d1c5..3b13753d98 100644 --- a/app/pdump/main.c +++ b/app/pdump/main.c @@ -496,7 +496,7 @@ pdump_rxtx(struct rte_ring *ring, uint8_t vdev_id, struct pdump_stats *stats) /* first dequeue packets from ring of primary process */ const uint16_t nb_in_deq = rte_ring_dequeue_burst(ring, - (void *)rxtx_bufs, BURST_SIZE); + (void *)rxtx_bufs, BURST_SIZE, NULL); stats->dequeue_pkts += nb_in_deq; if (nb_in_deq) { diff --git a/doc/guides/prog_guide/writing_efficient_code.rst b/doc/guides/prog_guide/writing_efficient_code.rst index 78d2afa94c..8223aceeab 100644 --- a/doc/guides/prog_guide/writing_efficient_code.rst +++ b/doc/guides/prog_guide/writing_efficient_code.rst @@ -124,7 +124,7 @@ The code algorithm that dequeues messages may be something similar to the follow while (1) { /* Process as many elements as can be dequeued. */ - count = rte_ring_dequeue_burst(ring, obj_table, MAX_BULK); + count = rte_ring_dequeue_burst(ring, obj_table, MAX_BULK, NULL); if (unlikely(count == 0)) continue; diff --git a/doc/guides/rel_notes/release_17_05.rst b/doc/guides/rel_notes/release_17_05.rst index b361a98717..c67e468a1f 100644 --- a/doc/guides/rel_notes/release_17_05.rst +++ b/doc/guides/rel_notes/release_17_05.rst @@ -140,6 +140,8 @@ API Changes * added an extra parameter to the burst/bulk enqueue functions to return the number of free spaces in the ring after enqueue. This can be used by an application to implement its own watermark functionality. + * added an extra parameter to the burst/bulk dequeue functions to return + the number elements remaining in the ring after dequeue. * changed the return value of the enqueue and dequeue bulk functions to match that of the burst equivalents. In all cases, ring functions which operate on multiple packets now return the number of elements enqueued @@ -152,6 +154,12 @@ API Changes - ``rte_ring_sc_dequeue_bulk`` - ``rte_ring_dequeue_bulk`` + NOTE: the above functions all have different parameters as well as + different return values, due to the other listed changes above. This + means that all instances of the functions in existing code will be + flagged by the compiler. The return value usage should be checked + while fixing the compiler error due to the extra parameter. + ABI Changes ----------- diff --git a/drivers/crypto/aesni_gcm/aesni_gcm_pmd.c b/drivers/crypto/aesni_gcm/aesni_gcm_pmd.c index a2d10a5726..638a95d51f 100644 --- a/drivers/crypto/aesni_gcm/aesni_gcm_pmd.c +++ b/drivers/crypto/aesni_gcm/aesni_gcm_pmd.c @@ -420,7 +420,7 @@ aesni_gcm_pmd_dequeue_burst(void *queue_pair, unsigned nb_dequeued; nb_dequeued = rte_ring_dequeue_burst(qp->processed_pkts, - (void **)ops, nb_ops); + (void **)ops, nb_ops, NULL); qp->qp_stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/aesni_mb/rte_aesni_mb_pmd.c b/drivers/crypto/aesni_mb/rte_aesni_mb_pmd.c index 432d239a75..05edb6c922 100644 --- a/drivers/crypto/aesni_mb/rte_aesni_mb_pmd.c +++ b/drivers/crypto/aesni_mb/rte_aesni_mb_pmd.c @@ -622,7 +622,7 @@ aesni_mb_pmd_dequeue_burst(void *queue_pair, struct rte_crypto_op **ops, unsigned nb_dequeued; nb_dequeued = rte_ring_dequeue_burst(qp->processed_ops, - (void **)ops, nb_ops); + (void **)ops, nb_ops, NULL); qp->stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/armv8/rte_armv8_pmd.c b/drivers/crypto/armv8/rte_armv8_pmd.c index 37ecd7babf..6376e9edd1 100644 --- a/drivers/crypto/armv8/rte_armv8_pmd.c +++ b/drivers/crypto/armv8/rte_armv8_pmd.c @@ -765,7 +765,7 @@ armv8_crypto_pmd_dequeue_burst(void *queue_pair, struct rte_crypto_op **ops, unsigned int nb_dequeued = 0; nb_dequeued = rte_ring_dequeue_burst(qp->processed_ops, - (void **)ops, nb_ops); + (void **)ops, nb_ops, NULL); qp->stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/kasumi/rte_kasumi_pmd.c b/drivers/crypto/kasumi/rte_kasumi_pmd.c index 1dd05cb84a..55bdb29b8f 100644 --- a/drivers/crypto/kasumi/rte_kasumi_pmd.c +++ b/drivers/crypto/kasumi/rte_kasumi_pmd.c @@ -542,7 +542,7 @@ kasumi_pmd_dequeue_burst(void *queue_pair, unsigned nb_dequeued; nb_dequeued = rte_ring_dequeue_burst(qp->processed_ops, - (void **)c_ops, nb_ops); + (void **)c_ops, nb_ops, NULL); qp->qp_stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/null/null_crypto_pmd.c b/drivers/crypto/null/null_crypto_pmd.c index ed5a9fca4b..f68ec8d2b8 100644 --- a/drivers/crypto/null/null_crypto_pmd.c +++ b/drivers/crypto/null/null_crypto_pmd.c @@ -155,7 +155,7 @@ null_crypto_pmd_dequeue_burst(void *queue_pair, struct rte_crypto_op **ops, unsigned nb_dequeued; nb_dequeued = rte_ring_dequeue_burst(qp->processed_pkts, - (void **)ops, nb_ops); + (void **)ops, nb_ops, NULL); qp->qp_stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/openssl/rte_openssl_pmd.c b/drivers/crypto/openssl/rte_openssl_pmd.c index e74c5cf9ad..09173b2120 100644 --- a/drivers/crypto/openssl/rte_openssl_pmd.c +++ b/drivers/crypto/openssl/rte_openssl_pmd.c @@ -1119,7 +1119,7 @@ openssl_pmd_dequeue_burst(void *queue_pair, struct rte_crypto_op **ops, unsigned int nb_dequeued = 0; nb_dequeued = rte_ring_dequeue_burst(qp->processed_ops, - (void **)ops, nb_ops); + (void **)ops, nb_ops, NULL); qp->stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/snow3g/rte_snow3g_pmd.c b/drivers/crypto/snow3g/rte_snow3g_pmd.c index 01c4e1ca4b..1042b311bf 100644 --- a/drivers/crypto/snow3g/rte_snow3g_pmd.c +++ b/drivers/crypto/snow3g/rte_snow3g_pmd.c @@ -533,7 +533,7 @@ snow3g_pmd_dequeue_burst(void *queue_pair, unsigned nb_dequeued; nb_dequeued = rte_ring_dequeue_burst(qp->processed_ops, - (void **)c_ops, nb_ops); + (void **)c_ops, nb_ops, NULL); qp->qp_stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/crypto/zuc/rte_zuc_pmd.c b/drivers/crypto/zuc/rte_zuc_pmd.c index 5e2dbf56c0..06ff503bde 100644 --- a/drivers/crypto/zuc/rte_zuc_pmd.c +++ b/drivers/crypto/zuc/rte_zuc_pmd.c @@ -433,7 +433,7 @@ zuc_pmd_dequeue_burst(void *queue_pair, unsigned nb_dequeued; nb_dequeued = rte_ring_dequeue_burst(qp->processed_ops, - (void **)c_ops, nb_ops); + (void **)c_ops, nb_ops, NULL); qp->qp_stats.dequeued_count += nb_dequeued; return nb_dequeued; diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c index f3ac9e273e..96638af013 100644 --- a/drivers/net/bonding/rte_eth_bond_pmd.c +++ b/drivers/net/bonding/rte_eth_bond_pmd.c @@ -1008,7 +1008,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs, struct port *port = &mode_8023ad_ports[slaves[i]]; slave_slow_nb_pkts[i] = rte_ring_dequeue_burst(port->tx_ring, - slow_pkts, BOND_MODE_8023AX_SLAVE_TX_PKTS); + slow_pkts, BOND_MODE_8023AX_SLAVE_TX_PKTS, + NULL); slave_nb_pkts[i] = slave_slow_nb_pkts[i]; for (j = 0; j < slave_slow_nb_pkts[i]; j++) diff --git a/drivers/net/ring/rte_eth_ring.c b/drivers/net/ring/rte_eth_ring.c index adbf478291..77ef3a19ac 100644 --- a/drivers/net/ring/rte_eth_ring.c +++ b/drivers/net/ring/rte_eth_ring.c @@ -88,7 +88,7 @@ eth_ring_rx(void *q, struct rte_mbuf **bufs, uint16_t nb_bufs) void **ptrs = (void *)&bufs[0]; struct ring_queue *r = q; const uint16_t nb_rx = (uint16_t)rte_ring_dequeue_burst(r->rng, - ptrs, nb_bufs); + ptrs, nb_bufs, NULL); if (r->rng->flags & RING_F_SC_DEQ) r->rx_pkts.cnt += nb_rx; else diff --git a/examples/distributor/main.c b/examples/distributor/main.c index 982b4fd7ea..8071f91955 100644 --- a/examples/distributor/main.c +++ b/examples/distributor/main.c @@ -334,7 +334,7 @@ lcore_distributor(struct lcore_params *p) printf("\nCore %u acting as distributor core.\n", rte_lcore_id()); while (!quit_signal_dist) { const uint16_t nb_rx = rte_ring_dequeue_burst(in_r, - (void *)bufs, BURST_SIZE*1); + (void *)bufs, BURST_SIZE*1, NULL); if (nb_rx) { app_stats.dist.in_pkts += nb_rx; @@ -403,7 +403,7 @@ lcore_tx(struct rte_ring *in_r) struct rte_mbuf *bufs[BURST_SIZE_TX]; const uint16_t nb_rx = rte_ring_dequeue_burst(in_r, - (void *)bufs, BURST_SIZE_TX); + (void *)bufs, BURST_SIZE_TX, NULL); app_stats.tx.dequeue_pkts += nb_rx; /* if we get no traffic, flush anything we have */ diff --git a/examples/load_balancer/runtime.c b/examples/load_balancer/runtime.c index 1645994fcb..8192c08127 100644 --- a/examples/load_balancer/runtime.c +++ b/examples/load_balancer/runtime.c @@ -349,7 +349,8 @@ app_lcore_io_tx( ret = rte_ring_sc_dequeue_bulk( ring, (void **) &lp->tx.mbuf_out[port].array[n_mbufs], - bsz_rd); + bsz_rd, + NULL); if (unlikely(ret == 0)) continue; @@ -504,7 +505,8 @@ app_lcore_worker( ret = rte_ring_sc_dequeue_bulk( ring_in, (void **) lp->mbuf_in.array, - bsz_rd); + bsz_rd, + NULL); if (unlikely(ret == 0)) continue; diff --git a/examples/multi_process/client_server_mp/mp_client/client.c b/examples/multi_process/client_server_mp/mp_client/client.c index dca9eb9f37..01b535c2bf 100644 --- a/examples/multi_process/client_server_mp/mp_client/client.c +++ b/examples/multi_process/client_server_mp/mp_client/client.c @@ -279,7 +279,8 @@ main(int argc, char *argv[]) uint16_t i, rx_pkts; uint8_t port; - rx_pkts = rte_ring_dequeue_burst(rx_ring, pkts, PKT_READ_SIZE); + rx_pkts = rte_ring_dequeue_burst(rx_ring, pkts, + PKT_READ_SIZE, NULL); if (unlikely(rx_pkts == 0)){ if (need_flush) diff --git a/examples/packet_ordering/main.c b/examples/packet_ordering/main.c index 569b6dadb0..49ae35b8ce 100644 --- a/examples/packet_ordering/main.c +++ b/examples/packet_ordering/main.c @@ -462,7 +462,7 @@ worker_thread(void *args_ptr) /* dequeue the mbufs from rx_to_workers ring */ burst_size = rte_ring_dequeue_burst(ring_in, - (void *)burst_buffer, MAX_PKTS_BURST); + (void *)burst_buffer, MAX_PKTS_BURST, NULL); if (unlikely(burst_size == 0)) continue; @@ -510,7 +510,7 @@ send_thread(struct send_thread_args *args) /* deque the mbufs from workers_to_tx ring */ nb_dq_mbufs = rte_ring_dequeue_burst(args->ring_in, - (void *)mbufs, MAX_PKTS_BURST); + (void *)mbufs, MAX_PKTS_BURST, NULL); if (unlikely(nb_dq_mbufs == 0)) continue; @@ -595,7 +595,7 @@ tx_thread(struct rte_ring *ring_in) /* deque the mbufs from workers_to_tx ring */ dqnum = rte_ring_dequeue_burst(ring_in, - (void *)mbufs, MAX_PKTS_BURST); + (void *)mbufs, MAX_PKTS_BURST, NULL); if (unlikely(dqnum == 0)) continue; diff --git a/examples/performance-thread/l3fwd-thread/main.c b/examples/performance-thread/l3fwd-thread/main.c index b4c0df1881..06f8962240 100644 --- a/examples/performance-thread/l3fwd-thread/main.c +++ b/examples/performance-thread/l3fwd-thread/main.c @@ -2077,7 +2077,7 @@ lthread_tx_per_ring(void *dummy) */ SET_CPU_BUSY(tx_conf, CPU_POLL); nb_rx = rte_ring_sc_dequeue_burst(ring, (void **)pkts_burst, - MAX_PKT_BURST); + MAX_PKT_BURST, NULL); SET_CPU_IDLE(tx_conf, CPU_POLL); if (nb_rx > 0) { @@ -2381,7 +2381,7 @@ pthread_tx(void *dummy) */ SET_CPU_BUSY(tx_conf, CPU_POLL); nb_rx = rte_ring_sc_dequeue_burst(tx_conf->ring, - (void **)pkts_burst, MAX_PKT_BURST); + (void **)pkts_burst, MAX_PKT_BURST, NULL); SET_CPU_IDLE(tx_conf, CPU_POLL); if (unlikely(nb_rx == 0)) { diff --git a/examples/qos_sched/app_thread.c b/examples/qos_sched/app_thread.c index 0c81a152c9..15f117f5f4 100644 --- a/examples/qos_sched/app_thread.c +++ b/examples/qos_sched/app_thread.c @@ -179,7 +179,7 @@ app_tx_thread(struct thread_conf **confs) while ((conf = confs[conf_idx])) { retval = rte_ring_sc_dequeue_bulk(conf->tx_ring, (void **)mbufs, - burst_conf.qos_dequeue); + burst_conf.qos_dequeue, NULL); if (likely(retval != 0)) { app_send_packets(conf, mbufs, burst_conf.qos_dequeue); @@ -218,7 +218,7 @@ app_worker_thread(struct thread_conf **confs) /* Read packet from the ring */ nb_pkt = rte_ring_sc_dequeue_burst(conf->rx_ring, (void **)mbufs, - burst_conf.ring_burst); + burst_conf.ring_burst, NULL); if (likely(nb_pkt)) { int nb_sent = rte_sched_port_enqueue(conf->sched_port, mbufs, nb_pkt); @@ -254,7 +254,7 @@ app_mixed_thread(struct thread_conf **confs) /* Read packet from the ring */ nb_pkt = rte_ring_sc_dequeue_burst(conf->rx_ring, (void **)mbufs, - burst_conf.ring_burst); + burst_conf.ring_burst, NULL); if (likely(nb_pkt)) { int nb_sent = rte_sched_port_enqueue(conf->sched_port, mbufs, nb_pkt); diff --git a/examples/quota_watermark/qw/main.c b/examples/quota_watermark/qw/main.c index 57df8efdb6..2dcddea6c2 100644 --- a/examples/quota_watermark/qw/main.c +++ b/examples/quota_watermark/qw/main.c @@ -247,7 +247,8 @@ pipeline_stage(__attribute__((unused)) void *args) } /* Dequeue up to quota mbuf from rx */ - nb_dq_pkts = rte_ring_dequeue_burst(rx, pkts, *quota); + nb_dq_pkts = rte_ring_dequeue_burst(rx, pkts, + *quota, NULL); if (unlikely(nb_dq_pkts < 0)) continue; @@ -305,7 +306,7 @@ send_stage(__attribute__((unused)) void *args) /* Dequeue packets from tx and send them */ nb_dq_pkts = (uint16_t) rte_ring_dequeue_burst(tx, - (void *) tx_pkts, *quota); + (void *) tx_pkts, *quota, NULL); rte_eth_tx_burst(dest_port_id, 0, tx_pkts, nb_dq_pkts); /* TODO: Check if nb_dq_pkts == nb_tx_pkts? */ diff --git a/examples/server_node_efd/node/node.c b/examples/server_node_efd/node/node.c index 9ec6a058fe..f780b92642 100644 --- a/examples/server_node_efd/node/node.c +++ b/examples/server_node_efd/node/node.c @@ -392,7 +392,7 @@ main(int argc, char *argv[]) */ while (rx_pkts > 0 && unlikely(rte_ring_dequeue_bulk(rx_ring, pkts, - rx_pkts) == 0)) + rx_pkts, NULL) == 0)) rx_pkts = (uint16_t)RTE_MIN(rte_ring_count(rx_ring), PKT_READ_SIZE); diff --git a/lib/librte_hash/rte_cuckoo_hash.c b/lib/librte_hash/rte_cuckoo_hash.c index 6552199f77..645c0cfab5 100644 --- a/lib/librte_hash/rte_cuckoo_hash.c +++ b/lib/librte_hash/rte_cuckoo_hash.c @@ -536,7 +536,8 @@ __rte_hash_add_key_with_hash(const struct rte_hash *h, const void *key, if (cached_free_slots->len == 0) { /* Need to get another burst of free slots from global ring */ n_slots = rte_ring_mc_dequeue_burst(h->free_slots, - cached_free_slots->objs, LCORE_CACHE_SIZE); + cached_free_slots->objs, + LCORE_CACHE_SIZE, NULL); if (n_slots == 0) return -ENOSPC; diff --git a/lib/librte_mempool/rte_mempool_ring.c b/lib/librte_mempool/rte_mempool_ring.c index 9b8fd2bd4f..5c132bf69e 100644 --- a/lib/librte_mempool/rte_mempool_ring.c +++ b/lib/librte_mempool/rte_mempool_ring.c @@ -58,14 +58,14 @@ static int common_ring_mc_dequeue(struct rte_mempool *mp, void **obj_table, unsigned n) { return rte_ring_mc_dequeue_bulk(mp->pool_data, - obj_table, n) == 0 ? -ENOBUFS : 0; + obj_table, n, NULL) == 0 ? -ENOBUFS : 0; } static int common_ring_sc_dequeue(struct rte_mempool *mp, void **obj_table, unsigned n) { return rte_ring_sc_dequeue_bulk(mp->pool_data, - obj_table, n) == 0 ? -ENOBUFS : 0; + obj_table, n, NULL) == 0 ? -ENOBUFS : 0; } static unsigned diff --git a/lib/librte_port/rte_port_frag.c b/lib/librte_port/rte_port_frag.c index 0fcace99b6..320407ecb7 100644 --- a/lib/librte_port/rte_port_frag.c +++ b/lib/librte_port/rte_port_frag.c @@ -186,7 +186,8 @@ rte_port_ring_reader_frag_rx(void *port, /* If "pkts" buffer is empty, read packet burst from ring */ if (p->n_pkts == 0) { p->n_pkts = rte_ring_sc_dequeue_burst(p->ring, - (void **) p->pkts, RTE_PORT_IN_BURST_SIZE_MAX); + (void **) p->pkts, RTE_PORT_IN_BURST_SIZE_MAX, + NULL); RTE_PORT_RING_READER_FRAG_STATS_PKTS_IN_ADD(p, p->n_pkts); if (p->n_pkts == 0) return n_pkts_out; diff --git a/lib/librte_port/rte_port_ring.c b/lib/librte_port/rte_port_ring.c index c5dbe07e7a..85fad44568 100644 --- a/lib/librte_port/rte_port_ring.c +++ b/lib/librte_port/rte_port_ring.c @@ -111,7 +111,8 @@ rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts) struct rte_port_ring_reader *p = (struct rte_port_ring_reader *) port; uint32_t nb_rx; - nb_rx = rte_ring_sc_dequeue_burst(p->ring, (void **) pkts, n_pkts); + nb_rx = rte_ring_sc_dequeue_burst(p->ring, (void **) pkts, + n_pkts, NULL); RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx); return nb_rx; @@ -124,7 +125,8 @@ rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts, struct rte_port_ring_reader *p = (struct rte_port_ring_reader *) port; uint32_t nb_rx; - nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts, n_pkts); + nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts, + n_pkts, NULL); RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx); return nb_rx; diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h index 439698be66..2b6896bf11 100644 --- a/lib/librte_ring/rte_ring.h +++ b/lib/librte_ring/rte_ring.h @@ -488,7 +488,8 @@ end: static inline unsigned int __attribute__((always_inline)) __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, - unsigned n, enum rte_ring_queue_behavior behavior) + unsigned int n, enum rte_ring_queue_behavior behavior, + unsigned int *available) { uint32_t cons_head, prod_tail; uint32_t cons_next, entries; @@ -497,11 +498,6 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, unsigned int i; uint32_t mask = r->mask; - /* Avoid the unnecessary cmpset operation below, which is also - * potentially harmful when n equals 0. */ - if (n == 0) - return 0; - /* move cons.head atomically */ do { /* Restore n as it may change every loop */ @@ -516,15 +512,11 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, entries = (prod_tail - cons_head); /* Set the actual entries for dequeue */ - if (n > entries) { - if (behavior == RTE_RING_QUEUE_FIXED) - return 0; - else { - if (unlikely(entries == 0)) - return 0; - n = entries; - } - } + if (n > entries) + n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : entries; + + if (unlikely(n == 0)) + goto end; cons_next = cons_head + n; success = rte_atomic32_cmpset(&r->cons.head, cons_head, @@ -543,7 +535,9 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, rte_pause(); r->cons.tail = cons_next; - +end: + if (available != NULL) + *available = entries - n; return n; } @@ -567,7 +561,8 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table, */ static inline unsigned int __attribute__((always_inline)) __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table, - unsigned n, enum rte_ring_queue_behavior behavior) + unsigned int n, enum rte_ring_queue_behavior behavior, + unsigned int *available) { uint32_t cons_head, prod_tail; uint32_t cons_next, entries; @@ -582,15 +577,11 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table, * and size(ring)-1. */ entries = prod_tail - cons_head; - if (n > entries) { - if (behavior == RTE_RING_QUEUE_FIXED) - return 0; - else { - if (unlikely(entries == 0)) - return 0; - n = entries; - } - } + if (n > entries) + n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : entries; + + if (unlikely(entries == 0)) + goto end; cons_next = cons_head + n; r->cons.head = cons_next; @@ -600,6 +591,9 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table, rte_smp_rmb(); r->cons.tail = cons_next; +end: + if (available != NULL) + *available = entries - n; return n; } @@ -751,13 +745,18 @@ rte_ring_enqueue(struct rte_ring *r, void *obj) * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * @param available + * If non-NULL, returns the number of remaining ring entries after the + * dequeue has finished. * @return * The number of objects dequeued, either 0 or n */ static inline unsigned int __attribute__((always_inline)) -rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n) +rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, + unsigned int n, unsigned int *available) { - return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED); + return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED, + available); } /** @@ -770,13 +769,18 @@ rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n) * @param n * The number of objects to dequeue from the ring to the obj_table, * must be strictly positive. + * @param available + * If non-NULL, returns the number of remaining ring entries after the + * dequeue has finished. * @return * The number of objects dequeued, either 0 or n */ static inline unsigned int __attribute__((always_inline)) -rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n) +rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, + unsigned int n, unsigned int *available) { - return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED); + return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED, + available); } /** @@ -792,16 +796,20 @@ rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n) * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * @param available + * If non-NULL, returns the number of remaining ring entries after the + * dequeue has finished. * @return * The number of objects dequeued, either 0 or n */ static inline unsigned int __attribute__((always_inline)) -rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n) +rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned int n, + unsigned int *available) { if (r->cons.single) - return rte_ring_sc_dequeue_bulk(r, obj_table, n); + return rte_ring_sc_dequeue_bulk(r, obj_table, n, available); else - return rte_ring_mc_dequeue_bulk(r, obj_table, n); + return rte_ring_mc_dequeue_bulk(r, obj_table, n, available); } /** @@ -822,7 +830,7 @@ rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n) static inline int __attribute__((always_inline)) rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p) { - return rte_ring_mc_dequeue_bulk(r, obj_p, 1) ? 0 : -ENOBUFS; + return rte_ring_mc_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS; } /** @@ -840,7 +848,7 @@ rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p) static inline int __attribute__((always_inline)) rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p) { - return rte_ring_sc_dequeue_bulk(r, obj_p, 1) ? 0 : -ENOBUFS; + return rte_ring_sc_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS; } /** @@ -862,7 +870,7 @@ rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p) static inline int __attribute__((always_inline)) rte_ring_dequeue(struct rte_ring *r, void **obj_p) { - return rte_ring_dequeue_bulk(r, obj_p, 1) ? 0 : -ENOBUFS; + return rte_ring_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOBUFS; } /** @@ -1057,13 +1065,18 @@ rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table, * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * @param available + * If non-NULL, returns the number of remaining ring entries after the + * dequeue has finished. * @return * - n: Actual number of objects dequeued, 0 if ring is empty */ static inline unsigned __attribute__((always_inline)) -rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n) +rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, + unsigned int n, unsigned int *available) { - return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE); + return __rte_ring_mc_do_dequeue(r, obj_table, n, + RTE_RING_QUEUE_VARIABLE, available); } /** @@ -1077,13 +1090,18 @@ rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n) * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * @param available + * If non-NULL, returns the number of remaining ring entries after the + * dequeue has finished. * @return * - n: Actual number of objects dequeued, 0 if ring is empty */ static inline unsigned __attribute__((always_inline)) -rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n) +rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, + unsigned int n, unsigned int *available) { - return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE); + return __rte_ring_sc_do_dequeue(r, obj_table, n, + RTE_RING_QUEUE_VARIABLE, available); } /** @@ -1099,16 +1117,20 @@ rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n) * A pointer to a table of void * pointers (objects) that will be filled. * @param n * The number of objects to dequeue from the ring to the obj_table. + * @param available + * If non-NULL, returns the number of remaining ring entries after the + * dequeue has finished. * @return * - Number of objects dequeued */ static inline unsigned __attribute__((always_inline)) -rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n) +rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, + unsigned int n, unsigned int *available) { if (r->cons.single) - return rte_ring_sc_dequeue_burst(r, obj_table, n); + return rte_ring_sc_dequeue_burst(r, obj_table, n, available); else - return rte_ring_mc_dequeue_burst(r, obj_table, n); + return rte_ring_mc_dequeue_burst(r, obj_table, n, available); } #ifdef __cplusplus diff --git a/test/test-pipeline/runtime.c b/test/test-pipeline/runtime.c index c06ff54cd3..8970e1c30a 100644 --- a/test/test-pipeline/runtime.c +++ b/test/test-pipeline/runtime.c @@ -121,7 +121,8 @@ app_main_loop_worker(void) { ret = rte_ring_sc_dequeue_bulk( app.rings_rx[i], (void **) worker_mbuf->array, - app.burst_size_worker_read); + app.burst_size_worker_read, + NULL); if (ret == 0) continue; @@ -151,7 +152,8 @@ app_main_loop_tx(void) { ret = rte_ring_sc_dequeue_bulk( app.rings_tx[i], (void **) &app.mbuf_tx[i].array[n_mbufs], - app.burst_size_tx_read); + app.burst_size_tx_read, + NULL); if (ret == 0) continue; diff --git a/test/test/test_link_bonding_mode4.c b/test/test/test_link_bonding_mode4.c index 8df28b4f15..15091b13db 100644 --- a/test/test/test_link_bonding_mode4.c +++ b/test/test/test_link_bonding_mode4.c @@ -193,7 +193,8 @@ static uint8_t lacpdu_rx_count[RTE_MAX_ETHPORTS] = {0, }; static int slave_get_pkts(struct slave_conf *slave, struct rte_mbuf **buf, uint16_t size) { - return rte_ring_dequeue_burst(slave->tx_queue, (void **)buf, size); + return rte_ring_dequeue_burst(slave->tx_queue, (void **)buf, + size, NULL); } /* diff --git a/test/test/test_pmd_ring_perf.c b/test/test/test_pmd_ring_perf.c index 045a7f255a..004882af17 100644 --- a/test/test/test_pmd_ring_perf.c +++ b/test/test/test_pmd_ring_perf.c @@ -67,7 +67,7 @@ test_empty_dequeue(void) const uint64_t sc_start = rte_rdtsc(); for (i = 0; i < iterations; i++) - rte_ring_sc_dequeue_bulk(r, burst, bulk_sizes[0]); + rte_ring_sc_dequeue_bulk(r, burst, bulk_sizes[0], NULL); const uint64_t sc_end = rte_rdtsc(); const uint64_t eth_start = rte_rdtsc(); @@ -99,7 +99,7 @@ test_single_enqueue_dequeue(void) rte_compiler_barrier(); for (i = 0; i < iterations; i++) { rte_ring_enqueue_bulk(r, &burst, 1, NULL); - rte_ring_dequeue_bulk(r, &burst, 1); + rte_ring_dequeue_bulk(r, &burst, 1, NULL); } const uint64_t sc_end = rte_rdtsc_precise(); rte_compiler_barrier(); @@ -133,7 +133,8 @@ test_bulk_enqueue_dequeue(void) for (i = 0; i < iterations; i++) { rte_ring_sp_enqueue_bulk(r, (void *)burst, bulk_sizes[sz], NULL); - rte_ring_sc_dequeue_bulk(r, (void *)burst, bulk_sizes[sz]); + rte_ring_sc_dequeue_bulk(r, (void *)burst, + bulk_sizes[sz], NULL); } const uint64_t sc_end = rte_rdtsc(); diff --git a/test/test/test_ring.c b/test/test/test_ring.c index b0ca88b65b..858ebc1da3 100644 --- a/test/test/test_ring.c +++ b/test/test/test_ring.c @@ -119,7 +119,8 @@ test_ring_basic_full_empty(void * const src[], void *dst[]) __func__, i, rand); TEST_RING_VERIFY(rte_ring_enqueue_bulk(r, src, rand, NULL) != 0); - TEST_RING_VERIFY(rte_ring_dequeue_bulk(r, dst, rand) == rand); + TEST_RING_VERIFY(rte_ring_dequeue_bulk(r, dst, rand, + NULL) == rand); /* fill the ring */ TEST_RING_VERIFY(rte_ring_enqueue_bulk(r, src, rsz, NULL) != 0); @@ -129,7 +130,8 @@ test_ring_basic_full_empty(void * const src[], void *dst[]) TEST_RING_VERIFY(0 == rte_ring_empty(r)); /* empty the ring */ - TEST_RING_VERIFY(rte_ring_dequeue_bulk(r, dst, rsz) == rsz); + TEST_RING_VERIFY(rte_ring_dequeue_bulk(r, dst, rsz, + NULL) == rsz); TEST_RING_VERIFY(rsz == rte_ring_free_count(r)); TEST_RING_VERIFY(0 == rte_ring_count(r)); TEST_RING_VERIFY(0 == rte_ring_full(r)); @@ -186,19 +188,19 @@ test_ring_basic(void) goto fail; printf("dequeue 1 obj\n"); - ret = rte_ring_sc_dequeue_bulk(r, cur_dst, 1); + ret = rte_ring_sc_dequeue_bulk(r, cur_dst, 1, NULL); cur_dst += 1; if (ret == 0) goto fail; printf("dequeue 2 objs\n"); - ret = rte_ring_sc_dequeue_bulk(r, cur_dst, 2); + ret = rte_ring_sc_dequeue_bulk(r, cur_dst, 2, NULL); cur_dst += 2; if (ret == 0) goto fail; printf("dequeue MAX_BULK objs\n"); - ret = rte_ring_sc_dequeue_bulk(r, cur_dst, MAX_BULK); + ret = rte_ring_sc_dequeue_bulk(r, cur_dst, MAX_BULK, NULL); cur_dst += MAX_BULK; if (ret == 0) goto fail; @@ -232,19 +234,19 @@ test_ring_basic(void) goto fail; printf("dequeue 1 obj\n"); - ret = rte_ring_mc_dequeue_bulk(r, cur_dst, 1); + ret = rte_ring_mc_dequeue_bulk(r, cur_dst, 1, NULL); cur_dst += 1; if (ret == 0) goto fail; printf("dequeue 2 objs\n"); - ret = rte_ring_mc_dequeue_bulk(r, cur_dst, 2); + ret = rte_ring_mc_dequeue_bulk(r, cur_dst, 2, NULL); cur_dst += 2; if (ret == 0) goto fail; printf("dequeue MAX_BULK objs\n"); - ret = rte_ring_mc_dequeue_bulk(r, cur_dst, MAX_BULK); + ret = rte_ring_mc_dequeue_bulk(r, cur_dst, MAX_BULK, NULL); cur_dst += MAX_BULK; if (ret == 0) goto fail; @@ -265,7 +267,7 @@ test_ring_basic(void) cur_src += MAX_BULK; if (ret == 0) goto fail; - ret = rte_ring_mc_dequeue_bulk(r, cur_dst, MAX_BULK); + ret = rte_ring_mc_dequeue_bulk(r, cur_dst, MAX_BULK, NULL); cur_dst += MAX_BULK; if (ret == 0) goto fail; @@ -303,13 +305,13 @@ test_ring_basic(void) printf("Cannot enqueue\n"); goto fail; } - ret = rte_ring_dequeue_bulk(r, cur_dst, num_elems); + ret = rte_ring_dequeue_bulk(r, cur_dst, num_elems, NULL); cur_dst += num_elems; if (ret == 0) { printf("Cannot dequeue\n"); goto fail; } - ret = rte_ring_dequeue_bulk(r, cur_dst, num_elems); + ret = rte_ring_dequeue_bulk(r, cur_dst, num_elems, NULL); cur_dst += num_elems; if (ret == 0) { printf("Cannot dequeue2\n"); @@ -390,19 +392,19 @@ test_ring_burst_basic(void) goto fail; printf("dequeue 1 obj\n"); - ret = rte_ring_sc_dequeue_burst(r, cur_dst, 1) ; + ret = rte_ring_sc_dequeue_burst(r, cur_dst, 1, NULL); cur_dst += 1; if ((ret & RTE_RING_SZ_MASK) != 1) goto fail; printf("dequeue 2 objs\n"); - ret = rte_ring_sc_dequeue_burst(r, cur_dst, 2); + ret = rte_ring_sc_dequeue_burst(r, cur_dst, 2, NULL); cur_dst += 2; if ((ret & RTE_RING_SZ_MASK) != 2) goto fail; printf("dequeue MAX_BULK objs\n"); - ret = rte_ring_sc_dequeue_burst(r, cur_dst, MAX_BULK); + ret = rte_ring_sc_dequeue_burst(r, cur_dst, MAX_BULK, NULL); cur_dst += MAX_BULK; if ((ret & RTE_RING_SZ_MASK) != MAX_BULK) goto fail; @@ -451,19 +453,19 @@ test_ring_burst_basic(void) printf("Test dequeue without enough objects \n"); for (i = 0; idata->dev_private; rx_count = rte_ring_dequeue_burst(dev_private->rx_queue, (void **) bufs, - nb_pkts); + nb_pkts, NULL); /* increments ipackets count */ dev_private->eth_stats.ipackets += rx_count; @@ -508,7 +508,7 @@ virtual_ethdev_get_mbufs_from_tx_queue(uint8_t port_id, dev_private = vrtl_eth_dev->data->dev_private; return rte_ring_dequeue_burst(dev_private->tx_queue, (void **)pkt_burst, - burst_length); + burst_length, NULL); } static uint8_t