在谈锁 - 简书

如果要深刻理解锁，我们需要知道锁的实现的细节。哪些方便需要硬件支持，哪些方便需要操作系统支持。

几方面去评估锁

在构建锁之前，我们首先需要知道我们的目标是什么。然后才可以实现一把有效的锁。下面列举几个条件，满足下列条件才可以称为是有效的锁。

最基本的互斥功能。一旦一个线程锁住后，都要阻止其他线程进入临界区。
公平性。但锁是空闲状态时，线程需要可以公平的去竞争这把锁。不会有线程在极端情况下，永远获取不到锁，会饿死。
高效性。加/解锁过程需要消耗很少的时间。这里也要区分几个情况的性能情况，单线程单核心，多线程单核心以及多线程多核心。

早期单核时代的解决方案是通过中断来实现锁的。伪代码如下：

void lock(){
  //关闭中断
}
void unlock(){
  //开启中断
}

一个线程关闭中断，执行临界区代码，执行完毕后再次打开中断。在打开中断前，其他线程是无法进入这个临界区的。

The short-term scheduler (also known as the CPU scheduler) decides which of the ready, in-memory processes is to be executed (allocated a CPU) after a clock interrupt, an I/O interrupt, an operating system call or another form of signal.

关闭中断后，会一直执行下去，直到执行结束。这也带来了几个负面因素。

一旦临界区里是个死循环，操作系统会卡死，只能重启系统了。
这个也不合适多线程在多核系统中，一个线程不论是否打开中断，不影响另外一个核去执行另外一个线程。
关闭中断一段时间，可能导致这个中断永远丢失。比如关闭中断读操作完成了，cpu缺丢失了这个中断，那操作系统也不会去唤醒正在等待这个中断完成的线程了。
打开和关闭中断本身也不是高效的操作。对比前3点，这点确实无足轻重的了。

spinLock

通过中断实现锁，在多核系统不现实后。后面有人通过一些硬件提供的特性来实现锁。test-and-set或者atomic exchange指令，在x86中就是原子的xchg指令。这个通过访问这个内存地址期间，锁定系统总线的方式来实现原子操作。需要注意的是如果不是抢占式的调度模式，在单核系统下使用自旋锁需要格外注意。

int lock;
void spinlock_init() {
    lock = 0;
}
void spinlock_lock() {
    while (__sync_lock_test_and_set(&lock,1)) {
    }
}
void spinlock_unlock() {
    __sync_lock_release(&lock);
}

我们有了自旋锁的实现，来谈谈以上实现锁的3个目标是否都满足。

首先看看锁是否满足互斥性。这个毫无疑问的。一个线程进入临界区后，其他线程无法进入。
公平性。spinlock不提供任何公平性的保证。极端情况下会导致有线程饿死的情况
性能。在单核系统中，spinlock是自旋时是白白消耗cpu时间片的，这个情况很糟糕。在多核系统中，spinlock的消耗是可以接受的了，这个时候拿到锁的线程正在执行临界区的代码，其他线程不论自旋等待还是挂起等待对整个程序的总的执行时间没有差别的。这个时候我们可以认为是高效的。

实现spinLock不仅可以用TestAndSet而且可以使用CompareAndSwap，而且CompareAndSwap会更高效一点。因为CompareAndSwap只有在对比两个值相同时才会写入。

  //cas版本的lock
void spinlock_lock() {
    while (__sync_bool_compare_and_swap(&lock,0,1) == 1) {
    }
}

还有其他锁的写法，通过__sync_fetch_and_add指令，

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) {
  lock->turn = lock->turn + 1;
}

在测试中发现，ticket-spinlock有个严重的问题，当线程数大于核心数的时候，很少几率解锁的情况。因为默认线程调度策略为SCHED_FIFO，线程占用cpu占用cpu会一直运行，直到有更高优先级线程到达或者自己主动交出cpu。所以很大概率会出现拿到锁的线程最后饿死的情况，没机会解锁。而其他的spinlock并没有这个情况因为原子操作本身没完成前会交出cpu时间片的。

实现了几种不同方式的spinlock的测试例子 https://github.com/iamjokerjun/test/tree/master/spinTest

spinlock存在的问题

当临界区内的发生上下文切换时怎么办，比如中断，或者本身消耗完了cpu时间片。这个时候其他线程会空自旋的，一直要等待拿到自旋锁的线程再次运行。
我们很容易想到的是，在自旋过程中交出剩下的时间片，使用yield。伪代码如下：

void lock(){
  while(TestAndSet(&lock,1) == 1)
    yield();//放弃当前时间片
}

yield本质上把线程从running状态切换到ready状态。
假设我们有2个线程在一个核上工作，这种方式可以处理的很好。一个线程拿到锁后，另外一个线程自旋时会放弃当前时间片，然后拿到锁的线程会继续完成剩下的工作。这种模式下可以工作的很好。
在假设有100个线程都在重复的去竞争这把锁，一旦一个线程拿到了锁，剩下的99个线程调用lock，发现锁被占用了，于是让出cpu。假设调度模式是某种循环执行的模式，那么在拿到锁的线程再次运行前，剩下的99个线程会执行，运行-->yield这种这种模式。虽然比自旋模式（浪费99个时间片）要好，这种模式仍然有较大消耗（上下文切换也有大量消耗）。
更严重的是，我们仍然没有解决线程饿死的问题。一个线程可能永远拿不到锁，其他的线程可能会一直拿到锁。一般使用自旋锁的场景下，多个线程一定要是优先级相同的情况。否则很大概率出现线程饿死的情况。

使用队列

上述我们用的spinlock最大的问题，是调度器来决定哪个线程会继续执行，这个就不可能能保证锁的公平性。不论是自旋等待，还是yield让出当前时间片。不论哪种方式都不能本质上避免线程饿死的情况。
下面用队列的方式实现的锁。

typedef struct queueLock {
    int flag;
    int guard;
    queue *q;
}Lock;

void lockInit(Lock* m){
    m->flag = 0;    
    m->guard = 0;   
    queueInit(m->q);
}

void lock(Lock *m){
    while(TestAndSet(&m->guard,1) == 1); //获取guard资源需要自旋获取
    if(m->flag == 0){       //锁没人用
        m->flag = 1;        //锁被获取了
        m->guard = 0;       //释放guard
    }
    else{
        queueAdd(m->q,gettid());    //gettid获取线程ID,把线程Id加入队列
        m->guard = 0;               //释放guard
        park();                     //让当前线程sleep
    }
}

void unlock(Lock*m){
    while(TestAndSet(&m->guard,1) == 1);//自旋获取guard资源
    if(queueEmpty(m->q))                //如果队列为空，说明没有任何线程需要锁
        m->flag = 0;                    //释放lock
    else
        unpark(queueRemove(m->q));      //从队列中移出元素，唤醒队列中最先如队列的线程 (队列先进先出保证了顺序)
    m->guard = 0;                       //释放guard资源
}

我们来分析分析基于队列的锁的代码。这个基于最原始的TAS的方式以及使用显式队列的方式制作的一把更高效的锁，我们使用队列保证了公平性，避免有线程被饿死的情况。
我们看到guard还是自旋的方式来获取，这种方式不能避免完全避免自旋，一个线程在lock/unlock时中断了，会导致其他正在调用lock/unlock的线程自旋下去。然而这种时间浪费是非常有限的，相比用户的锁定的临界区代码来说，这个临界区只是很少一些指令。
当一个线程不能获取锁时，首先把当前线程id加入队列，其次释放guard，再次sleep当前线程。这个顺序一定要很小心的保证。如果线程id和释放guard颠倒次序，就不能保证公平性了。比如flag资源被其他线程获取了。thread A调用lock时，释放guard后。接着thread B调用lock，执行到queueAdd，这样threadB的tId优先于threadA的tId放入队列了。
我们还注意到当一个线程unlock时，flag并没有被设置成0。当一个线程wake up，假设从park函数中醒过来，然而他并没有获取guard资源，所以不能设置flag为0。所以这个角度上看，锁只是从一个地方线程传递到另外一个线程，所以flag并没有设置为0.
最后我们来感知下锁竞争的问题。lock过程中在调用park之前，当发生时间转换时，拥有锁的线程释放锁了，调用unpark了，然后第一个调用lock的线程在调用park，于是这个线程会无限睡眠下去了。
Solaris解决这个问题的方法是引入第三个系统调用函数setpark,但是当中断发生后，其他线程在本线程调用park前调用unpark。随后的这个park立马返回而不是陷入睡眠。怀疑应该是用引用计数之类的办法来管理的。

 queueAdd(m->q,gettid());
 setpark();
 m->guard = 0;

另一个不同的解决方案是把这些代码传入内核，内核提供有一些原子的方法来释放锁和原子进出队列。

不同操作系统对锁的实现

不同操作系统提供的不同锁的实现细节。下面列举下linux的futex lock实现。

void mutex_lock (int *mutex) {
  int v;
  /* Bit 31 was clear, we got the mutex (the fastpath) */
  if (atomic_bit_test_set (mutex, 31) == 0)
    return;
  atomic_increment (mutex);
  while (1) {
        if (atomic_bit_test_set (mutex, 31) == 0) {
            atomic_decrement (mutex);
            return;
        }
        /* We have to waitFirst make sure the futex value
        we are monitoring is truly negative (locked). */
        v = *mutex;
        if (v >= 0)
            continue;
        futex_wait (mutex, v);
    }
}

void mutex_unlock (int *mutex) {
    /* Adding 0x80000000 to counter results in 0 if and
    only if there are not other interested threads */
    if (atomic_add_zero (mutex, 0x80000000))
        return;
    /* There are other threads waiting for this mutex,
    wake one of them up. */
    futex_wake (mutex);
}

下面我们来分析下这段代码，lock时if (atomic_bit_test_set (mutex, 31) == 0) return;这句主要为了一个线程时的优化。
我们来假设一种场景，当unlock执行一半时，也就是在 futex_wake执行之前，处于等待的线程都有机会拿到锁(不一定是是刚唤醒的线程)，这就可能导致刚唤醒的线程并没有拿到锁，而给正在调用lock的线程，这样的话刚wakeup的线程又重新sleep，同时从队列头加入队列尾了，导致饿死。lock代码选择2次校验的方式，第一次校验 v = *mutex;前已经unlock了，第二次校验防止 v = *mutex;后被unlock。
假设线程A拿到锁了，等待队列中无线程，线程B申请拿锁，执行到v = *mutex前，切好线程A执行unlock并执行完，这个时候线程B陷入sleep并且再也不会有机会wakeup。所以加上 if (v >= 0) continue;进行二次校验。
缺点：假设线程A拿到锁，线程B处于等待队列中，线程C申请拿锁，在执行到上v = *mutex前，线程A执行unlock并执行完。这样线程B和线程C都走到while里的 if (atomic_bit_test_set (mutex, 31) == 0) 这就有可能导致线程C拿到锁，线程B又进入sleep了。（这个也看操作系统的调度了，如果刚wakeup的线程能优先执行就能避免问题了）

unlock函数里的锁的释放和wakeup线程不是原子性的。运气不好的情况，wake之前，锁已经被其他线程拿走了。

尤此可见，mutex性能不低的，除非锁定一些“极小”的临界区，大部分时候mutex足够的了。

参考资料
https://en.wikipedia.org/wiki/Scheduling_(computing)
https://geidav.wordpress.com/2016/04/09/the-ticket-spinlock/
https://inst.eecs.berkeley.edu/~cs162/sp03/Lectures/L07.pdf
https://www.math.uni-hamburg.de/doc/java/tutorial/essential/threads/priority.html
https://woboq.com/blog/introduction-to-lockfree-programming.html