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     { \
  51         struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \
  52         int error = pthread_##TYPE##_##FUN(&l->lock); \
  53         if (OVS_UNLIKELY(error)) { \
  54             ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \
  55         } \
  56         l->where = where; \
  57     }
  58 LOCK_FUNCTION(mutex, lock);
  59 LOCK_FUNCTION(rwlock, rdlock);
  60 LOCK_FUNCTION(rwlock, wrlock);
  61
  62 #define TRY_LOCK_FUNCTION(TYPE, FUN) \
  63     int \
  64     ovs_##TYPE##_##FUN##_at(const struct ovs_##TYPE *l_, \
  65                             const char *where) \
  66     { \
  67         struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \
  68         int error = pthread_##TYPE##_##FUN(&l->lock); \
  69         if (OVS_UNLIKELY(error) && error != EBUSY) { \
  70             ovs_abort(error, "pthread_%s_%s failed", #TYPE, #FUN); \
  71         } \
  72         if (!error) { \
  73             l->where = where; \
  74         } \
  75         return error; \
  76     }
  77 TRY_LOCK_FUNCTION(mutex, trylock);
  78 TRY_LOCK_FUNCTION(rwlock, tryrdlock);
  79 TRY_LOCK_FUNCTION(rwlock, trywrlock);
  80
  81 #define UNLOCK_FUNCTION(TYPE, FUN) \
  82     void \
  83     ovs_##TYPE##_##FUN(const struct ovs_##TYPE *l_) \
  84     { \
  85         struct ovs_##TYPE *l = CONST_CAST(struct ovs_##TYPE *, l_); \
  86         int error; \
  87         l->where = NULL; \
  88         error = pthread_##TYPE##_##FUN(&l->lock); \
  89         if (OVS_UNLIKELY(error)) { \
  90             ovs_abort(error, "pthread_%s_%sfailed", #TYPE, #FUN); \
  91         } \
  92     }
  93 UNLOCK_FUNCTION(mutex, unlock);
  94 UNLOCK_FUNCTION(mutex, destroy);
  95 UNLOCK_FUNCTION(rwlock, unlock);
  96 UNLOCK_FUNCTION(rwlock, destroy);
  97
  98 #define XPTHREAD_FUNC1(FUNCTION, PARAM1)                \
  99     void                                                \
 100     x##FUNCTION(PARAM1 arg1)                            \
 101     {                                                   \
 102         int error = FUNCTION(arg1);                     \
 103         if (OVS_UNLIKELY(error)) {                      \
 104             ovs_abort(error, "%s failed", #FUNCTION);   \
 105         }                                               \
 106     }
 107 #define XPTHREAD_FUNC2(FUNCTION, PARAM1, PARAM2)        \
 108     void                                                \
 109     x##FUNCTION(PARAM1 arg1, PARAM2 arg2)               \
 110     {                                                   \
 111         int error = FUNCTION(arg1, arg2);               \
 112         if (OVS_UNLIKELY(error)) {                      \
 113             ovs_abort(error, "%s failed", #FUNCTION);   \
 114         }                                               \
 115     }
 116
 117 XPTHREAD_FUNC1(pthread_mutex_lock, pthread_mutex_t *);
 118 XPTHREAD_FUNC1(pthread_mutex_unlock, pthread_mutex_t *);
 119 XPTHREAD_FUNC1(pthread_mutexattr_init, pthread_mutexattr_t *);
 120 XPTHREAD_FUNC1(pthread_mutexattr_destroy, pthread_mutexattr_t *);
 121 XPTHREAD_FUNC2(pthread_mutexattr_settype, pthread_mutexattr_t *, int);
 122 XPTHREAD_FUNC2(pthread_mutexattr_gettype, pthread_mutexattr_t *, int *);
 123
 124 XPTHREAD_FUNC2(pthread_cond_init, pthread_cond_t *, pthread_condattr_t *);
 125 XPTHREAD_FUNC1(pthread_cond_destroy, pthread_cond_t *);
 126 XPTHREAD_FUNC1(pthread_cond_signal, pthread_cond_t *);
 127 XPTHREAD_FUNC1(pthread_cond_broadcast, pthread_cond_t *);
 128
 129 XPTHREAD_FUNC2(pthread_join, pthread_t, void **);
 130
 131 typedef void destructor_func(void *);
 132 XPTHREAD_FUNC2(pthread_key_create, pthread_key_t *, destructor_func *);
 133 XPTHREAD_FUNC2(pthread_setspecific, pthread_key_t, const void *);
 134
 135 static void
 136 ovs_mutex_init__(const struct ovs_mutex *l_, int type)
 137 {
 138     struct ovs_mutex *l = CONST_CAST(struct ovs_mutex *, l_);
 139     pthread_mutexattr_t attr;
 140     int error;
 141
 142     l->where = NULL;
 143     xpthread_mutexattr_init(&attr);
 144     xpthread_mutexattr_settype(&attr, type);
 145     error = pthread_mutex_init(&l->lock, &attr);
 146     if (OVS_UNLIKELY(error)) {
 147         ovs_abort(error, "pthread_mutex_init failed");
 148     }
 149     xpthread_mutexattr_destroy(&attr);
 150 }
 151
 152 /* Initializes 'mutex' as a normal (non-recursive) mutex. */
 153 void
 154 ovs_mutex_init(const struct ovs_mutex *mutex)
 155 {
 156     ovs_mutex_init__(mutex, PTHREAD_MUTEX_ERRORCHECK);
 157 }
 158
 159 /* Initializes 'mutex' as a recursive mutex. */
 160 void
 161 ovs_mutex_init_recursive(const struct ovs_mutex *mutex)
 162 {
 163     ovs_mutex_init__(mutex, PTHREAD_MUTEX_RECURSIVE);
 164 }
 165
 166 void
 167 ovs_rwlock_init(const struct ovs_rwlock *l_)
 168 {
 169     struct ovs_rwlock *l = CONST_CAST(struct ovs_rwlock *, l_);
 170     int error;
 171
 172     l->where = NULL;
 173     error = pthread_rwlock_init(&l->lock, NULL);
 174     if (OVS_UNLIKELY(error)) {
 175         ovs_abort(error, "pthread_rwlock_init failed");
 176     }
 177 }
 178
 179 void
 180 ovs_mutex_cond_wait(pthread_cond_t *cond, const struct ovs_mutex *mutex_)
 181 {
 182     struct ovs_mutex *mutex = CONST_CAST(struct ovs_mutex *, mutex_);
 183     int error = pthread_cond_wait(cond, &mutex->lock);
 184     if (OVS_UNLIKELY(error)) {
 185         ovs_abort(error, "pthread_cond_wait failed");
 186     }
 187 }
 188 \f
 189 DEFINE_EXTERN_PER_THREAD_DATA(ovsthread_id, 0);
 190
 191 struct ovsthread_aux {
 192     void *(*start)(void *);
 193     void *arg;
 194 };
 195
 196 static void *
 197 ovsthread_wrapper(void *aux_)
 198 {
 199     static atomic_uint next_id = ATOMIC_VAR_INIT(1);
 200
 201     struct ovsthread_aux *auxp = aux_;
 202     struct ovsthread_aux aux;
 203     unsigned int id;
 204
 205     atomic_add(&next_id, 1, &id);
 206     *ovsthread_id_get() = id;
 207
 208     aux = *auxp;
 209     free(auxp);
 210
 211     return aux.start(aux.arg);
 212 }
 213
 214 void
 215 xpthread_create(pthread_t *threadp, pthread_attr_t *attr,
 216                 void *(*start)(void *), void *arg)
 217 {
 218     struct ovsthread_aux *aux;
 219     pthread_t thread;
 220     int error;
 221
 222     forbid_forking("multiple threads exist");
 223     multithreaded = true;
 224
 225     aux = xmalloc(sizeof *aux);
 226     aux->start = start;
 227     aux->arg = arg;
 228
 229     error = pthread_create(threadp ? threadp : &thread, attr,
 230                            ovsthread_wrapper, aux);
 231     if (error) {
 232         ovs_abort(error, "pthread_create failed");
 233     }
 234 }
 235 \f
 236 bool
 237 ovsthread_once_start__(struct ovsthread_once *once)
 238 {
 239     ovs_mutex_lock(&once->mutex);
 240     if (!ovsthread_once_is_done__(once)) {
 241         return false;
 242     }
 243     ovs_mutex_unlock(&once->mutex);
 244     return true;
 245 }
 246
 247 void
 248 ovsthread_once_done(struct ovsthread_once *once)
 249 {
 250     atomic_store(&once->done, true);
 251     ovs_mutex_unlock(&once->mutex);
 252 }
 253 \f
 254 /* Asserts that the process has not yet created any threads (beyond the initial
 255  * thread).
 256  *
 257  * ('where' is used in logging.  Commonly one would use
 258  * assert_single_threaded() to automatically provide the caller's source file
 259  * and line number for 'where'.) */
 260 void
 261 assert_single_threaded_at(const char *where)
 262 {
 263     if (multithreaded) {
 264         VLOG_FATAL("%s: attempted operation not allowed when multithreaded",
 265                    where);
 266     }
 267 }
 268
 269 /* Forks the current process (checking that this is allowed).  Aborts with
 270  * VLOG_FATAL if fork() returns an error, and otherwise returns the value
 271  * returned by fork().
 272  *
 273  * ('where' is used in logging.  Commonly one would use xfork() to
 274  * automatically provide the caller's source file and line number for
 275  * 'where'.) */
 276 pid_t
 277 xfork_at(const char *where)
 278 {
 279     pid_t pid;
 280
 281     if (must_not_fork) {
 282         VLOG_FATAL("%s: attempted to fork but forking not allowed (%s)",
 283                    where, must_not_fork);
 284     }
 285
 286     pid = fork();
 287     if (pid < 0) {
 288         VLOG_FATAL("%s: fork failed (%s)", where, ovs_strerror(errno));
 289     }
 290     return pid;
 291 }
 292
 293 /* Notes that the process must not call fork() from now on, for the specified
 294  * 'reason'.  (The process may still fork() if it execs itself immediately
 295  * afterward.) */
 296 void
 297 forbid_forking(const char *reason)
 298 {
 299     ovs_assert(reason != NULL);
 300     must_not_fork = reason;
 301 }
 302
 303 /* Returns true if the process is allowed to fork, false otherwise. */
 304 bool
 305 may_fork(void)
 306 {
 307     return !must_not_fork;
 308 }
 309 #endif