X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fovs-thread.c;h=fbdd7fce0f19d347ea782ae136342660609029fc;hb=HEAD;hp=4dfccaf3e5f8aae2d323615e28e33aa7236626e3;hpb=db19b72b9f6d0d40b743214241d9666116e3b448;p=sliver-openvswitch.git diff --git a/lib/ovs-thread.c b/lib/ovs-thread.c index 4dfccaf3e..fbdd7fce0 100644 --- a/lib/ovs-thread.c +++ b/lib/ovs-thread.c @@ -22,6 +22,7 @@ #include #include "compiler.h" #include "hash.h" +#include "ovs-rcu.h" #include "poll-loop.h" #include "socket-util.h" #include "util.h" @@ -51,12 +52,17 @@ static bool multithreaded; OVS_NO_THREAD_SAFETY_ANALYSIS \ { \ struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \ - int error = pthread_##TYPE##_##FUN(&l->lock); \ + int error; \ + \ + /* Verify that 'l' was initialized. */ \ + ovs_assert(l->where); \ + \ + error = pthread_##TYPE##_##FUN(&l->lock); \ if (OVS_UNLIKELY(error)) { \ ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \ } \ l->where = where; \ - } + } LOCK_FUNCTION(mutex, lock); LOCK_FUNCTION(rwlock, rdlock); LOCK_FUNCTION(rwlock, wrlock); @@ -68,7 +74,12 @@ LOCK_FUNCTION(rwlock, wrlock); OVS_NO_THREAD_SAFETY_ANALYSIS \ { \ struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \ - int error = pthread_##TYPE##_##FUN(&l->lock); \ + int error; \ + \ + /* Verify that 'l' was initialized. */ \ + ovs_assert(l->where); \ + \ + error = pthread_##TYPE##_##FUN(&l->lock); \ if (OVS_UNLIKELY(error) && error != EBUSY) { \ ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \ } \ @@ -81,23 +92,27 @@ TRY_LOCK_FUNCTION(mutex, trylock); TRY_LOCK_FUNCTION(rwlock, tryrdlock); TRY_LOCK_FUNCTION(rwlock, trywrlock); -#define UNLOCK_FUNCTION(TYPE, FUN) \ +#define UNLOCK_FUNCTION(TYPE, FUN, WHERE) \ void \ ovs_##TYPE##_##FUN(const struct ovs_##TYPE *l_) \ OVS_NO_THREAD_SAFETY_ANALYSIS \ { \ struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \ int error; \ - l->where = NULL; \ + \ + /* Verify that 'l' was initialized. */ \ + ovs_assert(l->where); \ + \ + l->where = WHERE; \ error = pthread_##TYPE##_##FUN(&l->lock); \ if (OVS_UNLIKELY(error)) { \ ovs_abort(error, "pthread_%s_%sfailed", #TYPE, #FUN); \ } \ } -UNLOCK_FUNCTION(mutex, unlock); -UNLOCK_FUNCTION(mutex, destroy); -UNLOCK_FUNCTION(rwlock, unlock); -UNLOCK_FUNCTION(rwlock, destroy); +UNLOCK_FUNCTION(mutex, unlock, ""); +UNLOCK_FUNCTION(mutex, destroy, NULL); +UNLOCK_FUNCTION(rwlock, unlock, ""); +UNLOCK_FUNCTION(rwlock, destroy, NULL); #define XPTHREAD_FUNC1(FUNCTION, PARAM1) \ void \ @@ -117,6 +132,15 @@ UNLOCK_FUNCTION(rwlock, destroy); ovs_abort(error, "%s failed", #FUNCTION); \ } \ } +#define XPTHREAD_FUNC3(FUNCTION, PARAM1, PARAM2, PARAM3)\ + void \ + x##FUNCTION(PARAM1 arg1, PARAM2 arg2, PARAM3 arg3) \ + { \ + int error = FUNCTION(arg1, arg2, arg3); \ + if (OVS_UNLIKELY(error)) { \ + ovs_abort(error, "%s failed", #FUNCTION); \ + } \ + } XPTHREAD_FUNC1(pthread_mutex_lock, pthread_mutex_t *); XPTHREAD_FUNC1(pthread_mutex_unlock, pthread_mutex_t *); @@ -136,6 +160,10 @@ XPTHREAD_FUNC1(pthread_cond_destroy, pthread_cond_t *); 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 *); @@ -150,7 +178,7 @@ ovs_mutex_init__(const struct ovs_mutex *l_, int type) pthread_mutexattr_t attr; int error; - l->where = NULL; + l->where = ""; xpthread_mutexattr_init(&attr); xpthread_mutexattr_settype(&attr, type); error = pthread_mutex_init(&l->lock, &attr); @@ -192,7 +220,7 @@ ovs_rwlock_init(const struct ovs_rwlock *l_) pthread_rwlockattr_t attr; int error; - l->where = NULL; + l->where = ""; xpthread_rwlockattr_init(&attr); #ifdef PTHREAD_RWLOCK_WRITER_NONRECURSIVE_INITIALIZER_NP @@ -210,17 +238,39 @@ void ovs_mutex_cond_wait(pthread_cond_t *cond, const struct ovs_mutex *mutex_) { struct ovs_mutex *mutex = CONST_CAST(struct ovs_mutex *, mutex_); - int error = pthread_cond_wait(cond, &mutex->lock); + int error; + + ovsrcu_quiesce_start(); + error = pthread_cond_wait(cond, &mutex->lock); + ovsrcu_quiesce_end(); + if (OVS_UNLIKELY(error)) { ovs_abort(error, "pthread_cond_wait failed"); } } + +int +xpthread_barrier_wait(pthread_barrier_t *barrier) +{ + int error; + + ovsrcu_quiesce_start(); + error = pthread_barrier_wait(barrier); + ovsrcu_quiesce_end(); + + if (error && OVS_UNLIKELY(error != PTHREAD_BARRIER_SERIAL_THREAD)) { + ovs_abort(error, "pthread_barrier_wait failed"); + } + + return error; +} DEFINE_EXTERN_PER_THREAD_DATA(ovsthread_id, 0); struct ovsthread_aux { void *(*start)(void *); void *arg; + char name[16]; }; static void * @@ -238,12 +288,18 @@ ovsthread_wrapper(void *aux_) aux = *auxp; free(auxp); + /* The order of the following calls is important, because + * ovsrcu_quiesce_end() saves a copy of the thread name. */ + set_subprogram_name("%s%u", aux.name, id); + ovsrcu_quiesce_end(); + return aux.start(aux.arg); } -void -xpthread_create(pthread_t *threadp, pthread_attr_t *attr, - void *(*start)(void *), void *arg) +/* Starts a thread that calls 'start(arg)'. Sets the thread's name to 'name' + * (suffixed by its ovsthread_id()). Returns the new thread's pthread_t. */ +pthread_t +ovs_thread_create(const char *name, void *(*start)(void *), void *arg) { struct ovsthread_aux *aux; pthread_t thread; @@ -251,16 +307,18 @@ xpthread_create(pthread_t *threadp, pthread_attr_t *attr, forbid_forking("multiple threads exist"); multithreaded = true; + ovsrcu_quiesce_end(); aux = xmalloc(sizeof *aux); aux->start = start; aux->arg = arg; + ovs_strlcpy(aux->name, name, sizeof aux->name); - error = pthread_create(threadp ? threadp : &thread, attr, - ovsthread_wrapper, aux); + error = pthread_create(&thread, NULL, ovsthread_wrapper, aux); if (error) { ovs_abort(error, "pthread_create failed"); } + return thread; } bool @@ -281,6 +339,12 @@ ovsthread_once_done(struct ovsthread_once *once) ovs_mutex_unlock(&once->mutex); } +bool +single_threaded(void) +{ + return !multithreaded; +} + /* Asserts that the process has not yet created any threads (beyond the initial * thread). * @@ -296,6 +360,7 @@ assert_single_threaded_at(const char *where) } } +#ifndef _WIN32 /* Forks the current process (checking that this is allowed). Aborts with * VLOG_FATAL if fork() returns an error, and otherwise returns the value * returned by fork(). @@ -319,6 +384,7 @@ xfork_at(const char *where) } return pid; } +#endif /* Notes that the process must not call fork() from now on, for the specified * 'reason'. (The process may still fork() if it execs itself immediately @@ -337,83 +403,53 @@ may_fork(void) return !must_not_fork; } -/* ovsthread_counter. - * - * We implement the counter as an array of N_COUNTERS individual counters, each - * with its own lock. Each thread uses one of the counters chosen based on a - * hash of the thread's ID, the idea being that, statistically, different - * threads will tend to use different counters and therefore avoid - * interfering with each other. - * - * Undoubtedly, better implementations are possible. */ +/* ovsthread_stats. */ -/* Basic counter structure. */ -struct ovsthread_counter__ { - struct ovs_mutex mutex; - unsigned long long int value; -}; - -/* Pad the basic counter structure to 64 bytes to avoid cache line - * interference. */ -struct ovsthread_counter { - struct ovsthread_counter__ c; - char pad[ROUND_UP(sizeof(struct ovsthread_counter__), 64) - - sizeof(struct ovsthread_counter__)]; -}; - -#define N_COUNTERS 16 - -struct ovsthread_counter * -ovsthread_counter_create(void) +void +ovsthread_stats_init(struct ovsthread_stats *stats) { - struct ovsthread_counter *c; int i; - c = xmalloc(N_COUNTERS * sizeof *c); - for (i = 0; i < N_COUNTERS; i++) { - ovs_mutex_init(&c[i].c.mutex); - c[i].c.value = 0; + ovs_mutex_init(&stats->mutex); + for (i = 0; i < ARRAY_SIZE(stats->buckets); i++) { + stats->buckets[i] = NULL; } - return c; } void -ovsthread_counter_destroy(struct ovsthread_counter *c) +ovsthread_stats_destroy(struct ovsthread_stats *stats) { - if (c) { - int i; - - for (i = 0; i < N_COUNTERS; i++) { - ovs_mutex_destroy(&c[i].c.mutex); - } - free(c); - } + ovs_mutex_destroy(&stats->mutex); } -void -ovsthread_counter_inc(struct ovsthread_counter *c, unsigned long long int n) +void * +ovsthread_stats_bucket_get(struct ovsthread_stats *stats, + void *(*new_bucket)(void)) { - c = &c[hash_int(ovsthread_id_self(), 0) % N_COUNTERS]; - - ovs_mutex_lock(&c->c.mutex); - c->c.value += n; - ovs_mutex_unlock(&c->c.mutex); + unsigned int idx = ovsthread_id_self() & (ARRAY_SIZE(stats->buckets) - 1); + void *bucket = stats->buckets[idx]; + if (!bucket) { + ovs_mutex_lock(&stats->mutex); + bucket = stats->buckets[idx]; + if (!bucket) { + bucket = stats->buckets[idx] = new_bucket(); + } + ovs_mutex_unlock(&stats->mutex); + } + return bucket; } -unsigned long long int -ovsthread_counter_read(const struct ovsthread_counter *c) +size_t +ovs_thread_stats_next_bucket(const struct ovsthread_stats *stats, size_t i) { - unsigned long long int sum; - int i; - - sum = 0; - for (i = 0; i < N_COUNTERS; i++) { - ovs_mutex_lock(&c[i].c.mutex); - sum += c[i].c.value; - ovs_mutex_unlock(&c[i].c.mutex); + for (; i < ARRAY_SIZE(stats->buckets); i++) { + if (stats->buckets[i]) { + break; + } } - return sum; + return i; } + /* Parses /proc/cpuinfo for the total number of physical cores on this system * across all CPU packages, not counting hyper-threads. @@ -483,10 +519,16 @@ count_cpu_cores(void) static long int n_cores; if (ovsthread_once_start(&once)) { +#ifndef _WIN32 parse_cpuinfo(&n_cores); if (!n_cores) { n_cores = sysconf(_SC_NPROCESSORS_ONLN); } +#else + SYSTEM_INFO sysinfo; + GetSystemInfo(&sysinfo); + n_cores = sysinfo.dwNumberOfProcessors; +#endif ovsthread_once_done(&once); }