futures.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"

int ABT_future_create(uint32_t compartments, void (*cb_func)(void **arg),
                      ABT_future *newfuture)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_future *p_future;

    p_future = (ABTI_future *)ABTU_malloc(sizeof(ABTI_future));
    ABTI_spinlock_clear(&p_future->lock);
    ABTD_atomic_relaxed_store_uint32(&p_future->counter, 0);
    p_future->compartments = compartments;
    p_future->array = ABTU_malloc(compartments * sizeof(void *));
    p_future->p_callback = cb_func;
    p_future->p_head = NULL;
    p_future->p_tail = NULL;

    *newfuture = ABTI_future_get_handle(p_future);

    return abt_errno;
}

int ABT_future_free(ABT_future *future)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_future *p_future = ABTI_future_get_ptr(*future);
    ABTI_CHECK_NULL_FUTURE_PTR(p_future);

    /* The lock needs to be acquired to safely free the future structure.
     * However, we do not have to unlock it because the entire structure is
     * freed here. */
    ABTI_spinlock_acquire(&p_future->lock);

    ABTU_free(p_future->array);
    ABTU_free(p_future);

    *future = ABT_FUTURE_NULL;

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_future_wait(ABT_future future)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_future *p_future = ABTI_future_get_ptr(future);
    ABTI_CHECK_NULL_FUTURE_PTR(p_future);

    ABTI_spinlock_acquire(&p_future->lock);
    if (ABTD_atomic_relaxed_load_uint32(&p_future->counter) <
        p_future->compartments) {
        ABTI_thread *p_current;
        ABTI_unit *p_unit;
        ABT_unit_type type;
        ABTD_atomic_int32 ext_signal = ABTD_ATOMIC_INT32_STATIC_INITIALIZER(0);

        if (p_local != NULL) {
            p_current = p_local->p_thread;
#ifndef ABT_CONFIG_DISABLE_ERROR_CHECK
            if (p_current == NULL) {
                abt_errno = ABT_ERR_FUTURE;
                ABTI_spinlock_release(&p_future->lock);
                goto fn_fail;
            }
#endif
            type = ABT_UNIT_TYPE_THREAD;
            p_unit = &p_current->unit_def;
            p_unit->handle.thread = ABTI_thread_get_handle(p_current);
            p_unit->type = type;
        } else {
            /* external thread */
            type = ABT_UNIT_TYPE_EXT;
            p_unit = (ABTI_unit *)ABTU_calloc(1, sizeof(ABTI_unit));
            /* Check size if ext_signal can be stored in p_unit->handle.thread.
             */
            ABTI_STATIC_ASSERT(sizeof(ext_signal) <=
                               sizeof(p_unit->handle.thread));
            p_unit->handle.thread = (ABT_thread)&ext_signal;
            p_unit->type = type;
        }

        p_unit->p_next = NULL;
        if (p_future->p_head == NULL) {
            p_future->p_head = p_unit;
            p_future->p_tail = p_unit;
        } else {
            p_future->p_tail->p_next = p_unit;
            p_future->p_tail = p_unit;
        }

        if (type == ABT_UNIT_TYPE_THREAD) {
            ABTI_thread_set_blocked(p_current);

            ABTI_spinlock_release(&p_future->lock);

            /* Suspend the current ULT */
            ABTI_thread_suspend(&p_local, p_current);

        } else {
            ABTI_spinlock_release(&p_future->lock);

            /* External thread is waiting here polling ext_signal. */
            /* FIXME: need a better implementation */
            while (!ABTD_atomic_acquire_load_int32(&ext_signal))
                ;
            ABTU_free(p_unit);
        }
    } else {
        ABTI_spinlock_release(&p_future->lock);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_future_test(ABT_future future, ABT_bool *flag)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_future *p_future = ABTI_future_get_ptr(future);
    ABTI_CHECK_NULL_FUTURE_PTR(p_future);

    uint32_t counter = ABTD_atomic_acquire_load_uint32(&p_future->counter);
    *flag = (counter == p_future->compartments) ? ABT_TRUE : ABT_FALSE;

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_future_set(ABT_future future, void *value)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_future *p_future = ABTI_future_get_ptr(future);
    ABTI_CHECK_NULL_FUTURE_PTR(p_future);

    ABTI_spinlock_acquire(&p_future->lock);

    int counter = ABTD_atomic_relaxed_load_uint32(&p_future->counter);
#ifndef ABT_CONFIG_DISABLE_ERROR_CHECK
    if (counter >= p_future->compartments) {
        abt_errno = ABT_ERR_FUTURE;
        ABTI_spinlock_release(&p_future->lock);
        goto fn_fail;
    }
#endif
    p_future->array[counter] = value;
    counter++;
    ABTD_atomic_release_store_uint32(&p_future->counter, counter);

    if (counter == p_future->compartments) {
        if (p_future->p_callback != NULL)
            (*p_future->p_callback)(p_future->array);

        if (p_future->p_head == NULL) {
            ABTI_spinlock_release(&p_future->lock);
            goto fn_exit;
        }

        /* Wake up all waiting ULTs */
        ABTI_unit *p_head = p_future->p_head;
        ABTI_unit *p_unit = p_head;
        while (1) {
            ABTI_unit *p_next = p_unit->p_next;
            ABT_unit_type type = p_unit->type;

            p_unit->p_next = NULL;

            if (type == ABT_UNIT_TYPE_THREAD) {
                ABTI_thread *p_thread =
                    ABTI_thread_get_ptr(p_unit->handle.thread);
                ABTI_thread_set_ready(p_local, p_thread);
            } else {
                /* When the head is an external thread */
                ABTD_atomic_int32 *p_ext_signal =
                    (ABTD_atomic_int32 *)p_unit->handle.thread;
                ABTD_atomic_release_store_int32(p_ext_signal, 1);
            }

            /* Next ULT */
            if (p_next != NULL) {
                p_unit = p_next;
            } else {
                break;
            }
        }
        p_future->p_head = NULL;
        p_future->p_tail = NULL;
    }

    ABTI_spinlock_release(&p_future->lock);

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_future_reset(ABT_future future)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_future *p_future = ABTI_future_get_ptr(future);
    ABTI_CHECK_NULL_FUTURE_PTR(p_future);

    ABTI_spinlock_acquire(&p_future->lock);
    ABTD_atomic_release_store_uint32(&p_future->counter, 0);
    ABTI_spinlock_release(&p_future->lock);

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}