聊聊那些锁事

前言

最近看了一本书，名字叫做《Operating Systems: Three Easy Pieces》，它的中文版是《操作系统导论》，原书在豆瓣评分9.7分，质量还不错。该书围绕 虚拟化、并发和持久性 这三个主要概念展开，行文诙谐幽默却又鞭辟入里，不同于寻常的操作系统书籍。这些天看了并发的几个章节，我主要关注了”锁”的部分，细读下来，有了更深刻的认识。

所以这篇文章就是对《操作系统导论》中讲解锁的章节的一个 读书笔记 ，实在是忍不住想分享出来。

锁的基本思想

锁其实是一个变量，我们需要声明某种类型的锁变量才能使用，比如下例：

    lock_t mutex; //声明
    ...
    lock(&mutex); //加锁
    balance = balance + 1;
    unlock(&mutex); //解锁

锁变量保存了锁在某一时刻的状态，它要么是可用的（avaliable,或unlocked，或free），表示没有线程持有锁，要么是被占用的（acquired，或locked,或held），表示有一个线程持有锁，正处于临界区。

锁为程序员提供了最小程度的调度控制，线程可以视为程序员创建的实体，但是被操作系统调度，具体方式由操作系统选择，而锁让程序员获得一些控制权，通过给临界区加锁，可以保证临界区内只有一个线程活跃。

此外，POSIX库将锁称为 互斥量（mutex） ，因为它被用来提供线程之间的互斥，即当一个线程在临界区，它能够阻止其他线程进入直到本线程离开临界区。

如何实现一个锁

我们已经从程序员的角度，对锁如何工作有一定的理解。那如何实现一个锁呢？我们需要什么硬件支持？需要什么操作系统的支持？下面会进行解答。

而在实现锁之前，我们还需要明确目标，需要设立一些标准才能让“锁”工作的好，主要有3个标准：

• 提供互斥 ：最基本的，锁应该能够阻止多个线程进入临界区

• 公平性 ：当锁可用时，是否每一个竞争线程有公平的机会抢到锁？

• 性能 ： 需要考虑使用锁之后增加的时间开销。比如以下几种场景：

  • 没有竞争的情况下，即只有一个线程抢锁、释放锁的开支如何？
  • 一个CPU上多个线程竞争，性能如何？
  • 多个CPU、多个线程竞争时的性能？

这三个标准映射到Java层面来说：

• 互斥锁 ： JDK中的synchronized和JUC中的Lock就是互斥锁，保证一次最多只能由一个线程持有该锁
• 公平锁与非公平锁 ： Java中的公平锁是指多个线程在等待同一个锁时，必须按照申请锁的先后顺序来获取锁，而非公平锁是可以抢占的，公平锁可以使用new ReentrantLock(true)实现
• 锁的性能 ： JDK中的synchronized的锁升级——偏向锁、轻量级锁和重量级锁，这几个锁的实现与转换，就是为了提升synchronized锁的性能

测试并设置指令(test-and-set instruction)

测试并设置指令 (test-and-set instruction) ，在x86系统上，具体是指xchg (atomic exchange，原子交换) 指令，我们用如下的C代码片段来定义测试并设置指令做了什么：

    int TestAndSet(int *old_ptr,int new) {
        int old = *old_ptr; //fetch old value at old_ptr
        *old_ptr = new; //store 'new' into old_ptr
        return old;  //return the old value
    }

它返回 old_ptr 指向的旧值，同时更新为 new 的新值。同时需要注意的是，上述的伪代码是为了说明使用，直接传统编译肯定不能保证原子性，需要操作系统的硬件指令 ( x86的xchg指令 ) 支持以保证原子性。

因为既可以测试旧值，又可以设置新值，所以把这条指令叫作 “测试并设置”。依靠这一条指令完全可以实现一个简单的 自旋锁 (spin lock) ，代码如下：

    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)
        {
            ; // 自旋等待 (do something)
        }
    }
    
    void unlock(lock_t *lock) {
        lock->flag = 0;
    }

首先假设一个线程在运行，调用lock()，没有其他线程持有锁，所以flag = 0,当调用 TestAndSet(flag,1) 方法，返回0，线程会跳出 while 循环，获取锁。同时也还会原子的设置flag为1，标志锁已经被持有。当线程离开临界区，调用 unlock() 将 flag 清理为0。

依靠操作系统的硬件原语，将测试 (test 旧的锁值) 和设置 (set 新的值) 合并为一个原子操作之后，我们保证了只有一个线程能获取锁，这就实现了一个有效的互斥原语!

上面的代码也就是自旋锁 (spin lock) 的实现，这是最简单的一种锁，一直自旋，利用CPU周期，直到锁可用。现在按照之前的标准来评价基本的自旋锁：

• 能够互斥：自旋锁一次只允许一个线程进入临界区，能够保证锁的正确性
• 公平性: 自旋锁不提供任务公平性保证，实际上，自旋的线程在竞争条件下可能会永远自旋。自旋锁没有公平性，可能会导致饿死。
• 性能问题： 对于自旋锁，在单CPU的情况下，性能开销非常大，多个线程在竞争锁，放弃CPU之前，都会自旋一个时间片，浪费CPU周期。而在多CPU上，自旋锁性能不错（如果线程数大致等于CPU数），假设线程A在CPU1，线程B在CPU2竞争同一个锁。线程A（CPU1）占有锁时，线程B竞争锁就会自旋 （在CPU2上）。然而，临界区一般都很短，因此很快锁就可用，然后线程B获得锁

比较并交换 (compare-and-swap)

某些系统提供了另一个硬件原语，即比较并交换指令 (SPARC系统是compare-and-swap，x86系统是compare-and-exchange)，下面是这条指令的C语言伪代码。

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

比较并交换的基本思路是检测 ptr 指向的值是否和 expected 相等；如果是，更新 ptr 所指的值为新值。否则，什么也不做。不论哪种情况，都返回该内存地址的值

有了比较并交换指令，就可以实现一个锁，类似于用测试并设置那样。例如，我们只需要用下面的代码替换lock()函数：

    void lock(lock_t *lock) {
        while (CompareAndSwap(&lock->flag,0,1) == 1)
            ; //spin    
    }

比较并交换指令实际上就是 CAS 指令，在Java开发工作中，我们也会常常遇到，比如使用原子类AtomicXXX，内部实现就是使用了CAS操作。

获取并增加 (fetch-and-add)

还有一个硬件原语是： 获取并增加 (fetch-and-add) 指令，它能原子地返回特定地址的旧值，并且让该值自增1，在x86系统中，是xadd指令。获取并增加的C语言伪代码如下：

    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) 
            ;//spin
    }
    void unlock(lock_t *lock) {
        FetchAndAdd(&lock->turn);
    }

在这个例子中，我们用获取并增加指令，实现了一个ticket锁，其实就是一个公平锁。

该实现不是利用一个值，而是使用了ticket 和 turn 两个变量来构建。基本思想是：如果线程希望获得锁，首先对一个ticket值执行一个原子的获取并增加指令。这个值作为该线程的”turn”(即myturn,为该线程设定获取锁的顺序)。根据全局共享的lock->turn 变量，当某一个线程的 myturn == turn 时，则轮到这个线程进入临界区。unlock 则是增加 turn,从而下一个等待线程可以进入临界区。

本方法能够保证所有线程都能获到锁，而且是按线程来的先后顺序，也就是一种先进先出 (FIFO) 的公平性机制，该锁还有一种说法，叫做 排号自旋锁 (Ticket Lock)

如何避免过多的自旋

前面几种指令都可以实现自旋锁，其实现也是非常简单，但是自旋过多，怎么办呢？比如以两个线程运行在单处理器为例，当一个线程（线程0）持有锁时，被中断。第二个线程（线程1）去获取锁，发现锁已经被持有。因此，它就开始自旋，接着自旋。如果线程0长时间持有锁，那么线程2会一直自旋，浪费CPU时间，所以如何让锁减少不必要地自旋？

只有硬件支持是不够的，我们还需要操作系统支持！

使用队列：休眠替代自旋

我们可以利用 Solaris 提供的支持，它提供了两个调用：

• park() 能够让调用线程休眠
• unpark(threadID) 则会唤醒threadID 标识的线程

可以用这两个调用来实现锁，让调用者在获取不到锁时睡眠，在锁可用时被唤醒。下面是C语言实现的伪代码。

    typedef struct lock_t {
        int flag;
        int guard;
        queue_t *q;
    } lock_t;
    
    void lock_init(lock_t *m) {
        m->flag = 0;
        m->guard = 0;
        queueu_init(m->q);
    }
    
    void lock(lock_t *m) {
        while (TestAndSet(&m->guard,1) == 1)
           ; //acquire guard lock by spinning
        if (m->flag == 0) {
            m->flag = 1; //lock is acquired
            m->guard = 0;
        } else {
            queue_add(m->q,gettid());
            m->guard = 0;
            park();
        }
    }
    
    void unlock(lock_t *m) {
        while (TestAndSet(&m->guard,1) == 1)
            ; // acquire guard lock by spinning
        if (queue_empty(m->q)) {
            m->flag = 0; // let go of lock;on one wants it;
        } else {
            unpark(queue_remove(m->q)) ;//hold lock(for next thread!)
        }
        m->guard = 0;
    }

我们将之前的测试并设置和等待队列结合，实现了一个更高性能的锁，guard基本上起到了自旋锁的作用。其次，我们通过队列来控制谁会获得锁，避免饿死。

Java中的自旋锁

长时间的自旋等待会消耗处理器时间。对此，Java中的自旋锁也有一定的处理措施：自旋等待的时间必须要有一定的限度，如果自旋超过了限定次数没有成功获得锁，就应当挂起线程，限定次数默认为10次，可以使用 -XX:PreBlockSpin来更改。

自旋锁在JDK1.4.2中引入，使用-XX:+UseSpinning来开启，其实现原理就是之前提到的CAS。JDK 6中变为默认开启，并且引入了自适应的自旋锁（适应性自旋锁）。

自适应意味着自旋的时间（次数）不再固定，而是由前一次在同一个锁上的自旋时间及锁的拥有者的状态来决定。如果在同一个锁对象上，自旋等待刚刚成功获得过锁，并且持有锁的线程正在运行中，那么虚拟机就会认为这次自旋也是很有可能再次成功，进而它将允许自旋等待持续相对更长的时间

参考资料

• https://book.douban.com/subject/19973015//
• https://tech.meituan.com/2018/11/15/java-lock.html