1 .. _list_rcu_doc: 1 .. _list_rcu_doc:
2 2
3 Using RCU to Protect Read-Mostly Linked Lists 3 Using RCU to Protect Read-Mostly Linked Lists
4 ============================================= 4 =============================================
5 5
6 One of the most common uses of RCU is protecti 6 One of the most common uses of RCU is protecting read-mostly linked lists
7 (``struct list_head`` in list.h). One big adv 7 (``struct list_head`` in list.h). One big advantage of this approach is
8 that all of the required memory ordering is pr 8 that all of the required memory ordering is provided by the list macros.
9 This document describes several list-based RCU 9 This document describes several list-based RCU use cases.
10 10
11 When iterating a list while holding the rcu_re 11 When iterating a list while holding the rcu_read_lock(), writers may
12 modify the list. The reader is guaranteed to 12 modify the list. The reader is guaranteed to see all of the elements
13 which were added to the list before they acqui 13 which were added to the list before they acquired the rcu_read_lock()
14 and are still on the list when they drop the r 14 and are still on the list when they drop the rcu_read_unlock().
15 Elements which are added to, or removed from t 15 Elements which are added to, or removed from the list may or may not
16 be seen. If the writer calls list_replace_rcu 16 be seen. If the writer calls list_replace_rcu(), the reader may see
17 either the old element or the new element; the 17 either the old element or the new element; they will not see both,
18 nor will they see neither. 18 nor will they see neither.
19 19
20 20
21 Example 1: Read-mostly list: Deferred Destruct 21 Example 1: Read-mostly list: Deferred Destruction
22 ---------------------------------------------- 22 -------------------------------------------------
23 23
24 A widely used usecase for RCU lists in the ker 24 A widely used usecase for RCU lists in the kernel is lockless iteration over
25 all processes in the system. ``task_struct::ta 25 all processes in the system. ``task_struct::tasks`` represents the list node that
26 links all the processes. The list can be trave 26 links all the processes. The list can be traversed in parallel to any list
27 additions or removals. 27 additions or removals.
28 28
29 The traversal of the list is done using ``for_ 29 The traversal of the list is done using ``for_each_process()`` which is defined
30 by the 2 macros:: 30 by the 2 macros::
31 31
32 #define next_task(p) \ 32 #define next_task(p) \
33 list_entry_rcu((p)->tasks.next 33 list_entry_rcu((p)->tasks.next, struct task_struct, tasks)
34 34
35 #define for_each_process(p) \ 35 #define for_each_process(p) \
36 for (p = &init_task ; (p = nex 36 for (p = &init_task ; (p = next_task(p)) != &init_task ; )
37 37
38 The code traversing the list of all processes 38 The code traversing the list of all processes typically looks like::
39 39
40 rcu_read_lock(); 40 rcu_read_lock();
41 for_each_process(p) { 41 for_each_process(p) {
42 /* Do something with p */ 42 /* Do something with p */
43 } 43 }
44 rcu_read_unlock(); 44 rcu_read_unlock();
45 45
46 The simplified and heavily inlined code for re 46 The simplified and heavily inlined code for removing a process from a
47 task list is:: 47 task list is::
48 48
49 void release_task(struct task_struct * 49 void release_task(struct task_struct *p)
50 { 50 {
51 write_lock(&tasklist_lock); 51 write_lock(&tasklist_lock);
52 list_del_rcu(&p->tasks); 52 list_del_rcu(&p->tasks);
53 write_unlock(&tasklist_lock); 53 write_unlock(&tasklist_lock);
54 call_rcu(&p->rcu, delayed_put_ 54 call_rcu(&p->rcu, delayed_put_task_struct);
55 } 55 }
56 56
57 When a process exits, ``release_task()`` calls 57 When a process exits, ``release_task()`` calls ``list_del_rcu(&p->tasks)``
58 via __exit_signal() and __unhash_process() und 58 via __exit_signal() and __unhash_process() under ``tasklist_lock``
59 writer lock protection. The list_del_rcu() in 59 writer lock protection. The list_del_rcu() invocation removes
60 the task from the list of all tasks. The ``tas 60 the task from the list of all tasks. The ``tasklist_lock``
61 prevents concurrent list additions/removals fr 61 prevents concurrent list additions/removals from corrupting the
62 list. Readers using ``for_each_process()`` are 62 list. Readers using ``for_each_process()`` are not protected with the
63 ``tasklist_lock``. To prevent readers from not 63 ``tasklist_lock``. To prevent readers from noticing changes in the list
64 pointers, the ``task_struct`` object is freed 64 pointers, the ``task_struct`` object is freed only after one or more
65 grace periods elapse, with the help of call_rc 65 grace periods elapse, with the help of call_rcu(), which is invoked via
66 put_task_struct_rcu_user(). This deferring of 66 put_task_struct_rcu_user(). This deferring of destruction ensures that
67 any readers traversing the list will see valid 67 any readers traversing the list will see valid ``p->tasks.next`` pointers
68 and deletion/freeing can happen in parallel wi 68 and deletion/freeing can happen in parallel with traversal of the list.
69 This pattern is also called an **existence loc 69 This pattern is also called an **existence lock**, since RCU refrains
70 from invoking the delayed_put_task_struct() ca 70 from invoking the delayed_put_task_struct() callback function until
71 all existing readers finish, which guarantees 71 all existing readers finish, which guarantees that the ``task_struct``
72 object in question will remain in existence un 72 object in question will remain in existence until after the completion
73 of all RCU readers that might possibly have a 73 of all RCU readers that might possibly have a reference to that object.
74 74
75 75
76 Example 2: Read-Side Action Taken Outside of L 76 Example 2: Read-Side Action Taken Outside of Lock: No In-Place Updates
77 ---------------------------------------------- 77 ----------------------------------------------------------------------
78 78
79 Some reader-writer locking use cases compute a 79 Some reader-writer locking use cases compute a value while holding
80 the read-side lock, but continue to use that v 80 the read-side lock, but continue to use that value after that lock is
81 released. These use cases are often good cand 81 released. These use cases are often good candidates for conversion
82 to RCU. One prominent example involves networ 82 to RCU. One prominent example involves network packet routing.
83 Because the packet-routing data tracks the sta 83 Because the packet-routing data tracks the state of equipment outside
84 of the computer, it will at times contain stal 84 of the computer, it will at times contain stale data. Therefore, once
85 the route has been computed, there is no need 85 the route has been computed, there is no need to hold the routing table
86 static during transmission of the packet. Aft 86 static during transmission of the packet. After all, you can hold the
87 routing table static all you want, but that wo 87 routing table static all you want, but that won't keep the external
88 Internet from changing, and it is the state of 88 Internet from changing, and it is the state of the external Internet
89 that really matters. In addition, routing ent 89 that really matters. In addition, routing entries are typically added
90 or deleted, rather than being modified in plac 90 or deleted, rather than being modified in place. This is a rare example
91 of the finite speed of light and the non-zero 91 of the finite speed of light and the non-zero size of atoms actually
92 helping make synchronization be lighter weight 92 helping make synchronization be lighter weight.
93 93
94 A straightforward example of this type of RCU 94 A straightforward example of this type of RCU use case may be found in
95 the system-call auditing support. For example 95 the system-call auditing support. For example, a reader-writer locked
96 implementation of ``audit_filter_task()`` migh 96 implementation of ``audit_filter_task()`` might be as follows::
97 97
98 static enum audit_state audit_filter_t 98 static enum audit_state audit_filter_task(struct task_struct *tsk, char **key)
99 { 99 {
100 struct audit_entry *e; 100 struct audit_entry *e;
101 enum audit_state state; 101 enum audit_state state;
102 102
103 read_lock(&auditsc_lock); 103 read_lock(&auditsc_lock);
104 /* Note: audit_filter_mutex he 104 /* Note: audit_filter_mutex held by caller. */
105 list_for_each_entry(e, &audit_ 105 list_for_each_entry(e, &audit_tsklist, list) {
106 if (audit_filter_rules 106 if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {
107 if (state == A 107 if (state == AUDIT_STATE_RECORD)
108 *key = 108 *key = kstrdup(e->rule.filterkey, GFP_ATOMIC);
109 read_unlock(&a 109 read_unlock(&auditsc_lock);
110 return state; 110 return state;
111 } 111 }
112 } 112 }
113 read_unlock(&auditsc_lock); 113 read_unlock(&auditsc_lock);
114 return AUDIT_BUILD_CONTEXT; 114 return AUDIT_BUILD_CONTEXT;
115 } 115 }
116 116
117 Here the list is searched under the lock, but 117 Here the list is searched under the lock, but the lock is dropped before
118 the corresponding value is returned. By the t 118 the corresponding value is returned. By the time that this value is acted
119 on, the list may well have been modified. Thi 119 on, the list may well have been modified. This makes sense, since if
120 you are turning auditing off, it is OK to audi 120 you are turning auditing off, it is OK to audit a few extra system calls.
121 121
122 This means that RCU can be easily applied to t 122 This means that RCU can be easily applied to the read side, as follows::
123 123
124 static enum audit_state audit_filter_t 124 static enum audit_state audit_filter_task(struct task_struct *tsk, char **key)
125 { 125 {
126 struct audit_entry *e; 126 struct audit_entry *e;
127 enum audit_state state; 127 enum audit_state state;
128 128
129 rcu_read_lock(); 129 rcu_read_lock();
130 /* Note: audit_filter_mutex he 130 /* Note: audit_filter_mutex held by caller. */
131 list_for_each_entry_rcu(e, &au 131 list_for_each_entry_rcu(e, &audit_tsklist, list) {
132 if (audit_filter_rules 132 if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {
133 if (state == A 133 if (state == AUDIT_STATE_RECORD)
134 *key = 134 *key = kstrdup(e->rule.filterkey, GFP_ATOMIC);
135 rcu_read_unloc 135 rcu_read_unlock();
136 return state; 136 return state;
137 } 137 }
138 } 138 }
139 rcu_read_unlock(); 139 rcu_read_unlock();
140 return AUDIT_BUILD_CONTEXT; 140 return AUDIT_BUILD_CONTEXT;
141 } 141 }
142 142
143 The read_lock() and read_unlock() calls have b 143 The read_lock() and read_unlock() calls have become rcu_read_lock()
144 and rcu_read_unlock(), respectively, and the l 144 and rcu_read_unlock(), respectively, and the list_for_each_entry()
145 has become list_for_each_entry_rcu(). The **_ 145 has become list_for_each_entry_rcu(). The **_rcu()** list-traversal
146 primitives add READ_ONCE() and diagnostic chec 146 primitives add READ_ONCE() and diagnostic checks for incorrect use
147 outside of an RCU read-side critical section. 147 outside of an RCU read-side critical section.
148 148
149 The changes to the update side are also straig 149 The changes to the update side are also straightforward. A reader-writer lock
150 might be used as follows for deletion and inse 150 might be used as follows for deletion and insertion in these simplified
151 versions of audit_del_rule() and audit_add_rul 151 versions of audit_del_rule() and audit_add_rule()::
152 152
153 static inline int audit_del_rule(struc 153 static inline int audit_del_rule(struct audit_rule *rule,
154 struc 154 struct list_head *list)
155 { 155 {
156 struct audit_entry *e; 156 struct audit_entry *e;
157 157
158 write_lock(&auditsc_lock); 158 write_lock(&auditsc_lock);
159 list_for_each_entry(e, list, l 159 list_for_each_entry(e, list, list) {
160 if (!audit_compare_rul 160 if (!audit_compare_rule(rule, &e->rule)) {
161 list_del(&e->l 161 list_del(&e->list);
162 write_unlock(& 162 write_unlock(&auditsc_lock);
163 return 0; 163 return 0;
164 } 164 }
165 } 165 }
166 write_unlock(&auditsc_lock); 166 write_unlock(&auditsc_lock);
167 return -EFAULT; /* No 167 return -EFAULT; /* No matching rule */
168 } 168 }
169 169
170 static inline int audit_add_rule(struc 170 static inline int audit_add_rule(struct audit_entry *entry,
171 struc 171 struct list_head *list)
172 { 172 {
173 write_lock(&auditsc_lock); 173 write_lock(&auditsc_lock);
174 if (entry->rule.flags & AUDIT_ 174 if (entry->rule.flags & AUDIT_PREPEND) {
175 entry->rule.flags &= ~ 175 entry->rule.flags &= ~AUDIT_PREPEND;
176 list_add(&entry->list, 176 list_add(&entry->list, list);
177 } else { 177 } else {
178 list_add_tail(&entry-> 178 list_add_tail(&entry->list, list);
179 } 179 }
180 write_unlock(&auditsc_lock); 180 write_unlock(&auditsc_lock);
181 return 0; 181 return 0;
182 } 182 }
183 183
184 Following are the RCU equivalents for these tw 184 Following are the RCU equivalents for these two functions::
185 185
186 static inline int audit_del_rule(struc 186 static inline int audit_del_rule(struct audit_rule *rule,
187 struc 187 struct list_head *list)
188 { 188 {
189 struct audit_entry *e; 189 struct audit_entry *e;
190 190
191 /* No need to use the _rcu ite 191 /* No need to use the _rcu iterator here, since this is the only
192 * deletion routine. */ 192 * deletion routine. */
193 list_for_each_entry(e, list, l 193 list_for_each_entry(e, list, list) {
194 if (!audit_compare_rul 194 if (!audit_compare_rule(rule, &e->rule)) {
195 list_del_rcu(& 195 list_del_rcu(&e->list);
196 call_rcu(&e->r 196 call_rcu(&e->rcu, audit_free_rule);
197 return 0; 197 return 0;
198 } 198 }
199 } 199 }
200 return -EFAULT; /* No 200 return -EFAULT; /* No matching rule */
201 } 201 }
202 202
203 static inline int audit_add_rule(struc 203 static inline int audit_add_rule(struct audit_entry *entry,
204 struc 204 struct list_head *list)
205 { 205 {
206 if (entry->rule.flags & AUDIT_ 206 if (entry->rule.flags & AUDIT_PREPEND) {
207 entry->rule.flags &= ~ 207 entry->rule.flags &= ~AUDIT_PREPEND;
208 list_add_rcu(&entry->l 208 list_add_rcu(&entry->list, list);
209 } else { 209 } else {
210 list_add_tail_rcu(&ent 210 list_add_tail_rcu(&entry->list, list);
211 } 211 }
212 return 0; 212 return 0;
213 } 213 }
214 214
215 Normally, the write_lock() and write_unlock() 215 Normally, the write_lock() and write_unlock() would be replaced by a
216 spin_lock() and a spin_unlock(). But in this c 216 spin_lock() and a spin_unlock(). But in this case, all callers hold
217 ``audit_filter_mutex``, so no additional locki 217 ``audit_filter_mutex``, so no additional locking is required. The
218 auditsc_lock can therefore be eliminated, sinc 218 auditsc_lock can therefore be eliminated, since use of RCU eliminates the
219 need for writers to exclude readers. 219 need for writers to exclude readers.
220 220
221 The list_del(), list_add(), and list_add_tail( 221 The list_del(), list_add(), and list_add_tail() primitives have been
222 replaced by list_del_rcu(), list_add_rcu(), an 222 replaced by list_del_rcu(), list_add_rcu(), and list_add_tail_rcu().
223 The **_rcu()** list-manipulation primitives ad 223 The **_rcu()** list-manipulation primitives add memory barriers that are
224 needed on weakly ordered CPUs. The list_del_r 224 needed on weakly ordered CPUs. The list_del_rcu() primitive omits the
225 pointer poisoning debug-assist code that would 225 pointer poisoning debug-assist code that would otherwise cause concurrent
226 readers to fail spectacularly. 226 readers to fail spectacularly.
227 227
228 So, when readers can tolerate stale data and w 228 So, when readers can tolerate stale data and when entries are either added or
229 deleted, without in-place modification, it is 229 deleted, without in-place modification, it is very easy to use RCU!
230 230
231 231
232 Example 3: Handling In-Place Updates 232 Example 3: Handling In-Place Updates
233 ------------------------------------ 233 ------------------------------------
234 234
235 The system-call auditing code does not update 235 The system-call auditing code does not update auditing rules in place. However,
236 if it did, the reader-writer-locked code to do 236 if it did, the reader-writer-locked code to do so might look as follows
237 (assuming only ``field_count`` is updated, oth 237 (assuming only ``field_count`` is updated, otherwise, the added fields would
238 need to be filled in):: 238 need to be filled in)::
239 239
240 static inline int audit_upd_rule(struc 240 static inline int audit_upd_rule(struct audit_rule *rule,
241 struc 241 struct list_head *list,
242 __u32 242 __u32 newaction,
243 __u32 243 __u32 newfield_count)
244 { 244 {
245 struct audit_entry *e; 245 struct audit_entry *e;
246 struct audit_entry *ne; 246 struct audit_entry *ne;
247 247
248 write_lock(&auditsc_lock); 248 write_lock(&auditsc_lock);
249 /* Note: audit_filter_mutex he 249 /* Note: audit_filter_mutex held by caller. */
250 list_for_each_entry(e, list, l 250 list_for_each_entry(e, list, list) {
251 if (!audit_compare_rul 251 if (!audit_compare_rule(rule, &e->rule)) {
252 e->rule.action 252 e->rule.action = newaction;
253 e->rule.field_ 253 e->rule.field_count = newfield_count;
254 write_unlock(& 254 write_unlock(&auditsc_lock);
255 return 0; 255 return 0;
256 } 256 }
257 } 257 }
258 write_unlock(&auditsc_lock); 258 write_unlock(&auditsc_lock);
259 return -EFAULT; /* No 259 return -EFAULT; /* No matching rule */
260 } 260 }
261 261
262 The RCU version creates a copy, updates the co 262 The RCU version creates a copy, updates the copy, then replaces the old
263 entry with the newly updated entry. This sequ 263 entry with the newly updated entry. This sequence of actions, allowing
264 concurrent reads while making a copy to perfor 264 concurrent reads while making a copy to perform an update, is what gives
265 RCU (*read-copy update*) its name. 265 RCU (*read-copy update*) its name.
266 266
267 The RCU version of audit_upd_rule() is as foll 267 The RCU version of audit_upd_rule() is as follows::
268 268
269 static inline int audit_upd_rule(struc 269 static inline int audit_upd_rule(struct audit_rule *rule,
270 struc 270 struct list_head *list,
271 __u32 271 __u32 newaction,
272 __u32 272 __u32 newfield_count)
273 { 273 {
274 struct audit_entry *e; 274 struct audit_entry *e;
275 struct audit_entry *ne; 275 struct audit_entry *ne;
276 276
277 list_for_each_entry(e, list, l 277 list_for_each_entry(e, list, list) {
278 if (!audit_compare_rul 278 if (!audit_compare_rule(rule, &e->rule)) {
279 ne = kmalloc(s 279 ne = kmalloc(sizeof(*entry), GFP_ATOMIC);
280 if (ne == NULL 280 if (ne == NULL)
281 return 281 return -ENOMEM;
282 audit_copy_rul 282 audit_copy_rule(&ne->rule, &e->rule);
283 ne->rule.actio 283 ne->rule.action = newaction;
284 ne->rule.field 284 ne->rule.field_count = newfield_count;
285 list_replace_r 285 list_replace_rcu(&e->list, &ne->list);
286 call_rcu(&e->r 286 call_rcu(&e->rcu, audit_free_rule);
287 return 0; 287 return 0;
288 } 288 }
289 } 289 }
290 return -EFAULT; /* No 290 return -EFAULT; /* No matching rule */
291 } 291 }
292 292
293 Again, this assumes that the caller holds ``au 293 Again, this assumes that the caller holds ``audit_filter_mutex``. Normally, the
294 writer lock would become a spinlock in this so 294 writer lock would become a spinlock in this sort of code.
295 295
296 The update_lsm_rule() does something very simi 296 The update_lsm_rule() does something very similar, for those who would
297 prefer to look at real Linux-kernel code. 297 prefer to look at real Linux-kernel code.
298 298
299 Another use of this pattern can be found in th 299 Another use of this pattern can be found in the openswitch driver's *connection
300 tracking table* code in ``ct_limit_set()``. T 300 tracking table* code in ``ct_limit_set()``. The table holds connection tracking
301 entries and has a limit on the maximum entries 301 entries and has a limit on the maximum entries. There is one such table
302 per-zone and hence one *limit* per zone. The 302 per-zone and hence one *limit* per zone. The zones are mapped to their limits
303 through a hashtable using an RCU-managed hlist 303 through a hashtable using an RCU-managed hlist for the hash chains. When a new
304 limit is set, a new limit object is allocated 304 limit is set, a new limit object is allocated and ``ct_limit_set()`` is called
305 to replace the old limit object with the new o 305 to replace the old limit object with the new one using list_replace_rcu().
306 The old limit object is then freed after a gra 306 The old limit object is then freed after a grace period using kfree_rcu().
307 307
308 308
309 Example 4: Eliminating Stale Data 309 Example 4: Eliminating Stale Data
310 --------------------------------- 310 ---------------------------------
311 311
312 The auditing example above tolerates stale dat 312 The auditing example above tolerates stale data, as do most algorithms
313 that are tracking external state. After all, 313 that are tracking external state. After all, given there is a delay
314 from the time the external state changes befor 314 from the time the external state changes before Linux becomes aware
315 of the change, and so as noted earlier, a smal 315 of the change, and so as noted earlier, a small quantity of additional
316 RCU-induced staleness is generally not a probl 316 RCU-induced staleness is generally not a problem.
317 317
318 However, there are many examples where stale d 318 However, there are many examples where stale data cannot be tolerated.
319 One example in the Linux kernel is the System 319 One example in the Linux kernel is the System V IPC (see the shm_lock()
320 function in ipc/shm.c). This code checks a *d 320 function in ipc/shm.c). This code checks a *deleted* flag under a
321 per-entry spinlock, and, if the *deleted* flag 321 per-entry spinlock, and, if the *deleted* flag is set, pretends that the
322 entry does not exist. For this to be helpful, 322 entry does not exist. For this to be helpful, the search function must
323 return holding the per-entry spinlock, as shm_ 323 return holding the per-entry spinlock, as shm_lock() does in fact do.
324 324
325 .. _quick_quiz: 325 .. _quick_quiz:
326 326
327 Quick Quiz: 327 Quick Quiz:
328 For the deleted-flag technique to be h 328 For the deleted-flag technique to be helpful, why is it necessary
329 to hold the per-entry lock while retur 329 to hold the per-entry lock while returning from the search function?
330 330
331 :ref:`Answer to Quick Quiz <quick_quiz_answer> 331 :ref:`Answer to Quick Quiz <quick_quiz_answer>`
332 332
333 If the system-call audit module were to ever n 333 If the system-call audit module were to ever need to reject stale data, one way
334 to accomplish this would be to add a ``deleted 334 to accomplish this would be to add a ``deleted`` flag and a ``lock`` spinlock to the
335 ``audit_entry`` structure, and modify audit_fi 335 ``audit_entry`` structure, and modify audit_filter_task() as follows::
336 336
337 static enum audit_state audit_filter_t 337 static enum audit_state audit_filter_task(struct task_struct *tsk)
338 { 338 {
339 struct audit_entry *e; 339 struct audit_entry *e;
340 enum audit_state state; 340 enum audit_state state;
341 341
342 rcu_read_lock(); 342 rcu_read_lock();
343 list_for_each_entry_rcu(e, &au 343 list_for_each_entry_rcu(e, &audit_tsklist, list) {
344 if (audit_filter_rules 344 if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {
345 spin_lock(&e-> 345 spin_lock(&e->lock);
346 if (e->deleted 346 if (e->deleted) {
347 spin_u 347 spin_unlock(&e->lock);
348 rcu_re 348 rcu_read_unlock();
349 return 349 return AUDIT_BUILD_CONTEXT;
350 } 350 }
351 rcu_read_unloc 351 rcu_read_unlock();
352 if (state == A 352 if (state == AUDIT_STATE_RECORD)
353 *key = 353 *key = kstrdup(e->rule.filterkey, GFP_ATOMIC);
354 return state; 354 return state;
355 } 355 }
356 } 356 }
357 rcu_read_unlock(); 357 rcu_read_unlock();
358 return AUDIT_BUILD_CONTEXT; 358 return AUDIT_BUILD_CONTEXT;
359 } 359 }
360 360
361 The ``audit_del_rule()`` function would need t 361 The ``audit_del_rule()`` function would need to set the ``deleted`` flag under the
362 spinlock as follows:: 362 spinlock as follows::
363 363
364 static inline int audit_del_rule(struc 364 static inline int audit_del_rule(struct audit_rule *rule,
365 struc 365 struct list_head *list)
366 { 366 {
367 struct audit_entry *e; 367 struct audit_entry *e;
368 368
369 /* No need to use the _rcu ite 369 /* No need to use the _rcu iterator here, since this
370 * is the only deletion routin 370 * is the only deletion routine. */
371 list_for_each_entry(e, list, l 371 list_for_each_entry(e, list, list) {
372 if (!audit_compare_rul 372 if (!audit_compare_rule(rule, &e->rule)) {
373 spin_lock(&e-> 373 spin_lock(&e->lock);
374 list_del_rcu(& 374 list_del_rcu(&e->list);
375 e->deleted = 1 375 e->deleted = 1;
376 spin_unlock(&e 376 spin_unlock(&e->lock);
377 call_rcu(&e->r 377 call_rcu(&e->rcu, audit_free_rule);
378 return 0; 378 return 0;
379 } 379 }
380 } 380 }
381 return -EFAULT; /* No 381 return -EFAULT; /* No matching rule */
382 } 382 }
383 383
384 This too assumes that the caller holds ``audit 384 This too assumes that the caller holds ``audit_filter_mutex``.
385 385
386 Note that this example assumes that entries ar 386 Note that this example assumes that entries are only added and deleted.
387 Additional mechanism is required to deal corre 387 Additional mechanism is required to deal correctly with the update-in-place
388 performed by audit_upd_rule(). For one thing, 388 performed by audit_upd_rule(). For one thing, audit_upd_rule() would
389 need to hold the locks of both the old ``audit 389 need to hold the locks of both the old ``audit_entry`` and its replacement
390 while executing the list_replace_rcu(). 390 while executing the list_replace_rcu().
391 391
392 392
393 Example 5: Skipping Stale Objects 393 Example 5: Skipping Stale Objects
394 --------------------------------- 394 ---------------------------------
395 395
396 For some use cases, reader performance can be 396 For some use cases, reader performance can be improved by skipping
397 stale objects during read-side list traversal, 397 stale objects during read-side list traversal, where stale objects
398 are those that will be removed and destroyed a 398 are those that will be removed and destroyed after one or more grace
399 periods. One such example can be found in the 399 periods. One such example can be found in the timerfd subsystem. When a
400 ``CLOCK_REALTIME`` clock is reprogrammed (for 400 ``CLOCK_REALTIME`` clock is reprogrammed (for example due to setting
401 of the system time) then all programmed ``time 401 of the system time) then all programmed ``timerfds`` that depend on
402 this clock get triggered and processes waiting 402 this clock get triggered and processes waiting on them are awakened in
403 advance of their scheduled expiry. To facilita 403 advance of their scheduled expiry. To facilitate this, all such timers
404 are added to an RCU-managed ``cancel_list`` wh 404 are added to an RCU-managed ``cancel_list`` when they are setup in
405 ``timerfd_setup_cancel()``:: 405 ``timerfd_setup_cancel()``::
406 406
407 static void timerfd_setup_cancel(struc 407 static void timerfd_setup_cancel(struct timerfd_ctx *ctx, int flags)
408 { 408 {
409 spin_lock(&ctx->cancel_lock); 409 spin_lock(&ctx->cancel_lock);
410 if ((ctx->clockid == CLOCK_REA 410 if ((ctx->clockid == CLOCK_REALTIME ||
411 ctx->clockid == CLOCK_REA 411 ctx->clockid == CLOCK_REALTIME_ALARM) &&
412 (flags & TFD_TIMER_ABSTIME 412 (flags & TFD_TIMER_ABSTIME) && (flags & TFD_TIMER_CANCEL_ON_SET)) {
413 if (!ctx->might_cancel 413 if (!ctx->might_cancel) {
414 ctx->might_can 414 ctx->might_cancel = true;
415 spin_lock(&can 415 spin_lock(&cancel_lock);
416 list_add_rcu(& 416 list_add_rcu(&ctx->clist, &cancel_list);
417 spin_unlock(&c 417 spin_unlock(&cancel_lock);
418 } 418 }
419 } else { 419 } else {
420 __timerfd_remove_cance 420 __timerfd_remove_cancel(ctx);
421 } 421 }
422 spin_unlock(&ctx->cancel_lock) 422 spin_unlock(&ctx->cancel_lock);
423 } 423 }
424 424
425 When a timerfd is freed (fd is closed), then t 425 When a timerfd is freed (fd is closed), then the ``might_cancel``
426 flag of the timerfd object is cleared, the obj 426 flag of the timerfd object is cleared, the object removed from the
427 ``cancel_list`` and destroyed, as shown in thi 427 ``cancel_list`` and destroyed, as shown in this simplified and inlined
428 version of timerfd_release():: 428 version of timerfd_release()::
429 429
430 int timerfd_release(struct inode *inod 430 int timerfd_release(struct inode *inode, struct file *file)
431 { 431 {
432 struct timerfd_ctx *ctx = file 432 struct timerfd_ctx *ctx = file->private_data;
433 433
434 spin_lock(&ctx->cancel_lock); 434 spin_lock(&ctx->cancel_lock);
435 if (ctx->might_cancel) { 435 if (ctx->might_cancel) {
436 ctx->might_cancel = fa 436 ctx->might_cancel = false;
437 spin_lock(&cancel_lock 437 spin_lock(&cancel_lock);
438 list_del_rcu(&ctx->cli 438 list_del_rcu(&ctx->clist);
439 spin_unlock(&cancel_lo 439 spin_unlock(&cancel_lock);
440 } 440 }
441 spin_unlock(&ctx->cancel_lock) 441 spin_unlock(&ctx->cancel_lock);
442 442
443 if (isalarm(ctx)) 443 if (isalarm(ctx))
444 alarm_cancel(&ctx->t.a 444 alarm_cancel(&ctx->t.alarm);
445 else 445 else
446 hrtimer_cancel(&ctx->t 446 hrtimer_cancel(&ctx->t.tmr);
447 kfree_rcu(ctx, rcu); 447 kfree_rcu(ctx, rcu);
448 return 0; 448 return 0;
449 } 449 }
450 450
451 If the ``CLOCK_REALTIME`` clock is set, for ex 451 If the ``CLOCK_REALTIME`` clock is set, for example by a time server, the
452 hrtimer framework calls ``timerfd_clock_was_se 452 hrtimer framework calls ``timerfd_clock_was_set()`` which walks the
453 ``cancel_list`` and wakes up processes waiting 453 ``cancel_list`` and wakes up processes waiting on the timerfd. While iterating
454 the ``cancel_list``, the ``might_cancel`` flag 454 the ``cancel_list``, the ``might_cancel`` flag is consulted to skip stale
455 objects:: 455 objects::
456 456
457 void timerfd_clock_was_set(void) 457 void timerfd_clock_was_set(void)
458 { 458 {
459 ktime_t moffs = ktime_mono_to_ 459 ktime_t moffs = ktime_mono_to_real(0);
460 struct timerfd_ctx *ctx; 460 struct timerfd_ctx *ctx;
461 unsigned long flags; 461 unsigned long flags;
462 462
463 rcu_read_lock(); 463 rcu_read_lock();
464 list_for_each_entry_rcu(ctx, & 464 list_for_each_entry_rcu(ctx, &cancel_list, clist) {
465 if (!ctx->might_cancel 465 if (!ctx->might_cancel)
466 continue; 466 continue;
467 spin_lock_irqsave(&ctx 467 spin_lock_irqsave(&ctx->wqh.lock, flags);
468 if (ctx->moffs != moff 468 if (ctx->moffs != moffs) {
469 ctx->moffs = K 469 ctx->moffs = KTIME_MAX;
470 ctx->ticks++; 470 ctx->ticks++;
471 wake_up_locked 471 wake_up_locked_poll(&ctx->wqh, EPOLLIN);
472 } 472 }
473 spin_unlock_irqrestore 473 spin_unlock_irqrestore(&ctx->wqh.lock, flags);
474 } 474 }
475 rcu_read_unlock(); 475 rcu_read_unlock();
476 } 476 }
477 477
478 The key point is that because RCU-protected tr 478 The key point is that because RCU-protected traversal of the
479 ``cancel_list`` happens concurrently with obje 479 ``cancel_list`` happens concurrently with object addition and removal,
480 sometimes the traversal can access an object t 480 sometimes the traversal can access an object that has been removed from
481 the list. In this example, a flag is used to s 481 the list. In this example, a flag is used to skip such objects.
482 482
483 483
484 Summary 484 Summary
485 ------- 485 -------
486 486
487 Read-mostly list-based data structures that ca 487 Read-mostly list-based data structures that can tolerate stale data are
488 the most amenable to use of RCU. The simplest 488 the most amenable to use of RCU. The simplest case is where entries are
489 either added or deleted from the data structur 489 either added or deleted from the data structure (or atomically modified
490 in place), but non-atomic in-place modificatio 490 in place), but non-atomic in-place modifications can be handled by making
491 a copy, updating the copy, then replacing the 491 a copy, updating the copy, then replacing the original with the copy.
492 If stale data cannot be tolerated, then a *del 492 If stale data cannot be tolerated, then a *deleted* flag may be used
493 in conjunction with a per-entry spinlock in or 493 in conjunction with a per-entry spinlock in order to allow the search
494 function to reject newly deleted data. 494 function to reject newly deleted data.
495 495
496 .. _quick_quiz_answer: 496 .. _quick_quiz_answer:
497 497
498 Answer to Quick Quiz: 498 Answer to Quick Quiz:
499 For the deleted-flag technique to be h 499 For the deleted-flag technique to be helpful, why is it necessary
500 to hold the per-entry lock while retur 500 to hold the per-entry lock while returning from the search function?
501 501
502 If the search function drops the per-e 502 If the search function drops the per-entry lock before returning,
503 then the caller will be processing sta 503 then the caller will be processing stale data in any case. If it
504 is really OK to be processing stale da 504 is really OK to be processing stale data, then you don't need a
505 *deleted* flag. If processing stale d 505 *deleted* flag. If processing stale data really is a problem,
506 then you need to hold the per-entry lo 506 then you need to hold the per-entry lock across all of the code
507 that uses the value that was returned. 507 that uses the value that was returned.
508 508
509 :ref:`Back to Quick Quiz <quick_quiz>` 509 :ref:`Back to Quick Quiz <quick_quiz>`
Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.