pool.c

Go to the documentation of this file.

/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
/*
 * See COPYRIGHT in top-level directory.
 */
 
#include "abti.h"
 
ABTU_ret_err static int pool_create(ABTI_pool_def *def, ABT_pool_config config,
                                    ABT_bool automatic, ABT_bool is_builtin,
                                    ABTI_pool **pp_newpool);
 
int ABT_pool_create(ABT_pool_def *def, ABT_pool_config config,
                    ABT_pool *newpool)
{
#ifndef ABT_CONFIG_ENABLE_VER_20_API
    /* Argobots 1.x sets newpool to NULL on error. */
    *newpool = ABT_POOL_NULL;
#endif
    /* Copy def */
    ABTI_pool_def internal_def;
 
    internal_def.access = def->access;
    internal_def.u_is_in_pool = def->u_is_in_pool;
    internal_def.u_create_from_thread = def->u_create_from_thread;
    internal_def.u_free = def->u_free;
    internal_def.p_init = def->p_init;
    internal_def.p_get_size = def->p_get_size;
    internal_def.p_push = def->p_push;
    internal_def.p_pop = def->p_pop;
#ifdef ABT_CONFIG_ENABLE_VER_20_API
    internal_def.p_pop_wait = def->p_pop_wait;
#else
    internal_def.p_pop_wait = NULL;
#endif
    internal_def.p_pop_timedwait = def->p_pop_timedwait;
    internal_def.p_remove = def->p_remove;
    internal_def.p_free = def->p_free;
    internal_def.p_print_all = def->p_print_all;
 
    ABTI_pool *p_newpool;
    int abt_errno =
        pool_create(&internal_def, config, ABT_FALSE, ABT_FALSE, &p_newpool);
    ABTI_CHECK_ERROR(abt_errno);
 
    *newpool = ABTI_pool_get_handle(p_newpool);
    return ABT_SUCCESS;
}
 
int ABT_pool_create_basic(ABT_pool_kind kind, ABT_pool_access access,
                          ABT_bool automatic, ABT_pool *newpool)
{
#ifndef ABT_CONFIG_ENABLE_VER_20_API
    /* Argobots 1.x sets newpool to NULL on error. */
    *newpool = ABT_POOL_NULL;
#endif
    ABTI_pool *p_newpool;
    int abt_errno = ABTI_pool_create_basic(kind, access, automatic, &p_newpool);
    ABTI_CHECK_ERROR(abt_errno);
 
    *newpool = ABTI_pool_get_handle(p_newpool);
    return ABT_SUCCESS;
}
 
int ABT_pool_free(ABT_pool *pool)
{
    ABT_pool h_pool = *pool;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(h_pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    ABTI_pool_free(p_pool);
 
    *pool = ABT_POOL_NULL;
    return ABT_SUCCESS;
}
 
int ABT_pool_get_access(ABT_pool pool, ABT_pool_access *access)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *access = p_pool->access;
    return ABT_SUCCESS;
}
 
int ABT_pool_get_total_size(ABT_pool pool, size_t *size)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *size = ABTI_pool_get_total_size(p_pool);
    return ABT_SUCCESS;
}
 
int ABT_pool_get_size(ABT_pool pool, size_t *size)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *size = ABTI_pool_get_size(p_pool);
    return ABT_SUCCESS;
}
 
int ABT_pool_pop(ABT_pool pool, ABT_unit *p_unit)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *p_unit = ABTI_pool_pop(p_pool);
    return ABT_SUCCESS;
}
 
int ABT_pool_pop_wait(ABT_pool pool, ABT_unit *p_unit, double time_secs)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->p_pop_wait, ABT_ERR_POOL);
 
    *p_unit = ABTI_pool_pop_wait(p_pool, time_secs);
    return ABT_SUCCESS;
}
 
int ABT_pool_pop_timedwait(ABT_pool pool, ABT_unit *p_unit, double abstime_secs)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->p_pop_timedwait, ABT_ERR_POOL);
 
    *p_unit = ABTI_pool_pop_timedwait(p_pool, abstime_secs);
    return ABT_SUCCESS;
}
 
int ABT_pool_push(ABT_pool pool, ABT_unit unit)
{
    ABTI_global *p_global = ABTI_global_get_global();
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    ABTI_CHECK_TRUE(unit != ABT_UNIT_NULL, ABT_ERR_INV_UNIT);
 
    ABTI_thread *p_thread;
    int abt_errno =
        ABTI_unit_set_associated_pool(p_global, unit, p_pool, &p_thread);
    ABTI_CHECK_ERROR(abt_errno);
    /* ABTI_unit_set_associated_pool() might change unit, so "unit" must be read
     * again from p_thread. */
    ABTI_pool_push(p_pool, p_thread->unit);
    return ABT_SUCCESS;
}
 
int ABT_pool_remove(ABT_pool pool, ABT_unit unit)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->p_remove, ABT_ERR_POOL);
 
    /* unit must be in this pool, so we do not need to reset its associated
     * pool. */
    int abt_errno = ABTI_pool_remove(p_pool, unit);
    ABTI_CHECK_ERROR(abt_errno);
    return ABT_SUCCESS;
}
 
int ABT_pool_print_all(ABT_pool pool, void *arg,
                       void (*print_fn)(void *, ABT_unit))
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->p_print_all, ABT_ERR_POOL);
 
    p_pool->p_print_all(pool, arg, print_fn);
    return ABT_SUCCESS;
}
 
int ABT_pool_set_data(ABT_pool pool, void *data)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    p_pool->data = data;
    return ABT_SUCCESS;
}
 
int ABT_pool_get_data(ABT_pool pool, void **data)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *data = p_pool->data;
    return ABT_SUCCESS;
}
 
int ABT_pool_add_sched(ABT_pool pool, ABT_sched sched)
{
    ABTI_local *p_local = ABTI_local_get_local();
 
    ABTI_global *p_global;
    ABTI_SETUP_GLOBAL(&p_global);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    ABTI_sched *p_sched = ABTI_sched_get_ptr(sched);
    ABTI_CHECK_NULL_SCHED_PTR(p_sched);
 
    /* Mark the scheduler as it is used in pool */
#ifndef ABT_CONFIG_ENABLE_VER_20_API
    ABTI_CHECK_TRUE(p_sched->used == ABTI_SCHED_NOT_USED, ABT_ERR_INV_SCHED);
#endif
    p_sched->used = ABTI_SCHED_IN_POOL;
 
#ifndef ABT_CONFIG_ENABLE_VER_20_API
    /* In both ABT_SCHED_TYPE_ULT and ABT_SCHED_TYPE_TASK cases, we use ULT-type
     * scheduler to reduce the code maintenance cost. */
#endif
    int abt_errno =
        ABTI_ythread_create_sched(p_global, p_local, p_pool, p_sched);
    if (abt_errno != ABT_SUCCESS) {
        p_sched->used = ABTI_SCHED_NOT_USED;
        ABTI_HANDLE_ERROR(abt_errno);
    }
    return ABT_SUCCESS;
}
 
int ABT_pool_get_id(ABT_pool pool, int *id)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *id = (int)p_pool->id;
    return ABT_SUCCESS;
}
 
/*****************************************************************************/
/* Private APIs                                                              */
/*****************************************************************************/
 
ABTU_ret_err int ABTI_pool_create_basic(ABT_pool_kind kind,
                                        ABT_pool_access access,
                                        ABT_bool automatic,
                                        ABTI_pool **pp_newpool)
{
    int abt_errno;
    ABTI_pool_def def;
 
    ABTI_CHECK_TRUE(access == ABT_POOL_ACCESS_PRIV ||
                        access == ABT_POOL_ACCESS_SPSC ||
                        access == ABT_POOL_ACCESS_MPSC ||
                        access == ABT_POOL_ACCESS_SPMC ||
                        access == ABT_POOL_ACCESS_MPMC,
                    ABT_ERR_INV_POOL_ACCESS);
    switch (kind) {
        case ABT_POOL_FIFO:
            abt_errno = ABTI_pool_get_fifo_def(access, &def);
            break;
        case ABT_POOL_FIFO_WAIT:
            abt_errno = ABTI_pool_get_fifo_wait_def(access, &def);
            break;
        default:
            abt_errno = ABT_ERR_INV_POOL_KIND;
            break;
    }
    ABTI_CHECK_ERROR(abt_errno);
 
    abt_errno = pool_create(&def, ABT_POOL_CONFIG_NULL, automatic, ABT_TRUE,
                            pp_newpool);
    ABTI_CHECK_ERROR(abt_errno);
    return ABT_SUCCESS;
}
 
void ABTI_pool_free(ABTI_pool *p_pool)
{
    LOG_DEBUG("[P%" PRIu64 "] freed\n", p_pool->id);
    ABT_pool h_pool = ABTI_pool_get_handle(p_pool);
    if (p_pool->p_free) {
        p_pool->p_free(h_pool);
    }
    ABTU_free(p_pool);
}
 
void ABTI_pool_print(ABTI_pool *p_pool, FILE *p_os, int indent)
{
    if (p_pool == NULL) {
        fprintf(p_os, "%*s== NULL POOL ==\n", indent, "");
    } else {
        const char *access;
 
        switch (p_pool->access) {
            case ABT_POOL_ACCESS_PRIV:
                access = "PRIV";
                break;
            case ABT_POOL_ACCESS_SPSC:
                access = "SPSC";
                break;
            case ABT_POOL_ACCESS_MPSC:
                access = "MPSC";
                break;
            case ABT_POOL_ACCESS_SPMC:
                access = "SPMC";
                break;
            case ABT_POOL_ACCESS_MPMC:
                access = "MPMC";
                break;
            default:
                access = "UNKNOWN";
                break;
        }
 
        fprintf(p_os,
                "%*s== POOL (%p) ==\n"
                "%*sid            : %" PRIu64 "\n"
                "%*saccess        : %s\n"
                "%*sautomatic     : %s\n"
                "%*snum_scheds    : %d\n"
                "%*ssize          : %zu\n"
                "%*snum_blocked   : %d\n"
                "%*sdata          : %p\n",
                indent, "", (void *)p_pool, indent, "", p_pool->id, indent, "",
                access, indent, "",
                (p_pool->automatic == ABT_TRUE) ? "TRUE" : "FALSE", indent, "",
                ABTD_atomic_acquire_load_int32(&p_pool->num_scheds), indent, "",
                ABTI_pool_get_size(p_pool), indent, "",
                ABTD_atomic_acquire_load_int32(&p_pool->num_blocked), indent,
                "", p_pool->data);
    }
    fflush(p_os);
}
 
static ABTD_atomic_uint64 g_pool_id = ABTD_ATOMIC_UINT64_STATIC_INITIALIZER(0);
void ABTI_pool_reset_id(void)
{
    ABTD_atomic_release_store_uint64(&g_pool_id, 0);
}
 
/*****************************************************************************/
/* Internal static functions                                                 */
/*****************************************************************************/
 
static inline uint64_t pool_get_new_id(void);
ABTU_ret_err static int pool_create(ABTI_pool_def *def, ABT_pool_config config,
                                    ABT_bool automatic, ABT_bool is_builtin,
                                    ABTI_pool **pp_newpool)
{
    int abt_errno;
    ABTI_pool *p_pool;
    abt_errno = ABTU_malloc(sizeof(ABTI_pool), (void **)&p_pool);
    ABTI_CHECK_ERROR(abt_errno);
 
    p_pool->access = def->access;
    p_pool->automatic = automatic;
    p_pool->is_builtin = is_builtin;
    ABTD_atomic_release_store_int32(&p_pool->num_scheds, 0);
    ABTD_atomic_release_store_int32(&p_pool->num_blocked, 0);
    p_pool->data = NULL;
 
    /* Set up the pool functions from def */
    p_pool->u_is_in_pool = def->u_is_in_pool;
    p_pool->u_create_from_thread = def->u_create_from_thread;
    p_pool->u_free = def->u_free;
    p_pool->p_init = def->p_init;
    p_pool->p_get_size = def->p_get_size;
    p_pool->p_push = def->p_push;
    p_pool->p_pop = def->p_pop;
    p_pool->p_pop_wait = def->p_pop_wait;
    p_pool->p_pop_timedwait = def->p_pop_timedwait;
    p_pool->p_remove = def->p_remove;
    p_pool->p_free = def->p_free;
    p_pool->p_print_all = def->p_print_all;
    p_pool->id = pool_get_new_id();
    LOG_DEBUG("[P%" PRIu64 "] created\n", p_pool->id);
 
    /* Configure the pool */
    if (p_pool->p_init) {
        abt_errno = p_pool->p_init(ABTI_pool_get_handle(p_pool), config);
        if (abt_errno != ABT_SUCCESS) {
            ABTU_free(p_pool);
            return abt_errno;
        }
    }
    *pp_newpool = p_pool;
    return ABT_SUCCESS;
}
 
static inline uint64_t pool_get_new_id(void)
{
    return (uint64_t)ABTD_atomic_fetch_add_uint64(&g_pool_id, 1);
}