Linux中的等待队列

在linux同步方法中，有一种非常有效的同步数据结构:wait_queue_head_t，这里就具体分析其数据结构。

wait_head_t定义如下:

struct __wait_queue_head {
    spinlock_t lock;    //保护该链表的自旋锁
    struct list_head task_list;//双向链表
};
typedef struct __wait_queue_head wait_queue_head_t;

双向链表中保存的节点定义如下:

typedef struct __wait_queue  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;
};

声明一个等待队列函数定义如下:

#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)

在添加一个元素进入等待队列时，可以指明唤醒进程的方式：唤醒所有的睡眠进程，只唤醒所有的睡眠进程，等待队列入口可以设置WQ_FLAG_EXCLUSIVE标志来进行区分。

等待队列入口设置了WQ_FLAG_EXCLUSIVE标志时，则会被添加到等待队列的尾部，而没有这个标志的入口就会被添加到头部。在某个等待队列上调用wake_up时，它会在唤醒第一个具有WQ_FLAG_EXCLUSIVE标志的进程之后停止唤醒其他进程。

添加一个元素进入等待队列:

void fastcall add_wait_queue(wait_queue_head_t *q,wait_queue_t *wait)
{
    unsigned long flags;

    wait->flags &= ~WQ_FLAG_EXLUSIVE;
    spin_lock_irqsave(&q->lock,flags);
    __add_wait_queue(q,wait);//添加到等待队列的头部
    spin_unlock_irqrestore(&q->lock,flags);
}
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_irqrestore(&q->lock,flags);
    //添加元素到尾部的WQ_FLAG_EXCLUSIVE标志
    __add_wait_queue_tail(q,wait);
    spin_unlock_irqrestore(&q->lock,flags);
}

上层对应的函数接口为:

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);
}
void fastcall prepare_to_wait_exclusive(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_tail(q,wait);//加入队列尾部
if(is_sync_wait(wait))
set_current_state(state);
spin_unlock_irqrestore(&q->lock,flags);
}

上面两个函数直接设置进程的状态，在调用该函数之后，进程即可调用schedule,当然在这之前，进程应确保仍有必要等待，常见的处理流程如下:

set_current_state(TASK_INTERUPTIBLE);
if(need_to_sleep)  schedule();

一旦schedule返回，就到了清理时间了。这就需要下面的函数:

void fastcall  finish_wait(wait_queue_head_t *q,wait_queue_t *wait)
{
    unsigned long flags;
    __set_current_state(TASK_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);
    }
}

当满足条件后，系统开始唤醒睡眠在等待队列上的进程，唤醒函数如下:

wake_up(wait_queue_head_t *queue);
wake_up_interruptible(wait_queue_head_t *queue);
wake_up_nr(wait_queue_head_t *queue,int nr);
wake_up_interruptible_nr(wait_queue_head_t *queue,int nr);
wake_up_all(wait_queue_head_t *queue);
wake_up_intrruptible_all(wait_queue_head_t *queue);
wake_up_intrruptible_sync(wait_queue_head_t *queue);

函数定义如下:

//唤醒所有非独占进程且一个独占进程
#define wake_up(x)    __wake_up(x,TASK_UNINTERRUPTIBLE | TASK_INTERRUPTIBLE,1,NULL)
//唤醒所有非独占进程且nr个独占进程
#define wake_up_nr(x,nr)    __wake_up(x,TASK_UNINTERRUPTIBLE|TASK_INTERRUPTIBLE,nr,NULL)
//唤醒所有进程(不管是否独占或非独占)
#define wake_up_all(x,TASK_UNINTERRUPTIBLE|TASK_INTERRUPTIBLE,0,NULL)
//唤醒所有非独占，可中断进程以及独占可中断进程
#define wake_up_interruptible(x)  __wake_up(x,TASK_INTERRUPTIBLE,1,NULL)
//唤醒所有非独占可中断进程以及nr个独占可中断进程
#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)
//互斥唤醒所有可中断非独占进程以及一个可独占可中断进程
#define wake_up_interruptible_sync(x)    __wake_up_sync((x),TASK_INTERRUPTILBE,1)

据LDD介绍，被唤醒的进程可能会抢占当前的进程，并在wake_up返回前被调度到处理器上。换句话说，对wake_up的调用可能不是原子的。如果调用wake_up的进程运行在原子上下文(如拥有自旋锁或者是一个中断处理例程)中，则调度就不会发生。如果不希望在这时被调度出处理器，则可使用该函数，即如果包含有sync则wake_up进程就不会抢占当前正在执行唤醒的进程，结果是唤醒的高优先级进程(与当前进程同一cpu时)的执行出现延迟。

//如果已经有wait_queue_head_t的自旋锁就调用该函数，其它的与wake_up一致
#define wake_up_locked(x)    __wake_up_locked((x),TAKS_UNINTERRUPTIBLE | TASK_INTERRUPTIBLE)

上面的函数都调用了__wake_up该函数如下:

void fastcall __wake_up(wait_queue_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);
}
void fastcall __wake_up_sync(wait_queue_head_t *q,unsigned int mode,int nr_exclusive)
{
unsigned long flags;
int sync = 1;
if(unlikely(!q))	return;
if(unlikely(!nr_exclusive))	sync = 0;
spin_lock_irqsave(&q->lock,flags);
__wake_up_common(q,mode,nr_exclusive,sync,NULL);
spin_unlock_irqrestore(&q->lock,flags);
}
void fastcall __wake_up_locked(wait_queue_head_t *q,unsigned int mode)
{
__wake_up_common(q,mode,1,0,NULL);
}

__wake_up_common定义如下:

static void __wake_up_common(wait_queue_head_t *q,unsigned int mode,
    int nr_exclusive,int sync,void *key)
{
    wait_queue_t *curr,*next;
    list_for_each_entry_safe(curr,next,&q->task_list,task_list) {
        unsigned flags = curr->flags;
        //调用默认的注册回调函数autoremove_wake_function
        if(curr->func(curr,mode,sync,key) && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
            break;
    }
}

autoremove_wake_function则进行唤醒在队列上的进程(void*prviate):

int autoremove_wake_function(wait_queue_t *wait,unsigned mode,int sync,
        void *key)
{
    int ret = default_wake_function(wait,mode,sync,key);//唤醒该进程

    if(ret)
        list_del_init(&wait->task_list);//从等待队列中删除该进程
    return ret;
}

int default_wake_function(wait_queue_t *curr,unsigned mode,int sync,void *key)
{
    return try_to_wake_up(curr->private,mode,sync);
}

参考资料

1.中linux-2.6.24.3内核源码(linux/wait.c sched.c,include/wait.h)

2.linux设备驱动程序

3.深入理解linux内核(第三版)