lib/seq.c

   1 /*
   2  * Copyright (c) 2013, 2014 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
  19 #include "seq.h"
  20
  21 #include <stdbool.h>
  22
  23 #include "hash.h"
  24 #include "hmap.h"
  25 #include "latch.h"
  26 #include "list.h"
  27 #include "ovs-thread.h"
  28 #include "poll-loop.h"
  29
  30 /* A sequence number object. */
  31 struct seq {
  32     uint64_t value OVS_GUARDED;
  33     struct hmap waiters OVS_GUARDED; /* Contains 'struct seq_waiter's. */
  34 };
  35
  36 /* A thread waiting on a particular seq. */
  37 struct seq_waiter {
  38     struct seq *seq OVS_GUARDED;            /* Seq being waited for. */
  39     struct hmap_node hmap_node OVS_GUARDED; /* In 'seq->waiters'. */
  40     unsigned int ovsthread_id OVS_GUARDED;  /* Key in 'waiters' hmap. */
  41
  42     struct seq_thread *thread OVS_GUARDED;  /* Thread preparing to wait. */
  43     struct ovs_list list_node OVS_GUARDED;  /* In 'thread->waiters'. */
  44
  45     uint64_t value OVS_GUARDED; /* seq->value we're waiting to change. */
  46 };
  47
  48 /* A thread that might be waiting on one or more seqs. */
  49 struct seq_thread {
  50     struct ovs_list waiters OVS_GUARDED; /* Contains 'struct seq_waiter's. */
  51     struct latch latch OVS_GUARDED;  /* Wakeup latch for this thread. */
  52     bool waiting OVS_GUARDED;        /* True if latch_wait() already called. */
  53 };
  54
  55 static struct ovs_mutex seq_mutex = OVS_MUTEX_INITIALIZER;
  56
  57 static uint64_t seq_next OVS_GUARDED_BY(seq_mutex) = 1;
  58
  59 static pthread_key_t seq_thread_key;
  60
  61 static void seq_init(void);
  62 static struct seq_thread *seq_thread_get(void) OVS_REQUIRES(seq_mutex);
  63 static void seq_thread_exit(void *thread_) OVS_EXCLUDED(seq_mutex);
  64 static void seq_thread_woke(struct seq_thread *) OVS_REQUIRES(seq_mutex);
  65 static void seq_waiter_destroy(struct seq_waiter *) OVS_REQUIRES(seq_mutex);
  66 static void seq_wake_waiters(struct seq *) OVS_REQUIRES(seq_mutex);
  67
  68 /* Creates and returns a new 'seq' object. */
  69 struct seq * OVS_EXCLUDED(seq_mutex)
  70 seq_create(void)
  71 {
  72     struct seq *seq;
  73
  74     seq_init();
  75
  76     seq = xmalloc(sizeof *seq);
  77     ovs_mutex_lock(&seq_mutex);
  78     seq->value = seq_next++;
  79     hmap_init(&seq->waiters);
  80     ovs_mutex_unlock(&seq_mutex);
  81
  82     return seq;
  83 }
  84
  85 /* Destroys 'seq', waking up threads that were waiting on it, if any. */
  86 void
  87 seq_destroy(struct seq *seq)
  88      OVS_EXCLUDED(seq_mutex)
  89 {
  90     ovs_mutex_lock(&seq_mutex);
  91     seq_wake_waiters(seq);
  92     hmap_destroy(&seq->waiters);
  93     free(seq);
  94     ovs_mutex_unlock(&seq_mutex);
  95 }
  96
  97 /* Increments 'seq''s sequence number, waking up any threads that are waiting
  98  * on 'seq'. */
  99 void
 100 seq_change(struct seq *seq)
 101     OVS_EXCLUDED(seq_mutex)
 102 {
 103     ovs_mutex_lock(&seq_mutex);
 104     seq->value = seq_next++;
 105     seq_wake_waiters(seq);
 106     ovs_mutex_unlock(&seq_mutex);
 107 }
 108
 109 /* Returns 'seq''s current sequence number (which could change immediately).
 110  *
 111  * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 112  * when an object changes, even without an ability to lock the object.  See
 113  * Usage in seq.h for details. */
 114 uint64_t
 115 seq_read(const struct seq *seq)
 116     OVS_EXCLUDED(seq_mutex)
 117 {
 118     uint64_t value;
 119
 120     ovs_mutex_lock(&seq_mutex);
 121     value = seq->value;
 122     ovs_mutex_unlock(&seq_mutex);
 123
 124     return value;
 125 }
 126
 127 static void
 128 seq_wait__(struct seq *seq, uint64_t value, const char *where)
 129     OVS_REQUIRES(seq_mutex)
 130 {
 131     unsigned int id = ovsthread_id_self();
 132     uint32_t hash = hash_int(id, 0);
 133     struct seq_waiter *waiter;
 134
 135     HMAP_FOR_EACH_IN_BUCKET (waiter, hmap_node, hash, &seq->waiters) {
 136         if (waiter->ovsthread_id == id) {
 137             if (waiter->value != value) {
 138                 /* The current value is different from the value we've already
 139                  * waited for, */
 140                 poll_immediate_wake_at(where);
 141             } else {
 142                 /* Already waiting on 'value', nothing more to do. */
 143             }
 144             return;
 145         }
 146     }
 147
 148     waiter = xmalloc(sizeof *waiter);
 149     waiter->seq = seq;
 150     hmap_insert(&seq->waiters, &waiter->hmap_node, hash);
 151     waiter->ovsthread_id = id;
 152     waiter->value = value;
 153     waiter->thread = seq_thread_get();
 154     list_push_back(&waiter->thread->waiters, &waiter->list_node);
 155
 156     if (!waiter->thread->waiting) {
 157         latch_wait_at(&waiter->thread->latch, where);
 158         waiter->thread->waiting = true;
 159     }
 160 }
 161
 162 /* Causes the following poll_block() to wake up when 'seq''s sequence number
 163  * changes from 'value'.  (If 'seq''s sequence number isn't 'value', then
 164  * poll_block() won't block at all.)
 165  *
 166  * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 167  * when an object changes, even without an ability to lock the object.  See
 168  * Usage in seq.h for details.
 169  *
 170  * ('where' is used in debug logging.  Commonly one would use seq_wait() to
 171  * automatically provide the caller's source file and line number for
 172  * 'where'.) */
 173 void
 174 seq_wait_at(const struct seq *seq_, uint64_t value, const char *where)
 175     OVS_EXCLUDED(seq_mutex)
 176 {
 177     struct seq *seq = CONST_CAST(struct seq *, seq_);
 178
 179     ovs_mutex_lock(&seq_mutex);
 180     if (value == seq->value) {
 181         seq_wait__(seq, value, where);
 182     } else {
 183         poll_immediate_wake_at(where);
 184     }
 185     ovs_mutex_unlock(&seq_mutex);
 186 }
 187
 188 /* Called by poll_block() just before it returns, this function destroys any
 189  * seq_waiter objects associated with the current thread. */
 190 void
 191 seq_woke(void)
 192     OVS_EXCLUDED(seq_mutex)
 193 {
 194     struct seq_thread *thread;
 195
 196     seq_init();
 197
 198     thread = pthread_getspecific(seq_thread_key);
 199     if (thread) {
 200         ovs_mutex_lock(&seq_mutex);
 201         seq_thread_woke(thread);
 202         thread->waiting = false;
 203         ovs_mutex_unlock(&seq_mutex);
 204     }
 205 }
 206 \f
 207 static void
 208 seq_init(void)
 209 {
 210     static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;
 211
 212     if (ovsthread_once_start(&once)) {
 213         xpthread_key_create(&seq_thread_key, seq_thread_exit);
 214         ovsthread_once_done(&once);
 215     }
 216 }
 217
 218 static struct seq_thread *
 219 seq_thread_get(void)
 220     OVS_REQUIRES(seq_mutex)
 221 {
 222     struct seq_thread *thread = pthread_getspecific(seq_thread_key);
 223     if (!thread) {
 224         thread = xmalloc(sizeof *thread);
 225         list_init(&thread->waiters);
 226         latch_init(&thread->latch);
 227         thread->waiting = false;
 228
 229         xpthread_setspecific(seq_thread_key, thread);
 230     }
 231     return thread;
 232 }
 233
 234 static void
 235 seq_thread_exit(void *thread_)
 236     OVS_EXCLUDED(seq_mutex)
 237 {
 238     struct seq_thread *thread = thread_;
 239
 240     ovs_mutex_lock(&seq_mutex);
 241     seq_thread_woke(thread);
 242     latch_destroy(&thread->latch);
 243     free(thread);
 244     ovs_mutex_unlock(&seq_mutex);
 245 }
 246
 247 static void
 248 seq_thread_woke(struct seq_thread *thread)
 249     OVS_REQUIRES(seq_mutex)
 250 {
 251     struct seq_waiter *waiter, *next_waiter;
 252
 253     LIST_FOR_EACH_SAFE (waiter, next_waiter, list_node, &thread->waiters) {
 254         ovs_assert(waiter->thread == thread);
 255         seq_waiter_destroy(waiter);
 256     }
 257     latch_poll(&thread->latch);
 258 }
 259
 260 static void
 261 seq_waiter_destroy(struct seq_waiter *waiter)
 262     OVS_REQUIRES(seq_mutex)
 263 {
 264     hmap_remove(&waiter->seq->waiters, &waiter->hmap_node);
 265     list_remove(&waiter->list_node);
 266     free(waiter);
 267 }
 268
 269 static void
 270 seq_wake_waiters(struct seq *seq)
 271     OVS_REQUIRES(seq_mutex)
 272 {
 273     struct seq_waiter *waiter, *next_waiter;
 274
 275     HMAP_FOR_EACH_SAFE (waiter, next_waiter, hmap_node, &seq->waiters) {
 276         latch_set(&waiter->thread->latch);
 277         seq_waiter_destroy(waiter);
 278     }
 279 }