内核自旋锁_票号锁-CSDN博客

本文链接：https://blog.csdn.net/congchp/article/details/146865308

自旋锁用于处理器之间的互斥，适合保护很短的临界区，并且不允许在临界区睡眠。申请自旋锁的时候，如果自旋锁被其他处理器占有，本处理器自旋等待(也称为忙等待)。

进程、软中断和硬中断都可以使用自旋锁。

内核的自旋锁是排队自旋锁(queuedspinlock，也称为"FIFO ticket spinlock"，票号锁),算法类似于银行柜台的排队叫号。

（1）锁拥有排队号（也叫票号，ticket）和服务号，服务号是当前占有锁的进程的排队号。

（2）每个进程申请锁的时候，首先申请一个排队号，然后轮询锁的服务号是否等于自己的排队号，如果等于，表示自己占有锁，可以进入临界区，否则继续轮询。

（3）当进程释放锁时,把服务号加1，下一个进程看到服多号等于自己的排队号，退出自旋，进入临界区。

为什么中断中可以使用自旋锁？

中断中是不可以使用mutex的，因为mutex会引起进程sleep。中断上下文中是不允许**sleep**的。而spinlock中不会sleep，所以中断中可以使用它。

数据结构

typedef struct raw_spinlock {
	arch_spinlock_t raw_lock;
} raw_spinlock_t;

typedef struct spinlock {
	struct raw_spinlock rlock;
} spinlock_t;

spinlock_t-> raw_spinlock -> arch_spinlock_t，最后使用的实际是各处理器架构的arch_spinlock_t。

操作函数

初始化

定义并且初始化静态自旋锁的方法如下:

DEFINE_SPINLOCK(x)

在运行时动态初始化自旋锁的方法如下:

spin_lock_init(x)

申请自旋锁

申请自旋锁前，为什么要先禁止硬中断或者软中断？

申请自旋锁的函数如下。

（1）void spin_lock(spinlock_t *lock);

申请自旋锁,如果锁被其他处理器占有,当前处理器自旋等待。

（2）void spin_lock_bh(spinlock_t *lock);

申请自旋锁,并且禁止当前处理器的软中断。

（3）void spin_lock_irq(spinlock_t*lock);

申请自旋锁,并且禁止当前处理器的硬中断。

（4） spin_lock_irqsave(lock, flags);

申请自旋锁,保存当前处理器的硬中断状态,并且禁止当前处理器的硬中断。

（5） int spin_trylock(spinlock_t *lock);

申请自旋锁,如果申请成功,返回1;如果锁被其他处理器器占有,当前处理器不等待,立即返回0。

释放自旋锁

释放自旋锁的函数如下。

（1）void spin_unlock(spinlock_t*lock);

（2）void spin_unlock_bh(spinlock_t*lock);

释放自旋锁,并且开启当前处理器的软中断。

（3）void spin_unlock_irq(spinlock_t*lock);

释放自旋锁,并且开启当前处理器的硬中断。

（4）void spin_unlock_irqrestore(spinlock_t *lock, unsigned longflags);

释放自旋锁,并且恢复当前处理器的硬中断状态。

ARM64中spinlock实现

数据结构

typedef struct {
#ifdef __AARCH64EB__ // big endian
	u16 next;
	u16 owner;
#else                // little endian
	u16 owner;
	u16 next;
#endif
} __aligned(4) arch_spinlock_t;

arch_spinlock_t 是ARM64架构下的自旋锁数据结构。它包含两个16位的字段：

owner：表示当前持有锁的进程的票号。
next：表示下一个等待获取锁的进程的票号。

在ARM64架构中，spinlock 使用了“票号”（ticket）机制来实现公平锁。每个尝试获取锁的进程都会获得一个唯一的票号，各进程按票号的顺序依次获取锁。

函数实现

`arch_spin_lock`

spin_lock() -> raw_spin_lock() -> _raw_spin_lock() -> __raw_spin_lock() -> _ > do_raw_spin_lock() -> arch_spin_lock()

static inline void arch_spin_lock(arch_spinlock_t *lock)
{
	unsigned int tmp;
	arch_spinlock_t lockval, newval;

	asm volatile(
	/* Atomically increment the next ticket. */
	ARM64_LSE_ATOMIC_INSN(
	/* LL/SC */
"	prfm	pstl1strm, %3\n"
"1:	ldaxr	%w0, %3\n"
"	add	%w1, %w0, %w5\n"
"	stxr	%w2, %w1, %3\n"
"	cbnz	%w2, 1b\n",
	/* LSE atomics */
"	mov	%w2, %w5\n"
"	ldadda	%w2, %w0, %3\n"
	__nops(3)
	)

	/* Did we get the lock? */
"	eor	%w1, %w0, %w0, ror #16\n"
"	cbz	%w1, 3f\n"
	/*
	 * No: spin on the owner. Send a local event to avoid missing an
	 * unlock before the exclusive load.
	 */
"	sevl\n"
"2:	wfe\n"
"	ldaxrh	%w2, %4\n"
"	eor	%w1, %w2, %w0, lsr #16\n"
"	cbnz	%w1, 2b\n"
	/* We got the lock. Critical section starts here. */
"3:"
	: "=&r" (lockval), "=&r" (newval), "=&r" (tmp), "+Q" (*lock)
	: "Q" (lock->owner), "I" (1 << TICKET_SHIFT)
	: "memory");
}

arch_spin_lock 函数用于获取自旋锁。spinlock 实现使用了票号机制来保证锁的公平性。通过原子操作和忙等待的方式，确保在多核系统中只有一个CPU核心可以进入临界区。spinlock 的实现依赖于ARM64的原子指令和事件机制，以高效地实现锁的获取和释放。

它的主要步骤如下：

原子地增加**** **next** ****票号：
- 使用ARM64 LSE（Large System Extensions）原子指令或LL/SC（Load-Linked/Store-Conditional）指令实现原子地增加**next**票号。
 - LL/SC指令：使用 ldaxr 和 stxr 指令（Load-Exclusive 和 Store-Exclusive）来实现原子操作增加next票号。
 - 如果系统支持 LSE（Large System Extensions），则使用 ldadda 指令来原子地增加 next 票号。
- 下一个排队的进程申请的排队号就是增加后的next票号。
检查是否获取到锁：
- 通过 eor 和 cbz 指令检查当前 **owner** （服务号）是否等于自己持有的票号（增加前的next票号）。如果相等，则表示获取到了锁，跳转到标签 3，进入临界区。
- 如果不相等，则表示锁已经被其他核心持有，当前核心需要自旋等待（忙等待）。
忙等待：
- sevl (Set Event Local) 指令的功能是发送一个本地事件，避免错过其它处理器释放自旋锁时发送的事件。
- wfe (Wait For Event) 让CPU进入低功耗等待状态，等待其它处理器 sev 触发事件。
 - **wfe****不仅仅受 ****sev**唤醒，还可以被内存访问（如写入某个地址）间接触发。
 - arch_spin_unlock中并没有触发sev，但是将owner增加时，也能够唤醒执行wfe的处理器。
- 使用 ldaxrh 和eor指令不断检查 owner 字段，直到 owner 等于自己持有的票号，表示锁已经被释放，当前核心可以获取锁。
进入临界区：
- 当获取到锁后，代码跳转到标签 3，进入临界区。

`arch_spin_unlock`

spin_unlock() -> raw_spin_unlock() -> _raw_sspin_unlock() -> __raw_spin_unlock() -> do_raw_spin_unlock() -> arch_spin_unlock()

static inline void arch_spin_unlock(arch_spinlock_t *lock)
{
	unsigned long tmp;

	asm volatile(ARM64_LSE_ATOMIC_INSN(
	/* LL/SC */
	"	ldrh	%w1, %0\n"
	"	add	%w1, %w1, #1\n"
	"	stlrh	%w1, %0",
	/* LSE atomics */
	"	mov	%w1, #1\n"
	"	staddlh	%w1, %0\n"
	__nops(1))
	: "=Q" (lock->owner), "=&r" (tmp)
	:
	: "memory");
}

arch_spin_unlock 的功能是释放自旋锁，具体步骤如下：

增加**** **owner** ****字段：
- 将 lock->owner 的值加 1，表示当前持有锁的 CPU 核心已经释放了锁。
- 如果系统支持 LSE，则使用原子指令 staddlh 直接完成加 1 操作。
- 如果系统不支持 LSE，则使用 LL/SC 指令序列（ldrh、add、stlrh）完成加 1 操作。
内存同步：
- 使用 stlrh 或 staddlh 指令确保之前的加载和存储操作按顺序完成，避免内存访问乱序。
通知其他 CPU 核心：
- 通过增加 owner 字段的值，通知其他正在等待锁的 CPU 核心，锁已经被释放，它们可以尝试获取锁。

工作原理

为了更好地理解 ARM64 架构下 票号锁（ticket lock）的工作原理，我们可以通过图示来说明 owner 和 next 的变化过程。以下是详细的图示和解释。

1. 初始状态

在初始状态下，owner 和 next 的值均为 0，表示锁未被任何 CPU 核心持有，且没有 CPU 核心在等待锁。

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         0         |         0         |
+-------------------+-------------------+

**owner = 0**：表示当前持有锁的票号为 0。
**next = 0**：表示下一个等待锁的票号将从 0 开始分配。

2. 第一次执行 `arch_spin_lock`

当一个 CPU 核心（假设为 CPU 0）第一次调用 arch_spin_lock 时，会发生以下变化：

2.1 原子地增加 `next`

next 的值从 0 增加到 1。
CPU 0 获取的票号为 next 的旧值，即 0。

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         0         |         1         |
+-------------------+-------------------+

**owner = 0**：未变化。
**next = 1**：已增加，表示下一个等待锁的票号将从 1 开始分配。

2.2 检查是否获取到锁

CPU 0 的票号为 0，与 owner 的值相等，因此 CPU 0 成功获取锁。

3. 第一次执行 `arch_spin_unlock`

当 CPU 0 调用 arch_spin_unlock 释放锁时，会发生以下变化：

3.1 原子地增加 `owner`

owner 的值从 0 增加到 1。

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         1         |         1         |
+-------------------+-------------------+

**owner = 1**：已增加，表示锁已被释放，下一个等待锁的 CPU 核心可以获取锁。
**next = 1**：未变化。

4. 第二次执行 `arch_spin_lock`

当另一个 CPU 核心（假设为 CPU 1）调用 arch_spin_lock 时，会发生以下变化：

4.1 原子地增加 `next`

next 的值从 1 增加到 2。
CPU 1 获取的票号为 next 的旧值，即 1。

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         1         |         2         |
+-------------------+-------------------+

**owner = 1**：未变化。
**next = 2**：已增加，表示下一个等待锁的票号将从 2 开始分配。

4.2 检查是否获取到锁

CPU 1 的票号为 1，与 owner 的值相等，因此 CPU 1 成功获取锁。

5. 第二次执行 `arch_spin_unlock`

当 CPU 1 调用 arch_spin_unlock 释放锁时，会发生以下变化：

5.1 原子地增加 `owner`

owner 的值从 1 增加到 2。

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         2         |         2         |
+-------------------+-------------------+

**owner = 2**：已增加，表示锁已被释放，下一个等待锁的 CPU 核心可以获取锁。
**next = 2**：未变化。

6. 总结图示

初始状态

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         0         |         0         |
+-------------------+-------------------+

CPU 0 获取锁

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         0         |         1         |
+-------------------+-------------------+

CPU 0 释放锁

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         1         |         1         |
+-------------------+-------------------+

CPU 1 获取锁

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         1         |         2         |
+-------------------+-------------------+

CPU 1 释放锁

+-------------------+-------------------+
|      owner        |       next        |
+-------------------+-------------------+
|         2         |         2         |
+-------------------+-------------------+

7. 票号锁的工作原理

通过上述图示和过程，可以看出票号锁的工作原理：

**next**：表示下一个等待锁的票号，每次有 CPU 核心尝试获取锁时，next 的值会原子地增加。
**owner**：表示当前持有锁的票号，每次有 CPU 核心释放锁时，owner 的值会原子地增加。
公平性：CPU 核心按票号的顺序依次获取锁，确保每个 CPU 核心都能公平地获取锁。

这种机制避免了某些 CPU 核心长时间无法获取锁的情况，实现了公平的锁获取策略。

注意事项

一般使用spin_lock时，需要禁止内核抢占以及中断。内核实现spin_lock时，已经进行了preempt_disable禁止内核抢占。如果要禁止硬中断，需要使用spin_lock_irq。禁止软中断，使用spin_lock_bh。

禁止内核抢占、中断的原因

防止死锁：避免持有锁的任务被抢占，导致其他任务无法获取锁。
减少锁持有时间：确保自旋锁的持有时间尽可能短。
保证数据一致性：防止临界区内的操作被中断或抢占。

内核抢占的概念

内核抢占是指在内核模式下，当前任务可以被更高优先级的任务抢占，即使当前任务没有主动让出 CPU。

自旋锁、原子变量、互斥量比较

特性	自旋锁（spinlock_t）	互斥锁（mutex）	原子变量（atomic_t）
等待方式	忙等待	阻塞等待	无等待
是否可睡眠	不可睡眠	可睡眠	无睡眠
适用上下文	中断上下文、进程上下文	进程上下文	中断上下文、进程上下文
开销	低（短临界区）	高（上下文切换）	极低
适用场景	短临界区、锁竞争不激烈	长临界区、锁竞争激烈	简单操作、无锁数据结构
复杂性	中等	高	低

用户空间自旋锁

linux用户空间一般使用posix 线程库（pthread）中的自旋锁。它的实现原理、使用方法与内核自旋锁类似。

pthread中的spinlock既可以用于线程间通信，也可以用于进程间。因为linux不区分进程和线程。

用于进程间的spinlock需要依赖共享内存以及PTHREAD_PROCESS_SHARED。

线程间spinlock：

// gcc -o posix_spinlock_test posix_spinlock_test.c -pthread
#include <stdio.h>
#include <pthread.h>

pthread_spinlock_t lock; // 自旋锁

void *thread_func(void *arg) {
    pthread_spin_lock(&lock); // 获取锁
    printf("Thread %ld acquired the lock\n", (long)pthread_self());
    // 模拟临界区
    for (int i = 0; i < 1000000; i++) {}
    printf("Thread %ld released the lock\n", (long)pthread_self());
    pthread_spin_unlock(&lock); // 释放锁
    return NULL;
}

int main() {
    pthread_t thread1, thread2;

    // 初始化自旋锁
    pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE);

    // 创建两个线程
    pthread_create(&thread1, NULL, thread_func, NULL);
    pthread_create(&thread2, NULL, thread_func, NULL);

    // 等待线程结束
    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    // 销毁自旋锁
    pthread_spin_destroy(&lock);

    return 0;
}

进程间spinlock：

// gcc -o posix_spinlock_processA_test posix_spinlock_processA_test.c -pthread -lrt
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <unistd.h>

#define SHM_NAME "/spinlock_shm"

typedef struct {
    pthread_spinlock_t spinlock;
    int shared_counter;
} shared_data_t;

int main() {
    // 创建共享内存
    int shm_fd = shm_open(SHM_NAME, O_CREAT | O_RDWR, 0666);
    if (shm_fd < 0) {
        perror("shm_open");
        exit(EXIT_FAILURE);
    }

    // 调整共享内存大小
    ftruncate(shm_fd, sizeof(shared_data_t));

    // 映射共享内存
    shared_data_t *data = mmap(NULL, sizeof(shared_data_t), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (data == MAP_FAILED) {
        perror("mmap");
        exit(EXIT_FAILURE);
    }

    // 初始化自旋锁
    pthread_spin_init(&data->spinlock, PTHREAD_PROCESS_SHARED);
    data->shared_counter = 0;

    printf("Process A: Initialized spinlock and shared counter.\n");

    // 使用自旋锁
    for (int i = 0; i < 5; i++) {
        pthread_spin_lock(&data->spinlock);
        data->shared_counter++;
        printf("Process A: Incremented counter to %d\n", data->shared_counter);
        pthread_spin_unlock(&data->spinlock);
        sleep(1);
    }

    // 等待用户确认退出
    getchar();

    // 销毁自旋锁并清理共享内存
    pthread_spin_destroy(&data->spinlock);
    munmap(data, sizeof(shared_data_t));
    shm_unlink(SHM_NAME);

    return 0;
}

// gcc -o posix_spinlock_processB_test posix_spinlock_processB_test.c -pthread -lrt
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <unistd.h>

#define SHM_NAME "/spinlock_shm"

typedef struct {
    pthread_spinlock_t spinlock;
    int shared_counter;
} shared_data_t;

int main() {
    // 打开已经存在的共享内存
    int shm_fd = shm_open(SHM_NAME, O_RDWR, 0666);
    if (shm_fd < 0) {
        perror("shm_open");
        exit(EXIT_FAILURE);
    }

    // 映射共享内存
    shared_data_t *data = mmap(NULL, sizeof(shared_data_t), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (data == MAP_FAILED) {
        perror("mmap");
        exit(EXIT_FAILURE);
    }

    printf("Process B: Attached to shared memory.\n");

    // 使用自旋锁
    for (int i = 0; i < 5; i++) {
        pthread_spin_lock(&data->spinlock);
        data->shared_counter++;
        printf("Process B: Incremented counter to %d\n", data->shared_counter);
        pthread_spin_unlock(&data->spinlock);
        sleep(1);
    }

    // 清理共享内存映射
    munmap(data, sizeof(shared_data_t));

    return 0;
}