源码解读Linux等待队列

从源码角度来解读Linux等待队列机制，了解休眠与唤醒的运转原理

1kernel/include/linux/wait.h
2kernel/kernel/sched/wait.c
3kernel/include/linux/sched.h
4kernel/kernel/sched/core.c

一、概述

Linux内核的等待队列是非常重要的数据结构，在内核驱动中广为使用，它是以双循环链表为基础数据结构，与进程的休眠唤醒机制紧密相联，是实现异步事件通知、跨进程通信、同步资源访问等技术的底层技术支撑。

研究等待队列这个内核非常基础的数据结构，对于加深理解Linux非常有帮忙，等待队列有两种数据结构：等待队列头(wait_queue_head_t)和等待队列项(wait_queue_t)，两者都有一个list_head类型task_list。双向链表通过task_list将等待队列头和一系列等待队列项串起来，源码如下所示。

二、等待队列

2.1 struct wait_queue_head_t

1struct __wait_queue_head {
2    spinlock_t        lock;  //用于互斥访问的自旋锁
3    struct list_head    task_list;
4};
5typedef struct __wait_queue_head wait_queue_head_t;

可通过宏DECLARE_WAIT_QUEUE_HEAD(name)来创建类型为wait_queue_head_t的等待队列头name。

1#define DECLARE_WAIT_QUEUE_HEAD(name) \
2    struct wait_queue_head name = __WAIT_QUEUE_HEAD_INITIALIZER(name)
3    
4#define __WAIT_QUEUE_HEAD_INITIALIZER(name) {                    \
5    .lock        = __SPIN_LOCK_UNLOCKED(name.lock),            \
6    .head        = { &(name).head, &(name).head } }

2.2 struct wait_queue_t

1struct __wait_queue {
2    unsigned int        flags;
3    void            *private;  //指向等待队列的进程task_struct
4    wait_queue_func_t    func;  //唤醒函数
5    struct list_head    task_list; //链表元素，将wait_queue_t挂到wait_queue_head_t
6};
7typedef struct __wait_queue wait_queue_t;

可通过宏DECLARE_WAITQUEUE(name, tsk) 来创建类型为wait_queue_t的等待队列项name，并将tsk赋值给成员变量private， default_wake_function赋值给成员变量func。

1#define DECLARE_WAITQUEUE(name, tsk)                        \
2    struct wait_queue_entry name = __WAITQUEUE_INITIALIZER(name, tsk)
3
4#define __WAITQUEUE_INITIALIZER(name, tsk) {                    \
5    .private    = tsk,                            \
6    .func        = default_wake_function,                \
7    .entry        = { NULL, NULL } }

2.3 add_wait_queue

1void add_wait_queue(wait_queue_head_t *q, wait_queue_t *wait)
2{
3    unsigned long flags;
4    wait->flags &= ~WQ_FLAG_EXCLUSIVE;
5    spin_lock_irqsave(&q->lock, flags);
6    __add_wait_queue(q, wait);  //挂到队列头
7    spin_unlock_irqrestore(&q->lock, flags);
8}
9
10static inline void __add_wait_queue(wait_queue_head_t *head, wait_queue_t *new)
11{
12    list_add(&new->task_list, &head->task_list);
13}

该方法的功能是将wait等待队列项挂到等待队列头q中。

2.4 remove_wait_queue

1void remove_wait_queue(wait_queue_head_t *q, wait_queue_t *wait)
2{
3    unsigned long flags;
4    spin_lock_irqsave(&q->lock, flags);
5    __remove_wait_queue(q, wait);
6    spin_unlock_irqrestore(&q->lock, flags);
7}
8
9static inline void __remove_wait_queue(wait_queue_head_t *head, wait_queue_t *old)
10{
11    list_del(&old->task_list);
12}

该方法主要功能是将wait等待队列项从等待队列头q中移除。

到这里，已经介绍了wait_queue_head_t和wait_queue_t这两个创建方法，以及增加和删除等待队列元素的方法。接下里说一说如何在等待队列的基础上建立休眠与唤醒机制。

三、休眠与唤醒

3.1 wait_event

1#define wait_event(wq, condition)                    \
2do {                                    \
3    if (condition)                            \
4        break;                            \
5    __wait_event(wq, condition);                    \
6} while (0)
7
8#define __wait_event(wq, condition)                    \
9    (void)___wait_event(wq, condition, TASK_UNINTERRUPTIBLE, 0, 0, schedule())

将___wait_event()宏展开如下所示

1___wait_event(wq, condition, state, exclusive, ret, cmd){  
2    wait_queue_t __wait;                    
3    INIT_LIST_HEAD(&__wait.task_list);                
4    for (;;) {
5        //当检测进程是否有待处理信号则返回值__int不为0，【见3.1.1】
6        long __int = prepare_to_wait_event(&wq, &__wait, state);
7        if (condition)  //当满足条件，则跳出循环                    
8            break;                        
9                                    
10        //当有待处理信号且进程处于可中断状态(TASK_INTERRUPTIBLE或TASK_KILLABLE))，则跳出循环
11        if (___wait_is_interruptible(state) && __int) {        
12            __ret = __int;                    
13            break;                      
14        }                            
15        cmd; //schedule()，进入睡眠，从进程就绪队列选择一个高优先级进程来代替当前进程运行                       
16    }                                
17    finish_wait(&wq, &__wait);  //如果__wait还位于队列wq，则将__wait从wq中移除              
18}

3.1.1 prepare_to_wait_event

再来看看进入睡眠状态之前的准备工作，用于防止wait没有队列中，而事件已产生，则会无线等待。

1long prepare_to_wait_event(wait_queue_head_t *q, wait_queue_t *wait, int state)
2{
3    unsigned long flags;
4    if (signal_pending_state(state, current)) //信号检测
5        return -ERESTARTSYS;
6
7    wait->private = current;
8    wait->func = autoremove_wake_function; //设置func唤醒函数，【小节3.1.2】
9
10    spin_lock_irqsave(&q->lock, flags);
11    if (list_empty(&wait->task_list)) {  //当wait不在队列q，则加入其中，防止无法唤醒
12        if (wait->flags & WQ_FLAG_EXCLUSIVE)
13            __add_wait_queue_tail(q, wait);
14        else
15            __add_wait_queue(q, wait);
16    }
17    set_current_state(state);  //设置进程状态
18    spin_unlock_irqrestore(&q->lock, flags);
19
20    return 0;
21}

wait_event(wq, condition)：进入睡眠状态直到condition为true，在等待期进程状态为TASK_UNINTERRUPTIBLE。对应的唤醒方法是wake_up()，当等待队列wq被唤醒时会执行如下两个检测：

检查condition是否为true，满足条件，则跳出循环。
检测该进程task的成员thread_info->flags是否被设置TIF_SIGPENDING，被设置则说明有待处理的信号，则跳出循环。

wait_event_xxx有一组用于睡眠的函数，基于是否可中断(TASK_UNINTERRUPTIBLE)，是否有超时机制，在方法名后缀加上interruptible，timeout等信息，对应的含义就是允许中断(TASK_INTERRUPTINLE)和带有超时机制，比如wait_event_interruptible()，这里就不再列举。另外sleep_on()也是进入睡眠状态，没有condition，不过该方法有可能导致竞态，从kernel 3.15移除该方法，采用wait_event代替sleep_on()。

3.1.2 autoremove_wake_function

1int autoremove_wake_function(wait_queue_t *wait, unsigned mode, int sync, void *key)
2{
3    int ret = default_wake_function(wait, mode, sync, key); //唤醒函数
4    if (ret)
5        list_del_init(&wait->task_list); //从列表中移除wait
6    return ret;
7}

3.2 wake_up

1#define wake_up(x)            __wake_up(x, TASK_NORMAL, 1, NULL)
2
3
4void __wake_up(wait_queue_head_t *q, unsigned int mode,
5            int nr_exclusive, void *key)
6{
7    unsigned long flags;
8    spin_lock_irqsave(&q->lock, flags);
9    //核心方法
10    __wake_up_common(q, mode, nr_exclusive, 0, key);
11    spin_unlock_irqrestore(&q->lock, flags);
12}
13
14
15static void __wake_up_common(wait_queue_head_t *q, unsigned int mode,
16            int nr_exclusive, int wake_flags, void *key)
17{
18    wait_queue_t *curr, *next;
19
20    list_for_each_entry_safe(curr, next, &q->task_list, task_list) {
21        unsigned flags = curr->flags;
22        //调用唤醒函数 【小节3.2.1】
23        if (curr->func(curr, mode, wake_flags, key) &&
24                (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
25            break;
26    }
27}

wait_event(wq)遍历整个等待列表wq中的每一项wait_queue_t，依次调用唤醒函数来唤醒该等待队列中的所有项，唤醒函数如下：

对于通过宏DECLARE_WAITQUEUE(name, tsk) 来创建wait，再调用add_wait_queue(wq, wait)方法，则唤醒函数为default_wake_function
对于通过wait_event(wq, condition)方式加入的wait项，则经过调用prepare_to_wait_event()方法，则唤醒函数为autoremove_wake_function，由小节[1.3.5]可知，该方法主要还是调用default_wake_function来唤醒。

wake_up_xxx有一组用于唤醒的函数，跟wait_event配套使用。比如wait_event()与wake_up()，wait_event_interruptible()与wake_up_interruptible()。

3.2.1 default_wake_function

再来看看唤醒函数函数

1int default_wake_function(wait_queue_t *curr, unsigned mode, int wake_flags,
2              void *key)
3{
4    //获取wait所对应的进程 【小节3.2.2】
5    return try_to_wake_up(curr->private, mode, wake_flags);
6}

3.2.2 try_to_wake_up

1try_to_wake_up(struct task_struct *p, unsigned int state, int wake_flags)
2{
3    unsigned long flags;
4    int cpu, src_cpu, success = 0;
5
6    bool freq_notif_allowed = !(wake_flags & WF_NO_NOTIFIER);
7    bool check_group = false;
8    wake_flags &= ~WF_NO_NOTIFIER;
9
10    smp_mb__before_spinlock();
11    raw_spin_lock_irqsave(&p->pi_lock, flags); //关闭本地中断
12    src_cpu = cpu = task_cpu(p);
13
14    //如果当前进程状态不属于可唤醒状态集，则无法唤醒该进程
15    //wake_up()传递过来的TASK_NORMAL等于(TASK_INTERRUPTIBLE | TASK_UNINTERRUPTIBLE)
16    if (!(p->state & state)) 
17        goto out;
18
19    success = 1; 
20    smp_rmb();
21    if (p->on_rq && ttwu_remote(p, wake_flags)) //当前进程已处于rq运行队列，则无需唤醒
22        goto stat;
23    ...
24    
25    ttwu_queue(p, cpu); //【小节3.2.3】
26stat:
27    ttwu_stat(p, cpu, wake_flags);
28out:
29    raw_spin_unlock_irqrestore(&p->pi_lock, flags); //恢复本地中断
30    ...
31    return success;
32}

3.2.3 ttwu_queue

1static void ttwu_queue(struct task_struct *p, int cpu)
2{
3	struct rq *rq = cpu_rq(cpu); // 获取当前进程的运行队列
4	raw_spin_lock(&rq->lock);
5	lockdep_pin_lock(&rq->lock);
6	ttwu_do_activate(rq, p, 0); // 【小节3.2.4】
7	lockdep_unpin_lock(&rq->lock);
8	raw_spin_unlock(&rq->lock);
9}

在kernel/sched/core.c目录中发现有大量ttwu_xxx的方法，这个单词缩写可真是厉害，琢磨一会结合上下文，才明白原来是try to wake up的缩写，另外为了简化，这里只介绍单处理器的逻辑。

3.2.4 ttwu_do_activate

1static void ttwu_do_activate(struct rq *rq, struct task_struct *p, int wake_flags)
2{
3	ttwu_activate(rq, p, ENQUEUE_WAKEUP | ENQUEUE_WAKING);
4	ttwu_do_wakeup(rq, p, wake_flags);
5}
6
7static inline void ttwu_activate(struct rq *rq, struct task_struct *p, int en_flags)
8{
9	activate_task(rq, p, en_flags); //将进程task加入rq队列
10	p->on_rq = TASK_ON_RQ_QUEUED;
11
12	if (p->flags & PF_WQ_WORKER)
13		wq_worker_waking_up(p, cpu_of(rq)); //worker正在唤醒中，则通知工作队列
14}
15
16static void ttwu_do_wakeup(struct rq *rq, struct task_struct *p, int wake_flags)
17{
18	check_preempt_curr(rq, p, wake_flags);
19	p->state = TASK_RUNNING; //标记该进程为TASK_RUNNING状态
20    ...
21}

四、总结

通过DECLARE_WAIT_QUEUE_HEAD(name)可初始化wait_queue_head_t结构体，通过DECLARE_WAITQUEUE可初始化wait_queue_t结构体，由等待队列头(wait_queue_head_t)和等待队列项(wait_queue_t)构建一个双向链表。可通过add_wait_queue和remove_wait_queue分别向双向链表中添加或删除等待项。

休眠与唤醒流程：

进程A调用wait_event(wq, condition)就是向等待队列头中添加等待队列项wait_queue_t，该该等待队列项中的成员变量private记录当前进程，其成员变量func记录唤醒回调函数，然后调用schedule()使当前进程进入休眠状态。
进程B调用wake_up(wq)会遍历整个等待列表wq中的每一项wait_queue_t，依次调用每一项的唤醒函数try_to_wake_up()。这个过程会将private记录的进程加入rq运行队列，并设置进程状态为TASK_RUNNING。
进程A被唤醒后只执行如下检测：
- 检查condition是否为true，满足条件则跳出循环，再把wait_queue_t从wq队列中移除；
- 检测该进程task的成员thread_info->flags是否被设置TIF_SIGPENDING，被设置则说明有待处理的信号，则跳出循环，再把wait_queue_t从wq队列中移除；
- 否则，继续调用schedule()再次进入休眠等待状态，如果wait_queue_t不在wq队列，则再次加入wq队列。

本文转自 http://gityuan.com/2018/12/02/linux-wait-queue/，如有侵权，请联系删除。