Linux 内核自旋锁spinlock（三）--- MCS locks-EW帮帮网

文章目录

前言
一、MCS locks
- 1.1 原理简介
- 1.2 流程示例
二、源码说明
- 2.1 mcs_spin_lock
- 2.2 mcs_spin_unlock
三、MCS locks and qspinlocks
参考资料

前言

（1）老版本（2.6.25之前）的Linux内核的自旋锁请参考：Linux 内核自旋锁spinlock（一）
其缺点：该自旋锁是无序竞争的，不保证先申请的进程先获得锁，不公平。也就是所有的thread都是在无序的争抢spin lock，谁先抢到谁先得，不管thread等了很久还是刚刚开始spin。在冲突比较少的情况下，不公平不会体现的特别明显，然而，随着硬件的发展，多核处理器的数目越来越多，多核之间的冲突越来越剧烈，无序竞争的spinlock带来性能问题。

（2）2.6.25以后的Linux内核的自旋锁称为 ticket spinlock，基于 FIFO 算法的排队自选锁。请参考：Linux 内核自旋锁spinlock（二）— ticket spinlock
ticket spinlock可以让CPU按照到达的先后顺序，去获取spinlock的所有权，形成了有序竞争。
其缺点：所有申请锁的处理器在同一个变量上自旋等待，缓存同步的开销大，不适合处理器很多的系统。所有等待同一个自旋锁的处理器在同一个变量上自旋等待，申请或者释放锁的时候会修改锁，导致其他处理器存放自旋锁的缓存行失效，在拥有几百甚至几千个处理器的大型系统中，处理器申请自旋锁时竞争可能很激烈，缓存同步的开销很大，导致系统性能大幅度下降。

（3）基于ticket spinlock存在cache-line bouncing的问题，内核开发者提出MCS锁机制，MCS锁机制让每个CPU不再是等待同一个spinlock变量，而是基于各自不同的per-CPU的变量进行等待，那么每个CPU平时只需要查询自己对应的这个变量所在的本地cache line，仅在这个变量发生变化的时候，才需要读取内存和刷新这条cache line。
MCS锁机制会导致spinlock结构体变大（相比于ticket spinlock多了一个指针，64位平台指针大小8个字节），而内核很大数据结构都内嵌了spinlock结构体，比如struct page，因此MCS锁机制没有合入内核主线。

（4）4.2.0以后的Linux内核的自旋锁称为 queued spinlock，基于 MCS 算法的排队自选锁：
queued spinlock也属于排队自选锁，但是没有增加spinlock结构体的大小，进程按照申请锁的顺序排队，先申请的进程先获得锁。MCS自旋锁的策略是为每个处理器创建一个变量副本，每个处理器在自己的本地变量上自旋等待，解决了性能问题。

本文主要介绍MCS锁机制。

一、MCS locks

MCS（MCS是“Mellor-Crummey”和“Scott”这两个发明人的名字的首字母缩写）自旋锁解决了ticket spinlock的缺点，它的策略是为每个处理器创建一个变量副本，每个处理器在申请自旋锁的时候在自己的本地变量上自旋等待，避免缓存同步的开销。

它通过将锁的控制信息结构化为每CPU数据结构的方式，来减少缓存行的竞争和反弹。每个CPU都有自己的MCS节点，当一个CPU想要获取MCS锁时，它会将自己的MCS节点添加到队列的末尾，将等待锁的 CPU 组织成一个单向链表，每个 CPU 只需在自己的节点上自旋，每个 CPU 自旋在自己的本地变量上，减少对全局锁变量的访问，降低缓存行争用。然后等待前一个CPU通知它。前一个CPU负责通知后一个CPU，不需要对共享变量进行原子操作，因此减少了缓存行竞争。

MCS通过将自旋锁扩展为每CPU结构，能够在很大程度上减少简单锁所经历的cache-line bouncing，特别是在锁争用情况下。

在这里插入图片描述

1.1 原理简介

（1）mcs_spinlock 结构体

struct mcs_spinlock {
	struct mcs_spinlock *next;
	int locked; /* 1 if lock acquired */
	int count;  /* nesting count, see qspinlock.c */
};

如下图所示：
在这里插入图片描述
（2）当CPU0希望获取这个锁时，它会提供一个属于自己的mcs_spinlock结构。通过无条件的原子交换xchg操作，它将自己结构体的地址存储在锁的next字段中，并标记该锁为已被占用。如下图所示：

加入队列：
使用原子交换xchg将本地节点的地址写入全局锁的 next 字段。
如果返回的 next 为 NULL，表示锁未被占用，当前 CPU 成功获取锁。
如果返回的 next 非空，表示锁已被占用，当前 CPU 需要加入等待队列。

当处理器0申请自旋锁的时候，执行原子交换操作，使next 指向处理器0的mcs_spinlock结构体，并且返回next 的旧值。next 的旧值是空指针，说明自旋锁处于空闲状态，那么处理器0获得自旋锁。

这时候不需要设置node->locked为1，因为其他线程在等待自己的node->locked，当前线程不需要自旋，直接返回。

注意MCS locks获取到锁时，不需要将自己的mcs_spinlock结构体的locked值设置为1，0和1都无所谓。MCS locks释放锁的时候，如果next不为空，即有下一个CPU在等待，那么将下一个CPU的mcs_spinlock结构体的locked值设置为1，下一个CPU在busy wait发现自己的mcs_spinlock结构体的locked值由0变为1，那么获取到锁，下面会解释。

（3）在这种情况下，原子交换操作会返回next指针的先前值。由于该指针此前为null，获取锁的CPU会知道自己成功地获取了锁。一旦锁被占用，但没有争用。如果第二个CPU尝试获取这个锁，它将以相同的方式开始，将指向其mcs_spinlock结构的指针存储在主锁的next指针中。如下图所示：
在这里插入图片描述
CPU1调用mcs_spin_lock尝试获取锁，此时CPU0还没有释放锁，CPU1的prev不是NULL，说明锁已经被其他线程持有。这时候要把当前节点链接到prev的next，也就是等待队列的尾部。CPU1调用arch_mcs_spin_lock_contended这个函数，参数是&node->locked，在自旋等待，直到locked的值发生变化，由0变为1。

对于CPU1来说，其prev不为空，说明锁已经被其他线程持有，CPU1需要将自己添加到等待队列，并等待自己的locked变量值变为1被解锁。这里的等待队列是通过每个节点的next指针形成的链表，这样每个线程只需要在自己的locked变量上自旋，减少缓存行的竞争。

（4）当CPU1执行对主锁的原子交换操作时，它也会获得next字段的先前内容 — 即指向CPU 0的mcs_spinlock结构的指针。非NULL值告诉CPU1该锁不可用，而特定的指针值告诉CPU1谁在排队等待获取锁。CPU1会根据这种情况响应，将指向其mcs_spinlock结构的指针存储在CPU 0的结构体的next字段中。如下图所示：
在这里插入图片描述

CPU0调用 mcs_spin_unlock函数释放自旋锁，发现自己的mcs_spinlock结构体的成员next不是空指针，说明有申请者正在等待锁，于是把下一个节点的成员locked设置为1，CPU1获得自旋锁。

CPU0调用 mcs_spin_unlock函数在解锁时，CPU0设置next->locked = 1，即设置CPU1的locked值为1，CPU1调用arch_mcs_spin_lock_contended函数，在自旋等待，发现locked的值发生变化，由0变为1，CPU1退出循环，获得锁。

（5）一旦这个赋值完成，CPU 1将在其自己的mcs_spinlock结构中旋转锁定的值，而不是在主锁中的值。因此，它的自旋完全是CPU本地的，根本不会触及主锁。随着对锁的争用增加，这个过程可以无限期地进行下去，每个CPU都将自己排在已经在那里的CPU后面，每个CPU都在自己的锁的副本上旋转。因此，“主”锁中的指针始终表示等待CPU队列的尾部。

当CPU 0最终完成对锁的操作时，它将在主锁上执行比较并交换操作，尝试将next指针设置为NULL，假设该指针仍指向自己的结构体。如果该操作成功，说明锁从未被争用，工作已完成。然而，如果其他CPU像上面所述更改了该指针，比较并交换操作将失败。在这种情况下，CPU 0将完全不会更改主锁；相反，它将更改CPU 1结构中的锁定值，并从该情况中移除自己。
在这里插入图片描述

CPU1调用mcs_spin_unlock释放自旋锁，发现自己的mcs_spinlock结构体的成员next是空指针，说明自己是最后一个申请者，于是执行原子比较交换操作：如果next指向自己的mcs_spinlock结构体，那么把next设置为空指针。

在解锁的整个过程中，持有锁的这个CPU既没有将自己node中的"locked"设为0，也没有将"next"设为NULL，当该CPU释放锁把spinlock交到下一个node手里，它就等同于从这个spinlock的等待队列中移除了。

因此，MCS锁比普通自旋锁稍微复杂一些。但这种额外的复杂性消除了大部分争用情况下的cache-line bouncing；它也是完全公平的，按照CPU到达的顺序将锁传递给每个CPU。

1.2 流程示例

（1）- 线程A获取锁，xchg后prev为NULL，获得锁。
（2）- 线程B尝试获取锁，xchg后prev指向线程A的节点，然后将线程B的节点链接到线程A的next，然后自旋等待自己的locked变为1。
（3）- 线程A在解锁时，检查自己的next是否为NULL，如果非空（即线程B在等待），则设置线程B的locked为1，线程B退出自旋，获得锁。

这样，每个线程在自己的locked变量上自旋，减少了全局锁变量的争用，提高了扩展性。

当获得锁时不需要设置node->locked为1，因为其他线程只在自己的locked上自旋。而当前线程直接返回，不需要自旋。当作为等待者时，自旋在自己的locked上，直到解锁操作将其设为1。

所以，在mcs_spin_unlock中，如果有下一个节点，就设置下一个节点的locked为1，让下一个线程可以继续。

WRITE_ONCE(prev->next, node)是为了确保prev->next的写入是原子的，防止编译器优化或重排序，保证其他线程能够正确看到这个写入操作。

总的来说，MCS锁的获取逻辑通过原子操作构建等待队列，每个线程自旋在自己的本地变量上，减少缓存行的争用，提高多核环境下的性能。

二、源码说明

2.1 mcs_spin_lock

/*
 * In order to acquire the lock, the caller should declare a local node and
 * pass a reference of the node to this function in addition to the lock.
 * If the lock has already been acquired, then this will proceed to spin
 * on this node->locked until the previous lock holder sets the node->locked
 * in mcs_spin_unlock().
 */
static inline
void mcs_spin_lock(struct mcs_spinlock **lock, struct mcs_spinlock *node)
{
	struct mcs_spinlock *prev;

	/* Init node */
	node->locked = 0;
	node->next   = NULL;

	/*
	 * We rely on the full barrier with global transitivity implied by the
	 * below xchg() to order the initialization stores above against any
	 * observation of @node. And to provide the ACQUIRE ordering associated
	 * with a LOCK primitive.
	 */
	prev = xchg(lock, node);
	if (likely(prev == NULL)) {
		/*
		 * Lock acquired, don't need to set node->locked to 1. Threads
		 * only spin on its own node->locked value for lock acquisition.
		 * However, since this thread can immediately acquire the lock
		 * and does not proceed to spin on its own node->locked, this
		 * value won't be used. If a debug mode is needed to
		 * audit lock status, then set node->locked value here.
		 */
		return;
	}
	WRITE_ONCE(prev->next, node);

	/* Wait until the lock holder passes the lock down. */
	arch_mcs_spin_lock_contended(&node->locked);
}

函数是mcs_spin_lock，用来获取锁。用户需要声明一个本地节点，然后把节点和锁的指针传进去。

函数开始的时候，初始化节点的locked为0，next为NULL。然后有一个xchg操作，把锁的指针换成当前节点的地址，同时获取原来的锁指针prev。如果prev是NULL，说明锁没有被占用，直接返回，获取成功。否则，把当前节点链接到prev的next上，然后等待locked的值变化。

xchg操作是一个原子交换，把lock的值设置为node，并返回原来的lock值。这确保了在并发情况下，只有一个线程能成功将lock设置为自己的节点，从而获得锁。

如果prev是NULL，说明之前锁是空闲的，交换后锁现在指向当前节点，所以当前线程获得了锁。这时候不需要设置node->locked为1，因为其他线程在等待自己的node->locked，当前线程不需要自旋，直接返回。

如果prev不是NULL，说明锁已经被其他线程持有。这时候要把当前节点链接到prev的next，也就是等待队列的尾部。

arch_mcs_spin_lock_contended这个函数，参数是&node->locked。这个函数在自旋等待，直到locked的值发生变化，等待locked的值由0变为1。在解锁的时候，前一个线程会把当前节点的locked设为1。

/*
 * Using smp_cond_load_acquire() provides the acquire semantics
 * required so that subsequent operations happen after the
 * lock is acquired. Additionally, some architectures such as
 * ARM64 would like to do spin-waiting instead of purely
 * spinning, and smp_cond_load_acquire() provides that behavior.
 */
#define arch_mcs_spin_lock_contended(l)					\
do {									\
	smp_cond_load_acquire(l, VAL);					\
} while (0)

arch_mcs_spin_lock_contended函数是一个循环，不断检查node->locked是否为1。当解锁时，前一个线程设置node->locked为1，当前线程退出循环，获得锁。

如果prev不为空，说明锁已经被其他线程持有，当前线程需要将自己添加到等待队列，并等待自己的locked变量变为1被解锁。这里的等待队列是通过每个节点的next指针形成的链表，这样每个线程只需要在自己的locked变量上自旋，减少缓存行的竞争。

2.2 mcs_spin_unlock

/*
 * Releases the lock. The caller should pass in the corresponding node that
 * was used to acquire the lock.
 */
static inline
void mcs_spin_unlock(struct mcs_spinlock **lock, struct mcs_spinlock *node)
{
	(1)
	struct mcs_spinlock *next = READ_ONCE(node->next);

	if (likely(!next)) {
		/*
		 * Release the lock by setting it to NULL
		 */
		if (likely(cmpxchg_release(lock, node, NULL) == node))
			return;
		/* Wait until the next pointer is set */
		while (!(next = READ_ONCE(node->next)))
			cpu_relax();
	}

	/* Pass lock to next waiter. */
	arch_mcs_spin_unlock_contended(&next->locked);
}

参数说明：
lock：指向全局锁变量的指针（类型为 struct mcs_spinlock **）。
node：当前线程的本地节点（类型为 struct mcs_spinlock *）。

（1）读取后继节点
READ_ONCE：确保对 node->next 的读取是原子的，避免编译器优化或内存访问重排序。
next：指向当前节点的后继节点（若存在）。

（2）无后继节点（快速路径）
likely(!next)：提示编译器 next == NULL 是大概率事件（优化分支预测）。

cmpxchg_release：原子地将 *lock 从 node 修改为 NULL，并返回 *lock 的旧值。
如果操作成功，表示没有其他线程在等待锁，直接返回。
如果操作失败，表示有新线程加入了等待队列。

等待新节点：
使用 while 循环等待 node->next 被设置（新线程加入队列）。
cpu_relax()：降低 CPU 忙等待的开销（如执行 PAUSE 指令）。

（3）通知后继节点
arch_mcs_spin_unlock_contended：架构相关的通知实现，通常为设置 next->locked 为 1，即设置后继线程的locked值为1。
作用：通知后继线程，使其退出自旋并获取锁。

/*
 * smp_store_release() provides a memory barrier to ensure all
 * operations in the critical section has been completed before
 * unlocking.
 */
#define arch_mcs_spin_unlock_contended(l)				\
	smp_store_release((l), 1)

示例流程：
（1）线程A释放锁：
检查 node->next，发现无后继节点。
使用 cmpxchg_release 将锁置为 NULL，直接返回。

（2）线程B加入队列：
线程B在 mcs_spin_lock 中链接到线程A的 next。

（3）线程A释放锁：
检查 node->next，发现线程B在等待。
设置 next->locked = 1，通知线程B。

（4）线程B获得锁：
退出自旋，继续执行。

三、MCS locks and qspinlocks

在内核中MCS锁用于互斥锁的实现，但并没有取代现有的票据自旋锁实现。其中一个原因是大小：票据自旋锁适合单个32位字，而MCS锁则不适合。这事实证明是重要的：自旋锁嵌入到许多内核结构中，其中一些结构（特别是struct page）不能容忍大小的增加。

MCS锁用于互斥锁和信号量，4.2.0后面的版本qspinlocks取代现有的票据自旋锁
4.2.0以后的Linux内核的自旋锁称为 queued spinlock，基于 MCS 算法的排队自选锁。

参考资料

https://lwn.net/Articles/590243/
http://www.wowotech.net/kernel_synchronization/queued_spinlock.html
https://zhuanlan.zhihu.com/p/89058726
https://zhuanlan.zhihu.com/p/100546935

https://www.slideshare.net/slideshow/spinlockpdf/254977958

Linux 内核自旋锁spinlock（三）--- MCS locks

文章目录

前言

一、MCS locks

1.1 原理简介

1.2 流程示例

二、源码说明

2.1 mcs_spin_lock

2.2 mcs_spin_unlock

三、MCS locks and qspinlocks

参考资料

网站公告

今日签到

热门文章

最新发布