阅读 125

带您进入内核开发的大门 | 内核中的等待队列

配套的代码可以从本号的github下载: github.com/shuningzhan…

等待队列是一种基于资源状态的线程管理的机制,它可以使线程在资源不满足的情况下处于休眠状态,让出CPU资源,而资源状态满足时唤醒线程,使其继续进行业务的处理。 等待队列(wait queue)用于使线程等待某一特定的事件发生而无需频繁的轮询,进程在等待期间睡眠,在某件事发生时由内核自动唤醒。它是以双循环链表为基础数据结构,与进程的休眠唤醒机制紧密相联,是实现异步事件通知、跨进程通信、同步资源访问等技术的底层技术支撑。

一

基本接口

wait_queue_head_t 使用等待队列时,最基本的数据结构是struct wait_queue_head_t,也就是等待队列头,这个可以理解为等待队列的实体。队列头中包含一个双向链表,用于记录在该等待队列中处于等待状态的线程等信息。该结构体的定义如下:

struct __wait_queue_head {
    spinlock_t        lock;  //用于互斥访问的自旋锁
    struct list_head    task_list;
};
typedef struct __wait_queue_head wait_queue_head_t;
复制代码

可以通过宏定义 DECLARE_WAIT_QUEUE_HEAD直接定义一个队列头变量,并完成初始化,该宏定义如下:

#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 } }
复制代码

或者是通过结构体wait_queue_head_t定义后,调用函数init_waitqueue_head进行初始化。虽然方式不同,但基本原理是一样的,主要是对结构体内自旋锁和链表的初始化。

wait_event 函数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())
复制代码

wake_up 函数wake_up用于对处于阻塞状态的线程进行唤醒,其参数就是队列头。如下是该函数的定义,我们这里暂时不展开介绍。

#define wake_up(x)            __wake_up(x, TASK_NORMAL, 1, NULL)
复制代码

了解了上面1个数据结构及相关函数后就可以使用等待队列了,当然只是基本的使用。

示例程序

我们这里给出一个示例程序,程序很简单。示例程序中有2个线程,分别是服务线程和客户线程。其中服务线程起来后会检查条件是否满足,并视情况进入休眠状态。而客户进程会每隔5秒将条件变成可用状态,并唤醒服务线程。

/* 这个例程用于说明等待队列的用法,在本例程中有2个线程,分别是
 * 客户端和服务端。逻辑很简单,服务线程起来的时候会等待事件发生
 * 并阻塞,客户端每隔5秒中唤醒一次服务端。*/
#include <linux/init.h>
#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/mm.h>

#include <linux/in.h>
#include <linux/inet.h>
#include <linux/socket.h>
#include <net/sock.h>
#include <linux/kthread.h>
#include <linux/sched.h>
#include <linux/wait.h>

#define BUF_SIZE 1024

struct task_struct *main_task;
struct task_struct *client_task;
wait_queue_head_t wqh;

/* 这个结构体用于在线程之间共享数据 */
struct thread_stat
{
        int t_can_run;
};

static inline void sleep(unsigned sec)
{
        __set_current_state(TASK_INTERRUPTIBLE);
        schedule_timeout(sec * HZ);
}

static int multhread_server(void *data)
{
        int index = 0;
        struct thread_stat* ts = (struct thread_stat*) data;

        while (!kthread_should_stop()) {
                printk(KERN_NOTICE "server run %d\n", index);
                index ++; 
                /*在这里等待事件, 线程被阻塞在这里。 */
                wait_event(wqh, ts->t_can_run || kthread_should_stop());
                printk(KERN_NOTICE "server event over!\n");
                ts->t_can_run = 0;
        }

        printk(KERN_NOTICE "server thread end\n");
        return 0;
}
static int multhread_init(void)
{
        ssize_t ret = 0;
        struct thread_stat thread_s;
        thread_s.t_can_run = 0;



        printk("Hello, multhread \n");
        /* 初始化等待队列头 */
        init_waitqueue_head(&wqh);

        /* 分别启动2个线程 */
        main_task = kthread_run(multhread_server,
                                &thread_s,
                                "multhread_server");
        if (IS_ERR(main_task)) {
                ret = PTR_ERR(main_task);
                goto failed;
        }

        client_task = kthread_run(multhread_client,
                                  &thread_s,
                                  "multhread_client");
        if (IS_ERR(client_task)) {
                ret = PTR_ERR(client_task);
                goto client_failed;
        }

        return ret;
client_failed:
        kthread_stop(main_task);

failed:
        return ret;
}

static void multhread_exit(void)
{
        printk("Bye!\n");
        kthread_stop(main_task);
        kthread_stop(client_task);
}

module_init(multhread_init);
module_exit(multhread_exit);

MODULE_LICENSE("GPL");
MODULE_AUTHOR("SunnyZhang<shuningzhang@126.com>");
复制代码

等待队列的原理

关于等待队列的原理,有3点需要重点说明,理解了这几点,也就能够比较清晰的理解等待队列的原理。这3点分别是数据结构、等待函数和唤醒函数。 我们这里还是从结构体说起。这里主要有2个结构体,前面已经有所介绍。其中wait_queue_head是等待队列头,定义如下:

struct wait_queue_head {
        spinlock_t              lock;
        struct list_head        head;
};
复制代码

这里主要是双向链表,所有处于等待状态的线程都被加入到该双向链表中。等后续唤醒时根据该链表中的数据进行唤醒。另外一个数据结构是wait_queue_entry,该结构体是一个等待项,这个结构体对于普通用户通常不必关系,因为内核的API对其进行了封装。

struct wait_queue_entry {
        unsigned int            flags;
        void                    *private;
        wait_queue_func_t       func;
        struct list_head        entry;
};
复制代码

其中前一个结构体的head成员和后一个结构体的entry成员配合,形成所谓的双向链表。我们先看一下其大概的结构,具体如下图所示。

1.png

关于等待函数 关于等待函数,前面给出了一部分定义,下面我们继续深入介绍。在介绍之前,我们先介绍一下其大概流程,本质上就是将当前线程状态设置为TASK_UNINTERRUPTIBLE状态,然后调用schedule函数将本线程调度出去。理解了这个原理,代码就很容易理解,下面是函数的实现:

#define __wait_event(wq_head, condition)                                        \
        (void)___wait_event(wq_head, condition, TASK_UNINTERRUPTIBLE, 0, 0,     \
                            schedule())
复制代码

直接调用的___wait_event函数,注意观察一下这个函数的几个参数,其中TASK_UNINTERRUPTIBLE是目标状态,而schedule则是在内部要调用的函数。

#define ___wait_event(wq_head, condition, state, exclusive, ret, cmd)           \
({                                                                              \
        __label__ __out;                                                        \
        struct wait_queue_entry __wq_entry;                                     \
        long __ret = ret;       /* explicit shadow */                           \
        /* 这里初始化了前文所说的第二个结构体,也就是等待队列项 */    \
        init_wait_entry(&__wq_entry, exclusive ? WQ_FLAG_EXCLUSIVE : 0);        \
        for (;;) {                                                              \
                /* 这个函数设置线程状态,并将等待队列项添加到等待队列中
                */
                long __int = prepare_to_wait_event(&wq_head, &__wq_entry, state);\
                /* 满足条件的情况下退出等待 */                        \
                if (condition)                                                  \
                        break;                                                  \
                                                                                \
                if (___wait_is_interruptible(state) && __int) {                 \
                        __ret = __int;                                          \
                        goto __out;                                             \
                }                                                               \
                /* 将线程调度出去 */                                   \
                cmd;                                                            \
        }                                         \
        /*将状态重新设置为TASK_RUNNING,并将队列项移出 */                      \
        finish_wait(&wq_head, &__wq_entry);                                     \
__out:  __ret;                                                                  \
})
复制代码

这个函数里面所调用的函数的具体实现就不再解释了,代码贴过来太冗余了,本身也比较简单。

关于唤醒函数 唤醒函数前面也做过简单介绍,我们这里直接进入主体,介绍其实现函数。

static void __wake_up_common_lock(struct wait_queue_head *wq_head, 
          unsigned int mode, 
          int nr_exclusive, int wake_flags, void *key)
{
        unsigned long flags;
        ... ...
        spin_lock_irqsave(&wq_head->lock, flags);
        nr_exclusive = __wake_up_common(wq_head, mode, 
                                            nr_exclusive, 
                                            wake_flags, key, &bookmark);
        spin_unlock_irqrestore(&wq_head->lock, flags);

        ...  ...
}
复制代码

具体实现在函数__wake_up_common中。代码比较长,我们这里删除不必要的代码,只保留必要的代码逻辑。

static int __wake_up_common(struct wait_queue_head *wq_head, unsigned int mode,
                        int nr_exclusive, int wake_flags, void *key,
                        wait_queue_entry_t *bookmark)
{
        wait_queue_entry_t *curr, *next;
        ... ...
        /* 主要是这个循环,完成所有等待线程的唤醒, 这里关键是调用func */
        list_for_each_entry_safe_from(curr, next, &wq_head->head, entry) {
                unsigned flags = curr->flags;
                int ret;
                /* 这个函数是在init_wait_entry中初始化的,函数的名字是
                 * autoremove_wake_function,主要完成线程唤醒的动作。 */
                ret = curr->func(curr, mode, wake_flags, key);
                if (ret < 0)
                        break;
                if (ret && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
                        break;
                ... ...
        }
        return nr_exclusive;
}

复制代码

相信介绍到这里,大家应该对等待队列有了比较清晰的认识。总结起来就是要等待的线程加入队列并休眠,当条件满足时有其它线程将处于休眠状态的线程唤醒

其它接口

本文只介绍了基本的接口,其实系统还提供了很多扩展功能接口,以wake_up为例,还包括如下接口:

#define wake_up(x)                      __wake_up(x, TASK_NORMAL, 1, NULL)
#define wake_up_nr(x, nr)               __wake_up(x, TASK_NORMAL, nr, NULL)
#define wake_up_all(x)                  __wake_up(x, TASK_NORMAL, 0, NULL)
#define wake_up_locked(x)               __wake_up_locked((x), TASK_NORMAL, 1)
#define wake_up_all_locked(x)           __wake_up_locked((x), TASK_NORMAL, 0)

#define wake_up_interruptible(x)        __wake_up(x, TASK_INTERRUPTIBLE, 1, 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)
#define wake_up_interruptible_sync(x)   __wake_up_sync((x), TASK_INTERRUPTIBLE, 1)
复制代码

接口比较多,这里就不一一介绍了,但使用方法是类似的。

关注下面的标签,发现更多相似文章
评论