path: root/libglusterfs/src/mem-pool.c

/*
  Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.com>
  This file is part of GlusterFS.

  This file is licensed to you under your choice of the GNU Lesser
  General Public License, version 3 or any later version (LGPLv3 or
  later), or the GNU General Public License, version 2 (GPLv2), in all
  cases as published by the Free Software Foundation.
*/

#include "glusterfs/mem-pool.h"
#include "glusterfs/logging.h"
#include "glusterfs/xlator.h"
#include <stdlib.h>
#include <stdarg.h>

#include "unittest/unittest.h"
#include "glusterfs/libglusterfs-messages.h"

void
gf_mem_acct_enable_set(void *data)
{
    glusterfs_ctx_t *ctx = NULL;

    REQUIRE(data != NULL);

    ctx = data;

    GF_ASSERT(ctx != NULL);

    ctx->mem_acct_enable = 1;

    ENSURE(1 == ctx->mem_acct_enable);

    return;
}

int
gf_mem_set_acct_info(xlator_t *xl, char **alloc_ptr, size_t size, uint32_t type,
                     const char *typestr)
{
    void *ptr = NULL;
    struct mem_header *header = NULL;

    if (!alloc_ptr)
        return -1;

    ptr = *alloc_ptr;

    GF_ASSERT(xl != NULL);

    GF_ASSERT(xl->mem_acct != NULL);

    GF_ASSERT(type <= xl->mem_acct->num_types);

    LOCK(&xl->mem_acct->rec[type].lock);
    {
        if (!xl->mem_acct->rec[type].typestr)
            xl->mem_acct->rec[type].typestr = typestr;
        xl->mem_acct->rec[type].size += size;
        xl->mem_acct->rec[type].num_allocs++;
        xl->mem_acct->rec[type].total_allocs++;
        xl->mem_acct->rec[type].max_size = max(xl->mem_acct->rec[type].max_size,
                                               xl->mem_acct->rec[type].size);
        xl->mem_acct->rec[type].max_num_allocs = max(
            xl->mem_acct->rec[type].max_num_allocs,
            xl->mem_acct->rec[type].num_allocs);
    }
    UNLOCK(&xl->mem_acct->rec[type].lock);

    GF_ATOMIC_INC(xl->mem_acct->refcnt);

    header = (struct mem_header *)ptr;
    header->type = type;
    header->size = size;
    header->mem_acct = xl->mem_acct;
    header->magic = GF_MEM_HEADER_MAGIC;

#ifdef DEBUG
    INIT_LIST_HEAD(&header->acct_list);
    LOCK(&xl->mem_acct->rec[type].lock);
    {
        list_add(&header->acct_list, &(xl->mem_acct->rec[type].obj_list));
    }
    UNLOCK(&xl->mem_acct->rec[type].lock);
#endif
    ptr += sizeof(struct mem_header);
    /* data follows in this gap of 'size' bytes */
    *(uint32_t *)(ptr + size) = GF_MEM_TRAILER_MAGIC;

    *alloc_ptr = ptr;
    return 0;
}

void *
__gf_calloc(size_t nmemb, size_t size, uint32_t type, const char *typestr)
{
    size_t tot_size = 0;
    size_t req_size = 0;
    char *ptr = NULL;
    xlator_t *xl = NULL;

    if (!THIS->ctx->mem_acct_enable)
        return CALLOC(nmemb, size);

    xl = THIS;

    req_size = nmemb * size;
    tot_size = req_size + GF_MEM_HEADER_SIZE + GF_MEM_TRAILER_SIZE;

    ptr = calloc(1, tot_size);

    if (!ptr) {
        gf_msg_nomem("", GF_LOG_ALERT, tot_size);
        return NULL;
    }
    gf_mem_set_acct_info(xl, &ptr, req_size, type, typestr);

    return (void *)ptr;
}

void *
__gf_malloc(size_t size, uint32_t type, const char *typestr)
{
    size_t tot_size = 0;
    char *ptr = NULL;
    xlator_t *xl = NULL;

    if (!THIS->ctx->mem_acct_enable)
        return MALLOC(size);

    xl = THIS;

    tot_size = size + GF_MEM_HEADER_SIZE + GF_MEM_TRAILER_SIZE;

    ptr = malloc(tot_size);
    if (!ptr) {
        gf_msg_nomem("", GF_LOG_ALERT, tot_size);
        return NULL;
    }
    gf_mem_set_acct_info(xl, &ptr, size, type, typestr);

    return (void *)ptr;
}

void *
__gf_realloc(void *ptr, size_t size)
{
    size_t tot_size = 0;
    char *new_ptr;
    struct mem_header *old_header = NULL;
    struct mem_header *new_header = NULL;
    struct mem_header tmp_header;

    if (!THIS->ctx->mem_acct_enable)
        return REALLOC(ptr, size);

    REQUIRE(NULL != ptr);

    old_header = (struct mem_header *)(ptr - GF_MEM_HEADER_SIZE);
    GF_ASSERT(old_header->magic == GF_MEM_HEADER_MAGIC);
    tmp_header = *old_header;

#ifdef DEBUG
    int type = 0;
    size_t copy_size = 0;

    /* Making these changes for realloc is not straightforward. So
     * I am simulating realloc using calloc and free
     */

    type = tmp_header.type;
    new_ptr = __gf_calloc(1, size, type,
                          tmp_header.mem_acct->rec[type].typestr);
    if (new_ptr) {
        copy_size = (size > tmp_header.size) ? tmp_header.size : size;
        memcpy(new_ptr, ptr, copy_size);
        __gf_free(ptr);
    }

    /* This is not quite what the man page says should happen */
    return new_ptr;
#endif

    tot_size = size + GF_MEM_HEADER_SIZE + GF_MEM_TRAILER_SIZE;
    new_ptr = realloc(old_header, tot_size);
    if (!new_ptr) {
        gf_msg_nomem("", GF_LOG_ALERT, tot_size);
        return NULL;
    }

    /*
     * We used to pass (char **)&ptr as the second
     * argument after the value of realloc was saved
     * in ptr, but the compiler warnings complained
     * about the casting to and forth from void ** to
     * char **.
     * TBD: it would be nice to adjust the memory accounting info here,
     * but calling gf_mem_set_acct_info here is wrong because it bumps
     * up counts as though this is a new allocation - which it's not.
     * The consequence of doing nothing here is only that the sizes will be
     * wrong, but at least the counts won't be.
    uint32_t           type = 0;
    xlator_t          *xl = NULL;
    type = header->type;
    xl = (xlator_t *) header->xlator;
    gf_mem_set_acct_info (xl, &new_ptr, size, type, NULL);
     */

    new_header = (struct mem_header *)new_ptr;
    *new_header = tmp_header;
    new_header->size = size;

    new_ptr += sizeof(struct mem_header);
    /* data follows in this gap of 'size' bytes */
    *(uint32_t *)(new_ptr + size) = GF_MEM_TRAILER_MAGIC;

    return (void *)new_ptr;
}

int
gf_vasprintf(char **string_ptr, const char *format, va_list arg)
{
    va_list arg_save;
    char *str = NULL;
    int size = 0;
    int rv = 0;

    if (!string_ptr || !format)
        return -1;

    va_copy(arg_save, arg);

    size = vsnprintf(NULL, 0, format, arg);
    size++;
    str = GF_MALLOC(size, gf_common_mt_asprintf);
    if (str == NULL) {
        /* log is done in GF_MALLOC itself */
        va_end(arg_save);
        return -1;
    }
    rv = vsnprintf(str, size, format, arg_save);

    *string_ptr = str;
    va_end(arg_save);
    return (rv);
}

int
gf_asprintf(char **string_ptr, const char *format, ...)
{
    va_list arg;
    int rv = 0;

    va_start(arg, format);
    rv = gf_vasprintf(string_ptr, format, arg);
    va_end(arg);

    return rv;
}

#ifdef DEBUG
void
__gf_mem_invalidate(void *ptr)
{
    struct mem_header *header = ptr;
    void *end = NULL;

    struct mem_invalid inval = {
        .magic = GF_MEM_INVALID_MAGIC,
        .mem_acct = header->mem_acct,
        .type = header->type,
        .size = header->size,
        .baseaddr = ptr + GF_MEM_HEADER_SIZE,
    };

    /* calculate the last byte of the allocated area */
    end = ptr + GF_MEM_HEADER_SIZE + inval.size + GF_MEM_TRAILER_SIZE;

    /* overwrite the old mem_header */
    memcpy(ptr, &inval, sizeof(inval));
    ptr += sizeof(inval);

    /* zero out remaining (old) mem_header bytes) */
    memset(ptr, 0x00, sizeof(*header) - sizeof(inval));
    ptr += sizeof(*header) - sizeof(inval);

    /* zero out the first byte of data */
    *(uint32_t *)(ptr) = 0x00;
    ptr += 1;

    /* repeated writes of invalid structurein data area */
    while ((ptr + (sizeof(inval))) < (end - 1)) {
        memcpy(ptr, &inval, sizeof(inval));
        ptr += sizeof(inval);
    }

    /* fill out remaining data area with 0xff */
    memset(ptr, 0xff, end - ptr);
}
#endif /* DEBUG */

/* Coverity taint NOTE: pointers passed to free, would operate on
pointer-GF_MEM_HEADER_SIZE content and if the pointer was used for any IO
related purpose, the pointer stands tainted, and hence coverity would consider
access to the said region as tainted. The following directive to coverity hence
sanitizes the pointer, thus removing any taint to the same within this function.
If the pointer is accessed outside the scope of this function without any
checks on content read from an IO operation, taints will still be reported, and
needs appropriate addressing. */

/* coverity[ +tainted_data_sanitize : arg-0 ] */
static void
gf_free_sanitize(void *s)
{
}

void
__gf_free(void *free_ptr)
{
    void *ptr = NULL;
    struct mem_acct *mem_acct;
    struct mem_header *header = NULL;

    if (!THIS->ctx->mem_acct_enable) {
        FREE(free_ptr);
        return;
    }

    if (!free_ptr)
        return;

    gf_free_sanitize(free_ptr);
    ptr = free_ptr - GF_MEM_HEADER_SIZE;
    header = (struct mem_header *)ptr;

    // Possible corruption, assert here
    GF_ASSERT(GF_MEM_HEADER_MAGIC == header->magic);

    mem_acct = header->mem_acct;
    if (!mem_acct) {
        goto free;
    }

    // This points to a memory overrun
    GF_ASSERT(GF_MEM_TRAILER_MAGIC ==
              *(uint32_t *)((char *)free_ptr + header->size));

    LOCK(&mem_acct->rec[header->type].lock);
    {
        GF_ASSERT(mem_acct->rec[header->type].size >= header->size);
        mem_acct->rec[header->type].size -= header->size;
        mem_acct->rec[header->type].num_allocs--;
        /* If all the instances are freed up then ensure typestr is set
         * to NULL */
        if (!mem_acct->rec[header->type].num_allocs)
            mem_acct->rec[header->type].typestr = NULL;
#ifdef DEBUG
        list_del(&header->acct_list);
#endif
    }
    UNLOCK(&mem_acct->rec[header->type].lock);

    if (GF_ATOMIC_DEC(mem_acct->refcnt) == 0) {
        FREE(mem_acct);
    }

free:
#ifdef DEBUG
    __gf_mem_invalidate(ptr);
#endif

    FREE(ptr);
}

#define POOL_SMALLEST 7 /* i.e. 128 */
#define POOL_LARGEST 20 /* i.e. 1048576 */
#define NPOOLS (POOL_LARGEST - POOL_SMALLEST + 1)

static pthread_key_t pool_key;
static pthread_mutex_t pool_lock = PTHREAD_MUTEX_INITIALIZER;
static struct list_head pool_threads;
static pthread_mutex_t pool_free_lock = PTHREAD_MUTEX_INITIALIZER;
static struct list_head pool_free_threads;
static struct mem_pool_shared pools[NPOOLS];
static size_t pool_list_size;

#if !defined(GF_DISABLE_MEMPOOL)
#define N_COLD_LISTS 1024
#define POOL_SWEEP_SECS 30

static unsigned long sweep_times;
static unsigned long sweep_usecs;
static unsigned long frees_to_system;

typedef struct {
    struct list_head death_row;
    pooled_obj_hdr_t *cold_lists[N_COLD_LISTS];
    unsigned int n_cold_lists;
} sweep_state_t;

enum init_state {
    GF_MEMPOOL_INIT_NONE = 0,
    GF_MEMPOOL_INIT_PREINIT,
    GF_MEMPOOL_INIT_EARLY,
    GF_MEMPOOL_INIT_LATE,
    GF_MEMPOOL_INIT_DESTROY
};

static enum init_state init_done = GF_MEMPOOL_INIT_NONE;
static pthread_mutex_t init_mutex = PTHREAD_MUTEX_INITIALIZER;
static unsigned int init_count = 0;
static pthread_t sweeper_tid;

gf_boolean_t
collect_garbage(sweep_state_t *state, per_thread_pool_list_t *pool_list)
{
    unsigned int i;
    per_thread_pool_t *pt_pool;
    gf_boolean_t poisoned;

    (void)pthread_spin_lock(&pool_list->lock);

    poisoned = pool_list->poison != 0;
    if (!poisoned) {
        for (i = 0; i < NPOOLS; ++i) {
            pt_pool = &pool_list->pools[i];
            if (pt_pool->cold_list) {
                if (state->n_cold_lists >= N_COLD_LISTS) {
                    break;
                }
                state->cold_lists[state->n_cold_lists++] = pt_pool->cold_list;
            }
            pt_pool->cold_list = pt_pool->hot_list;
            pt_pool->hot_list = NULL;
        }
    }

    (void)pthread_spin_unlock(&pool_list->lock);

    return poisoned;
}

void
free_obj_list(pooled_obj_hdr_t *victim)
{
    pooled_obj_hdr_t *next;

    while (victim) {
        next = victim->next;
        free(victim);
        victim = next;
        ++frees_to_system;
    }
}

void *
pool_sweeper(void *arg)
{
    sweep_state_t state;
    per_thread_pool_list_t *pool_list;
    per_thread_pool_list_t *next_pl;
    per_thread_pool_t *pt_pool;
    unsigned int i;
    struct timeval begin_time;
    struct timeval end_time;
    struct timeval elapsed;
    gf_boolean_t poisoned;

    /*
     * This is all a bit inelegant, but the point is to avoid doing
     * expensive things (like freeing thousands of objects) while holding a
     * global lock.  Thus, we split each iteration into three passes, with
     * only the first and fastest holding the lock.
     */

    for (;;) {
        sleep(POOL_SWEEP_SECS);
        (void)pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
        INIT_LIST_HEAD(&state.death_row);
        state.n_cold_lists = 0;

        /* First pass: collect stuff that needs our attention. */
        (void)gettimeofday(&begin_time, NULL);
        (void)pthread_mutex_lock(&pool_lock);
        list_for_each_entry_safe(pool_list, next_pl, &pool_threads, thr_list)
        {
            (void)pthread_mutex_unlock(&pool_lock);
            poisoned = collect_garbage(&state, pool_list);
            (void)pthread_mutex_lock(&pool_lock);

            if (poisoned) {
                list_move(&pool_list->thr_list, &state.death_row);
            }
        }
        (void)pthread_mutex_unlock(&pool_lock);
        (void)gettimeofday(&end_time, NULL);
        timersub(&end_time, &begin_time, &elapsed);
        sweep_usecs += elapsed.tv_sec * 1000000 + elapsed.tv_usec;
        sweep_times += 1;

        /* Second pass: free dead pools. */
        (void)pthread_mutex_lock(&pool_free_lock);
        list_for_each_entry_safe(pool_list, next_pl, &state.death_row, thr_list)
        {
            for (i = 0; i < NPOOLS; ++i) {
                pt_pool = &pool_list->pools[i];
                free_obj_list(pt_pool->cold_list);
                free_obj_list(pt_pool->hot_list);
                pt_pool->hot_list = pt_pool->cold_list = NULL;
            }
            list_del(&pool_list->thr_list);
            list_add(&pool_list->thr_list, &pool_free_threads);
        }
        (void)pthread_mutex_unlock(&pool_free_lock);

        /* Third pass: free cold objects from live pools. */
        for (i = 0; i < state.n_cold_lists; ++i) {
            free_obj_list(state.cold_lists[i]);
        }
        (void)pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    }
}

void
pool_destructor(void *arg)
{
    per_thread_pool_list_t *pool_list = arg;

    /* The pool-sweeper thread will take it from here.
     *
     * We can change 'poison' here without taking locks because the change
     * itself doesn't interact with other parts of the code and a simple write
     * is already atomic from the point of view of the processor.
     *
     * This change can modify what mem_put() does, but both possibilities are
     * fine until the sweeper thread kicks in. The real synchronization must be
     * between mem_put() and the sweeper thread. */
    pool_list->poison = 1;
}

static __attribute__((constructor)) void
mem_pools_preinit(void)
{
    unsigned int i;

    INIT_LIST_HEAD(&pool_threads);
    INIT_LIST_HEAD(&pool_free_threads);

    for (i = 0; i < NPOOLS; ++i) {
        pools[i].power_of_two = POOL_SMALLEST + i;

        GF_ATOMIC_INIT(pools[i].allocs_hot, 0);
        GF_ATOMIC_INIT(pools[i].allocs_cold, 0);
        GF_ATOMIC_INIT(pools[i].allocs_stdc, 0);
        GF_ATOMIC_INIT(pools[i].frees_to_list, 0);
    }

    pool_list_size = sizeof(per_thread_pool_list_t) +
                     sizeof(per_thread_pool_t) * (NPOOLS - 1);

    init_done = GF_MEMPOOL_INIT_PREINIT;
}

/* Use mem_pools_init_early() function for basic initialization. There will be
 * no cleanup done by the pool_sweeper thread until mem_pools_init_late() has
 * been called. Calling mem_get() will be possible after this function has
 * setup the basic structures. */
void
mem_pools_init_early(void)
{
    pthread_mutex_lock(&init_mutex);
    /* Use a pthread_key destructor to clean up when a thread exits.
     *
     * We won't increase init_count here, that is only done when the
     * pool_sweeper thread is started too.
     */
    if (init_done == GF_MEMPOOL_INIT_PREINIT ||
        init_done == GF_MEMPOOL_INIT_DESTROY) {
        /* key has not been created yet */
        if (pthread_key_create(&pool_key, pool_destructor) != 0) {
            gf_log("mem-pool", GF_LOG_CRITICAL,
                   "failed to initialize mem-pool key");
        }

        init_done = GF_MEMPOOL_INIT_EARLY;
    } else {
        gf_log("mem-pool", GF_LOG_CRITICAL,
               "incorrect order of mem-pool initialization "
               "(init_done=%d)",
               init_done);
    }

    pthread_mutex_unlock(&init_mutex);
}

/* Call mem_pools_init_late() once threading has been configured completely.
 * This prevent the pool_sweeper thread from getting killed once the main()
 * thread exits during deamonizing. */
void
mem_pools_init_late(void)
{
    pthread_mutex_lock(&init_mutex);
    if ((init_count++) == 0) {
        (void)gf_thread_create(&sweeper_tid, NULL, pool_sweeper, NULL,
                               "memsweep");

        init_done = GF_MEMPOOL_INIT_LATE;
    }
    pthread_mutex_unlock(&init_mutex);
}

void
mem_pools_fini(void)
{
    pthread_mutex_lock(&init_mutex);
    switch (init_count) {
        case 0:
            /*
             * If init_count is already zero (as e.g. if somebody called
             * this before mem_pools_init_late) then the sweeper was
             * probably never even started so we don't need to stop it.
             * Even if there's some crazy circumstance where there is a
             * sweeper but init_count is still zero, that just means we'll
             * leave it running.  Not perfect, but far better than any
             * known alternative.
             */
            break;
        case 1: {
            per_thread_pool_list_t *pool_list;
            per_thread_pool_list_t *next_pl;
            unsigned int i;

            /* if only mem_pools_init_early() was called, sweeper_tid will
             * be invalid and the functions will error out. That is not
             * critical. In all other cases, the sweeper_tid will be valid
             * and the thread gets stopped. */
            (void)pthread_cancel(sweeper_tid);
            (void)pthread_join(sweeper_tid, NULL);

            /* Need to clean the pool_key to prevent further usage of the
             * per_thread_pool_list_t structure that is stored for each
             * thread.
             * This also prevents calling pool_destructor() when a thread
             * exits, so there is no chance on a use-after-free of the
             * per_thread_pool_list_t structure. */
            (void)pthread_key_delete(pool_key);

            /* free all objects from all pools */
            list_for_each_entry_safe(pool_list, next_pl, &pool_threads,
                                     thr_list)
            {
                for (i = 0; i < NPOOLS; ++i) {
                    free_obj_list(pool_list->pools[i].hot_list);
                    free_obj_list(pool_list->pools[i].cold_list);
                    pool_list->pools[i].hot_list = NULL;
                    pool_list->pools[i].cold_list = NULL;
                }

                list_del(&pool_list->thr_list);
                FREE(pool_list);
            }

            list_for_each_entry_safe(pool_list, next_pl, &pool_free_threads,
                                     thr_list)
            {
                list_del(&pool_list->thr_list);
                FREE(pool_list);
            }

            init_done = GF_MEMPOOL_INIT_DESTROY;
            /* Fall through. */
        }
        default:
            --init_count;
    }
    pthread_mutex_unlock(&init_mutex);
}

#else
void
mem_pools_init_early(void)
{
}
void
mem_pools_init_late(void)
{
}
void
mem_pools_fini(void)
{
}
#endif

struct mem_pool *
mem_pool_new_fn(glusterfs_ctx_t *ctx, unsigned long sizeof_type,
                unsigned long count, char *name)
{
    unsigned int i;
    struct mem_pool *new = NULL;
    struct mem_pool_shared *pool = NULL;

    if (!sizeof_type) {
        gf_msg_callingfn("mem-pool", GF_LOG_ERROR, EINVAL, LG_MSG_INVALID_ARG,
                         "invalid argument");
        return NULL;
    }

    for (i = 0; i < NPOOLS; ++i) {
        if (sizeof_type <= AVAILABLE_SIZE(pools[i].power_of_two)) {
            pool = &pools[i];
            break;
        }
    }

    if (!pool) {
        gf_msg_callingfn("mem-pool", GF_LOG_ERROR, EINVAL, LG_MSG_INVALID_ARG,
                         "invalid argument");
        return NULL;
    }

    new = GF_CALLOC(sizeof(struct mem_pool), 1, gf_common_mt_mem_pool);
    if (!new)
        return NULL;

    new->ctx = ctx;
    new->sizeof_type = sizeof_type;
    new->count = count;
    new->name = name;
    new->pool = pool;
    GF_ATOMIC_INIT(new->active, 0);
    INIT_LIST_HEAD(&new->owner);

    LOCK(&ctx->lock);
    {
        list_add(&new->owner, &ctx->mempool_list);
    }
    UNLOCK(&ctx->lock);

    return new;
}

void *
mem_get0(struct mem_pool *mem_pool)
{
    void *ptr = NULL;

    if (!mem_pool) {
        gf_msg_callingfn("mem-pool", GF_LOG_ERROR, EINVAL, LG_MSG_INVALID_ARG,
                         "invalid argument");
        return NULL;
    }

    ptr = mem_get(mem_pool);
    if (ptr) {
#if defined(GF_DISABLE_MEMPOOL)
        memset(ptr, 0, mem_pool->sizeof_type);
#else
        memset(ptr, 0, AVAILABLE_SIZE(mem_pool->pool->power_of_two));
#endif
    }

    return ptr;
}

per_thread_pool_list_t *
mem_get_pool_list(void)
{
    per_thread_pool_list_t *pool_list;
    unsigned int i;

    pool_list = pthread_getspecific(pool_key);
    if (pool_list) {
        return pool_list;
    }

    (void)pthread_mutex_lock(&pool_free_lock);
    if (!list_empty(&pool_free_threads)) {
        pool_list = list_entry(pool_free_threads.next, per_thread_pool_list_t,
                               thr_list);
        list_del(&pool_list->thr_list);
    }
    (void)pthread_mutex_unlock(&pool_free_lock);

    if (!pool_list) {
        pool_list = MALLOC(pool_list_size);
        if (!pool_list) {
            return NULL;
        }

        INIT_LIST_HEAD(&pool_list->thr_list);
        (void)pthread_spin_init(&pool_list->lock, PTHREAD_PROCESS_PRIVATE);
        for (i = 0; i < NPOOLS; ++i) {
            pool_list->pools[i].parent = &pools[i];
            pool_list->pools[i].hot_list = NULL;
            pool_list->pools[i].cold_list = NULL;
        }
    }

    (void)pthread_mutex_lock(&pool_lock);
    pool_list->poison = 0;
    list_add(&pool_list->thr_list, &pool_threads);
    (void)pthread_mutex_unlock(&pool_lock);

    (void)pthread_setspecific(pool_key, pool_list);
    return pool_list;
}

pooled_obj_hdr_t *
mem_get_from_pool(struct mem_pool *mem_pool)
{
    per_thread_pool_list_t *pool_list;
    per_thread_pool_t *pt_pool;
    pooled_obj_hdr_t *retval;

    pool_list = mem_get_pool_list();
    if (!pool_list || pool_list->poison) {
        return NULL;
    }

    pt_pool = &pool_list->pools[mem_pool->pool->power_of_two - POOL_SMALLEST];

    (void)pthread_spin_lock(&pool_list->lock);

    retval = pt_pool->hot_list;
    if (retval) {
        pt_pool->hot_list = retval->next;
        (void)pthread_spin_unlock(&pool_list->lock);
        GF_ATOMIC_INC(pt_pool->parent->allocs_hot);
    } else {
        retval = pt_pool->cold_list;
        if (retval) {
            pt_pool->cold_list = retval->next;
            (void)pthread_spin_unlock(&pool_list->lock);
            GF_ATOMIC_INC(pt_pool->parent->allocs_cold);
        } else {
            (void)pthread_spin_unlock(&pool_list->lock);
            GF_ATOMIC_INC(pt_pool->parent->allocs_stdc);
            retval = malloc(1 << pt_pool->parent->power_of_two);
        }
    }

    if (retval != NULL) {
        retval->magic = GF_MEM_HEADER_MAGIC;
        retval->pool = mem_pool;
        retval->pool_list = pool_list;
        retval->power_of_two = mem_pool->pool->power_of_two;
    }

    return retval;
}

void *
mem_get(struct mem_pool *mem_pool)
{
#if defined(GF_DISABLE_MEMPOOL)
    return GF_MALLOC(mem_pool->sizeof_type, gf_common_mt_mem_pool);
#else
    pooled_obj_hdr_t *retval;

    if (!mem_pool) {
        gf_msg_callingfn("mem-pool", GF_LOG_ERROR, EINVAL, LG_MSG_INVALID_ARG,
                         "invalid argument");
        return NULL;
    }

    retval = mem_get_from_pool(mem_pool);
    if (!retval) {
        return NULL;
    }

    GF_ATOMIC_INC(mem_pool->active);

    return retval + 1;
#endif /* GF_DISABLE_MEMPOOL */
}

void
mem_put(void *ptr)
{
#if defined(GF_DISABLE_MEMPOOL)
    GF_FREE(ptr);
#else
    pooled_obj_hdr_t *hdr;
    per_thread_pool_list_t *pool_list;
    per_thread_pool_t *pt_pool;

    if (!ptr) {
        gf_msg_callingfn("mem-pool", GF_LOG_ERROR, EINVAL, LG_MSG_INVALID_ARG,
                         "invalid argument");
        return;
    }

    hdr = ((pooled_obj_hdr_t *)ptr) - 1;
    if (hdr->magic != GF_MEM_HEADER_MAGIC) {
        /* Not one of ours; don't touch it. */
        return;
    }
    pool_list = hdr->pool_list;
    pt_pool = &pool_list->pools[hdr->power_of_two - POOL_SMALLEST];

    GF_ATOMIC_DEC(hdr->pool->active);

    hdr->magic = GF_MEM_INVALID_MAGIC;

    (void)pthread_spin_lock(&pool_list->lock);
    if (!pool_list->poison) {
        hdr->next = pt_pool->hot_list;
        pt_pool->hot_list = hdr;
        (void)pthread_spin_unlock(&pool_list->lock);
        GF_ATOMIC_INC(pt_pool->parent->frees_to_list);
    } else {
        /* If the owner thread of this element has terminated, we simply
         * release its memory. */
        (void)pthread_spin_unlock(&pool_list->lock);
        free(hdr);
    }
#endif /* GF_DISABLE_MEMPOOL */
}

void
mem_pool_destroy(struct mem_pool *pool)
{
    if (!pool)
        return;

    /* remove this pool from the owner (glusterfs_ctx_t) */
    LOCK(&pool->ctx->lock);
    {
        list_del(&pool->owner);
    }
    UNLOCK(&pool->ctx->lock);

    /* free this pool, but keep the mem_pool_shared */
    GF_FREE(pool);

    /*
     * Pools are now permanent, so the mem_pool->pool is kept around. All
     * of the objects *in* the pool will eventually be freed via the
     * pool-sweeper thread, and this way we don't have to add a lot of
     * reference-counting complexity.
     */
}