原始碼解讀Linux等待佇列
摘要：
從原始碼角度來解讀Linux等待佇列機制，瞭解休眠與喚醒的運轉原理
kernel/include/linux/wait.h
kernel/kernel/sched/wait.c
kernel/include/linux/sched.h
kernel/kernel/sched/core.c ...
從原始碼角度來解讀Linux等待佇列機制，瞭解休眠與喚醒的運轉原理
kernel/include/linux/wait.h kernel/kernel/sched/wait.c kernel/include/linux/sched.h kernel/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
struct __wait_queue_head {
spinlock_tlock;//用於互斥訪問的自旋鎖
struct list_headtask_list;
};
typedef struct __wait_queue_head wait_queue_head_t;
可通過巨集 DECLARE_WAIT_QUEUE_HEAD(name) 來建立型別為wait_queue_head_t的等待佇列頭name。
#define DECLARE_WAIT_QUEUE_HEAD(name) \
struct wait_queue_head name = __WAIT_QUEUE_HEAD_INITIALIZER(name)
#define __WAIT_QUEUE_HEAD_INITIALIZER(name) {\
.lock= __SPIN_LOCK_UNLOCKED(name.lock),\
.head= { &(name).head, &(name).head } }
2.2 struct wait_queue_t
struct __wait_queue {
unsigned intflags;
void*private;//指向等待佇列的程序task_struct
wait_queue_func_tfunc;//喚醒函式
struct list_headtask_list; //連結串列元素，將wait_queue_t掛到wait_queue_head_t
};
typedef struct __wait_queue wait_queue_t;
可通過巨集 DECLARE_WAITQUEUE(name, tsk) 來建立型別為wait_queue_t的等待佇列項name，並將tsk賦值給成員變數private， default_wake_function賦值給成員變數func。
#define DECLARE_WAITQUEUE(name, tsk)\
struct wait_queue_entry name = __WAITQUEUE_INITIALIZER(name, tsk)
#define __WAITQUEUE_INITIALIZER(name, tsk) {\
.private= tsk,\
.func= default_wake_function,\
.entry= { NULL, NULL } }
2.3 add_wait_queue
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);
}
static inline void __add_wait_queue(wait_queue_head_t *head, wait_queue_t *new)
{
list_add(&new->task_list, &head->task_list);
}
該方法的功能是將wait等待佇列項 掛到等待佇列頭q中。
2.4 remove_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);
}
該方法主要功能是將wait等待佇列項 從等待佇列頭q中移除。
到這裡，已經介紹了wait_queue_head_t和wait_queue_t這兩個建立方法，以及增加和刪除等待佇列元素的方法。接下里說一說如何在等待佇列的基礎上建立休眠與喚醒機制。
三、休眠與喚醒
3.1 wait_event
#define wait_event(wq, condition)\
do {\
if (condition)\
break;\
__wait_event(wq, condition);\
} while (0)
#define __wait_event(wq, condition)\
(void)___wait_event(wq, condition, TASK_UNINTERRUPTIBLE, 0, 0, schedule())
將___wait_event()巨集展開如下所示
___wait_event(wq, condition, state, exclusive, ret, cmd){
wait_queue_t __wait;
INIT_LIST_HEAD(&__wait.task_list);
for (;;) {
//當檢測程序是否有待處理訊號則返回值__int不為0，【見3.1.1】
long __int = prepare_to_wait_event(&wq, &__wait, state);
if (condition)//當滿足條件，則跳出迴圈
break;
//當有待處理訊號且程序處於可中斷狀態(TASK_INTERRUPTIBLE或TASK_KILLABLE))，則跳出迴圈
if (___wait_is_interruptible(state) && __int) {
__ret = __int;
break;
}
cmd; //schedule()，進入睡眠，從程序就緒佇列選擇一個高優先順序程序來代替當前程序執行
}
finish_wait(&wq, &__wait);//如果__wait還位於佇列wq，則將__wait從wq中移除
}
3.1.1 prepare_to_wait_event
再來看看進入睡眠狀態之前的準備工作，用於防止wait沒有佇列中，而事件已產生，則會無線等待。
long prepare_to_wait_event(wait_queue_head_t *q, wait_queue_t *wait, int state)
{
unsigned long flags;
if (signal_pending_state(state, current)) //訊號檢測
return -ERESTARTSYS;
wait->private = current;
wait->func = autoremove_wake_function; //設定func喚醒函式，【小節3.1.2】
spin_lock_irqsave(&q->lock, flags);
if (list_empty(&wait->task_list)) {//當wait不在佇列q，則加入其中，防止無法喚醒
if (wait->flags & WQ_FLAG_EXCLUSIVE)
__add_wait_queue_tail(q, wait);
else
__add_wait_queue(q, wait);
}
set_current_state(state);//設定程序狀態
spin_unlock_irqrestore(&q->lock, flags);
return 0;
}
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
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); //從列表中移除wait
return ret;
}
3.2 wake_up
#define wake_up(x)__wake_up(x, TASK_NORMAL, 1, NULL)
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);
}
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;
//呼叫喚醒函式 【小節3.2.1】
if (curr->func(curr, mode, wake_flags, key) &&
(flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
break;
}
}
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
再來看看喚醒函式函式
int default_wake_function(wait_queue_t *curr, unsigned mode, int wake_flags,
void *key)
{
//獲取wait所對應的程序 【小節3.2.2】
return try_to_wake_up(curr->private, mode, wake_flags);
}
3.2.2 try_to_wake_up
try_to_wake_up(struct task_struct *p, unsigned int state, int wake_flags)
{
unsigned long flags;
int cpu, src_cpu, success = 0;
bool freq_notif_allowed = !(wake_flags & WF_NO_NOTIFIER);
bool check_group = false;
wake_flags &= ~WF_NO_NOTIFIER;
smp_mb__before_spinlock();
raw_spin_lock_irqsave(&p->pi_lock, flags); //關閉本地中斷
src_cpu = cpu = task_cpu(p);
//如果當前程序狀態不屬於可喚醒狀態集，則無法喚醒該程序
//wake_up()傳遞過來的TASK_NORMAL等於(TASK_INTERRUPTIBLE | TASK_UNINTERRUPTIBLE)
if (!(p->state & state))
goto out;
success = 1;
smp_rmb();
if (p->on_rq && ttwu_remote(p, wake_flags)) //當前程序已處於rq執行佇列，則無需喚醒
goto stat;
...
ttwu_queue(p, cpu); //【小節3.2.3】
stat:
ttwu_stat(p, cpu, wake_flags);
out:
raw_spin_unlock_irqrestore(&p->pi_lock, flags); //恢復本地中斷
...
return success;
}
3.2.3 ttwu_queue
static void ttwu_queue(struct task_struct *p, int cpu)
{
struct rq *rq = cpu_rq(cpu); // 獲取當前程序的執行佇列
raw_spin_lock(&rq->lock);
lockdep_pin_lock(&rq->lock);
ttwu_do_activate(rq, p, 0); // 【小節3.2.4】
lockdep_unpin_lock(&rq->lock);
raw_spin_unlock(&rq->lock);
}
在kernel/sched/core.c目錄中發現有大量ttwu_xxx的方法，這個單詞縮寫可真是厲害，琢磨一會結合上下文，才明白原來是try to wake up的縮寫，另外為了簡化，這裡只介紹單處理器的邏輯。
3.2.4 ttwu_do_activate
static void ttwu_do_activate(struct rq *rq, struct task_struct *p, int wake_flags)
{
ttwu_activate(rq, p, ENQUEUE_WAKEUP | ENQUEUE_WAKING);
ttwu_do_wakeup(rq, p, wake_flags);
}
static inline void ttwu_activate(struct rq *rq, struct task_struct *p, int en_flags)
{
activate_task(rq, p, en_flags); //將程序task加入rq佇列
p->on_rq = TASK_ON_RQ_QUEUED;
if (p->flags & PF_WQ_WORKER)
wq_worker_waking_up(p, cpu_of(rq)); //worker正在喚醒中，則通知工作佇列
}
static void ttwu_do_wakeup(struct rq *rq, struct task_struct *p, int wake_flags)
{
check_preempt_curr(rq, p, wake_flags);
p->state = TASK_RUNNING; //標記該程序為TASK_RUNNING狀態
...
}
四、總結
通過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，該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佇列。
喜歡的話請幫忙轉發一下能讓更多有需要的人看到吧，有些技術上的問題大家可以多探討一下。
