From: rean' via OSv Development <[email protected]>
Committer: Nadav Har'El <[email protected]>
Branch: master
pthread_barrier*: Add OSv-specific pthread_barrier_init, wait, destroy +
test
- 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)
- reformat patch per OSv style (r1)
- reformat patch per OSv style (r2)
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]>
Message-Id: <[email protected]>
---
diff --git a/include/osv/latch.hh b/include/osv/latch.hh
--- 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
--- 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 {
@@ -1128,3 +1128,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
--- 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
--- a/tests/tst-pthread-barrier.cc
+++ 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;
+}
--
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.
