12. 读写锁与自旋锁.md 23 KB

12 读写锁与自旋锁

12.1 完整概念讲解

12.1.1 读写锁概念

读写锁(Read-Write Lock) 是一种特殊的同步原语,专为读多写少场景设计。与互斥锁不同,读写锁允许多个读线程同时持有锁,但写线程必须独占访问。

核心思想:

  • 读操作不修改数据:多个线程可以同时读取共享资源
  • 写操作修改数据:写线程必须独占,不能有其他读或写
  • 优化目标:提高读并发性,减少读线程间的竞争

读写锁状态:

状态 描述 允许的操作
未锁定 无任何线程持有锁 读锁、写锁
读锁定 一个或多个线程持有读锁 读锁(继续加锁)、写锁(阻塞)
写锁定 一个线程持有写锁 无(所有操作阻塞)

12.1.2 读写锁的优先级问题

读写锁存在读优先写优先两种实现策略,不同的策略会导致不同的性能特性:

读优先(Read-Preference):

  • 当读线程持有锁时,新的读线程可以立即获取锁
  • 写线程必须等待所有读线程释放锁
  • 优点:读并发性高
  • 缺点:写线程可能长时间饥饿(Writer Starvation)

写优先(Write-Preference):

  • 当写线程等待获取锁时,新的读线程必须排队
  • 写线程获取锁后,后续的读线程必须等待
  • 优点:写线程不会饥饿
  • 缺点:读并发性降低

Linux实现(NPTL):

  • Linux的pthread_rwlock默认采用读优先策略
  • 可以通过pthread_rwlockattr_setkind_np()设置写优先

12.1.3 自旋锁概念

自旋锁(Spinlock) 是一种忙等待(Busy-Waiting)锁机制。当线程无法获取锁时,它不会睡眠或让出CPU,而是在一个循环中不断尝试获取锁。

核心特点:

  • 忙等待:线程在循环中不断检查锁状态
  • 不释放CPU:线程不会睡眠,持续占用CPU时间片
  • 适用场景:锁持有时间极短的场景

自旋锁的工作流程:

线程尝试获取锁
    ↓
锁可用? ──是──→ 获取锁,继续执行
    │
    否
    │
    └──→ 在循环中不断重试(自旋)
            ↓
        锁被释放? ──是──→ 获取锁,继续执行
            │
            否
            └──→ 继续自旋

12.1.4 自旋锁 vs 互斥锁

特性 自旋锁 互斥锁
等待方式 忙等待(自旋) 睡眠等待
CPU占用 持续占用CPU 不占用CPU
上下文切换
适用场景 锁持有时间极短 锁持有时间较长
多CPU/多核 适合 适合
单CPU 不适合(浪费CPU) 适合
中断上下文 可以使用 不能使用(不能睡眠)

选择原则:

  • 锁持有时间 < 上下文切换时间 → 自旋锁
  • 锁持有时间 > 上下文切换时间 → 互斥锁
  • 中断处理函数中 → 必须用自旋锁

12.1.5 嵌入式应用场景

在嵌入式系统中,读写锁和自旋锁有特定的应用场景:

自旋锁的典型应用:

  1. 中断处理:中断处理函数不能睡眠,必须用自旋锁保护共享数据
  2. 多核SOC的核间同步:CPU核心间的短暂同步
  3. 短临界区:保护只有几条指令的临界区
  4. 内核与驱动:内核中广泛使用自旋锁

读写锁的典型应用:

  1. 配置数据:读取配置频繁,修改配置很少
  2. 路由表:查询频繁,更新较少
  3. 日志系统:多线程写日志,偶尔读取
  4. 缓存系统:频繁读取缓存,偶尔更新

12.2 核心API/语法

12.2.1 读写锁函数

#include <pthread.h>

// 初始化读写锁
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
                        const pthread_rwlockattr_t *restrict attr);

// 销毁读写锁
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

// 获取读锁(阻塞)
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);

// 非阻塞获取读锁
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

// 获取写锁(阻塞)
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

// 非阻塞获取写锁
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);

// 释放读写锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

参数说明:

  • rwlock:指向读写锁的指针
  • attr:读写锁属性,NULL使用默认属性

返回值:

  • 成功:0
  • 失败:非零错误码

12.2.2 读写锁属性函数

#include <pthread.h>

// 初始化读写锁属性
int pthread_rwlockattr_init(pthread_rwlockattr_t *attr);

// 销毁读写锁属性
int pthread_rwlockattr_destroy(pthread_rwlockattr_t *attr);

// 获取/设置进程共享属性
int pthread_rwlockattr_getpshared(const pthread_rwlockattr_t *restrict attr,
                                  int *restrict pshared);
int pthread_rwlockattr_setpshared(pthread_rwlockattr_t *attr, int pshared);

// 获取/设置写优先属性(非标准扩展)
int pthread_rwlockattr_getkind_np(const pthread_rwlockattr_t *attr,
                                  int *pref);
int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);

写优先设置:

// 设置为写优先
pthread_rwlockattr_t attr;
pthread_rwlockattr_init(&attr);
pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);

12.2.3 自旋锁函数

#include <pthread.h>

// 初始化自旋锁
int pthread_spin_init(pthread_spinlock_t *lock, int pshared);

// 销毁自旋锁
int pthread_spin_destroy(pthread_spinlock_t *lock);

// 获取自旋锁(阻塞)
int pthread_spin_lock(pthread_spinlock_t *lock);

// 非阻塞获取自旋锁
int pthread_spin_trylock(pthread_spinlock_t *lock);

// 释放自旋锁
int pthread_spin_unlock(pthread_spinlock_t *lock);

参数说明:

  • lock:指向自旋锁的指针
  • pshared:进程共享属性
    • PTHREAD_PROCESS_PRIVATE:仅在进程内使用(默认)
    • PTHREAD_PROCESS_SHARED:可在进程间共享

返回值:

  • 成功:0
  • 失败:非零错误码

12.3 代码示例

12.3.1 读写锁基础示例

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define NUM_READERS 5
#define NUM_WRITERS 2
#define NUM_OPS 10

// 共享资源
int shared_data = 0;
pthread_rwlock_t rwlock;

// 读线程函数
void *reader(void *arg) {
    int id = *(int *)arg;

    for (int i = 0; i < NUM_OPS; i++) {
        // 获取读锁
        pthread_rwlock_rdlock(&rwlock);

        // 读取共享数据
        printf("[Reader %d] Read: %d\n", id, shared_data);

        // 模拟读操作耗时
        usleep(100);

        // 释放读锁
        pthread_rwlock_unlock(&rwlock);

        usleep(50);
    }

    return NULL;
}

// 写线程函数
void *writer(void *arg) {
    int id = *(int *)arg;

    for (int i = 0; i < NUM_OPS; i++) {
        // 获取写锁
        pthread_rwlock_wrlock(&rwlock);

        // 修改共享数据
        shared_data++;
        printf("[Writer %d] Write: %d\n", id, shared_data);

        // 模拟写操作耗时
        usleep(200);

        // 释放写锁
        pthread_rwlock_unlock(&rwlock);

        usleep(100);
    }

    return NULL;
}

int main() {
    pthread_t readers[NUM_READERS], writers[NUM_WRITERS];
    int reader_ids[NUM_READERS], writer_ids[NUM_WRITERS];

    // 初始化读写锁
    pthread_rwlock_init(&rwlock, NULL);

    // 创建读线程
    for (int i = 0; i < NUM_READERS; i++) {
        reader_ids[i] = i;
        pthread_create(&readers[i], NULL, reader, &reader_ids[i]);
    }

    // 创建写线程
    for (int i = 0; i < NUM_WRITERS; i++) {
        writer_ids[i] = i;
        pthread_create(&writers[i], NULL, writer, &writer_ids[i]);
    }

    // 等待所有线程完成
    for (int i = 0; i < NUM_READERS; i++) {
        pthread_join(readers[i], NULL);
    }
    for (int i = 0; i < NUM_WRITERS; i++) {
        pthread_join(writers[i], NULL);
    }

    // 销毁读写锁
    pthread_rwlock_destroy(&rwlock);

    printf("Final shared_data: %d\n", shared_data);
    return 0;
}

编译运行:

gcc -pthread -o rwlock_basic rwlock_basic.c
./rwlock_basic

12.3.2 自旋锁基础示例

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define NUM_THREADS 4
#define NUM_OPS 100

// 共享资源
int shared_counter = 0;
pthread_spinlock_t spinlock;

// 线程函数
void *thread_func(void *arg) {
    int id = *(int *)arg;

    for (int i = 0; i < NUM_OPS; i++) {
        // 获取自旋锁
        pthread_spin_lock(&spinlock);

        // 临界区操作
        shared_counter++;

        // 释放自旋锁
        pthread_spin_unlock(&spinlock);
    }

    printf("[Thread %d] Finished, counter = %d\n", id, shared_counter);
    return NULL;
}

int main() {
    pthread_t threads[NUM_THREADS];
    int thread_ids[NUM_THREADS];

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

    // 创建线程
    for (int i = 0; i < NUM_THREADS; i++) {
        thread_ids[i] = i;
        pthread_create(&threads[i], NULL, thread_func, &thread_ids[i]);
    }

    // 等待所有线程完成
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }

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

    printf("Final counter: %d (expected: %d)\n", shared_counter, NUM_THREADS * NUM_OPS);
    return 0;
}

编译运行:

gcc -pthread -o spinlock_basic spinlock_basic.c
./spinlock_basic

12.3.3 自旋锁 vs 互斥锁性能对比

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <time.h>

#define NUM_THREADS 4
#define NUM_OPS 1000000

// 共享资源
int shared_counter = 0;
pthread_mutex_t mutex;
pthread_spinlock_t spinlock;

// 互斥锁线程函数
void *mutex_thread(void *arg) {
    for (int i = 0; i < NUM_OPS; i++) {
        pthread_mutex_lock(&mutex);
        shared_counter++;
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

// 自旋锁线程函数
void *spin_thread(void *arg) {
    for (int i = 0; i < NUM_OPS; i++) {
        pthread_spin_lock(&spinlock);
        shared_counter++;
        pthread_spin_unlock(&spinlock);
    }
    return NULL;
}

double get_time_ms(struct timespec *start, struct timespec *end) {
    return (end->tv_sec - start->tv_sec) * 1000.0 +
           (end->tv_nsec - start->tv_nsec) / 1000000.0;
}

int main() {
    pthread_t threads[NUM_THREADS];
    struct timespec start, end;
    double mutex_time, spin_time;

    // 初始化
    pthread_mutex_init(&mutex, NULL);
    pthread_spin_init(&spinlock, PTHREAD_PROCESS_PRIVATE);

    // 测试互斥锁
    shared_counter = 0;
    clock_gettime(CLOCK_MONOTONIC, &start);
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_create(&threads[i], NULL, mutex_thread, NULL);
    }
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }
    clock_gettime(CLOCK_MONOTONIC, &end);
    mutex_time = get_time_ms(&start, &end);
    printf("Mutex: counter=%d, time=%.2f ms\n", shared_counter, mutex_time);

    // 测试自旋锁
    shared_counter = 0;
    clock_gettime(CLOCK_MONOTONIC, &start);
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_create(&threads[i], NULL, spin_thread, NULL);
    }
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }
    clock_gettime(CLOCK_MONOTONIC, &end);
    spin_time = get_time_ms(&start, &end);
    printf("Spinlock: counter=%d, time=%.2f ms\n", shared_counter, spin_time);

    // 清理
    pthread_mutex_destroy(&mutex);
    pthread_spin_destroy(&spinlock);

    printf("Spinlock is %.2fx faster than mutex\n", mutex_time / spin_time);
    return 0;
}

编译运行:

gcc -pthread -o lock_compare lock_compare.c
./lock_compare

12.3.4 嵌入式场景:中断处理中的自旋锁

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <signal.h>
#include <unistd.h>

// 模拟中断共享数据
volatile int sensor_data = 0;
pthread_spinlock_t irq_spinlock;

// 模拟中断处理函数(在真实嵌入式系统中这是ISR)
void irq_handler(int sig) {
    // 在中断处理函数中使用自旋锁保护共享数据
    pthread_spin_lock(&irq_spinlock);

    // 更新传感器数据
    sensor_data++;

    pthread_spin_unlock(&irq_spinlock);
}

// 主线程:读取传感器数据
void *sensor_reader(void *arg) {
    int last_value = 0;

    for (int i = 0; i < 20; i++) {
        pthread_spin_lock(&irq_spinlock);
        int current_value = sensor_data;
        pthread_spin_unlock(&irq_spinlock);

        if (current_value != last_value) {
            printf("[Reader] Sensor value changed: %d -> %d\n",
                   last_value, current_value);
            last_value = current_value;
        }

        usleep(100000);  // 100ms
    }

    return NULL;
}

int main() {
    pthread_t reader_thread;

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

    // 注册信号处理函数(模拟中断)
    struct sigaction sa;
    sa.sa_handler = irq_handler;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = 0;
    sigaction(SIGUSR1, &sa, NULL);

    // 创建读线程
    pthread_create(&reader_thread, NULL, sensor_reader, NULL);

    // 模拟中断触发
    for (int i = 0; i < 10; i++) {
        usleep(50000);  // 50ms
        kill(getpid(), SIGUSR1);  // 发送信号模拟中断
    }

    // 等待读线程完成
    pthread_join(reader_thread, NULL);

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

    printf("Final sensor value: %d\n", sensor_data);
    return 0;
}

编译运行:

gcc -pthread -o irq_spinlock irq_spinlock.c
./irq_spinlock

12.3.5 嵌入式场景:配置数据的读写锁

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

#define MAX_CONFIG_ENTRIES 10

// 配置结构体
typedef struct {
    char key[64];
    char value[128];
} ConfigEntry;

// 配置管理器
typedef struct {
    ConfigEntry entries[MAX_CONFIG_ENTRIES];
    int count;
    pthread_rwlock_t rwlock;
} ConfigManager;

// 初始化配置管理器
void config_init(ConfigManager *mgr) {
    mgr->count = 0;
    pthread_rwlock_init(&mgr->rwlock, NULL);

    // 添加一些默认配置
    strcpy(mgr->entries[0].key, "device_name");
    strcpy(mgr->entries[0].value, "Embedded_System_v1.0");
    strcpy(mgr->entries[1].key, "ip_address");
    strcpy(mgr->entries[1].value, "192.168.1.100");
    strcpy(mgr->entries[2].key, "port");
    strcpy(mgr->entries[2].value, "8080");
    mgr->count = 3;
}

// 读取配置(频繁调用)
const char *config_get(ConfigManager *mgr, const char *key) {
    pthread_rwlock_rdlock(&mgr->rwlock);

    for (int i = 0; i < mgr->count; i++) {
        if (strcmp(mgr->entries[i].key, key) == 0) {
            const char *value = mgr->entries[i].value;
            pthread_rwlock_unlock(&mgr->rwlock);
            return value;
        }
    }

    pthread_rwlock_unlock(&mgr->rwlock);
    return NULL;
}

// 更新配置(偶尔调用)
int config_set(ConfigManager *mgr, const char *key, const char *value) {
    pthread_rwlock_wrlock(&mgr->rwlock);

    // 查找已存在的key
    for (int i = 0; i < mgr->count; i++) {
        if (strcmp(mgr->entries[i].key, key) == 0) {
            strcpy(mgr->entries[i].value, value);
            pthread_rwlock_unlock(&mgr->rwlock);
            return 0;  // 更新成功
        }
    }

    // 添加新key
    if (mgr->count < MAX_CONFIG_ENTRIES) {
        strcpy(mgr->entries[mgr->count].key, key);
        strcpy(mgr->entries[mgr->count].value, value);
        mgr->count++;
        pthread_rwlock_unlock(&mgr->rwlock);
        return 0;  // 添加成功
    }

    pthread_rwlock_unlock(&mgr->rwlock);
    return -1;  // 配置已满
}

// 销毁配置管理器
void config_destroy(ConfigManager *mgr) {
    pthread_rwlock_destroy(&mgr->rwlock);
}

// 读线程:频繁读取配置
void *reader_thread(void *arg) {
    ConfigManager *mgr = (ConfigManager *)arg;

    for (int i = 0; i < 20; i++) {
        const char *name = config_get(mgr, "device_name");
        const char *ip = config_get(mgr, "ip_address");
        const char *port = config_get(mgr, "port");

        printf("[Reader] device=%s, ip=%s, port=%s\n",
               name ? name : "N/A",
               ip ? ip : "N/A",
               port ? port : "N/A");

        usleep(100000);  // 100ms
    }

    return NULL;
}

// 写线程:偶尔更新配置
void *writer_thread(void *arg) {
    ConfigManager *mgr = (ConfigManager *)arg;

    for (int i = 0; i < 5; i++) {
        usleep(500000);  // 500ms

        char value[128];
        sprintf(value, "192.168.1.%d", 100 + i);
        config_set(mgr, "ip_address", value);

        printf("[Writer] Updated ip_address to %s\n", value);
    }

    return NULL;
}

int main() {
    ConfigManager mgr;
    pthread_t reader, writer;

    config_init(&mgr);

    // 创建线程
    pthread_create(&reader, NULL, reader_thread, &mgr);
    pthread_create(&writer, NULL, writer_thread, &mgr);

    // 等待线程完成
    pthread_join(reader, NULL);
    pthread_join(writer, NULL);

    config_destroy(&mgr);

    printf("Config test completed.\n");
    return 0;
}

编译运行:

gcc -pthread -o config_rwlock config_rwlock.c
./config_rwlock

12.4 注意事项与易错点

12.4.1 读写锁注意事项

问题 描述 解决方案
写线程饥饿 读线程持续持有锁,写线程无法获取 设置写优先属性
锁升级死锁 持有读锁时尝试获取写锁 先释放读锁,再获取写锁
锁降级风险 持有写锁时直接降级为读锁 可能导致数据不一致
未初始化 使用未初始化的读写锁 确保调用pthread_rwlock_init
未销毁 程序结束时未销毁读写锁 使用pthread_rwlock_destroy

12.4.2 自旋锁注意事项

问题 描述 解决方案
CPU浪费 长时间自旋导致CPU空转 确保临界区足够短
死锁 在单CPU系统上自旋时被中断 中断处理函数中使用自旋锁时需禁用中断
优先级反转 低优先级线程持有自旋锁,高优先级线程自旋 使用优先级继承机制
不可嵌套 自旋锁不支持递归获取 使用pthread_spin_lock前检查锁状态
中断上下文 中断处理函数不能使用互斥锁 必须使用自旋锁

12.4.3 嵌入式场景最佳实践

中断处理:

// 中断处理函数中使用自旋锁
void irq_handler(void) {
    // 禁用本地中断(如果是单核系统)
    // spin_lock_irqsave(&lock, flags);

    pthread_spin_lock(&spinlock);
    // 访问共享数据
    pthread_spin_unlock(&spinlock);

    // 恢复中断状态
    // spin_unlock_irqrestore(&lock, flags);
}

配置数据:

// 读多写少场景使用读写锁
const char *get_config(const char *key) {
    pthread_rwlock_rdlock(&config_lock);
    // 读取配置
    pthread_rwlock_unlock(&config_lock);
    return value;
}

void update_config(const char *key, const char *value) {
    pthread_rwlock_wrlock(&config_lock);
    // 更新配置
    pthread_rwlock_unlock(&config_lock);
}

12.5 面试要点

Q1: 读写锁和互斥锁的区别是什么?各适用于什么场景?

答:

  • 互斥锁:任意时刻只有一个线程可以持有锁,适用于读写都需要互斥的场景
  • 读写锁:允许多个读线程同时持有锁,但写线程必须独占,适用于读多写少的场景
  • 选择原则
    • 读写比例高(>10:1)→ 读写锁
    • 读写比例低 → 互斥锁
    • 写操作频繁 → 互斥锁

Q2: 读写锁的读优先和写优先有什么区别?

答:

  • 读优先:读线程持有锁时,新读线程可以立即获取锁,写线程可能饥饿
  • 写优先:写线程等待时,新读线程必须排队,写线程不会饥饿
  • Linux实现:默认读优先,可通过pthread_rwlockattr_setkind_np()设置写优先

Q3: 自旋锁和互斥锁的区别是什么?

答:

特性 自旋锁 互斥锁
等待方式 忙等待(占用CPU) 睡眠等待(不占用CPU)
上下文切换
适用场景 锁持有时间极短 锁持有时间较长
单CPU 不适合 适合
中断上下文 可以使用 不能使用

Q4: 为什么中断处理函数中必须使用自旋锁?

答:

  • 中断处理函数不能睡眠,否则会导致系统崩溃
  • 互斥锁在获取失败时会使线程睡眠
  • 自旋锁在获取失败时只是忙等待,不会睡眠
  • 因此中断处理函数中必须使用自旋锁

Q5: 如何避免读写锁的写线程饥饿?

答:

  • 设置写优先属性:pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP)
  • 使用公平性调度策略
  • 在应用层实现写线程优先级提升

Q6: 自旋锁在什么情况下会导致性能下降?

答:

  • 锁持有时间过长:自旋等待时间超过上下文切换时间
  • 单CPU系统:自旋线程无法被调度,持锁线程无法释放锁
  • 高竞争场景:多个线程同时自旋,浪费CPU时间
  • 解决方案:评估临界区执行时间,选择合适的锁类型

Q7: 读写锁的锁升级/降级问题如何处理?

答:

  • 锁升级:持有读锁时尝试获取写锁,会导致死锁
    • 解决方案:先释放读锁,再获取写锁
  • 锁降级:持有写锁时直接获取读锁,可能释放后数据被修改
    • 解决方案:先获取读锁,再释放写锁(如果支持)
  • 最佳实践:避免锁升级/降级,重新设计锁的粒度