聊聊那些"锁"事

前言

最近看了一本书，名字叫作《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中变为默认开启，而且引入了自适应的自旋锁（适应性自旋锁）。

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

参考资料

• book.douban.com/subject/199…
• tech.meituan.com/2018/11/15/…