Linux等待队列

多路IO复用select采用等待队列机制让用户程序没有资源读/写时睡眠，有资源读/写时唤醒用户程序。

等待队列以双向循环链表为基础数据结构，与进程调度紧密结合，用于实现内核的异步事件通知机制，也可用于同步对系统资源的访问。

1. 数据结构

(1) 每个等待队列都有一个等待队列头，该结构定义如下：

struct __wait_queue_head {
spinlock_t		lock;           //自旋锁，在对task_list操作时，使用该锁实现对等待队列的互斥访问
struct list_head	task_list;      // 双向循环链表
};

每次对等待队列的操作都需要加自旋锁，防止其他进程对等待队列的写操作，实现对等待队列的互斥访问，并且保证临界区的原子性。使用task_list连接一个等待队列项。

(2) 等待队列项，定义如下：

//等待队列项。每个等待任务都会抽象成为一个_wait_queue,并且挂载在__wait_queue_head
struct __wait_queue {
unsigned int		flags;       // 互斥/非互斥进程
#define WQ_FLAG_EXCLUSIVE   0x01
void			*private;    // 指向一个task_struct实例
wait_queue_func_t	func;        // 函数指针，用于唤醒进程
struct list_head	task_list;  // 挂入__wait_queue_head
};
typedef struct __wait_queue_head wait_queue_head_t;
typedef struct __wait_queue wait_queue_t;

#define WQ_FLAG_EXCLUSIVE 0x01 互斥进程标志，private指向一个task_struct实例，每一个进程使用结构体task_struct来表示。Func是一个函数指针，声明如下：

typedef int (*wait_queue_func_t)(wait_queue_t *wait, unsigned mode, int flags, void *key);

该函数指针在唤醒进程函数__wake_up_common调用，func默认调用default_wake_function唤醒进程。使用task_list与__wait_queue_head连接。

2. 初始化

(1) 等待队列头初始化。

//初始化
#define DECLARE_WAIT_QUEUE_HEAD(name)                              \
wait_queue_head_t name = __WAIT_QUEUE_HEAD_INITIALIZER(name)
// 初始化未加锁
#define __WAIT_QUEUE_HEAD_INITIALIZER(name) {			   \
.lock		= __SPIN_LOCK_UNLOCKED(name.lock),         \
.task_list	= { &(name).task_list, &(name).task_list } }

等待队列头初始化时未加锁，双向链表的prev，next指针都指向wait_queue_head的task_list，也就是双向链表的头节点。

(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;
}
#define DEFINE_WAIT_FUNC(name, function)				\
wait_queue_t name = {					        \
.private	= current,			        \
.func		= function,				\
.task_list	= LIST_HEAD_INIT((name).task_list),	\
}
#define DEFINE_WAIT(name) DEFINE_WAIT_FUNC(name, autoremove_wake_function)  // autoremove_wake_function 唤醒进程，调用default_wake_function，将被唤醒进程从等待队列删除
// 双向链表prev，next指向空，唤醒函数为default_wake_function，并指定等待的进程。
#define __WAITQUEUE_INITIALIZER(name, tsk) {				\
.private	= tsk,					        \
.func		= default_wake_function,			\
.task_list	= { NULL, NULL } }
#define DECLARE_WAITQUEUE(name, tsk)					\
wait_queue_t name = __WAITQUEUE_INITIALIZER(name, tsk)

3. 添加/删除等待队列项

(1) 添加(头插)

为了保证进程对等待队列的互斥访问，需要加自旋锁，使进程进入临界区

void 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); // 释放锁
}
// 采用头插的方法list_add，将new节点插入到head之后
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_exclusive(wait_queue_head_t *q, wait_queue_t *wait)
{
wait->flags |= WQ_FLAG_EXCLUSIVE;
__add_wait_queue(q, wait);
}

(2) 删除

//进程唤醒，从wait_queue删除该节点
void 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);
}
static inline void
__remove_wait_queue(wait_queue_head_t *head, wait_queue_t *old)
{
list_del(&old->task_list);
}

4. 等待事件

<span style="font-size:10px;">#define wait_event(wq, condition)					\
do {							                \
might_sleep();					                \
if (condition)					                \
break;
\
__wait_event(wq, condition);				        \
} while (0)

#define __wait_event(wq, condition)					\
(void)___wait_event(wq, condition, TASK_UNINTERRUPTIBLE, 0, 0,	\
schedule())

#define ___wait_event(wq, condition, state, exclusive, ret, cmd)	  \
({									  \
__label__ __out;						  \
wait_queue_t __wait;						  \
long __ret = ret;	/* explicit shadow */			  \
\
INIT_LIST_HEAD(&__wait.task_list);				  \
if (exclusive)							  \
__wait.flags = WQ_FLAG_EXCLUSIVE;			  \
else								  \
__wait.flags = 0;					  \
\
for (;;) {							  \
long __int = prepare_to_wait_event(&wq, &__wait, state);  \
\
if (condition)						  \
break;					          \
\
if (___wait_is_interruptible(state) && __int) {		  \
__ret = __int;					  \
if (exclusive) {			          \
abort_exclusive_wait(&wq, &__wait,	  \
state, NULL);	  \
goto __out;				  \
}					          \
break;						  \
}							  \
\
cmd;							  \
}	                                                          \
finish_wait(&wq, &__wait);				          \
__out:	__ret;								  \
})</span>

在等待队列中，进程睡眠直到condition为真。在等待期间，调用___wait_event(wq, condition, state, exclusive, ret, cmd)。该宏主要完成：

a. 创建等待队列成员

b. 在for(;;)中，prepare_to_wait_event使进程在等待队列上等待，并将进程状态置为不可中断TASK_UNINTERRUPTIBLE。

c. 当进程被唤醒时，检查condition是否满足条件，不满足，调用schedule进行进程调度，进程继续睡眠。相反，调用finis_wait将进程状态设置为TASK_RUNNING，并从等待队列删除该进程。

与wait_event类似的函数

wait_event_interruptible(queue,condition);//可被信号打断

wait_event_hrtimeout(queue,condition,timeout);//阻塞等待的超时时间，时间到了，不论condition是否满足，都要返回

wait_event_interruptible_timeout(queue,condition,timeout)

5.唤醒事件

//唤醒等待队列.可唤醒处于TASK_INTERRUPTIBLE和TASK_UNINTERUPTIBLE状态的进程,和wait_event/wait_event_timeout成对使用
void __wake_up(wait_queue_head_t *q, unsigned int mode,
int nr_exclusive, void *key)
{
unsigned long flags;
spin_lock_irqsave(&q->lock, flags);
__wake_up_common(q, mode, nr_exclusive, 0, key);
spin_unlock_irqrestore(&q->lock, flags);
}
#define wake_up_interruptible(x)    __wake_up(x, TASK_INTERRUPTIBLE, 1, NULL)
//和wake_up()唯一的区别是它只能唤醒TASK_INTERRUPTIBLE状态的进程.,与wait_event_interruptible， wait_event_interruptible_timeout，wait_event_interruptible_exclusive成对使用

#define wake_up_all(x)          __wake_up(x, TASK_NORMAL, 0, NULL)
#define wake_up_interruptible_nr(x, nr) __wake_up(x, TASK_INTERRUPTIBLE, nr, NULL)
#define wake_up_interruptible_all(x)    __wake_up(x, TASK_INTERRUPTIBLE, 0, NULL)
//这些也基本都和wake_up/wake_up_interruptible一样
__wake_up 在临界区调用__wake_up_common()
static void __wake_up_common(wait_queue_head_t *q, unsigned int mode,
int nr_exclusive, int wake_flags, void *key)
{
wait_queue_t *curr, *next;

list_for_each_entry_safe(curr, next, &q->task_list, task_list) {
unsigned flags = curr->flags;

if (curr->func(curr, mode, wake_flags, key) &&
(flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
break;
}
}

根据进程状态mode，扫描等待队列，调用func(默认为default_wake_function)唤醒进程，直至没有更多的进程被唤醒，或者被唤醒的的独占进程数目已经达到规定数目。

参考资料

http://www.cnblogs.com/gdk-0078/p/5172941.html

http://blog.chinaunix.net/uid-27714502-id-3450323.html