内核中读写信号量（rwsem）实现的关键注释

Posted 2021-05-21 程序猿Ricky

本文代码摘自于linux-3.10.0-327.el7版本内核：

读锁实现注释：
读锁在获取时主入口时down_read，在这个函数中会先尝试__down_read_trylock，如果__down_read_trylock失败，就会执行__down_read，在这个函数中实际运行的是慢速路径，它的逻辑实现在如下：

struct rw_semaphore __sched *rwsem_down_read_failed(struct rw_semaphore *sem)
{
    //1.默认是减去这个值，在运行到这里之前，会先try，在try时增加了read标记，但是获取锁失败了，所以要undo这个操作
    long count, adjustment = -RWSEM_ACTIVE_READ_BIAS;
    struct rwsem_waiter waiter;
    struct task_struct *tsk = current;

    /* set up my own style of waitqueue */
    waiter.task = tsk;
    waiter.type = RWSEM_WAITING_FOR_READ;
    get_task_struct(tsk);

    raw_spin_lock_irq(&sem->wait_lock);
    //2.如果list是空的，那么我们将要添加一个waiter，因此需要添加wait flag到count中.
    if (list_empty(&sem->wait_list))
        adjustment += RWSEM_WAITING_BIAS;
    list_add_tail(&waiter.list, &sem->wait_list);

    /* we're now waiting on the lock, but no longer actively locking */

    //3.此时我们已经添加到了wait_list，因此需要把前面获取到的adjustment更新到count标记
    count = rwsem_atomic_update(adjustment, sem);

    /* If there are no active locks, wake the front queued process(es).
     *
     * If there are no writers and we are first in the queue,
     * wake our own waiter to join the existing active readers !
     */
    //4.唤醒wait_list中前面的进程，可能为一个写者，或者多个读者(可能包含自己)。
    //如果唤醒本进程，会把对应的读waiter中的task设置为NULL
    if (count == RWSEM_WAITING_BIAS ||
        (count > RWSEM_WAITING_BIAS &&
         adjustment != -RWSEM_ACTIVE_READ_BIAS))
        sem = __rwsem_do_wake(sem, RWSEM_WAKE_ANY);

    raw_spin_unlock_irq(&sem->wait_lock);

    /* wait to be given the lock */
    while (true) {
        set_task_state(tsk, TASK_UNINTERRUPTIBLE);
        //判断task为NULL，说明不用继续等待了，可以持锁了.
        if (!waiter.task)
            break;
        schedule();
    }   

    tsk->state = TASK_RUNNING;

    return sem;
}

写锁实现注释：
写锁在获取时主入口时down_write，在这个函数中会先尝试__down_write_trylock，如果__down_write_trylock失败，就会执行__down_write，在这个函数中实际运行的是慢速路径，它的逻辑实现在如下：

struct rw_semaphore __sched *rwsem_down_write_failed(struct rw_semaphore *sem)
{
    //1.默认是减去这个值，在try时增加了write标记，但是获取锁失败了，所以要undo这个操作
    long count, adjustment = -RWSEM_ACTIVE_WRITE_BIAS;
    struct rwsem_waiter waiter;
    struct task_struct *tsk = current;

    /* set up my own style of waitqueue */
    waiter.task = tsk;
    waiter.type = RWSEM_WAITING_FOR_WRITE;

    raw_spin_lock_irq(&sem->wait_lock);
    //2.如果list是空的，那么我们将要添加一个waiter，因此需要添加wait flag到count中.
    if (list_empty(&sem->wait_list))
        adjustment += RWSEM_WAITING_BIAS;
    list_add_tail(&waiter.list, &sem->wait_list);

    /* we're now waiting on the lock, but no longer actively locking */

    //3.此时我们已经添加到了wait_list，因此需要把前面获取到的adjustment更新到count标记
    count = rwsem_atomic_update(adjustment, sem);

    /* If there were already threads queued before us and there are no
     * active writers, the lock must be read owned; so we try to wake
     * any read locks that were queued ahead of us. */
    //判断如果wait_list中已经存在等待的读者时，
    //为了加快读者执行完成临界区执行，需要先把他们唤醒
    //这样当前写者才能更快的获取到写锁
    if (count > RWSEM_WAITING_BIAS &&
        adjustment == -RWSEM_ACTIVE_WRITE_BIAS)
        sem = __rwsem_do_wake(sem, RWSEM_WAKE_READERS);

    /* wait until we successfully acquire the lock */
    set_task_state(tsk, TASK_UNINTERRUPTIBLE);
    while (true) {
        if (!(count & RWSEM_ACTIVE_MASK)) {
            /* Try acquiring the write lock. */
            //4.没有active时，尝试获取锁，添加write active标记
            count = RWSEM_ACTIVE_WRITE_BIAS;
            if (!list_is_singular(&sem->wait_list))
                count += RWSEM_WAITING_BIAS;

            if (sem->count == RWSEM_WAITING_BIAS &&
                cmpxchg(&sem->count, RWSEM_WAITING_BIAS, count) ==
                            RWSEM_WAITING_BIAS)
                break;
        }

        raw_spin_unlock_irq(&sem->wait_lock);

        /* Block until there are no active lockers. */
        do {
            schedule();
            set_task_state(tsk, TASK_UNINTERRUPTIBLE);
        } while ((count = sem->count) & RWSEM_ACTIVE_MASK);

        raw_spin_lock_irq(&sem->wait_lock);
    }

    list_del(&waiter.list);  
    raw_spin_unlock_irq(&sem->wait_lock);
    tsk->state = TASK_RUNNING;

    return sem;
}

读写锁的实现总结：

• 当读者持有锁时，且没有写者等待时，后续的读者可以直接获取锁成功；
• 当有一个写者等待时，后续的写者和读者都不能获取锁；
• 当写者持有锁时，后续的写者和读者都不能获取锁；
• 等待wait_list上，第一个都是写者；
• 当写者释放锁时，等待wait_list上的第一个将会获取锁（不管其是读者还是写者）；

---

参考：http://linux.laoqinren.net/kernel/rw-semaphore/