eventual.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_eventual_create(int nbytes, ABT_eventual *neweventual)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_eventual *p_eventual;

    p_eventual = (ABTI_eventual *)ABTU_malloc(sizeof(ABTI_eventual));
    ABTI_spinlock_clear(&p_eventual->lock);
    p_eventual->ready = ABT_FALSE;
    p_eventual->nbytes = nbytes;
    p_eventual->value = (nbytes == 0) ? NULL : ABTU_malloc(nbytes);
    p_eventual->p_head = NULL;
    p_eventual->p_tail = NULL;

    *neweventual = ABTI_eventual_get_handle(p_eventual);

    return abt_errno;
}

int ABT_eventual_free(ABT_eventual *eventual)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_eventual *p_eventual = ABTI_eventual_get_ptr(*eventual);
    ABTI_CHECK_NULL_EVENTUAL_PTR(p_eventual);

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

    if (p_eventual->value)
        ABTU_free(p_eventual->value);
    ABTU_free(p_eventual);

    *eventual = ABT_EVENTUAL_NULL;

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_eventual_wait(ABT_eventual eventual, void **value)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_eventual *p_eventual = ABTI_eventual_get_ptr(eventual);
    ABTI_CHECK_NULL_EVENTUAL_PTR(p_eventual);

    ABTI_spinlock_acquire(&p_eventual->lock);
    if (p_eventual->ready == ABT_FALSE) {
        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;
            ABTI_CHECK_TRUE(p_current != NULL, ABT_ERR_EVENTUAL);

            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_eventual->p_head == NULL) {
            p_eventual->p_head = p_unit;
            p_eventual->p_tail = p_unit;
        } else {
            p_eventual->p_tail->p_next = p_unit;
            p_eventual->p_tail = p_unit;
        }

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

            ABTI_spinlock_release(&p_eventual->lock);

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

        } else {
            ABTI_spinlock_release(&p_eventual->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_eventual->lock);
    }
    if (value)
        *value = p_eventual->value;

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_eventual_test(ABT_eventual eventual, void **value, int *is_ready)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_eventual *p_eventual = ABTI_eventual_get_ptr(eventual);
    ABTI_CHECK_NULL_EVENTUAL_PTR(p_eventual);
    int flag = ABT_FALSE;

    ABTI_spinlock_acquire(&p_eventual->lock);
    if (p_eventual->ready != ABT_FALSE) {
        if (value)
            *value = p_eventual->value;
        flag = ABT_TRUE;
    }
    ABTI_spinlock_release(&p_eventual->lock);

    *is_ready = flag;

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_eventual_set(ABT_eventual eventual, void *value, int nbytes)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_eventual *p_eventual = ABTI_eventual_get_ptr(eventual);
    ABTI_CHECK_NULL_EVENTUAL_PTR(p_eventual);
    ABTI_CHECK_TRUE(nbytes <= p_eventual->nbytes, ABT_ERR_INV_EVENTUAL);

    ABTI_spinlock_acquire(&p_eventual->lock);

    p_eventual->ready = ABT_TRUE;
    if (p_eventual->value)
        memcpy(p_eventual->value, value, nbytes);

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

    /* Wake up all waiting ULTs */
    ABTI_unit *p_head = p_eventual->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_eventual->p_head = NULL;
    p_eventual->p_tail = NULL;

    ABTI_spinlock_release(&p_eventual->lock);

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_eventual_reset(ABT_eventual eventual)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_eventual *p_eventual = ABTI_eventual_get_ptr(eventual);
    ABTI_CHECK_NULL_EVENTUAL_PTR(p_eventual);

    ABTI_spinlock_acquire(&p_eventual->lock);
    p_eventual->ready = ABT_FALSE;
    ABTI_spinlock_release(&p_eventual->lock);

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}