关于信号量与线程互斥锁的区别与实现

本文转自

http://jimmee.iteye.com/blog/618740

之前一直没有怎么关注过这个问题，前些日子在面试一家公司的时候，面试官提到了pthread_cond_wait/pthread_cond_signal的实现，当时答的不是很好，回来就查了nptl的代码。前天，水木上又有人问到了信号量和互斥锁的问题，我想还是对它们的区别与实现总结一下。 

首先了解一些信号量和线程互斥锁的语义上的区别： 

>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 

援引CU上一篇帖子的内容： 
“信号量用在多线程多任务同步的，一个线程完成了某一个动作就通过信号量告诉别的线程，别的线程再进行某些动作（大家都在sem_wait的时候，就阻塞在那里）。而互斥锁是用在多线程多任务互斥的，一个线程占用了某一个资源，那么别的线程就无法访问，直到这个线程unlock，其他的线程才开始可以利用这个资源。比如对全局变量的访问，有时要加锁，操作完了，在解锁。有的时候锁和信号量会同时使用的” 
也就是说，信号量不一定是锁定某一个资源，而是流程上的概念，比如：有A,B两个线程，B线程要等A线程完成某一任务以后再进行自己下面的步骤，这个任务并不一定是锁定某一资源，还可以是进行一些计算或者数据处理之类。而线程互斥量则是“锁住某一资源”的概念，在锁定期间内，其他线程无法对被保护的数据进行操作。在有些情况下两者可以互换。 

两者之间的区别: 

作用域 
信号量: 进程间或线程间(linux仅线程间) 
互斥锁: 线程间 

上锁时 
信号量: 只要信号量的value大于0，其他线程就可以sem_wait成功，成功后信号量的value减一。若value值不大于0，则sem_wait阻塞，直到sem_post释放后value值加一。一句话，信号量的value>=0。 
互斥锁: 只要被锁住，其他任何线程都不可以访问被保护的资源。如果没有锁，获得资源成功，否则进行阻塞等待资源可用。一句话，线程互斥锁的vlaue可以为负数。 

<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< 

接下来，我们需要分析一下信号量和线程互斥锁的实现机制。 

在Linux下，信号量和线程互斥锁的实现都是通过futex系统调用。 

futex（快速用户区互斥的简称）是一个在Linux上实现锁定和构建高级抽象锁如信号量和POSIX互斥的基本工具。它们第一次出现在内核开发的2.5.7版；其语义在2.5.40固定下来，然后在2.6.x系列稳定版内核中出现。 

Futex 是fast userspace mutex的缩写，意思是快速用户空间互斥体。Linux内核把它们作为快速的用户空间的锁和信号量的预制构件提供给开发者。Futex非常基础，借助其自身的优异性能，构建更高级别的锁的抽象，如POSIX互斥体。大多数程序员并不需要直接使用Futex，它一般用来实现像NPTL这样的系统库。 

Futex 由一块能够被多个进程共享的内存空间（一个对齐后的整型变量）组成；这个整型变量的值能够通过汇编语言调用CPU提供的原子操作指令来增加或减少，并且一个进程可以等待直到那个值变成正数。Futex 的操作几乎全部在应用程序空间完成；只有当操作结果不一致从而需要仲裁时，才需要进入操作系统内核空间执行。这种机制允许使用 futex 的锁定原语有非常高的执行效率：由于绝大多数的操作并不需要在多个进程之间进行仲裁，所以绝大多数操作都可以在应用程序空间执行，而不需要使用（相对高代价的）内核系统调用。 

---------------------------------------------------------------- 
插播一段关于x86原子操作指令的说明： 

cmpxchg 比较交换指令，其语义为： 

int CompareAndExchange(int *ptr, int old, int new)
{
int actual = *ptr;
if (actual == old)
*ptr = new;
return actual;
}

Intel白皮书上的说明如下： 
(* Accumulator = AL, AX, EAX, or RAX depending on whether a byte, word, doubleword, or 
quadword comparison is being performed *) 
IF accumulator = DEST 
    THEN 
        ZF ← 1; 
        DEST ← SRC; 
    ELSE 
        ZF ← 0; 
        accumulator ← DEST; 
FI; 

使用此原子操作可以实现自旋锁，之前有一篇文章中描述了实现： 

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

关于smp下的原子操作的一些说明： 
  原子操作是不可分割的，在执行完毕不会被任何其它任务或事件中断。在单处理器系统(UniProcessor)中，能够在单条指令中完成的操作都可以认为是" 原子操作"，因为中断只能发生于指令之间。这也是某些CPU指令系统中引入了test_and_set、test_and_clear等指令用于临界资源互斥的原因。在对称多处理器(Symmetric Multi-Processor)结构中就不同了，由于系统中有多个处理器在独立地运行，即使能在单条指令中完成的操作也有可能受到干扰。 
  在x86 平台上，CPU提供了在指令执行期间对总线加锁的手段。CPU芯片上有一条引线#HLOCK pin，如果汇编语言的程序中在一条指令前面加上前缀"LOCK"，经过汇编以后的机器代码就使CPU在执行这条指令的时候把#HLOCK pin的电位拉低，持续到这条指令结束时放开，从而把总线锁住，这样同一总线上别的CPU就暂时不能通过总线访问内存了，保证了这条指令在多处理器环境中的原子性。 
  当然，并不是所有的指令前面都可以加lock前缀的，只有ADD, ADC, AND, BTC, BTR, BTS, CMPXCHG,DEC, INC, NEG, NOT, OR, SBB, SUB, XOR, XADD, 和 XCHG指令前面可以加lock指令，实现原子操作。 
---------------------------------------------------------------- 

广告回来了，我们继续。 

futex保存在用户空间的共享内存中，并且通过原子操作进行操作。在大部分情况下，资源不存在争用的情况下，进程或者线程可以立刻获得资源成功，实际上就没有必要调用系统调用，陷入内核了。实际上，futex的作用就在于减少系统调用的次数，来提高系统的性能。 

线程互斥锁pthread_mutex_t的实现原理： 

pthread_mutex_lock:
atomic_dec(pthread_mutex_t.value);
if(pthread_mutex_t.value!=0)
   futex(WAIT)
else
   success

pthread_mutex_unlock:
atomic_inc(pthread_mutex_t.value);
if(pthread_mutex_t.value!=1)
   futex(WAKEUP)
else
   success

信号量sem_t的实现原理（直接从glibc/nptl/DESIGN-sem.txt中摘的）： 

sem_wait(sem_t *sem)
{
  for (;;) {

    if (atomic_decrement_if_positive(sem->count))
      break;

    futex_wait(&sem->count, 0)
  }
}

sem_post(sem_t *sem)
{
  n = atomic_increment(sem->count);
  // Pass the new value of sem->count
  futex_wake(&sem->count, n + 1);

对比，pthread_mutex_unlock()和sem_post()的实现，我们发现一个不同点，sem_post()无论如何都会调用futex_wake()，进行系统调用。但是pthread_mutex_unlock()却符合futex的初衷，只有在需要仲裁的时候才调用futex_wake()。那么什么是仲裁条件呢？ 

前面说过信号量和线程互斥锁语义上的区别在于信号量的value>=0，而线程互斥锁的value可以为负数。 
对于lock操作，这两个倒是没有多少差别。信号量只要value>0就可以获得资源，线程互斥锁需要value=1。 
但是对于unlock操作，这两个就有一些差别了。信号量和线程互斥锁，都会增加对应的value。如果加1后，value为1，对于线程互斥锁来讲，实际上表明资源可用，并且之前没有其他的线程在等待这个资源；否则说明还有其他线程在等待这个资源，需要调用futex系统调用唤醒它们。但是对于信号量，由于value必须>=0。那么加1后，即使value为1，也无法判定现在没有其他的进程或线程正在等待资源，所以必须调用futex系统调用。例如： 

#include
#include
#include

sem_t sem_a;
void *task1();

int main(void)
{
 int ret=0;
 pthread_t thrd1;
 pthread_t thrd2;
 sem_init(&sem_a,0,1);
 ret=pthread_create(&thrd1,NULL,task1,NULL); //创建子线程
 ret=pthread_create(&thrd2,NULL,task1,NULL); //创建子线程
 pthread_join(thrd1,NULL); //等待子线程结束
 pthread_join(thrd2,NULL); //等待子线程结束
}

void *task1()
{
  int sval = 0;
  sem_wait(&sem_a); //持有信号量
  sleep(5); //do_nothing
  sem_getvalue(&sem_a,&sval);
  printf("sem value = %d\n",sval);
  sem_post(&sem_a); //释放信号量
}

上面sem的value初始化为1，但是有两个线程争用资源。那么第一个线程获得资源成功，当它unlock的时候，sem的value变为1。但是，这个时候，实际上还有一个线程在等待资源。因此，必须要进行futex_wake()系统调用，唤醒等待资源的线程。 

感兴趣的同学可以使用strace跟踪一下，进行验证。要注意忽略程序运行初始化的那个futex_wake ;-) 

参考： 
http://www.eetop.cn/blog/html/04/343504-14125.html 
http://javadino.blog.sohu.com/99256728.html 
http://javadino.blog.sohu.com/99256835.html 
http://javadino.blog.sohu.com/99256921.html 

文章出处：DIY部落()