thread.h

Go to the documentation of this file.

/*
 * CDDL HEADER START
 *
 * This file and its contents are supplied under the terms of the
 * Common Development and Distribution License ("CDDL"), version 1.0.
 * You may only use this file in accordance with the terms of version
 * 1.0 of the CDDL.
 *
 * A full copy of the text of the CDDL should have accompanied this
 * source.  A copy of the CDDL is also available via the Internet at
 * http://www.illumos.org/license/CDDL.
 *
 * CDDL HEADER END
*/
/*
 * Copyright 2021 Saso Kiselkov. All rights reserved.
 */
 
#ifndef _ACF_UTILS_THREAD_H_
#define _ACF_UTILS_THREAD_H_
 
#include <errno.h>
#include <stddef.h>
#include <stdlib.h>
#include <string.h>
 
#if APL || LIN
#include <pthread.h>
#include <stdint.h>
#include <time.h>
#else   /* !APL && !LIN */
#include <windows.h>
#endif  /* !APL && !LIN */
 
#if __STDC_VERSION__ >= 201112L && !defined(__STDC_NO_ATOMICS__) && \
    (!defined(__GNUC__) || __GNUC__ >= 7 || defined(__clang__))
#define _USE_STDATOMICS
#endif
 
#ifdef  _USE_STDATOMICS
#include <stdatomic.h>
#elif   APL
#include <libkern/OSAtomic.h>
#endif
 
#if LIN
#include <sys/syscall.h>
#include <unistd.h>
#endif  /* LIN */
 
#include "assert.h"
#include "helpers.h"
#include "list.h"
#include "time.h"
#include "safe_alloc.h"
 
#ifdef __cplusplus
extern "C" {
#endif
 
typedef struct {
    void        (*proc)(void *);
    void        *arg;
    const char  *filename;
    int     linenum;
    list_node_t node;
} lacf_thread_info_t;
 
/*
 * !!!! CAUTION !!!!
 * MacOS's kernel API contains a function named `thread_create' that'll
 * clash with the naming below. To work around this, we declare the
 * function name to be a macro to our own scoped version. The user must
 * include /usr/include/mach/task.h before us, otherwise we could be
 * rewriting the function name in the header too. The flip side is that
 * the user will get calls to libacfutils' thread_create instead.
 */
#if !APL || !defined(_task_user_)
#define thread_create(thrp, start_proc, arg) \
    lacf_thread_create((thrp), (start_proc), (arg), \
        log_basename(__FILE__), __LINE__)
#endif
 
#ifdef  _USE_STDATOMICS
#define atomic32_t      _Atomic int32_t
#define atomic_inc_32(x)    atomic_fetch_add((x), 1)
#define atomic_dec_32(x)    atomic_fetch_add((x), -1)
#define atomic_set_32(x, y) atomic_store((x), (y))
#define atomic_add_32(x, y) atomic_fetch_add((x), (y))
#define atomic64_t      _Atomic int64_t
#define atomic_inc_64(x)    atomic_fetch_add((x), 1)
#define atomic_dec_64(x)    atomic_fetch_add((x), -1)
#define atomic_set_64(x, y) atomic_store((x), (y))
#define atomic_add_64(x, y) atomic_fetch_add((x), (y))
#elif   IBM
#define atomic32_t      volatile LONG
#define atomic_inc_32(x)    InterlockedIncrement((x))
#define atomic_dec_32(x)    InterlockedDecrement((x))
#define atomic_add_32(x, y) InterlockedAdd((x), (y))
#define atomic_set_32(x, y) InterlockedExchange((x), (y))
#define atomic64_t      volatile LONG64
#define atomic_inc_64(x)    InterlockedIncrement64((x))
#define atomic_dec_64(x)    InterlockedDecrement64((x))
#define atomic_add_64(x, y) InterlockedAdd64((x), (y))
#define atomic_set_64(x, y) InterlockedExchange64((x), (y))
#elif   APL
#define atomic32_t      volatile int32_t
#define atomic_inc_32(x)    OSAtomicAdd32(1, (x))
#define atomic_dec_32(x)    OSAtomicAdd32(-1, (x))
#define atomic_add_32(x, y) OSAtomicAdd32((y), (x))
#define atomic_set_32(x, y) (x) = (y)   /* No such op on OSX */
#define atomic64_t      volatile int64_t
#define atomic_inc_64(x)    OSAtomicAdd64(1, (x))
#define atomic_dec_64(x)    OSAtomicAdd64(-1, (x))
#define atomic_add_64(x, y) OSAtomicAdd64((y), (x))
#define atomic_set_64(x, y) (x) = (y)   /* No such op on OSX */
#else   /* LIN */
#define atomic32_t      volatile int32_t
#define atomic_inc_32(x)    __sync_add_and_fetch((x), 1)
#define atomic_dec_32(x)    __sync_add_and_fetch((x), -1)
#define atomic_add_32(x, y) __sync_add_and_fetch((x), (y))
#define atomic_set_32(x, y) __atomic_store_n((x), (y), __ATOMIC_RELAXED)
#define atomic64_t      volatile int64_t
#define atomic_inc_64(x)    __sync_add_and_fetch((x), 1)
#define atomic_dec_64(x)    __sync_add_and_fetch((x), -1)
#define atomic_add_64(x, y) __sync_add_and_fetch((x), (y))
#define atomic_set_64(x, y) __atomic_store_n((x), (y), __ATOMIC_RELAXED)
#endif  /* LIN */
 
#if APL || LIN
 
#if APL
#define thread_t    pthread_t
#define thread_id_t pthread_t
#define mutex_t pthread_mutex_t
#define condvar_t   pthread_cond_t
#else   /* !APL */
typedef pthread_t thread_t;
typedef pthread_t thread_id_t;
typedef pthread_mutex_t mutex_t;
typedef pthread_cond_t condvar_t;
#endif  /* !APL */
#define curthread_id        pthread_self()
#define curthread       pthread_self()
 
static inline bool_t
thread_equal(thread_id_t t1, thread_id_t t2)
{
    return (pthread_equal(t1, t2));
}
 
static inline void
mutex_init(mutex_t *mtx)
{
    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(mtx, &attr);
}
 
static inline void
mutex_destroy(mutex_t *mtx)
{
    pthread_mutex_destroy(mtx);
}
 
static inline void
mutex_enter(mutex_t *mtx)
{
    pthread_mutex_lock(mtx);
}
 
static inline void
mutex_exit(mutex_t *mtx)
{
    pthread_mutex_unlock(mtx);
}
 
#if LIN
#define VERIFY_MUTEX_HELD(mtx)  \
    VERIFY((mtx)->__data.__owner == syscall(SYS_gettid))
#define VERIFY_MUTEX_NOT_HELD(mtx) \
    VERIFY((mtx)->__data.__owner != syscall(SYS_gettid))
#else   /* APL */
#define VERIFY_MUTEX_HELD(mtx)      (void)1
#define VERIFY_MUTEX_NOT_HELD(mtx)  (void)1
#endif  /* APL */
 
#else   /* !APL && !LIN */
 
typedef HANDLE thread_t;
typedef DWORD thread_id_t;
typedef struct {
    bool_t          inited;
    CRITICAL_SECTION    cs;
} mutex_t;
typedef CONDITION_VARIABLE condvar_t;
 
#define curthread_id            GetCurrentThreadId()
#define curthread           GetCurrentThread()
 
/*
 * Compares thread IDs in a platform-independent way.
 * @return Returns true if thread ID `tid1` is equal to `tid2`. These
 *  thread IDs are as returned by curthread_id().
 * @see curthread_id
 */
static inline bool_t
thread_equal(thread_id_t tid1, thread_id_t tid2)
{
    return (tid1 == tid2);
}
 
static inline void
mutex_init(mutex_t *mtx)
{
    mtx->inited = B_TRUE;
    InitializeCriticalSection(&mtx->cs);
}
 
static inline void
mutex_destroy(mutex_t *mtx)
{
    if (mtx->inited) {
        DeleteCriticalSection(&mtx->cs);
        mtx->inited = B_FALSE;
    }
}
 
static inline void
mutex_enter(mutex_t *mtx)
{
    ASSERT(mtx->inited);
    EnterCriticalSection(&mtx->cs);
}
 
static inline void
mutex_exit(mutex_t *mtx)
{
    ASSERT(mtx->inited);
    LeaveCriticalSection(&mtx->cs);
}
 
#define VERIFY_MUTEX_HELD(mtx)      (void)1
#define VERIFY_MUTEX_NOT_HELD(mtx)  (void)1
 
#endif  /* !APL && !LIN */
 
/*
 * This is the thread cleanup tracking machinery.
 */
API_EXPORT extern bool_t    lacf_thread_list_inited;
API_EXPORT extern mutex_t   lacf_thread_list_lock;
API_EXPORT extern list_t    lacf_thread_list;
 
static inline void
_lacf_thread_list_init(void)
{
    if (!lacf_thread_list_inited) {
        mutex_init(&lacf_thread_list_lock);
        list_create(&lacf_thread_list, sizeof (lacf_thread_info_t),
            offsetof(lacf_thread_info_t, node));
        lacf_thread_list_inited = B_TRUE;
    }
}
 
static inline void
_lacf_thread_list_fini(void)
{
    if (lacf_thread_list_inited) {
        mutex_enter(&lacf_thread_list_lock);
        if (list_count(&lacf_thread_list) == 0) {
            list_destroy(&lacf_thread_list);
            mutex_exit(&lacf_thread_list_lock);
            mutex_destroy(&lacf_thread_list_lock);
            lacf_thread_list_inited = B_FALSE;
        } else {
            mutex_exit(&lacf_thread_list_lock);
        }
    }
}
 
static inline void
_lacf_thread_list_add(lacf_thread_info_t *ti)
{
    ASSERT(ti != NULL);
    ASSERT(lacf_thread_list_inited);
    mutex_enter(&lacf_thread_list_lock);
    list_insert_tail(&lacf_thread_list, ti);
    mutex_exit(&lacf_thread_list_lock);
}
 
static inline void
_lacf_thread_list_remove(lacf_thread_info_t *ti)
{
    ASSERT(ti != NULL);
    ASSERT(lacf_thread_list_inited);
    mutex_enter(&lacf_thread_list_lock);
    list_remove(&lacf_thread_list, ti);
    mutex_exit(&lacf_thread_list_lock);
}
 
static inline void
lacf_threads_fini(void)
{
    if (!lacf_thread_list_inited)
        return;
 
    mutex_enter(&lacf_thread_list_lock);
    for (lacf_thread_info_t *ti =
        (lacf_thread_info_t *)list_head(&lacf_thread_list); ti != NULL;
        ti = (lacf_thread_info_t *)list_next(&lacf_thread_list, ti)) {
        logMsg("Leaked thread, created here %s:%d", ti->filename,
            ti->linenum);
    }
    VERIFY0(list_count(&lacf_thread_list));
    mutex_exit(&lacf_thread_list_lock);
 
    _lacf_thread_list_fini();
}
 
#if APL || LIN
 
void *lacf_thread_start_routine(void *arg);
 
WARN_UNUSED_RES_ATTR static inline bool_t
lacf_thread_create(thread_t *thrp, void (*proc)(void *), void *arg,
    const char *filename, int linenum)
{
    lacf_thread_info_t *ti =
        (lacf_thread_info_t *)safe_calloc(1, sizeof (*ti));
    ti->proc = proc;
    ti->arg = arg;
    ti->filename = filename;
    ti->linenum = linenum;
    _lacf_thread_list_init();
    _lacf_thread_list_add(ti);
    if (pthread_create(thrp, NULL, lacf_thread_start_routine, ti) == 0) {
        /* Start success */
        return (B_TRUE);
    }
    /* Start failure - remove from list and try to destroy the list */
    _lacf_thread_list_remove(ti);
    _lacf_thread_list_fini();
    free(ti);
    return (B_FALSE);
}
 
static inline void
thread_join(thread_t *thrp)
{
    pthread_join(*thrp, NULL);
    _lacf_thread_list_fini();
}
 
#if LIN
 
static inline void
thread_set_name(const char *name)
{
    pthread_setname_np(pthread_self(), name);
}
 
#else   /* APL */
 
static inline void
thread_set_name(const char *name)
{
    pthread_setname_np(name);
}
 
#endif  /* APL */
 
static inline void
cv_wait(condvar_t *cv, mutex_t *mtx)
{
    pthread_cond_wait(cv, mtx);
}
 
static inline int
cv_timedwait(condvar_t *cv, mutex_t *mtx, uint64_t limit)
{
    struct timespec ts = { .tv_sec = (time_t)(limit / 1000000),
        .tv_nsec = (long)((limit % 1000000) * 1000) };
    return (pthread_cond_timedwait(cv, mtx, &ts));
}
 
static inline void
cv_init(condvar_t *cv)
{
    pthread_cond_init(cv, NULL);
}
 
static inline void
cv_destroy(condvar_t *cv)
{
    pthread_cond_destroy(cv);
}
 
static inline void
cv_signal(condvar_t *cv)
{
    pthread_cond_signal(cv);
}
 
static inline void
cv_broadcast(condvar_t *cv)
{
    pthread_cond_broadcast(cv);
}
 
#if !APL
#define THREAD_PRIO_IDLE    sched_get_priority_min()
#define THREAD_PRIO_VERY_LOW    (THREAD_PRIO_NORM - 2)
#define THREAD_PRIO_LOW     (THREAD_PRIO_NORM - 1)
#define THREAD_PRIO_NORM    0   /* Default priority on Linux */
#define THREAD_PRIO_HIGH    (THREAD_PRIO_NORM + 1)
#define THREAD_PRIO_VERY_HIGH   (THREAD_PRIO_NORM + 2)
#define THREAD_PRIO_RT      sched_get_priority_max()
static inline void
thread_set_prio(thread_t thr, int prio)
{
    struct sched_param param = {0};
    param.sched_priority = (prio);
    pthread_setschedparam(thr, SCHED_OTHER, &param);
}
 
#else   /* APL */
/*
 * BIG CAVEAT: Apparently idle thread prioritization is causing massive
 * thread scheduling stability issues on MacOS Monterey with its Rosetta
 * x86 emulation. Threads either don't get scheduled, or they run in
 * "slow mo", gradually speeding up and generally just behave entirely
 * erratically.
 */
#define THREAD_PRIO_IDLE    0
#define THREAD_PRIO_VERY_LOW    0
#define THREAD_PRIO_LOW     0
#define THREAD_PRIO_NORM    0
#define THREAD_PRIO_HIGH    0
#define THREAD_PRIO_VERY_HIGH   0
#define THREAD_PRIO_RT      0
static inline void
thread_set_prio(thread_t thr, int prio)
{
    LACF_UNUSED(thr);
    LACF_UNUSED(prio);
}
 
#endif  /* APL */
 
#else   /* !APL && !LIN */
 
DWORD lacf_thread_start_routine(void *arg);
 
WARN_UNUSED_RES_ATTR static inline bool_t
lacf_thread_create(thread_t *thrp, void (*proc)(void *), void *arg,
    const char *filename, int linenum)
{
    lacf_thread_info_t *ti =
        (lacf_thread_info_t *)safe_calloc(1, sizeof (*ti));
    ti->proc = proc;
    ti->arg = arg;
    ti->filename = filename;
    ti->linenum = linenum;
    _lacf_thread_list_init();
    _lacf_thread_list_add(ti);
    if ((*(thrp) = CreateThread(NULL, 0, lacf_thread_start_routine, ti,
        0, NULL)) != NULL) {
        /* Start success */
        return (B_TRUE);
    }
    /* Start failure - remove from list and try to destroy the list */
    _lacf_thread_list_remove(ti);
    _lacf_thread_list_fini();
    free(ti);
    return (B_FALSE);
}
 
static inline void
thread_join(thread_t *thrp)
{
    VERIFY3S(WaitForSingleObject(*thrp, INFINITE), ==, WAIT_OBJECT_0);
    _lacf_thread_list_fini();
}
 
static inline void
thread_set_name(const char *name)
{
    LACF_UNUSED(name);
}
 
static inline void
cv_wait(condvar_t *cv, mutex_t *mtx)
{
    VERIFY(SleepConditionVariableCS(cv, &mtx->cs, INFINITE));
}
 
static inline int
cv_timedwait(condvar_t *cv, mutex_t *mtx, uint64_t limit)
{
    uint64_t now = microclock();
    if (now < limit) {
        /*
         * The only way to guarantee that when we return due to a
         * timeout the full microsecond-accurate quantum has elapsed
         * is to round-up to the nearest millisecond.
         */
        if (SleepConditionVariableCS(cv, &mtx->cs,
            ceil((limit - now) / 1000.0)) != 0) {
            return (0);
        }
        if (GetLastError() == ERROR_TIMEOUT)
            return (ETIMEDOUT);
        return (-1);
    }
    return (ETIMEDOUT);
}
 
static inline void
cv_init(condvar_t *cv)
{
    InitializeConditionVariable(cv);
}
 
static inline void
cv_destroy(condvar_t *cv)
{
    LACF_UNUSED(cv);
}
 
static inline void
cv_signal(condvar_t *cv)
{
    WakeConditionVariable(cv);
}
 
static inline void
cv_broadcast(condvar_t *cv)
{
    WakeAllConditionVariable(cv);
}
 
#define THREAD_PRIO_IDLE        THREAD_PRIORITY_IDLE
#define THREAD_PRIO_VERY_LOW        THREAD_PRIORITY_LOWEST
#define THREAD_PRIO_LOW         THREAD_PRIORITY_BELOW_NORMAL
#define THREAD_PRIO_NORM        THREAD_PRIORITY_NORMAL
#define THREAD_PRIO_HIGH        THREAD_PRIORITY_ABOVE_NORMAL
#define THREAD_PRIO_VERY_HIGH       THREAD_PRIORITY_HIGHEST
#define THREAD_PRIO_RT          THREAD_PRIORITY_TIME_CRITICAL
static inline void
thread_set_prio(thread_t thr, int prio)
{
    SetThreadPriority(thr, prio);
}
 
#endif  /* !APL && !LIN */
 
API_EXPORT void lacf_mask_sigpipe(void);
 
typedef struct {
    mutex_t         lock;
    condvar_t       cv;
    bool_t          write_locked;
    thread_id_t     writer;
    unsigned        refcount;
    list_t          waiters;
} rwmutex_t;
 
typedef struct {
    bool_t          write;
    list_node_t     node;
} rwlock_waiter_t;
 
UNUSED_ATTR static void
rwmutex_init(rwmutex_t *rw)
{
    memset(rw, 0, sizeof (*rw));
    mutex_init(&rw->lock);
    cv_init(&rw->cv);
    list_create(&rw->waiters, sizeof (rwlock_waiter_t),
        offsetof(rwlock_waiter_t, node));
}
 
UNUSED_ATTR static void
rwmutex_destroy(rwmutex_t *rw)
{
    ASSERT3U(rw->refcount, ==, 0);
    list_destroy(&rw->waiters);
    cv_destroy(&rw->cv);
    mutex_destroy(&rw->lock);
}
 
static inline bool_t
rwmutex_held_write(rwmutex_t *rw)
{
    return (rw->write_locked && rw->writer == curthread_id);
}
 
static inline bool_t
rwmutex_can_enter_impl(const rwmutex_t *rw, const rwlock_waiter_t *wt_self)
{
    for (const rwlock_waiter_t *wt = (rwlock_waiter_t *)list_head(
        &rw->waiters);;
        wt = (rwlock_waiter_t *)list_next(&rw->waiters, wt)) {
        /* Our wt_self MUST be somewhere in rw->waiters! */
        VERIFY(wt != NULL);
        if (wt == wt_self)
            return (B_TRUE);
        if (wt->write)
            return (B_FALSE);
    }
}
 
UNUSED_ATTR static void
rwmutex_enter(rwmutex_t *rw, bool_t write)
{
    rwlock_waiter_t wt = {0};
 
    wt.write = write;
    /*
     * No recursion allowed! We can't check for recursive read attempts,
     * only write (since readers don't retain any ownership information),
     * so it's best to avoid recursion altogether.
     */
    ASSERT_MSG(!rwmutex_held_write(rw), "%s", "Attempted to recursively "
        "acquire an rwmutex_t. This is NOT supported!");
 
    mutex_enter(&rw->lock);
    /*
     * Enter the queue of threads waiting to acquire the mutex.
     */
    list_insert_tail(&rw->waiters, &wt);
 
    if (write) {
        /*
         * Wait until everybody else is out of the mutex
         * and we're next to enter.
         */
        while (rw->refcount != 0 ||
            list_head(&rw->waiters) != (void *)&wt) {
            cv_wait(&rw->cv, &rw->lock);
        }
        /*
         * We're clear to proceed, mark the mutex as
         * write-locked by us.
         */
        rw->writer = curthread_id;
        rw->write_locked = B_TRUE;
    } else {
        /*
         * If the mutex is currently held by a writer, or
         * there's another writer ahead of us, wait.
         */
        while (rw->write_locked ||
            !rwmutex_can_enter_impl(rw, &wt)) {
            cv_wait(&rw->cv, &rw->lock);
        }
    }
    /*
     * Exit the wait queue. We've now acquired the mutex.
     */
    list_remove(&rw->waiters, &wt);
    rw->refcount++;
 
    mutex_exit(&rw->lock);
}
 
UNUSED_ATTR static void
rwmutex_exit(rwmutex_t *rw)
{
    mutex_enter(&rw->lock);
    ASSERT(rw->refcount != 0);
    rw->refcount--;
    if (rw->refcount == 0 && rw->write_locked) {
        ASSERT3U(rw->writer, ==, curthread_id);
        rw->write_locked = B_FALSE;
    }
    if (list_head(&rw->waiters) != NULL)
        cv_broadcast(&rw->cv);
    mutex_exit(&rw->lock);
}
 
static inline void
rwmutex_upgrade(rwmutex_t *rw)
{
    rwmutex_exit(rw);
    rwmutex_enter(rw, B_TRUE);
}
 
#ifdef  DEBUG
#define ASSERT_MUTEX_HELD(mtx)      VERIFY_MUTEX_HELD(mtx)
#define ASSERT_MUTEX_NOT_HELD(mtx)  VERIFY_MUTEX_NOT_HELD(mtx)
#else   /* !DEBUG */
#define ASSERT_MUTEX_HELD(mtx)
#define ASSERT_MUTEX_NOT_HELD(mtx)
#endif  /* !DEBUG */
 
#ifdef __cplusplus
}
#endif
 
#endif  /* _ACF_UTILS_THREAD_H_ */