[apple/libdispatch.git] / src / semaphore.c

/*
 * Copyright (c) 2008-2011 Apple Inc. All rights reserved.
 *
 * @APPLE_APACHE_LICENSE_HEADER_START@
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * @APPLE_APACHE_LICENSE_HEADER_END@
 */

#include "internal.h"

// semaphores are too fundamental to use the dispatch_assume*() macros
#if USE_MACH_SEM
#define DISPATCH_SEMAPHORE_VERIFY_KR(x) do { \
                if (slowpath(x)) { \
                        DISPATCH_CRASH("flawed group/semaphore logic"); \
                } \
        } while (0)
#elif USE_POSIX_SEM
#define DISPATCH_SEMAPHORE_VERIFY_RET(x) do { \
                if (slowpath((x) == -1)) { \
                        DISPATCH_CRASH("flawed group/semaphore logic"); \
                } \
        } while (0)
#endif

DISPATCH_WEAK // rdar://problem/8503746
long _dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema);

static long _dispatch_group_wake(dispatch_semaphore_t dsema);

#pragma mark -
#pragma mark dispatch_semaphore_t

static void
_dispatch_semaphore_init(long value, dispatch_object_t dou)
{
        dispatch_semaphore_t dsema = dou._dsema;

        dsema->do_next = DISPATCH_OBJECT_LISTLESS;
        dsema->do_targetq = dispatch_get_global_queue(
                        DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
        dsema->dsema_value = value;
        dsema->dsema_orig = value;
#if USE_POSIX_SEM
        int ret = sem_init(&dsema->dsema_sem, 0, 0);
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
#endif
}

dispatch_semaphore_t
dispatch_semaphore_create(long value)
{
        dispatch_semaphore_t dsema;

        // If the internal value is negative, then the absolute of the value is
        // equal to the number of waiting threads. Therefore it is bogus to
        // initialize the semaphore with a negative value.
        if (value < 0) {
                return NULL;
        }

        dsema = _dispatch_alloc(DISPATCH_VTABLE(semaphore),
                        sizeof(struct dispatch_semaphore_s));
        _dispatch_semaphore_init(value, dsema);
        return dsema;
}

#if USE_MACH_SEM
static void
_dispatch_semaphore_create_port(semaphore_t *s4)
{
        kern_return_t kr;
        semaphore_t tmp;

        if (*s4) {
                return;
        }
        _dispatch_safe_fork = false;

        // lazily allocate the semaphore port

        // Someday:
        // 1) Switch to a doubly-linked FIFO in user-space.
        // 2) User-space timers for the timeout.
        // 3) Use the per-thread semaphore port.

        while ((kr = semaphore_create(mach_task_self(), &tmp,
                        SYNC_POLICY_FIFO, 0))) {
                DISPATCH_VERIFY_MIG(kr);
                sleep(1);
        }

        if (!dispatch_atomic_cmpxchg(s4, 0, tmp)) {
                kr = semaphore_destroy(mach_task_self(), tmp);
                DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        }
}
#endif

void
_dispatch_semaphore_dispose(dispatch_object_t dou)
{
        dispatch_semaphore_t dsema = dou._dsema;

        if (dsema->dsema_value < dsema->dsema_orig) {
                DISPATCH_CLIENT_CRASH(
                                "Semaphore/group object deallocated while in use");
        }

#if USE_MACH_SEM
        kern_return_t kr;
        if (dsema->dsema_port) {
                kr = semaphore_destroy(mach_task_self(), dsema->dsema_port);
                DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        }
        if (dsema->dsema_waiter_port) {
                kr = semaphore_destroy(mach_task_self(), dsema->dsema_waiter_port);
                DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        }
#elif USE_POSIX_SEM
        int ret = sem_destroy(&dsema->dsema_sem);
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
#endif
}

size_t
_dispatch_semaphore_debug(dispatch_object_t dou, char *buf, size_t bufsiz)
{
        dispatch_semaphore_t dsema = dou._dsema;

        size_t offset = 0;
        offset += snprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
                        dx_kind(dsema), dsema);
        offset += _dispatch_object_debug_attr(dsema, &buf[offset], bufsiz - offset);
#if USE_MACH_SEM
        offset += snprintf(&buf[offset], bufsiz - offset, "port = 0x%u, ",
                        dsema->dsema_port);
#endif
        offset += snprintf(&buf[offset], bufsiz - offset,
                        "value = %ld, orig = %ld }", dsema->dsema_value, dsema->dsema_orig);
        return offset;
}

DISPATCH_NOINLINE
long
_dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema)
{
        // Before dsema_sent_ksignals is incremented we can rely on the reference
        // held by the waiter. However, once this value is incremented the waiter
        // may return between the atomic increment and the semaphore_signal(),
        // therefore an explicit reference must be held in order to safely access
        // dsema after the atomic increment.
        _dispatch_retain(dsema);

        (void)dispatch_atomic_inc2o(dsema, dsema_sent_ksignals);

#if USE_MACH_SEM
        _dispatch_semaphore_create_port(&dsema->dsema_port);
        kern_return_t kr = semaphore_signal(dsema->dsema_port);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
#elif USE_POSIX_SEM
        int ret = sem_post(&dsema->dsema_sem);
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
#endif

        _dispatch_release(dsema);
        return 1;
}

long
dispatch_semaphore_signal(dispatch_semaphore_t dsema)
{
        dispatch_atomic_release_barrier();
        long value = dispatch_atomic_inc2o(dsema, dsema_value);
        if (fastpath(value > 0)) {
                return 0;
        }
        if (slowpath(value == LONG_MIN)) {
                DISPATCH_CLIENT_CRASH("Unbalanced call to dispatch_group_leave() or "
                                "dispatch_semaphore_signal()");
        }
        return _dispatch_semaphore_signal_slow(dsema);
}

DISPATCH_NOINLINE
static long
_dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema,
                dispatch_time_t timeout)
{
        long orig;

again:
        // Mach semaphores appear to sometimes spuriously wake up. Therefore,
        // we keep a parallel count of the number of times a Mach semaphore is
        // signaled (6880961).
        while ((orig = dsema->dsema_sent_ksignals)) {
                if (dispatch_atomic_cmpxchg2o(dsema, dsema_sent_ksignals, orig,
                                orig - 1)) {
                        return 0;
                }
        }

#if USE_MACH_SEM
        mach_timespec_t _timeout;
        kern_return_t kr;

        _dispatch_semaphore_create_port(&dsema->dsema_port);

        // From xnu/osfmk/kern/sync_sema.c:
        // wait_semaphore->count = -1; /* we don't keep an actual count */
        //
        // The code above does not match the documentation, and that fact is
        // not surprising. The documented semantics are clumsy to use in any
        // practical way. The above hack effectively tricks the rest of the
        // Mach semaphore logic to behave like the libdispatch algorithm.

        switch (timeout) {
        default:
                do {
                        uint64_t nsec = _dispatch_timeout(timeout);
                        _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
                        _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
                        kr = slowpath(semaphore_timedwait(dsema->dsema_port, _timeout));
                } while (kr == KERN_ABORTED);

                if (kr != KERN_OPERATION_TIMED_OUT) {
                        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
                        break;
                }
                // Fall through and try to undo what the fast path did to
                // dsema->dsema_value
        case DISPATCH_TIME_NOW:
                while ((orig = dsema->dsema_value) < 0) {
                        if (dispatch_atomic_cmpxchg2o(dsema, dsema_value, orig, orig + 1)) {
                                return KERN_OPERATION_TIMED_OUT;
                        }
                }
                // Another thread called semaphore_signal().
                // Fall through and drain the wakeup.
        case DISPATCH_TIME_FOREVER:
                do {
                        kr = semaphore_wait(dsema->dsema_port);
                } while (kr == KERN_ABORTED);
                DISPATCH_SEMAPHORE_VERIFY_KR(kr);
                break;
        }
#elif USE_POSIX_SEM
        struct timespec _timeout;
        int ret;

        switch (timeout) {
        default:
                do {
                        uint64_t nsec = _dispatch_timeout(timeout);
                        _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
                        _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
                        ret = slowpath(sem_timedwait(&dsema->dsema_sem, &_timeout));
                } while (ret == -1 && errno == EINTR);

                if (ret == -1 && errno != ETIMEDOUT) {
                        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
                        break;
                }
                // Fall through and try to undo what the fast path did to
                // dsema->dsema_value
        case DISPATCH_TIME_NOW:
                while ((orig = dsema->dsema_value) < 0) {
                        if (dispatch_atomic_cmpxchg2o(dsema, dsema_value, orig, orig + 1)) {
                                errno = ETIMEDOUT;
                                return -1;
                        }
                }
                // Another thread called semaphore_signal().
                // Fall through and drain the wakeup.
        case DISPATCH_TIME_FOREVER:
                do {
                        ret = sem_wait(&dsema->dsema_sem);
                } while (ret != 0);
                DISPATCH_SEMAPHORE_VERIFY_RET(ret);
                break;
        }
#endif

        goto again;
}

long
dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
        long value = dispatch_atomic_dec2o(dsema, dsema_value);
        dispatch_atomic_acquire_barrier();
        if (fastpath(value >= 0)) {
                return 0;
        }
        return _dispatch_semaphore_wait_slow(dsema, timeout);
}

#pragma mark -
#pragma mark dispatch_group_t

dispatch_group_t
dispatch_group_create(void)
{
        dispatch_group_t dg = _dispatch_alloc(DISPATCH_VTABLE(group),
                        sizeof(struct dispatch_semaphore_s));
        _dispatch_semaphore_init(LONG_MAX, dg);
        return dg;
}

void
dispatch_group_enter(dispatch_group_t dg)
{
        dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;

        (void)dispatch_semaphore_wait(dsema, DISPATCH_TIME_FOREVER);
}

DISPATCH_NOINLINE
static long
_dispatch_group_wake(dispatch_semaphore_t dsema)
{
        struct dispatch_sema_notify_s *next, *head, *tail = NULL;
        long rval;

        head = dispatch_atomic_xchg2o(dsema, dsema_notify_head, NULL);
        if (head) {
                // snapshot before anything is notified/woken <rdar://problem/8554546>
                tail = dispatch_atomic_xchg2o(dsema, dsema_notify_tail, NULL);
        }
        rval = dispatch_atomic_xchg2o(dsema, dsema_group_waiters, 0);
        if (rval) {
                // wake group waiters
#if USE_MACH_SEM
                _dispatch_semaphore_create_port(&dsema->dsema_waiter_port);
                do {
                        kern_return_t kr = semaphore_signal(dsema->dsema_waiter_port);
                        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
                } while (--rval);
#elif USE_POSIX_SEM
                do {
                        int ret = sem_post(&dsema->dsema_sem);
                        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
                } while (--rval);
#endif
        }
        if (head) {
                // async group notify blocks
                do {
                        dispatch_async_f(head->dsn_queue, head->dsn_ctxt, head->dsn_func);
                        _dispatch_release(head->dsn_queue);
                        next = fastpath(head->dsn_next);
                        if (!next && head != tail) {
                                while (!(next = fastpath(head->dsn_next))) {
                                        _dispatch_hardware_pause();
                                }
                        }
                        free(head);
                } while ((head = next));
                _dispatch_release(dsema);
        }
        return 0;
}

void
dispatch_group_leave(dispatch_group_t dg)
{
        dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;

        dispatch_semaphore_signal(dsema);
        if (dsema->dsema_value == dsema->dsema_orig) {
                (void)_dispatch_group_wake(dsema);
        }
}

DISPATCH_NOINLINE
static long
_dispatch_group_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
        long orig;

again:
        // check before we cause another signal to be sent by incrementing
        // dsema->dsema_group_waiters
        if (dsema->dsema_value == dsema->dsema_orig) {
                return _dispatch_group_wake(dsema);
        }
        // Mach semaphores appear to sometimes spuriously wake up. Therefore,
        // we keep a parallel count of the number of times a Mach semaphore is
        // signaled (6880961).
        (void)dispatch_atomic_inc2o(dsema, dsema_group_waiters);
        // check the values again in case we need to wake any threads
        if (dsema->dsema_value == dsema->dsema_orig) {
                return _dispatch_group_wake(dsema);
        }

#if USE_MACH_SEM
        mach_timespec_t _timeout;
        kern_return_t kr;

        _dispatch_semaphore_create_port(&dsema->dsema_waiter_port);

        // From xnu/osfmk/kern/sync_sema.c:
        // wait_semaphore->count = -1; /* we don't keep an actual count */
        //
        // The code above does not match the documentation, and that fact is
        // not surprising. The documented semantics are clumsy to use in any
        // practical way. The above hack effectively tricks the rest of the
        // Mach semaphore logic to behave like the libdispatch algorithm.

        switch (timeout) {
        default:
                do {
                        uint64_t nsec = _dispatch_timeout(timeout);
                        _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
                        _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
                        kr = slowpath(semaphore_timedwait(dsema->dsema_waiter_port,
                                        _timeout));
                } while (kr == KERN_ABORTED);

                if (kr != KERN_OPERATION_TIMED_OUT) {
                        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
                        break;
                }
                // Fall through and try to undo the earlier change to
                // dsema->dsema_group_waiters
        case DISPATCH_TIME_NOW:
                while ((orig = dsema->dsema_group_waiters)) {
                        if (dispatch_atomic_cmpxchg2o(dsema, dsema_group_waiters, orig,
                                        orig - 1)) {
                                return KERN_OPERATION_TIMED_OUT;
                        }
                }
                // Another thread called semaphore_signal().
                // Fall through and drain the wakeup.
        case DISPATCH_TIME_FOREVER:
                do {
                        kr = semaphore_wait(dsema->dsema_waiter_port);
                } while (kr == KERN_ABORTED);
                DISPATCH_SEMAPHORE_VERIFY_KR(kr);
                break;
        }
#elif USE_POSIX_SEM
        struct timespec _timeout;
        int ret;

        switch (timeout) {
        default:
                do {
                        uint64_t nsec = _dispatch_timeout(timeout);
                        _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
                        _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
                        ret = slowpath(sem_timedwait(&dsema->dsema_sem, &_timeout));
                } while (ret == -1 && errno == EINTR);

                if (!(ret == -1 && errno == ETIMEDOUT)) {
                        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
                        break;
                }
                // Fall through and try to undo the earlier change to
                // dsema->dsema_group_waiters
        case DISPATCH_TIME_NOW:
                while ((orig = dsema->dsema_group_waiters)) {
                        if (dispatch_atomic_cmpxchg2o(dsema, dsema_group_waiters, orig,
                                        orig - 1)) {
                                errno = ETIMEDOUT;
                                return -1;
                        }
                }
                // Another thread called semaphore_signal().
                // Fall through and drain the wakeup.
        case DISPATCH_TIME_FOREVER:
                do {
                        ret = sem_wait(&dsema->dsema_sem);
                } while (ret == -1 && errno == EINTR);
                DISPATCH_SEMAPHORE_VERIFY_RET(ret);
                break;
        }
#endif

        goto again;
}

long
dispatch_group_wait(dispatch_group_t dg, dispatch_time_t timeout)
{
        dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;

        if (dsema->dsema_value == dsema->dsema_orig) {
                return 0;
        }
        if (timeout == 0) {
#if USE_MACH_SEM
                return KERN_OPERATION_TIMED_OUT;
#elif USE_POSIX_SEM
                errno = ETIMEDOUT;
                return (-1);
#endif
        }
        return _dispatch_group_wait_slow(dsema, timeout);
}

DISPATCH_NOINLINE
void
dispatch_group_notify_f(dispatch_group_t dg, dispatch_queue_t dq, void *ctxt,
                void (*func)(void *))
{
        dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;
        struct dispatch_sema_notify_s *dsn, *prev;

        // FIXME -- this should be updated to use the continuation cache
        while (!(dsn = calloc(1, sizeof(*dsn)))) {
                sleep(1);
        }

        dsn->dsn_queue = dq;
        dsn->dsn_ctxt = ctxt;
        dsn->dsn_func = func;
        _dispatch_retain(dq);
        dispatch_atomic_store_barrier();
        prev = dispatch_atomic_xchg2o(dsema, dsema_notify_tail, dsn);
        if (fastpath(prev)) {
                prev->dsn_next = dsn;
        } else {
                _dispatch_retain(dg);
                dsema->dsema_notify_head = dsn;
                if (dsema->dsema_value == dsema->dsema_orig) {
                        _dispatch_group_wake(dsema);
                }
        }
}

#ifdef __BLOCKS__
void
dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
                dispatch_block_t db)
{
        dispatch_group_notify_f(dg, dq, _dispatch_Block_copy(db),
                        _dispatch_call_block_and_release);
}
#endif

#pragma mark -
#pragma mark _dispatch_thread_semaphore_t

DISPATCH_NOINLINE
static _dispatch_thread_semaphore_t
_dispatch_thread_semaphore_create(void)
{
        _dispatch_safe_fork = false;
#if USE_MACH_SEM
        semaphore_t s4;
        kern_return_t kr;
        while (slowpath(kr = semaphore_create(mach_task_self(), &s4,
                        SYNC_POLICY_FIFO, 0))) {
                DISPATCH_VERIFY_MIG(kr);
                sleep(1);
        }
        return s4;
#elif USE_POSIX_SEM
        sem_t s4;
        int ret = sem_init(&s4, 0, 0);
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
        return s4;
#endif
}

DISPATCH_NOINLINE
void
_dispatch_thread_semaphore_dispose(_dispatch_thread_semaphore_t sema)
{
#if USE_MACH_SEM
        semaphore_t s4 = (semaphore_t)sema;
        kern_return_t kr = semaphore_destroy(mach_task_self(), s4);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
#elif USE_POSIX_SEM
        sem_t s4 = (sem_t)sema;
        int ret = sem_destroy(&s4);
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
#endif
}

void
_dispatch_thread_semaphore_signal(_dispatch_thread_semaphore_t sema)
{
#if USE_MACH_SEM
        semaphore_t s4 = (semaphore_t)sema;
        kern_return_t kr = semaphore_signal(s4);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
#elif USE_POSIX_SEM
        sem_t s4 = (sem_t)sema;
        int ret = sem_post(&s4);
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
#endif
}

void
_dispatch_thread_semaphore_wait(_dispatch_thread_semaphore_t sema)
{
#if USE_MACH_SEM
        semaphore_t s4 = (semaphore_t)sema;
        kern_return_t kr;
        do {
                kr = semaphore_wait(s4);
        } while (slowpath(kr == KERN_ABORTED));
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
#elif USE_POSIX_SEM
        sem_t s4 = (sem_t)sema;
        int ret;
        do {
                ret = sem_wait(&s4);
        } while (slowpath(ret != 0));
        DISPATCH_SEMAPHORE_VERIFY_RET(ret);
#endif
}

_dispatch_thread_semaphore_t
_dispatch_get_thread_semaphore(void)
{
        _dispatch_thread_semaphore_t sema = (_dispatch_thread_semaphore_t)
                        _dispatch_thread_getspecific(dispatch_sema4_key);
        if (slowpath(!sema)) {
                return _dispatch_thread_semaphore_create();
        }
        _dispatch_thread_setspecific(dispatch_sema4_key, NULL);
        return sema;
}

void
_dispatch_put_thread_semaphore(_dispatch_thread_semaphore_t sema)
{
        _dispatch_thread_semaphore_t old_sema = (_dispatch_thread_semaphore_t)
                        _dispatch_thread_getspecific(dispatch_sema4_key);
        _dispatch_thread_setspecific(dispatch_sema4_key, (void*)sema);
        if (slowpath(old_sema)) {
                return _dispatch_thread_semaphore_dispose(old_sema);
        }
}