mutex.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"
#include "abti_thread_htable.h"

int ABT_mutex_create(ABT_mutex *newmutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_mutex *p_newmutex;

    p_newmutex = (ABTI_mutex *)ABTU_calloc(1, sizeof(ABTI_mutex));
    ABTI_mutex_init(p_newmutex);

    /* Return value */
    *newmutex = ABTI_mutex_get_handle(p_newmutex);

    return abt_errno;
}

int ABT_mutex_create_with_attr(ABT_mutex_attr attr, ABT_mutex *newmutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_mutex_attr *p_attr = ABTI_mutex_attr_get_ptr(attr);
    ABTI_CHECK_NULL_MUTEX_ATTR_PTR(p_attr);
    ABTI_mutex *p_newmutex;

    p_newmutex = (ABTI_mutex *)ABTU_malloc(sizeof(ABTI_mutex));
    ABTI_mutex_init(p_newmutex);
    ABTI_mutex_attr_copy(&p_newmutex->attr, p_attr);

    /* Return value */
    *newmutex = ABTI_mutex_get_handle(p_newmutex);

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_free(ABT_mutex *mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABT_mutex h_mutex = *mutex;
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(h_mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    ABTI_mutex_fini(p_mutex);
    ABTU_free(p_mutex);

    /* Return value */
    *mutex = ABT_MUTEX_NULL;

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_lock(ABT_mutex mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    if (p_mutex->attr.attrs == ABTI_MUTEX_ATTR_NONE) {
        /* default attributes */
        ABTI_mutex_lock(&p_local, p_mutex);

    } else if (p_mutex->attr.attrs & ABTI_MUTEX_ATTR_RECURSIVE) {
        /* recursive mutex */
        ABTI_unit_id self_id = ABTI_self_get_unit_id(p_local);
        if (self_id != p_mutex->attr.owner_id) {
            ABTI_mutex_lock(&p_local, p_mutex);
            p_mutex->attr.owner_id = self_id;
            ABTI_ASSERT(p_mutex->attr.nesting_cnt == 0);
        } else {
            p_mutex->attr.nesting_cnt++;
        }

    } else {
        /* unknown attributes */
        ABTI_mutex_lock(&p_local, p_mutex);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

static inline void ABTI_mutex_lock_low(ABTI_local **pp_local,
                                       ABTI_mutex *p_mutex)
{
#ifdef ABT_CONFIG_USE_SIMPLE_MUTEX
    ABTI_local *p_local = *pp_local;
    ABT_unit_type type = ABTI_self_get_type(p_local);
    if (type == ABT_UNIT_TYPE_THREAD) {
        LOG_EVENT("%p: lock_low - try\n", p_mutex);
        while (!ABTD_atomic_bool_cas_weak_uint32(&p_mutex->val, 0, 1)) {
            ABTI_thread_yield(pp_local, p_local->p_thread);
            p_local = *pp_local;
        }
        LOG_EVENT("%p: lock_low - acquired\n", p_mutex);
    } else {
        ABTI_mutex_spinlock(p_mutex);
    }
#else
    int abt_errno;
    ABTI_local *p_local = *pp_local;
    ABT_unit_type type = ABTI_self_get_type(p_local);

    /* Only ULTs can yield when the mutex has been locked. For others,
     * just call mutex_spinlock. */
    if (type == ABT_UNIT_TYPE_THREAD) {
        LOG_EVENT("%p: lock_low - try\n", p_mutex);
        int c;

        /* If other ULTs associated with the same ES are waiting on the
         * low-mutex queue, we give the header ULT a chance to try to get
         * the mutex by context switching to it. */
        ABTI_thread_htable *p_htable = p_mutex->p_htable;
        ABTI_thread *p_self = p_local->p_thread;
        ABTI_xstream *p_xstream = p_self->p_last_xstream;
        int rank = (int)p_xstream->rank;
        ABTI_thread_queue *p_queue = &p_htable->queue[rank];
        if (p_queue->low_num_threads > 0) {
            ABT_bool ret = ABTI_thread_htable_switch_low(pp_local, p_queue,
                                                         p_self, p_htable);
            if (ret == ABT_TRUE) {
                /* This ULT became a waiter in the mutex queue */
                goto check_handover;
            }
        }

        if ((c = ABTD_atomic_val_cas_strong_uint32(&p_mutex->val, 0, 1)) != 0) {
            if (c != 2) {
                c = ABTD_atomic_exchange_uint32(&p_mutex->val, 2);
            }
            while (c != 0) {
                ABTI_mutex_wait_low(pp_local, p_mutex, 2);

            check_handover:
                /* If the mutex has been handed over to the current ULT from
                 * other ULT on the same ES, we don't need to change the mutex
                 * state. */
                if (p_mutex->p_handover) {
                    if (p_self == p_mutex->p_handover) {
                        p_mutex->p_handover = NULL;
                        ABTD_atomic_release_store_uint32(&p_mutex->val, 2);

                        /* Push the previous ULT to its pool */
                        ABTI_thread *p_giver = p_mutex->p_giver;
                        ABTD_atomic_release_store_int(&p_giver->state,
                                                      ABT_THREAD_STATE_READY);
                        ABTI_POOL_PUSH(p_giver->p_pool, p_giver->unit,
                                       ABTI_self_get_native_thread_id(
                                           *pp_local));
                        break;
                    }
                }

                c = ABTD_atomic_exchange_uint32(&p_mutex->val, 2);
            }
        }
        LOG_EVENT("%p: lock_low - acquired\n", p_mutex);
    } else {
        ABTI_mutex_spinlock(p_mutex);
    }

fn_exit:
    return;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
#endif
}

int ABT_mutex_lock_low(ABT_mutex mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    if (p_mutex->attr.attrs == ABTI_MUTEX_ATTR_NONE) {
        /* default attributes */
        ABTI_mutex_lock_low(&p_local, p_mutex);

    } else if (p_mutex->attr.attrs & ABTI_MUTEX_ATTR_RECURSIVE) {
        /* recursive mutex */
        ABTI_unit_id self_id = ABTI_self_get_unit_id(p_local);
        if (self_id != p_mutex->attr.owner_id) {
            ABTI_mutex_lock_low(&p_local, p_mutex);
            p_mutex->attr.owner_id = self_id;
            ABTI_ASSERT(p_mutex->attr.nesting_cnt == 0);
        } else {
            p_mutex->attr.nesting_cnt++;
        }

    } else {
        /* unknown attributes */
        ABTI_mutex_lock_low(&p_local, p_mutex);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_lock_high(ABT_mutex mutex)
{
    return ABT_mutex_lock(mutex);
}

int ABT_mutex_trylock(ABT_mutex mutex)
{
    int abt_errno;
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    if (p_mutex->attr.attrs == ABTI_MUTEX_ATTR_NONE) {
        /* default attributes */
        abt_errno = ABTI_mutex_trylock(p_mutex);

    } else if (p_mutex->attr.attrs & ABTI_MUTEX_ATTR_RECURSIVE) {
        /* recursive mutex */
        ABTI_local *p_local = ABTI_local_get_local();
        ABTI_unit_id self_id = ABTI_self_get_unit_id(p_local);
        if (self_id != p_mutex->attr.owner_id) {
            abt_errno = ABTI_mutex_trylock(p_mutex);
            if (abt_errno == ABT_SUCCESS) {
                p_mutex->attr.owner_id = self_id;
                ABTI_ASSERT(p_mutex->attr.nesting_cnt == 0);
            }
        } else {
            p_mutex->attr.nesting_cnt++;
            abt_errno = ABT_SUCCESS;
        }

    } else {
        /* unknown attributes */
        abt_errno = ABTI_mutex_trylock(p_mutex);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_spinlock(ABT_mutex mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    if (p_mutex->attr.attrs == ABTI_MUTEX_ATTR_NONE) {
        /* default attributes */
        ABTI_mutex_spinlock(p_mutex);

    } else if (p_mutex->attr.attrs & ABTI_MUTEX_ATTR_RECURSIVE) {
        /* recursive mutex */
        ABTI_local *p_local = ABTI_local_get_local();
        ABTI_unit_id self_id = ABTI_self_get_unit_id(p_local);
        if (self_id != p_mutex->attr.owner_id) {
            ABTI_mutex_spinlock(p_mutex);
            p_mutex->attr.owner_id = self_id;
            ABTI_ASSERT(p_mutex->attr.nesting_cnt == 0);
        } else {
            p_mutex->attr.nesting_cnt++;
        }

    } else {
        /* unknown attributes */
        ABTI_mutex_spinlock(p_mutex);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_unlock(ABT_mutex mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    if (p_mutex->attr.attrs == ABTI_MUTEX_ATTR_NONE) {
        /* default attributes */
        ABTI_mutex_unlock(p_local, p_mutex);

    } else if (p_mutex->attr.attrs & ABTI_MUTEX_ATTR_RECURSIVE) {
        /* recursive mutex */
        ABTI_CHECK_TRUE(ABTI_self_get_unit_id(p_local) ==
                            p_mutex->attr.owner_id,
                        ABT_ERR_INV_THREAD);
        if (p_mutex->attr.nesting_cnt == 0) {
            p_mutex->attr.owner_id = 0;
            ABTI_mutex_unlock(p_local, p_mutex);
        } else {
            p_mutex->attr.nesting_cnt--;
        }

    } else {
        /* unknown attributes */
        ABTI_mutex_unlock(p_local, p_mutex);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

/* Hand over the mutex to other ULT on the same ES */
static inline int ABTI_mutex_unlock_se(ABTI_local **pp_local,
                                       ABTI_mutex *p_mutex)
{
    int abt_errno = ABT_SUCCESS;

#ifdef ABT_CONFIG_USE_SIMPLE_MUTEX
    ABTD_atomic_release_store_uint32(&p_mutex->val, 0);
    LOG_EVENT("%p: unlock_se\n", p_mutex);
    ABTI_local *p_local = *pp_local;
#ifndef ABT_CONFIG_DISABLE_EXT_THREAD
    if (ABTI_self_get_type(p_local) == ABT_UNIT_TYPE_THREAD)
        ABTI_thread_yield(pp_local, p_local->p_thread);
#else
    ABTI_thread_yield(pp_local, p_local->p_thread);
#endif
#else
    int i;
    ABTI_xstream *p_xstream;
    ABTI_thread *p_next = NULL;
    ABTI_thread *p_thread;
    ABTI_thread_queue *p_queue;

    /* Unlock the mutex */
    /* If p_mutex->val is 1 before decreasing it, it means there is no any
     * waiter in the mutex queue.  We can just return. */
    if (ABTD_atomic_fetch_sub_uint32(&p_mutex->val, 1) == 1) {
        LOG_EVENT("%p: unlock_se\n", p_mutex);
#ifndef ABT_CONFIG_DISABLE_EXT_THREAD
        if (ABTI_self_get_type(*pp_local) == ABT_UNIT_TYPE_THREAD)
            ABTI_thread_yield(pp_local, (*pp_local)->p_thread);
#else
        ABTI_thread_yield(pp_local, (*pp_local)->p_thread);
#endif
        return abt_errno;
    }

    /* There are ULTs waiting in the mutex queue */
    ABTI_thread_htable *p_htable = p_mutex->p_htable;

    p_thread = (*pp_local)->p_thread;
    p_xstream = p_thread->p_last_xstream;
    ABTI_ASSERT(p_xstream == (*pp_local)->p_xstream);
    i = (int)p_xstream->rank;
    p_queue = &p_htable->queue[i];

check_cond:
    /* Check whether the mutex handover is possible */
    if (p_queue->num_handovers >= p_mutex->attr.max_handovers) {
        ABTD_atomic_release_store_uint32(&p_mutex->val, 0); /* Unlock */
        LOG_EVENT("%p: unlock_se\n", p_mutex);
        ABTI_mutex_wake_de(*pp_local, p_mutex);
        p_queue->num_handovers = 0;
        ABTI_thread_yield(pp_local, p_thread);
        return abt_errno;
    }

    /* Hand over the mutex to high-priority ULTs */
    if (p_queue->num_threads <= 1) {
        if (p_htable->h_list != NULL) {
            ABTD_atomic_release_store_uint32(&p_mutex->val, 0); /* Unlock */
            LOG_EVENT("%p: unlock_se\n", p_mutex);
            ABTI_mutex_wake_de(*pp_local, p_mutex);
            ABTI_thread_yield(pp_local, p_thread);
            return abt_errno;
        }
    } else {
        p_next = ABTI_thread_htable_pop(p_htable, p_queue);
        if (p_next == NULL)
            goto check_cond;
        else
            goto handover;
    }

    /* When we don't have high-priority ULTs and other ESs don't either,
     * we hand over the mutex to low-priority ULTs. */
    if (p_queue->low_num_threads <= 1) {
        ABTD_atomic_release_store_uint32(&p_mutex->val, 0); /* Unlock */
        LOG_EVENT("%p: unlock_se\n", p_mutex);
        ABTI_mutex_wake_de(*pp_local, p_mutex);
        ABTI_thread_yield(pp_local, p_thread);
        return abt_errno;
    } else {
        p_next = ABTI_thread_htable_pop_low(p_htable, p_queue);
        if (p_next == NULL)
            goto check_cond;
    }

handover:
    /* We don't push p_thread to the pool. Instead, we will yield_to p_thread
     * directly at the end of this function. */
    p_queue->num_handovers++;

    /* We are handing over the mutex */
    p_mutex->p_handover = p_next;
    p_mutex->p_giver = p_thread;

    LOG_EVENT("%p: handover -> U%" PRIu64 "\n", p_mutex,
              ABTI_thread_get_id(p_next));

    /* yield_to the next ULT */
    while (ABTD_atomic_acquire_load_uint32(&p_next->request) &
           ABTI_THREAD_REQ_BLOCK)
        ;
    ABTI_pool_dec_num_blocked(p_next->p_pool);
    ABTD_atomic_release_store_int(&p_next->state, ABT_THREAD_STATE_RUNNING);
    ABTI_thread_context_switch_thread_to_thread(pp_local, p_thread, p_next);
#endif

    return abt_errno;
}

int ABT_mutex_unlock_se(ABT_mutex mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    if (p_mutex->attr.attrs == ABTI_MUTEX_ATTR_NONE) {
        /* default attributes */
        ABTI_mutex_unlock_se(&p_local, p_mutex);

    } else if (p_mutex->attr.attrs & ABTI_MUTEX_ATTR_RECURSIVE) {
        /* recursive mutex */
        ABTI_CHECK_TRUE(ABTI_self_get_unit_id(p_local) ==
                            p_mutex->attr.owner_id,
                        ABT_ERR_INV_THREAD);
        if (p_mutex->attr.nesting_cnt == 0) {
            p_mutex->attr.owner_id = 0;
            ABTI_mutex_unlock_se(&p_local, p_mutex);
        } else {
            p_mutex->attr.nesting_cnt--;
        }

    } else {
        /* unknown attributes */
        ABTI_mutex_unlock_se(&p_local, p_mutex);
    }

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_unlock_de(ABT_mutex mutex)
{
    int abt_errno = ABT_SUCCESS;
    ABTI_local *p_local = ABTI_local_get_local();
    ABTI_mutex *p_mutex = ABTI_mutex_get_ptr(mutex);
    ABTI_CHECK_NULL_MUTEX_PTR(p_mutex);

    ABTI_mutex_unlock(p_local, p_mutex);

fn_exit:
    return abt_errno;

fn_fail:
    HANDLE_ERROR_FUNC_WITH_CODE(abt_errno);
    goto fn_exit;
}

int ABT_mutex_equal(ABT_mutex mutex1, ABT_mutex mutex2, ABT_bool *result)
{
    ABTI_mutex *p_mutex1 = ABTI_mutex_get_ptr(mutex1);
    ABTI_mutex *p_mutex2 = ABTI_mutex_get_ptr(mutex2);
    *result = ABTI_mutex_equal(p_mutex1, p_mutex2);
    return ABT_SUCCESS;
}

void ABTI_mutex_wait(ABTI_local **pp_local, ABTI_mutex *p_mutex, int val)
{
    ABTI_local *p_local = *pp_local;
    ABTI_thread_htable *p_htable = p_mutex->p_htable;
    ABTI_thread *p_self = p_local->p_thread;
    ABTI_xstream *p_xstream = p_self->p_last_xstream;

    int rank = (int)p_xstream->rank;
    ABTI_ASSERT(rank < p_htable->num_rows);
    ABTI_thread_queue *p_queue = &p_htable->queue[rank];

    ABTI_THREAD_HTABLE_LOCK(p_htable->mutex);

    if (ABTD_atomic_acquire_load_uint32(&p_mutex->val) != val) {
        ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);
        return;
    }

    if (p_queue->p_h_next == NULL) {
        ABTI_thread_htable_add_h_node(p_htable, p_queue);
    }

    /* Change the ULT's state to BLOCKED */
    ABTI_thread_set_blocked(p_self);

    /* Push the current ULT to the queue */
    ABTI_thread_htable_push(p_htable, rank, p_self);

    /* Unlock */
    ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);

    /* Suspend the current ULT */
    ABTI_thread_suspend(pp_local, p_self);
}

void ABTI_mutex_wait_low(ABTI_local **pp_local, ABTI_mutex *p_mutex, int val)
{
    ABTI_local *p_local = *pp_local;
    ABTI_thread_htable *p_htable = p_mutex->p_htable;
    ABTI_thread *p_self = p_local->p_thread;
    ABTI_xstream *p_xstream = p_self->p_last_xstream;

    int rank = (int)p_xstream->rank;
    ABTI_ASSERT(rank < p_htable->num_rows);
    ABTI_thread_queue *p_queue = &p_htable->queue[rank];

    ABTI_THREAD_HTABLE_LOCK(p_htable->mutex);

    if (ABTD_atomic_acquire_load_uint32(&p_mutex->val) != val) {
        ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);
        return;
    }

    if (p_queue->p_l_next == NULL) {
        ABTI_thread_htable_add_l_node(p_htable, p_queue);
    }

    /* Change the ULT's state to BLOCKED */
    ABTI_thread_set_blocked(p_self);

    /* Push the current ULT to the queue */
    ABTI_thread_htable_push_low(p_htable, rank, p_self);

    /* Unlock */
    ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);

    /* Suspend the current ULT */
    ABTI_thread_suspend(pp_local, p_self);
}

void ABTI_mutex_wake_de(ABTI_local *p_local, ABTI_mutex *p_mutex)
{
    int n;
    ABTI_thread *p_thread;
    ABTI_thread_htable *p_htable = p_mutex->p_htable;
    int num = p_mutex->attr.max_wakeups;
    ABTI_thread_queue *p_start, *p_curr;

    /* Wake up num ULTs in a round-robin manner */
    for (n = 0; n < num; n++) {
        p_thread = NULL;

        ABTI_THREAD_HTABLE_LOCK(p_htable->mutex);

        if (ABTD_atomic_acquire_load_uint32(&p_htable->num_elems) == 0) {
            ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);
            break;
        }

        /* Wake up the high-priority ULTs */
        p_start = p_htable->h_list;
        for (p_curr = p_start; p_curr;) {
            p_thread = ABTI_thread_htable_pop(p_htable, p_curr);
            if (p_curr->num_threads == 0) {
                ABTI_thread_htable_del_h_head(p_htable);
            } else {
                p_htable->h_list = p_curr->p_h_next;
            }
            if (p_thread != NULL)
                goto done;
            p_curr = p_htable->h_list;
            if (p_curr == p_start)
                break;
        }

        /* Wake up the low-priority ULTs */
        p_start = p_htable->l_list;
        for (p_curr = p_start; p_curr;) {
            p_thread = ABTI_thread_htable_pop_low(p_htable, p_curr);
            if (p_curr->low_num_threads == 0) {
                ABTI_thread_htable_del_l_head(p_htable);
            } else {
                p_htable->l_list = p_curr->p_l_next;
            }
            if (p_thread != NULL)
                goto done;
            p_curr = p_htable->l_list;
            if (p_curr == p_start)
                break;
        }

        /* Nothing to wake up */
        ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);
        LOG_EVENT("%p: nothing to wake up\n", p_mutex);
        break;

    done:
        ABTI_THREAD_HTABLE_UNLOCK(p_htable->mutex);

        /* Push p_thread to the scheduler's pool */
        LOG_EVENT("%p: wake up U%" PRIu64 ":E%d\n", p_mutex,
                  ABTI_thread_get_id(p_thread),
                  ABTI_thread_get_xstream_rank(p_thread));
        ABTI_thread_set_ready(p_local, p_thread);
    }
}