【摘要】本文分析了内核的同步及互斥的几种机制：原子运算符（atomic operator）、自旋锁Spinlock、等待队列Waitqueue、事件Event、completion、信号量Semaphore及其优化版互斥锁，详细分析了其实现流程。Event及Semaphore本质上都是基于Waitqueue和自旋锁实现的。本文还探讨了每种机制最适合应用到哪些地方，以及如何构建安全高效的内核及驱动代码。
【关键词】原子操作；Spinlock；Waitqueue；completion；Event；Semaphore
---------------------------------------------------------------------------------------------------------
---------------------------------------------------------------------------------------------------------
1 休眠与同步
一个驱动当它无法立刻满足请求应当如何响应? 一个对 read 的调用可能当没有数据时到来, 而以后会期待更多的数据。或者一个进程可能试图写, 但是你的设备没有准备好接受数据, 因为你的输出缓冲满了。调用进程往往不关心这种问题; 程序员只希望调用 read 或 write 并且使调用返回, 在必要的工作已完成后. 这样, 在这样的情形中。驱动应当(缺省地)阻塞进程, 使它进入睡眠直到请求可继续。
进程被置为休眠，意味着它被标识为处于一个特殊的状态并且从调度器的运行队列中移走。直到发生某些事情改变了那个状态，否则这个进程将不被任何 CPU调度运行。
安全地进入休眠的两条规则：
1) 永远不要在原子上下文中进入休眠。
当驱动在持有一个自旋锁、seqlock或者 RCU 锁时不能睡眠；
关闭中断也不能睡眠；
持有一个信号量时休眠是合法的，但你应当仔细查看代码：如果代码在持有一个信号量时睡眠，任何其他的等待这个信号量的线程也会休眠。因此发生在持有信号量时的休眠必须短暂，而且决不能阻塞那个将最终唤醒你的进程。
2) 当进程被唤醒，重新检查其所需资源。
它并不知道休眠了多长时间以及休眠时发生什么；也不知道是否另有进程也在休眠等待同一事件，且那个进程可能在它之前醒来并获取了所等待的资源。所以不能对唤醒后的系统状态做任何的假设，并必须重新检查等待条件来确保正确的响应。
除非确信其他进程会在其他地方唤醒休眠的进程，否则也不能睡眠。使进程可被找到意味着：需要维护一个称为等待队列的数据结构。它是一个进程链表，其中饱含了等待某个特定事件的所有进程。在 Linux 中， 一个等待队列由一个wait_queue_head_t 结构体来管理。
2 休眠的基础
2.1 wait_queue系列数据结构
2.1.1 wait_queue_head_t
\include\linux\wait.h
struct __wait_queue_head {
spinlock_t lock;
struct list_head task_list;
};
typedef struct __wait_queue_head wait_queue_head_t;
它包含一个自旋锁和一个链表。这个链表是一个等待队列入口。
关于自定义结构体的风格，若需要提供别名，则原始类型前面加”__”或者“tag_”，表示其为内部数据类型，对外是不可见的。typedef之后的类型为了和原始类型分开一般会在后面添加“_t”，表示是typedef的，对外使用。
#define __WAIT_QUEUE_HEAD_INITIALIZER(name) { \
.lock = __SPIN_LOCK_UNLOCKED(name.lock)， \
.task_list = { &(name).task_list， &(name).task_list } \
}
因为Linux内核对于链表的遍历方式的问题，通常一个双向循环链表中有一个头节点，其与其他节点的结构不一样，并且通常无有效信息。此处的等待队列头有两个域：
1) 操作循环链表的互斥锁；
2) 嵌入到等待队列头中的链表头。
为了用“.”域的形式初始化成员不能采用单独的初始化锁和链表头部的宏，但可以采用声明一个结构体类型的宏，如__SPIN_LOCK_UNLOCKED(name.lock)；.task_list的初始化应该采用LIST_HEAD_INIT宏的，这样提高了可移植性。定义等待队列头部的同时，初始化了其成员，尤其是链表头的初始化是添加后续等待队列的前提。
#define DECLARE_WAIT_QUEUE_HEAD(name) \
wait_queue_head_t name = __WAIT_QUEUE_HEAD_INITIALIZER(name)
定义一个等待队列头同时分配内存并进行初始化。对外的接口。
extern void init_waitqueue_head(wait_queue_head_t *q);
void init_waitqueue_head(wait_queue_head_t *q)
{
spin_lock_init(&q->lock);
INIT_LIST_HEAD(&q->task_list);
}
动态初始化一个已经分配了内存的wait_queue_head_t结构。当wait_queue_head_t类型成员内嵌到其他结构体中时需要采用此方法，而不能采用DECLARE_WAIT_QUEUE_HEAD全局或在栈中定义初始化一个wait_queue_head_t结构。
2.1.2 wait_queue_t
struct __wait_queue {
unsigned int flags; //在等待队列上唤醒时是否具备排他性
#define WQ_FLAG_EXCLUSIVE 0x01
void *private;
wait_queue_func_t func; //从等待队列中唤醒进程时执行的统一操作，将进程更改为可运行状态，具体谁运行将由调度策略决定
struct list_head task_list; //与该等待队列对应的链表
};
/*Macros for declaration and initialisaton of the datatypes*/
#define __WAITQUEUE_INITIALIZER(name， tsk) { \
.private = tsk， \
.func = default_wake_function， \
.task_list = { NULL， NULL } \
}
GNU语法中对于结构体成员赋初值采用了域的形式，如“.private =”，其好处在于：
a) 可以选择性的对部分成员赋值。当结构体成员变量较多而大部分无须初始值时，此方法显得尤为重要，因此减少了不必要的赋值。
b) 赋值顺序与数据结构定义中成员的顺序无关，因此若结构体成员顺序变化，初始化部分不会受到任何影响。
c) 各个域之间用“，”分隔，最后一个无“，”。
#define DECLARE_WAITQUEUE(name， tsk) \
wait_queue_t name = __WAITQUEUE_INITIALIZER(name， tsk)
全局或者在栈中定义一个wait_queue_t类型变量的同时对其初始化，这保证了系统的可靠性，避免因用户忘记初始化时导致的问题。通用的初始化宏，tsk为任意指针。分两步：
1) 内部宏__WAITQUEUE_INITIALIZER初始化相应成员；当wq内嵌在别的结构体中时，此宏很重要，提高了可移植性；
2) 提供给外部的接口，定义一个变量，并将第一步的结果赋值给该变量。
static inline void init_waitqueue_entry(wait_queue_t *q， struct task_struct *p)
{
q->flags = 0;
q->private = p;
q->func = default_wake_function;
}
动态初始化一个等待队列入口项，将其和当前进程关联起来，以便唤醒当前进程。
2.1.3 数据结构设计规则
今后凡遇到新设计一类结构体，若此类结构体变量必须初始化且有相对集中的操作，则应提供以下两个操作接口：
a) 定义新建一个结构体变量，并初始化之；
b) 动态初始化一个已经分配内存的该类变量
为了适应在堆栈及全局等任意地方分配的该变量，其应该接收指向该类变量的指针。
2.2 陈旧sleep_on系列
//初始化一个wait_queue_t
#define SLEEP_ON_VAR \
unsigned long flags; \
wait_queue_t wait; \
init_waitqueue_entry(&wait， current);
//添加到队列中
#define SLEEP_ON_HEAD \
spin_lock_irqsave(&q->lock，flags); \
__add_wait_queue(q， &wait); \
spin_unlock(&q->lock);
//从队列中删除
#define SLEEP_ON_TAIL \
spin_lock_irq(&q->lock); \
__remove_wait_queue(q， &wait); \
spin_unlock_irqrestore(&q->lock， flags);
SLEEP_ON_VAR、SLEEP_ON_HEAD及SLEEP_ON_ TAIL总是同时出现，不可分割。上述宏不是函数，只是连续的表达式而已，因为函数就将他们隔离开来了，函数他退出后变量就无意义了。也不能写成do――while的多语句宏，变量定义离开“{}”后就没有意义了。是为了编写代码更清晰明了，同时避免多写字，实际上就是代码封装复用。
void fastcall __sched interruptible_sleep_on(wait_queue_head_t *q)
{
SLEEP_ON_VAR //注意没“；”
current->state = TASK_INTERRUPTIBLE;
SLEEP_ON_HEAD
schedule(); //此处可能出问题
SLEEP_ON_TAIL
}
EXPORT_SYMBOL(interruptible_sleep_on);
添加到队列和从队列中删除由同一个模块做，符合模块设计规则，减小了耦合性。
唤醒的wakeup只负责改变进程状态，进程重新获得cpu后从队列中删除。
long fastcall __sched
interruptible_sleep_on_timeout(wait_queue_head_t *q， long timeout)
{
SLEEP_ON_VAR
current->state = TASK_INTERRUPTIBLE;
SLEEP_ON_HEAD
timeout = schedule_timeout(timeout);
SLEEP_ON_TAIL
return timeout;
}
EXPORT_SYMBOL(interruptible_sleep_on_timeout);
void fastcall __sched sleep_on(wait_queue_head_t *q)
{
SLEEP_ON_VAR
current->state = TASK_UNINTERRUPTIBLE;
SLEEP_ON_HEAD
schedule();
SLEEP_ON_TAIL
}
EXPORT_SYMBOL(sleep_on);
long fastcall __sched sleep_on_timeout(wait_queue_head_t *q， long timeout)
{
SLEEP_ON_VAR
current->state = TASK_UNINTERRUPTIBLE;
SLEEP_ON_HEAD
timeout = schedule_timeout(timeout);
SLEEP_ON_TAIL
return timeout;
}
EXPORT_SYMBOL(sleep_on_timeout);
在决定调用sleep_on系列函数到真正调用schedule系列函数期间，若等待的条件为真，若此时继续schedule，相当于丢失了一次唤醒机会。因此sleep_on系列函数会引入竞态，导致系统的不安全。
另外对于interruptible系列函数，其返回时并不能确定是因为资源可用返回还是遇到了signal，因此在程序中用户需要再次判断资源是否可用。如：
static ssize_t at91_mcp2510_read(struct file *filp, char *buffer, size_t count, loff_t *ppos)
{
CanData candata_ret;
retry:
if (mcp2510dev.nCanReadpos != mcp2510dev.nCanRevpos) {// 需求的资源
int count;
count = MCP2510_RevRead(&candata_ret);
if (count) copy_to_user(buffer, (char *)&candata_ret, count);
return count;
} else { //不可用
if (filp->f_flags & O_NONBLOCK)
return -EAGAIN;
interruptible_sleep_on(&(mcp2510dev.wq));
if (signal_pending(current)) // 遇到信号，返回
return -ERESTARTSYS;
goto retry; //重新判断资源是否可用
}
DPRINTK("read data size=%d\n", sizeof(candata_ret));
return sizeof(candata_ret);
}
综合上述两个因素，sleep_on系列函数应避免在驱动程序中出现，未来的2.7版内核中将删除此类函数。
2.3 等待队列的接口
2.3.1 初始化等待队列
#define DEFINE_WAIT(name) \
wait_queue_t name = { \
.private = current, \
.func = autoremove_wake_function, \
.task_list = LIST_HEAD_INIT((name).task_list), \
}
#define init_wait(wait) \
do { \
(wait)->private = current; \
(wait)->func = autoremove_wake_function; \
INIT_LIST_HEAD(&(wait)->task_list); \
} while (0)
专用的初始化等待队列函数，将当前进程添加到等待队列中，注意和通用的接口DECLARE_WAITQUEUE及init_waitqueue_entry区别。同时唤醒处理不一样，autoremove_wake_function在default_wake_function的基础之上，将当前进程从等待队列中删除。
2.3.2 添加或从等待队列中删除
添加删除的原始接口
static inline void __add_wait_queue(wait_queue_head_t *head, wait_queue_t *new)
{
list_add(&new->task_list, &head->task_list);
}
static inline void __add_wait_queue_tail(wait_queue_head_t *head,
wait_queue_t *new)
{
list_add_tail(&new->task_list, &head->task_list);
}
static inline void __remove_wait_queue(wait_queue_head_t *head,
wait_queue_t *old)
{
list_del(&old->task_list);
}
等待队列是公用资源，但上述接口没有加任何保护措施，适用于已经获得锁的情况下使用。
带锁并设置属性的添加删除
void fastcall add_wait_queue(wait_queue_head_t *q, wait_queue_t *wait)
{
unsigned long flags;
wait->flags &= ~WQ_FLAG_EXCLUSIVE;
spin_lock_irqsave(&q->lock, flags);
__add_wait_queue(q, wait);
spin_unlock_irqrestore(&q->lock, flags);
}
EXPORT_SYMBOL(add_wait_queue);
void fastcall add_wait_queue_exclusive(wait_queue_head_t *q, wait_queue_t *wait)
{
unsigned long flags;
wait->flags |= WQ_FLAG_EXCLUSIVE;
spin_lock_irqsave(&q->lock, flags);
__add_wait_queue_tail(q, wait);
spin_unlock_irqrestore(&q->lock, flags);
}
EXPORT_SYMBOL(add_wait_queue_exclusive);
void fastcall remove_wait_queue(wait_queue_head_t *q, wait_queue_t *wait)
{
unsigned long flags;
spin_lock_irqsave(&q->lock, flags);
__remove_wait_queue(q, wait);
spin_unlock_irqrestore(&q->lock, flags);
}
EXPORT_SYMBOL(remove_wait_queue);
此三对函数的特点是调用同名的内部函数，同时添加一些保障安全特性的代码。WQ_FLAG_EXCLUSIVE属性的进程添加到队尾，而非WQ_FLAG_EXCLUSIVE从队头添加。这样可以保证每次都能唤醒所有的非WQ_FLAG_EXCLUSIVE进程。
无锁但设置属性的添加删除
static inline void add_wait_queue_exclusive_locked(wait_queue_head_t *q,
wait_queue_t * wait)
{
wait->flags |= WQ_FLAG_EXCLUSIVE;
__add_wait_queue_tail(q, wait);
}
/* Must be called with the spinlock in the wait_queue_head_t held.*/
static inline void remove_wait_queue_locked(wait_queue_head_t *q,
wait_queue_t * wait)
{
__remove_wait_queue(q, wait);
}
Locked系列适用于在已经获得锁的情况下调用，通常用于信号量后者complete系列函数中。
上述三组函数，__add_wait_queue是内部函数，无任何保护，无任何属性设置；而另外两组分别适用于当前是否加锁的两种场合，是对外的接口函数。这种根据适用场合不同提供不同的接口函数的方法值得借鉴。
2.3.3 prepare_to_wait和finish_wait
/*
* Used to distinguish between sync and async io wait context:
* sync i/o typically specifies a NULL wait queue entry or a wait
* queue entry bound to a task (current task) to wake up.
* aio specifies a wait queue entry with an async notification
* callback routine， not associated with any task.
*/
#define is_sync_wait(wait) (!(wait) || ((wait)->private))
同步io等待将唤醒当前进程，异步io等待和当前进程无关，时间到后执行安装的回调函数
void fastcall
prepare_to_wait(wait_queue_head_t *q， wait_queue_t *wait， int state)
{
unsigned long flags;
wait->flags &= ~WQ_FLAG_EXCLUSIVE; //非排它性的等待将都被唤醒
spin_lock_irqsave(&q->lock， flags);
//等待节点尚未添加到任何等待队列中，防止重复加入
if (list_empty(&wait->task_list))
__add_wait_queue(q， wait);
if (is_sync_wait(wait))
set_current_state(state);
spin_unlock_irqrestore(&q->lock， flags);
}
prepare_to_wait_exclusive(wait_queue_head_t *q, wait_queue_t *wait, int state)
wait->flags |= WQ_FLAG_EXCLUSIVE;
排他性等待，其余和prepare_to_wait一样
void fastcall finish_wait(wait_queue_head_t *q， wait_queue_t *wait)
{
unsigned long flags;
__set_current_state(TASK_RUNNING); //确保进程状态为running
//若有等待进程，则将其从等待队列中删除
if (!list_empty_careful(&wait->task_list)) {
spin_lock_irqsave(&q->lock， flags);
list_del_init(&wait->task_list);
spin_unlock_irqrestore(&q->lock， flags);
}
}
3 等待事件event
Linux 内核中最简单的休眠方式是称为 wait_event的宏(及其变种)，它实现了休眠和进程等待的条件的检查。形式如下：
wait_event(queue, condition)/*不可中断休眠,不推荐*/ wait_event_interruptible(queue, condition)/*推荐，返回非零值意味着休眠被中断,且驱动应返回 -ERESTARTSYS*/ wait_event_timeout(queue, condition, timeout) wait_event_interruptible_timeout(queue, condition, timeout) /*有限的时间的休眠；若超时，则不管条件为何值返回0，*/
上述四个宏函数为内核对外的接口，其他的休眠函数应避免使用。因为宏并不是函数，参数所做的任何修改对调用环境仍然有效，所以queue都是“值传递”，在宏内部会调用底层函数，采用的指针传递。Linux内核中存在大部分这样的宏，其都在接口上都是值传递。
3.1 wait_event
认真地看简单休眠中的 wait_event(queue, condition) 和 wait_event_interruptible(queue, condition) 底层源码会发现，其实他们只是手工休眠中的函数的组合。因此在驱动程序中应避免使用手动休眠代码，而应该采用内核已经封装好的四个wait_event系列函数。
\include\linux\wait.h
#define __wait_event(wq，condition) \
do { \
DEFINE_WAIT(__wait); \
\
for(;;) { \
prepare_to_wait(&wq，&__wait，TASK_UNINTERRUPTIBLE); \
/// 添加到等待队列中，同时更改进程状态；若已经加入则不会重复添加
if (condition) \
break; \
schedule(); //何时返回呢？？？ \
} \
finish_wait(&wq，&__wait); \
} while (0)
// “__”表示内部函数，默认为condition不满足，添加至等待队列，调度
注意prepare_to_wait和finish_wait的匹配关系
#define wait_event(wq，condition) \
do { \
if(condition) \
break; \
__wait_event(wq，condition); \
}while (0)
//对外的接口函数，需要判断condition，若假则等待；若真则直接退出
--------------------------------------------------------------------------------------------------------------
等待系列函数架构设计:
3.2 wait_event_timeout
#define __wait_event_timeout(wq， condition， ret) \
do { \
DEFINE_WAIT(__wait); \
\
for (;;) { \
prepare_to_wait(&wq， &__wait， TASK_UNINTERRUPTIBLE); \
if (condition) \
break; \
ret = schedule_timeout(ret); \
if (!ret) \
break; //延时到，退出 \
} \
finish_wait(&wq， &__wait); \
} while (0)
#define wait_event_timeout(wq，condition，timeout) \
({ \
long __ret=timeout; \
if( !(condition) ) \
__wait_event_timeout( wq，condition，__ret); \
__ret; \
})
3.3 wait_event_interruptible
#define __wait_event_interruptible(wq, condition, ret) \
do { \
DEFINE_WAIT(__wait); \
&n
阅读(1626) | 评论(0) | 转发(0) |
