线程数增加致读写锁实现的多线程链表程序性能下降问题

问题背景 #

用读写锁实现多线程安全链表，功能正常但性能不符合预期：

• 10万次操作、1线程：耗时6.59秒
• 10万次操作、2线程：耗时6.68秒
• 10万次操作、4线程：耗时6.69秒

多线程版本未达串行性能，甚至略有下降；调整操作顺序后问题仍存在。另外MacOS下无法运行操作数超过20万的程序。

相关代码 #

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

struct list_node_s {
    int data;
    struct list_node_s * next;
};

int Member(int value, struct list_node_s ** head);
int Insert(int value, struct list_node_s ** head);
int Delete(int value, struct list_node_s ** head);
void * threadOp(void * my_rank);

pthread_rwlock_t lock;
long n;
long thread_count;
struct list_node_s * head_p; 


int main(int argc, char ** argv){
    double i = 0;
    clock_t begin = clock();
    n = strtol(argv[1], NULL, 10);
    thread_count = strtol(argv[2], NULL, 10);
    pthread_t * thread_handles = malloc(sizeof(pthread_t) * thread_count);
    long thread;
    

    printf("Will create %ld threads \n", thread_count);
    pthread_rwlock_init(&lock, NULL);
    
    struct list_node_s head = {.data = 0, .next = NULL};
    head_p = &head;
    
    for (i = 1.0; i < n; i++){
        Insert(i, &head_p);
    }

    for (thread = 0; thread < thread_count; thread++)
        pthread_create(&thread_handles[thread], NULL, threadOp, (void *) thread);

    for (thread = 0; thread < thread_count; thread++)
        pthread_join(thread_handles[thread], NULL);

    pthread_rwlock_destroy(&lock);
    clock_t end = clock();

    double time_spent = (double) (end - begin) / (CLOCKS_PER_SEC);

    printf("Elapsed time %lf \n", time_spent);
}

int Member(int value, struct list_node_s ** head){
    struct list_node_s * curr_p = * head;
    
    // traverse list until end or we got to right value
    while(curr_p != NULL && curr_p -> data < value)
        curr_p = curr_p -> next;

    if (curr_p == NULL || curr_p -> data > value){
        return 0;
    }
    else{
        return 1;
    }
}

int Insert(int value, struct list_node_s ** head){

    struct list_node_s * curr_p = * head;
    struct list_node_s * pred_p = NULL; 
    struct list_node_s * temp_p;
    
    while (curr_p != NULL && curr_p -> data < value){
        pred_p = curr_p;
        curr_p = curr_p -> next;
    }
    if (curr_p == NULL || curr_p -> data > value){
        temp_p = malloc(sizeof(struct list_node_s));
        temp_p -> data = value;
        temp_p -> next = curr_p;
        if (pred_p == NULL)
            *head = temp_p;
        else
            pred_p -> next = temp_p;
        return 1;
    }
    else {
        return 0;
    }
}

int Delete(int value, struct list_node_s ** head){
    struct list_node_s * curr_p = * head;
    struct list_node_s * pred_p = NULL;

    while (curr_p != NULL && curr_p -> data < value){
        pred_p = curr_p;
        curr_p = curr_p -> next;
    }
    // value was found and curr_p is not null
    if (curr_p != NULL && curr_p -> data == value){
        // first element in list
        if (pred_p == NULL){ 
            * head = curr_p -> next;
            free(curr_p);
        } else{
            pred_p -> next = curr_p -> next;
            free(curr_p);
        }
        return 1;
    } 
    // value not in list 
    else{ 
        return 0;
    }
}


void * threadOp(void * my_rank){
    long thread = (long) my_rank;
    long local_n = n / thread_count;
    long a = local_n * thread;
    long b = a + local_n;

    printf("Thread %ld is executing\n", thread);
    // half of threads will execute n members
    for (long i = a; i < b; i++){
        pthread_rwlock_rdlock(&lock);
        Member(i % 15, &head_p);
        pthread_rwlock_unlock(&lock);
        if (i % 10000 == 0){ 
            // other half will execute n/4 Ins and n/4 Del
            pthread_rwlock_wrlock(&lock);
            Insert(i, &head_p);
            pthread_rwlock_unlock(&lock);
        }
        else if (i % 10002 == 0)
        { 
            // other half will execute n/4 Ins and n/4 Del
            pthread_rwlock_wrlock(&lock);
            Delete(i, &head_p);
            pthread_rwlock_unlock(&lock);
        }
    }
    return NULL;
}

核心问题分析 #

1. 全局锁粒度过大导致竞争加剧 #

当前用全局读写锁保护整个链表，带来两个关键问题：

• 读操作（Member）的加解锁开销远超过操作本身：因为Member只查询i%15，遍历最多15个节点，耗时极短，但每次都要申请/释放读锁，多线程下锁的上下文切换开销完全抵消了并行收益。
• 写操作完全阻塞所有读：只要有一个线程执行Insert/Delete，所有读线程必须等待，全局锁变成了事实上的串行锁，多线程并行性被彻底扼杀。

2. 操作特性放大锁开销 #

Member操作的查询范围被限制在15个固定值，遍历长度极短，锁操作的耗时占总耗时比例极高，多线程的并行优势根本无法覆盖这部分开销。

3. MacOS下的内存问题 #

20万次操作时程序崩溃，大概率是内存分配失败：初始化阶段循环调用Insert会分配n个节点，加上malloc的内存碎片和管理开销，可能触发系统内存限制；另外MacOS的pthread默认栈大小较小，若线程数较多也可能导致栈溢出，但核心原因是内存分配问题。

解决方案 #

1. 改用细粒度节点锁（Hand-over-Hand Locking） #

放弃全局锁，给每个节点单独加读写锁，遍历链表时采用"手递手"方式锁节点——锁定下一个节点后再解锁前一个节点，这样读操作可以并行遍历不同节点段，写操作只需要锁定目标节点和前驱节点，大幅降低锁竞争。

修改节点结构

struct list_node_s {
    int data;
    struct list_node_s * next;
    pthread_rwlock_t lock;  // 每个节点的读写锁
};

调整Member操作（手递手读锁）

int Member(int value, struct list_node_s ** head) {
    struct list_node_s *curr_p = *head;
    struct list_node_s *prev_p = NULL;

    // 锁定头节点
    pthread_rwlock_rdlock(&curr_p->lock);
    while (curr_p != NULL && curr_p->data < value) {
        prev_p = curr_p;
        curr_p = curr_p->next;
        if (curr_p != NULL) {
            pthread_rwlock_rdlock(&curr_p->lock);  // 锁定下一个节点
        }
        pthread_rwlock_unlock(&prev_p->lock);      // 解锁前一个节点
    }

    int result = 0;
    if (curr_p != NULL && curr_p->data == value) {
        result = 1;
    }

    if (curr_p != NULL) {
        pthread_rwlock_unlock(&curr_p->lock);
    }
    return result;
}

Insert和Delete操作需用写锁锁定前驱和当前节点，确保修改时的原子性，逻辑类似。

2. 优化锁使用与操作比例 #

• 减少锁频率：读操作占比极高时，可采用乐观读——先无锁遍历找到目标节点，再加锁验证节点是否被修改，未修改则返回结果，否则重试，避免频繁加锁。
• 调整读写比例：进一步降低写操作频率，让读写锁的并发读优势得以体现；或者增加读操作的遍历长度，让锁开销占比降低。

3. 解决MacOS内存问题 #

• 预分配内存池：初始化时一次性分配n个节点的内存，再逐个初始化，减少malloc的开销和内存碎片。
• 增加内存检查：在Insert中添加malloc失败的判断，避免程序崩溃：

temp_p = malloc(sizeof(struct list_node_s));
if (temp_p == NULL) {
    fprintf(stderr, "内存分配失败\n");
    exit(EXIT_FAILURE);
}

4. 其他细节优化 #

• 移除threadOp中的printf：多线程下IO竞争会额外增加耗时，调试完成后可注释该语句。
• 改用墙钟计时：clock_t统计的是CPU总时间（所有线程累加），无法准确反映实际运行时间，建议用clock_gettime(CLOCK_MONOTONIC, ...)统计墙钟时间。

内容的提问来源于stack exchange，提问作者Luca Mautino