ovsdb: Add ovsdb-client options for testing lock
[cascardo/ovs.git] / lib / ovs-rcu.c
1 /*
2  * Copyright (c) 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 #include <errno.h>
19 #include "ovs-rcu.h"
20 #include "fatal-signal.h"
21 #include "guarded-list.h"
22 #include "openvswitch/list.h"
23 #include "ovs-thread.h"
24 #include "poll-loop.h"
25 #include "seq.h"
26 #include "timeval.h"
27 #include "openvswitch/vlog.h"
28
29 VLOG_DEFINE_THIS_MODULE(ovs_rcu);
30
31 struct ovsrcu_cb {
32     void (*function)(void *aux);
33     void *aux;
34 };
35
36 struct ovsrcu_cbset {
37     struct ovs_list list_node;
38     struct ovsrcu_cb cbs[16];
39     int n_cbs;
40 };
41
42 struct ovsrcu_perthread {
43     struct ovs_list list_node;  /* In global list. */
44
45     struct ovs_mutex mutex;
46     uint64_t seqno;
47     struct ovsrcu_cbset *cbset;
48     char name[16];              /* This thread's name. */
49 };
50
51 static struct seq *global_seqno;
52
53 static pthread_key_t perthread_key;
54 static struct ovs_list ovsrcu_threads;
55 static struct ovs_mutex ovsrcu_threads_mutex;
56
57 static struct guarded_list flushed_cbsets;
58 static struct seq *flushed_cbsets_seq;
59
60 static void ovsrcu_init_module(void);
61 static void ovsrcu_flush_cbset__(struct ovsrcu_perthread *, bool);
62 static void ovsrcu_flush_cbset(struct ovsrcu_perthread *);
63 static void ovsrcu_unregister__(struct ovsrcu_perthread *);
64 static bool ovsrcu_call_postponed(void);
65 static void *ovsrcu_postpone_thread(void *arg OVS_UNUSED);
66
67 static struct ovsrcu_perthread *
68 ovsrcu_perthread_get(void)
69 {
70     struct ovsrcu_perthread *perthread;
71
72     ovsrcu_init_module();
73
74     perthread = pthread_getspecific(perthread_key);
75     if (!perthread) {
76         const char *name = get_subprogram_name();
77
78         perthread = xmalloc(sizeof *perthread);
79         ovs_mutex_init(&perthread->mutex);
80         perthread->seqno = seq_read(global_seqno);
81         perthread->cbset = NULL;
82         ovs_strlcpy(perthread->name, name[0] ? name : "main",
83                     sizeof perthread->name);
84
85         ovs_mutex_lock(&ovsrcu_threads_mutex);
86         ovs_list_push_back(&ovsrcu_threads, &perthread->list_node);
87         ovs_mutex_unlock(&ovsrcu_threads_mutex);
88
89         pthread_setspecific(perthread_key, perthread);
90     }
91     return perthread;
92 }
93
94 /* Indicates the end of a quiescent state.  See "Details" near the top of
95  * ovs-rcu.h.
96  *
97  * Quiescent states don't stack or nest, so this always ends a quiescent state
98  * even if ovsrcu_quiesce_start() was called multiple times in a row. */
99 void
100 ovsrcu_quiesce_end(void)
101 {
102     ovsrcu_perthread_get();
103 }
104
105 static void
106 ovsrcu_quiesced(void)
107 {
108     if (single_threaded()) {
109         ovsrcu_call_postponed();
110     } else {
111         static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;
112         if (ovsthread_once_start(&once)) {
113             ovs_thread_create("urcu", ovsrcu_postpone_thread, NULL);
114             ovsthread_once_done(&once);
115         }
116     }
117 }
118
119 /* Indicates the beginning of a quiescent state.  See "Details" near the top of
120  * ovs-rcu.h. */
121 void
122 ovsrcu_quiesce_start(void)
123 {
124     struct ovsrcu_perthread *perthread;
125
126     ovsrcu_init_module();
127     perthread = pthread_getspecific(perthread_key);
128     if (perthread) {
129         pthread_setspecific(perthread_key, NULL);
130         ovsrcu_unregister__(perthread);
131     }
132
133     ovsrcu_quiesced();
134 }
135
136 /* Indicates a momentary quiescent state.  See "Details" near the top of
137  * ovs-rcu.h.
138  *
139  * Provides a full memory barrier via seq_change().
140  */
141 void
142 ovsrcu_quiesce(void)
143 {
144     struct ovsrcu_perthread *perthread;
145
146     perthread = ovsrcu_perthread_get();
147     perthread->seqno = seq_read(global_seqno);
148     if (perthread->cbset) {
149         ovsrcu_flush_cbset(perthread);
150     }
151     seq_change(global_seqno);
152
153     ovsrcu_quiesced();
154 }
155
156 int
157 ovsrcu_try_quiesce(void)
158 {
159     struct ovsrcu_perthread *perthread;
160     int ret = EBUSY;
161
162     ovs_assert(!single_threaded());
163     perthread = ovsrcu_perthread_get();
164     if (!seq_try_lock()) {
165         perthread->seqno = seq_read_protected(global_seqno);
166         if (perthread->cbset) {
167             ovsrcu_flush_cbset__(perthread, true);
168         }
169         seq_change_protected(global_seqno);
170         seq_unlock();
171         ovsrcu_quiesced();
172         ret = 0;
173     }
174     return ret;
175 }
176
177 bool
178 ovsrcu_is_quiescent(void)
179 {
180     ovsrcu_init_module();
181     return pthread_getspecific(perthread_key) == NULL;
182 }
183
184 void
185 ovsrcu_synchronize(void)
186 {
187     unsigned int warning_threshold = 1000;
188     uint64_t target_seqno;
189     long long int start;
190
191     if (single_threaded()) {
192         return;
193     }
194
195     target_seqno = seq_read(global_seqno);
196     ovsrcu_quiesce_start();
197     start = time_msec();
198
199     for (;;) {
200         uint64_t cur_seqno = seq_read(global_seqno);
201         struct ovsrcu_perthread *perthread;
202         char stalled_thread[16];
203         unsigned int elapsed;
204         bool done = true;
205
206         ovs_mutex_lock(&ovsrcu_threads_mutex);
207         LIST_FOR_EACH (perthread, list_node, &ovsrcu_threads) {
208             if (perthread->seqno <= target_seqno) {
209                 ovs_strlcpy(stalled_thread, perthread->name,
210                             sizeof stalled_thread);
211                 done = false;
212                 break;
213             }
214         }
215         ovs_mutex_unlock(&ovsrcu_threads_mutex);
216
217         if (done) {
218             break;
219         }
220
221         elapsed = time_msec() - start;
222         if (elapsed >= warning_threshold) {
223             VLOG_WARN("blocked %u ms waiting for %s to quiesce",
224                       elapsed, stalled_thread);
225             warning_threshold *= 2;
226         }
227         poll_timer_wait_until(start + warning_threshold);
228
229         seq_wait(global_seqno, cur_seqno);
230         poll_block();
231     }
232     ovsrcu_quiesce_end();
233 }
234
235 /* Registers 'function' to be called, passing 'aux' as argument, after the
236  * next grace period.
237  *
238  * The call is guaranteed to happen after the next time all participating
239  * threads have quiesced at least once, but there is no quarantee that all
240  * registered functions are called as early as possible, or that the functions
241  * registered by different threads would be called in the order the
242  * registrations took place.  In particular, even if two threads provably
243  * register a function each in a specific order, the functions may still be
244  * called in the opposite order, depending on the timing of when the threads
245  * call ovsrcu_quiesce(), how many functions they postpone, and when the
246  * ovs-rcu thread happens to grab the functions to be called.
247  *
248  * All functions registered by a single thread are guaranteed to execute in the
249  * registering order, however.
250  *
251  * This function is more conveniently called through the ovsrcu_postpone()
252  * macro, which provides a type-safe way to allow 'function''s parameter to be
253  * any pointer type. */
254 void
255 ovsrcu_postpone__(void (*function)(void *aux), void *aux)
256 {
257     struct ovsrcu_perthread *perthread = ovsrcu_perthread_get();
258     struct ovsrcu_cbset *cbset;
259     struct ovsrcu_cb *cb;
260
261     cbset = perthread->cbset;
262     if (!cbset) {
263         cbset = perthread->cbset = xmalloc(sizeof *perthread->cbset);
264         cbset->n_cbs = 0;
265     }
266
267     cb = &cbset->cbs[cbset->n_cbs++];
268     cb->function = function;
269     cb->aux = aux;
270
271     if (cbset->n_cbs >= ARRAY_SIZE(cbset->cbs)) {
272         ovsrcu_flush_cbset(perthread);
273     }
274 }
275
276 static bool
277 ovsrcu_call_postponed(void)
278 {
279     struct ovsrcu_cbset *cbset;
280     struct ovs_list cbsets;
281
282     guarded_list_pop_all(&flushed_cbsets, &cbsets);
283     if (ovs_list_is_empty(&cbsets)) {
284         return false;
285     }
286
287     ovsrcu_synchronize();
288
289     LIST_FOR_EACH_POP (cbset, list_node, &cbsets) {
290         struct ovsrcu_cb *cb;
291
292         for (cb = cbset->cbs; cb < &cbset->cbs[cbset->n_cbs]; cb++) {
293             cb->function(cb->aux);
294         }
295         free(cbset);
296     }
297
298     return true;
299 }
300
301 static void *
302 ovsrcu_postpone_thread(void *arg OVS_UNUSED)
303 {
304     pthread_detach(pthread_self());
305
306     for (;;) {
307         uint64_t seqno = seq_read(flushed_cbsets_seq);
308         if (!ovsrcu_call_postponed()) {
309             seq_wait(flushed_cbsets_seq, seqno);
310             poll_block();
311         }
312     }
313
314     OVS_NOT_REACHED();
315 }
316
317 static void
318 ovsrcu_flush_cbset__(struct ovsrcu_perthread *perthread, bool protected)
319 {
320     struct ovsrcu_cbset *cbset = perthread->cbset;
321
322     if (cbset) {
323         guarded_list_push_back(&flushed_cbsets, &cbset->list_node, SIZE_MAX);
324         perthread->cbset = NULL;
325
326         if (protected) {
327             seq_change_protected(flushed_cbsets_seq);
328         } else {
329             seq_change(flushed_cbsets_seq);
330         }
331     }
332 }
333
334 static void
335 ovsrcu_flush_cbset(struct ovsrcu_perthread *perthread)
336 {
337     ovsrcu_flush_cbset__(perthread, false);
338 }
339
340 static void
341 ovsrcu_unregister__(struct ovsrcu_perthread *perthread)
342 {
343     if (perthread->cbset) {
344         ovsrcu_flush_cbset(perthread);
345     }
346
347     ovs_mutex_lock(&ovsrcu_threads_mutex);
348     ovs_list_remove(&perthread->list_node);
349     ovs_mutex_unlock(&ovsrcu_threads_mutex);
350
351     ovs_mutex_destroy(&perthread->mutex);
352     free(perthread);
353
354     seq_change(global_seqno);
355 }
356
357 static void
358 ovsrcu_thread_exit_cb(void *perthread)
359 {
360     ovsrcu_unregister__(perthread);
361 }
362
363 /* Cancels the callback to ovsrcu_thread_exit_cb().
364  *
365  * Cancelling the call to the destructor during the main thread exit
366  * is needed while using pthreads-win32 library in Windows. It has been
367  * observed that in pthreads-win32, a call to the destructor during
368  * main thread exit causes undefined behavior. */
369 static void
370 ovsrcu_cancel_thread_exit_cb(void *aux OVS_UNUSED)
371 {
372     pthread_setspecific(perthread_key, NULL);
373 }
374
375 static void
376 ovsrcu_init_module(void)
377 {
378     static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;
379     if (ovsthread_once_start(&once)) {
380         global_seqno = seq_create();
381         xpthread_key_create(&perthread_key, ovsrcu_thread_exit_cb);
382         fatal_signal_add_hook(ovsrcu_cancel_thread_exit_cb, NULL, NULL, true);
383         ovs_list_init(&ovsrcu_threads);
384         ovs_mutex_init(&ovsrcu_threads_mutex);
385
386         guarded_list_init(&flushed_cbsets);
387         flushed_cbsets_seq = seq_create();
388
389         ovsthread_once_done(&once);
390     }
391 }