Merge tag 'gfs2-4.8.fixes' of git://git.kernel.org/pub/scm/linux/kernel/git/gfs2...
[cascardo/linux.git] / kernel / locking / rwsem-xadd.c
1 /* rwsem.c: R/W semaphores: contention handling functions
2  *
3  * Written by David Howells (dhowells@redhat.com).
4  * Derived from arch/i386/kernel/semaphore.c
5  *
6  * Writer lock-stealing by Alex Shi <alex.shi@intel.com>
7  * and Michel Lespinasse <walken@google.com>
8  *
9  * Optimistic spinning by Tim Chen <tim.c.chen@intel.com>
10  * and Davidlohr Bueso <davidlohr@hp.com>. Based on mutexes.
11  */
12 #include <linux/rwsem.h>
13 #include <linux/sched.h>
14 #include <linux/init.h>
15 #include <linux/export.h>
16 #include <linux/sched/rt.h>
17 #include <linux/osq_lock.h>
18
19 #include "rwsem.h"
20
21 /*
22  * Guide to the rw_semaphore's count field for common values.
23  * (32-bit case illustrated, similar for 64-bit)
24  *
25  * 0x0000000X   (1) X readers active or attempting lock, no writer waiting
26  *                  X = #active_readers + #readers attempting to lock
27  *                  (X*ACTIVE_BIAS)
28  *
29  * 0x00000000   rwsem is unlocked, and no one is waiting for the lock or
30  *              attempting to read lock or write lock.
31  *
32  * 0xffff000X   (1) X readers active or attempting lock, with waiters for lock
33  *                  X = #active readers + # readers attempting lock
34  *                  (X*ACTIVE_BIAS + WAITING_BIAS)
35  *              (2) 1 writer attempting lock, no waiters for lock
36  *                  X-1 = #active readers + #readers attempting lock
37  *                  ((X-1)*ACTIVE_BIAS + ACTIVE_WRITE_BIAS)
38  *              (3) 1 writer active, no waiters for lock
39  *                  X-1 = #active readers + #readers attempting lock
40  *                  ((X-1)*ACTIVE_BIAS + ACTIVE_WRITE_BIAS)
41  *
42  * 0xffff0001   (1) 1 reader active or attempting lock, waiters for lock
43  *                  (WAITING_BIAS + ACTIVE_BIAS)
44  *              (2) 1 writer active or attempting lock, no waiters for lock
45  *                  (ACTIVE_WRITE_BIAS)
46  *
47  * 0xffff0000   (1) There are writers or readers queued but none active
48  *                  or in the process of attempting lock.
49  *                  (WAITING_BIAS)
50  *              Note: writer can attempt to steal lock for this count by adding
51  *              ACTIVE_WRITE_BIAS in cmpxchg and checking the old count
52  *
53  * 0xfffe0001   (1) 1 writer active, or attempting lock. Waiters on queue.
54  *                  (ACTIVE_WRITE_BIAS + WAITING_BIAS)
55  *
56  * Note: Readers attempt to lock by adding ACTIVE_BIAS in down_read and checking
57  *       the count becomes more than 0 for successful lock acquisition,
58  *       i.e. the case where there are only readers or nobody has lock.
59  *       (1st and 2nd case above).
60  *
61  *       Writers attempt to lock by adding ACTIVE_WRITE_BIAS in down_write and
62  *       checking the count becomes ACTIVE_WRITE_BIAS for successful lock
63  *       acquisition (i.e. nobody else has lock or attempts lock).  If
64  *       unsuccessful, in rwsem_down_write_failed, we'll check to see if there
65  *       are only waiters but none active (5th case above), and attempt to
66  *       steal the lock.
67  *
68  */
69
70 /*
71  * Initialize an rwsem:
72  */
73 void __init_rwsem(struct rw_semaphore *sem, const char *name,
74                   struct lock_class_key *key)
75 {
76 #ifdef CONFIG_DEBUG_LOCK_ALLOC
77         /*
78          * Make sure we are not reinitializing a held semaphore:
79          */
80         debug_check_no_locks_freed((void *)sem, sizeof(*sem));
81         lockdep_init_map(&sem->dep_map, name, key, 0);
82 #endif
83         atomic_long_set(&sem->count, RWSEM_UNLOCKED_VALUE);
84         raw_spin_lock_init(&sem->wait_lock);
85         INIT_LIST_HEAD(&sem->wait_list);
86 #ifdef CONFIG_RWSEM_SPIN_ON_OWNER
87         sem->owner = NULL;
88         osq_lock_init(&sem->osq);
89 #endif
90 }
91
92 EXPORT_SYMBOL(__init_rwsem);
93
94 enum rwsem_waiter_type {
95         RWSEM_WAITING_FOR_WRITE,
96         RWSEM_WAITING_FOR_READ
97 };
98
99 struct rwsem_waiter {
100         struct list_head list;
101         struct task_struct *task;
102         enum rwsem_waiter_type type;
103 };
104
105 enum rwsem_wake_type {
106         RWSEM_WAKE_ANY,         /* Wake whatever's at head of wait list */
107         RWSEM_WAKE_READERS,     /* Wake readers only */
108         RWSEM_WAKE_READ_OWNED   /* Waker thread holds the read lock */
109 };
110
111 /*
112  * handle the lock release when processes blocked on it that can now run
113  * - if we come here from up_xxxx(), then:
114  *   - the 'active part' of count (&0x0000ffff) reached 0 (but may have changed)
115  *   - the 'waiting part' of count (&0xffff0000) is -ve (and will still be so)
116  * - there must be someone on the queue
117  * - the wait_lock must be held by the caller
118  * - tasks are marked for wakeup, the caller must later invoke wake_up_q()
119  *   to actually wakeup the blocked task(s) and drop the reference count,
120  *   preferably when the wait_lock is released
121  * - woken process blocks are discarded from the list after having task zeroed
122  * - writers are only marked woken if downgrading is false
123  */
124 static void __rwsem_mark_wake(struct rw_semaphore *sem,
125                               enum rwsem_wake_type wake_type,
126                               struct wake_q_head *wake_q)
127 {
128         struct rwsem_waiter *waiter, *tmp;
129         long oldcount, woken = 0, adjustment = 0;
130
131         /*
132          * Take a peek at the queue head waiter such that we can determine
133          * the wakeup(s) to perform.
134          */
135         waiter = list_first_entry(&sem->wait_list, struct rwsem_waiter, list);
136
137         if (waiter->type == RWSEM_WAITING_FOR_WRITE) {
138                 if (wake_type == RWSEM_WAKE_ANY) {
139                         /*
140                          * Mark writer at the front of the queue for wakeup.
141                          * Until the task is actually later awoken later by
142                          * the caller, other writers are able to steal it.
143                          * Readers, on the other hand, will block as they
144                          * will notice the queued writer.
145                          */
146                         wake_q_add(wake_q, waiter->task);
147                 }
148
149                 return;
150         }
151
152         /*
153          * Writers might steal the lock before we grant it to the next reader.
154          * We prefer to do the first reader grant before counting readers
155          * so we can bail out early if a writer stole the lock.
156          */
157         if (wake_type != RWSEM_WAKE_READ_OWNED) {
158                 adjustment = RWSEM_ACTIVE_READ_BIAS;
159  try_reader_grant:
160                 oldcount = atomic_long_fetch_add(adjustment, &sem->count);
161                 if (unlikely(oldcount < RWSEM_WAITING_BIAS)) {
162                         /*
163                          * If the count is still less than RWSEM_WAITING_BIAS
164                          * after removing the adjustment, it is assumed that
165                          * a writer has stolen the lock. We have to undo our
166                          * reader grant.
167                          */
168                         if (atomic_long_add_return(-adjustment, &sem->count) <
169                             RWSEM_WAITING_BIAS)
170                                 return;
171
172                         /* Last active locker left. Retry waking readers. */
173                         goto try_reader_grant;
174                 }
175                 /*
176                  * It is not really necessary to set it to reader-owned here,
177                  * but it gives the spinners an early indication that the
178                  * readers now have the lock.
179                  */
180                 rwsem_set_reader_owned(sem);
181         }
182
183         /*
184          * Grant an infinite number of read locks to the readers at the front
185          * of the queue. We know that woken will be at least 1 as we accounted
186          * for above. Note we increment the 'active part' of the count by the
187          * number of readers before waking any processes up.
188          */
189         list_for_each_entry_safe(waiter, tmp, &sem->wait_list, list) {
190                 struct task_struct *tsk;
191
192                 if (waiter->type == RWSEM_WAITING_FOR_WRITE)
193                         break;
194
195                 woken++;
196                 tsk = waiter->task;
197
198                 wake_q_add(wake_q, tsk);
199                 list_del(&waiter->list);
200                 /*
201                  * Ensure that the last operation is setting the reader
202                  * waiter to nil such that rwsem_down_read_failed() cannot
203                  * race with do_exit() by always holding a reference count
204                  * to the task to wakeup.
205                  */
206                 smp_store_release(&waiter->task, NULL);
207         }
208
209         adjustment = woken * RWSEM_ACTIVE_READ_BIAS - adjustment;
210         if (list_empty(&sem->wait_list)) {
211                 /* hit end of list above */
212                 adjustment -= RWSEM_WAITING_BIAS;
213         }
214
215         if (adjustment)
216                 atomic_long_add(adjustment, &sem->count);
217 }
218
219 /*
220  * Wait for the read lock to be granted
221  */
222 __visible
223 struct rw_semaphore __sched *rwsem_down_read_failed(struct rw_semaphore *sem)
224 {
225         long count, adjustment = -RWSEM_ACTIVE_READ_BIAS;
226         struct rwsem_waiter waiter;
227         struct task_struct *tsk = current;
228         WAKE_Q(wake_q);
229
230         waiter.task = tsk;
231         waiter.type = RWSEM_WAITING_FOR_READ;
232
233         raw_spin_lock_irq(&sem->wait_lock);
234         if (list_empty(&sem->wait_list))
235                 adjustment += RWSEM_WAITING_BIAS;
236         list_add_tail(&waiter.list, &sem->wait_list);
237
238         /* we're now waiting on the lock, but no longer actively locking */
239         count = atomic_long_add_return(adjustment, &sem->count);
240
241         /*
242          * If there are no active locks, wake the front queued process(es).
243          *
244          * If there are no writers and we are first in the queue,
245          * wake our own waiter to join the existing active readers !
246          */
247         if (count == RWSEM_WAITING_BIAS ||
248             (count > RWSEM_WAITING_BIAS &&
249              adjustment != -RWSEM_ACTIVE_READ_BIAS))
250                 __rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
251
252         raw_spin_unlock_irq(&sem->wait_lock);
253         wake_up_q(&wake_q);
254
255         /* wait to be given the lock */
256         while (true) {
257                 set_task_state(tsk, TASK_UNINTERRUPTIBLE);
258                 if (!waiter.task)
259                         break;
260                 schedule();
261         }
262
263         __set_task_state(tsk, TASK_RUNNING);
264         return sem;
265 }
266 EXPORT_SYMBOL(rwsem_down_read_failed);
267
268 /*
269  * This function must be called with the sem->wait_lock held to prevent
270  * race conditions between checking the rwsem wait list and setting the
271  * sem->count accordingly.
272  */
273 static inline bool rwsem_try_write_lock(long count, struct rw_semaphore *sem)
274 {
275         /*
276          * Avoid trying to acquire write lock if count isn't RWSEM_WAITING_BIAS.
277          */
278         if (count != RWSEM_WAITING_BIAS)
279                 return false;
280
281         /*
282          * Acquire the lock by trying to set it to ACTIVE_WRITE_BIAS. If there
283          * are other tasks on the wait list, we need to add on WAITING_BIAS.
284          */
285         count = list_is_singular(&sem->wait_list) ?
286                         RWSEM_ACTIVE_WRITE_BIAS :
287                         RWSEM_ACTIVE_WRITE_BIAS + RWSEM_WAITING_BIAS;
288
289         if (atomic_long_cmpxchg_acquire(&sem->count, RWSEM_WAITING_BIAS, count)
290                                                         == RWSEM_WAITING_BIAS) {
291                 rwsem_set_owner(sem);
292                 return true;
293         }
294
295         return false;
296 }
297
298 #ifdef CONFIG_RWSEM_SPIN_ON_OWNER
299 /*
300  * Try to acquire write lock before the writer has been put on wait queue.
301  */
302 static inline bool rwsem_try_write_lock_unqueued(struct rw_semaphore *sem)
303 {
304         long old, count = atomic_long_read(&sem->count);
305
306         while (true) {
307                 if (!(count == 0 || count == RWSEM_WAITING_BIAS))
308                         return false;
309
310                 old = atomic_long_cmpxchg_acquire(&sem->count, count,
311                                       count + RWSEM_ACTIVE_WRITE_BIAS);
312                 if (old == count) {
313                         rwsem_set_owner(sem);
314                         return true;
315                 }
316
317                 count = old;
318         }
319 }
320
321 static inline bool rwsem_can_spin_on_owner(struct rw_semaphore *sem)
322 {
323         struct task_struct *owner;
324         bool ret = true;
325
326         if (need_resched())
327                 return false;
328
329         rcu_read_lock();
330         owner = READ_ONCE(sem->owner);
331         if (!rwsem_owner_is_writer(owner)) {
332                 /*
333                  * Don't spin if the rwsem is readers owned.
334                  */
335                 ret = !rwsem_owner_is_reader(owner);
336                 goto done;
337         }
338
339         ret = owner->on_cpu;
340 done:
341         rcu_read_unlock();
342         return ret;
343 }
344
345 /*
346  * Return true only if we can still spin on the owner field of the rwsem.
347  */
348 static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem)
349 {
350         struct task_struct *owner = READ_ONCE(sem->owner);
351
352         if (!rwsem_owner_is_writer(owner))
353                 goto out;
354
355         rcu_read_lock();
356         while (sem->owner == owner) {
357                 /*
358                  * Ensure we emit the owner->on_cpu, dereference _after_
359                  * checking sem->owner still matches owner, if that fails,
360                  * owner might point to free()d memory, if it still matches,
361                  * the rcu_read_lock() ensures the memory stays valid.
362                  */
363                 barrier();
364
365                 /* abort spinning when need_resched or owner is not running */
366                 if (!owner->on_cpu || need_resched()) {
367                         rcu_read_unlock();
368                         return false;
369                 }
370
371                 cpu_relax_lowlatency();
372         }
373         rcu_read_unlock();
374 out:
375         /*
376          * If there is a new owner or the owner is not set, we continue
377          * spinning.
378          */
379         return !rwsem_owner_is_reader(READ_ONCE(sem->owner));
380 }
381
382 static bool rwsem_optimistic_spin(struct rw_semaphore *sem)
383 {
384         bool taken = false;
385
386         preempt_disable();
387
388         /* sem->wait_lock should not be held when doing optimistic spinning */
389         if (!rwsem_can_spin_on_owner(sem))
390                 goto done;
391
392         if (!osq_lock(&sem->osq))
393                 goto done;
394
395         /*
396          * Optimistically spin on the owner field and attempt to acquire the
397          * lock whenever the owner changes. Spinning will be stopped when:
398          *  1) the owning writer isn't running; or
399          *  2) readers own the lock as we can't determine if they are
400          *     actively running or not.
401          */
402         while (rwsem_spin_on_owner(sem)) {
403                 /*
404                  * Try to acquire the lock
405                  */
406                 if (rwsem_try_write_lock_unqueued(sem)) {
407                         taken = true;
408                         break;
409                 }
410
411                 /*
412                  * When there's no owner, we might have preempted between the
413                  * owner acquiring the lock and setting the owner field. If
414                  * we're an RT task that will live-lock because we won't let
415                  * the owner complete.
416                  */
417                 if (!sem->owner && (need_resched() || rt_task(current)))
418                         break;
419
420                 /*
421                  * The cpu_relax() call is a compiler barrier which forces
422                  * everything in this loop to be re-loaded. We don't need
423                  * memory barriers as we'll eventually observe the right
424                  * values at the cost of a few extra spins.
425                  */
426                 cpu_relax_lowlatency();
427         }
428         osq_unlock(&sem->osq);
429 done:
430         preempt_enable();
431         return taken;
432 }
433
434 /*
435  * Return true if the rwsem has active spinner
436  */
437 static inline bool rwsem_has_spinner(struct rw_semaphore *sem)
438 {
439         return osq_is_locked(&sem->osq);
440 }
441
442 #else
443 static bool rwsem_optimistic_spin(struct rw_semaphore *sem)
444 {
445         return false;
446 }
447
448 static inline bool rwsem_has_spinner(struct rw_semaphore *sem)
449 {
450         return false;
451 }
452 #endif
453
454 /*
455  * Wait until we successfully acquire the write lock
456  */
457 static inline struct rw_semaphore *
458 __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
459 {
460         long count;
461         bool waiting = true; /* any queued threads before us */
462         struct rwsem_waiter waiter;
463         struct rw_semaphore *ret = sem;
464         WAKE_Q(wake_q);
465
466         /* undo write bias from down_write operation, stop active locking */
467         count = atomic_long_sub_return(RWSEM_ACTIVE_WRITE_BIAS, &sem->count);
468
469         /* do optimistic spinning and steal lock if possible */
470         if (rwsem_optimistic_spin(sem))
471                 return sem;
472
473         /*
474          * Optimistic spinning failed, proceed to the slowpath
475          * and block until we can acquire the sem.
476          */
477         waiter.task = current;
478         waiter.type = RWSEM_WAITING_FOR_WRITE;
479
480         raw_spin_lock_irq(&sem->wait_lock);
481
482         /* account for this before adding a new element to the list */
483         if (list_empty(&sem->wait_list))
484                 waiting = false;
485
486         list_add_tail(&waiter.list, &sem->wait_list);
487
488         /* we're now waiting on the lock, but no longer actively locking */
489         if (waiting) {
490                 count = atomic_long_read(&sem->count);
491
492                 /*
493                  * If there were already threads queued before us and there are
494                  * no active writers, the lock must be read owned; so we try to
495                  * wake any read locks that were queued ahead of us.
496                  */
497                 if (count > RWSEM_WAITING_BIAS) {
498                         WAKE_Q(wake_q);
499
500                         __rwsem_mark_wake(sem, RWSEM_WAKE_READERS, &wake_q);
501                         /*
502                          * The wakeup is normally called _after_ the wait_lock
503                          * is released, but given that we are proactively waking
504                          * readers we can deal with the wake_q overhead as it is
505                          * similar to releasing and taking the wait_lock again
506                          * for attempting rwsem_try_write_lock().
507                          */
508                         wake_up_q(&wake_q);
509                 }
510
511         } else
512                 count = atomic_long_add_return(RWSEM_WAITING_BIAS, &sem->count);
513
514         /* wait until we successfully acquire the lock */
515         set_current_state(state);
516         while (true) {
517                 if (rwsem_try_write_lock(count, sem))
518                         break;
519                 raw_spin_unlock_irq(&sem->wait_lock);
520
521                 /* Block until there are no active lockers. */
522                 do {
523                         if (signal_pending_state(state, current))
524                                 goto out_nolock;
525
526                         schedule();
527                         set_current_state(state);
528                 } while ((count = atomic_long_read(&sem->count)) & RWSEM_ACTIVE_MASK);
529
530                 raw_spin_lock_irq(&sem->wait_lock);
531         }
532         __set_current_state(TASK_RUNNING);
533         list_del(&waiter.list);
534         raw_spin_unlock_irq(&sem->wait_lock);
535
536         return ret;
537
538 out_nolock:
539         __set_current_state(TASK_RUNNING);
540         raw_spin_lock_irq(&sem->wait_lock);
541         list_del(&waiter.list);
542         if (list_empty(&sem->wait_list))
543                 atomic_long_add(-RWSEM_WAITING_BIAS, &sem->count);
544         else
545                 __rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
546         raw_spin_unlock_irq(&sem->wait_lock);
547         wake_up_q(&wake_q);
548
549         return ERR_PTR(-EINTR);
550 }
551
552 __visible struct rw_semaphore * __sched
553 rwsem_down_write_failed(struct rw_semaphore *sem)
554 {
555         return __rwsem_down_write_failed_common(sem, TASK_UNINTERRUPTIBLE);
556 }
557 EXPORT_SYMBOL(rwsem_down_write_failed);
558
559 __visible struct rw_semaphore * __sched
560 rwsem_down_write_failed_killable(struct rw_semaphore *sem)
561 {
562         return __rwsem_down_write_failed_common(sem, TASK_KILLABLE);
563 }
564 EXPORT_SYMBOL(rwsem_down_write_failed_killable);
565
566 /*
567  * handle waking up a waiter on the semaphore
568  * - up_read/up_write has decremented the active part of count if we come here
569  */
570 __visible
571 struct rw_semaphore *rwsem_wake(struct rw_semaphore *sem)
572 {
573         unsigned long flags;
574         WAKE_Q(wake_q);
575
576         /*
577          * If a spinner is present, it is not necessary to do the wakeup.
578          * Try to do wakeup only if the trylock succeeds to minimize
579          * spinlock contention which may introduce too much delay in the
580          * unlock operation.
581          *
582          *    spinning writer           up_write/up_read caller
583          *    ---------------           -----------------------
584          * [S]   osq_unlock()           [L]   osq
585          *       MB                           RMB
586          * [RmW] rwsem_try_write_lock() [RmW] spin_trylock(wait_lock)
587          *
588          * Here, it is important to make sure that there won't be a missed
589          * wakeup while the rwsem is free and the only spinning writer goes
590          * to sleep without taking the rwsem. Even when the spinning writer
591          * is just going to break out of the waiting loop, it will still do
592          * a trylock in rwsem_down_write_failed() before sleeping. IOW, if
593          * rwsem_has_spinner() is true, it will guarantee at least one
594          * trylock attempt on the rwsem later on.
595          */
596         if (rwsem_has_spinner(sem)) {
597                 /*
598                  * The smp_rmb() here is to make sure that the spinner
599                  * state is consulted before reading the wait_lock.
600                  */
601                 smp_rmb();
602                 if (!raw_spin_trylock_irqsave(&sem->wait_lock, flags))
603                         return sem;
604                 goto locked;
605         }
606         raw_spin_lock_irqsave(&sem->wait_lock, flags);
607 locked:
608
609         if (!list_empty(&sem->wait_list))
610                 __rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
611
612         raw_spin_unlock_irqrestore(&sem->wait_lock, flags);
613         wake_up_q(&wake_q);
614
615         return sem;
616 }
617 EXPORT_SYMBOL(rwsem_wake);
618
619 /*
620  * downgrade a write lock into a read lock
621  * - caller incremented waiting part of count and discovered it still negative
622  * - just wake up any readers at the front of the queue
623  */
624 __visible
625 struct rw_semaphore *rwsem_downgrade_wake(struct rw_semaphore *sem)
626 {
627         unsigned long flags;
628         WAKE_Q(wake_q);
629
630         raw_spin_lock_irqsave(&sem->wait_lock, flags);
631
632         if (!list_empty(&sem->wait_list))
633                 __rwsem_mark_wake(sem, RWSEM_WAKE_READ_OWNED, &wake_q);
634
635         raw_spin_unlock_irqrestore(&sem->wait_lock, flags);
636         wake_up_q(&wake_q);
637
638         return sem;
639 }
640 EXPORT_SYMBOL(rwsem_downgrade_wake);