一种直观简单的解决方案是用一个计数器 nr_readers 记录当前同时持有锁的数目,读者获得锁前原子递增 nr_readers;释放锁的时候原子递减,如果递减后 nr_readers 为 0,表示自己此刻是最后一个持有锁的读者。此外还需要一个记录等待线程中的第一个写者的指针 next_writer。这两个变量也应当放到锁的数据结构中。
读者并发还带来一个棘手的问题。还是以图 2 为例,当读者 A1获得锁时,如果此时知道 A2已经到来了(通过检查自己的 next 指针),那么应该由 A1通知 A2。但是可能 A2到来的时候 A1已经在访问共享资源或者正在释放锁,那么 A2应该自行获得锁。因为获得锁涉及到原子递增 nr_readers,所以 A1和 A2必须达成一致。我们的方案是:读者到来时先检查其直接前驱读者是否还在自旋,若是,则原子地在其前驱的节点数据结构中标记一个特殊值。如果标记成功,表明前驱尚未结束自旋,那么由前驱负责通知;如果标记失败,说明前驱已经退出自旋而获得锁,那么读者直接获得锁。标记值可以选得有意义一些,这儿我们选择线程的角色,于是线程很容易知道直接后继的类型,而不用通过 next 指针进一步查看。我们在节点数据结构中增加变量 successor_type。
下面我们详细阐述读写自旋锁的算法。假设申请线程为 A。
A 是读者,申请锁:
A 使用原子交换操作将读写自旋锁的 tail 指针指向自己的 qnode 结构以确定在链表中的位置,并返回原来的 tail 值作为自己的直接前驱 pred。即使多个线程同时申请锁,由于交换操作的原子性,每个执行线程的申请顺序将会被唯一确定,不会出现不一致的现象。
如果 pred 等于 NULL,说明锁无人持有或只有读者持有(A 的直接前驱是读者,且已经释放锁),那么 A 可以立即持有锁,跳到第 4 步。
此时 pred 不等于 NULL。如果 pred 是写者,A 只能依赖 pred 在释放锁的时候通知自己,所以 A 所要做的只是将 pred->next 指向自己的,然后自旋等待。如果 pred 是读者,那么需要按照前面描述的那样,向 pred 标记 A 的读者角色。如果标记成功,则取得锁,跳到第 4 步;否则自旋等待。
A 准备结束函数调用,但还需要检查一下自己的 successor_type 变量,如果直接后继是读者,那么先等它将链表构建完整,然后通知其结束自旋。
A 有直接后继 B,先需要等待 B 将链表构建完整,然后检查 B 是不是写者,若是,则将锁结构中的 next_writer 指针指向 B。
A 判断自己是否是最后一个持有者,若是,且 next_writer 指针不为 NULL,则应该通知 next_writer。具体做法是:A 原子递减 nr_readers,如果 nr_readers 新值大于 0,说明还有读者持有锁,于是 A 直接拍拍屁股走人。如果 nr_readers 新值等于 0,这并不能说明 A 在后面通知 next_writer 时还是最后一个持有者,因为在 A 执行后续操作的间隙中可能依次来了新的读者 C 和 D。假定读者 D 以极快的速度获得并执行释放锁,将 tail 指针置为 NULL。此刻又来了写者 W,因为 W 发现 tail 为 NULL,那么 W 必须将 next_writer 置为自己,否则 A 或 D 无法通知 W。现在轮到 A 继续执行了,A 发现 next_writer 不为 NULL,于是试图通知 W 结束自旋,但是 C 仍然持有锁,导致错误。这个例子说明通知 next_writer 前还需要再检查一下 nr_readers 的值,如果 nr_reader 仍然等于 0,才能说明是最后一个持有锁的读者。如果 C 也迅速释放锁,那么 A 和 C 可能同时试图通知 W。通知一个已经退出自旋的线程是危险的,因为该线程可能重新申请锁,而导致提前退出自旋。因此 A 通知 W 时,需要检测 next_writer 指针是否没有变化并原子地将其置为 NULL,这样 C 会发现已经有人通知过 W 了。
A 是写者,申请锁:
A 使用原子交换操作将读写自旋锁的 tail 指针指向自己的 qnode 结构以确定在链表中的位置,并返回原来的 tail 值作为自己的直接前驱 pred。
如果 pred 等于 NULL,说明锁无人持有或仍有读者持有。A 先将 next_writer 指向自己,然后检查 nr_readers,如果 nr_readers 大于 0,说明尚有读者持有锁,那么自旋等待。如果 nr_readers 等于 0,也有可能某个(些)读者正在释放锁,因此 A 需要检测 next_writer 指针是否仍然指向自己并原子地将其置为 NULL。如果成功置为 NULL,获得锁并返回;否则等待前面的读者通知自己。
如果 pred 不等于 NULL,那么 A 标记自己的角色并将链表构建好,然后自旋等待。
A 是写者,释放锁:
A 先检查是否存在直接后继。这可以用原子比较 tail 指针是否还是指向自己并更新为 NULL 的操作实现。如果成功地将 tail 指针更新为 NULL,说明没有直接后继,那么直接返回。
A 有直接后继 B,先需要等待 B 将链表构建完整,然后通过 next 指针通知 B 结束自旋。如果 B 是读者,还需要先原子递增 nr_readers。