操作系统概念-内核同步-信号量

       继续内核同步的话题。自旋锁是一种快速简单的锁实现，可以用于临界区很短的情况。但是由于等待锁的进程要不断检测锁的状态，会造成一定的CPU资源浪费。对于这个问题的解决方案就是信号量。信号量是一种睡眠锁，当一个进程试图获得正在使用的锁时，它不会像自旋锁那样原地等待，而是会被丢入一个等待队列，等到这个锁被释放时，等待队列会中的（第一个）进程会被唤醒，并获得该锁。

       还是用前面门和锁的例子来说明。当进程试图进入屋内（临界区），发现屋内有人（有进程在占用），它不会在门口等待，而是把自己的名字（进程ID）记录在门口的一张表上，然后回去睡觉了（进程休眠）。等到屋内的进程出来（离开临界区），系统会检查门口的表，如果表上有等待者的名字，它就去叫醒第一个等待的人（唤醒进程），并把钥匙交给它。信号量的好处是它提供了更好的处理器利用率，没有在徘徊等待上浪费CPU。但另一方面，信号量的开销要比自旋锁大，包括休眠、唤醒以及维护等待队列的开销。所以信号量适合于长期持有锁（长期占据临界区）的情形，而自旋锁适合于短期持有锁的情形。

       信号量的概念有Dijkstra在1968年[1]首次提出，并明确提出了信号量应该有的两个操作：P和V操作（荷兰语，proberen测试和verhogen增加），即down和up操作。进入临界区前执行down操作，检测临界区是否可以进入，离开时再执行up操作恢复标记。

linux下的实现

       信号量（semaphore）的接口放在<linux/semaphore.h>文件内，这个接口文件非常小，只包括了信号量的定义以及两个基本操作：up和down的函数声明：

/*
* Copyright (c) 2008 Intel Corporation
* Author: Matthew Wilcox <willy@linux.intel.com>
*
* Distributed under the terms of the GNU GPL, version 2
*
* Please see kernel/semaphore.c for documentation of these functions
*/
#ifndef __LINUX_SEMAPHORE_H
#define__LINUX_SEMAPHORE_H
#include<linux/list.h>
#include<linux/spinlock.h>

/* Please don't access any members of this structure directly */
structsemaphore {
raw_spinlock_t		lock;
unsignedint		count;
structlist_head	wait_list;
};
#define__SEMAPHORE_INITIALIZER(name, n)				\
{									\
.lock		= __RAW_SPIN_LOCK_UNLOCKED((name).lock),	\
.count		= n,						\
.wait_list	= LIST_HEAD_INIT((name).wait_list),		\
}
#defineDEFINE_SEMAPHORE(name)	\
struct semaphore name = __SEMAPHORE_INITIALIZER(name, 1)
staticinlinevoid sema_init(structsemaphore *sem, intval)
{
staticstructlock_class_key __key;
*sem = (structsemaphore) __SEMAPHORE_INITIALIZER(*sem, val);
lockdep_init_map(&sem->lock.dep_map, "semaphore->lock", &__key, 0);
}
externvoid down(structsemaphore *sem);
externint __must_check down_interruptible(struct semaphore *sem);
externint __must_check down_killable(struct semaphore *sem);
externint __must_check down_trylock(struct semaphore *sem);
externint __must_check down_timeout(struct semaphore *sem, long jiffies);
externvoid up(structsemaphore *sem);
#endif/* __LINUX_SEMAPHORE_H */

       初始化函数sema_init指明了可以同时进入临界区的进程的数量val，这一点特性标表明信号量与自旋锁不同：自旋锁只允许一个进程获取锁，但信号量可以允许val个进程同时获取锁。semaphore的定义很简单，一个count用于计数空闲信号量的数量，即允许进程进入信号量的数量；lock表示count的自旋锁，还有一个保存等待列表的list。

其他函数的定义位于<kernel/semaphore.c>，

void down(structsemaphore *sem)
{
unsignedlong flags;
raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(sem->count > 0))
sem->count--;
else
__down(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);
}
void up(structsemaphore *sem)
{
unsignedlong flags;
raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(list_empty(&sem->wait_list)))
sem->count++;
else
__up(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);
}

static noinline void __sched __down(struct semaphore *sem)
{
__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}

staticinlineint __sched __down_common(struct semaphore *sem, long state,
long timeout)
{
struct task_struct *task = current;
struct semaphore_waiter waiter;

list_add_tail(&waiter.list, &sem->wait_list);
waiter.task = task;
waiter.up = false;

for (;;) {
if (signal_pending_state(state, task))
goto interrupted;
if (unlikely(timeout <= 0))
goto timed_out;
__set_task_state(task, state);
raw_spin_unlock_irq(&sem->lock);
timeout = schedule_timeout(timeout);
raw_spin_lock_irq(&sem->lock);
if (waiter.up)
return 0;
}
timed_out:
list_del(&waiter.list);
return -ETIME;

interrupted:
list_del(&waiter.list);
return -EINTR;
}

static noinline void __sched __up(struct semaphore *sem)
{
struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
struct semaphore_waiter, list);
list_del(&waiter->list);
waiter->up = true;
wake_up_process(waiter->task);
}

先来看up函数。这个函数在进程离开临界区的时候被调用，它会检查等待队列是否为空（使用自旋锁锁定count，然后操作），如果不为空，则调用__up函数。这个函数进一步地唤醒等待队列第一个等待的进程，这个进程会拿到钥匙进入临界区开始执行。

而down函数是在进程进入临界区的时候被调用的。它同样用到了自旋锁来控制count临界区。如果没能进入临界区，就获取当前进程（current）并把它附加到等待列表（wait_list）的最后。然后进入for循环，其中

if (signal_pending_state(state, task))
goto interrupted;

表示如果是TASK_INTERRUPTIBLE或TASK_WAKEKILL状态，并且有信号将进程唤醒，则将当前进程从等待队列中取出并退出：

interrupted:
list_del(&waiter.list);
return -EINTR;

另外一个引起错误的原因是，该进程等待了太长的时间：

if (unlikely(timeout <= 0))
goto timed_out;
timed_out:
list_del(&waiter.list);
return -ETIME;

正常的情况下，则是根据指定的睡眠时间（timeout），调用schedule_timeout：

timeout = schedule_timeout(timeout);

此时进程被休眠，停在上一语句。而当它被唤醒的时候，一定是锁被释放，这是该进程可以获取锁并继续执行：

raw_spin_lock_irq(&sem->lock);
if (waiter.up)
return 0;

根据down函数的注释，这个函数是不推荐使用的，因为在__down内调用__down_common函数指定的state参数是TASK_UNINTERRUPTIBLE，也就是不可抢断，这样会使得进入睡眠的该进程无法被唤醒，只能自行醒来检测有没有锁可以使用。推荐使用的函数有down_interruptible和down_killable等，他们在实现上的区别仅仅是调用__down_common时传入的参数的不同：

信号量接口	功能	调用__down_common的参数
void down(struct semaphore *sem)	获取信号量，不可被信号唤醒（不推荐）	__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
int down_interruptible(struct semaphore *sem)	获取信号量，但可被信号唤醒	__down_common(sem, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
int down_killable(struct semaphore *sem)	获取信号量，但可被致命信号唤醒	__down_common(sem, TASK_KILLABLE, MAX_SCHEDULE_TIMEOUT);
int down_trylock(struct semaphore *sem)	获取信号量，如果信号量不可用，则返回失败而不等待	未调用__down_common
int down_timeout(struct semaphore *sem)	获取信号量，如果在指定的时间内，任务没有被信号量唤醒，则返回失败	__down_common(sem, TASK_UNINTERRUPTIBLE, timeout);

       可以看出，信号量的实现非常精炼，接口也很齐全。调用的方法也比较容易。

读写信号量

除了普通的信号量，linux还实现了读写信号量（相应的，也实现了读写自旋锁，方法类似），主要是为了更高效地处理读者写着问题。其特点为：

1. 同一时刻最多有一个写者（writer）获得信号量；

2. 同一时刻可以有多个读者（reader）获得信号量；

3. 同一时刻写者和读者不能同时获得信号量；

读写信号量的具体定义位于<linxu/Rwsem.h>内：

structrw_semaphore {
atomic_long_t count;
structlist_head wait_list;
raw_spinlock_t wait_lock;
#ifdef CONFIG_RWSEM_SPIN_ON_OWNER
struct optimistic_spin_queue osq; /* spinner MCS lock */
struct task_struct *owner;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map	dep_map;
#endif
};

不同于普通的信号量，读写信号量的count不是记录有多少个进程可以同时获取信号量的。因为对写操作是排他的，即一次只能允许最多一个进程写，但是如果没有进程写，是可以允许若干个进程同时读的。读写信号量的count用于标记多少个进程在读、写该信号量，用起来非常繁琐，因为要同时标记读的进程数量和写的进程数量，实现上均采用内嵌汇编的方法，基于不同架构实现。简单来说，

1， 如果等待队列为空，有N个进程在读，则count为N；

2， 如果等待队列为空，有1个进程在写，则count为0xffff0001，即前16位取反，然后加1；

3， 如果等待队列不空，有N个进程在读，则count为0xffff000N，即前16位取反，然后加N；

4， 如果等待队列不空，有1个进程在写，则count为0xfffe000N。

也就是说，对一个新到的申请读的进程，如果count<0，则该进程无法读，只能进入等待列表；如果count≥0，则该进程可以直接读；对于新到的申请写的进程，如果count=0，即没有进程占有信号量，则可以写，否则进入等待队列等待分配。而等待队列也不是单纯的按先到先得的顺序获取信号量，而是当信号量空闲时对等待队列的进程重新进程分配，决定下一步是读还是写、由哪个/些进程读/写。

读写信号量主要有以下函数：

读写信号量接口	说明
init_rwsem(sem)	宏，初始化读写信号量
int rwsem_is_contended(struct rw_semaphore *sem)	检测信号量的等待列表，非空返回非0
void down_read(struct rw_semaphore *sem);	获取读的锁
int down_read_trylock(struct rw_semaphore *sem);	获取读的锁，成功则返回1
void down_write(struct rw_semaphore *sem);	获取写的锁
int down_write_killable(struct rw_semaphore *sem);	获取写的锁，可被致命信号唤醒
int down_write_trylock(struct rw_semaphore *sem);	获取写的锁，成功则返回1
void up_read(struct rw_semaphore *sem);	释放读的锁
void up_write(struct rw_semaphore *sem);	释放写的锁
void downgrade_write(struct rw_semaphore *sem);	动态的将写锁改为读锁

这些函数与普通信号量无异，因此不再赘述。与普通信号量不同的是downgrade_write这个函数，它可以将写操作的锁降级为读操作锁，是读写信号量特有的操作。

信号量的使用

       信号量的使用比较容易：

/*定义信号量*/
semaphore sem;
/*初始化信号量*/
sema_init(sem, 1);
// ...
/*试图获取信号量*/
if (down_interruptible(&sem))
{
/*错误处理*/
}
/*临界区*/
// ...
/*释放信号量*/
up(&sem);

在linux源码中，主要是device部分用到了信号量。对读写信号量的使用也是类似的流程。

互斥体

       互斥体是简化版的信号量，在同一时刻只允许一个进程获取锁，即满足互斥性。它定义在<linux/Mutex.h>下：

structmutex {
/* 1: unlocked, 0: locked, negative: locked, possible waiters */
atomic_t		count;
spinlock_t		wait_lock;
structlist_head	wait_list;
// ...
};

如注释中所说，count只能为1,0，负值，小于0的时候其绝对值表示等待队列内的进程数量。

mutex相关的函数有：

互斥体接口	说明
define mutex_init(mutex)	宏，初始化互斥体
void mutex_lock(struct mutex *lock);	互斥体上锁
int mutex_lock_interruptible(struct mutex *lock);	互斥体上锁，但可被信号唤醒
int mutex_lock_killable(struct mutex *lock);	互斥体上锁，但可被致命信号唤醒
int mutex_trylock(struct mutex *lock);	互斥体上锁，如果信号量不可用，则返回失败而不等待
void mutex_unlock(struct mutex *lock);	互斥体释放锁
int atomic_dec_and_mutex_lock(atomic_t *cnt, struct mutex *lock)	递减cnt，如果结果为0则互斥体上锁并返回1，否则返回0

上面列出的最后一个函数将原子操作与互斥体结合了起来，相当于

if (atomic_dec_and_test(cnt))
mutex_lock(lock);
else
return 0;
return 1;

可以看出，这些函数与信号量非常相似，互斥体的实现细节和用法也与信号量一样。