lib/ovs-thread.c

   1 /*
   2  * Copyright (c) 2013 Nicira, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <config.h>
  18 #include "ovs-thread.h"
  19 #include <errno.h>
  20 #include <poll.h>
  21 #include <stdlib.h>
  22 #include <unistd.h>
  23 #include "compiler.h"
  24 #include "poll-loop.h"
  25 #include "socket-util.h"
  26 #include "util.h"
  27
  28 #ifdef __CHECKER__
  29 /* Omit the definitions in this file because they are somewhat difficult to
  30  * write without prompting "sparse" complaints, without ugliness or
  31  * cut-and-paste.  Since "sparse" is just a checker, not a compiler, it
  32  * doesn't matter that we don't define them. */
  33 #else
  34 #include "vlog.h"
  35
  36 VLOG_DEFINE_THIS_MODULE(ovs_thread);
  37
  38 /* If there is a reason that we cannot fork anymore (unless the fork will be
  39  * immediately followed by an exec), then this points to a string that
  40  * explains why. */
  41 static const char *must_not_fork;
  42
  43 /* True if we created any threads beyond the main initial thread. */
  44 static bool multithreaded;
  45
  46 #define LOCK_FUNCTION(TYPE, FUN) \
  47     void \
  48     ovs_##TYPE##_##FUN##_at(const struct ovs_##TYPE *l_, \
  49                             const char *where) \
  50         OVS_NO_THREAD_SAFETY_ANALYSIS \
  51     { \
  52         struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \
  53         int error = pthread_##TYPE##_##FUN(&l->lock); \
  54         if (OVS_UNLIKELY(error)) { \
  55             ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \
  56         } \
  57         l->where = where; \
  58     }
  59 LOCK_FUNCTION(mutex, lock);
  60 LOCK_FUNCTION(rwlock, rdlock);
  61 LOCK_FUNCTION(rwlock, wrlock);
  62
  63 #define TRY_LOCK_FUNCTION(TYPE, FUN) \
  64     int \
  65     ovs_##TYPE##_##FUN##_at(const struct ovs_##TYPE *l_, \
  66                             const char *where) \
  67         OVS_NO_THREAD_SAFETY_ANALYSIS \
  68     { \
  69         struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \
  70         int error = pthread_##TYPE##_##FUN(&l->lock); \
  71         if (OVS_UNLIKELY(error) && error != EBUSY) { \
  72             ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \
  73         } \
  74         if (!error) { \
  75             l->where = where; \
  76         } \
  77         return error; \
  78     }
  79 TRY_LOCK_FUNCTION(mutex, trylock);
  80 TRY_LOCK_FUNCTION(rwlock, tryrdlock);
  81 TRY_LOCK_FUNCTION(rwlock, trywrlock);
  82
  83 #define UNLOCK_FUNCTION(TYPE, FUN) \
  84     void \
  85     ovs_##TYPE##_##FUN(const struct ovs_##TYPE *l_) \
  86         OVS_NO_THREAD_SAFETY_ANALYSIS \
  87     { \
  88         struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \
  89         int error; \
  90         l->where = NULL; \
  91         error = pthread_##TYPE##_##FUN(&l->lock); \
  92         if (OVS_UNLIKELY(error)) { \
  93             ovs_abort(error, "pthread_%s_%sfailed", #TYPE, #FUN); \
  94         } \
  95     }
  96 UNLOCK_FUNCTION(mutex, unlock);
  97 UNLOCK_FUNCTION(mutex, destroy);
  98 UNLOCK_FUNCTION(rwlock, unlock);
  99 UNLOCK_FUNCTION(rwlock, destroy);
 100
 101 #define XPTHREAD_FUNC1(FUNCTION, PARAM1)                \
 102     void                                                \
 103     x##FUNCTION(PARAM1 arg1)                            \
 104     {                                                   \
 105         int error = FUNCTION(arg1);                     \
 106         if (OVS_UNLIKELY(error)) {                      \
 107             ovs_abort(error, "%s failed", #FUNCTION);   \
 108         }                                               \
 109     }
 110 #define XPTHREAD_FUNC2(FUNCTION, PARAM1, PARAM2)        \
 111     void                                                \
 112     x##FUNCTION(PARAM1 arg1, PARAM2 arg2)               \
 113     {                                                   \
 114         int error = FUNCTION(arg1, arg2);               \
 115         if (OVS_UNLIKELY(error)) {                      \
 116             ovs_abort(error, "%s failed", #FUNCTION);   \
 117         }                                               \
 118     }
 119
 120 XPTHREAD_FUNC1(pthread_mutex_lock, pthread_mutex_t *);
 121 XPTHREAD_FUNC1(pthread_mutex_unlock, pthread_mutex_t *);
 122 XPTHREAD_FUNC1(pthread_mutexattr_init, pthread_mutexattr_t *);
 123 XPTHREAD_FUNC1(pthread_mutexattr_destroy, pthread_mutexattr_t *);
 124 XPTHREAD_FUNC2(pthread_mutexattr_settype, pthread_mutexattr_t *, int);
 125 XPTHREAD_FUNC2(pthread_mutexattr_gettype, pthread_mutexattr_t *, int *);
 126
 127 XPTHREAD_FUNC2(pthread_cond_init, pthread_cond_t *, pthread_condattr_t *);
 128 XPTHREAD_FUNC1(pthread_cond_destroy, pthread_cond_t *);
 129 XPTHREAD_FUNC1(pthread_cond_signal, pthread_cond_t *);
 130 XPTHREAD_FUNC1(pthread_cond_broadcast, pthread_cond_t *);
 131
 132 XPTHREAD_FUNC2(pthread_join, pthread_t, void **);
 133
 134 typedef void destructor_func(void *);
 135 XPTHREAD_FUNC2(pthread_key_create, pthread_key_t *, destructor_func *);
 136 XPTHREAD_FUNC2(pthread_setspecific, pthread_key_t, const void *);
 137
 138 static void
 139 ovs_mutex_init__(const struct ovs_mutex *l_, int type)
 140 {
 141     struct ovs_mutex *l = CONST_CAST(struct ovs_mutex *, l_);
 142     pthread_mutexattr_t attr;
 143     int error;
 144
 145     l->where = NULL;
 146     xpthread_mutexattr_init(&attr);
 147     xpthread_mutexattr_settype(&attr, type);
 148     error = pthread_mutex_init(&l->lock, &attr);
 149     if (OVS_UNLIKELY(error)) {
 150         ovs_abort(error, "pthread_mutex_init failed");
 151     }
 152     xpthread_mutexattr_destroy(&attr);
 153 }
 154
 155 /* Initializes 'mutex' as a normal (non-recursive) mutex. */
 156 void
 157 ovs_mutex_init(const struct ovs_mutex *mutex)
 158 {
 159     ovs_mutex_init__(mutex, PTHREAD_MUTEX_ERRORCHECK);
 160 }
 161
 162 /* Initializes 'mutex' as a recursive mutex. */
 163 void
 164 ovs_mutex_init_recursive(const struct ovs_mutex *mutex)
 165 {
 166     ovs_mutex_init__(mutex, PTHREAD_MUTEX_RECURSIVE);
 167 }
 168
 169 void
 170 ovs_rwlock_init(const struct ovs_rwlock *l_)
 171 {
 172     struct ovs_rwlock *l = CONST_CAST(struct ovs_rwlock *, l_);
 173     int error;
 174
 175     l->where = NULL;
 176     error = pthread_rwlock_init(&l->lock, NULL);
 177     if (OVS_UNLIKELY(error)) {
 178         ovs_abort(error, "pthread_rwlock_init failed");
 179     }
 180 }
 181
 182 void
 183 ovs_mutex_cond_wait(pthread_cond_t *cond, const struct ovs_mutex *mutex_)
 184 {
 185     struct ovs_mutex *mutex = CONST_CAST(struct ovs_mutex *, mutex_);
 186     int error = pthread_cond_wait(cond, &mutex->lock);
 187     if (OVS_UNLIKELY(error)) {
 188         ovs_abort(error, "pthread_cond_wait failed");
 189     }
 190 }
 191 \f
 192 DEFINE_EXTERN_PER_THREAD_DATA(ovsthread_id, 0);
 193
 194 struct ovsthread_aux {
 195     void *(*start)(void *);
 196     void *arg;
 197 };
 198
 199 static void *
 200 ovsthread_wrapper(void *aux_)
 201 {
 202     static atomic_uint next_id = ATOMIC_VAR_INIT(1);
 203
 204     struct ovsthread_aux *auxp = aux_;
 205     struct ovsthread_aux aux;
 206     unsigned int id;
 207
 208     atomic_add(&next_id, 1, &id);
 209     *ovsthread_id_get() = id;
 210
 211     aux = *auxp;
 212     free(auxp);
 213
 214     return aux.start(aux.arg);
 215 }
 216
 217 void
 218 xpthread_create(pthread_t *threadp, pthread_attr_t *attr,
 219                 void *(*start)(void *), void *arg)
 220 {
 221     struct ovsthread_aux *aux;
 222     pthread_t thread;
 223     int error;
 224
 225     forbid_forking("multiple threads exist");
 226     multithreaded = true;
 227
 228     aux = xmalloc(sizeof *aux);
 229     aux->start = start;
 230     aux->arg = arg;
 231
 232     error = pthread_create(threadp ? threadp : &thread, attr,
 233                            ovsthread_wrapper, aux);
 234     if (error) {
 235         ovs_abort(error, "pthread_create failed");
 236     }
 237 }
 238 \f
 239 bool
 240 ovsthread_once_start__(struct ovsthread_once *once)
 241 {
 242     ovs_mutex_lock(&once->mutex);
 243     if (!ovsthread_once_is_done__(once)) {
 244         return false;
 245     }
 246     ovs_mutex_unlock(&once->mutex);
 247     return true;
 248 }
 249
 250 void
 251 ovsthread_once_done(struct ovsthread_once *once)
 252 {
 253     atomic_store(&once->done, true);
 254     ovs_mutex_unlock(&once->mutex);
 255 }
 256 \f
 257 /* Asserts that the process has not yet created any threads (beyond the initial
 258  * thread).
 259  *
 260  * ('where' is used in logging.  Commonly one would use
 261  * assert_single_threaded() to automatically provide the caller's source file
 262  * and line number for 'where'.) */
 263 void
 264 assert_single_threaded_at(const char *where)
 265 {
 266     if (multithreaded) {
 267         VLOG_FATAL("%s: attempted operation not allowed when multithreaded",
 268                    where);
 269     }
 270 }
 271
 272 /* Forks the current process (checking that this is allowed).  Aborts with
 273  * VLOG_FATAL if fork() returns an error, and otherwise returns the value
 274  * returned by fork().
 275  *
 276  * ('where' is used in logging.  Commonly one would use xfork() to
 277  * automatically provide the caller's source file and line number for
 278  * 'where'.) */
 279 pid_t
 280 xfork_at(const char *where)
 281 {
 282     pid_t pid;
 283
 284     if (must_not_fork) {
 285         VLOG_FATAL("%s: attempted to fork but forking not allowed (%s)",
 286                    where, must_not_fork);
 287     }
 288
 289     pid = fork();
 290     if (pid < 0) {
 291         VLOG_FATAL("%s: fork failed (%s)", where, ovs_strerror(errno));
 292     }
 293     return pid;
 294 }
 295
 296 /* Notes that the process must not call fork() from now on, for the specified
 297  * 'reason'.  (The process may still fork() if it execs itself immediately
 298  * afterward.) */
 299 void
 300 forbid_forking(const char *reason)
 301 {
 302     ovs_assert(reason != NULL);
 303     must_not_fork = reason;
 304 }
 305
 306 /* Returns true if the process is allowed to fork, false otherwise. */
 307 bool
 308 may_fork(void)
 309 {
 310     return !must_not_fork;
 311 }
 312 \f
 313 /* Returns the total number of cores on this system, or 0 if the number cannot
 314  * be determined. */
 315 unsigned int
 316 count_cpu_cores(void)
 317 {
 318     long int n_cores = sysconf(_SC_NPROCESSORS_ONLN);
 319
 320     return n_cores > 0 ? n_cores : 0;
 321 }
 322 #endif