《Operating System：Three Easy Pieces》第二十九章基于锁的并发数据结构

通过锁可以使数据结构线程安全（thread safe）。

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

并发计数器

计数器是最简单的一种数据结构，使用广泛而且接口简单。

c
1 typedef struct counter_t {
2     int value;
3 } counter_t;
4
5 void init(counter_t *c) {
6     c->value = 0;
7 }
8
9 void increment(counter_t *c) {
10     c->value++;
11 }
12
13 void decrement(counter_t *c) {
14     c->value--;
15 }
16
17 int get(counter_t *c) {
18     return c->value;
19 }

简单但无法扩展

如何让这段代码线程安全（thread safe）：

c
1 typedef struct counter_t {
2     int value;
3     pthread_mutex_t lock;
4 } counter_t;
5
6 void init(counter_t *c) {
7     c->value = 0;
8     Pthread_mutex_init(&c->lock, NULL);
9 }
10
11 void increment(counter_t *c) {
12     Pthread_mutex_lock(&c->lock);
13     c->value++;
14     Pthread_mutex_unlock(&c->lock);
15 }
16
17 void decrement(counter_t *c) {
18     Pthread_mutex_lock(&c->lock);
19     c->value--;
20     Pthread_mutex_unlock(&c->lock);
21 }
22
23 int get(counter_t *c) {
24     Pthread_mutex_lock(&c->lock);
25     int rc = c->value;
26     Pthread_mutex_unlock(&c->lock);
27     return rc;
28 }

这个并发计数器简单、正确。实际上，它遵循了最简单、最基本的并发数据结构中常见的数据模式：它只是加了一把锁，在调用函数操作该数据结构时获取锁，从调用返回时释放锁。现在，有了一个并发数据结构，问题可能就是性能了。如果这个结构导致运行速度太慢，那么除了简单加锁，还需要进行优化。

可以看出，同步的计数器扩展性不好。线程更多时，性能更差。

这样做在多CPU环境下性能很差。因为这种锁导致了多 CPU 情况下也只允许一个线程在运行，其他都在自旋等待，没发挥出多 CPU 的优势。

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

可扩展的计数

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

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

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

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

c
typedef struct counter_t
{
    int global;   // global count
    pthread_mutex_t glock;  // global lock
    int local[NUMCPUS];   //    local count (per cpu)
    pthread_mutex_t llock[NUMCPUS];// ... and locks
    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 > 0if (c->local[threadID]>= c->threshold)
    {// transfer to globalpthread_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!
}

并发链表

其实就是在对链表的头指针操作时加上锁

c
// basic node structure
typedef struct node_t
{
    int key;
    struct node_t* next;
} node_t;

// basic list structure (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()是线程安全的，每个线程都可以调用它，不需要担心竞争条件和其他并发缺陷。只有在更新共享列表时需要持有锁。下面展示了这些修改的细节。

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

c
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）：

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

它增加了链表操作的并发程度。但是实际上，在遍历的时候，每个结点获取锁、释放锁的开销巨大，很难比单锁的方法快。

并发队列

c
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)
{
// tmp是假结点
    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 使用了一个技巧，添加了一个假节点（在队列初始化的代码里分配的）。该假节点分开了头和尾操作。

并发散列表

c
#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);
}

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

这个简单的并发散列表扩展性（性能）极好，相较于单纯的链表。原因就是缩小了临界区，减少了不同线程间操作的冲突

小结

控制流变化时注意获取锁和释放锁；增加并发不一定能提高性能；有性能问题的时候再做优化。

避免不成熟的优化（premature optimization，对于所有关心性能的开发者都有用。我们让整个应用的某一小部分变快，却没有提高整体性能，其实没有价值。

参考

Operating Systems: Three Easy Pieces 中文版

《Operating System：Three Easy Pieces》第二十九章 基于锁的并发数据结构