改进的工作队列

#include
#include
#include
#include
#include
#include
#include
#include

#include "temp.h"

static struct workqueue_struct *sender;
static struct workqueue_struct *recver;

static DECLARE_WORK(sendwork, sendwork_func);
static DECLARE_WORK(recvwork, recvwork_func);

static wait_queue_head_t waitqueuehead;
static struct kmem_cache *cache;

static struct buff_struct_head head = {
        .prev = (struct buff_struct *)&head,
        .next = (struct buff_struct *)&head,
        .lock = SPIN_LOCK_UNLOCKED,
};

static void buff_queue_tail(struct buff_struct_head *list, struct buff_struct *newsk)
{
        struct buff_struct *prev, *next;
        unsigned long flags;

        spin_lock_irqsave(&list->lock, flags);
        next = (struct buff_struct *)list;
        prev = next->prev;
        newsk->next = next;
        newsk->prev = prev;
        next->prev  = prev->next = newsk;
        spin_unlock_irqrestore(&list->lock, flags);
}

static struct buff_struct * buff_peek(struct buff_struct_head *list_)
{
        struct buff_struct *list;
        unsigned long flags;

        spin_lock_irqsave(&list_->lock, flags);
        list = ((struct buff_struct *)list_)->next;
        if (list == (struct buff_struct *)list_)
                list = NULL;
        spin_unlock_irqrestore(&list_->lock, flags);

        return list;
}

static struct buff_struct * buff_dequeue(struct buff_struct_head *list)
{
        struct buff_struct *next, *prev, *result;
        unsigned long flags;

        spin_lock_irqsave(&list->lock, flags);
        prev = (struct buff_struct *) list;
        next = prev->next;
        result = NULL;
        if (next != prev) {
                result       = next;
                next         = next->next;
                next->prev   = prev;
                prev->next   = next;
                result->next = result->prev = NULL;
        }
        spin_unlock_irqrestore(&list->lock, flags);
        return result;
}

static void sendwork_func(struct work_struct *arg)
{
        struct buff_struct *t;
        char data[] = "hello linux";
        int len = (sizeof(struct buff_struct) + strlen(data) + DALIGN) & ~DALIGN;
        int i;

        if ((cache = kmem_cache_create("cache", len, 0, SLAB_HWCACHE_ALIGN, NULL)) == NULL) {
                err("kmem_cache_create");
                return;
        }

        for (i = 0; i < 1000; i++) {
                t = kmem_cache_alloc(cache, GFP_KERNEL);
                if (!t) {
                        err("kmem_cache_alloc");
                        continue;
                }

                memset(t, '\0', len);
                memcpy(t->data, data, strlen(data));

                buff_queue_tail(&head, t);
                wake_up(&waitqueuehead);
        }
}

static void recvwork_func(struct work_struct *arg)
{
        DECLARE_WAITQUEUE(waitqueue, current);
        struct buff_struct *t;
        int count = 0;

        while (1) {
                add_wait_queue(&waitqueuehead, &waitqueue);
                while (!buff_peek(&head)) {
                        set_current_state(TASK_INTERRUPTIBLE);
                        schedule();
                        if (signal_pending(current)) {
                                err("signal_pending");
                                continue;
                        }
                }
                set_current_state(TASK_RUNNING);
                remove_wait_queue(&waitqueuehead, &waitqueue);

                while ((t = buff_dequeue(&head)) != NULL) {
                        printk(KERN_ALERT "count = %d, t->data: %s\n", count, t->data);
                        kmem_cache_free(cache, t);
                        count++;
                }

                if (count == 1000)
                        break;
        }
}

static int workqueue_init(void)
{
        init_waitqueue_head(&waitqueuehead);

        sender = create_workqueue("sender");
        recver = create_workqueue("recver");

        queue_work(sender, &sendwork);
        queue_work(recver, &recvwork);

        return 0;
}

static void workqueue_exit(void)
{
        flush_workqueue(sender);
        flush_workqueue(recver);
        destroy_workqueue(sender);
        destroy_workqueue(recver);
        kmem_cache_destroy(cache);

        printk(KERN_ALERT "workqueue_exit\n");
}

module_init(workqueue_init);
module_exit(workqueue_exit);

MODULE_LICENSE("GPL");