From f0bf079dca01acb6dfc07af03226a0cae7a90010 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Sun, 1 Jan 2017 13:32:32 -0800 Subject: New thread infrastructure -- not used anywhere yet, but tested. --- src/CMakeLists.txt | 2 + src/core/nng_impl.h | 1 + src/core/platform.h | 16 ++++ src/core/thread.c | 150 +++++++++++++++++++++++++++++++++ src/core/thread.h | 52 ++++++++++++ src/platform/posix/posix_impl.h | 17 ++++ src/platform/posix/posix_thread.c | 171 +++++++++++++++++++++++++++++++++++++- 7 files changed, 407 insertions(+), 2 deletions(-) create mode 100644 src/core/thread.c create mode 100644 src/core/thread.h (limited to 'src') diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 0f10d2ac..4b09bedc 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -51,6 +51,8 @@ set (NNG_SOURCES core/protocol.h core/socket.c core/socket.h + core/thread.c + core/thread.h core/transport.c core/transport.h diff --git a/src/core/nng_impl.h b/src/core/nng_impl.h index 46341863..4c00a522 100644 --- a/src/core/nng_impl.h +++ b/src/core/nng_impl.h @@ -31,6 +31,7 @@ #include "core/panic.h" #include "core/platform.h" #include "core/protocol.h" +#include "core/thread.h" #include "core/transport.h" // These have to come after the others - particularly transport.h diff --git a/src/core/platform.h b/src/core/platform.h index 21793086..bdec8bb6 100644 --- a/src/core/platform.h +++ b/src/core/platform.h @@ -63,41 +63,53 @@ extern void nni_free(void *, size_t); typedef struct nni_mutex nni_mutex; typedef struct nni_cond nni_cond; +typedef struct nni_plat_mtx nni_plat_mtx; +typedef struct nni_plat_cv nni_plat_cv; +typedef struct nni_plat_thr nni_plat_thr; + // Mutex handling. // nni_mutex_init initializes a mutex structure. This may require dynamic // allocation, depending on the platform. It can return NNG_ENOMEM if that // fails. +extern int nni_plat_mtx_init(nni_plat_mtx *); extern int nni_mutex_init(nni_mutex *); // nni_mutex_fini destroys the mutex and releases any resources allocated for // it's use. extern void nni_mutex_fini(nni_mutex *); +extern void nni_plat_mtx_fini(nni_plat_mtx *); // nni_mutex_enter locks the mutex. This is not recursive -- a mutex can only // be entered once. extern void nni_mutex_enter(nni_mutex *); +extern void nni_plat_mtx_lock(nni_plat_mtx *); // nni_mutex_exit unlocks the mutex. This can only be performed by the thread // that owned the mutex. extern void nni_mutex_exit(nni_mutex *); +extern void nni_plat_mtx_unlock(nni_plat_mtx *); // nni_mutex_tryenter tries to lock the mutex. If it can't, it may return // NNG_EBUSY. extern int nni_mutex_tryenter(nni_mutex *); +extern int nni_plat_mtx_trylock(nni_plat_mtx *); // nni_cond_init initializes a condition variable. We require a mutex be // supplied with it, and that mutex must always be held when performing any // operations on the condition variable (other than fini.) This may require // dynamic allocation, and if so this operation may fail with NNG_ENOMEM. extern int nni_cond_init(nni_cond *, nni_mutex *); +extern int nni_plat_cv_init(nni_plat_cv *, nni_plat_mtx *); // nni_cond_fini releases all resources associated with condition variable. extern void nni_cond_fini(nni_cond *); +extern void nni_plat_cv_fini(nni_plat_cv *); // nni_cond_broadcast wakes all waiters on the condition. This should be // called with the lock held. extern void nni_cond_broadcast(nni_cond *); +extern void nni_plat_cv_wake(nni_plat_cv *); // nni_cond_signal wakes a signal waiter. extern void nni_cond_signal(nni_cond *); @@ -106,23 +118,27 @@ extern void nni_cond_signal(nni_cond *); // associated lock is atomically released and reacquired upon wake up. // Callers can be spuriously woken. The associated lock must be held. extern void nni_cond_wait(nni_cond *); +extern void nni_plat_cv_wait(nni_plat_cv *); // nni_cond_waituntil waits for a wakeup on the condition variable, or // until the system time reaches the specified absolute time. (It is an // absolute form of nni_cond_timedwait.) Early wakeups are possible, so // check the condition. It will return either NNG_ETIMEDOUT, or 0. extern int nni_cond_waituntil(nni_cond *, nni_time); +extern int nni_plat_cv_until(nni_plat_cv *, nni_time); typedef struct nni_thread nni_thread; // nni_thread_creates a thread that runs the given function. The thread // receives a single argument. extern int nni_thread_create(nni_thread **, void (*fn)(void *), void *); +extern int nni_plat_thr_init(nni_plat_thr *, void (*)(void *), void *); // nni_thread_reap waits for the thread to exit, and then releases any // resources associated with the thread. After this returns, it // is an error to reference the thread in any further way. extern void nni_thread_reap(nni_thread *); +extern void nni_plat_thr_fini(nni_plat_thr *); // nn_clock returns a number of microseconds since some arbitrary time // in the past. The values returned by nni_clock must use the same base diff --git a/src/core/thread.c b/src/core/thread.c new file mode 100644 index 00000000..4f714341 --- /dev/null +++ b/src/core/thread.c @@ -0,0 +1,150 @@ +// +// Copyright 2016 Garrett D'Amore +// +// This software is supplied under the terms of the MIT License, a +// copy of which should be located in the distribution where this +// file was obtained (LICENSE.txt). A copy of the license may also be +// found online at https://opensource.org/licenses/MIT. +// + +#include "core/nng_impl.h" + +int +nni_mtx_init(nni_mtx *mtx) +{ + return (nni_plat_mtx_init(&mtx->mtx)); +} + +void +nni_mtx_fini(nni_mtx *mtx) +{ + nni_plat_mtx_fini(&mtx->mtx); +} + +void +nni_mtx_lock(nni_mtx *mtx) +{ + nni_plat_mtx_lock(&mtx->mtx); +} + +void +nni_mtx_unlock(nni_mtx *mtx) +{ + nni_plat_mtx_unlock(&mtx->mtx); +} + +int +nni_mtx_trylock(nni_mtx *mtx) +{ + return (nni_plat_mtx_trylock(&mtx->mtx)); +} + +int +nni_cv_init(nni_cv *cv, nni_mtx *mtx) +{ + return (nni_plat_cv_init(&cv->cv, &mtx->mtx)); +} + +void +nni_cv_fini(nni_cv *cv) +{ + nni_plat_cv_fini(&cv->cv); +} + +void +nni_cv_wait(nni_cv *cv) +{ + nni_plat_cv_wait(&cv->cv); +} + +int +nni_cv_until(nni_cv *cv, nni_time until) +{ + // Some special cases for times. Catching these here means that + // platforms can assume a valid time is presented to them. + if (until == NNI_TIME_NEVER) { + nni_plat_cv_wait(&cv->cv); + return (0); + } + if (until == NNI_TIME_ZERO) { + return (NNG_EAGAIN); + } + + return (nni_plat_cv_until(&cv->cv, until)); +} + +void +nni_cv_wake(nni_cv *cv) +{ + return (nni_plat_cv_wake(&cv->cv)); +} + +static void +nni_thr_wrap(void *arg) +{ + nni_thr *thr = arg; + int stop; + + nni_plat_mtx_lock(&thr->mtx); + while (((stop = thr->stop) == 0) && (thr->start == 0)) { + nni_plat_cv_wait(&thr->cv); + } + nni_plat_mtx_unlock(&thr->mtx); + if (!stop) { + thr->fn(thr->arg); + } + nni_plat_mtx_lock(&thr->mtx); + thr->done = 1; + nni_plat_cv_wake(&thr->cv); + nni_plat_mtx_unlock(&thr->mtx); +} + +int +nni_thr_init(nni_thr *thr, nni_thr_func fn, void *arg) +{ + int rv; + + thr->done = 0; + thr->start = 0; + thr->stop = 0; + thr->fn = fn; + thr->arg = arg; + + if ((rv = nni_plat_mtx_init(&thr->mtx)) != 0) { + return (rv); + } + if ((rv = nni_plat_cv_init(&thr->cv, &thr->mtx)) != 0) { + nni_plat_mtx_fini(&thr->mtx); + return (rv); + } + if ((rv = nni_plat_thr_init(&thr->thr, nni_thr_wrap, thr)) != 0) { + nni_plat_cv_fini(&thr->cv); + nni_plat_mtx_fini(&thr->mtx); + return (rv); + } + return (0); +} + +void +nni_thr_run(nni_thr *thr) +{ + nni_plat_mtx_lock(&thr->mtx); + thr->start = 1; + nni_plat_cv_wake(&thr->cv); + nni_plat_mtx_unlock(&thr->mtx); +} + +void +nni_thr_fini(nni_thr *thr) +{ + nni_plat_mtx_lock(&thr->mtx); + thr->stop = 1; + nni_plat_cv_wake(&thr->cv); + while (!thr->done) { + nni_plat_cv_wait(&thr->cv); + } + nni_plat_mtx_unlock(&thr->mtx); + nni_plat_thr_fini(&thr->thr); + nni_plat_cv_fini(&thr->cv); + nni_plat_mtx_fini(&thr->mtx); +} \ No newline at end of file diff --git a/src/core/thread.h b/src/core/thread.h new file mode 100644 index 00000000..062941a1 --- /dev/null +++ b/src/core/thread.h @@ -0,0 +1,52 @@ +// +// Copyright 2016 Garrett D'Amore +// +// This software is supplied under the terms of the MIT License, a +// copy of which should be located in the distribution where this +// file was obtained (LICENSE.txt). A copy of the license may also be +// found online at https://opensource.org/licenses/MIT. +// + +#ifndef CORE_THREAD_H +#define CORE_THREAD_H + +#include "core/nng_impl.h" + +typedef struct { + nni_plat_mtx mtx; +} nni_mtx; + +typedef struct { + nni_plat_cv cv; +} nni_cv; + +typedef void (*nni_thr_func)(void *); + +typedef struct { + nni_plat_thr thr; + nni_plat_mtx mtx; + nni_plat_cv cv; + nni_thr_func fn; + void *arg; + int start; + int stop; + int done; +} nni_thr; + +extern int nni_mtx_init(nni_mtx *mtx); +extern void nni_mtx_fini(nni_mtx *mtx); +extern void nni_mtx_lock(nni_mtx *mtx); +extern void nni_mtx_unlock(nni_mtx *mtx); +extern int nni_mtx_trylock(nni_mtx *mtx); + +extern int nni_cv_init(nni_cv *cv, nni_mtx *); +extern void nni_cv_fini(nni_cv *cv); +extern void nni_cv_wake(nni_cv *cv); +extern void nni_cv_wait(nni_cv *cv); +extern int nni_cv_until(nni_cv *cv, nni_time when); + +extern int nni_thr_init(nni_thr *thr, nni_thr_func fn, void *arg); +extern void nni_thr_fini(nni_thr *thr); +extern void nni_thr_run(nni_thr *thr); + +#endif CORE_THREAD_H \ No newline at end of file diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h index 2df96c71..5ca7812f 100644 --- a/src/platform/posix/posix_impl.h +++ b/src/platform/posix/posix_impl.h @@ -41,6 +41,23 @@ struct nni_cond { pthread_cond_t cv; pthread_mutex_t * mx; }; + +// These types are provided for here, to permit them to be directly inlined +// elsewhere. + +struct nni_plat_mtx { + pthread_mutex_t mtx; +}; + +struct nni_plat_thr { + pthread_t tid; +}; + +struct nni_plat_cv { + pthread_cond_t cv; + pthread_mutex_t *mtx; +}; + #endif #endif // PLATFORM_POSIX_IMPL_H diff --git a/src/platform/posix/posix_thread.c b/src/platform/posix/posix_thread.c index f5cad7a9..0c928d84 100644 --- a/src/platform/posix/posix_thread.c +++ b/src/platform/posix/posix_thread.c @@ -31,6 +31,9 @@ static int nni_plat_inited = 0; static int nni_plat_forked = 0; static int nni_plat_next = 0; +pthread_condattr_t nni_cvattr; +pthread_mutexattr_t nni_mxattr; + uint32_t nni_plat_nextid(void) { @@ -86,6 +89,172 @@ nni_thread_reap(nni_thread *thr) } +int +nni_plat_mtx_init(nni_plat_mtx *mtx) +{ + int rv; + + if ((rv = pthread_mutex_init(&mtx->mtx, &nni_mxattr)) != 0) { + switch (rv) { + case EAGAIN: + case ENOMEM: + return (NNG_ENOMEM); + + default: + nni_panic("pthread_mutex_init: %s", strerror(rv)); + } + } + return (0); +} + + +void +nni_plat_mtx_fini(nni_plat_mtx *mtx) +{ + int rv; + + if ((rv = pthread_mutex_destroy(&mtx->mtx)) != 0) { + nni_panic("pthread_mutex_fini: %s", strerror(rv)); + } +} + + +void +nni_plat_mtx_lock(nni_plat_mtx *mtx) +{ + int rv; + + if ((rv = pthread_mutex_lock(&mtx->mtx)) != 0) { + nni_panic("pthread_mutex_lock: %s", strerror(rv)); + } +} + + +void +nni_plat_mtx_unlock(nni_plat_mtx *mtx) +{ + int rv; + + if ((rv = pthread_mutex_unlock(&mtx->mtx)) != 0) { + nni_panic("pthread_mutex_unlock: %s", strerror(rv)); + } +} + + +int +nni_plat_mtx_trylock(nni_plat_mtx *mtx) +{ + int rv; + + if ((rv = pthread_mutex_trylock(&mtx->mtx)) == EBUSY) { + return (NNG_EBUSY); + } + if (rv != 0) { + nni_panic("pthread_mutex_trylock: %s", strerror(rv)); + } + return (0); +} + + +int +nni_plat_cv_init(nni_plat_cv *cv, nni_plat_mtx *mtx) +{ + int rv; + + if ((rv = pthread_cond_init(&cv->cv, &nni_cvattr)) != 0) { + switch (rv) { + case ENOMEM: + case EAGAIN: + return (NNG_ENOMEM); + + default: + nni_panic("pthread_cond_init: %s", strerror(rv)); + } + } + cv->mtx = &mtx->mtx; + return (0); +} + +void +nni_plat_cv_wake(nni_plat_cv *cv) +{ + int rv; + + if ((rv = pthread_cond_broadcast(&cv->cv)) != 0) { + nni_panic("pthread_cond_broadcast: %s", strerror(rv)); + } +} + +void +nni_plat_cv_wait(nni_plat_cv *cv) +{ + int rv; + + if ((rv = pthread_cond_wait(&cv->cv, cv->mtx)) != 0) { + nni_panic("pthread_cond_wait: %s", strerror(rv)); + } +} + +int +nni_plat_cv_until(nni_plat_cv *cv, nni_time until) +{ + struct timespec ts; + int rv; + + // Our caller has already guaranteed a sane value for until. + ts.tv_sec = until / 1000000; + ts.tv_nsec = (until % 1000000) * 1000; + + rv = pthread_cond_timedwait(&cv->cv, cv->mtx, &ts); + if (rv == ETIMEDOUT) { + if (nni_clock() < until) { + // Buggy pthreads implementation!! Seen with + // CLOCK_MONOTONIC on macOS Sierra. + nni_panic("nni_plat_cv_until: Premature wake up!"); + } + return (NNG_ETIMEDOUT); + } else if (rv != 0) { + nni_panic("pthread_cond_timedwait: %d", rv); + } + return (0); + +} + +void +nni_plat_cv_fini(nni_plat_cv *cv) +{ + int rv; + + if ((rv = pthread_cond_destroy(&cv->cv)) != 0) { + nni_panic("pthread_cond_destroy: %s", strerror(rv)); + } +} + +int +nni_plat_thr_init(nni_plat_thr *thr, void (*fn)(void *), void *arg) +{ + int rv; + + // POSIX wants functions to return a void *, but we don't care. + if ((rv = pthread_create(&thr->tid, NULL, (void *) fn, arg)) != 0) { + //nni_printf("pthread_create: %s", strerror(rv)); + return (NNG_ENOMEM); + } + return (0); +} + + +void +nni_plat_thr_fini(nni_plat_thr *thr) +{ + int rv; + + if ((rv = pthread_join(thr->tid, NULL))) { + nni_panic("pthread_join: %s", strerror(rv)); + } +} + + void nni_atfork_child(void) { @@ -93,8 +262,6 @@ nni_atfork_child(void) } -pthread_condattr_t nni_cvattr; -pthread_mutexattr_t nni_mxattr; int nni_plat_init(int (*helper)(void)) -- cgit v1.2.3-70-g09d2