---

title: 锁的底层实现原理
date: 2021-06-08 17:00:00
categories: 技术杂文
tags:
- 技术杂文

---

锁的基本思想

在并发编程中的一个基本问题是，希望原子执行一系列命令，但是由于单处理器上的中断，或者多处理器上的多线程并发执行，需要锁机制来解决这样的问题。

使用锁的方式如下：

lock_t mutex;
...
lock(&mutex);
/* 临界区代码 */
unlock(&mutex);

lock与unlock函数的语义比较简单，调用lock尝试获取锁，如果没有其他的线程持有锁，则该锁是可用的，该线程会获得锁，从而进入临界区，这个线程被称为锁的持有者。

如果另外一个线程对相同的锁变量调用lock，因为锁被别的线程所持有，该调用不会返回，这个线程就无法进入临界区，直到锁的持有者释放了锁为止。

评价锁的实现机制

如何评价一种锁的实现机制的好坏？可以制定以下的一些标准：

• 互斥性，即锁是否有效，是否能够阻止多线程进入临界区；
• 公平性，即当锁可用时，是否每一个竞争线程有公平的机会抢到锁，从另一方面看，是否有竞争锁的线程会饿死；
• 性能，即使用锁之后增加的时间开销。

锁的实现机制

机制1：控制中断

在早期，实现互斥的解决方案，就是在临界区关闭中断，这是为单处理器系统开发的，实现的伪代码如下：

void lock()
{
    disableInterrupts();
}

void unlock()
{
    enableInterrupts();
}

关闭中断与打开中断，使用的是特权的硬件指令。通过在进入临界区之前关闭中断，可以保证临界区的代码不会被中断，从而原子地执行。临界区代码执行完成后，重新打开中断，程序继续正常运行。没有了中断，线程可以确信其代码会一直执行下去，不会被其他的线程干扰。

但是该方案的缺点很多：

• 该方案要求允许所有的线程都可以执行打开与关闭中断的特权操作，恶意的程序会滥用该机制，关闭中断后不在打开中断，从而独占处理器；
• 该方案不支持多处理器；
• 关闭中断会导致中断丢失，可能会导致非常严重的系统问题，比如磁盘IO操作完成，但是中断被关闭，CPU错过了这个事件，等待该IO操作的进程就无法被唤醒；
• 该方案效率较低，现代CPU对关闭与打开中断的代码执行较慢。

在计算机的发展历史上，出于某些原因，研究人员在早期热衷于研究不依赖硬件支持的锁机制，从后来的结果证明，这些工作没有太多的意义，只需要很少的硬件支持，就可以实现更好的锁机制。

机制2：测试并设置指令

锁的机制实现，需要硬件的支持。最简单的硬件支持是测试并设置指令（test-and-set instruction），也被称为原子交换（atomic exchange），实现的伪代码如下：

int TestAndSet(int *old_ptr, int new)
{
    int old = *old_ptr;
    *old_ptr = new;
    return old;
}

硬件指令，可以实现原子执行以上的一系列操作，可以根据这个硬件指令，实现一个简单的自旋锁：

typedef struct lock_t
{
    int flag;
}lock_t;

void init(lock_t *lock)
{
    //0表示锁是可用的，1表示锁被其他线程占用
    lock->flag = 0;
}

void lock(lock_t *lock)
{
    while(TestAndSet(&lock->flag, 1) == 1)
    {
        ; //自旋，等待
    }
}

void unlock(lock_t *lock)
{
    lock->flag = 0;
}

理解这个锁机制为什么能够工作，可以假设某个线程正在运行，调用lock后，没有其他的线程持有锁，flag值为0，调用TestAndSet后，指令会原子地将flag设置为1，并返回0，线程则跳出while循环。

当某个线程已经持有锁，此时flag值为1，本线程调用lock尝试获取锁，TestAndSet会返回1，则本线程会在while循环中自旋，直到另一个线程释放锁，即将flag置为0为止。

机制3：比较并交换指令

硬件支持的另一个原语，是比较并交换指令（compare-and-swap或者compare-and-exchange）。该指令的伪代码为：

int CompareAndSwap(int *ptr, int expected, int new)
{
    int actual = *ptr;
    if(actual == expected)
    {
        *ptr = new;
    }
    return actual;
}

比较并交换指令的基本思路是，检测ptr指向的值是否与expected相等，若相等，则更新ptr指向的值为新的值，否则，什么也不做，指令返回ptr原先指向的值。

使用比较并交换指令，实现的锁机制可以为：

typedef struct lock_t
{
    int flag;
}lock_t;

void init(lock_t *lock)
{
    //0表示锁是可用的，1表示锁被其他线程占用
    lock->flag = 0;
}

void lock(lock_t *lock)
{
    while(CompareAndSwap(&lock->flag, 0, 1) == 1)
    {
        ; //自旋，等待
    }
}

void unlock(lock_t *lock)
{
    lock->flag = 0;
}

该硬件指令实现的自旋锁，与机制2所实现的等价。CompareAndSwap(&lock->flag, 0, 1) 可以理解为，检查flag是否与0相等（0表示锁未被占用），若相等，则置flag为1，并返回flag的旧值0。

机制4：链接的加载和条件式存储指令

一些平台提供了链接的加载（load-linked）和条件式存储（store-conditional）指令，可以配合使用，来实现锁机制，指令的实现伪代码为：

int LoadLinked(int *ptr)
{
    return *ptr;
}

int StoreConditional(int *ptr, int value)
{
    if(no one has updated *ptr since the LoadLinked to this address)
    {
        *ptr = value;
        return 1; /* success */
    }
    else
    {
        return 0; /* failed to update */
    }
}

这里关键的是条件式存储指令，只有上一次加载的地址在期间都没有更新时，才会更新成功。

使用该指令实现的锁机制的伪代码为：

typedef struct lock_t
{
    int flag;
}lock_t;

void init(lock_t *lock)
{
    //0表示锁是可用的，1表示锁被其他线程占用
    lock->flag = 0;
}

void lock(lock_t *lock)
{
    while(1)
    {
        while(LoadLinked(&lock->flag) == 1)
        {
            ; /* 自旋，等待flag为0 */
        }

        if(StoreConditional(&lock->flag, 1) == 1)
        {
            return; /* 条件存储成功则获取锁，否则重新LoadLinked */
        }
    }
}

void unlock(lock_t *lock)
{
    lock->flag = 0;
}

这里需要关注条件式存储是如何失败的。一个线程调用了lock，执行链接的加载指令，返回0，表示当前锁可用，但是在执行条件式存储之前，被调度程序中断了，另一个线程占用了CPU，也调用了lock，也执行了链接的加载指令，同样此时锁是可用，返回0。

现在两个线程都执行了链接的加载指令，将要执行条件式存储指令，重点是只有一个线程能够成功更新flag为1。先执行条件式存储指令的线程将会成功获取锁，而后执行条件式存储指令的线程将失败。

机制5：获取并增加指令

另一个硬件原语是获取并增加指令（fetch-and-add），该指令能够原子地返回特定地址的旧值，并将该旧值增加1。指令的伪代码为：

int FetchAndAdd(int *ptr)
{
    int old = *ptr;
    *ptr = old + 1;
    return old;
}

根据获取并增加指令，实现的锁机制可以为：

typedef struct lock_t
{
    int ticket;
    int turn;
}lock_t;

void lock_init(lock_t *lock)
{
    lock->ticket = 0;
    lock->turn = 0;
}

void lock(lock_t *lock)
{
    int myturn = FetchAndAdd(&lock->ticket);
    while(lock->turn != myturn)
    {
        ; /* 自旋 */
    }
}

void unlock(lock_t *lock)
{
    FetchAndAdd(&lock->turn);
}