- pthread_barrier* patch clean up (address review comments: r1) - pthread_barrier* patch clean up (address review comments: r2) - Abridged test output with 7 microsec usleep, 10 threads, 4 rounds of barrier crossings (tst-pthread-barrier.so) - pthread_barrier* patch clean up (address review comments: r3)
Special return value PTHREAD_BARRIER_SERIAL_THREAD == -1 [Thread 8] crossed barrier with -1 (round 1) [Thread 9] crossed barrier with -1 (round 2) [Thread 9] crossed barrier with -1 (round 3) [Thread 3] crossed barrier with -1 (round 4) Signed-off-by: rean <[email protected]> --- include/osv/latch.hh | 12 +++++ libc/pthread.cc | 116 ++++++++++++++++++++++++++++++++++++++++++- modules/tests/Makefile | 2 +- tests/tst-pthread-barrier.cc | 109 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 237 insertions(+), 2 deletions(-) create mode 100644 tests/tst-pthread-barrier.cc diff --git a/include/osv/latch.hh b/include/osv/latch.hh index 6ff78a8..6b6b349 100644 --- a/include/osv/latch.hh +++ b/include/osv/latch.hh @@ -58,6 +58,18 @@ public: std::unique_lock<std::mutex> l(_mutex); return _condvar.wait_for(l, duration, [&] () -> bool { return is_released(); }); } + + // Useful if latches are being used as a primitive for implementing + // pthread_barrier_t so threads can wait on a barrier multiple times + // (over multiple rounds). There should be no other threads calling + // count_down or await when we're resetting. The caller needs to use + // external locking to avoid disaster (having a thread concurrently + // in count_down or await and one concurrently in reset). Without + // external locking resetting the latch is unsafe. + void unsafe_reset(int count) + { + _count.store(count, std::memory_order_relaxed); + } }; class thread_barrier diff --git a/libc/pthread.cc b/libc/pthread.cc index 148b79e..e8f8ef1 100644 --- a/libc/pthread.cc +++ b/libc/pthread.cc @@ -29,7 +29,7 @@ #include <api/time.h> #include <osv/rwlock.h> - +#include <osv/latch.hh> #include "pthread.hh" namespace pthread_private { @@ -1121,3 +1121,117 @@ int pthread_attr_getaffinity_np(const pthread_attr_t *attr, size_t cpusetsize, return 0; } + +// Private definitions of the internal structs backing pthread_barrier_t and +// pthread_barrierattr_t +typedef struct +{ + unsigned int out; + unsigned int count; + latch *ltch; + pthread_mutex_t *mtx; +} pthread_barrier_t_int; + +typedef struct +{ + unsigned pshared; +} pthread_barrierattr_t_int; + +int pthread_barrier_init(pthread_barrier_t *barrier_opq, + const pthread_barrierattr_t *attr_opq, + unsigned count) +{ + pthread_barrier_t_int *barrier = (pthread_barrier_t_int*) barrier_opq; + static_assert(sizeof(pthread_barrier_t_int) <= sizeof(pthread_barrier_t), + "pthread_barrier_t_int is larger than pthread_barrier_t"); + + // Linux returns EINVAL if count == 0 or INT_MAX so we do too. + // In theory, we could go up to UINT_MAX since count is unsigned. + if (!barrier || count == 0 || count >= INT_MAX) { + return EINVAL; + } + + // Always ignore attr, it has no meaning in the context of a unikernel. + // pthread_barrierattr_t has a single member variable pshared that can be set + // to PTHREAD_PROCESS_PRIVATE or PTHREAD_PROCESS_SHARED. These have the + // same effect in a unikernel - there is only a single process and all + // threads can manipulate the memory area associated with the + // pthread_barrier_t so it doesn't matter what the value of pshared is set to + barrier->count = count; + barrier->out = 0; + barrier->ltch = new latch(count); + barrier->mtx = new pthread_mutex_t; + pthread_mutex_init(barrier->mtx, NULL); + return 0; +} + +int pthread_barrier_wait(pthread_barrier_t *barrier_opq) +{ + pthread_barrier_t_int *barrier = (pthread_barrier_t_int*) barrier_opq; + static_assert(sizeof(pthread_barrier_t_int) <= sizeof(pthread_barrier_t), + "pthread_barrier_t_int is larger than pthread_barrier_t"); + + if (!barrier || !barrier->ltch || !barrier->mtx) { + return EINVAL; + } + + int retval = 0; + pthread_mutex_t *mtx = barrier->mtx; + + pthread_mutex_lock(mtx); + pthread_mutex_unlock(mtx); + + latch *l = barrier->ltch; + l->count_down(); + // All threads stuck here until we get at least 'count' waiters + l->await(); + + // If the last thread (thread x) to wait on the barrier is descheduled here + // (immediately after being the count'th thread crossing the barrier) + // the barrier remains open (a new waiting thread will cross) until + // the barrier is reset below (when thread x is rescheduled), which doesn't + // seem technically incorrect. Only one of the crossing threads will get a + // retval of PTHREAD_BARRIER_SERIAL_THREAD, when + // barrier->out == barrier->count. + // All other crossing threads will get a retval of 0. + + pthread_mutex_lock(mtx); + barrier->out++; + // Make the last thread out responsible for resetting the barrier's latch. + // The last thread also gets the special return value + // PTHREAD_BARRIER_SERIAL_THREAD. Every other thread gets a retval of 0 + if (barrier->out == barrier->count) { + retval = PTHREAD_BARRIER_SERIAL_THREAD; + // Reset the latch for the next round of waiters. We're using an + // external lock (mtx) to ensure that no other thread is calling + // count_down or in await when we're resetting it. Without the external + // lock, resetting the latch isn't safe. + l->unsafe_reset(barrier->count); + // Reset the 'out' counter so that the equality check above works across + // multiple rounds of threads waiting on the barrier + barrier->out = 0; + } + pthread_mutex_unlock(mtx); + return retval; +} + +int pthread_barrier_destroy(pthread_barrier_t *barrier_opq) +{ + pthread_barrier_t_int *barrier = (pthread_barrier_t_int*) barrier_opq; + + static_assert(sizeof(pthread_barrier_t_int) <= sizeof(pthread_barrier_t), + "pthread_barrier_t_int is larger than pthread_barrier_t"); + + if (!barrier || !barrier->ltch || !barrier->mtx) { + return EINVAL; + } + + delete barrier->ltch; + barrier->ltch = nullptr; + + pthread_mutex_destroy(barrier->mtx); + delete barrier->mtx; + barrier->mtx = nullptr; + + return 0; +} diff --git a/modules/tests/Makefile b/modules/tests/Makefile index 3f9cb59..feeba12 100644 --- a/modules/tests/Makefile +++ b/modules/tests/Makefile @@ -85,7 +85,7 @@ tests := tst-pthread.so misc-ramdisk.so tst-vblk.so tst-bsd-evh.so \ payload-merge-env.so misc-execve.so misc-execve-payload.so misc-mutex2.so \ tst-pthread-setcancelstate.so tst-syscall.so tst-pin.so tst-run.so \ tst-ifaddrs.so tst-pthread-affinity-inherit.so tst-sem-timed-wait.so \ - tst-ttyname.so + tst-ttyname.so tst-pthread-barrier.so # libstatic-thread-variable.so tst-static-thread-variable.so \ diff --git a/tests/tst-pthread-barrier.cc b/tests/tst-pthread-barrier.cc new file mode 100644 index 0000000..d9326e5 --- /dev/null +++ b/tests/tst-pthread-barrier.cc @@ -0,0 +1,109 @@ +#include <stdio.h> +#include <unistd.h> +#include <memory.h> +#include <errno.h> +#include <stdbool.h> +#include <pthread.h> +#include <limits.h> +#include <stdlib.h> +#include <atomic> + +unsigned int tests_total = 0, tests_failed = 0; + +void report(const char* name, bool passed) +{ + static const char* status[] = {"FAIL", "PASS"}; + printf("%s: %s\n", status[passed], name); + tests_total += 1; + tests_failed += !passed; +} + +// Opaque type 32 bytes in size +static pthread_barrier_t barrier; +// Opaque type 4 bytes in size +pthread_barrierattr_t attr; +// Number of crossings across the barrier +static int numCrossings; +// Counter to track the number of special return values to threads +static std::atomic<int> specialRetVals; + +static void* thread_func(void *arg) +{ + int threadNum = arg ? *((int*) arg): 0; + int retval = 0; + printf("[Thread %d] starting...\n", threadNum); + + for (int crossing = 0; crossing < numCrossings; crossing++) { + // Force threads to sleep for a random interval so we randomize + // which thread might get the special return value + // PTHREAD_BARRIER_SERIAL_THREAD + int delay = random() % 7 + 1; + usleep(delay); + + printf("[Thread %d] waiting on barrier\n", threadNum); + retval = pthread_barrier_wait(&barrier); + if (retval == PTHREAD_BARRIER_SERIAL_THREAD) { + printf("[Thread %d] crossed barrier with %d\n", threadNum, + PTHREAD_BARRIER_SERIAL_THREAD); + // Increment the counter of special return values + specialRetVals.fetch_add(1, std::memory_order_relaxed); + } else if (retval == 0) { + printf("[Thread %d] crossed barrier with %d\n", threadNum, retval); + } + } + return 0; +} + +int main(void) +{ + // Number of threads that must call into the barrier before they all unblock + const int numThreads = 10; + // Pass through the barrier k times + numCrossings = 4; + specialRetVals.store(0, std::memory_order_relaxed); + pthread_t threads[numThreads]; + int threadIds[numThreads]; + int retval = -1; + printf("Sizeof pthread_barrier_t : %ld\n", sizeof(barrier)); + report("sizeof pthread_barrier_t is 32 bytes\n", sizeof(barrier) == 32); + printf("Sizeof pthread_barrierattr_t: %ld\n", sizeof(attr)); + report("sizeof pthread_barrierattr_t is 4 bytes\n", sizeof(attr) == 4); + + // Try an invalid initialization (-1 or 0 or a null pthread_barrier_t*) + retval = pthread_barrier_init(NULL, NULL, 4); + report("pthread_barrier_init (pthread_barrier_t* == NULL)", + retval == EINVAL); + retval = pthread_barrier_init(&barrier, NULL, -1); + report("pthread_barrier_init (count == -1)", retval == EINVAL); + retval = pthread_barrier_init(&barrier, NULL, 0); + report("pthread_barrier_init (count == 0)", retval == EINVAL); + retval = pthread_barrier_init(&barrier, NULL, INT_MAX); + report("pthread_barrier_init (count == INT_MAX)", retval == EINVAL); + + // Initalize a barrier with NULL attributes. In general + // it doesn't really matter what we do with pthread_barrierattr_t + // PTHREAD_PROCESS_PRIVATE vs PTHREAD_PROCESS_SHARED have the same effect + // in a unikernel - there's only a single process and all threads can + // manipulate the barrier so we can just ignore pthread_barrierattr_t + retval = pthread_barrier_init(&barrier, NULL, numThreads); + report("pthread_barrier_init", retval == 0); + if (retval != 0) { + printf("Early exit, pthread_barrier_init returned %d instead of 0\n", + retval); + goto exit; + } + + for (int t = 0; t < numThreads; t++) { + threadIds[t] = t; + retval = pthread_create(&threads[t], NULL, thread_func, &threadIds[t]); + } + exit: + for (int t = 0; t < numThreads; t++) { + pthread_join(threads[t], NULL); + } + report("pthread_barrier_wait (special retvals)", + specialRetVals.load(std::memory_order_relaxed) == numCrossings); + pthread_barrier_destroy(&barrier); + printf("SUMMARY: %u tests / %u failures\n", tests_total, tests_failed); + return tests_failed == 0 ? 0 : 1; +} -- 2.7.4 -- You received this message because you are subscribed to the Google Groups "OSv Development" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. For more options, visit https://groups.google.com/d/optout.
