线程的应用

上节中，我们成功地编译运行了一个linux模 块。可惜的是，它只有两个函数，hello_init在模块加载时调用，hello_exit 在模块卸载时调用。这样下去，模块纵使有天大的本事，也只能压缩在这两个函数中。为了避免这种悲剧发生，本节就来学习一种让模块在加载后能一直运行下去的 方法——内核线程。

要创建一个内核线程有许多种方法，我们这里要学的是最简单的一种。打开 include/linux/kthread.h，你就看到了它全部的API，一共三个函数。

struct task_struct kthread_run(int (*threadfn)(void *data),
    void *data, const char namefmt[],...);
int kthread_stop(struct task_struct *k);
int kthread_should_stop(void);

      kthread_run()负责内核线程的创建，参数包括入口函数 threadfn，参数data，线程名称namefmt。可以看到线程的名字可以是类似sprintf方式组成的字符串。如果实际看到 kthread.h文件，就会发现kthread_run实际是一个宏定义，它由kthread_create()和wake_up_process() 两部分组成，这样的好处是用kthread_run()创建的线程可以直接运行，使用方便。

      kthread_stop()负责结束创建的线程，参数是创建时返回的task_struct指针。kthread设置标志should_stop，并等 待线程主动结束，返回线程的返回值。线程可能在kthread_stop()调用前就结束。（经过实际验证，如果线程在kthread_stop()调用 之前就结束，之后kthread_stop()再调用会发生可怕地事情—调用kthread_stop()的进程crash！！之所以如此，是由于 kthread实现上的弊端，之后会专门写文章讲到）

      kthread_should_stop()返回should_stop标志。它用于创建的线程检查结束标志，并决定是否退出。线程完全可以在完成自己的 工作后主动结束，不需等待should_stop标志。

 

    下面就来尝试一下运行不息的内核线程吧。

1、把上节建立的hello子目录，复制为新的kthread子目录。

 

2、修改hello.c，使其内容如下。

#include <linux/init.h>
#include <linux/module.h>
#include <linux/kthread.h>

MODULE_LICENSE("Dual BSD/GPL");

static struct task_struct *tsk;

static int thread_function(void *data)
{
    int time_count = 0;
    do {
        printk(KERN_INFO "thread_function: %d times", ++time_count);
        msleep(1000);
    }while(!kthread_should_stop() && time_count<=30);
    return time_count;
}

static int hello_init(void)
{
    printk(KERN_INFO "Hello, world!\n");

    tsk = kthread_run(thread_function, NULL, "mythread%d", 1);
    if (IS_ERR(tsk)) {
        printk(KERN_INFO "create kthread failed!\n");
    }
    else {
        printk(KERN_INFO "create ktrhead ok!\n");
    }
    return 0;
}

static void hello_exit(void)
{
    printk(KERN_INFO "Hello, exit!\n");
    if (!IS_ERR(tsk)){
        int ret = kthread_stop(tsk);
        printk(KERN_INFO "thread function has run %ds\n", ret);
    }
}

module_init(hello_init);
module_exit(hello_exit);

为了不让创建的内核线程一直运行浪费CPU，代码中采用周期性延迟的方式，每次循 环用msleep(1000)延迟1s。为了防止线程一直运行下去，代码中使用了两个结束条件：一个是模块要求线程结束，一个是打印满一定次数，后者是为 了防止printk输出信息太多。最后在hello_exit中结束线程，并打印线程运行的时间。

这里要注意的是kthread_run的返回值tsk。不能用tsk是否为 NULL进行检查，而要用IS_ERR()宏定义检查，这是因为返回的是错误码，大致从0xfffff000~0xffffffff。 

3、编译运行模块，步骤参照前例。在运行过程中使用ps -e命令，可以看到有名字位mythread1的内核线程在运行。 

经过本节，我们学习了内核线程的创建使用方法，现在要创建一大堆的线程在内核中已 经易如反掌。你会逐渐相信，我们模块的拓展空间是无限的。 

附注：

      我们的重点在模块编程，不断学习内核API的使用。但如果能知其然，而知其所以然就更好了。所以有了文章后的附注部分。在附注部分，我们会尽量解释内核 API的实现原理，对相关linux内核代码做简单的分析，以帮助大家学习理解相关的代码。分析的代码包含在linux-2.6.32中，但这些代码在相 近版本中都变化不大。作者水平有限，请大家见谅。

      kthread的实现在kernel/kthread.c中，头文件是include/linux/kthread.h。内核中一直运行一个线程 kthreadd，它运行kthread.c中的kthreadd函数。在kthreadd()中，不断检查一个kthread_create_list 链表。kthread_create_list中的每个节点都是一个创建内核线程的请求，kthreadd()发现链表不为空，就将其第一个节点退出链 表，并调用create_kthread()创建相应的线程。create_kthread()则进一步调用更深层的kernel_thread()创建 线程，入口函数设在kthread()中。

      外界调用kthread_run创建运行线程。kthread_run是个宏定义，首先调用kthread_create()创建线程，如果创建成功，再 调用wake_up_process()唤醒新创建的线程。kthread_create()根据参数向kthread_create_list中发送一 个请求，并唤醒kthreadd，之后会调用wait_for_completion(&create.done)等待线程创建完成。新创建的线 程开始运行后，入口在kthread()，kthread()调用complete(&create->done)唤醒阻塞的模块进程，并 使用schedule()调度出去。kthread_create()被唤醒后，设置新线程的名称，并返回到kthread_run中。 kthread_run调用wake_up_process()重新唤醒新创建线程，此时新线程才开始运行kthread_run参数中的入口函数。

      外界调用kthread_stop()删除线程。kthread_stop首先设置结束标志should_stop，然后调用 wake_for_completion(&kthread->exited)上，这个其实是新线程task_struct上的 vfork_done，会在线程结束调用do_exit()时设置。

上节中我们已经掌握了创建大量内核线程的能力，可惜线程之间还缺乏配合。要知道学习ITC(inter thread communication)，和学习IPC(inter process communication)一样，不是件简单的事情。本节就暂且解释一种最简单的线程同步手段—completion。

打开include/linux/completion.h，你就会看到 completion使用的全部API。这里简单介绍一下。

struct completion{
    unsigned int done;
    wait_queue_head_t wait;
};

void init_completion(struct completion *x);
void wait_for_completion(struct completion *x);
void wait_for_completion_interruptible(struct completion *x);
void wait_for_completion_killable(struct completion *x);
unsigned long wait_for_completion_timeout(struct completion *x,
    unsigned long timeout);
unsigned long wait_for_completion_interruptible_timeout(struct completion *x,
    unsigned long timeout);
bool try_wait_for_completion(struct completion *x);
bool completion_done(struct completion *x);
void complete(struct completion *x);
void complete_all(struct completion *x);

       首先是struct completion的结构，由一个计数值和一个等待队列组成。我们就大致明白，completion是类似于信号量的东西，用 completion.done来表示资源是否可用，获取不到的线程会阻塞在completion.wait的等待队列上，直到其它线程释放 completion。这样理解在实现上不错，但我认为completion不是与具体的资源绑定，而是单纯作为一种线程间同步的机制，它在概念上要比信 号量清晰得多。以后会逐渐看到，线程间事件的同步大多靠completion，而资源临界区的保护大多靠信号量。所以说，completion是一种线程 间的约会。

           init_completion初始化 completion结构。初此之外，linux当然还有在定义变量时初始化的方法，都在completion.h中。

      wait_for_completion等待在completion上。如果加了interruptible，就表示线程等待可被外部发来的信号打断；如 果加了killable，就表示线程只可被kill信号打断；如果加了timeout，表示等待超出一定时间会自动结束等待，timeout的单位是系统 所用的时间片jiffies(多为1ms)。

      try_wait_for_completion则是非阻塞地获取completion。它相当于 wait_for_completion_timeout调用中的timeout值为0。

      completion_done检查是否有线程阻塞在completion上。但这个API并不准确，它只是检查completion.done是否为 0，为0则认为有线程阻塞。这个API并不会去检查实际的等待队列，所以用时要注意。

      complete唤醒阻塞在completion上的首个线程。

      complete_all唤醒阻塞在completion上的所有线程。它的实现手法很粗糙，把completion.done的值设为 UINT_MAX/2，自然所有等待的线程都醒了。所以如果complete_all之后还要使用这个completion，就要把它重新初始化。

 

      好，completion介绍完毕，下面就来设计我们的模块吧。

      我们模拟5个周期性线程的运行。每个周期性线程period_thread的周期各不相同，但都以秒为单位，有各自的completion变量。period_thread每个周期运行一次，然后等待在 自己的completion变量上。为了唤醒period_thread，我们使用一个watchdog_thread来模拟时钟，每隔1s watchdog_thread就会检查哪个period_thread下一周期是否到来，并用相应的completion唤醒线程。

下面就动手实现吧。

1、把上节建立的kthread子目录，复制为新的completion子目录。

2、修改hello.c，使其内容如下。

#include <linux/init.h>
#include <linux/module.h>
#include <linux/kthread.h>
#include <linux/completion.h>

MODULE_LICENSE("Dual BSD/GPL");

#define PERIOD_THREAD_NUM 5

static int periods[PERIOD_THREAD_NUM] =
    { 1, 2, 4, 8, 16 };

static struct task_struct *period_tsks[PERIOD_THREAD_NUM];

static struct task_struct watchdog_tsk;

static struct completion wakeups[PERIOD_THREAD_NUM];


static int period_thread(void *data)
{
    int k = (int)data;
    int count = -1;

    do{
        printk("thread%d: period=%ds, count=%d\n", k, periods[k], ++count);
        wait_for_completion(&wakeups[k]);
    }while(!kthread_should_stop());
    return count;
}

static int watchdog_thread(void *data)
{
    int k;
    int count = 0;

    do{
        msleep(1000);
        count++;
        for(k=0; k<PERIOD_THREAD_NUM; k++){
            if (count%periods[k] == 0)
                complete(&wakeups[k]);
        }
    }while(!kthread_should_stop());
    return count;
}

static int hello_init(void)
{
    int k;

    printk(KERN_INFO "Hello, world!\n");

    for(k=0; k<PERIOD_THREAD_NUM; k++){
        init_completion(&wakeups[k]);
    }

    watchdog_tsk = kthread_run(watchdog_thread, NULL, "watchdog_thread");

    if(IS_ERR(watchdog_tsk)){
        printk(KERN_INFO "create watchdog_thread failed!\n");
        return 1;
    }

    for(k=0; k<PERIOD_THREAD_NUM; k++){
        period_tsks[k] = kthread_run(period_thread, (void*)k, "period_thread%d", k);
        if(IS_ERR(period_tsks[k]))
            printk(KERN_INFO "create period_thread%d failed!\n", k);
    }
    return 0;
}

static void hello_exit(void)
{
    int k;
    int count[5], watchdog_count;

    printk(KERN_INFO "Hello, exit!\n");
    for(k=0; k<PERIOD_THREAD_NUM]; k++){
        count[k] = 0;
        if(!IS_ERR(period_tsks[k]))
            count[k] = kthread_stop(period_tsks[k]);
    }
    watchdog_count = 0;
    if(!IS_ERR(watchdog_tsk))
        watchdog_count = kthread_stop(watchdog_tsk);

    printk("running total time: %ds\n", watchdog_count);
    for(k=0; k<PERIOD_THREAD_NUM; k++)
        printk("thread%d: period %d, running %d times\n", k, periods[k], count[k]);
}

module_init(hello_init);
module_exit(hello_exit);

3、编译运行模块，步骤参照前例。为保持模块的简洁性，我们仍然使用了 kthread_stop结束线程，这种方法虽然简单，但在卸载模块时等待时间太长，而且这个时间会随线程个数和周期的增长而增长。 

4、使用统一的exit_flag标志来表示结束请求，hello_exit发送 completion信号给所有的周期线程，最后调用kthread_stop来回收线程返回值。这样所有的周期线程都是在被唤醒后看到 exit_flag，自动结束，卸载模块时间大大缩短。下面是改进过后的hello.c，之前的那个姑且叫做hello-v1.c好了。

#include <linux/init.h>
#include <linux/module.h>
#include <linux/kthread.h>
#include <linux/completion.h>

MODULE_LICENSE("Dual BSD/GPL");

#define PERIOD_THREAD_NUM 5

static int periods[PERIOD_THREAD_NUM] =
    { 1, 2, 4, 8, 16 };

static struct task_struct *period_tsks[PERIOD_THREAD_NUM];

static struct task_struct watchdog_tsk;

static struct completion wakeups[PERIOD_THREAD_NUM];

static int exit_flag = 0;

static int period_thread(void *data)
{
    int k = (int)data;
    int count = -1;

    do{
        printk("thread%d: period=%ds, count=%d\n", k, periods[k], ++count);
        wait_for_completion(&wakeups[k]);
    }while(!exit_flag);
    return count;
}

static int watchdog_thread(void *data)
{
    int k;
    int count = 0;

    do{
        msleep(1000);
        count++;
        for(k=0; k<PERIOD_THREAD_NUM; k++){
            if (count%periods[k] == 0)
                complete(&wakeups[k]);
        }
    }while(!exit_flag);
    return count;
}

static int hello_init(void)
{
    int k;

    printk(KERN_INFO "Hello, world!\n");

    for(k=0; k<PERIOD_THREAD_NUM; k++){
        init_completion(&wakeups[k]);
    }

    watchdog_tsk = kthread_run(watchdog_thread, NULL, "watchdog_thread");

    if(IS_ERR(watchdog_tsk)){
        printk(KERN_INFO "create watchdog_thread failed!\n");
        return 1;
    }

    for(k=0; k<PERIOD_THREAD_NUM; k++){
        period_tsks[k] = kthread_run(period_thread, (void*)k, "period_thread%d", k);
        if(IS_ERR(period_tsks[k]))
            printk(KERN_INFO "create period_thread%d failed!\n", k);
    }
    return 0;
}

static void hello_exit(void)
{
    int k;
    int count[5], watchdog_count;

    printk(KERN_INFO "Hello, exit!\n");
    exit_flag = 1;
    for(k=0; k<PERIOD_THREAD_NUM]; k++)
        complete_all(&wakeups[k]);

    for(k=0; k<PERIOD_THREAD_NUM]; k++){
        count[k] = 0;
        if(!IS_ERR(period_tsks[k]))
            count[k] = kthread_stop(period_tsks[k]);
    }
    watchdog_count = 0;
    if(!IS_ERR(watchdog_tsk))
        watchdog_count = kthread_stop(watchdog_tsk);

    printk("running total time: %ds\n", watchdog_count);
    for(k=0; k<PERIOD_THREAD_NUM; k++)
        printk("thread%d: period %d, running %d times\n", k, periods[k], count[k]);
}

module_init(hello_init);
module_exit(hello_exit);

5、编译运行改进过后的模块。可以看到模块卸载时间大大减少，不会超过1s。

经过本节，我们学会了一种内核线程间同步的机制—completion。线程们已 经开始注意相互配合，以完成复杂的工作。相信它们会越来越聪明的。

附注：

completion的实现在kernel/sched.c中。这里的每个API 都较短，实现也较为简单。completion背后的实现机制其实是等待队列。等待队列的实现会涉及到较多的调度问题，这里先简单略过。

通过之前几节，我们已经了解了内核线程的创建方法kthread，内核同步的工具completion。现在我们就来学学内核线程传递消息的方法 list。或许大家会说，list不是链表吗。不错，list是链表，但它可以变成承担消息传递的消息队列。消息的发送者把消息放到链表上，并通过同步工 具（如completion）通知接收线程，接收线程再从链表上取回消息，就这么简单。linux内核或许没有给我们定制好的东西，但却给了我们可随意变 换的、基础的工具，把这些工具稍加组合就能完成复杂的功能。list又是这些万能工具中最常用的。

前 面两篇文章的惯例是先对新增的功能做出介绍，并解释要用到的API。但我感觉这种既要解释原理，又要分析代码，又要写应用样例的十全文章，写起来实在吃 力，而且漏洞百出。与其如此，我还不如把两部分分开，这里的模块编程就专心设计模块，编写内核API的组合使用样例；而原理介绍、API代码分析的部分， 会转到linux内核部件分析的部分。这样我一方面能安心设计样例，一方面能把API介绍地更全面一些。

模块设计

本模块目的是展示list作为消息队列使用时的情况。所以会创建一个全局链表work_list，定义一种消息的结构struct work_event，并创建两个内核线程work_thread和watchdog_thread。work_thread是消息的接收者，它循环检查 work_list，如果其上有消息，就将其取出并执行，否则阻塞。watchdog_thread是消息的发送者，它周期性地发送消息到 work_list，并唤醒work_thread。 

模块实现

1、建立list子目录。

2、编写list.c，使其内容如下。

#include <linux/init.h>
#include <linux/module.h>
#include <linux/list.h>
#include <linux/completion.h>
#include <linux/kthread.h>

MODULE_LICENSE("Dual BSD/GPL");

static struct task_struct *work_tsk;

static struct task_struct *watchdog_tsk;

static DECLARE_SPINLOCK(work_list_lock);

static LIST_HEAD(work_list);

static DECLARE_COMPLETION(work_wait);

enum WORK_EVENT_TYPE {
    EVENT_TIMER,
    EVENT_EXIT
};

struct work_event {
    enum WORK_EVENT_TYPE type;
    int need_free;
    list_head list;
};

static int work_thread(void *data)
{
    int count = 0;

    while(1){
        if(list_empty(&work_list))
            wait_for_completion(&work_wait);
        spin_lock(&work_list_lock);
        while(!list_empty(&work_list)){
            struct work_event *event;
            event = list_entry(work_list.next, struct work_event, list);
            list_del(&event->list);
            spin_unlock(&work_list_lock);

            if(event->type == EVENT_TIMER){
                printk(KERN_INFO "event timer: count = %d\n", ++count);
            }
            else if (event->type == EVENT_EXIT){
                if(event->need_free)
                    kfree(event);
                goto exit;
            }

            if(event->need_free)
                kfree(event);
            spin_lock(&work_list_lock);
        }
    }
exit:
    return count;
}

static int watchdog_thread(void *data)
{
    int count = 0;
    while(!kthread_should_stop()){
        msleep(1000);
        count++;
        if(count%5 == 0){
            struct work_event *event;
            event = kmalloc(sizeof(struct work_event), GFP_KERNEL);
            if(event == NULL){
                printk(KERN_INFO "watchdog_thread: kmalloc failed!\n");
                break;
            }
            event->type = EVENT_TIMER;
            event->need_free = 1;
            spin_lock(&work_list_lock);
            list_add_tail(&event->list, &work_list);
            spin_unlock(&work_list_lock);
            complete(&work_wait);
        }
    }
    return count;
}

static int list_init()
{
    printk(KERN_INFO "list_init()\n");

    watchdog_tsk = kthread_run(watchdog_thread, NULL, "watchdog_thread");
    if(IS_ERR(watchdog_tsk))
        goto err1;

    work_tsk = kthread_run(work_thread, NULL, "work_thread");
    if(IS_ERR(work_tsk))
        goto err2;

    return 0;

err2:
    kthread_stop(watchdog_tsk);
err1:
    return 1;
}

static void list_exit()
{
    printk(KERN_INFO "list_exit()\n");

    if(!IS_ERR(watchdog_tsk)){
        int count = kthread_stop(watchdog_tsk);
        printk(KERN_INFO "watchdog_thread: running for %ss\n", count);
    }
    if(!IS_ERR(work_tsk)){
        get_task_struct(&work_tsk);
        struct work_event event;
        event.type = EVENT_EXIT;
        event.need_free = 0;
        spin_lock(&work_list_lock);
        list_add(&event.list, &work_list);
        spin_unlock(&work_list_lock);
        complete(&work_wait);

        int count = kthread_stop(work_tsk);
        printk(KERN_INFO "work_thread: period 5s, running %d times\n", count);
    }
}

module_init(list_init);
module_exit(list_exit);

整个模块较为简单，work_thread只是接收work_list中的消息并处理，所以在list_exit退出时也要给它发 EVENT_EXIT类型的消息，使其退出。至于在list_exit发消息之前调用的get_task_struct，实在是无奈之举。因为我们发送 EVENT_EXIT消息后work_thread会在kthread_stop调用前就马上结束，导致之后的kthread_stop错误。所以要先 get_task_struct防止work_thread退出后释放任务结构中的某些内容，虽然有对应的put_task_struct，但我们并未使 用，因为put_task_struct并未引出到符号表。当然，这种情况只是一时之举，等我们学习了更强大的线程同步机制，或者更灵活的线程管理方法， 就能把代码改得更流畅。

      注意到代码中使用spin_lock/spin_unlock来保护work_list。如果之前不了解spin_lock，很容易认为它不足以保护。实 际上spin_lock不止是使用自旋锁，在此之前还调用了preempt_disable来禁止本cpu任务调度。只要同步行为只发生在线程之 间，spin_lock就足以保护，如果有中断或者软中断参与进来，就需要用spin_lock_irqsave了。

3、 编译运行模块。

可能本节介绍的list显得过于质朴，但只有简单的东西才能长久保留。不久之后，我们或许不再用kthread，不再用completion，但 list一定会一直用下去。