cf8b5cec46d2c10afaff8c53a56e9adfa4e27eb1
[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 #include <stdio.h>
27 #include <pthread.h>
28 #include <signal.h>
29 #include <assert.h>
30 #include <stdlib.h>
31 #include <string.h>
32 #include <errno.h>
33 #include <poll.h>
34
35 #include "urcu-qsbr-map.h"
36
37 #define BUILD_QSBR_LIB
38 #include "urcu-qsbr-static.h"
39 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
40 #include "urcu-qsbr.h"
41
42 void __attribute__((destructor)) rcu_exit(void);
43
44 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
45
46 int gp_futex;
47
48 /*
49 * Global grace period counter.
50 */
51 unsigned long rcu_gp_ctr = RCU_GP_ONLINE;
52
53 /*
54 * Written to only by each individual reader. Read by both the reader and the
55 * writers.
56 */
57 struct rcu_reader __thread rcu_reader;
58
59 #ifdef DEBUG_YIELD
60 unsigned int yield_active;
61 unsigned int __thread rand_yield;
62 #endif
63
64 static CDS_LIST_HEAD(registry);
65
66 static void mutex_lock(pthread_mutex_t *mutex)
67 {
68 int ret;
69
70 #ifndef DISTRUST_SIGNALS_EXTREME
71 ret = pthread_mutex_lock(mutex);
72 if (ret) {
73 perror("Error in pthread mutex lock");
74 exit(-1);
75 }
76 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
77 while ((ret = pthread_mutex_trylock(mutex)) != 0) {
78 if (ret != EBUSY && ret != EINTR) {
79 printf("ret = %d, errno = %d\n", ret, errno);
80 perror("Error in pthread mutex lock");
81 exit(-1);
82 }
83 poll(NULL,0,10);
84 }
85 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
86 }
87
88 static void mutex_unlock(pthread_mutex_t *mutex)
89 {
90 int ret;
91
92 ret = pthread_mutex_unlock(mutex);
93 if (ret) {
94 perror("Error in pthread mutex unlock");
95 exit(-1);
96 }
97 }
98
99 /*
100 * synchronize_rcu() waiting. Single thread.
101 */
102 static void wait_gp(void)
103 {
104 /* Read reader_gp before read futex */
105 cmm_smp_rmb();
106 if (uatomic_read(&gp_futex) == -1)
107 futex_noasync(&gp_futex, FUTEX_WAIT, -1,
108 NULL, NULL, 0);
109 }
110
111 static void update_counter_and_wait(void)
112 {
113 CDS_LIST_HEAD(qsreaders);
114 int wait_loops = 0;
115 struct rcu_reader *index, *tmp;
116
117 #if (CAA_BITS_PER_LONG < 64)
118 /* Switch parity: 0 -> 1, 1 -> 0 */
119 CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
120 #else /* !(CAA_BITS_PER_LONG < 64) */
121 /* Increment current G.P. */
122 CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
123 #endif /* !(CAA_BITS_PER_LONG < 64) */
124
125 /*
126 * Must commit rcu_gp_ctr update to memory before waiting for
127 * quiescent state. Failure to do so could result in the writer
128 * waiting forever while new readers are always accessing data
129 * (no progress). Enforce compiler-order of store to rcu_gp_ctr
130 * before load rcu_reader ctr.
131 */
132 cmm_barrier();
133
134 /*
135 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
136 * model easier to understand. It does not have a big performance impact
137 * anyway, given this is the write-side.
138 */
139 cmm_smp_mb();
140
141 /*
142 * Wait for each thread rcu_reader_qs_gp count to become 0.
143 */
144 for (;;) {
145 wait_loops++;
146 if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
147 uatomic_dec(&gp_futex);
148 /* Write futex before read reader_gp */
149 cmm_smp_mb();
150 }
151
152 cds_list_for_each_entry_safe(index, tmp, &registry, node) {
153 if (!rcu_gp_ongoing(&index->ctr))
154 cds_list_move(&index->node, &qsreaders);
155 }
156
157 if (cds_list_empty(&registry)) {
158 if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
159 /* Read reader_gp before write futex */
160 cmm_smp_mb();
161 uatomic_set(&gp_futex, 0);
162 }
163 break;
164 } else {
165 if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
166 wait_gp();
167 } else {
168 #ifndef HAS_INCOHERENT_CACHES
169 caa_cpu_relax();
170 #else /* #ifndef HAS_INCOHERENT_CACHES */
171 cmm_smp_mb();
172 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
173 }
174 }
175 }
176 /* put back the reader list in the registry */
177 cds_list_splice(&qsreaders, &registry);
178 }
179
180 /*
181 * Using a two-subphases algorithm for architectures with smaller than 64-bit
182 * long-size to ensure we do not encounter an overflow bug.
183 */
184
185 #if (CAA_BITS_PER_LONG < 64)
186 void synchronize_rcu(void)
187 {
188 unsigned long was_online;
189
190 was_online = rcu_reader.ctr;
191
192 /* All threads should read qparity before accessing data structure
193 * where new ptr points to.
194 */
195 /* Write new ptr before changing the qparity */
196 cmm_smp_mb();
197
198 /*
199 * Mark the writer thread offline to make sure we don't wait for
200 * our own quiescent state. This allows using synchronize_rcu()
201 * in threads registered as readers.
202 */
203 if (was_online)
204 CMM_STORE_SHARED(rcu_reader.ctr, 0);
205
206 mutex_lock(&rcu_gp_lock);
207
208 if (cds_list_empty(&registry))
209 goto out;
210
211 /*
212 * Wait for previous parity to be empty of readers.
213 */
214 update_counter_and_wait(); /* 0 -> 1, wait readers in parity 0 */
215
216 /*
217 * Must finish waiting for quiescent state for parity 0 before
218 * committing next rcu_gp_ctr update to memory. Failure to
219 * do so could result in the writer waiting forever while new
220 * readers are always accessing data (no progress). Enforce
221 * compiler-order of load rcu_reader ctr before store to
222 * rcu_gp_ctr.
223 */
224 cmm_barrier();
225
226 /*
227 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
228 * model easier to understand. It does not have a big performance impact
229 * anyway, given this is the write-side.
230 */
231 cmm_smp_mb();
232
233 /*
234 * Wait for previous parity to be empty of readers.
235 */
236 update_counter_and_wait(); /* 1 -> 0, wait readers in parity 1 */
237 out:
238 mutex_unlock(&rcu_gp_lock);
239
240 /*
241 * Finish waiting for reader threads before letting the old ptr being
242 * freed.
243 */
244 if (was_online)
245 _CMM_STORE_SHARED(rcu_reader.ctr,
246 CMM_LOAD_SHARED(rcu_gp_ctr));
247 cmm_smp_mb();
248 }
249 #else /* !(CAA_BITS_PER_LONG < 64) */
250 void synchronize_rcu(void)
251 {
252 unsigned long was_online;
253
254 was_online = rcu_reader.ctr;
255
256 /*
257 * Mark the writer thread offline to make sure we don't wait for
258 * our own quiescent state. This allows using synchronize_rcu()
259 * in threads registered as readers.
260 */
261 cmm_smp_mb();
262 if (was_online)
263 CMM_STORE_SHARED(rcu_reader.ctr, 0);
264
265 mutex_lock(&rcu_gp_lock);
266 if (cds_list_empty(&registry))
267 goto out;
268 update_counter_and_wait();
269 out:
270 mutex_unlock(&rcu_gp_lock);
271
272 if (was_online)
273 _CMM_STORE_SHARED(rcu_reader.ctr,
274 CMM_LOAD_SHARED(rcu_gp_ctr));
275 cmm_smp_mb();
276 }
277 #endif /* !(CAA_BITS_PER_LONG < 64) */
278
279 /*
280 * library wrappers to be used by non-LGPL compatible source code.
281 */
282
283 void rcu_read_lock(void)
284 {
285 _rcu_read_lock();
286 }
287
288 void rcu_read_unlock(void)
289 {
290 _rcu_read_unlock();
291 }
292
293 void rcu_quiescent_state(void)
294 {
295 _rcu_quiescent_state();
296 }
297
298 void rcu_thread_offline(void)
299 {
300 _rcu_thread_offline();
301 }
302
303 void rcu_thread_online(void)
304 {
305 _rcu_thread_online();
306 }
307
308 void rcu_register_thread(void)
309 {
310 rcu_reader.tid = pthread_self();
311 assert(rcu_reader.ctr == 0);
312
313 mutex_lock(&rcu_gp_lock);
314 cds_list_add(&rcu_reader.node, &registry);
315 mutex_unlock(&rcu_gp_lock);
316 _rcu_thread_online();
317 }
318
319 void rcu_unregister_thread(void)
320 {
321 /*
322 * We have to make the thread offline otherwise we end up dealocking
323 * with a waiting writer.
324 */
325 _rcu_thread_offline();
326 mutex_lock(&rcu_gp_lock);
327 cds_list_del(&rcu_reader.node);
328 mutex_unlock(&rcu_gp_lock);
329 }
330
331 void rcu_exit(void)
332 {
333 assert(cds_list_empty(&registry));
334 }
335
336 #include "urcu-call-rcu-impl.h"
337 #include "urcu-defer-impl.h"
This page took 0.035396 seconds and 3 git commands to generate.