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(
    ABT_pool_access access, const ABTI_pool_required_def *p_required_def,
    const ABTI_pool_optional_def *p_optional_def,
    const ABTI_pool_deprecated_def *p_deprecated_def,
    const ABTI_pool_old_def *p_old_def, ABTI_pool_config *p_config,
    ABT_bool def_automatic, ABT_bool is_builtin, ABTI_pool **pp_newpool);
static void
pool_create_def_from_old_def(const ABT_pool_def *p_def,
                             ABTI_pool_old_def *p_old_def,
                             ABTI_pool_required_def *p_required_def,
                             ABTI_pool_optional_def *p_optional_def,
                             ABTI_pool_deprecated_def *p_deprecated_def);
static inline int pool_pop_thread_ex(ABT_pool pool, ABT_thread *thread,
                                     ABT_pool_context pool_ctx);
static inline int pool_pop_threads_ex(ABT_pool pool, ABT_thread *threads,
                                      size_t len, size_t *num,
                                      ABT_pool_context pool_ctx);
static inline int pool_push_thread_ex(ABT_pool pool, ABT_thread thread,
                                      ABT_pool_context pool_ctx);
static inline int pool_push_threads_ex(ABT_pool pool, const ABT_thread *threads,
                                       size_t num, ABT_pool_context pool_ctx);
static inline int pool_pop_wait_thread_ex(ABT_pool pool, ABT_thread *thread,
                                          double time_secs,
                                          ABT_pool_context pool_ctx);
typedef struct {
    void *arg;
    void (*print_fn)(void *, ABT_unit);
} pool_print_thread_to_unit_arg_t;
static void pool_print_thread_to_unit(void *arg, ABT_thread thread);
typedef struct {
    void *arg;
    void (*print_fn)(void *, ABT_thread);
} pool_print_unit_to_thread_arg_t;
static void pool_print_unit_to_thread(void *arg, ABT_unit unit);
 
int ABT_pool_create(ABT_pool_user_def def, ABT_pool_config config,
                    ABT_pool *newpool)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(newpool);
    ABTI_UB_ASSERT(def);
#ifndef ABT_CONFIG_ENABLE_VER_20_API
    /* Argobots 1.x sets newpool to NULL on error. */
    *newpool = ABT_POOL_NULL;
#endif
    ABT_pool_access access;
    ABTI_pool_required_def required_def, *p_required_def;
    ABTI_pool_optional_def optional_def, *p_optional_def;
    ABTI_pool_deprecated_def deprecated_def, *p_deprecated_def;
    ABTI_pool_old_def old_def, *p_old_def;
    /* Copy def */
    if (ABTI_pool_user_def_is_new(def)) {
        /* New ABTI_pool_user_def */
        access = ABT_POOL_ACCESS_MPMC;
        ABTI_pool_user_def *p_def = ABTI_pool_user_def_get_ptr(def);
        ABTI_CHECK_NULL_POOL_USER_DEF_PTR(p_def);
        p_required_def = &p_def->required_def;
        p_optional_def = &p_def->optional_def;
        p_deprecated_def = NULL;
        p_old_def = NULL;
    } else {
        /* Old ABT_pool_def */
        ABTI_UB_ASSERT(def->u_create_from_thread);
        ABTI_UB_ASSERT(def->u_free);
        ABTI_UB_ASSERT(def->p_get_size);
        ABTI_UB_ASSERT(def->p_push);
        ABTI_UB_ASSERT(def->p_pop);
        access = def->access;
        pool_create_def_from_old_def(def, &old_def, &required_def,
                                     &optional_def, &deprecated_def);
        p_required_def = &required_def;
        p_optional_def = &optional_def;
        p_deprecated_def = &deprecated_def;
        p_old_def = &old_def;
    }
 
    ABTI_pool *p_newpool;
    ABTI_pool_config *p_config = ABTI_pool_config_get_ptr(config);
    const ABT_bool def_automatic = ABT_FALSE;
    int abt_errno =
        pool_create(access, p_required_def, p_optional_def, p_deprecated_def,
                    p_old_def, p_config, def_automatic, 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)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT_BOOL(automatic);
    ABTI_UB_ASSERT(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)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(pool);
 
    ABT_pool h_pool = *pool;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(h_pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_UB_ASSERT(ABTI_pool_is_empty(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_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(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_is_empty(ABT_pool pool, ABT_bool *is_empty)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(is_empty);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    *is_empty = ABTI_pool_is_empty(p_pool);
    return ABT_SUCCESS;
}
 
int ABT_pool_get_total_size(ABT_pool pool, size_t *size)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(size);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_get_size, ABT_ERR_POOL);
 
    *size = ABTI_pool_get_total_size(p_pool);
    return ABT_SUCCESS;
}
 
int ABT_pool_get_size(ABT_pool pool, size_t *size)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(size);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_get_size, ABT_ERR_POOL);
 
    *size = ABTI_pool_get_size(p_pool);
    return ABT_SUCCESS;
}
 
int ABT_pool_pop_thread(ABT_pool pool, ABT_thread *thread)
{
    return pool_pop_thread_ex(pool, thread, ABT_POOL_CONTEXT_OP_POOL_OTHER);
}
 
int ABT_pool_pop_thread_ex(ABT_pool pool, ABT_thread *thread,
                           ABT_pool_context pool_ctx)
{
    return pool_pop_thread_ex(pool, thread, pool_ctx);
}
 
int ABT_pool_pop_threads(ABT_pool pool, ABT_thread *threads, size_t len,
                         size_t *num)
{
    return pool_pop_threads_ex(pool, threads, len, num,
                               ABT_POOL_CONTEXT_OP_POOL_OTHER);
}
 
int ABT_pool_pop_threads_ex(ABT_pool pool, ABT_thread *threads, size_t len,
                            size_t *num, ABT_pool_context pool_ctx)
{
    return pool_pop_threads_ex(pool, threads, len, num, pool_ctx);
}
 
int ABT_pool_push_thread(ABT_pool pool, ABT_thread thread)
{
    return pool_push_thread_ex(pool, thread, ABT_POOL_CONTEXT_OP_POOL_OTHER);
}
 
int ABT_pool_push_thread_ex(ABT_pool pool, ABT_thread thread,
                            ABT_pool_context pool_ctx)
{
    return pool_push_thread_ex(pool, thread, pool_ctx);
}
 
int ABT_pool_push_threads(ABT_pool pool, const ABT_thread *threads, size_t num)
{
    return pool_push_threads_ex(pool, threads, num,
                                ABT_POOL_CONTEXT_OP_POOL_OTHER);
}
 
int ABT_pool_push_threads_ex(ABT_pool pool, const ABT_thread *threads,
                             size_t num, ABT_pool_context pool_ctx)
{
    return pool_push_threads_ex(pool, threads, num, pool_ctx);
}
 
int ABT_pool_pop_wait_thread(ABT_pool pool, ABT_thread *thread,
                             double time_secs)
{
    return pool_pop_wait_thread_ex(pool, thread, time_secs,
                                   ABT_POOL_CONTEXT_OP_POOL_OTHER);
}
 
int ABT_pool_pop_wait_thread_ex(ABT_pool pool, ABT_thread *thread,
                                double time_secs, ABT_pool_context pool_ctx)
{
    return pool_pop_wait_thread_ex(pool, thread, time_secs, pool_ctx);
}
 
int ABT_pool_print_all_threads(ABT_pool pool, void *arg,
                               void (*print_fn)(void *arg, ABT_thread))
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(print_fn);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_print_all, ABT_ERR_POOL);
 
    p_pool->optional_def.p_print_all(pool, arg, print_fn);
    return ABT_SUCCESS;
}
 
int ABT_pool_pop(ABT_pool pool, ABT_unit *p_unit)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(p_unit);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
 
    ABT_thread thread = ABTI_pool_pop(p_pool, ABT_POOL_CONTEXT_OP_POOL_OTHER);
    if (thread != ABT_THREAD_NULL) {
        ABTI_thread *p_thread = ABTI_thread_get_ptr(thread);
        *p_unit = p_thread->unit;
    } else {
        *p_unit = ABT_UNIT_NULL;
    }
    return ABT_SUCCESS;
}
 
int ABT_pool_pop_wait(ABT_pool pool, ABT_unit *p_unit, double time_secs)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(p_unit);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_pop_wait, ABT_ERR_POOL);
 
    ABT_thread thread =
        ABTI_pool_pop_wait(p_pool, time_secs, ABT_POOL_CONTEXT_OP_POOL_OTHER);
    if (thread != ABT_THREAD_NULL) {
        ABTI_thread *p_thread = ABTI_thread_get_ptr(thread);
        *p_unit = p_thread->unit;
    } else {
        *p_unit = ABT_UNIT_NULL;
    }
    return ABT_SUCCESS;
}
 
int ABT_pool_pop_timedwait(ABT_pool pool, ABT_unit *p_unit, double abstime_secs)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(p_unit);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->deprecated_def.p_pop_timedwait, ABT_ERR_POOL);
 
    ABT_thread thread = ABTI_pool_pop_timedwait(p_pool, abstime_secs);
    if (thread != ABT_THREAD_NULL) {
        ABTI_thread *p_thread = ABTI_thread_get_ptr(thread);
        *p_unit = p_thread->unit;
    } else {
        *p_unit = ABT_UNIT_NULL;
    }
    return ABT_SUCCESS;
}
 
int ABT_pool_push(ABT_pool pool, ABT_unit unit)
{
    ABTI_UB_ASSERT(ABTI_initialized());
 
    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, ABT_POOL_CONTEXT_OP_POOL_OTHER);
    return ABT_SUCCESS;
}
 
int ABT_pool_remove(ABT_pool pool, ABT_unit unit)
{
    ABTI_UB_ASSERT(ABTI_initialized());
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->deprecated_def.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_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(print_fn);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_print_all, ABT_ERR_POOL);
 
    pool_print_thread_to_unit_arg_t func_arg = { arg, print_fn };
    p_pool->optional_def.p_print_all(pool, (void *)&func_arg,
                                     pool_print_thread_to_unit);
    return ABT_SUCCESS;
}
 
int ABT_pool_set_data(ABT_pool pool, void *data)
{
    ABTI_UB_ASSERT(ABTI_initialized());
 
    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_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(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_UB_ASSERT(ABTI_initialized());
 
    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);
#else
    ABTI_UB_ASSERT(p_sched->used == ABTI_SCHED_NOT_USED);
#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_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(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_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);
 
    ABTI_pool_required_def required_def;
    ABTI_pool_optional_def optional_def;
    ABTI_pool_deprecated_def deprecated_def;
    switch (kind) {
        case ABT_POOL_FIFO:
            abt_errno = ABTI_pool_get_fifo_def(access, &required_def,
                                               &optional_def, &deprecated_def);
            break;
        case ABT_POOL_FIFO_WAIT:
            abt_errno =
                ABTI_pool_get_fifo_wait_def(access, &required_def,
                                            &optional_def, &deprecated_def);
            break;
        case ABT_POOL_RANDWS:
            abt_errno =
                ABTI_pool_get_randws_def(access, &required_def, &optional_def,
                                         &deprecated_def);
            break;
        default:
            abt_errno = ABT_ERR_INV_POOL_KIND;
            break;
    }
    ABTI_CHECK_ERROR(abt_errno);
 
    abt_errno =
        pool_create(access, &required_def, &optional_def, &deprecated_def, NULL,
                    NULL, automatic, ABT_TRUE, pp_newpool);
    ABTI_CHECK_ERROR(abt_errno);
    return ABT_SUCCESS;
}
 
void ABTI_pool_free(ABTI_pool *p_pool)
{
    ABT_pool h_pool = ABTI_pool_get_handle(p_pool);
    if (p_pool->optional_def.p_free) {
        p_pool->optional_def.p_free(h_pool);
    }
    ABTU_free(p_pool);
}
 
ABT_thread ABTI_pool_pop_timedwait(ABTI_pool *p_pool, double abstime_secs)
{
    ABTI_UB_ASSERT(p_pool->deprecated_def.p_pop_timedwait);
    ABT_unit unit =
        p_pool->deprecated_def.p_pop_timedwait(ABTI_pool_get_handle(p_pool),
                                               abstime_secs);
    if (unit == ABT_UNIT_NULL) {
        return ABT_THREAD_NULL;
    } else {
        ABTI_thread *p_thread =
            ABTI_unit_get_thread(ABTI_global_get_global(), unit);
        ABT_thread thread = ABTI_thread_get_handle(p_thread);
        LOG_DEBUG_POOL_POP(p_pool, thread);
        return thread;
    }
}
 
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"
                "%*sis_empty      : %s\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_is_empty(p_pool) ? "TRUE" : "FALSE"), indent, "",
                (p_pool->optional_def.p_get_size ? ABTI_pool_get_size(p_pool)
                                                 : 0),
                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 ABT_unit pool_create_unit_wrapper(ABT_pool pool, ABT_thread thread)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    return p_pool->old_def.u_create_from_thread(thread);
}
 
static void pool_free_unit_wrapper(ABT_pool pool, ABT_unit unit)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    p_pool->old_def.u_free(&unit);
}
 
static ABT_bool pool_is_empty_wrapper(ABT_pool pool)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    size_t size = p_pool->old_def.p_get_size(pool);
    return (size == 0) ? ABT_TRUE : ABT_FALSE;
}
 
static ABT_thread pool_pop_wrapper(ABT_pool pool, ABT_pool_context context)
{
    (void)context;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABT_unit unit = p_pool->old_def.p_pop(pool);
    if (unit != ABT_UNIT_NULL) {
        ABTI_global *p_global = ABTI_global_get_global();
        ABTI_thread *p_thread = ABTI_unit_get_thread(p_global, unit);
        return ABTI_thread_get_handle(p_thread);
    } else {
        return ABT_THREAD_NULL;
    }
}
 
static void pool_push_wrapper(ABT_pool pool, ABT_unit unit,
                              ABT_pool_context context)
{
    (void)context;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    p_pool->old_def.p_push(pool, unit);
}
 
static int pool_init_wrapper(ABT_pool pool, ABT_pool_config config)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    return p_pool->old_def.p_init(pool, config);
}
 
static void pool_free_wrapper(ABT_pool pool)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    p_pool->old_def.p_free(pool);
}
 
static size_t pool_get_size_wrapper(ABT_pool pool)
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    return p_pool->old_def.p_get_size(pool);
}
 
static ABT_thread pool_pop_wait_wrapper(ABT_pool pool, double time_secs,
                                        ABT_pool_context context)
{
    (void)context;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABT_unit unit = p_pool->old_def.p_pop_wait(pool, time_secs);
    if (unit != ABT_UNIT_NULL) {
        ABTI_global *p_global = ABTI_global_get_global();
        ABTI_thread *p_thread = ABTI_unit_get_thread(p_global, unit);
        return ABTI_thread_get_handle(p_thread);
    } else {
        return ABT_THREAD_NULL;
    }
}
 
static void pool_pop_many_wrapper(ABT_pool pool, ABT_thread *threads,
                                  size_t max_threads, size_t *num_popped,
                                  ABT_pool_context context)
{
    (void)context;
    size_t i;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    for (i = 0; i < max_threads; i++) {
        ABT_unit unit = p_pool->old_def.p_pop(pool);
        if (unit != ABT_UNIT_NULL) {
            ABTI_global *p_global = ABTI_global_get_global();
            ABTI_thread *p_thread = ABTI_unit_get_thread(p_global, unit);
            threads[i] = ABTI_thread_get_handle(p_thread);
        } else {
            break;
        }
    }
    *num_popped = i;
}
 
static void pool_push_many_wrapper(ABT_pool pool, const ABT_unit *units,
                                   size_t num_units, ABT_pool_context context)
{
    (void)context;
    size_t i;
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    for (i = 0; i < num_units; i++) {
        p_pool->old_def.p_push(pool, units[i]);
    }
}
 
static void pool_print_all_wrapper(ABT_pool pool, void *arg,
                                   void (*print_f)(void *, ABT_thread))
{
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    pool_print_unit_to_thread_arg_t wrapper_arg = { arg, print_f };
    p_pool->old_def.p_print_all(pool, (void *)&wrapper_arg,
                                pool_print_unit_to_thread);
}
 
static void
pool_create_def_from_old_def(const ABT_pool_def *p_def,
                             ABTI_pool_old_def *p_old_def,
                             ABTI_pool_required_def *p_required_def,
                             ABTI_pool_optional_def *p_optional_def,
                             ABTI_pool_deprecated_def *p_deprecated_def)
{
    /* Create p_old_def*/
    p_old_def->u_create_from_thread = p_def->u_create_from_thread;
    p_old_def->u_free = p_def->u_free;
    p_old_def->p_init = p_def->p_init;
    p_old_def->p_get_size = p_def->p_get_size;
    p_old_def->p_push = p_def->p_push;
    p_old_def->p_pop = p_def->p_pop;
#ifdef ABT_CONFIG_ENABLE_VER_20_API
    p_old_def->p_pop_wait = p_def->p_pop_wait;
#else
    p_old_def->p_pop_wait = NULL;
#endif
    p_old_def->p_free = p_def->p_free;
    p_old_def->p_print_all = p_def->p_print_all;
 
    /* Set up p_required_def */
    p_required_def->p_create_unit = pool_create_unit_wrapper;
    p_required_def->p_free_unit = pool_free_unit_wrapper;
    p_required_def->p_is_empty = pool_is_empty_wrapper;
    p_required_def->p_pop = pool_pop_wrapper;
    p_required_def->p_push = pool_push_wrapper;
 
    /* Set up p_optional_def */
    /* Must be created from ABT_pool_def */
    p_optional_def->p_get_size = pool_get_size_wrapper;
    p_optional_def->p_pop_many = pool_pop_many_wrapper;
    p_optional_def->p_push_many = pool_push_many_wrapper;
    /* Optional */
    p_optional_def->p_init = p_old_def->p_init ? pool_init_wrapper : NULL;
    p_optional_def->p_free = p_old_def->p_free ? pool_free_wrapper : NULL;
    p_optional_def->p_pop_wait =
        p_old_def->p_pop_wait ? pool_pop_wait_wrapper : NULL;
    p_optional_def->p_print_all =
        p_old_def->p_print_all ? pool_print_all_wrapper : NULL;
 
    /* Set up p_deprecated_def */
    p_deprecated_def->u_is_in_pool = p_def->u_is_in_pool;
    p_deprecated_def->p_pop_timedwait = p_def->p_pop_timedwait;
    p_deprecated_def->p_remove = p_def->p_remove;
}
 
static inline uint64_t pool_get_new_id(void);
ABTU_ret_err static int
pool_create(ABT_pool_access access,
            const ABTI_pool_required_def *p_required_def,
            const ABTI_pool_optional_def *p_optional_def,
            const ABTI_pool_deprecated_def *p_deprecated_def,
            const ABTI_pool_old_def *p_old_def, ABTI_pool_config *p_config,
            ABT_bool def_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);
 
    /* Read the config and set the configured parameter */
    ABT_bool automatic = def_automatic;
    if (p_config) {
        int automatic_val = 0;
        abt_errno =
            ABTI_pool_config_read(p_config, ABT_pool_config_automatic.key,
                                  &automatic_val);
        if (abt_errno == ABT_SUCCESS) {
            automatic = (automatic_val == 0) ? ABT_FALSE : ABT_TRUE;
        }
    }
 
    p_pool->access = 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;
    memcpy(&p_pool->required_def, p_required_def,
           sizeof(ABTI_pool_required_def));
    if (p_optional_def) {
        memcpy(&p_pool->optional_def, p_optional_def,
               sizeof(ABTI_pool_optional_def));
    } else {
        memset(&p_pool->optional_def, 0, sizeof(ABTI_pool_optional_def));
    }
    if (p_deprecated_def) {
        memcpy(&p_pool->deprecated_def, p_deprecated_def,
               sizeof(ABTI_pool_deprecated_def));
    } else {
        memset(&p_pool->deprecated_def, 0, sizeof(ABTI_pool_deprecated_def));
    }
    if (p_old_def) {
        memcpy(&p_pool->old_def, p_old_def, sizeof(ABTI_pool_old_def));
    } else {
        memset(&p_pool->old_def, 0, sizeof(ABTI_pool_old_def));
    }
    p_pool->id = pool_get_new_id();
 
    /* Configure the pool */
    if (p_pool->optional_def.p_init) {
        ABT_pool_config config = ABTI_pool_config_get_handle(p_config);
        abt_errno =
            p_pool->optional_def.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);
}
 
static inline int pool_pop_thread_ex(ABT_pool pool, ABT_thread *thread,
                                     ABT_pool_context pool_ctx)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(thread);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    *thread = ABTI_pool_pop(p_pool, pool_ctx);
    return ABT_SUCCESS;
}
 
static inline int pool_pop_threads_ex(ABT_pool pool, ABT_thread *threads,
                                      size_t len, size_t *num,
                                      ABT_pool_context pool_ctx)
{
    ABTI_STATIC_ASSERT(sizeof(ABT_unit) == sizeof(ABT_thread));
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(threads || len == 0);
    ABTI_UB_ASSERT(num);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_pop_many, ABT_ERR_POOL);
 
    if (len > 0) {
        ABTI_pool_pop_many(p_pool, threads, len, num, pool_ctx);
    }
    return ABT_SUCCESS;
}
 
static inline int pool_push_thread_ex(ABT_pool pool, ABT_thread thread,
                                      ABT_pool_context pool_ctx)
{
    ABTI_UB_ASSERT(ABTI_initialized());
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_thread *p_thread = ABTI_thread_get_ptr(thread);
 
    if (p_thread) {
        ABTI_global *p_global = ABTI_global_get_global();
        int abt_errno =
            ABTI_thread_set_associated_pool(p_global, p_thread, p_pool);
        ABTI_CHECK_ERROR(abt_errno);
        ABTI_pool_push(p_pool, p_thread->unit, pool_ctx);
    }
    return ABT_SUCCESS;
}
 
static inline int pool_push_threads_ex(ABT_pool pool, const ABT_thread *threads,
                                       size_t num, ABT_pool_context pool_ctx)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(threads || num == 0);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_push_many, ABT_ERR_POOL);
 
    if (num > 0) {
        ABTI_global *p_global = ABTI_global_get_global();
        int abt_errno;
        ABT_unit *push_units, push_units_buffer[64];
        if (num > sizeof(push_units_buffer) / sizeof(push_units_buffer[0])) {
            abt_errno =
                ABTU_malloc(sizeof(ABT_unit) * num, (void **)&push_units);
            ABTI_CHECK_ERROR(abt_errno);
        } else {
            push_units = push_units_buffer;
        }
 
        size_t i, num_units = 0;
        for (i = 0; i < num; i++) {
            /* FIXME: the following can break the intermediate mapping if an
             * error happens. */
            ABTI_thread *p_thread = ABTI_thread_get_ptr(threads[i]);
            if (p_thread) {
                abt_errno =
                    ABTI_thread_set_associated_pool(p_global, p_thread, p_pool);
                if (abt_errno != ABT_SUCCESS) {
                    if (push_units != push_units_buffer)
                        ABTU_free(push_units);
                    ABTI_HANDLE_ERROR(abt_errno);
                }
                push_units[num_units++] = p_thread->unit;
            }
        }
        if (num_units > 0) {
            ABTI_pool_push_many(p_pool, push_units, num_units, pool_ctx);
        }
        if (push_units != push_units_buffer)
            ABTU_free(push_units);
    }
    return ABT_SUCCESS;
}
 
static inline int pool_pop_wait_thread_ex(ABT_pool pool, ABT_thread *thread,
                                          double time_secs,
                                          ABT_pool_context pool_ctx)
{
    ABTI_UB_ASSERT(ABTI_initialized());
    ABTI_UB_ASSERT(thread);
 
    ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
    ABTI_CHECK_NULL_POOL_PTR(p_pool);
    ABTI_CHECK_TRUE(p_pool->optional_def.p_pop_wait, ABT_ERR_POOL);
 
    *thread = ABTI_pool_pop_wait(p_pool, time_secs, pool_ctx);
    return ABT_SUCCESS;
}
 
static void pool_print_thread_to_unit(void *arg, ABT_thread thread)
{
    pool_print_thread_to_unit_arg_t *p_arg =
        (pool_print_thread_to_unit_arg_t *)arg;
    ABTI_thread *p_thread = ABTI_thread_get_ptr(thread);
    p_arg->print_fn(p_arg->arg, p_thread->unit);
}
 
static void pool_print_unit_to_thread(void *arg, ABT_unit unit)
{
    pool_print_unit_to_thread_arg_t *p_arg =
        (pool_print_unit_to_thread_arg_t *)arg;
    ABTI_global *p_global = ABTI_global_get_global();
    ABTI_thread *p_thread = ABTI_unit_get_thread(p_global, unit);
    ABT_thread thread = ABTI_thread_get_handle(p_thread);
    p_arg->print_fn(p_arg->arg, thread);
}