kernel/locking/rwbase_rt.c

0001 // SPDX-License-Identifier: GPL-2.0-only
0002
0003 /*
0004  * RT-specific reader/writer semaphores and reader/writer locks
0005  *
0006  * down_write/write_lock()
0007  *  1) Lock rtmutex
0008  *  2) Remove the reader BIAS to force readers into the slow path
0009  *  3) Wait until all readers have left the critical section
0010  *  4) Mark it write locked
0011  *
0012  * up_write/write_unlock()
0013  *  1) Remove the write locked marker
0014  *  2) Set the reader BIAS, so readers can use the fast path again
0015  *  3) Unlock rtmutex, to release blocked readers
0016  *
0017  * down_read/read_lock()
0018  *  1) Try fast path acquisition (reader BIAS is set)
0019  *  2) Take tmutex::wait_lock, which protects the writelocked flag
0020  *  3) If !writelocked, acquire it for read
0021  *  4) If writelocked, block on tmutex
0022  *  5) unlock rtmutex, goto 1)
0023  *
0024  * up_read/read_unlock()
0025  *  1) Try fast path release (reader count != 1)
0026  *  2) Wake the writer waiting in down_write()/write_lock() #3
0027  *
0028  * down_read/read_lock()#3 has the consequence, that rw semaphores and rw
0029  * locks on RT are not writer fair, but writers, which should be avoided in
0030  * RT tasks (think mmap_sem), are subject to the rtmutex priority/DL
0031  * inheritance mechanism.
0032  *
0033  * It's possible to make the rw primitives writer fair by keeping a list of
0034  * active readers. A blocked writer would force all newly incoming readers
0035  * to block on the rtmutex, but the rtmutex would have to be proxy locked
0036  * for one reader after the other. We can't use multi-reader inheritance
0037  * because there is no way to support that with SCHED_DEADLINE.
0038  * Implementing the one by one reader boosting/handover mechanism is a
0039  * major surgery for a very dubious value.
0040  *
0041  * The risk of writer starvation is there, but the pathological use cases
0042  * which trigger it are not necessarily the typical RT workloads.
0043  *
0044  * Fast-path orderings:
0045  * The lock/unlock of readers can run in fast paths: lock and unlock are only
0046  * atomic ops, and there is no inner lock to provide ACQUIRE and RELEASE
0047  * semantics of rwbase_rt. Atomic ops should thus provide _acquire()
0048  * and _release() (or stronger).
0049  *
0050  * Common code shared between RT rw_semaphore and rwlock
0051  */
0052
0053 static __always_inline int rwbase_read_trylock(struct rwbase_rt *rwb)
0054 {
0055     int r;
0056
0057     /*
0058      * Increment reader count, if sem->readers < 0, i.e. READER_BIAS is
0059      * set.
0060      */
0061     for (r = atomic_read(&rwb->readers); r < 0;) {
0062         if (likely(atomic_try_cmpxchg_acquire(&rwb->readers, &r, r + 1)))
0063             return 1;
0064     }
0065     return 0;
0066 }
0067
0068 static int __sched __rwbase_read_lock(struct rwbase_rt *rwb,
0069                       unsigned int state)
0070 {
0071     struct rt_mutex_base *rtm = &rwb->rtmutex;
0072     int ret;
0073
0074     raw_spin_lock_irq(&rtm->wait_lock);
0075     /*
0076      * Allow readers, as long as the writer has not completely
0077      * acquired the semaphore for write.
0078      */
0079     if (atomic_read(&rwb->readers) != WRITER_BIAS) {
0080         atomic_inc(&rwb->readers);
0081         raw_spin_unlock_irq(&rtm->wait_lock);
0082         return 0;
0083     }
0084
0085     /*
0086      * Call into the slow lock path with the rtmutex->wait_lock
0087      * held, so this can't result in the following race:
0088      *
0089      * Reader1      Reader2     Writer
0090      *          down_read()
0091      *                  down_write()
0092      *                  rtmutex_lock(m)
0093      *                  wait()
0094      * down_read()
0095      * unlock(m->wait_lock)
0096      *          up_read()
0097      *          wake(Writer)
0098      *                  lock(m->wait_lock)
0099      *                  sem->writelocked=true
0100      *                  unlock(m->wait_lock)
0101      *
0102      *                  up_write()
0103      *                  sem->writelocked=false
0104      *                  rtmutex_unlock(m)
0105      *          down_read()
0106      *                  down_write()
0107      *                  rtmutex_lock(m)
0108      *                  wait()
0109      * rtmutex_lock(m)
0110      *
0111      * That would put Reader1 behind the writer waiting on
0112      * Reader2 to call up_read(), which might be unbound.
0113      */
0114
0115     trace_contention_begin(rwb, LCB_F_RT | LCB_F_READ);
0116
0117     /*
0118      * For rwlocks this returns 0 unconditionally, so the below
0119      * !ret conditionals are optimized out.
0120      */
0121     ret = rwbase_rtmutex_slowlock_locked(rtm, state);
0122
0123     /*
0124      * On success the rtmutex is held, so there can't be a writer
0125      * active. Increment the reader count and immediately drop the
0126      * rtmutex again.
0127      *
0128      * rtmutex->wait_lock has to be unlocked in any case of course.
0129      */
0130     if (!ret)
0131         atomic_inc(&rwb->readers);
0132     raw_spin_unlock_irq(&rtm->wait_lock);
0133     if (!ret)
0134         rwbase_rtmutex_unlock(rtm);
0135
0136     trace_contention_end(rwb, ret);
0137     return ret;
0138 }
0139
0140 static __always_inline int rwbase_read_lock(struct rwbase_rt *rwb,
0141                         unsigned int state)
0142 {
0143     if (rwbase_read_trylock(rwb))
0144         return 0;
0145
0146     return __rwbase_read_lock(rwb, state);
0147 }
0148
0149 static void __sched __rwbase_read_unlock(struct rwbase_rt *rwb,
0150                      unsigned int state)
0151 {
0152     struct rt_mutex_base *rtm = &rwb->rtmutex;
0153     struct task_struct *owner;
0154     DEFINE_RT_WAKE_Q(wqh);
0155
0156     raw_spin_lock_irq(&rtm->wait_lock);
0157     /*
0158      * Wake the writer, i.e. the rtmutex owner. It might release the
0159      * rtmutex concurrently in the fast path (due to a signal), but to
0160      * clean up rwb->readers it needs to acquire rtm->wait_lock. The
0161      * worst case which can happen is a spurious wakeup.
0162      */
0163     owner = rt_mutex_owner(rtm);
0164     if (owner)
0165         rt_mutex_wake_q_add_task(&wqh, owner, state);
0166
0167     /* Pairs with the preempt_enable in rt_mutex_wake_up_q() */
0168     preempt_disable();
0169     raw_spin_unlock_irq(&rtm->wait_lock);
0170     rt_mutex_wake_up_q(&wqh);
0171 }
0172
0173 static __always_inline void rwbase_read_unlock(struct rwbase_rt *rwb,
0174                            unsigned int state)
0175 {
0176     /*
0177      * rwb->readers can only hit 0 when a writer is waiting for the
0178      * active readers to leave the critical section.
0179      *
0180      * dec_and_test() is fully ordered, provides RELEASE.
0181      */
0182     if (unlikely(atomic_dec_and_test(&rwb->readers)))
0183         __rwbase_read_unlock(rwb, state);
0184 }
0185
0186 static inline void __rwbase_write_unlock(struct rwbase_rt *rwb, int bias,
0187                      unsigned long flags)
0188 {
0189     struct rt_mutex_base *rtm = &rwb->rtmutex;
0190
0191     /*
0192      * _release() is needed in case that reader is in fast path, pairing
0193      * with atomic_try_cmpxchg_acquire() in rwbase_read_trylock().
0194      */
0195     (void)atomic_add_return_release(READER_BIAS - bias, &rwb->readers);
0196     raw_spin_unlock_irqrestore(&rtm->wait_lock, flags);
0197     rwbase_rtmutex_unlock(rtm);
0198 }
0199
0200 static inline void rwbase_write_unlock(struct rwbase_rt *rwb)
0201 {
0202     struct rt_mutex_base *rtm = &rwb->rtmutex;
0203     unsigned long flags;
0204
0205     raw_spin_lock_irqsave(&rtm->wait_lock, flags);
0206     __rwbase_write_unlock(rwb, WRITER_BIAS, flags);
0207 }
0208
0209 static inline void rwbase_write_downgrade(struct rwbase_rt *rwb)
0210 {
0211     struct rt_mutex_base *rtm = &rwb->rtmutex;
0212     unsigned long flags;
0213
0214     raw_spin_lock_irqsave(&rtm->wait_lock, flags);
0215     /* Release it and account current as reader */
0216     __rwbase_write_unlock(rwb, WRITER_BIAS - 1, flags);
0217 }
0218
0219 static inline bool __rwbase_write_trylock(struct rwbase_rt *rwb)
0220 {
0221     /* Can do without CAS because we're serialized by wait_lock. */
0222     lockdep_assert_held(&rwb->rtmutex.wait_lock);
0223
0224     /*
0225      * _acquire is needed in case the reader is in the fast path, pairing
0226      * with rwbase_read_unlock(), provides ACQUIRE.
0227      */
0228     if (!atomic_read_acquire(&rwb->readers)) {
0229         atomic_set(&rwb->readers, WRITER_BIAS);
0230         return 1;
0231     }
0232
0233     return 0;
0234 }
0235
0236 static int __sched rwbase_write_lock(struct rwbase_rt *rwb,
0237                      unsigned int state)
0238 {
0239     struct rt_mutex_base *rtm = &rwb->rtmutex;
0240     unsigned long flags;
0241
0242     /* Take the rtmutex as a first step */
0243     if (rwbase_rtmutex_lock_state(rtm, state))
0244         return -EINTR;
0245
0246     /* Force readers into slow path */
0247     atomic_sub(READER_BIAS, &rwb->readers);
0248
0249     raw_spin_lock_irqsave(&rtm->wait_lock, flags);
0250     if (__rwbase_write_trylock(rwb))
0251         goto out_unlock;
0252
0253     rwbase_set_and_save_current_state(state);
0254     trace_contention_begin(rwb, LCB_F_RT | LCB_F_WRITE);
0255     for (;;) {
0256         /* Optimized out for rwlocks */
0257         if (rwbase_signal_pending_state(state, current)) {
0258             rwbase_restore_current_state();
0259             __rwbase_write_unlock(rwb, 0, flags);
0260             trace_contention_end(rwb, -EINTR);
0261             return -EINTR;
0262         }
0263
0264         if (__rwbase_write_trylock(rwb))
0265             break;
0266
0267         raw_spin_unlock_irqrestore(&rtm->wait_lock, flags);
0268         rwbase_schedule();
0269         raw_spin_lock_irqsave(&rtm->wait_lock, flags);
0270
0271         set_current_state(state);
0272     }
0273     rwbase_restore_current_state();
0274     trace_contention_end(rwb, 0);
0275
0276 out_unlock:
0277     raw_spin_unlock_irqrestore(&rtm->wait_lock, flags);
0278     return 0;
0279 }
0280
0281 static inline int rwbase_write_trylock(struct rwbase_rt *rwb)
0282 {
0283     struct rt_mutex_base *rtm = &rwb->rtmutex;
0284     unsigned long flags;
0285
0286     if (!rwbase_rtmutex_trylock(rtm))
0287         return 0;
0288
0289     atomic_sub(READER_BIAS, &rwb->readers);
0290
0291     raw_spin_lock_irqsave(&rtm->wait_lock, flags);
0292     if (__rwbase_write_trylock(rwb)) {
0293         raw_spin_unlock_irqrestore(&rtm->wait_lock, flags);
0294         return 1;
0295     }
0296     __rwbase_write_unlock(rwb, 0, flags);
0297     return 0;
0298 }