Linux内核中实现生产者与消费者(避免无效唤醒)

本文关注的重点是，避免内核线程的无效唤醒，并且主要是关注消费者线程的设计。

因此，为了省事，这里关与生产者，消费者本身的处理流程可能不够严密。

1. 生产者

一个内核线程，每生产一个商品后，就唤醒消费者，然后自己睡眠1秒钟。

2. 消费者

一个内核线程，每当被唤醒后，就消费商品，然后进入睡眠。

对于消费者线程的这种设计，有几个好处：响应快，平时不占任何cpu。

但这种设计有一点要注意，那就是要避免线程的无效唤醒。如何实现，看看消费者线程的代码就知道了。

/*
 * kernel programming test code
 *
 * Copyright (C) 2014 Sun Mingbao <[email protected]>
 * Dual licensed under the MIT and/or GPL licenses.
 *
 */
 
#include <linux/init.h>
#include <linux/module.h>
#include <linux/types.h>
#include <linux/kernel.h>
#include <linux/kthread.h>
#include <linux/proc_fs.h>
#include <linux/string.h>


MODULE_AUTHOR("Sun Mingbao <[email protected]>");
MODULE_DESCRIPTION("kernel programming test code");
MODULE_VERSION("1.0");
MODULE_LICENSE("Dual MIT/GPL");


#define  MODULE_NAME    "test"


#define    WRITE_CONSOLE(fmt, args...) \
    do \
    { \
        printk(KERN_ALERT fmt,##args); \
    } while (0)


#define    DBG_PRINT(fmt, args...) \
    do \
    { \
        WRITE_CONSOLE(MODULE_NAME"_DBG:%s(%d)-%s:\n"fmt"\n", __FILE__,__LINE__,__FUNCTION__,##args); \
    } while (0)




static struct task_struct *consumer_thread;
static struct task_struct *producer_thread;


static u32 cnt_consumer, cnt_producer;
static int has_something_to_consume = 0;
static void consume()
{
    has_something_to_consume = 0;
    cnt_consumer++;
}


static void produce()
{
    has_something_to_consume = 1;
    cnt_producer++;
}


static int consumer_thread_func(void * data)
{
while (!kthread_should_stop()) 
    {
        if (has_something_to_consume)
        {
            consume();
        }


        set_current_state(TASK_INTERRUPTIBLE);
        if (has_something_to_consume)
        {
            set_current_state(TASK_RUNNING);
            continue;
        }


        schedule();
    }


    if (has_something_to_consume)
    {
        consume();
    }


}


static int producer_thread_func(void * data)
{


while (!kthread_should_stop()) 
    {
        produce();
        if (consumer_thread->state & TASK_INTERRUPTIBLE)
        {
            wake_up_process(consumer_thread);
        }
        
        set_current_state(TASK_INTERRUPTIBLE);
        schedule_timeout(HZ);
    }


}


static int __init create_threads(void)
{
    consumer_thread=kthread_run(consumer_thread_func, NULL, "consumer_thread");
    producer_thread=kthread_run(producer_thread_func, NULL, "producer_thread");


    return 0;
}


static struct proc_dir_entry *my_proc_dir;
    
static int misc_info_read_proc(char *buffer, char **start, off_t offset, int length, int *eof, void *data)
{
    int ret;
    static char proc_file_contents[128];
    static int proc_file_len = 0;
    if (0==offset || 0==proc_file_len)
    {
        proc_file_len=sprintf(proc_file_contents, "cnt_producer:%u\n""cnt_consumer:%u\n", cnt_producer, cnt_consumer);
    }
    
    ret=snprintf(buffer, length, "%s", proc_file_contents+offset);


    if(ret+offset==proc_file_len)
        *eof = 1;
    
    return ret;
}


static int __init create_my_proc_entries(void)
{
    my_proc_dir = proc_mkdir(MODULE_NAME, NULL);
    
    create_proc_read_entry("misc_info"
        ,0
        , my_proc_dir
        , misc_info_read_proc
        , NULL);


    return 0;
}


static void __exit remove_my_proc_entries(void)
{
    remove_proc_entry("misc_info", my_proc_dir);
    remove_proc_entry(MODULE_NAME, NULL);
}


static int __init test_init(void)
{
    int retval;


    DBG_PRINT("start");
    retval=create_threads();
    if (retval < 0)
    {
   goto EXIT;
    }


    create_my_proc_entries();
    
    DBG_PRINT("start succeed");
    
EXIT:
    return retval;
}




static void __exit stop_threads(void)
{
    kthread_stop(consumer_thread);
    kthread_stop(producer_thread);
}




static void __exit test_exit(void)
{
    DBG_PRINT("quit");
    remove_my_proc_entries();
    stop_threads();
}


module_init(test_init);
module_exit(test_exit);