内核同步机制之（五）：rwsem - 读写锁

/*

* For an uncontended rwsem, count and owner are the only fields a task

* needs to touch when acquiring the rwsem. So they are put next to each

* other to increase the chance that they will share the same cacheline.

*

* In a contended rwsem, the owner is likely the most frequently accessed

* field in the structure as the optimistic waiter that holds the osq lock

* will spin on owner. For an embedded rwsem, other hot fields in the

* containing structure should be moved further away from the rwsem to

* reduce the chance that they will share the same cacheline causing

* cacheline bouncing problem.

*/

struct rw_semaphore {

    //count的不用的bit代表不同的含义

    atomic_long_t count;

    /*

     * Write owner or one of the read owners as well flags regarding

     * the current state of the rwsem. Can be used as a speculative

     * check to see if the write owner is running on the cpu.

     */

    //如果是读持锁，只能记录最后一个进入临界区的reader

    //如果是写持锁，因为临界区只允许一个writer，owner记录的是那个唯一的writer

    atomic_long_t owner;

#ifdef CONFIG_RWSEM_SPIN_ON_OWNER

    struct optimistic_spin_queue osq; /* spinner MCS lock */

#endif

    raw_spinlock_t wait_lock;

    //阻塞的线程都挂在这个链表上，新加进来的task挂在链表末尾，唤醒从链表最前面开始唤醒

    struct list_head wait_list;

#ifdef CONFIG_DEBUG_RWSEMS

    void *magic;

#endif

#ifdef CONFIG_DEBUG_LOCK_ALLOC

    struct lockdep_map dep_map;

#endif

};

/*

* On 64-bit architectures, the bit definitions of the count are:

*

* Bit 0 - writer locked bit

* Bit 1 - waiters present bit

* Bit 2 - lock handoff bit

* Bits 3-7 - reserved

* Bits 8-62 - 55-bit reader count

* Bit 63 - read fail bit

*

* On 32-bit architectures, the bit definitions of the count are:

*

* Bit 0 - writer locked bit

* Bit 1 - waiters present bit

* Bit 2 - lock handoff bit

* Bits 3-7 - reserved

* Bits 8-30 - 23-bit reader count

* Bit 31 - read fail bit

*

* It is not likely that the most significant bit (read fail bit) will ever

* be set. This guard bit is still checked anyway in the down_read() fastpath

* just in case we need to use up more of the reader bits for other purpose

* in the future.

*

* atomic_long_fetch_add() is used to obtain reader lock, whereas

* atomic_long_cmpxchg() will be used to obtain writer lock.

*

* There are three places where the lock handoff bit may be set or cleared.

* 1) rwsem_mark_wake() for readers.

* 2) rwsem_try_write_lock() for writers.

* 3) Error path of rwsem_down_write_slowpath().

*

* For all the above cases, wait_lock will be held. A writer must also

* be the first one in the wait_list to be eligible for setting the handoff

* bit. So concurrent setting/clearing of handoff bit is not possible.

*/

//RWSEM_WRITER_LOCKED : 标记当前是reader还是writer在临界区

//RWSEM_FLAG_WAITERS : 标记当前wait_list链表上有阻塞的线程

//RWSEM_FLAG_HANDOFF : 禁止自旋"偷锁"，下一次将锁交给wait_list最前面的waiter

//RWSEM_FLAG_READFAIL : reader在临界区，并且临界区的reader已经非常多了，计数已经溢出

// 因为其位于最高bit，所以代码中一般通过判断正负，来判断是否有太多

// reader在临界区了

#define RWSEM_WRITER_LOCKED    (1UL << 0)                                //bit0

#define RWSEM_FLAG_WAITERS    (1UL << 1)                                //bit1

#define RWSEM_FLAG_HANDOFF    (1UL << 2)                                //bit2

#define RWSEM_FLAG_READFAIL    (1UL << (BITS_PER_LONG - 1))            //最高bit

//RWSEM_READER_BIAS : 实现对count字段加1

//RWSEM_READER_MASK : 不为0时，表示当前有reader在临界区

#define RWSEM_READER_SHIFT    8

#define RWSEM_READER_BIAS        (1UL << RWSEM_READER_SHIFT)            //可直线count字段加1

#define RWSEM_READER_MASK        (~(RWSEM_READER_BIAS - 1))            //bit[8:63]

//RWSEM_WRITER_MASK : 为1时，表示当前writer在临界区

//RWSEM_LOCK_MASK : 该字段不为0，表示当前有人在临界区，不管是reader还是writer

#define RWSEM_WRITER_MASK        RWSEM_WRITER_LOCKED                        //bit0

#define RWSEM_LOCK_MASK        (RWSEM_WRITER_MASK|RWSEM_READER_MASK)    //bit0和bit[8:63]

//除去count和resv字段之外的所有字段，如果该字段不为0，表示后

//来的reader不能通过快速路径拿到锁，必须进行乐观自旋或者睡眠了

// a) 已经有writer进入临界区

// b) 或有人（可能是reader或writer）阻塞在这把锁上

// c) 或当前已近禁止自旋"偷锁"

// d) 临界区的reader已经撑爆了

#define RWSEM_READ_FAILED_MASK    (RWSEM_WRITER_MASK|RWSEM_FLAG_WAITERS|\

                 RWSEM_FLAG_HANDOFF|RWSEM_FLAG_READFAIL)

/*

* The least significant 3 bits of the owner value has the following

* meanings when set.

* - Bit 0: RWSEM_READER_OWNED - The rwsem is owned by readers

* - Bit 1: RWSEM_RD_NONSPINNABLE - Readers cannot spin on this lock.

* - Bit 2: RWSEM_WR_NONSPINNABLE - Writers cannot spin on this lock.

*

* When the rwsem is either owned by an anonymous writer, or it is

* reader-owned, but a spinning writer has timed out, both nonspinnable

* bits will be set to disable optimistic spinning by readers and writers.

* In the later case, the last unlocking reader should then check the

* writer nonspinnable bit and clear it only to give writers preference

* to acquire the lock via optimistic spinning, but not readers. Similar

* action is also done in the reader slowpath.

* When a writer acquires a rwsem, it puts its task_struct pointer

* into the owner field. It is cleared after an unlock.

*

* When a reader acquires a rwsem, it will also puts its task_struct

* pointer into the owner field with the RWSEM_READER_OWNED bit set.

* On unlock, the owner field will largely be left untouched. So

* for a free or reader-owned rwsem, the owner value may contain

* information about the last reader that acquires the rwsem.

*

* That information may be helpful in debugging cases where the system

* seems to hang on a reader owned rwsem especially if only one reader

* is involved. Ideally we would like to track all the readers that own

* a rwsem, but the overhead is simply too big.

*

* Reader optimistic spinning is helpful when the reader critical section

* is short and there aren't that many readers around. It makes readers

* relatively more preferred than writers. When a writer times out spinning

* on a reader-owned lock and set the nospinnable bits, there are two main

* reasons for that.

*

* 1) The reader critical section is long, perhaps the task sleeps after

* acquiring the read lock.

* 2) There are just too many readers contending the lock causing it to

* take a while to service all of them.

*

* In the former case, long reader critical section will impede the progress

* of writers which is usually more important for system performance. In

* the later case, reader optimistic spinning tends to make the reader

* groups that contain readers that acquire the lock together smaller

* leading to more of them. That may hurt performance in some cases. In

* other words, the setting of nonspinnable bits indicates that reader

* optimistic spinning may not be helpful for those workloads that cause

* it.

*

* Therefore, any writers that had observed the setting of the writer

* nonspinnable bit for a given rwsem after they fail to acquire the lock

* via optimistic spinning will set the reader nonspinnable bit once they

* acquire the write lock. Similarly, readers that observe the setting

* of reader nonspinnable bit at slowpath entry will set the reader

* nonspinnable bits when they acquire the read lock via the wakeup path.

*

* Once the reader nonspinnable bit is on, it will only be reset when

* a writer is able to acquire the rwsem in the fast path or somehow a

* reader or writer in the slowpath doesn't observe the nonspinable bit.

*

* This is to discourage reader optmistic spinning on that particular

* rwsem and make writers more preferred. This adaptive disabling of reader

* optimistic spinning will alleviate the negative side effect of this

* feature.

*/

//RWSEM_READER_OWNED : 标记这把锁是否被reader持有，1读持锁；0写持锁

//RWSEM_RD_NONSPINNABLE : 是否允许reader执行乐观自旋逻辑，还是直接进阻塞状态

//RWSEM_WR_NONSPINNABLE : 是否允许writer执行乐观自旋逻辑，还是直接进阻塞状态

#define RWSEM_READER_OWNED        (1UL << 0)

#define RWSEM_RD_NONSPINNABLE        (1UL << 1)

#define RWSEM_WR_NONSPINNABLE        (1UL << 2)

#define RWSEM_NONSPINNABLE        (RWSEM_RD_NONSPINNABLE | RWSEM_WR_NONSPINNABLE)

#define RWSEM_OWNER_FLAGS_MASK    (RWSEM_READER_OWNED | RWSEM_NONSPINNABLE)

struct rwsem_waiter {

    //用于将该结构挂在rwsem->wait_list链表上

    struct list_head list;

    //waiter的task_struct结构

    struct task_struct *task;

    //标记老子是一个reader还是writer

    enum rwsem_waiter_type type;

    //一个时间戳，waiter如果在链表上阻塞超过4ms，并且当前也已经是

    //链表最前面的一个waiter了，就设置HANDOFF标记防止被自旋者"偷锁"

    unsigned long timeout;

    //记录在自旋或睡眠前，也就是申请锁的时候锁的owner

    //上一次在rwsem_down_read_slowpath中，上一次读时候的锁的owner

    unsigned long last_rowner;

};

enum rwsem_waiter_type {

    RWSEM_WAITING_FOR_WRITE,            //当前线程是一个writer

    RWSEM_WAITING_FOR_READ            //当前线程是一个reader

};

struct optimistic_spin_queue {

    /*

     * Stores an encoded value of the CPU # of the tail node in the queue.

     * If the queue is empty, then it's set to OSQ_UNLOCKED_VAL.

     */

    atomic_t tail;

};

/*

* An MCS like lock especially tailored for optimistic spinning for sleeping

* lock implementations (mutex, rwsem, etc).

*/

struct optimistic_spin_node {

    struct optimistic_spin_node *next, *prev;

    int locked; /* 1 if lock acquired */

    int cpu; /* encoded CPU # + 1 value */

};

bool osq_lock(struct optimistic_spin_queue *lock)

{

    struct optimistic_spin_node *node = this_cpu_ptr(&osq_node);

    struct optimistic_spin_node *prev, *next;

    int curr = encode_cpu(smp_processor_id());

    int old;

    node->locked = 0;

    node->next = NULL;

    node->cpu = curr;

    /*

     * We need both ACQUIRE (pairs with corresponding RELEASE in

     * unlock() uncontended, or fastpath) and RELEASE (to publish

     * the node fields we just initialised) semantics when updating

     * the lock tail.

     */

    old = atomic_xchg(&lock->tail, curr);

    if (old == OSQ_UNLOCKED_VAL)

        return true;

    prev = decode_cpu(old);

    node->prev = prev;

    /*

     * osq_lock()            unqueue

     *

     * node->prev = prev        osq_wait_next()

     * WMB                MB

     * prev->next = node        next->prev = prev // unqueue-C

     *

     * Here 'node->prev' and 'next->prev' are the same variable and we need

     * to ensure these stores happen in-order to avoid corrupting the list.

     */

    smp_wmb();

    WRITE_ONCE(prev->next, node);

    /*

     * Normally @prev is untouchable after the above store; because at that

     * moment unlock can proceed and wipe the node element from stack.

     *

     * However, since our nodes are static per-cpu storage, we're

     * guaranteed their existence -- this allows us to apply

     * cmpxchg in an attempt to undo our queueing.

     */

    /*

     * Wait to acquire the lock or cancelation. Note that need_resched()

     * will come with an IPI, which will wake smp_cond_load_relaxed() if it

     * is implemented with a monitor-wait. vcpu_is_preempted() relies on

     * polling, be careful.

     */

    if (smp_cond_load_relaxed(&node->locked, VAL || need_resched() ||

                 vcpu_is_preempted(node_cpu(node->prev))))

        return true;

    /* unqueue */

    /*

     * Step - A -- stabilize @prev

     *

     * Undo our @prev->next assignment; this will make @prev's

     * unlock()/unqueue() wait for a next pointer since @lock points to us

     * (or later).

     */

    for (;;) {

        /*

         * cpu_relax() below implies a compiler barrier which would

         * prevent this comparison being optimized away.

         */

        if (data_race(prev->next) == node &&

         cmpxchg(&prev->next, node, NULL) == node)

            break;

        /*

         * We can only fail the cmpxchg() racing against an unlock(),

         * in which case we should observe @node->locked becomming

         * true.

         */

        if (smp_load_acquire(&node->locked))

            return true;

        cpu_relax();

        /*

         * Or we race against a concurrent unqueue()'s step-B, in which

         * case its step-C will write us a new @node->prev pointer.

         */

        prev = READ_ONCE(node->prev);

    }

    /*

     * Step - B -- stabilize @next

     *

     * Similar to unlock(), wait for @node->next or move @lock from @node

     * back to @prev.

     */

    next = osq_wait_next(lock, node, prev);

    if (!next)

        return false;

    /*

     * Step - C -- unlink

     *

     * @prev is stable because its still waiting for a new @prev->next

     * pointer, @next is stable because our @node->next pointer is NULL and

     * it will wait in Step-A.

     */

    WRITE_ONCE(next->prev, prev);

    WRITE_ONCE(prev->next, next);

    return false;

}

/*

* We use the value 0 to represent "no CPU", thus the encoded value

* will be the CPU number incremented by 1.

*/

static inline int encode_cpu(int cpu_nr)

{

    return cpu_nr + 1;

}

static inline int node_cpu(struct optimistic_spin_node *node)

{

    return node->cpu - 1;

}

static inline struct optimistic_spin_node *decode_cpu(int encoded_cpu_val)

{

    int cpu_nr = encoded_cpu_val - 1;

    return per_cpu_ptr(&osq_node, cpu_nr);

}

void osq_unlock(struct optimistic_spin_queue *lock)

{

    struct optimistic_spin_node *node, *next;

    int curr = encode_cpu(smp_processor_id());

    /*

     * Fast path for the uncontended case.

     */

    if (likely(atomic_cmpxchg_release(&lock->tail, curr,

                     OSQ_UNLOCKED_VAL) == curr))

        return;

    /*

     * Second most likely case.

     */

    node = this_cpu_ptr(&osq_node);

    next = xchg(&node->next, NULL);

    if (next) {

        WRITE_ONCE(next->locked, 1);

        return;

    }

    next = osq_wait_next(lock, node, NULL);

    if (next)

        WRITE_ONCE(next->locked, 1);

}

#ifdef CONFIG_RWSEM_SPIN_ON_OWNER

static inline bool rwsem_can_spin_on_owner(

            struct rw_semaphore *sem,

            unsigned long nonspinnable)        //锁的申请者是读者自旋还是写者自旋

{

    struct task_struct *owner;

    unsigned long flags;

    bool ret = true;

    //1.如果正在申请锁的线程current在下一个调度时机即将调度出去，那么也没有必要自旋了

    if (need_resched()) {

        lockevent_inc(rwsem_opt_fail);

        return false;

    }

    preempt_disable();

    rcu_read_lock();

    //2.代码走到这里，表示申请锁的这个线程一时半会是不会调度出去的，此时就需要判断

    // 是不是需要自旋等待了，而一个线程是否需要自旋等待取决于当前正在持锁的这个线程

    // 下面剥离owner的flags和task_struct字段，分别保存在flags和owner变量中

    owner = rwsem_owner_flags(sem, &flags);

    /*

     * Don't check the read-owner as the entry may be stale.

     */

    //3.下面场景也不允许自旋

    // a) 锁已经被打上nonspinnable标签（已经禁止自旋"偷锁"），那就不用自旋

    // b) 锁虽然没有被打上nonspinnable标签，但是锁已经被writer持有，

    // 但是writer当前不是处于running状态，可能是runnable或者已

    // 经进入阻塞状态，但是不管是那种状态，都表示这个writer短时间

    // 不会释放锁（runnable表示至少一个调度时间片都没有完成，也表

    // 明短时间不会释放锁）

    // 注意：当锁没有被打上nonspinnable标签时，并且满足下面两种情况允许自旋

    // a) 如果是reader在临界区的话，后面来的不管是reader还是writer，

    // 都可以自旋一会，为啥捏？？？

    // 这主要考虑reader很快就能结束，自旋一会可能很快就能拿到锁

    // b) 如果是writer在临界区的话，并且这个writer当前正在running

    // 那么也会返回true，笔者感觉这里有赌的成分，因为内核简单的认

    // 为只有在runnable或阻塞状态的writer，才表示这个writer短时

    // 间之内不会释放锁，那么既然writer很快能退出，那么我们也等一会把

    if ((flags & nonspinnable) ||

     (owner && !(flags & RWSEM_READER_OWNED) && !owner_on_cpu(owner)))

        ret = false;

    rcu_read_unlock();

    preempt_enable();

    lockevent_cond_inc(rwsem_opt_fail, !ret);

    return ret;

}

#else

static inline bool rwsem_can_spin_on_owner(

            struct rw_semaphore *sem,

            unsigned long nonspinnable)

{

    return false;

}

#endif

/*

* Return the real task structure pointer of the owner and the embedded

* flags in the owner. pflags must be non-NULL.

*/

static inline struct task_struct * rwsem_owner_flags(

            struct rw_semaphore *sem,

            unsigned long *pflags)

{

    unsigned long owner = atomic_long_read(&sem->owner);

    //1.返回低3bit

    *pflags = owner & RWSEM_OWNER_FLAGS_MASK;

    //2.屏蔽掉低3bit，返回真正的task_struct结构

    return (struct task_struct *)(owner & ~RWSEM_OWNER_FLAGS_MASK);

}

static inline bool owner_on_cpu(struct task_struct *owner)

{

    /*

     * As lock holder preemption issue, we both skip spinning if

     * task is not on cpu or its cpu is preempted

     */

    //第一个条件表示持锁者当前正在running，不是出于runnable和阻塞状态

    //第二个条件对arm64来说始终为false，我们暂不深究

    return owner->on_cpu && !vcpu_is_preempted(task_cpu(owner));

}

#ifdef CONFIG_RWSEM_SPIN_ON_OWNER

static bool rwsem_optimistic_spin(

            struct rw_semaphore *sem,

            bool wlock)                //指示申请锁的人是reader还是writer，

{                                        //readr传false，writer传true

    bool taken = false;

    int prev_owner_state = OWNER_NULL;

    int loop = 0;

    u64 rspin_threshold = 0;

    //1.wlock标记申请锁的人是想以读的身份还是写的身份进入临界区，

    // 申请者要读临界区传false，要写临界区传true，下面转换标记位

    unsigned long nonspinnable = wlock ?

                RWSEM_WR_NONSPINNABLE : RWSEM_RD_NONSPINNABLE;

    preempt_disable();

    /* sem->wait_lock should not be held when doing optimistic spinning */

    //2.标记乐观自旋开始

    if (!osq_lock(&sem->osq))

        goto done;

    /*

     * Optimistically spin on the owner field and attempt to acquire the

     * lock whenever the owner changes. Spinning will be stopped when:

     * 1) the owning writer isn't running; or

     * 2) readers own the lock and spinning time has exceeded limit.

     */

    //3.乐观自旋外循环，当出现下面情况时，退出自旋

    // a) writer持锁，但是writer并没有在运行，而是已经进入阻塞状态

    // b) reader持锁，但是申请锁的人是writer，并且这个writer已经自旋超时

    //注意：只要没有超时，当reader在临界区时，后来申请锁的人就可以一直等下去

    for (;;) {

        enum owner_state owner_state;

        //3.1 内循环，顾名思义，围着这个owner转，直到owner或者flags发生变化

        // 才会退出小循环，实际上，当rwsem_spin_on_owner围着owner转时，这

        // 个owner一定是writer，其他情况该函数直接返回。函数返回值可为：

        // a) OWNER_NULL: 没人在临界区

        // b) OWNER_WRITER: writer在临界区

        // c) OWNER_READER: reader在临界区

        // d) OWNER_NONSPINNABLE: 锁已经被打上nospin标记

        // OWNER_SPINNABLE定义为(OWNER_NULL | OWNER_WRITER | OWNER_READER)

        // 也就是说，当临界区没人、写在临界区、读在临界区时，都是有可能自旋的

        // 下面的逻辑表示，一旦发现锁已经被打上nospin标记（禁止自旋），就直接

        // 退出大循环，并返回false，自旋拿锁失败，后面将会进入睡眠状态

        owner_state = rwsem_spin_on_owner(sem, nonspinnable);

        if (!(owner_state & OWNER_SPINNABLE))

            break;

        //3.2 代码走到这里，这把锁一定没有被打上nospinable标记，此时的临界区要么

        // 没人，要么是reader或者writer，此时需要进一步判断是否需要继续自旋

        /*

         * Try to acquire the lock

         */

        //3.3 执行"偷锁"逻辑

        // 从上面小循环中返回，代码走到这里有下面三种情况

        // a) OWNER_NULL: 临界区没人，那么可以在这里尝试获取锁

        // b) OWNER_READER: reader在临界区，此时如果申请锁的人也是reader的话，

        // 那么可以直接进入临界区

        // c) OWNER_WRITER: writer在临界区，在经过上面rwsem_spin_on_owner中

        // 返回，表示有可能owner或者owner的flags发生了变化，

        // 下面尝试获取锁

        //注意：

        // 如果wait_list链表上有线程阻塞着，则这些阻塞的线程肯定是比这里正在自旋的

        // 申请者要先申请锁，但是在这里，后来的申请者却通过自旋的方式却先拿到了锁，

        // 就像是后来的申请者插队了，这公平吗？这就是传说中的"偷锁"!!!

        //注意：

        // 当一个reader在申请锁的时候，不管临界区时reader还是writer，即使这把锁

        // 被打上了RWSEM_FLAG_HANDOFF标记，如果reader在这里"偷锁"失败，reader

        // 也会继续自旋下去，直到在上面rwsem_spin_on_owner中发现这把锁已经被打上

        // nonspinable标记时（临界区reader已经撑爆，或者有一个writer已经自旋超时

        // 了），才会结束自旋

        taken = wlock ? rwsem_try_write_lock_unqueued(sem)

                         : rwsem_try_read_lock_unqueued(sem);

        if (taken)

            break;

        /*

         * Time-based reader-owned rwsem optimistic spinning

         */

        //3.4 下面的逻辑表示当前是writer在自旋申请锁，但是当前是reader在临界区，

        // 因为我们认为reader一般很快会退出临界区，所以此时writer会再等一会，

        // 但是需要注意的是，由上面对rwsem_try_read_lock_unqueued的分析可知，

        // 在writer等待期间，当reader在临界区并且锁没有被打上RWSEM_FLAG_HANDOFF

        // 标记时，那么后来的reader会通过自旋"偷锁"的方式源源不断的进入临界区，

        // 也就是说在writer自旋期间，可能会不断的有reader插队，这样writer不就

        // 永远拿不到锁了吗？为了确保writer不被饿死，不能一直让reader源源不断的

        // 进入临界区吧！解决办法就是禁止reader自旋。

        //注意：

        // 这里将锁的RWSEM_RD_NONSPINNABLE和RWSEM_WR_NONSPINNABLE位全部置位，

        // 因为此时writer已经长时间自旋都没有拿到锁了，全部置位主要有两个原因：

        // i) 设置RWSEM_WR_NONSPINNABLE，告诉后来的writer不要做无意义的

        // 尝试了，前辈已翻车你就不要自旋了，直接进sleep状态

        // ii) 设置RWSEM_RD_NONSPINNABLE，后来的reader也不不会再执行自旋

        // 的逻辑，也就不会再出现reader源源不断进入临界区的情况，从源头

        // 上防止reader的临界区越来越大

        if (wlock && (owner_state == OWNER_READER)) {

            /*

             * Re-initialize rspin_threshold every time when

             * the owner state changes from non-reader to reader.

             * This allows a writer to steal the lock in between

             * 2 reader phases and have the threshold reset at

             * the beginning of the 2nd reader phase.

             */

            //3.5 在reader刚进临界区的时候计算writer自旋的阈值

            // （这个writer在上一次循环时，临界区还不是reader（可能没人，

            // 也可能是writer在临界区），但是这次循环却发现当前临界区已经是

            // reader了，说明这些reader是刚被放进来的）

            if (prev_owner_state != OWNER_READER) {

                //3.5.1 在重新设置阈值之前，再次检查锁是否被打上了nospinnable

                // 标记，如果已经被打上了nonspinnable标记，则表示已经禁止

                // writer自旋，此时直接结束自旋，（注意：此时nonspinnable

                // 为RWSEM_WR_NONSPINNABLE）

                // reader在临界区时，什么时候会被打上nospin标记呢？

                // a) 临界区已经有很多reader了，已经溢出了

                // b) 有一个writer已经自旋等了很长时间，再等就饿死了

                if (rwsem_test_oflags(sem, nonspinnable))

                    break;

                //3.5.2 重新计算自旋的时间

                rspin_threshold = rwsem_rspin_threshold(sem);

                loop = 0;

            }

            /*

             * Check time threshold once every 16 iterations to

             * avoid calling sched_clock() too frequently so

             * as to reduce the average latency between the times

             * when the lock becomes free and when the spinner

             * is ready to do a trylock.

             */

            //3.6 代码如果走到这个分支，则表示这个writer在上一次循环时发现临界区时reader，

            // 这次循环发现临界区还是reader，因为当reader在临界区时，后续申请锁的reader

            // 是可以直接进入临界区的（只要临界区的reader没有被撑爆，并且锁没有被置为

            // RWSEM_FLAG_HANDOFF标记，详见rwsem_try_read_lock_unqueued），也就是说，

            // 当这个writer在这里自旋时，可能会源源不断的有新的reader进入临界区，如

            // 此下去的话，writer会一直自旋下去也拿不到锁，这显然不合理，所以我们在这

            // 里设置了一个阈值，当writer自旋一段时间后没有拿到锁，就结束自旋，返回false

            //注意到：

            // writer在自旋超期拿锁失败退出时，将锁的nospin标记置位，这是为啥捏？

            // 试想，一旦nospin标记被置位，那么后来的申请锁的人（不管是reader还

            // 是writer），都不会自旋了，而是直接进入D状态。为啥要这么做捏？

            // 这还是防止这个writer饿死！试想，只要允许自旋，后来的reader就有可

            // 能在自旋期间拿到锁直接进入临界区（只要临界区的reader没有被撑爆，并

            // 且锁没有被置为RWSEM_FLAG_HANDOFF标记，详见rwsem_try_read_lock_unqueued），

            // 要知道wait_list上阻塞的线程是先申请锁的啊，这里通过自旋拿锁不就相

            // 当于插队一样吗，这对于wait_list上阻塞的线程显然是不公平的，因为这

            // 样wait_list链表上的writer就会越等越久，因为前面不断有人插队，这

            // 样writer不就饿死了吗！所以，这里为了防止writer饿死，在进D状态之

            // 前，关闭插队机制，即对这把锁打上nospin标记

            // 换句话说，只要wait_list上有writer，后来的申请者，不管是reader

            // 还是writer，都一定不能不能自旋

            // 另外一个注意点，下面每循环16次检查一次自旋时间是否已经超过预期限制

            // 的值，这主要是为了防止频繁的调用sched_clock

            else if (!(++loop & 0xf) && (sched_clock() > rspin_threshold)) {

                rwsem_set_nonspinnable(sem);

                lockevent_inc(rwsem_opt_nospin);

                break;

            }

        }

        /*

         * An RT task cannot do optimistic spinning if it cannot

         * be sure the lock holder is running or live-lock may

         * happen if the current task and the lock holder happen

         * to run in the same CPU. However, aborting optimistic

         * spinning while a NULL owner is detected may miss some

         * opportunity where spinning can continue without causing

         * problem.

         *

         * There are 2 possible cases where an RT task may be able

         * to continue spinning.

         *

         * 1) The lock owner is in the process of releasing the

         * lock, sem->owner is cleared but the lock has not

         * been released yet.

         * 2) The lock was free and owner cleared, but another

         * task just comes in and acquire the lock before

         * we try to get it. The new owner may be a spinnable

         * writer.

         *

         * To take advantage of two scenarios listed agove, the RT

         * task is made to retry one more time to see if it can

         * acquire the lock or continue spinning on the new owning

         * writer. Of course, if the time lag is long enough or the

         * new owner is not a writer or spinnable, the RT task will

         * quit spinning.

         *

         * If the owner is a writer, the need_resched() check is

         * done inside rwsem_spin_on_owner(). If the owner is not

         * a writer, need_resched() check needs to be done here.

         */

        //3.7 关于rt线程的优化

        // 由上面的注释可知，当rt线程和持锁的线程运行在同一个cpu上时，可能会

        // 产生live-lock，（笔者在这里把他翻译为"活锁"）。那么什么是"活锁"呢？

        // 试想，在同一个cpu上，一个普通的cfs线程先拿到锁进入临界区，此时一个

        // rt线程来申请锁，那么这个rt线程必然会打断这个cfs线程，只要这个rt

        // 在自旋，那么这个cfs线程就永远得不到运行，也就永远无法释放锁，这就是

        // "活锁"

        // 所以，这里理论上只要探测到时rt线程就应该立刻返回，实际上原来的代码就

        // 是这样的逻辑，修改前代码如下，在990fa7384a305这笔提交做了修改，因为

        // 作者Waiman Long通过实验发现，让rt线程在这里多尝试一次是可以提升性能，

        // 具体参见上面的patch

        // if (!sem->owner && (need_resched() || rt_task(current)))

        // break;

        // 下面我们来分析一下这段代码的具体逻辑，从哪看出是让rt线程"再试一次"的呢

        // 实际上，如果申请锁的人是rt的话，代码走到这里时，owner_state一定不是

        // OWNER_WRITER，只能是OWNER_NULL或者OWNER_READER。为啥捏，我们先假设

        // owner_state是OWNER_WRITER，看看会发生什么情况：此时临界区时writer，

        // 并且这个rt线程和这个writer是在同一个cpu上，那么这个rt线程一定会抢

        // 这个writer的cpu资源，导致这个writer的owner_on_cpu为false，所以在

        // 上面的rwsem_spin_on_owner中就会返回OWNER_NONSPINNABLE，所以代码根本

        // 不会走到这里！！！所以，对于申请锁的人如果是rt的话，代码走到这里owner_state

        // 一定是OWNER_NULL或者OWNER_READER

        // 代码走到这里，这个rt线程还是没抢到锁，有下面三个可能

        // a) owner_state为OWNER_NULL的第一种情况：当前持锁者正在释放锁，sem->owner

        // 已经被设置为NULL（注意：只有up_writer中才会调用rwsem_clear_owner

        // 接口将owner设置为NULL，reader退出临界区时并不会设置owner），导致上

        // 面rwsem_spin_on_owner中返回的是OWNER_NULL，但锁还没有完全被释放

        // （这里说的没有完全释放的意思是，在up_writer释放锁的时候，锁的

        // RWSEM_FLAG_HANDOFF标记并没有清除，所以下一个会将锁交给wait_list最

        // 前面的那个线程，到时我们在上面"偷锁"的逻辑中也获取不到锁）

        // b) owner_state为OWNER_NULL的另一种情况：虽然锁已经被完全释放了，sem->owner

        // 已经被设置为NULL，并且RWSEM_FLAG_HANDOFF标记也已经被清除了，但是因

        // 为当前在其他cpu上还可能存在其他线程也在自旋抢这把锁，因为上面"偷锁"

        // 逻辑本质上也是atomic操作，所以在上面"偷锁"逻辑中，很可能这个rt线程

        // 没有抢过其他cpu上的线程（即使其他cpu上的线程时cfs线程）

        // c) 而如果owner_state为OWNER_READER的话，则表示当前临界区是reader。那

        // 么如果这个rt线程是reader，（临界区的reader没有被撑爆，并且锁没有被置

        // 为RWSEM_FLAG_HANDOFF标记，后续的reader可以直接进入临界区），此时这个

        // rt线程是可以直接进入临界区的，此时代码不会走到这里，因为在上面"偷锁"

        // 的位置就拿到锁退出了，所以这种情况也是不可能的；如果这个rt是writer，

        // 则只能自旋。

        // prev_owner_state记录的是这个rt线程上一次自旋时临界区的状态，

        // owner_state记录的是本次自旋时临界区的状态。由上面分析可知，rt

        // 线程走到这里，"owner_state != OWNER_WRITER"一定成立，但是

        // "owner_state != OWNER_WRITER"成立并不表示这个申请锁的线程是rt

        // 如果一个rt线程连续两次走到这里，则表示其已经自旋了两次了，如果自旋

        // 了两次owner_state状态都没有变化，则停止rt的自旋

        if (owner_state != OWNER_WRITER) {

            //3.7.1 如果当前正在申请锁的线程马上要调度出去了，则返回false，停止自旋

            if (need_resched())

                break;

            //3.7.2 如图rt自旋两次，owner_state都没有发生变化，则停止自旋

            if (rt_task(current) &&

             (prev_owner_state != OWNER_WRITER))

                break;

        }

        prev_owner_state = owner_state;

        /*

         * The cpu_relax() call is a compiler barrier which forces

         * everything in this loop to be re-loaded. We don't need

         * memory barriers as we'll eventually observe the right

         * values at the cost of a few extra spins.

         */

        //3.8 原来这就是他妈的自旋？和spin没有半毛钱关系

        cpu_relax();

    }

    //4.标记自旋结束

    osq_unlock(&sem->osq);

done:

    preempt_enable();

    lockevent_cond_inc(rwsem_opt_fail, !taken);

    //5.乐观自旋拿到锁后返回true，否则返回false

    return taken;

}

#else

static inline bool rwsem_optimistic_spin(struct rw_semaphore *sem, bool wlock)

{

    return false;

}

#endif

/*

* Try to acquire read lock before the reader is put on wait queue.

* Lock acquisition isn't allowed if the rwsem is locked or a writer handoff

* is ongoing.

*/

static inline bool rwsem_try_read_lock_unqueued(struct rw_semaphore *sem)

{

    long count = atomic_long_read(&sem->count);

    //1.下面两种场景reader是"偷"不到锁的

    // a) 有一个writer在临界区

    // b) 锁已经被打上RWSEM_FLAG_HANDOFF标记，禁止"偷锁"

    if (count & (RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))

        return false;

    //2.代码走到这里，临界区可能没人或是reader，并且当前是允许"偷锁"的

    // rwsem允许多个reader同时进入临界区，count字段记录的当前有多少个

    // reader在临界区，下面对count执行加1标记拿锁。但是在这个原子操作

    // 的过程中这把锁还是有可能被其他cpu上的其他线程拿走的，所以这里要判断

    // 下面原子操作的返回值是加之前的值，所以这里需要注意，即使原子加完之后，

    // 这个reader也是有可能拿不到锁的，此时函数返回false

    count = atomic_long_fetch_add_acquire(RWSEM_READER_BIAS, &sem->count);

    if (!(count & (RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))) {

        rwsem_set_reader_owned(sem);

        lockevent_inc(rwsem_opt_rlock);

        return true;

    }

    /* Back out the change */

    //3.拿锁失败，说明在执行原子操作期间，存在下面两种情况导致拿锁失败，

    // 此时需要将count重新减回去

    // a) 原子操作期间，锁被另一个cpu上的一个writer给抢了

    // c) 原子操作期间，锁突然被打上RWSEM_FLAG_HANDOFF标记，禁止偷锁

    atomic_long_add(-RWSEM_READER_BIAS, &sem->count);

    return false;

}

/*

* Try to acquire write lock before the writer has been put on wait queue.

*/

static inline bool rwsem_try_write_lock_unqueued(struct rw_semaphore *sem)

{

    long count = atomic_long_read(&sem->count);

    //1.下面任一条件，writer在这里都会"偷锁"失败

    // a) 临界区有人（RWSEM_LOCK_MASK标记被置位），因为对于writer来说，

    // 只要临界区有人，writer都不能进入临界区

    // b) 锁已经被打上RWSEM_FLAG_HANDOFF标记，禁止"偷锁"

    while (!(count & (RWSEM_LOCK_MASK|RWSEM_FLAG_HANDOFF))) {

        //2.进入该循环表示当前临界区没人，并且当前还是允许"偷锁"的

        // writer拿锁就是通过原子操作将count的RWSEM_WRITER_LOCKED标记置位，

        // 当然，这个原子操作拿锁的过程，也有可能其他cpu上的线程拿走，下面原子

        // 操作返回true表示xchg操作真的成功，也就是说这个writer已经拿到锁了；

        // 如果原子操作返回false，则表示锁被其他cpu上的其他线程拿走了，此时继

        // 续while循环判断，如果真的是锁被别人拿走了，就返回false拿锁失败

        // 注意：每次原子操作，count变量都会被修改

        if (atomic_long_try_cmpxchg_acquire(&sem->count, &count,

                    count | RWSEM_WRITER_LOCKED)) {

            rwsem_set_owner(sem);

            lockevent_inc(rwsem_opt_wlock);

            return true;

        }

    }

    return false;

}

/*

* Calculate reader-owned rwsem spinning threshold for writer

*

* The more readers own the rwsem, the longer it will take for them to

* wind down and free the rwsem. So the empirical formula used to

* determine the actual spinning time limit here is:

*

* Spinning threshold = (10 + nr_readers/2)us

*

* The limit is capped to a maximum of 25us (30 readers). This is just

* a heuristic and is subjected to change in the future.

*/

static inline u64 rwsem_rspin_threshold(struct rw_semaphore *sem)

{

    long count = atomic_long_read(&sem->count);

    int readers = count >> RWSEM_READER_SHIFT;

    u64 delta;

    if (readers > 30)

        readers = 30;

    delta = (20 + readers) * NSEC_PER_USEC / 2;

    return sched_clock() + delta;

}

static noinline enum owner_state rwsem_spin_on_owner(

            struct rw_semaphore *sem,

            unsigned long nonspinnable)    //可为RWSEM_WR_NONSPINNABLE或RWSEM_RD_NONSPINNABLE

{

    struct task_struct *new, *owner;

    unsigned long flags, new_flags;

    enum owner_state state;

    //1.由rwsem_owner_state的实现可知，当返回的state不为OWNER_WRITER时，有下面几种可能

    // a) 没人在临界区，这时候后来的申请者不需要自旋就能拿到锁，退出小循环在大循环中尝试拿锁

    // b) reader在临界区，因为多个reader可以直接进入临界区，退出小循环在大循环中尝试拿锁

    // c) 锁被打上了nonspinable标记，显式打上了该标签，还自旋个毛啊

    owner = rwsem_owner_flags(sem, &flags);

    state = rwsem_owner_state(owner, flags, nonspinnable);

    if (state != OWNER_WRITER)

        return state;

    //2.代码都到这里说明这把锁是被writer持有，也就是说spin on owner一定是围着writer转

    // 下面这个循环实际是在等锁切换writer，当writer发生切换时退出这里的循环

    rcu_read_lock();

    for (;;) {

        /*

         * When a waiting writer set the handoff flag, it may spin

         * on the owner as well. Once that writer acquires the lock,

         * we can spin on it. So we don't need to quit even when the

         * handoff bit is set.

         */

        //3.由下面的逻辑可知，下面两种情况都会被认为是切换了writer，退出小循环

        // a) 持锁的task发生变化

        // b) 持锁的task没有变，但是flag发生了变化，例如nonspinable标记发生了变化

        // 注意: 这里只是返回，并不是已经拿到到锁了，是否需要继续自旋由新的持锁者决定

        new = rwsem_owner_flags(sem, &new_flags);

        if ((new != owner) || (new_flags != flags)) {

            state = rwsem_owner_state(new, new_flags, nonspinnable);

            break;

        }

        /*

         * Ensure we emit the owner->on_cpu, dereference _after_

         * checking sem->owner still matches owner, if that fails,

         * owner might point to free()d memory, if it still matches,

         * the rcu_read_lock() ensures the memory stays valid.

         */

        barrier();

        //4.代码走到这里说明持锁的writer还没有释放锁，状态也没变，

        // 但是满足下面条件，也不需要继续自旋了

        // a) 锁的申请者即将调度出去，（自己都要挂了，还自旋个毛啊）

        // b) 或者当前持锁的writer不是正在running，可能是处于runnable

        // 状态，或者是进入D状态了。真是舔狗的宿命...

        if (need_resched() || !owner_on_cpu(owner)) {

            state = OWNER_NONSPINNABLE;

            break;

        }

        //5.代码走到这里表示writer还没有从临界区出来，并且当前还在运行，那么就再等一下吧

        cpu_relax();

    }

    rcu_read_unlock();

    return state;

}

/*

* The rwsem_spin_on_owner() function returns the folowing 4 values

* depending on the lock owner state.

* OWNER_NULL : owner is currently NULL

* OWNER_WRITER: when owner changes and is a writer

* OWNER_READER: when owner changes and the new owner may be a reader.

* OWNER_NONSPINNABLE:

*         when optimistic spinning has to stop because either the

*         owner stops running, is unknown, or its timeslice has

*         been used up.

*/

enum owner_state {

    OWNER_NULL                = 1 << 0,            //没人在临界区

    OWNER_WRITER            = 1 << 1,            //写在临界区

    OWNER_READER            = 1 << 2,            //读在临界区

    OWNER_NONSPINNABLE        = 1 << 3,            //owner结束自旋

};

//由下面的注释可知，当临界区有reader, writer或者根本就没人在临界区的时候，都有可能使申请者自旋

#define OWNER_SPINNABLE        (OWNER_NULL | OWNER_WRITER | OWNER_READER)

static noinline enum owner_state rwsem_spin_on_owner(

            struct rw_semaphore *sem,

            unsigned long nonspinnable)

{

    struct task_struct *new, *owner;

    unsigned long flags, new_flags;

    enum owner_state state;

    owner = rwsem_owner_flags(sem, &flags);

    state = rwsem_owner_state(owner, flags, nonspinnable);

    if (state != OWNER_WRITER)

        return state;

    ...

}

static inline enum owner_state rwsem_owner_state(

            struct task_struct *owner,

            unsigned long flags,            //owner成员的低3bit的flags

            unsigned long nonspinnable)    //可为RWSEM_WR_NONSPINNABLE或RWSEM_RD_NONSPINNABLE

{

    //1.判断临界区当前是否允许reader或writer自旋

    // 传入的nonspinnable可为RWSEM_WR_NONSPINNABLE或RWSEM_RD_NONSPINNABLE

    // 该字段由申请锁的人传入，如果申请锁的人是reader，则传入RWSEM_RD_NONSPINNABLE，

    // 表示这个reader是想判断这个临界区当前是否允许读者自旋；如果申请锁的是人writer，

    // 则传入RWSEM_WR_NONSPINNABLE，表示这个writer是想判断这个临界区当前是否允许写者自旋

    if (flags & nonspinnable)

        return OWNER_NONSPINNABLE;

    //2.下面判断到底是谁在临界区，是reader还是writer，或者根本就没人

    // 但是这并不表示申请锁的人就是可以自旋的，而是表示需要进一步判断

    //2.1 reader在临界区

    if (flags & RWSEM_READER_OWNED)

        return OWNER_READER;

    //2.2 代码走到这里只有两种可能，这两种情况也是可能产生自旋的

    // a) 写持锁

    // b) 没人持锁

    return owner ? OWNER_WRITER : OWNER_NULL;

}

#define rwsem_first_waiter(sem) \

    list_first_entry(&sem->wait_list, struct rwsem_waiter, list)

/*

* handle the lock release when processes blocked on it that can now run

* - if we come here from up_xxxx(), then the RWSEM_FLAG_WAITERS bit must

* have been set.

* - there must be someone on the queue

* - the wait_lock must be held by the caller

* - tasks are marked for wakeup, the caller must later invoke wake_up_q()

* to actually wakeup the blocked task(s) and drop the reference count,

* preferably when the wait_lock is released

* - woken process blocks are discarded from the list after having task zeroed

* - writers are only marked woken if downgrading is false

*/

static void rwsem_mark_wake(

            struct rw_semaphore *sem,

            enum rwsem_wake_type wake_type,            //要唤醒什么类型的task

            struct wake_q_head *wake_q)                //将要唤醒的task先添加到一个临时链表上

{

    struct rwsem_waiter *waiter, *tmp;

    long oldcount, woken = 0, adjustment = 0;

    struct list_head wlist;

    lockdep_assert_held(&sem->wait_lock);

    /*

     * Take a peek at the queue head waiter such that we can determine

     * the wakeup(s) to perform.

     */

    //1.获取wait_list最前面的task，队列最前面的task是等待最久的，就唤醒他了

    waiter = rwsem_first_waiter(sem);

    //2.如果wait_list链表最前面是writer，则不会跳过这个writer去唤醒reader

    // 下面我们情景分析一下这段代码，满足下面if条件表示链表最前面的是writer:

    // a) 如果唤醒类型是RWSEM_WAKE_ANY，那么直接将这个writer唤醒

    // b) 如果唤醒类型不是RWSEM_WAKE_ANY，那么wake_type只能是RWSEM_WAKE_READERS

    // 或RWSEM_WAKE_READ_OWNED，也就是说本次一定是想唤醒reader，我们看到下面的

    // 代码逻辑就直接return了，并不会真的执行唤醒reader的动作，为啥呢？

    // 试想，如果真的去唤醒reader的话，那么这个writer就要再多等一会，所以为了防

    // 止writer饿死，这个时候不去唤醒任何人

    // RWSEM_WAKE_ANY参数应该这么理解：本次要唤醒谁取决于链表最前面线程

    // a) 如果链表最前面的是reader，那么本次会唤醒链表上最前面的256个reader一起进入临界区；

    // b) 如果链表最前面的是writer，那么本次只会唤醒链表最前面的这个writer进入临界区；

    // 这么设计的原因主要是防止writer饿死，详见__up_read中的描述

    // wake_type可取的值还可能是RWSEM_WAKE_READERS或RWSEM_WAKE_READ_OWNED

    // 这两个type都是表示要唤醒reader，但是背景不一样

    // RWSEM_WAKE_READERS : 唤醒链表上的reader

    // RWSEM_WAKE_READ_OWNED: 用于标记waker已经拿到了reader锁，我是来带领

    // wait_list链表上的其他reader一起进入临界区的

    if (waiter->type == RWSEM_WAITING_FOR_WRITE) {

        if (wake_type == RWSEM_WAKE_ANY) {

            /*

             * Mark writer at the front of the queue for wakeup.

             * Until the task is actually later awoken later by

             * the caller, other writers are able to steal it.

             * Readers, on the other hand, will block as they

             * will notice the queued writer.

             */

            wake_q_add(wake_q, waiter->task);

            lockevent_inc(rwsem_wake_writer);

        }

        return;

    }

    //3.代码走到这里，wait_list最前面一定是一个reader，唤醒时一定是从链表最前

    // 面开始向后依次唤醒，所以最前面的是reader的话，下一个一定是唤醒reader，

    // 下面的逻辑是唤醒reader

    /*

     * No reader wakeup if there are too many of them already.

     */

    //4.出现负数表示最高bit位RWSEM_FLAG_READFAIL已经被置一

    // 也就是说当前临界区里已经有很多reader了，此时就不要再新的reader了

    if (unlikely(atomic_long_read(&sem->count) < 0))

        return;

    /*

     * Writers might steal the lock before we grant it to the next reader.

     * We prefer to do the first reader grant before counting readers

     * so we can bail out early if a writer stole the lock.

     */

    //5.如果当前wake_type不是RWSEM_WAKE_READ_OWNED，则表示当前的waker

    // 还没有拿到reader锁，这里唤醒的reader是第一次进入临界区

    if (wake_type != RWSEM_WAKE_READ_OWNED) {

        struct task_struct *owner;

        //5.1 如果writer偷了reader的锁，并且最前面的reader阻塞超时，

        // 则设置HANDOFF标记，禁止偷锁

        // 在当前cpu上要唤醒链表上的reader去拿锁，同时在其他cpu上可能存在

        // 一个writer也在申请锁（通过自旋的方式），他们通过原子操作公平的竞争，

        // 谁先抢到谁拿锁，如果下面的if条件成立，则表示在另一个cpu上的那个

        // writer已经通过自旋的方式成功拿到锁了，当前cpu上的这个reader拿锁

        // 失败，这就是writer"偷"锁的一种场景！！！

        // 这样链表最前面的reader又要继续处于D状态等下去，但是如果这个reader

        // 处于D状态的时间已经很长了，再阻塞下去可能就会导致系统crash，这时就会

        // 强制设置HANDOFF标记，禁止"偷锁"

        adjustment = RWSEM_READER_BIAS;

        oldcount = atomic_long_fetch_add(adjustment, &sem->count);

        if (unlikely(oldcount & RWSEM_WRITER_MASK)) {

            /*

             * When we've been waiting "too" long (for writers

             * to give up the lock), request a HANDOFF to

             * force the issue.

             */

            //注意：

            //这里在设置RWSEM_FLAG_HANDOFF标记时有个小技巧，if条件里

            //面的adjustment是减去RWSEM_FLAG_HANDOFF，和后面原子操作

            //中的atomic_long_add(-adjustment, &sem->count)结合，就

            //是设置RWSEM_FLAG_HANDOFF位了

            if (!(oldcount & RWSEM_FLAG_HANDOFF) &&

             time_after(jiffies, waiter->timeout)) {

                adjustment -= RWSEM_FLAG_HANDOFF;

                lockevent_inc(rwsem_rlock_handoff);

            }

            atomic_long_add(-adjustment, &sem->count);

            return;

        }

        /*

         * Set it to reader-owned to give spinners an early

         * indication that readers now have the lock.

         * The reader nonspinnable bit seen at slowpath entry of

         * the reader is copied over.

         */

        //5.2 代码走到这里，表示这个wait_list链表最前面的reader已经成功拿到了锁

        // 下面就需要设置锁的owner，但是在设置owner之前，还有一个工作要做，

        // 就是继承last_rowner的nonspin标记，为啥捏？？？？？？？？？？

        owner = waiter->task;

        if (waiter->last_rowner & RWSEM_RD_NONSPINNABLE) {

            owner = (void *)((unsigned long)owner | RWSEM_RD_NONSPINNABLE);

            lockevent_inc(rwsem_opt_norspin);

        }

        __rwsem_set_reader_owned(sem, owner);

    }

    /*

     * Grant up to MAX_READERS_WAKEUP read locks to all the readers in the

     * queue. We know that the woken will be at least 1 as we accounted

     * for above. Note we increment the 'active part' of the count by the

     * number of readers before waking any processes up.

     *

     * This is an adaptation of the phase-fair R/W locks where at the

     * reader phase (first waiter is a reader), all readers are eligible

     * to acquire the lock at the same time irrespective of their order

     * in the queue. The writers acquire the lock according to their

     * order in the queue.

     *

     * We have to do wakeup in 2 passes to prevent the possibility that

     * the reader count may be decremented before it is incremented. It

     * is because the to-be-woken waiter may not have slept yet. So it

     * may see waiter->task got cleared, finish its critical section and

     * do an unlock before the reader count increment.

     *

     * 1) Collect the read-waiters in a separate list, count them and

     * fully increment the reader count in rwsem.

     * 2) For each waiters in the new list, clear waiter->task and

     * put them into wake_q to be woken up later.

     */

    //6.代码走到这里，一定是reader在临界区了，因为rwsem是允许多个reader

    // 同时进入临界区，所以下面需要唤醒wait_list链表上的其他reader，手牵

    // 手一起进入临界区，一次最多唤醒256个reader，所以对于reader，rwsem

    // 是一个相对公平的锁，因为wait_list链表上的reader不管顺序是怎么排序

    // 的，他们都可以一起进去临界区，同时获取锁，但是对于writer，是有顺序之

    // 分的，wait_list链表前面的writer肯定是比后面的writer先获取到锁

    //收集要唤醒的reader的工作必须分两步

    // a) 先将这些reader收集在一个单独的链表上，并完成计数，并将计数的值先加

    // 到sem->count中去，需要注意的是，这时候这些reader还没有真正的被唤醒

    // b) 对上面链表上收集到的每一个reader，清空waiter->task成员，并将这

    // 些task加到wake_q中去，以便稍后唤醒

    //为啥要分两步呢？上面的注释已经给出答案：

    // 防止rwsem->count成员的count字段，在增加之前被减少的可能，这是因为

    // 可能存在这一的情况：即将被唤醒的waiter可能根本就没有睡眠（参见

    // rwsem_down_read_slowpath的实现），所以在慢速路径中这个waiter可能会

    // 看到waiter->task成员被清空，于是就结束临界区，并且执行unlock操作，

    // 而在unlock中会将rwsem->count减1，但是这时候rwsem->count成员的

    // count字段还没有执行加操作，先减的话可能会出问题

    //6.1 第一步，先收集要唤醒的reader，最多不能超过256个，先将要唤醒的

    // reader添加到临时链表wlist上，等到第二步将其放到wake_q上

    //注意：这里在遍历链表时，跳过了writer，只唤醒reader

    INIT_LIST_HEAD(&wlist);

    list_for_each_entry_safe(waiter, tmp, &sem->wait_list, list) {

        if (waiter->type == RWSEM_WAITING_FOR_WRITE)

            continue;

        woken++;

        list_move_tail(&waiter->list, &wlist);

        /*

         * Limit # of readers that can be woken up per wakeup call.

         */

        if (woken >= MAX_READERS_WAKEUP)

            break;

    }

    //6.1.1 计算count要调整的值

    adjustment = woken * RWSEM_READER_BIAS - adjustment;

    lockevent_cond_inc(rwsem_wake_reader, woken);

    //6.1.2 如果此时wait_list为空的话，则取消RWSEM_FLAG_WAITERS标记，

    // 表示链表上已经没有waiter了

    if (list_empty(&sem->wait_list)) {

        /* hit end of list above */

        adjustment -= RWSEM_FLAG_WAITERS;

    }

    /*

     * When we've woken a reader, we no longer need to force writers

     * to give up the lock and we can clear HANDOFF.

     */

    //6.1.3 由上面的分析可知，上面设置HANDOFF是为了防止其他writer偷了链表

    // 最前面的reader的锁，从而临时禁止偷锁，现在链表最前面的reader已

    // 经得到唤醒，需要取消HANDOFF标记，重新允许自旋的线程偷锁

    if (woken && (atomic_long_read(&sem->count) & RWSEM_FLAG_HANDOFF))

        adjustment -= RWSEM_FLAG_HANDOFF;

    if (adjustment)

        atomic_long_add(adjustment, &sem->count);

    /* 2nd pass */

    //6.2 第二步：将临时链表wlist上的添加进唤醒队列wake_q中去，以便后面唤醒

    list_for_each_entry_safe(waiter, tmp, &wlist, list) {

        struct task_struct *tsk;

        tsk = waiter->task;

        get_task_struct(tsk);

        /*

         * Ensure calling get_task_struct() before setting the reader

         * waiter to nil such that rwsem_down_read_slowpath() cannot

         * race with do_exit() by always holding a reference count

         * to the task to wakeup.

         */

        //6.2 1 为啥要将waiter->task清空呢？

        // 这主要和rwsem_down_read_slowpath中的唤醒逻辑有关，在线程被

        // 唤醒时，会先判断这个waiter->task成员，如果该成员被清空，则表

        // 示其被真正的唤醒了，退出for循环，详见rwsem_down_read_slowpath实现

        smp_store_release(&waiter->task, NULL);

        /*

         * Ensure issuing the wakeup (either by us or someone else)

         * after setting the reader waiter to nil.

         */

        //6.2.2 确保在waiter->task设置为NULL后，再执行唤醒动作

        // 这里还不是唤醒，只是将其添加到队列中

        wake_q_add_safe(wake_q, tsk);

    }

}

enum rwsem_wake_type {

    RWSEM_WAKE_ANY,            /* Wake whatever's at head of wait list */

    RWSEM_WAKE_READERS,        /* Wake readers only */

    RWSEM_WAKE_READ_OWNED        /* Waker thread holds the read lock */

};

/*

* handle waking up a waiter on the semaphore

* - up_read/up_write has decremented the active part of count if we come here

*/

static struct rw_semaphore *rwsem_wake(struct rw_semaphore *sem, long count)

{

    unsigned long flags;

    DEFINE_WAKE_Q(wake_q);

    //1.先将要唤醒的线程放到临时链表wake_q上，然后一起唤醒

    raw_spin_lock_irqsave(&sem->wait_lock, flags);

    if (!list_empty(&sem->wait_list))

        rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);

    raw_spin_unlock_irqrestore(&sem->wait_lock, flags);

    //2.执行唤醒的动作

    wake_up_q(&wake_q);

    return sem;

}

void __sched down_read(struct rw_semaphore *sem)

{

    //1.增加抢占点，可能会进入睡眠

    // 打开CONFIG_PREEMPT_VOLUNTARY宏表示自愿抢占，是内核的一种抢占模型，

    // 打开该宏后，might_sleep、might_resched这类接口才会有实现，否则就是

    // 空的，在内核中会有多个模块调用这两个函数，例如mutex、信号量、驱动等。

    // 这两个函数要完成的工作实际就是判断current的TIF_NEED_RESCHED标记是

    // 否被打上，也就是说判断是否要自愿让出cpu，如果被打上的话，就主动调用

    // schedule函数调度其他线程运行，通过这样的方式，相当于在代码中会增加了

    // 抢占点，这里先简单了解一下，后面在调度章节详细分析

    might_sleep();

    //2.只有在使能CONFIG_DEBUG_LOCK_ALLOC时才有效

    rwsem_acquire_read(&sem->dep_map, 0, 0, _RET_IP_);

    //3.这里才是真正的持锁

    LOCK_CONTENDED(sem, __down_read_trylock, __down_read);

}

#ifdef CONFIG_LOCK_STAT

#define LOCK_CONTENDED(_lock, try, lock)            \

do {                                \

    if (!try(_lock)) {                    \

        lock_contended(&(_lock)->dep_map, _RET_IP_);    \

        lock(_lock);                    \

    }                            \

    lock_acquired(&(_lock)->dep_map, _RET_IP_);            \

} while (0)

#else

#define LOCK_CONTENDED(_lock, try, lock) \

    lock(_lock)

#endif

static inline int __down_read_trylock(struct rw_semaphore *sem)

{

    long tmp;

    //1.调试相关，暂时不关注

    DEBUG_RWSEMS_WARN_ON(sem->magic != sem, sem);

    /*

     * Optimize for the case when the rwsem is not locked at all.

     */

    //2.尝试直接拿锁，（所谓拿到锁，就是成功对这个原子变量执行了加1操作）

    // reader拿到锁时完成下面两个工作

    // a) 将sem->count的count字段加1

    // b) 将sem->owner设置为current

    // 需要注意的是，对于reader来说，owner记录的是最后一个进入临界区的reader

    // 但是并不表示owner指向的task一定就在临界区，详见rwsem_clear_reader_owned

    // 首先来看一下atomic_try_cmpxchg_acquire原子操作：

    // 如果第一个参数和第二个参数相等，则将第三个参数赋值给第一个参数，并返回true，

    // 否则不执行赋值操作，并返回false，关键字try用于标记是否完成和chg操作

    // 注意：

    // 不管是否交换成功，每次调用atomic_try_cmpxchg_acquire操作后，tmp

    // 的值都会被修改，tmp会被改为原子变量的旧值，结合这里的do-while表示:

    // 只要count没有被打上RWSEM_READ_FAILED_MASK标记，这个reader就

    // 会一直循环尝试修改这个原子变量的值，直到成功

    // RWSEM_READ_FAILED_MASK定义如下：

    // (RWSEM_WRITER_MASK|RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF|RWSEM_FLAG_READFAIL)，

    // 这表示满足下面任意一个条件，reader就会退出循环，快速持锁路径都会失败：

    // a) 如果有writer已经进入临界区

    // b) wait_list上有人（可能是reader或writer），即有人阻塞在这把锁上

    // c) 锁已经被打上HANDOFF标记

    // d) 临界区的reader已经撑爆了

    // 换句话说，只有同时满足下面条件，reader才能直接拿到锁

    // a) 当前在reader在临界区

    // b) 并且wait_list上没有阻塞的线程时

    // c) 锁没有被打上HANDOFF标记，即允许偷锁

    // （严格意义上来讲，这里不应该叫偷锁，应该叫插队拿锁）

    // 还有一个问题，那么这里为什么要不停的循环尝试呢？

    // 这主要是防止多个cpu同时修改这个count值撞到一块去了，也就是多个cpu

    // 上在同一时刻恰好都有reader需要进入临界区时，这时候多个cpu会同时来修

    // 改这个count的值，一个cpu在写count时，另一个cpu就必须等着

    // 返回值：直接拿锁成功返回1，失败返回0

    tmp = RWSEM_UNLOCKED_VALUE;

    do {

        if (atomic_long_try_cmpxchg_acquire(&sem->count, &tmp,

                    tmp + RWSEM_READER_BIAS)) {

            rwsem_set_reader_owned(sem);

            return 1;

        }

    } while (!(tmp & RWSEM_READ_FAILED_MASK));

    return 0;

}

static inline void rwsem_set_reader_owned(struct rw_semaphore *sem)

{

    //将当前线程current标记为读的owner

    __rwsem_set_reader_owned(sem, current);

}

/*

* The task_struct pointer of the last owning reader will be left in

* the owner field.

*

* Note that the owner value just indicates the task has owned the rwsem

* previously, it may not be the real owner or one of the real owners

* anymore when that field is examined, so take it with a grain of salt.

*

* The reader non-spinnable bit is preserved.

*/

static inline void __rwsem_set_reader_owned(

            struct rw_semaphore *sem,

            struct task_struct *owner)

{

    //1.在设置read_owner时，除了设置task_struct之外，还完成下面3个工作

    // a) 标记RWSEM_READER_OWNED，这个很好理解，因为现在是reader持锁

    // b) 保留owner中原来的RWSEM_RD_NONSPINNABLE位的值

    // c) 需要特别注意的是，此处还清除了RWSEM_WR_NONSPINNABLE位被清零，

    // 也就是说，当reader在临界区时，我们认为reader很快就会退出临界

    // 区，writer应该再自旋等一下，不用直接进D状态

    // 需要注意的是，只要有reader进入临界区，都会清一次RWSEM_RD_NONSPINNABLE

    // 标记位，也就是说后来的writer都会自旋

    unsigned long val = (unsigned long)owner | RWSEM_READER_OWNED |

        (atomic_long_read(&sem->owner) & RWSEM_RD_NONSPINNABLE);

    atomic_long_set(&sem->owner, val);

}

static inline void __down_read(struct rw_semaphore *sem)

{

    //背景：代码走到__down_read这里，一定表示前面在调用

    // __down_read_trylock时失败了，也就是说此时可能

    // a) 临界区里是writer

    // b) 或者临界区里是reader，但是wait_list上有线程阻塞着

    //1.首先是尝试走快速路径持锁，拿到锁后就直接设置owner了

    if (!rwsem_read_trylock(sem)) {

        //2.上面快速路径拿锁失败，不得不走到这里执行慢速路径

        rwsem_down_read_slowpath(sem, TASK_UNINTERRUPTIBLE);

        DEBUG_RWSEMS_WARN_ON(!is_rwsem_reader_owned(sem), sem);

    } else {

        //3.快速路径持锁成功后设置owner

        // 对于reader，owner记录的是最后一次进入临界区的reader

        rwsem_set_reader_owned(sem);

    }

}

static inline bool rwsem_read_trylock(struct rw_semaphore *sem)

{

    //1.原子操作返回的值是加完1之后的值，注意这里的逻辑

    // 不管这个reader是否能够拿到锁，也不管wait_list链表上是否有线程在等着，

    // 上来就先将rwsem->count成员的count字段加1，要知道，count字段中记录的

    // 是在临界区的reader的个数啊，你这个reader还不一定能成功拿到锁呢，凭啥上

    // 来就加1呢？这么自信吗？

    // 实际上如果这里持锁失败，则会进入慢速路径，在慢速路径中会将这个1减去的

    // 那么为啥要提前加呢？难道是起到了提前预警的作用？

    long cnt = atomic_long_add_return_acquire(RWSEM_READER_BIAS, &sem->count);

    //2.这里判断小于0实际是判断count的最高bit位RWSEM_FLAG_READFAIL

    // RWSEM_FLAG_READFAIL位被置一表示已经有很多reader在临界区了

    // 这时候设置nospinable标记，也就是后续不管是reader还是writer

    // 都不需要再自旋了，直接进阻塞状态，因为服务完这么多reader需要很长时间

    if (WARN_ON_ONCE(cnt < 0))

        rwsem_set_nonspinnable(sem);

    //3.RWSEM_READ_FAILED_MASK的定义如下

    // (RWSEM_WRITER_MASK|RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF|RWSEM_FLAG_READFAIL)

    // 这表示：

    // a) 如果临界区的是writer，后来的reader快速持锁失败

    // b) 如果wait_list上有等待的线程，则后来的reader也会持锁失败

    // c) 如果锁已经被打上HANDOFF标记，则后来的reader持锁失败

    // d) 如果临界区的reader已经特别多了，后来的reader也会持锁失败

    // 也就是说，只有在当临界区全是reader，并且wait_list上没有阻塞的线程时，

    // 后来的reader才可以直接拿到锁进入临界区

    return !(cnt & RWSEM_READ_FAILED_MASK);

}

/*

* Set the RWSEM_NONSPINNABLE bits if the RWSEM_READER_OWNED flag

* remains set. Otherwise, the operation will be aborted.

*/

static inline void rwsem_set_nonspinnable(struct rw_semaphore *sem)

{

    unsigned long owner = atomic_long_read(&sem->owner);

    do {

        //1.下面的条件表示：

        // a) 如果是reader在临界区时，继续向下，一定会设置RWSEM_NONSPINNABLE，

        // 此时后来的reader或者writer都不允许自旋

        // b) 如果是writer或者没人在临界区时，直接返回，RWSEM_NONSPINNABLE标记

        // 保持原来的状态，此时后来的reader或者writer允许自旋，为啥呀？

        // 该函数的调用时机，都是reader在临界区，根本的目的是禁止reader通过

        // 乐观自旋的方式源源不断的进入临界区，导致reader的临界区越来越大，这

        // 也是防止writer饿死的一种机制

        if (!(owner & RWSEM_READER_OWNED))

            break;

        if (owner & RWSEM_NONSPINNABLE)

            break;

    //2.代码走到这里，表示reader在临界区，并且还没有设置

    // RWSEM_NONSPINNABLE标记，下面就需要设置该标记位

    // atomic_long_try_cmpxchg有三个参数，如果第一个参数等于第二个参数，

    // 则执行xchg操作将第三个参数的值赋值给第一个参数，并返回true；不相

    // 等则不执行xchg操作，也就不执行赋值操作，并返回false；并且每次原子

    // 操作后，不管是否成功，owner每次都会被更改为最新读取到的sem->owner

    // 的值，函数返回标记是否已经成功执行了xchg操作；

    // 所以，这里的循环的意思是：只有真正的执行了xchg操作时，atomic_long_try_cmpxchg

    // 才会返回true，表示已经成功将RWSEM_NONSPINNABLE设置上，此时退出循环；

    // 如果在执行原子变量期间owner成员发生变化（可能值持锁者发生变化，也可能

    // 是标记发生变化），atomic_long_try_cmpxchg没有真正的执行xchg操作，

    // 此时返回false，继续循环操作，直到成功设置RWSEM_NONSPINNABLE标记为止

    } while (!atomic_long_try_cmpxchg(&sem->owner, &owner, owner | RWSEM_NONSPINNABLE));

}

/*

* Wait for the read lock to be granted

*/

static struct rw_semaphore __sched * rwsem_down_read_slowpath(

            struct rw_semaphore *sem,

            int state)                        //持锁失败时线程进入的状态

{

    //1.adjustment的初值为-1，表示对count字段减1

    // 因为在调用rwsem_down_read_slowpath之前一定已经调用过rwsem_read_trylock

    // 由前面的分析可知，在rwsem_read_trylock中不管三七二十一直接对count加1了

    // 走到这里肯定是没拿到锁，需要减掉了，但是为啥要提前加呢？？？？

    long count, adjustment = -RWSEM_READER_BIAS;

    struct rwsem_waiter waiter;

    //2.定义一个等待队列，批量唤醒，这个在后面分析

    DEFINE_WAKE_Q(wake_q);

    bool wake = false;

    /*

     * Save the current read-owner of rwsem, if available, and the

     * reader nonspinnable bit.

     */

    //3.备份当前rwsem的owner，owner的值可为：

    // a) 如果是读持锁，只能记录最后一个进入临界区的reader，但是需要注意的是，这个

    // owner指向的task当前并不一定在临界区哦，详见rwsem_clear_reader_owned实现

    // b) 如果是写持锁，因为临界区只允许一个writer，owner记录的是那个唯一的writer

    // 如果当前是writer在临界区，则只保留RWSEM_RD_NONSPINNABLE标记，为啥捏？？？？

    // 原因就是方便在后面rwsem_reader_phase_trylock中判断锁的持有者是否被更换了。

    // 这里主要是考虑到了这种情况：同一个task，可能一会是reader，而在另一个时刻却

    // 摇身一变成了writer，过一会又会在变成reader，这种情况下Linux也会被看做是不

    // 同的owner持了锁，但是这种情况下owner的task_struct字段却又是一样的，这样在

    // rwsem_reader_phase_trylock中通过(owner ^ last_rowner)的方式判断是否切换过

    // owner就会失败，所以这里如果是writer在临界区，会将task_struct字段清空

    // 题外话，不得不吐槽一下这里的代码逻辑，真是太乱了，怎么审核过的！！！

    waiter.last_rowner = atomic_long_read(&sem->owner);

    if (!(waiter.last_rowner & RWSEM_READER_OWNED))

        waiter.last_rowner &= RWSEM_RD_NONSPINNABLE;

    //4.判断这个reader是否满足乐观自旋的条件，下面情况不满足乐观自旋的条件，

    // 详见rwsem_can_spin_on_owner实现