当前位置：首页 > news >正文

操作系统导论——第29章基于锁的并发数据结构

news 来源：原创 2025/7/10 10:40:08

通过锁可以使数据结构线程安全（thread safe）。当然，具体如何加锁决定了该数据结构的正确性和效率？挑战是：

关键问题：如何给数据结构加锁？

对于特定数据结构，如何加锁才能让该结构功能正确？进一步，如何对该数据结构加锁，能够保证高性能，让许多线程同时访问该结构，即并发访问（concurrently）？

一、并发计数器

计数器是最简单的一种数据结构，使用广泛而且接口简单。图29.1 中定义了一个非并发的计数器。

简单但无法扩展

可以看到，没有同步机制的计数器很简单，只需要很少代码就能实现。下一个挑战是：如何让这段代码线程安全（thread safe）？图29.2展示了我们的做法。

这个并发计数器简单、正确。实际上，它遵循了最简单、最基本的并发数据结构中常见的数据模式：它只是加了一把锁，在调用函数操作该数据结构时获取锁，从调用返回时释放锁。这种方式类似基于观察者（monitor）[BH73]的数据结构，在调用、退出对象方法时，会自动获取锁、释放锁。

现在有了一个并发数据结构，问题就是性能了。如果这个结构导致运行速度太慢，那么除了简单加锁，还需要进行优化。当然，如果数据结构导致的运行速度太慢，那就没事。

理解简单方法的性能成本，运行一个基准测试，每个线程更新同一个共享计数器固定次数，然后改变线程数。图29.3 给出了运行 1 个线程到 4 个线程的总耗时，其中每个线程更新 100 万次计数器。本实验是在 4 核 Intel 2.7GHz i5 CPU 的 iMac 上运行。通过增加 CPU，我们希望单位时间能够完成更多的任务。

从图29.3 上方的曲线（标为“精确”）可以看出，同步的计数器扩展性不好。单线程完成 100万次更新只需要很短的时间（大约0.03s），而两个线程并发执行，每个更新100万次，性能下降很多（超过5s！）。线程更多时，性能更差。

理想情况下，你会看到多处理上运行的多线程就像单线程一样快。达到这种状态称为完美扩展（perfect scaling）。虽然总工作量增多，但是并行执行后，完成任务的时间并没有增加。

可扩展的计数（懒惰计数器）

可扩展的计数器很重要，没有可扩展的计数，一些运行在 Linux 上的工作在多核机器上将遇到严重的扩展性问题。将介绍一种特定的方法，称作懒惰计数器（sloppy counter）。

懒惰计数器通过多个局部计数器和一个全局计数器来实现一个逻辑计数器，其中每个 CPU 核心有一个局部计数器。具体来说，在 4个CPU 的机器上，有 4 个局部计数器和 1 个全局计数器。除了这些计数器，还有锁：每个局部计数器有一个锁，全局计数器有一个。

懒惰计数器的基本思想：如果一个核心上的线程想增加计数器，那就增加它的局部计数器，访问这个局部计数器是通过对应的局部锁同步的。因为每个CPU有自己的局部计数器，不同CPU上的线程不会竞争，所以计数器的更新操作可扩展性好。

但是，为了保持全局计数器更新（以防某个线程要读取该值），局部值会定期转移给全局计数器，方法是获取全局锁，让全局计数器加上局部计数器的值，然后将局部计数器置零。

这种局部转全局的频度，取决于一个阈值，这里称为S（表示sloppiness）。S越小，懒惰计数器则越趋近于非扩展的计数器。S越大，扩展性越强，但是全局计数器与实际计数的偏差越大。我们可以抢占所有的局部锁和全局锁（以特定的顺序，避免死锁），以获得精确值，但这种方法没有扩展性。

为了弄清楚这一点，来看一个例子（见表 29.1）。在这个例子中，阈值 S 设置为 5，4 个CPU 上分别有一个线程更新局部计数器L1,…, L4。随着时间增加，全局计数器G的值也会记录下来。每一段时间，局部计数器可能会增加。如果局部计数值增加到阈值S，就把局部值转移到全局计数器，局部计数器清零。

图 29.3 中下方的线，是阈值S为 1024 时懒惰计数器的性能。性能很高，4个处理器更新400万次的时间和一个处理器更新100万次的几乎一样。

图 29.4 展示了阈值S的重要性，在 4 个CPU上的 4 个线程，分别增加计数器100万次。如果S 小，性能很差（但是全局计数器精确度高）。如果S大，性能很好，但是全局计数器会有延时。懒惰计数器就是在准确性和性能之间折中。

图 29.5 是这种懒惰计数器的基本实现。

typedef struct  counter_t { int             global;            // 全局计数器pthread_mutex_t glock;             // 全局锁 int             local[NUMCPUS];    // 局部计数器 (per cpu) pthread_mutex_t llock[NUMCPUS];    // 局部锁int             threshold;         // update frequency 
} counter_t; // init: record threshold, init locks, init values 
// of all local counts and global count 
void init(counter_t *c, int threshold) { c->threshold = threshold; c->global = 0; pthread_mutex_init(&c->glock,  NULL); int i; for (i = 0; i < NUMCPUS; i++) { c->local[i] = 0; pthread_mutex_init(&c->llock[i],  NULL); } 
} // update: usually, just grab local lock and update local amount 
//        once local count has risen by 'threshold', grab global 
//        lock and transfer local values to it 
void update(counter_t *c, int threadID, int amt) { pthread_mutex_lock(&c->llock[threadID]); c->local[threadID] += amt;               // assumes amt > 0 if (c->local[threadID] >= c->threshold) { // transfer to global pthread_mutex_lock(&c->glock); c->global += c->local[threadID]; pthread_mutex_unlock(&c->glock); c->local[threadID] = 0; } pthread_mutex_unlock(&c->llock[threadID]); 
} // get: just return global amount (which may not be perfect) 
int get(counter_t *c) { pthread_mutex_lock(&c->glock); int val = c->global; pthread_mutex_unlock(&c->glock); return val; // only approximate! 
}

二、并发列表

更复杂的数据结构——链表。同样从一个基础实现开始。简单起见，只关注链表的插入操作。图 29.6 展示了这个基本数据结构的代码。

// 链表指标
typedef struct  node_t { int                key; struct  node_t        *next; 
} node_t; // 链表结构 (one used per list) 
typedef struct  list_t { node_t                *head; pthread_mutex_t    lock; 
} list_t; // 初始化链表
void List_Init(list_t *L) { L->head = NULL;pthread_mutex_init(&L->lock,  NULL); 
} // 插入元素
int List_Insert(list_t *L, int key) { pthread_mutex_lock(&L->lock); node_t *new = malloc(sizeof(node_t)); if (new == NULL) { perror("malloc"); pthread_mutex_unlock(&L->lock); return -1; // fail } new->key = key; new->next = L->head; L->head = new; pthread_mutex_unlock(&L->lock); return 0; // success 
} // 查询元素
int List_Lookup(list_t *L, int key) { pthread_mutex_lock(&L->lock); node_t *curr = L->head; while (curr) { if (curr->key == key) { pthread_mutex_unlock(&L->lock); return 0; // success } curr = curr->next; 
} pthread_mutex_unlock(&L->lock); return -1; // failure 
}

从代码中可以看出，代码插入函数入口处获取锁，结束时释放锁。如果 malloc 失败（在极少的时候），会有一点小问题，在这种情况下，代码在插入失败之前，必须释放锁。

事实表明，这种异常控制流容易产生错误。挑战来了：能否重写插入和查找函数，既要保证并发插入的正确性，又要避免在插入操作失败的情况下，仍需调用释放锁的操作呢？

调整代码，让获取锁和释放锁只环绕插入代码的真正临界区，前面的方法有效是因为部分工作实际上不需要锁，假定 malloc() 是线程安全的，每个线程都可以调用它，不需要担心竞争条件和其他并发缺陷。只有在更新共享列表时需要持有锁。图 29.7 展示了这些修改细节。

对于查找函数，进行了简单的代码调整，跳出主查找循环，到单一的返回路径。这样做减少了代码中需要获取锁、释放锁的地方，降低了代码中不小心引入缺陷（诸如在返回前忘记释放锁）的可能性。

void List_Init(list_t *L) { L->head = NULL; pthread_mutex_init(&L->lock,  NULL); 
} void List_Insert(list_t *L, int key) { 
// synchronization not needed node_t *new = malloc(sizeof(node_t)); if (new == NULL) { perror("malloc"); return; } new->key = key; // just lock critical section pthread_mutex_lock(&L->lock);  // 这里设置锁new->next = L->head; L->head = new; pthread_mutex_unlock(&L->lock); 
} int List_Lookup(list_t *L, int key) { int rv = -1; pthread_mutex_lock(&L->lock); node_t *curr = L->head; while (curr) { if (curr->key == key) { rv = 0; break; } curr = curr->next; } pthread_mutex_unlock(&L->lock); 、return rv; // now both success and failure 
}

扩展链表

有了基本的并发链表，但又遇到了这个链表扩展性不好的问题。研究人员发现的增加链表并发的技术中，有一种叫作过手锁（hand-over-hand locking，也叫作锁耦合， lock coupling）。

原理很简单。每个节点都有一个锁，替代之前整个链表一个锁。遍历链表的时候，首先抢占下一个节点的锁，然后释放当前节点的锁。

从概念上说，过手锁链表有点道理，它增加了链表操作的并发程度。但是实际上，在遍历的时候，每个节点获取锁、释放锁的开销巨大，很难比单锁的方法快。即使有大量的线程和很大的链表，这种并发的方案也不一定会比单锁的方案快。也许某种杂合的方案（一定数量的节点用一个锁）值得去研究。

三、并发队列

总有一个标准的方法来创建一个并发数据结构：添加一把锁。对于一个队列，跳过这种方法

下列展示了用于该队列的数据结构和代码：

typedef struct  node_t {  // 定义链表int                 value; struct  node_t     *next; 
} node_t; typedef struct  queue_t {  // 队列头部 尾部node_t            *head; node_t            *tail; pthread_mutex_t    headLock; pthread_mutex_t    tailLock; 
} queue_t; void Queue_Init(queue_t *q) {  // 初始化队列node_t *tmp = malloc(sizeof(node_t)); tmp->next = NULL; q->head = q->tail = tmp; pthread_mutex_init(&q->headLock,  NULL); pthread_mutex_init(&q->tailLock,  NULL); 
} void Queue_Enqueue(queue_t *q, int value) { node_t *tmp = malloc(sizeof(node_t)); assert(tmp != NULL); tmp->value = value; tmp->next = NULL; pthread_mutex_lock(&q->tailLock); q->tail->next = tmp; q->tail = tmp; pthread_mutex_unlock(&q->tailLock); 
} int Queue_Dequeue(queue_t *q, int *value) { pthread_mutex_lock(&q->headLock); node_t *tmp = q->head; node_t *newHead = tmp->next; if (newHead == NULL) { pthread_mutex_unlock(&q->headLock); return -1; // queue was empty } *value = newHead->value; q->head = newHead; pthread_mutex_unlock(&q->headLock); free(tmp); return 0; 
}

仔细研究这段代码，会发现有两个锁，一个负责队列头，另一个负责队列尾。这两个锁使得入队列操作和出队列操作可以并发执行，因为入队列只访问 tail 锁，而出队列只访问head 锁。

Michael 和 Scott 使用了一个技巧，添加了一个假节点（在队列初始化的代码里分配的）。 该假节点分开了头和尾操作。

四、并发散列表

最后一个讨论散列表。只关注不需要调整大小的简单散列表。支持调整大小还需要一些工作

#define BUCKETS (101) typedef struct  hash_t { list_t lists[BUCKETS]; 
} hash_t; void Hash_Init(hash_t *H) { int i; for (i = 0; i < BUCKETS; i++) { List_Init(&H->lists[i]); } 
} int Hash_Insert(hash_t *H, int key) { int bucket = key % BUCKETS; return List_Insert(&H->lists[bucket], key); 
} int Hash_Lookup(hash_t *H, int key) { int bucket = key % BUCKETS; return List_Lookup(&H->lists[bucket], key); 
}

本例的散列表使用我们之前实现的并发链表，性能特别好。每个散列桶（每个桶都是一个链表）都有一个锁，而不是整个散列表只有一个锁，从而支持许多并发操作。

图 29.10 展示了并发更新下的散列表的性能（同样在4 CPU的 iMac，4个线程，每个线程分别执行1万～5万次并发更新）。同时，作为比较，我们也展示了单锁链表的性能。可以看出，这个简单的并发散列表扩展性极好，而链表则相反。

五、小结

我们已经介绍了一些并发数据结构，从计数器到链表队列，最后到大量使用的散列表

一、并发计数器

二、并发列表

三、并发队列

四、并发散列表

五、小结

相关文章：