rte_mempool.h

Go to the documentation of this file.

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright(c) 2010-2014 Intel Corporation.
 * Copyright(c) 2016 6WIND S.A.
 * Copyright(c) 2022 SmartShare Systems
 */

#ifndef _RTE_MEMPOOL_H_
#define _RTE_MEMPOOL_H_

#include <stdalign.h>
#include <stdio.h>
#include <stdint.h>
#include <inttypes.h>

#include <rte_compat.h>
#include <rte_config.h>
#include <rte_spinlock.h>
#include <rte_debug.h>
#include <rte_lcore.h>
#include <rte_log.h>
#include <rte_branch_prediction.h>
#include <rte_ring.h>
#include <rte_memcpy.h>
#include <rte_common.h>

#include "rte_mempool_trace_fp.h"

#ifdef __cplusplus
extern "C" {
#endif

#define RTE_MEMPOOL_HEADER_COOKIE1  0xbadbadbadadd2e55ULL 
#define RTE_MEMPOOL_HEADER_COOKIE2  0xf2eef2eedadd2e55ULL 
#define RTE_MEMPOOL_TRAILER_COOKIE  0xadd2e55badbadbadULL 
#ifdef RTE_LIBRTE_MEMPOOL_STATS

struct __rte_cache_aligned rte_mempool_debug_stats {
    uint64_t put_bulk;             
    uint64_t put_objs;             
    uint64_t put_common_pool_bulk; 
    uint64_t put_common_pool_objs; 
    uint64_t get_common_pool_bulk; 
    uint64_t get_common_pool_objs; 
    uint64_t get_success_bulk;     
    uint64_t get_success_objs;     
    uint64_t get_fail_bulk;        
    uint64_t get_fail_objs;        
    uint64_t get_success_blks;     
    uint64_t get_fail_blks;        
    RTE_CACHE_GUARD;
};
#endif

struct __rte_cache_aligned rte_mempool_cache {
    uint32_t size;        
    uint32_t flushthresh; 
    uint32_t len;         
#ifdef RTE_LIBRTE_MEMPOOL_STATS
    uint32_t unused;
    /*
     * Alternative location for the most frequently updated mempool statistics (per-lcore),
     * providing faster update access when using a mempool cache.
     */
    struct {
        uint64_t put_bulk;          
        uint64_t put_objs;          
        uint64_t get_success_bulk;  
        uint64_t get_success_objs;  
    } stats;                        
#endif

    alignas(RTE_CACHE_LINE_SIZE) void *objs[RTE_MEMPOOL_CACHE_MAX_SIZE * 2];
};

struct rte_mempool_objsz {
    uint32_t elt_size;     
    uint32_t header_size;  
    uint32_t trailer_size; 
    uint32_t total_size;
};

#define RTE_MEMPOOL_NAMESIZE (RTE_RING_NAMESIZE - \
                  sizeof(RTE_MEMPOOL_MZ_PREFIX) + 1)
#define RTE_MEMPOOL_MZ_PREFIX "MP_"

/* "MP_<name>" */
#define RTE_MEMPOOL_MZ_FORMAT   RTE_MEMPOOL_MZ_PREFIX "%s"

#ifndef RTE_MEMPOOL_ALIGN

#define RTE_MEMPOOL_ALIGN   RTE_CACHE_LINE_SIZE
#endif

#define RTE_MEMPOOL_ALIGN_MASK  (RTE_MEMPOOL_ALIGN - 1)

struct rte_mempool_objhdr {
    RTE_STAILQ_ENTRY(rte_mempool_objhdr) next; 
    struct rte_mempool *mp;          
    rte_iova_t iova;                 
#ifdef RTE_LIBRTE_MEMPOOL_DEBUG
    uint64_t cookie;                 
#endif
};

RTE_STAILQ_HEAD(rte_mempool_objhdr_list, rte_mempool_objhdr);

#ifdef RTE_LIBRTE_MEMPOOL_DEBUG

struct rte_mempool_objtlr {
    uint64_t cookie;                 
};

#endif

extern int rte_mempool_logtype;
#define RTE_LOGTYPE_MEMPOOL rte_mempool_logtype
#define RTE_MEMPOOL_LOG(level, ...) \
    RTE_LOG_LINE(level, MEMPOOL, "" __VA_ARGS__)

RTE_STAILQ_HEAD(rte_mempool_memhdr_list, rte_mempool_memhdr);

typedef void (rte_mempool_memchunk_free_cb_t)(struct rte_mempool_memhdr *memhdr,
    void *opaque);

struct rte_mempool_memhdr {
    RTE_STAILQ_ENTRY(rte_mempool_memhdr) next; 
    struct rte_mempool *mp;  
    void *addr;              
    rte_iova_t iova;         
    size_t len;              
    rte_mempool_memchunk_free_cb_t *free_cb; 
    void *opaque;            
};

struct __rte_cache_aligned rte_mempool_info {
    unsigned int contig_block_size;
};

struct __rte_cache_aligned rte_mempool {
    char name[RTE_MEMPOOL_NAMESIZE]; 
    union {
        void *pool_data;         
        uint64_t pool_id;        
    };
    void *pool_config;               
    const struct rte_memzone *mz;    
    unsigned int flags;              
    int socket_id;                   
    uint32_t size;                   
    uint32_t cache_size;
    uint32_t elt_size;               
    uint32_t header_size;            
    uint32_t trailer_size;           
    unsigned private_data_size;      
    int32_t ops_index;

    struct rte_mempool_cache *local_cache; 
    uint32_t populated_size;         
    struct rte_mempool_objhdr_list elt_list; 
    uint32_t nb_mem_chunks;          
    struct rte_mempool_memhdr_list mem_list; 
#ifdef RTE_LIBRTE_MEMPOOL_STATS

    struct rte_mempool_debug_stats stats[RTE_MAX_LCORE + 1];
#endif
};

#define RTE_MEMPOOL_F_NO_SPREAD     0x0001

#define MEMPOOL_F_NO_SPREAD     RTE_MEMPOOL_F_NO_SPREAD

#define RTE_MEMPOOL_F_NO_CACHE_ALIGN    0x0002

#define MEMPOOL_F_NO_CACHE_ALIGN    RTE_MEMPOOL_F_NO_CACHE_ALIGN

#define RTE_MEMPOOL_F_SP_PUT        0x0004

#define MEMPOOL_F_SP_PUT        RTE_MEMPOOL_F_SP_PUT

#define RTE_MEMPOOL_F_SC_GET        0x0008

#define MEMPOOL_F_SC_GET        RTE_MEMPOOL_F_SC_GET

#define RTE_MEMPOOL_F_POOL_CREATED  0x0010

#define RTE_MEMPOOL_F_NO_IOVA_CONTIG    0x0020

#define MEMPOOL_F_NO_IOVA_CONTIG    RTE_MEMPOOL_F_NO_IOVA_CONTIG

#define RTE_MEMPOOL_F_NON_IO        0x0040

#define RTE_MEMPOOL_VALID_USER_FLAGS (RTE_MEMPOOL_F_NO_SPREAD \
    | RTE_MEMPOOL_F_NO_CACHE_ALIGN \
    | RTE_MEMPOOL_F_SP_PUT \
    | RTE_MEMPOOL_F_SC_GET \
    | RTE_MEMPOOL_F_NO_IOVA_CONTIG \
    )

#ifdef RTE_LIBRTE_MEMPOOL_STATS
#define RTE_MEMPOOL_STAT_ADD(mp, name, n) do {                                  \
        unsigned int __lcore_id = rte_lcore_id();                       \
        if (likely(__lcore_id < RTE_MAX_LCORE))                         \
            (mp)->stats[__lcore_id].name += (n);                    \
        else                                                            \
            rte_atomic_fetch_add_explicit(&((mp)->stats[RTE_MAX_LCORE].name),  \
                       (n), rte_memory_order_relaxed);              \
    } while (0)
#else
#define RTE_MEMPOOL_STAT_ADD(mp, name, n) do {} while (0)
#endif

#ifdef RTE_LIBRTE_MEMPOOL_STATS
#define RTE_MEMPOOL_CACHE_STAT_ADD(cache, name, n) ((cache)->stats.name += (n))
#else
#define RTE_MEMPOOL_CACHE_STAT_ADD(cache, name, n) do {} while (0)
#endif

#define RTE_MEMPOOL_HEADER_SIZE(mp, cs) \
    (sizeof(*(mp)) + (((cs) == 0) ? 0 : \
    (sizeof(struct rte_mempool_cache) * RTE_MAX_LCORE)))

/* return the header of a mempool object (internal) */
static inline struct rte_mempool_objhdr *
rte_mempool_get_header(void *obj)
{
    return (struct rte_mempool_objhdr *)RTE_PTR_SUB(obj,
        sizeof(struct rte_mempool_objhdr));
}

static inline struct rte_mempool *rte_mempool_from_obj(void *obj)
{
    struct rte_mempool_objhdr *hdr = rte_mempool_get_header(obj);
    return hdr->mp;
}

/* return the trailer of a mempool object (internal) */
static inline struct rte_mempool_objtlr *rte_mempool_get_trailer(void *obj)
{
    struct rte_mempool *mp = rte_mempool_from_obj(obj);
    return (struct rte_mempool_objtlr *)RTE_PTR_ADD(obj, mp->elt_size);
}

void rte_mempool_check_cookies(const struct rte_mempool *mp,
    void * const *obj_table_const, unsigned n, int free);

#ifdef RTE_LIBRTE_MEMPOOL_DEBUG
#define RTE_MEMPOOL_CHECK_COOKIES(mp, obj_table_const, n, free) \
    rte_mempool_check_cookies(mp, obj_table_const, n, free)
#else
#define RTE_MEMPOOL_CHECK_COOKIES(mp, obj_table_const, n, free) do {} while (0)
#endif /* RTE_LIBRTE_MEMPOOL_DEBUG */

void rte_mempool_contig_blocks_check_cookies(const struct rte_mempool *mp,
    void * const *first_obj_table_const, unsigned int n, int free);

#ifdef RTE_LIBRTE_MEMPOOL_DEBUG
#define RTE_MEMPOOL_CONTIG_BLOCKS_CHECK_COOKIES(mp, first_obj_table_const, n, \
                        free) \
    rte_mempool_contig_blocks_check_cookies(mp, first_obj_table_const, n, \
                        free)
#else
#define RTE_MEMPOOL_CONTIG_BLOCKS_CHECK_COOKIES(mp, first_obj_table_const, n, \
                        free) \
    do {} while (0)
#endif /* RTE_LIBRTE_MEMPOOL_DEBUG */

#define RTE_MEMPOOL_OPS_NAMESIZE 32 
typedef int (*rte_mempool_alloc_t)(struct rte_mempool *mp);

typedef void (*rte_mempool_free_t)(struct rte_mempool *mp);

typedef int (*rte_mempool_enqueue_t)(struct rte_mempool *mp,
        void * const *obj_table, unsigned int n);

typedef int (*rte_mempool_dequeue_t)(struct rte_mempool *mp,
        void **obj_table, unsigned int n);

typedef int (*rte_mempool_dequeue_contig_blocks_t)(struct rte_mempool *mp,
         void **first_obj_table, unsigned int n);

typedef unsigned (*rte_mempool_get_count)(const struct rte_mempool *mp);

typedef ssize_t (*rte_mempool_calc_mem_size_t)(const struct rte_mempool *mp,
        uint32_t obj_num,  uint32_t pg_shift,
        size_t *min_chunk_size, size_t *align);

ssize_t rte_mempool_op_calc_mem_size_helper(const struct rte_mempool *mp,
        uint32_t obj_num, uint32_t pg_shift, size_t chunk_reserve,
        size_t *min_chunk_size, size_t *align);

ssize_t rte_mempool_op_calc_mem_size_default(const struct rte_mempool *mp,
        uint32_t obj_num, uint32_t pg_shift,
        size_t *min_chunk_size, size_t *align);

typedef void (rte_mempool_populate_obj_cb_t)(struct rte_mempool *mp,
        void *opaque, void *vaddr, rte_iova_t iova);

typedef int (*rte_mempool_populate_t)(struct rte_mempool *mp,
        unsigned int max_objs,
        void *vaddr, rte_iova_t iova, size_t len,
        rte_mempool_populate_obj_cb_t *obj_cb, void *obj_cb_arg);

#define RTE_MEMPOOL_POPULATE_F_ALIGN_OBJ 0x0001

int rte_mempool_op_populate_helper(struct rte_mempool *mp,
        unsigned int flags, unsigned int max_objs,
        void *vaddr, rte_iova_t iova, size_t len,
        rte_mempool_populate_obj_cb_t *obj_cb, void *obj_cb_arg);

int rte_mempool_op_populate_default(struct rte_mempool *mp,
        unsigned int max_objs,
        void *vaddr, rte_iova_t iova, size_t len,
        rte_mempool_populate_obj_cb_t *obj_cb, void *obj_cb_arg);

typedef int (*rte_mempool_get_info_t)(const struct rte_mempool *mp,
        struct rte_mempool_info *info);


struct __rte_cache_aligned rte_mempool_ops {
    char name[RTE_MEMPOOL_OPS_NAMESIZE]; 
    rte_mempool_alloc_t alloc;       
    rte_mempool_free_t free;         
    rte_mempool_enqueue_t enqueue;   
    rte_mempool_dequeue_t dequeue;   
    rte_mempool_get_count get_count; 
    rte_mempool_calc_mem_size_t calc_mem_size;
    rte_mempool_populate_t populate;
    rte_mempool_get_info_t get_info;
    rte_mempool_dequeue_contig_blocks_t dequeue_contig_blocks;
};

#define RTE_MEMPOOL_MAX_OPS_IDX 16  
struct __rte_cache_aligned rte_mempool_ops_table {
    rte_spinlock_t sl;     
    uint32_t num_ops;      
    struct rte_mempool_ops ops[RTE_MEMPOOL_MAX_OPS_IDX];
};

extern struct rte_mempool_ops_table rte_mempool_ops_table;

static inline struct rte_mempool_ops *
rte_mempool_get_ops(int ops_index)
{
    RTE_VERIFY((ops_index >= 0) && (ops_index < RTE_MEMPOOL_MAX_OPS_IDX));

    return &rte_mempool_ops_table.ops[ops_index];
}

int
rte_mempool_ops_alloc(struct rte_mempool *mp);

static inline int
rte_mempool_ops_dequeue_bulk(struct rte_mempool *mp,
        void **obj_table, unsigned n)
{
    struct rte_mempool_ops *ops;
    int ret;

    rte_mempool_trace_ops_dequeue_bulk(mp, obj_table, n);
    ops = rte_mempool_get_ops(mp->ops_index);
    ret = ops->dequeue(mp, obj_table, n);
    if (ret == 0) {
        RTE_MEMPOOL_STAT_ADD(mp, get_common_pool_bulk, 1);
        RTE_MEMPOOL_STAT_ADD(mp, get_common_pool_objs, n);
    }
    return ret;
}

static inline int
rte_mempool_ops_dequeue_contig_blocks(struct rte_mempool *mp,
        void **first_obj_table, unsigned int n)
{
    struct rte_mempool_ops *ops;

    ops = rte_mempool_get_ops(mp->ops_index);
    RTE_ASSERT(ops->dequeue_contig_blocks != NULL);
    rte_mempool_trace_ops_dequeue_contig_blocks(mp, first_obj_table, n);
    return ops->dequeue_contig_blocks(mp, first_obj_table, n);
}

static inline int
rte_mempool_ops_enqueue_bulk(struct rte_mempool *mp, void * const *obj_table,
        unsigned n)
{
    struct rte_mempool_ops *ops;
    int ret;

    RTE_MEMPOOL_STAT_ADD(mp, put_common_pool_bulk, 1);
    RTE_MEMPOOL_STAT_ADD(mp, put_common_pool_objs, n);
    rte_mempool_trace_ops_enqueue_bulk(mp, obj_table, n);
    ops = rte_mempool_get_ops(mp->ops_index);
    ret = ops->enqueue(mp, obj_table, n);
#ifdef RTE_LIBRTE_MEMPOOL_DEBUG
    if (unlikely(ret < 0))
        RTE_MEMPOOL_LOG(CRIT, "cannot enqueue %u objects to mempool %s",
            n, mp->name);
#endif
    return ret;
}

unsigned
rte_mempool_ops_get_count(const struct rte_mempool *mp);

ssize_t rte_mempool_ops_calc_mem_size(const struct rte_mempool *mp,
                      uint32_t obj_num, uint32_t pg_shift,
                      size_t *min_chunk_size, size_t *align);

int rte_mempool_ops_populate(struct rte_mempool *mp, unsigned int max_objs,
                 void *vaddr, rte_iova_t iova, size_t len,
                 rte_mempool_populate_obj_cb_t *obj_cb,
                 void *obj_cb_arg);

int rte_mempool_ops_get_info(const struct rte_mempool *mp,
             struct rte_mempool_info *info);

void
rte_mempool_ops_free(struct rte_mempool *mp);

int
rte_mempool_set_ops_byname(struct rte_mempool *mp, const char *name,
        void *pool_config);

int rte_mempool_register_ops(const struct rte_mempool_ops *ops);

#define RTE_MEMPOOL_REGISTER_OPS(ops)               \
    RTE_INIT(mp_hdlr_init_##ops)                \
    {                           \
        rte_mempool_register_ops(&ops);         \
    }

typedef void (rte_mempool_obj_cb_t)(struct rte_mempool *mp,
        void *opaque, void *obj, unsigned obj_idx);
typedef rte_mempool_obj_cb_t rte_mempool_obj_ctor_t; /* compat */

typedef void (rte_mempool_mem_cb_t)(struct rte_mempool *mp,
        void *opaque, struct rte_mempool_memhdr *memhdr,
        unsigned mem_idx);

typedef void (rte_mempool_ctor_t)(struct rte_mempool *, void *);

struct rte_mempool *
rte_mempool_create(const char *name, unsigned n, unsigned elt_size,
           unsigned cache_size, unsigned private_data_size,
           rte_mempool_ctor_t *mp_init, void *mp_init_arg,
           rte_mempool_obj_cb_t *obj_init, void *obj_init_arg,
           int socket_id, unsigned flags);

struct rte_mempool *
rte_mempool_create_empty(const char *name, unsigned n, unsigned elt_size,
    unsigned cache_size, unsigned private_data_size,
    int socket_id, unsigned flags);
void
rte_mempool_free(struct rte_mempool *mp);

int rte_mempool_populate_iova(struct rte_mempool *mp, char *vaddr,
    rte_iova_t iova, size_t len, rte_mempool_memchunk_free_cb_t *free_cb,
    void *opaque);

int
rte_mempool_populate_virt(struct rte_mempool *mp, char *addr,
    size_t len, size_t pg_sz, rte_mempool_memchunk_free_cb_t *free_cb,
    void *opaque);

int rte_mempool_populate_default(struct rte_mempool *mp);

int rte_mempool_populate_anon(struct rte_mempool *mp);

uint32_t rte_mempool_obj_iter(struct rte_mempool *mp,
    rte_mempool_obj_cb_t *obj_cb, void *obj_cb_arg);

uint32_t rte_mempool_mem_iter(struct rte_mempool *mp,
    rte_mempool_mem_cb_t *mem_cb, void *mem_cb_arg);

void rte_mempool_dump(FILE *f, struct rte_mempool *mp);

struct rte_mempool_cache *
rte_mempool_cache_create(uint32_t size, int socket_id);

void
rte_mempool_cache_free(struct rte_mempool_cache *cache);

static __rte_always_inline struct rte_mempool_cache *
rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
{
    if (mp->cache_size == 0)
        return NULL;

    if (lcore_id >= RTE_MAX_LCORE)
        return NULL;

    rte_mempool_trace_default_cache(mp, lcore_id,
        &mp->local_cache[lcore_id]);
    return &mp->local_cache[lcore_id];
}

static __rte_always_inline void
rte_mempool_cache_flush(struct rte_mempool_cache *cache,
            struct rte_mempool *mp)
{
    if (cache == NULL)
        cache = rte_mempool_default_cache(mp, rte_lcore_id());
    if (cache == NULL || cache->len == 0)
        return;
    rte_mempool_trace_cache_flush(cache, mp);
    rte_mempool_ops_enqueue_bulk(mp, cache->objs, cache->len);
    cache->len = 0;
}

static __rte_always_inline void
rte_mempool_do_generic_put(struct rte_mempool *mp, void * const *obj_table,
               unsigned int n, struct rte_mempool_cache *cache)
{
    void **cache_objs;

    /* No cache provided */
    if (unlikely(cache == NULL))
        goto driver_enqueue;

    /* increment stat now, adding in mempool always success */
    RTE_MEMPOOL_CACHE_STAT_ADD(cache, put_bulk, 1);
    RTE_MEMPOOL_CACHE_STAT_ADD(cache, put_objs, n);

    /* The request itself is too big for the cache */
    if (unlikely(n > cache->flushthresh))
        goto driver_enqueue_stats_incremented;

    /*
     * The cache follows the following algorithm:
     *   1. If the objects cannot be added to the cache without crossing
     *      the flush threshold, flush the cache to the backend.
     *   2. Add the objects to the cache.
     */

    if (cache->len + n <= cache->flushthresh) {
        cache_objs = &cache->objs[cache->len];
        cache->len += n;
    } else {
        cache_objs = &cache->objs[0];
        rte_mempool_ops_enqueue_bulk(mp, cache_objs, cache->len);
        cache->len = n;
    }

    /* Add the objects to the cache. */
    rte_memcpy(cache_objs, obj_table, sizeof(void *) * n);

    return;

driver_enqueue:

    /* increment stat now, adding in mempool always success */
    RTE_MEMPOOL_STAT_ADD(mp, put_bulk, 1);
    RTE_MEMPOOL_STAT_ADD(mp, put_objs, n);

driver_enqueue_stats_incremented:

    /* push objects to the backend */
    rte_mempool_ops_enqueue_bulk(mp, obj_table, n);
}


static __rte_always_inline void
rte_mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
            unsigned int n, struct rte_mempool_cache *cache)
{
    rte_mempool_trace_generic_put(mp, obj_table, n, cache);
    RTE_MEMPOOL_CHECK_COOKIES(mp, obj_table, n, 0);
    rte_mempool_do_generic_put(mp, obj_table, n, cache);
}

static __rte_always_inline void
rte_mempool_put_bulk(struct rte_mempool *mp, void * const *obj_table,
             unsigned int n)
{
    struct rte_mempool_cache *cache;
    cache = rte_mempool_default_cache(mp, rte_lcore_id());
    rte_mempool_trace_put_bulk(mp, obj_table, n, cache);
    rte_mempool_generic_put(mp, obj_table, n, cache);
}

static __rte_always_inline void
rte_mempool_put(struct rte_mempool *mp, void *obj)
{
    rte_mempool_put_bulk(mp, &obj, 1);
}

static __rte_always_inline int
rte_mempool_do_generic_get(struct rte_mempool *mp, void **obj_table,
               unsigned int n, struct rte_mempool_cache *cache)
{
    int ret;
    unsigned int remaining;
    uint32_t index, len;
    void **cache_objs;

    /* No cache provided */
    if (unlikely(cache == NULL)) {
        remaining = n;
        goto driver_dequeue;
    }

    /* The cache is a stack, so copy will be in reverse order. */
    cache_objs = &cache->objs[cache->len];

    if (__rte_constant(n) && n <= cache->len) {
        /*
         * The request size is known at build time, and
         * the entire request can be satisfied from the cache,
         * so let the compiler unroll the fixed length copy loop.
         */
        cache->len -= n;
        for (index = 0; index < n; index++)
            *obj_table++ = *--cache_objs;

        RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_bulk, 1);
        RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_objs, n);

        return 0;
    }

    /*
     * Use the cache as much as we have to return hot objects first.
     * If the request size 'n' is known at build time, the above comparison
     * ensures that n > cache->len here, so omit RTE_MIN().
     */
    len = __rte_constant(n) ? cache->len : RTE_MIN(n, cache->len);
    cache->len -= len;
    remaining = n - len;
    for (index = 0; index < len; index++)
        *obj_table++ = *--cache_objs;

    /*
     * If the request size 'n' is known at build time, the case
     * where the entire request can be satisfied from the cache
     * has already been handled above, so omit handling it here.
     */
    if (!__rte_constant(n) && remaining == 0) {
        /* The entire request is satisfied from the cache. */

        RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_bulk, 1);
        RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_objs, n);

        return 0;
    }

    /* if dequeue below would overflow mem allocated for cache */
    if (unlikely(remaining > RTE_MEMPOOL_CACHE_MAX_SIZE))
        goto driver_dequeue;

    /* Fill the cache from the backend; fetch size + remaining objects. */
    ret = rte_mempool_ops_dequeue_bulk(mp, cache->objs,
            cache->size + remaining);
    if (unlikely(ret < 0)) {
        /*
         * We are buffer constrained, and not able to allocate
         * cache + remaining.
         * Do not fill the cache, just satisfy the remaining part of
         * the request directly from the backend.
         */
        goto driver_dequeue;
    }

    /* Satisfy the remaining part of the request from the filled cache. */
    cache_objs = &cache->objs[cache->size + remaining];
    for (index = 0; index < remaining; index++)
        *obj_table++ = *--cache_objs;

    cache->len = cache->size;

    RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_bulk, 1);
    RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_objs, n);

    return 0;

driver_dequeue:

    /* Get remaining objects directly from the backend. */
    ret = rte_mempool_ops_dequeue_bulk(mp, obj_table, remaining);

    if (ret < 0) {
        if (likely(cache != NULL)) {
            cache->len = n - remaining;
            /*
             * No further action is required to roll the first part
             * of the request back into the cache, as objects in
             * the cache are intact.
             */
        }

        RTE_MEMPOOL_STAT_ADD(mp, get_fail_bulk, 1);
        RTE_MEMPOOL_STAT_ADD(mp, get_fail_objs, n);
    } else {
        if (likely(cache != NULL)) {
            RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_bulk, 1);
            RTE_MEMPOOL_CACHE_STAT_ADD(cache, get_success_objs, n);
        } else {
            RTE_MEMPOOL_STAT_ADD(mp, get_success_bulk, 1);
            RTE_MEMPOOL_STAT_ADD(mp, get_success_objs, n);
        }
    }

    return ret;
}

static __rte_always_inline int
rte_mempool_generic_get(struct rte_mempool *mp, void **obj_table,
            unsigned int n, struct rte_mempool_cache *cache)
{
    int ret;
    ret = rte_mempool_do_generic_get(mp, obj_table, n, cache);
    if (ret == 0)
        RTE_MEMPOOL_CHECK_COOKIES(mp, obj_table, n, 1);
    rte_mempool_trace_generic_get(mp, obj_table, n, cache);
    return ret;
}

static __rte_always_inline int
rte_mempool_get_bulk(struct rte_mempool *mp, void **obj_table, unsigned int n)
{
    struct rte_mempool_cache *cache;
    cache = rte_mempool_default_cache(mp, rte_lcore_id());
    rte_mempool_trace_get_bulk(mp, obj_table, n, cache);
    return rte_mempool_generic_get(mp, obj_table, n, cache);
}

static __rte_always_inline int
rte_mempool_get(struct rte_mempool *mp, void **obj_p)
{
    return rte_mempool_get_bulk(mp, obj_p, 1);
}

static __rte_always_inline int
rte_mempool_get_contig_blocks(struct rte_mempool *mp,
                  void **first_obj_table, unsigned int n)
{
    int ret;

    ret = rte_mempool_ops_dequeue_contig_blocks(mp, first_obj_table, n);
    if (ret == 0) {
        RTE_MEMPOOL_STAT_ADD(mp, get_success_bulk, 1);
        RTE_MEMPOOL_STAT_ADD(mp, get_success_blks, n);
        RTE_MEMPOOL_CONTIG_BLOCKS_CHECK_COOKIES(mp, first_obj_table, n,
                            1);
    } else {
        RTE_MEMPOOL_STAT_ADD(mp, get_fail_bulk, 1);
        RTE_MEMPOOL_STAT_ADD(mp, get_fail_blks, n);
    }

    rte_mempool_trace_get_contig_blocks(mp, first_obj_table, n);
    return ret;
}

unsigned int rte_mempool_avail_count(const struct rte_mempool *mp);

unsigned int
rte_mempool_in_use_count(const struct rte_mempool *mp);

static inline int
rte_mempool_full(const struct rte_mempool *mp)
{
    return rte_mempool_avail_count(mp) == mp->size;
}

static inline int
rte_mempool_empty(const struct rte_mempool *mp)
{
    return rte_mempool_avail_count(mp) == 0;
}

static inline rte_iova_t
rte_mempool_virt2iova(const void *elt)
{
    const struct rte_mempool_objhdr *hdr;
    hdr = (const struct rte_mempool_objhdr *)RTE_PTR_SUB(elt,
        sizeof(*hdr));
    return hdr->iova;
}

void rte_mempool_audit(struct rte_mempool *mp);

static inline void *rte_mempool_get_priv(struct rte_mempool *mp)
{
    return (char *)mp +
        RTE_MEMPOOL_HEADER_SIZE(mp, mp->cache_size);
}

void rte_mempool_list_dump(FILE *f);

struct rte_mempool *rte_mempool_lookup(const char *name);

uint32_t rte_mempool_calc_obj_size(uint32_t elt_size, uint32_t flags,
    struct rte_mempool_objsz *sz);

void rte_mempool_walk(void (*func)(struct rte_mempool *, void *arg),
              void *arg);

struct rte_mempool_mem_range_info {
    void *start;
    size_t length;
    bool is_contiguous;
};

__rte_experimental
int
rte_mempool_get_mem_range(const struct rte_mempool *mp,
    struct rte_mempool_mem_range_info *mem_range);

__rte_experimental
size_t
rte_mempool_get_obj_alignment(const struct rte_mempool *mp);

int
rte_mempool_get_page_size(struct rte_mempool *mp, size_t *pg_sz);

enum rte_mempool_event {
    RTE_MEMPOOL_EVENT_READY = 0,
    RTE_MEMPOOL_EVENT_DESTROY = 1,
};

typedef void (rte_mempool_event_callback)(
        enum rte_mempool_event event,
        struct rte_mempool *mp,
        void *user_data);

__rte_internal
int
rte_mempool_event_callback_register(rte_mempool_event_callback *func,
                    void *user_data);

__rte_internal
int
rte_mempool_event_callback_unregister(rte_mempool_event_callback *func,
                      void *user_data);

#ifdef __cplusplus
}
#endif

#endif /* _RTE_MEMPOOL_H_ */