summaryrefslogtreecommitdiff
path: root/lib/seq.c
blob: 6581cb06baa164123ffc15ba82842f14a6f9a327 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
/*
 * Copyright (c) 2013, 2014 Nicira, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <config.h>

#include "seq.h"

#include <stdbool.h>

#include "coverage.h"
#include "hash.h"
#include "openvswitch/hmap.h"
#include "latch.h"
#include "openvswitch/list.h"
#include "ovs-thread.h"
#include "openvswitch/poll-loop.h"

COVERAGE_DEFINE(seq_change);

/* A sequence number object. */
struct seq {
    uint64_t value OVS_GUARDED;
    struct hmap waiters OVS_GUARDED; /* Contains 'struct seq_waiter's. */
};

/* A thread waiting on a particular seq. */
struct seq_waiter {
    struct hmap_node hmap_node OVS_GUARDED; /* In 'seq->waiters'. */
    struct seq *seq OVS_GUARDED;            /* Seq being waited for. */
    unsigned int ovsthread_id OVS_GUARDED;  /* Key in 'waiters' hmap. */

    struct seq_thread *thread OVS_GUARDED;  /* Thread preparing to wait. */
    struct ovs_list list_node OVS_GUARDED;  /* In 'thread->waiters'. */

    uint64_t value OVS_GUARDED; /* seq->value we're waiting to change. */
};

/* A thread that might be waiting on one or more seqs. */
struct seq_thread {
    struct ovs_list waiters OVS_GUARDED; /* Contains 'struct seq_waiter's. */
    struct latch latch OVS_GUARDED;  /* Wakeup latch for this thread. */
    bool waiting OVS_GUARDED;        /* True if latch_wait() already called. */
};

static struct ovs_mutex seq_mutex = OVS_MUTEX_INITIALIZER;

static uint64_t seq_next OVS_GUARDED_BY(seq_mutex) = 1;

static pthread_key_t seq_thread_key;

static void seq_init(void);
static struct seq_thread *seq_thread_get(void) OVS_REQUIRES(seq_mutex);
static void seq_thread_exit(void *thread_) OVS_EXCLUDED(seq_mutex);
static void seq_thread_woke(struct seq_thread *) OVS_REQUIRES(seq_mutex);
static void seq_waiter_destroy(struct seq_waiter *) OVS_REQUIRES(seq_mutex);
static void seq_wake_waiters(struct seq *) OVS_REQUIRES(seq_mutex);

/* Creates and returns a new 'seq' object. */
struct seq * OVS_EXCLUDED(seq_mutex)
seq_create(void)
{
    struct seq *seq;

    seq_init();

    seq = xmalloc(sizeof *seq);

    COVERAGE_INC(seq_change);

    ovs_mutex_lock(&seq_mutex);
    seq->value = seq_next++;
    hmap_init(&seq->waiters);
    ovs_mutex_unlock(&seq_mutex);

    return seq;
}

/* Destroys 'seq', waking up threads that were waiting on it, if any. */
void
seq_destroy(struct seq *seq)
     OVS_EXCLUDED(seq_mutex)
{
    ovs_mutex_lock(&seq_mutex);
    seq_wake_waiters(seq);
    hmap_destroy(&seq->waiters);
    free(seq);
    ovs_mutex_unlock(&seq_mutex);
}

int
seq_try_lock(void)
{
    return ovs_mutex_trylock(&seq_mutex);
}

void
seq_lock(void)
    OVS_ACQUIRES(seq_mutex)
{
    ovs_mutex_lock(&seq_mutex);
}

void
seq_unlock(void)
    OVS_RELEASES(seq_mutex)
{
    ovs_mutex_unlock(&seq_mutex);
}

/* Increments 'seq''s sequence number, waking up any threads that are waiting
 * on 'seq'. */
void
seq_change_protected(struct seq *seq)
    OVS_REQUIRES(seq_mutex)
{
    COVERAGE_INC(seq_change);

    seq->value = seq_next++;
    seq_wake_waiters(seq);
}

/* Increments 'seq''s sequence number, waking up any threads that are waiting
 * on 'seq'. */
void
seq_change(struct seq *seq)
    OVS_EXCLUDED(seq_mutex)
{
    ovs_mutex_lock(&seq_mutex);
    seq_change_protected(seq);
    ovs_mutex_unlock(&seq_mutex);
}

/* Returns 'seq''s current sequence number (which could change immediately).
 *
 * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 * when an object changes, even without an ability to lock the object.  See
 * Usage in seq.h for details. */
uint64_t
seq_read_protected(const struct seq *seq)
    OVS_REQUIRES(seq_mutex)
{
    return seq->value;
}

/* Returns 'seq''s current sequence number (which could change immediately).
 *
 * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 * when an object changes, even without an ability to lock the object.  See
 * Usage in seq.h for details. */
uint64_t
seq_read(const struct seq *seq)
    OVS_EXCLUDED(seq_mutex)
{
    uint64_t value;

    ovs_mutex_lock(&seq_mutex);
    value = seq_read_protected(seq);
    ovs_mutex_unlock(&seq_mutex);

    return value;
}

static void
seq_wait__(struct seq *seq, uint64_t value, const char *where)
    OVS_REQUIRES(seq_mutex)
{
    unsigned int id = ovsthread_id_self();
    uint32_t hash = hash_int(id, 0);
    struct seq_waiter *waiter;

    HMAP_FOR_EACH_IN_BUCKET (waiter, hmap_node, hash, &seq->waiters) {
        if (waiter->ovsthread_id == id) {
            if (waiter->value != value) {
                /* The current value is different from the value we've already
                 * waited for, */
                poll_immediate_wake_at(where);
            } else {
                /* Already waiting on 'value', nothing more to do. */
            }
            return;
        }
    }

    waiter = xmalloc(sizeof *waiter);
    waiter->seq = seq;
    hmap_insert(&seq->waiters, &waiter->hmap_node, hash);
    waiter->ovsthread_id = id;
    waiter->value = value;
    waiter->thread = seq_thread_get();
    ovs_list_push_back(&waiter->thread->waiters, &waiter->list_node);

    if (!waiter->thread->waiting) {
        latch_wait_at(&waiter->thread->latch, where);
        waiter->thread->waiting = true;
    }
}

/* Causes the following poll_block() to wake up when 'seq''s sequence number
 * changes from 'value'.  (If 'seq''s sequence number isn't 'value', then
 * poll_block() won't block at all.)
 *
 * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 * when an object changes, even without an ability to lock the object.  See
 * Usage in seq.h for details.
 *
 * ('where' is used in debug logging.  Commonly one would use seq_wait() to
 * automatically provide the caller's source file and line number for
 * 'where'.) */
void
seq_wait_at(const struct seq *seq_, uint64_t value, const char *where)
    OVS_EXCLUDED(seq_mutex)
{
    struct seq *seq = CONST_CAST(struct seq *, seq_);

    ovs_mutex_lock(&seq_mutex);
    if (value == seq->value) {
        seq_wait__(seq, value, where);
    } else {
        poll_immediate_wake_at(where);
    }
    ovs_mutex_unlock(&seq_mutex);
}

/* Called by poll_block() just before it returns, this function destroys any
 * seq_waiter objects associated with the current thread. */
void
seq_woke(void)
    OVS_EXCLUDED(seq_mutex)
{
    struct seq_thread *thread;

    seq_init();

    thread = pthread_getspecific(seq_thread_key);
    if (thread) {
        ovs_mutex_lock(&seq_mutex);
        seq_thread_woke(thread);
        thread->waiting = false;
        ovs_mutex_unlock(&seq_mutex);
    }
}

static void
seq_init(void)
{
    static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;

    if (ovsthread_once_start(&once)) {
        xpthread_key_create(&seq_thread_key, seq_thread_exit);
        ovsthread_once_done(&once);
    }
}

static struct seq_thread *
seq_thread_get(void)
    OVS_REQUIRES(seq_mutex)
{
    struct seq_thread *thread = pthread_getspecific(seq_thread_key);
    if (!thread) {
        thread = xmalloc(sizeof *thread);
        ovs_list_init(&thread->waiters);
        latch_init(&thread->latch);
        thread->waiting = false;

        xpthread_setspecific(seq_thread_key, thread);
    }
    return thread;
}

static void
seq_thread_exit(void *thread_)
    OVS_EXCLUDED(seq_mutex)
{
    struct seq_thread *thread = thread_;

    ovs_mutex_lock(&seq_mutex);
    seq_thread_woke(thread);
    latch_destroy(&thread->latch);
    free(thread);
    ovs_mutex_unlock(&seq_mutex);
}

static void
seq_thread_woke(struct seq_thread *thread)
    OVS_REQUIRES(seq_mutex)
{
    struct seq_waiter *waiter, *next_waiter;

    LIST_FOR_EACH_SAFE (waiter, next_waiter, list_node, &thread->waiters) {
        ovs_assert(waiter->thread == thread);
        seq_waiter_destroy(waiter);
    }
    latch_poll(&thread->latch);
}

static void
seq_waiter_destroy(struct seq_waiter *waiter)
    OVS_REQUIRES(seq_mutex)
{
    hmap_remove(&waiter->seq->waiters, &waiter->hmap_node);
    ovs_list_remove(&waiter->list_node);
    free(waiter);
}

static void
seq_wake_waiters(struct seq *seq)
    OVS_REQUIRES(seq_mutex)
{
    struct seq_waiter *waiter, *next_waiter;

    HMAP_FOR_EACH_SAFE (waiter, next_waiter, hmap_node, &seq->waiters) {
        latch_set(&waiter->thread->latch);
        seq_waiter_destroy(waiter);
    }
}