Merge branch 'master' into urcu/ht-shrink
[urcu.git] / urcu-qsbr.c
1 /*
2 * urcu-qsbr.c
3 *
4 * Userspace RCU QSBR library
5 *
6 * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
7 * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
8 *
9 * This library is free software; you can redistribute it and/or
10 * modify it under the terms of the GNU Lesser General Public
11 * License as published by the Free Software Foundation; either
12 * version 2.1 of the License, or (at your option) any later version.
13 *
14 * This library is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * Lesser General Public License for more details.
18 *
19 * You should have received a copy of the GNU Lesser General Public
20 * License along with this library; if not, write to the Free Software
21 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
22 *
23 * IBM's contributions to this file may be relicensed under LGPLv2 or later.
24 */
25
26 #define _GNU_SOURCE
27 #define _LGPL_SOURCE
28 #include <stdio.h>
29 #include <pthread.h>
30 #include <signal.h>
31 #include <assert.h>
32 #include <stdlib.h>
33 #include <stdint.h>
34 #include <string.h>
35 #include <errno.h>
36 #include <poll.h>
37
38 #include "urcu/wfqueue.h"
39 #include "urcu/map/urcu-qsbr.h"
40 #define BUILD_QSBR_LIB
41 #include "urcu/static/urcu-qsbr.h"
42
43 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
44 #undef _LGPL_SOURCE
45 #include "urcu-qsbr.h"
46 #define _LGPL_SOURCE
47
48 void __attribute__((destructor)) rcu_exit(void);
49
50 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
51
52 int32_t gp_futex;
53
54 /*
55 * Global grace period counter.
56 */
57 unsigned long rcu_gp_ctr = RCU_GP_ONLINE;
58
59 /*
60 * Active attempts to check for reader Q.S. before calling futex().
61 */
62 #define RCU_QS_ACTIVE_ATTEMPTS 100
63
64 /*
65 * Written to only by each individual reader. Read by both the reader and the
66 * writers.
67 */
68 struct rcu_reader __thread rcu_reader;
69
70 #ifdef DEBUG_YIELD
71 unsigned int yield_active;
72 unsigned int __thread rand_yield;
73 #endif
74
75 static CDS_LIST_HEAD(registry);
76
77 static void mutex_lock(pthread_mutex_t *mutex)
78 {
79 int ret;
80
81 #ifndef DISTRUST_SIGNALS_EXTREME
82 ret = pthread_mutex_lock(mutex);
83 if (ret) {
84 perror("Error in pthread mutex lock");
85 exit(-1);
86 }
87 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
88 while ((ret = pthread_mutex_trylock(mutex)) != 0) {
89 if (ret != EBUSY && ret != EINTR) {
90 printf("ret = %d, errno = %d\n", ret, errno);
91 perror("Error in pthread mutex lock");
92 exit(-1);
93 }
94 poll(NULL,0,10);
95 }
96 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
97 }
98
99 static void mutex_unlock(pthread_mutex_t *mutex)
100 {
101 int ret;
102
103 ret = pthread_mutex_unlock(mutex);
104 if (ret) {
105 perror("Error in pthread mutex unlock");
106 exit(-1);
107 }
108 }
109
110 /*
111 * synchronize_rcu() waiting. Single thread.
112 */
113 static void wait_gp(void)
114 {
115 /* Read reader_gp before read futex */
116 cmm_smp_rmb();
117 if (uatomic_read(&gp_futex) == -1)
118 futex_noasync(&gp_futex, FUTEX_WAIT, -1,
119 NULL, NULL, 0);
120 }
121
122 static void update_counter_and_wait(void)
123 {
124 CDS_LIST_HEAD(qsreaders);
125 int wait_loops = 0;
126 struct rcu_reader *index, *tmp;
127
128 #if (CAA_BITS_PER_LONG < 64)
129 /* Switch parity: 0 -> 1, 1 -> 0 */
130 CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
131 #else /* !(CAA_BITS_PER_LONG < 64) */
132 /* Increment current G.P. */
133 CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
134 #endif /* !(CAA_BITS_PER_LONG < 64) */
135
136 /*
137 * Must commit rcu_gp_ctr update to memory before waiting for
138 * quiescent state. Failure to do so could result in the writer
139 * waiting forever while new readers are always accessing data
140 * (no progress). Enforce compiler-order of store to rcu_gp_ctr
141 * before load rcu_reader ctr.
142 */
143 cmm_barrier();
144
145 /*
146 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
147 * model easier to understand. It does not have a big performance impact
148 * anyway, given this is the write-side.
149 */
150 cmm_smp_mb();
151
152 /*
153 * Wait for each thread rcu_reader_qs_gp count to become 0.
154 */
155 for (;;) {
156 wait_loops++;
157 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
158 uatomic_set(&gp_futex, -1);
159 /*
160 * Write futex before write waiting (the other side
161 * reads them in the opposite order).
162 */
163 cmm_smp_wmb();
164 cds_list_for_each_entry(index, &registry, node) {
165 _CMM_STORE_SHARED(index->waiting, 1);
166 }
167 /* Write futex before read reader_gp */
168 cmm_smp_mb();
169 }
170 cds_list_for_each_entry_safe(index, tmp, &registry, node) {
171 if (!rcu_gp_ongoing(&index->ctr))
172 cds_list_move(&index->node, &qsreaders);
173 }
174
175 if (cds_list_empty(&registry)) {
176 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
177 /* Read reader_gp before write futex */
178 cmm_smp_mb();
179 uatomic_set(&gp_futex, 0);
180 }
181 break;
182 } else {
183 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
184 wait_gp();
185 } else {
186 #ifndef HAS_INCOHERENT_CACHES
187 caa_cpu_relax();
188 #else /* #ifndef HAS_INCOHERENT_CACHES */
189 cmm_smp_mb();
190 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
191 }
192 }
193 }
194 /* put back the reader list in the registry */
195 cds_list_splice(&qsreaders, &registry);
196 }
197
198 /*
199 * Using a two-subphases algorithm for architectures with smaller than 64-bit
200 * long-size to ensure we do not encounter an overflow bug.
201 */
202
203 #if (CAA_BITS_PER_LONG < 64)
204 void synchronize_rcu(void)
205 {
206 unsigned long was_online;
207
208 was_online = rcu_reader.ctr;
209
210 /* All threads should read qparity before accessing data structure
211 * where new ptr points to. In the "then" case, rcu_thread_offline
212 * includes a memory barrier.
213 *
214 * Mark the writer thread offline to make sure we don't wait for
215 * our own quiescent state. This allows using synchronize_rcu()
216 * in threads registered as readers.
217 */
218 if (was_online)
219 rcu_thread_offline();
220 else
221 cmm_smp_mb();
222
223 mutex_lock(&rcu_gp_lock);
224
225 if (cds_list_empty(&registry))
226 goto out;
227
228 /*
229 * Wait for previous parity to be empty of readers.
230 */
231 update_counter_and_wait(); /* 0 -> 1, wait readers in parity 0 */
232
233 /*
234 * Must finish waiting for quiescent state for parity 0 before
235 * committing next rcu_gp_ctr update to memory. Failure to
236 * do so could result in the writer waiting forever while new
237 * readers are always accessing data (no progress). Enforce
238 * compiler-order of load rcu_reader ctr before store to
239 * rcu_gp_ctr.
240 */
241 cmm_barrier();
242
243 /*
244 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
245 * model easier to understand. It does not have a big performance impact
246 * anyway, given this is the write-side.
247 */
248 cmm_smp_mb();
249
250 /*
251 * Wait for previous parity to be empty of readers.
252 */
253 update_counter_and_wait(); /* 1 -> 0, wait readers in parity 1 */
254 out:
255 mutex_unlock(&rcu_gp_lock);
256
257 /*
258 * Finish waiting for reader threads before letting the old ptr being
259 * freed.
260 */
261 if (was_online)
262 rcu_thread_online();
263 else
264 cmm_smp_mb();
265 }
266 #else /* !(CAA_BITS_PER_LONG < 64) */
267 void synchronize_rcu(void)
268 {
269 unsigned long was_online;
270
271 was_online = rcu_reader.ctr;
272
273 /*
274 * Mark the writer thread offline to make sure we don't wait for
275 * our own quiescent state. This allows using synchronize_rcu()
276 * in threads registered as readers.
277 */
278 if (was_online)
279 rcu_thread_offline();
280 else
281 cmm_smp_mb();
282
283 mutex_lock(&rcu_gp_lock);
284 if (cds_list_empty(&registry))
285 goto out;
286 update_counter_and_wait();
287 out:
288 mutex_unlock(&rcu_gp_lock);
289
290 if (was_online)
291 rcu_thread_online();
292 else
293 cmm_smp_mb();
294 }
295 #endif /* !(CAA_BITS_PER_LONG < 64) */
296
297 /*
298 * library wrappers to be used by non-LGPL compatible source code.
299 */
300
301 void rcu_read_lock(void)
302 {
303 _rcu_read_lock();
304 }
305
306 void rcu_read_unlock(void)
307 {
308 _rcu_read_unlock();
309 }
310
311 void rcu_quiescent_state(void)
312 {
313 _rcu_quiescent_state();
314 }
315
316 void rcu_thread_offline(void)
317 {
318 _rcu_thread_offline();
319 }
320
321 void rcu_thread_online(void)
322 {
323 _rcu_thread_online();
324 }
325
326 void rcu_register_thread(void)
327 {
328 rcu_reader.tid = pthread_self();
329 assert(rcu_reader.ctr == 0);
330
331 mutex_lock(&rcu_gp_lock);
332 cds_list_add(&rcu_reader.node, &registry);
333 mutex_unlock(&rcu_gp_lock);
334 _rcu_thread_online();
335 }
336
337 void rcu_unregister_thread(void)
338 {
339 /*
340 * We have to make the thread offline otherwise we end up dealocking
341 * with a waiting writer.
342 */
343 _rcu_thread_offline();
344 mutex_lock(&rcu_gp_lock);
345 cds_list_del(&rcu_reader.node);
346 mutex_unlock(&rcu_gp_lock);
347 }
348
349 void rcu_exit(void)
350 {
351 /*
352 * Assertion disabled because call_rcu threads are now rcu
353 * readers, and left running at exit.
354 * assert(cds_list_empty(&registry));
355 */
356 }
357
358 #include "urcu-call-rcu-impl.h"
359 #include "urcu-defer-impl.h"
This page took 0.052225 seconds and 4 git commands to generate.