#include "ovs-thread.h"
#include <errno.h>
#include <poll.h>
+#ifndef _WIN32
+#include <signal.h>
+#endif
#include <stdlib.h>
#include <unistd.h>
#include "compiler.h"
#include "hash.h"
+#include "netdev-dpdk.h"
#include "ovs-rcu.h"
#include "poll-loop.h"
+#include "seq.h"
#include "socket-util.h"
#include "util.h"
int error; \
\
/* Verify that 'l' was initialized. */ \
- ovs_assert(l->where); \
+ if (OVS_UNLIKELY(!l->where)) { \
+ ovs_abort(0, "%s: %s() passed uninitialized ovs_"#TYPE, \
+ where, __func__); \
+ } \
\
error = pthread_##TYPE##_##FUN(&l->lock); \
if (OVS_UNLIKELY(error)) { \
- ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \
+ ovs_abort(error, "%s: pthread_%s_%s failed", where, #TYPE, #FUN); \
} \
l->where = where; \
}
int error; \
\
/* Verify that 'l' was initialized. */ \
- ovs_assert(l->where); \
+ if (OVS_UNLIKELY(!l->where)) { \
+ ovs_abort(0, "%s: %s() passed uninitialized ovs_"#TYPE, \
+ where, __func__); \
+ } \
\
error = pthread_##TYPE##_##FUN(&l->lock); \
if (OVS_UNLIKELY(error) && error != EBUSY) { \
- ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \
+ ovs_abort(error, "%s: pthread_%s_%s failed", where, #TYPE, #FUN); \
} \
if (!error) { \
l->where = where; \
XPTHREAD_FUNC1(pthread_cond_signal, pthread_cond_t *);
XPTHREAD_FUNC1(pthread_cond_broadcast, pthread_cond_t *);
-XPTHREAD_FUNC3(pthread_barrier_init, pthread_barrier_t *,
- pthread_barrierattr_t *, unsigned int);
-XPTHREAD_FUNC1(pthread_barrier_destroy, pthread_barrier_t *);
-
XPTHREAD_FUNC2(pthread_join, pthread_t, void **);
typedef void destructor_func(void *);
XPTHREAD_FUNC1(pthread_key_delete, pthread_key_t);
XPTHREAD_FUNC2(pthread_setspecific, pthread_key_t, const void *);
+#ifndef _WIN32
+XPTHREAD_FUNC3(pthread_sigmask, int, const sigset_t *, sigset_t *);
+#endif
+
static void
ovs_mutex_init__(const struct ovs_mutex *l_, int type)
{
}
}
-int
-xpthread_barrier_wait(pthread_barrier_t *barrier)
+/* Initializes the 'barrier'. 'size' is the number of threads
+ * expected to hit the barrier. */
+void
+ovs_barrier_init(struct ovs_barrier *barrier, uint32_t size)
{
- int error;
+ barrier->size = size;
+ atomic_count_init(&barrier->count, 0);
+ barrier->seq = seq_create();
+}
- ovsrcu_quiesce_start();
- error = pthread_barrier_wait(barrier);
- ovsrcu_quiesce_end();
+/* Destroys the 'barrier'. */
+void
+ovs_barrier_destroy(struct ovs_barrier *barrier)
+{
+ seq_destroy(barrier->seq);
+}
- if (error && OVS_UNLIKELY(error != PTHREAD_BARRIER_SERIAL_THREAD)) {
- ovs_abort(error, "pthread_barrier_wait failed");
+/* Makes the calling thread block on the 'barrier' until all
+ * 'barrier->size' threads hit the barrier.
+ * ovs_barrier provides the necessary acquire-release semantics to make
+ * the effects of prior memory accesses of all the participating threads
+ * visible on return and to prevent the following memory accesses to be
+ * reordered before the ovs_barrier_block(). */
+void
+ovs_barrier_block(struct ovs_barrier *barrier)
+{
+ uint64_t seq = seq_read(barrier->seq);
+ uint32_t orig;
+
+ orig = atomic_count_inc(&barrier->count);
+ if (orig + 1 == barrier->size) {
+ atomic_count_set(&barrier->count, 0);
+ /* seq_change() serves as a release barrier against the other threads,
+ * so the zeroed count is visible to them as they continue. */
+ seq_change(barrier->seq);
+ } else {
+ /* To prevent thread from waking up by other event,
+ * keeps waiting for the change of 'barrier->seq'. */
+ while (seq == seq_read(barrier->seq)) {
+ seq_wait(barrier->seq, seq);
+ poll_block();
+ }
}
-
- return error;
}
\f
DEFINE_EXTERN_PER_THREAD_DATA(ovsthread_id, 0);
static void *
ovsthread_wrapper(void *aux_)
{
- static atomic_uint next_id = ATOMIC_VAR_INIT(1);
+ static atomic_count next_id = ATOMIC_COUNT_INIT(1);
struct ovsthread_aux *auxp = aux_;
struct ovsthread_aux aux;
unsigned int id;
- atomic_add(&next_id, 1, &id);
+ id = atomic_count_inc(&next_id);
*ovsthread_id_get() = id;
aux = *auxp;
set_subprogram_name("%s%u", aux.name, id);
ovsrcu_quiesce_end();
+ thread_set_nonpmd();
+
return aux.start(aux.arg);
}
ovsthread_once_start__(struct ovsthread_once *once)
{
ovs_mutex_lock(&once->mutex);
- if (!ovsthread_once_is_done__(once)) {
- return false;
+ /* Mutex synchronizes memory, so we get the current value of 'done'. */
+ if (!once->done) {
+ return true;
}
ovs_mutex_unlock(&once->mutex);
- return true;
+ return false;
}
void
ovsthread_once_done(struct ovsthread_once *once)
{
- atomic_store(&once->done, true);
+ /* We need release semantics here, so that the following store may not
+ * be moved ahead of any of the preceding initialization operations.
+ * A release atomic_thread_fence provides that prior memory accesses
+ * will not be reordered to take place after the following store. */
+ atomic_thread_fence(memory_order_release);
+ once->done = true;
ovs_mutex_unlock(&once->mutex);
}
\f
break;
}
- if (cpu & (1 << id)) {
+ if (cpu & (1ULL << id)) {
/* We've already counted this package's cores. */
continue;
}
- cpu |= 1 << id;
+ cpu |= 1ULL << id;
/* Find the number of cores for this package. */
while (fgets(line, sizeof line, stream)) {
projects / cascardo / ovs.git / blobdiff