在多线程编程的世界里，锁机制是保证数据一致性的关键。当我们面对"读多写少"的场景时，传统的互斥锁往往成为性能瓶颈。Linux内核如何巧妙地解决这一问题？

一、背景：为什么需要读写锁？

1.1、传统互斥锁的局限性

在并发编程中，互斥锁（mutex）是最基础的同步原语，但它采用“一刀切”的策略：任何时候只允许一个线程访问受保护资源。这种设计在多读少写的场景中存在明显的性能瓶颈。

典型场景示例：

// 配置信息结构体（读多写少）
struct config {
    rwlock_t lock;
    int max_connections;
    int timeout_ms;
    char version[32];
};
// 读取线程（高频调用）
void read_config(struct config* cfg) {
    mutex_lock(&cfg->lock);          // 🔴 所有读者串行化
    printf("Timeout: %d\n", cfg->timeout_ms);
    mutex_unlock(&cfg->lock);
}
// 写入线程（低频调用）
void update_config(struct config* cfg) {
    mutex_lock(&cfg->lock);
    cfg->timeout_ms = 5000;          // ⚡️ 阻塞所有读者
    mutex_unlock(&cfg->lock);
}

1.2 读写锁的核心思想

读写锁（Read-Write Semaphore, Rwsem）引入了一种更精细的并发控制策略：

锁类型

共享特性

适用场景

读锁

共享访问

N个读者可同时访问

写锁

独占访问

仅一个写者，且排斥所有读者

这种设计的理论基础是Amdahl定律：通过并行化读操作来提升整体吞吐量。

1.3 Linux的选择：Rwsem vs RCU

Linux内核提供了多种同步机制，Rwsem在其中定位明确：

// Linux同步机制选择指南
if (读操作 >> 写操作) {
    if (数据量小 && 实时性要求高) {
        使用读写锁(rwsem);          // ✅ 本文重点
    } else if (容忍旧数据 && 追求极致性能) {
        使用RCU;                   // ⚡️ Read-Copy-Update
    }
} else if (读写相当) {
    使用互斥锁(mutex);            // 🔄 Simple is best
} else if (等待时间极短) {
    使用自旋锁(spinlock);         // 🌀 CPU忙等待
}

二、数据结构设计：如何抽象读写状态？

Linux内核中，读写锁的核心实现是struct rw_semaphore（简称rwsem），不同内核版本的结构细节略有差异，但核心字段的设计逻辑一致。

2.1 Rwsem核心结构体分析

// include/linux/rwsem.h
struct rw_semaphore {
    atomic_long_t count;            // 📊 64位状态计数器


#ifdef CONFIG_RWSEM_SPIN_ON_OWNER
    struct optimistic_spin_queue osq;  /* ✨ 乐观自旋队列 */
#endif


    raw_spinlock_t wait_lock;       // 🔒 保护等待队列的自旋锁
    struct list_head wait_list;     // 📋 等待任务链表
};

2.2 count字段的位分解密

count字段采用位域编码，巧妙地将多个状态压缩到一个原子变量中：

/* 64位count字段布局：
 * ┌────────────────┬──────────────┬────────────────┬──────────────┐
 * │  写者计数(16)   │  预留位(16)   │  读者计数(24)   │  活动标志(8) │  
 * └────────────────┴──────────────┴────────────────┴──────────────┘
 */
// 具体定义：
#define RWSEM_WRITER_LOCKED     (1UL << 0)      // Bit0: 写锁定标志
#define RWSEM_FLAG_WAITERS      (1UL << 1)      // Bit1: 存在等待者
#define RWSEM_FLAG_HANDOFF      (1UL << 2)      // Bit2: handoff进行中
#define RWSEM_FLAG_READER_OWNED (1UL << (BITS_PER_LONG - reader_shift))
// 关键宏定义：
#define RWSEM_READER_SHIFT      (BITS_PER_LONG - reader_bits)
#define RWSEM_READER_BIAS       (1UL << RWSEM_READER_SHIFT)
#define RWSEM_WRITER_MASK       (0xFFFFUL << (RWSEM_READER_SHIFT + 24))

这种设计的精妙之处在于：

  • 原子性操作：单个atomic_long_t保证状态变更的原子性
  • 内存高效：避免为每个状态维护独立变量带来的缓存行争用

2.3 等待队列实现机制

采用双链表结构管理等待任务，每个等待节点包含：

struct rwsem_waiter {
    struct list_head list;      // 📌 链表节点，用于挂载到等待队列
    struct task_struct *task;   // 🧵 等待的任务指针
    enum rwsem_waiter_type type;// 🔖 等待类型(READER/WRITER)
    unsigned long timeout;      // ⏱️ 可选的超时时间(如有需要)
};

三、API接口层

与互斥锁的区别：读写锁API通过区分读操作和写操作，提供了更细粒度的同步控制。例如，down_read允许多个读者同时进入临界区，而down_write则必须等待所有读者和写者释放锁后才能获取 。

API使用规范与示例

（1）基本使用模式：

struct rw_semaphore sem;
// 读者线程
void reader_thread(void) 
{
    down_read(&sem);          // 🔒 获取读锁


    /* 临界区: 只读操作 */
    // 📖 多个读者可同时进入


    up_read(&sem);            // 🔓 释放读锁
}
// 写者线程 
void writer_thread(void) 
{
    down_write(&sem);         // ✏️ 获取写锁


    /* 临界区: 写入操作 */
    // ✍️ 独占访问，无其他读者/写者


    up_write(&sem);           // 🚪 释放写锁
}

（2）嵌套使用注意事项：

// 读写锁升级的正确方式
down_read(&sem);                // 🔒 首先获取读锁
if (need_upgrade) {
    up_read(&sem);              // 🔓 先释放读锁
    down_write(&sem);           // ✏️ 再获取写锁


    /* 临界区: 写入操作 */
    // ✍️ 独占访问保证


    up_write(&sem);             // 🚪 释放写锁
} else {
    /* 临界区: 只读操作 */
    // 📖 共享访问


    up_read(&sem);              // 🔓 释放读锁
}
/* 错误示例 - 会导致死锁: */
// down_read(&sem);
// down_write(&sem);  // ❌ 尝试直接升级(可能死锁)
// up_read(&sem);
// up_write(&sem);

四、尝试获取读锁

与 down_read 不同，down_read_trylock 仅尝试获取读锁。若成功，则直接返回 1；若失败，不会阻塞，而是返回 0。其核心逻辑位于 __down_read_trylock 函数中，代码如下：

tmp = RWSEM_UNLOCKED_VALUE;
do {
    if (atomic_long_try_cmpxchg_acquire(&sem->count, &tmp, tmp + RWSEM_READER_BIAS)) { // A
        rwsem_set_reader_owned(sem);
        return 1;
    }
} while (!(tmp & RWSEM_READ_FAILED_MASK)); // B
return 0;
  • 初始状态：tmp 初始化为 RWSEM_UNLOCKED_VALUE（值为 0）。首次循环针对空锁场景进行优化：若当前 sem->count 等于 0，则将其更新为 RWSEM_READER_BIAS，标记持锁成功，随后设置 owner 并返回 1。
  • 快速获取失败后的处理：若快速获取空锁失败（此时 tmp 已更新为当前的 sem->count），需根据当前计数值判断是否可进入临界区。以下情况会导致持读锁失败：
  • RWSEM_WRITER_MASK：已有写者持有锁。
  • RWSEM_FLAG_WAITERS：存在写者或读者在等待队列中。
  • RWSEM_FLAG_HANDOFF：锁正处于转交过程中（即已有预定的 owner）。
  • RWSEM_FLAG_READFAIL：临界区内读者过多（通常为异常状态）。
  • 重试机制：若判断可进入读临界区（即仅有读者且无写者等待），则重新进入循环。如果在此期间 sem->count 未发生变化，则持锁成功：
  • 若其他线程在此期间修改了计数值，则需重新判断是否满足持锁条件并重复上述循环。
  • 增加读者计数。
  • 设置 owner task。
  • 保持 non-spinnable bit不变。
  • 退出条件：若判断无法进入临界区，则退出循环并返回失败（0）。

五、获取读锁

Reader获取读锁的核心逻辑集中在__down_read_common函数中，该函数整合了快速路径、慢速路径的完整处理流程，是读锁获取的核心实现：

// 基于Linux 5.x内核
int __down_read_common(struct rw_semaphore *sem, long state, unsigned long timeout)
{
    struct rwsem_waiter waiter;
    struct wake_q_head wqh;
    unsigned int adjustment = -RWSEM_READER_BIAS;
    long cnt, orig_cnt;
    bool first_reader = false;
    // 快速路径：尝试非阻塞获取读锁
    orig_cnt = cnt = rwsem_read_trylock(sem);
    if (cnt >= 0)
        return 0;
    // 慢速路径：快速路径失败后的处理逻辑
    wake_q_init(&wqh);
    for (;;) {
        // 防止等待队列任务饿死的判断
        if (rwsem_is_reader_owned(sem) && rwsem_has_waiters(sem)) {
            goto queue;
        }
        // 乐观偷锁（optimistic lock stealing）逻辑
        if (!rwsem_is_writer_owned(sem) && !rwsem_handoff_pending(sem) &&
            rwsem_is_waking(sem) && rwsem_has_waiters(sem)) {
            if (rwsem_try_read_lock(sem, cnt)) {
                // 偷锁成功，设置owner
                rwsem_set_reader_owned(sem);
                first_reader = (cnt == RWSEM_READER_BIAS);
                goto wake_readers;
            }
        }
        // 偷锁失败，进入阻塞流程
queue:
        // 初始化等待节点
        waiter.task = current;
        waiter.type = RWSEM_WAITER_READ;
        waiter.timeout = jiffies + 1; // 手机平台HZ=250，对应4ms超时
        init_waitqueue_entry(&waiter.wait, current);
        // 等待队列为空的特殊处理
        if (list_empty(&sem->wait_list)) {
            set_bit(RWSEM_FLAG_WAITERS, &sem->flags);
            // 最后一次尝试持锁
            if (rwsem_try_read_lock(sem, cnt)) {
                clear_bit(RWSEM_FLAG_WAITERS, &sem->flags);
                rwsem_set_reader_owned(sem);
                return 0;
            }
        }
        // 加入等待队列尾部
        rwsem_add_waiter(&waiter, sem, RWSEM_WAITER_READ);
        // 修正之前错误增加的counter
        cnt = atomic_long_add_return(adjustment, &sem->count);
        // 阻塞前的最后唤醒检查
        if (rwsem_is_unlocked(sem)) {
            clear_bit(RWSEM_FLAG_NONSPINNABLE, &sem->flags);
            rwsem_mark_wake(sem, &wqh, RWSEM_WAKE_ANY, 0);
            wake_up_q(&wqh);
        }
        // 阻塞等待
        set_current_state(state);
        if (waiter.task)
            schedule_timeout(timeout);
        __set_current_state(TASK_RUNNING);
        // 检查唤醒状态
        if (!waiter.task) {
            // 正常唤醒，持锁成功
            return 0;
        }
        if (signal_pending(current)) {
            // 信号中断，持锁失败
            rwsem_remove_waiter(&waiter, sem);
            return -EINTR;
        }
        // 超时后设置handoff，禁止乐观自旋
        if (time_after(jiffies, waiter.timeout)) {
            set_bit(RWSEM_FLAG_HANDOFF, &sem->flags);
        }
    }
wake_readers:
    // 唤醒队列中的多个reader（最多MAX_READERS_WAKEUP=256个）
    if (first_reader && !rwsem_is_writer_waiter(sem->wait_list.next)) {
        rwsem_mark_wake(sem, &wqh, RWSEM_WAKE_READERS, MAX_READERS_WAKEUP);
        wake_up_q(&wqh);
    }
    return 0;
}

1. 快速路径

快速路径的核心是rwsem_read_trylock函数，通过原子操作快速尝试获取读锁，避免进入复杂的慢速路径：

// 读锁快速路径：
long rwsem_read_trylock(struct rw_semaphore *sem)
{
    long tmp, cnt;
    tmp = atomic_long_read(&sem->count);
    for (;;) {
        // 检查是否能获取读锁（无writer持有且无读失败标记）
        if (tmp & RWSEM_READ_FAILED_MASK)
            return -1;
        // 原子增加读锁计数（RWSEM_READER_BIAS=1）
        cnt = atomic_long_cmpxchg(&sem->count, tmp, tmp + RWSEM_READER_BIAS);
        if (cnt == tmp)
            break;
        tmp = cnt;
    }
    // 持锁成功，设置reader owned标记
    if (cnt >= 0) {
        rwsem_set_reader_owned(sem);
        // 读线程过多时禁止乐观自旋
        if (cnt > RWSEM_MAX_READERS)
            set_bit(RWSEM_FLAG_NONSPINNABLE, &sem->flags);
        return cnt;
    }
    // 持锁失败，回退计数
    atomic_long_sub(RWSEM_READER_BIAS, &sem->count);
    return -1;
}

核心逻辑解析：

  • Reader直接对sem->count加RWSEM_READER_BIAS（通常为1）来增加读临界区线程数；
  • 若操作成功，设置RWSEM_READER_OWNED标记并返回；若失败则回退计数，进入慢速路径；
  • 当读线程数量超过RWSEM_MAX_READERS时，禁止乐观自旋，避免过度并发导致的性能问题。

2. 慢速路径

快速路径失败后进入慢速路径，其核心是平衡reader吞吐量与waiter饥饿，分为三个关键阶段：

（1）防止waiter饿死的处理

// 防止等待队列任务饿死的判断逻辑
if (rwsem_is_reader_owned(sem) && rwsem_has_waiters(sem)) {
    goto queue;
}

核心逻辑解析：

  • 若锁已被reader持有且等待队列有任务，直接进入排队流程，不再乐观偷锁；
  • 此举是为了平衡：既避免连续reader偷锁导致writer饿死，又能通过后续批量唤醒reader提升吞吐量。

（2）乐观偷锁（Optimistic Lock Stealing）

// 乐观偷锁的核心判断与实现
if (!rwsem_is_writer_owned(sem) && !rwsem_handoff_pending(sem) &&
    rwsem_is_waking(sem) && rwsem_has_waiters(sem)) {
    if (rwsem_try_read_lock(sem, cnt)) {
        rwsem_set_reader_owned(sem);
        first_reader = (cnt == RWSEM_READER_BIAS);
        goto wake_readers;
    }
}

核心逻辑解析：

  • 偷锁的条件：无writer持锁、无handoff标记、正在唤醒top waiter且有任务排队；
  • 若偷锁成功，若为第一个reader，会唤醒队列中最多MAX_READERS_WAKEUP（256）个reader，提升并发效率；
  • 偷锁失败则需回退计数，避免计数错误。

（3）阻塞等待处理

// 阻塞等待的核心代码
// 初始化等待节点
waiter.task = current;
waiter.type = RWSEM_WAITER_READ;
waiter.timeout = jiffies + 1; // 4ms超时（手机平台HZ=250）
// 加入等待队列
if (list_empty(&sem->wait_list)) {
    set_bit(RWSEM_FLAG_WAITERS, &sem->flags);
    // 最后一次尝试持锁
    if (rwsem_try_read_lock(sem, cnt)) {
        clear_bit(RWSEM_FLAG_WAITERS, &sem->flags);
        return 0;
    }
}
rwsem_add_waiter(&waiter, sem, RWSEM_WAITER_READ);
cnt = atomic_long_add_return(adjustment, &sem->count);
// 阻塞前的唤醒检查
if (rwsem_is_unlocked(sem)) {
    clear_bit(RWSEM_FLAG_NONSPINNABLE, &sem->flags);
    rwsem_mark_wake(sem, &wqh, RWSEM_WAKE_ANY, 0);
    wake_up_q(&wqh);
}
// 进入阻塞
set_current_state(state);
if (waiter.task)
    schedule_timeout(timeout);
__set_current_state(TASK_RUNNING);

核心逻辑解析：

  • 等待节点的timeout为4ms（手机平台），超时后触发handoff机制，禁止乐观自旋；
  • 入队前会做最后一次持锁尝试，失败后加入队列尾部，并修正计数；
  • 阻塞后被唤醒时，通过waiter.task是否为NULL判断是否正常唤醒，正常则持锁成功，否则处理信号中断。

六、释放读锁

释放读锁的逻辑在__up_read函数中实现：

void __up_read(struct rw_semaphore *sem)
{
    long cnt;
    cnt = atomic_long_dec_return(&sem->count);
    if (cnt >= 0)
        return;
    rwsem_wake(sem);
}

这里仅减少读计数，不清除owner指针。当等待队列有任务且临界区为空时，系统会唤醒等待队列中的线程。

在唤醒过程中，系统使用rwsem_mark_wake函数，参数为RWSEM_WAKE_ANY：

static void rwsem_mark_wake(struct rw_semaphore *sem, enum rwsem_wake_type wake_type)
{
    struct rwsem_waiter *waiter;
    waiter = list_first_entry(&sem->wait_list, struct rwsem_waiter, list);
    if (waiter->flags & RWSEM_WAITING_FOR_WRITE) {
        if (wake_type == RWSEM_WAKE_ANY)
            rwsem_wake_writer(sem, waiter);
    } else {
        if (wake_type == RWSEM_WAKE_READERS || wake_type == RWSEM_WAKE_READ_OWNED)
            rwsem_wake_readers(sem, waiter);
    }
}

当top waiter是写者时，系统只需唤醒该写者；当top waiter是读者时，系统会唤醒一组读者。

唤醒读者的具体过程如下：

static void rwsem_wake_readers(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
{
    struct rwsem_waiter *pos, *tmp;
    struct list_head wlist;
    int count = 0;
    INIT_LIST_HEAD(&wlist);
    list_for_each_entry_safe(pos, tmp, &sem->wait_list, list) {
        if (pos->flags & RWSEM_WAITING_FOR_WRITE)
            break;
        count++;
        list_move(&pos->list, &wlist);
    }
    if (count > MAX_READERS_WAKEUP)
        count = MAX_READERS_WAKEUP;
    sem->count += count * RWSEM_READER_BIAS;
    sem->waiters -= count;
    list_for_each_entry_safe(pos, tmp, &wlist, list) {
        pos->task = NULL;
        wake_q_add(&wake_q, pos->task);
    }
    wake_q_add(&wake_q, NULL);
    wake_up_q(&wake_q);
}

系统会先唤醒top reader，然后根据情况唤醒其他读者。唤醒的读者数量最多为256个，以避免饿死写者。

七、尝试获取写锁

与down_write不同，down_write_trylock只尝试获取锁而不阻塞：

static inline int rwsem_write_trylock(struct rw_semaphore *sem)
{
    long cnt;
    cnt = atomic_long_read(&sem->count);
    if (cnt == RWSEM_UNLOCKED_VALUE) {
        if (atomic_long_try_cmpxchg_acquire(&sem->count, &cnt, RWSEM_WRITER_LOCKED)) {
            sem->owner = current;
            return 1;
        }
    }
    return 0;
}

这里，系统检查锁是否为空闲状态（RWSEM_UNLOCKED_VALUE=0），如果是，尝试使用原子操作获取写锁。

八、获取写锁

写者获取锁的主流程在__down_write_common函数中：

static void __down_write_common(struct rw_semaphore *sem, int *cntp)
{
    long cnt;
    cnt = atomic_long_read(&sem->count);
    if (likely(cnt == RWSEM_UNLOCKED_VALUE)) {
        if (rwsem_write_trylock(sem))
            return;
    }
    rwsem_down_write_slowpath(sem, cntp);
}

慢速路径中，系统会将自己加入等待队列并尝试唤醒：

static void rwsem_down_write_slowpath(struct rw_semaphore *sem, int *cntp)
{
    struct rwsem_waiter waiter;
    long cnt;
    cnt = atomic_long_read(&sem->count);
    if (cnt < 0) {
        waiter.task = current;
        waiter.flags = RWSEM_WAITING_FOR_WRITE;
        waiter.timeout = jiffies + HZ/250;
        rwsem_add_waiter(sem, &waiter);
        if (rwsem_wake_readers(sem, &waiter))
            return;
    } else {
        if (atomic_long_try_cmpxchg_acquire(&sem->count, &cnt, RWSEM_WRITER_LOCKED)) {
            sem->owner = current;
            return;
        }
    }


    while (1) {
        if (rwsem_try_write_lock(sem)) {
            return;
        }
        if (signal_pending(current))
            return -EINTR;
        if (rwsem_is_handoff(sem)) {
            if (rwsem_spin_on_owner(sem))
                continue;
        }
        set_current_state(TASK_INTERRUPTIBLE);
        schedule();
    }
}

写者进入阻塞状态后，会尝试通过rwsem_try_write_lock获取锁：

static inline int rwsem_try_write_lock(struct rw_semaphore *sem)
{
    long cnt;
    cnt = atomic_long_read(&sem->count);
    if (cnt == RWSEM_UNLOCKED_VALUE) {
        if (atomic_long_try_cmpxchg_acquire(&sem->count, &cnt, RWSEM_WRITER_LOCKED)) {
            sem->owner = current;
            return 1;
        }
    }
    return 0;
}

九、释放写锁

释放写锁的逻辑与释放读锁类似，但需要清除owner指针：

void __up_write(struct rw_semaphore *sem)
{
    long cnt;
    cnt = atomic_long_read(&sem->count);
    if (cnt == RWSEM_WRITER_LOCKED) {
        sem->owner = NULL;
        rwsem_wake(sem);
        return;
    }
    rwsem_wake(sem);

往期干货文章｜C/C++ 开发专属

做 C/C++ 开发的小伙伴，大概率都会碰到选方向、定学习路线、求职面试、练实操这些头疼的问题，特意整理了一波针对性干货，大家按需自取～

选方向 / 求职迷茫？不知道该往哪走、求职没机会、拿不到 offer，甚至分不清是技术不到位还是简历拖后腿？速看👉为什么很多人劝退学 C++，但大厂核心岗位还是要 C++？

目标大厂 Linux C/C++ 后端岗？想找套科学系统的进阶指南，避开学习弯路？必看👉【大厂标准】Linux C/C++ 后端进阶学习路线

想入局音视频流媒体赛道？想掌握核心学习路径，搭建完整技术体系？看这篇👉音视频流媒体高级开发 - 学习路线

想做桌面 / 嵌入式开发，吃透 C++ Qt 技术？需要一套完整的学习闭环？收藏这篇👉C++ Qt 学习路线一条龙！（桌面开发 & 嵌入式开发）

想深耕底层技术，挑战 Linux 内核开发？需要硬核的学习方法和修炼手册？安排👉Linux 内核学习指南，硬核修炼手册

备战 C/C++ 面试？需要高频八股文题库刷题冲刺，夯实面试基础？刷这篇👉C/C++ 高频八股文面试题 1000 题（三）

# c++
Logo
openEuler 社区

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐

cover

DeepSeek Harness 完整排错指南：从安装启动到插件、模型与子agent故障

avatar openEuler 社区
cover

面对小时级漏洞攻击，终端补丁管理该如何破局？

avatar openEuler 社区
cover

从空间内卷到时间破局：摩尔定律与韬定律的底层逻辑对比

avatar openEuler 社区