linux工作队列和异步io_linux 工作队列线程和io线程

工作队列：
内核中所有的工作队列都是由helper_wq工作队列创建的，那么helper_wq是谁创建的呢？答案是直接执行的。看一下代码：
static __init int helper_init(void)
{
         helper_wq = create_singlethread_workqueue("kthread");
         BUG_ON(!helper_wq);
          return 0;
}
core_initcall(helper_init);
看看create_workqueue函数是怎么实现的：
struct workqueue_struct *__create_workqueue(const char *name, int singlethread)
{
         int cpu, destroy = 0;
         struct workqueue_struct *wq;
         struct task_struct *p;
         wq = kzalloc(sizeof(*wq), GFP_KERNEL);
         if (!wq)
                 return NULL;
         wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
         if (!wq->cpu_wq) {
                 kfree(wq);
                 return NULL;
         }
         wq->name = name;
         lock_cpu_hotplug();
         if (singlethread) {
                 INIT_LIST_HEAD(&wq->list);
                 p = create_workqueue_thread(wq, singlethread_cpu);
                 if (!p)
                         destroy = 1;
                 else
                         wake_up_process(p);
         } else {
                 spin_lock(&workqueue_lock);
                 list_add(&wq->list, &workqueues);
                 spin_unlock(&workqueue_lock);
                 for_each_online_cpu(cpu) {
                         p = create_workqueue_thread(wq, cpu);
                         if (p) {
                                 kthread_bind(p, cpu);
                                 wake_up_process(p);
                         } else
                                 destroy = 1;
                 }
         }
         unlock_cpu_hotplug();
         if (destroy) {
                 destroy_workqueue(wq);
                 wq = NULL;
         }
         return wq;
}
其中wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct)这一句创建了
struct cpu_workqueue_struct {
         spinlock_t lock;
         long remove_sequence;  
         long insert_sequence;  
         struct list_head worklist;
         wait_queue_head_t more_work;
         wait_queue_head_t work_done;
         struct workqueue_struct *wq;
         task_t *thread;
         int run_depth;         
} ____cacheline_aligned;
以上的这个结构是每个cpu结构，它是内嵌在
struct workqueue_struct {
         struct cpu_workqueue_struct *cpu_wq;
         const char *name;
         struct list_head list;  /* Empty if single thread */
};
的cpu_wq数组字段里面的。作为消遣，看看__alloc_percpu函数：
void *__alloc_percpu(size_t size)
{
         int i;
         struct percpu_data *pdata = kmalloc(sizeof(*pdata), GFP_KERNEL);
          if (!pdata)
                 return NULL;
         for_each_possible_cpu(i) {
                 int node = cpu_to_node(i);
                  if (node_online(node))
                         pdata->ptrs[i] = kmalloc_node(size, GFP_KERNEL, node);
                 else
                         pdata->ptrs[i] = kmalloc(size, GFP_KERNEL);
                  if (!pdata->ptrs[i])
                         goto unwind_oom;
                 memset(pdata->ptrs[i], 0, size);
         }
         return (void *)(~(unsigned long)pdata);
  unwind_oom:
         while (--i >= 0) {
                 if (!cpu_possible(i))
                         continue;
                 kfree(pdata->ptrs[i]);
         }
         kfree(pdata);
         return NULL;
}
下面开始真正的重要过程：
static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq,
                                                    int cpu)
{
         struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
         struct task_struct *p;
          spin_lock_init(&cwq->lock);
         cwq->wq = wq;
         cwq->thread = NULL;
         cwq->insert_sequence = 0;
         cwq->remove_sequence = 0;
         INIT_LIST_HEAD(&cwq->worklist);
         init_waitqueue_head(&cwq->more_work);
         init_waitqueue_head(&cwq->work_done);
          if (is_single_threaded(wq))
                 p = kthread_create(worker_thread, cwq, "%s", wq->name);
         else
                 p = kthread_create(worker_thread, cwq, "%s/%d", wq->name, cpu);
         if (IS_ERR(p))
                 return NULL;
         cwq->thread = p;
         return p;
}
struct task_struct *kthread_create(int (*threadfn)(void *data),
                                    void *data,
                                    const char namefmt[],
                                    ...)
{
         struct kthread_create_info create;
         DECLARE_WORK(work, keventd_create_kthread, &create);
          create.threadfn = threadfn;
         create.data = data;
         init_completion(&create.started);
         init_completion(&create.done);
          if (!helper_wq)//注意这一句，如果没有helper_wq，那么就直接执行回调函数，这里就是keventd_create_kthread
                 work.func(work.data);
         else {
                 queue_work(helper_wq, &work);
                 wait_for_completion(&create.done);
         }
         if (!IS_ERR(create.result)) {
                 va_list args;
                 va_start(args, namefmt);
                 vsnprintf(create.result->comm, sizeof(create.result->comm),
                           namefmt, args);
                 va_end(args);
         }
          return create.result;
}
上面的过程确实有点绕，实际上work.func和queue_work调用执行的是同一个过程，就是执行keventd_create_kthread函数，而前面传入的threadfn也就是worker_thread则是封装进了kthread_create_info结构里面了，要想知道往下该做什么了，就要看keventd_create_kthread函数：
static void keventd_create_kthread(void *_create)
{
         struct kthread_create_info *create = _create;
         int pid;
         pid = kernel_thread(kthread, create, CLONE_FS | CLONE_FILES | SIGCHLD);
         if (pid < 0) {
                 create->result = ERR_PTR(pid);
         } else {
                 wait_for_completion(&create->started);
                 read_lock(&tasklist_lock);
                 create->result = find_task_by_pid(pid);
                 read_unlock(&tasklist_lock);
         }
         complete(&create->done);
}
到这里才真正启动一个内核线程kthread，然后这个内核线程函数以create为参数，到这儿，几乎可以猜出接下来要做什么了
static int kthread(void *_create)
{
         struct kthread_create_info *create = _create;
         int (*threadfn)(void *data);
         void *data;
         sigset_t blocked;
         int ret = -EINTR;
          kthread_exit_files();
         threadfn = create->threadfn;
         data = create->data;
         sigfillset(&blocked);
         sigprocmask(SIG_BLOCK, &blocked, NULL);
         flush_signals(current);
          set_cpus_allowed(current, CPU_MASK_ALL);
          __set_current_state(TASK_INTERRUPTIBLE);
         complete(&create->started);
         schedule();
          if (!kthread_should_stop())
                 ret = threadfn(data);
          if (kthread_should_stop()) {
                 kthread_stop_info.err = ret;
                 complete(&kthread_stop_info.done);
         }
         return 0;
}
唉，最终threadfn(data)搞定了一切，这个threadfn(data)实际上就是
static int worker_thread(void *__cwq)
{
         struct cpu_workqueue_struct *cwq = __cwq;
         DECLARE_WAITQUEUE(wait, current);
         struct k_sigaction sa;
         sigset_t blocked;
         current->flags |= PF_NOFREEZE;
         set_user_nice(current, -5);
         sigfillset(&blocked);
         sigprocmask(SIG_BLOCK, &blocked, NULL);
         flush_signals(current);
         sa.sa.sa_handler = SIG_IGN;
         sa.sa.sa_flags = 0;
         siginitset(&sa.sa.sa_mask, sigmask(SIGCHLD));
         do_sigaction(SIGCHLD, &sa, (struct k_sigaction *)0);
         set_current_state(TASK_INTERRUPTIBLE);
         while (!kthread_should_stop()) {
                 add_wait_queue(&cwq->more_work, &wait);
                 if (list_empty(&cwq->worklist))
                         schedule();
                 else
                         __set_current_state(TASK_RUNNING);
                 remove_wait_queue(&cwq->more_work, &wait);
                 if (!list_empty(&cwq->worklist))
                         run_workqueue(cwq);
                 set_current_state(TASK_INTERRUPTIBLE);
         }
         __set_current_state(TASK_RUNNING);
         return 0;
}
再来一层调用就完事了：run_workqueue(cwq)
static void run_workqueue(struct cpu_workqueue_struct *cwq)
{
         unsigned long flags;
         spin_lock_irqsave(&cwq->lock, flags);
         cwq->run_depth++;
         if (cwq->run_depth > 3) {
                 printk("%s: recursion depth exceeded: %d/n",
                         __FUNCTION__, cwq->run_depth);
                 dump_stack();
         }
         while (!list_empty(&cwq->worklist)) {
                 struct work_struct *work = list_entry(cwq->worklist.next,
                                                struct work_struct, entry);
                 void (*f) (void *) = work->func;
                 void *data = work->data;
                 list_del_init(cwq->worklist.next);
                 spin_unlock_irqrestore(&cwq->lock, flags);
                 BUG_ON(work->wq_data != cwq);
                 clear_bit(0, &work->pending);
                 f(data);
                 spin_lock_irqsave(&cwq->lock, flags);
                 cwq->remove_sequence++;
                 wake_up(&cwq->work_done);
         }
         cwq->run_depth--;
         spin_unlock_irqrestore(&cwq->lock, flags);
}
以上就是工作队列的全过程，这是一个完美的层次嵌套结构，很完美！如果任何时间想用工作队列解决一些问题那么你只需要做两件事情，一是建立一个工作队列，二是往里面排入任务。
看完了上述代码，下面看一个应用，就是linux的异步io。
static int __init aio_setup(void)
{
         kiocb_cachep = kmem_cache_create("kiocb", sizeof(struct kiocb),
                                 0, SLAB_HWCACHE_ALIGN|SLAB_PANIC, NULL, NULL);
         kioctx_cachep = kmem_cache_create("kioctx", sizeof(struct kioctx),
                                 0, SLAB_HWCACHE_ALIGN|SLAB_PANIC, NULL, NULL);
         aio_wq = create_workqueue("aio");
         pr_debug("aio_setup: sizeof(struct page) = %d/n", (int)sizeof(struct page));
         return 0;
}
在新内核里，睡眠唤醒和2.4的不同了：为了更加灵活，用回调函数来实现。
#define wake_up(x)   __wake_up(x, TASK_UNINTERRUPTIBLE | TASK_INTERRUPTIBLE, 1, NULL)
void fastcall __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 sync, void *key)
{
         struct list_head *tmp, *next;
         list_for_each_safe(tmp, next, &q->task_list) {
                 wait_queue_t *curr;
                 unsigned flags;
                 curr = list_entry(tmp, wait_queue_t, task_list);
                 flags = curr->flags;
                 if (curr->func(curr, mode, sync, key) &&
                     (flags & WQ_FLAG_EXCLUSIVE) &&!--nr_exclusive)
                         break;
         }
}
看看curr->func(curr, mode, sync, key) 这一句，wait_queue_t结构体里面加入了一个函数指针。对于一般的唤醒，就是default_wake_function函数，而对于异步io，就是aio_wake_function：
static int aio_wake_function(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
         struct kiocb *iocb = container_of(wait, struct kiocb, ki_wait);
         list_del_init(&wait->task_list);
         kick_iocb(iocb);
         return 1;
}
kiocb结构是与异步io相关的结构，暂且现在不用深究。接着看kick_iocb
void fastcall kick_iocb(struct kiocb *iocb)
{
         if (is_sync_kiocb(iocb)) {
                 kiocbSetKicked(iocb);
                 wake_up_process(iocb->ki_obj.tsk);
                 return;
         }
         try_queue_kicked_iocb(iocb);
}
重点就是try_queue_kicked_iocb：
static void try_queue_kicked_iocb(struct kiocb *iocb)
{
         struct kioctx   *ctx = iocb->ki_ctx;
         unsigned long flags;
         int run = 0;
         BUG_ON((!list_empty(&iocb->ki_wait.task_list)));
         spin_lock_irqsave(&ctx->ctx_lock, flags);
         if (!kiocbTryKick(iocb))
                 run = __queue_kicked_iocb(iocb);
         spin_unlock_irqrestore(&ctx->ctx_lock, flags);
         if (run)
                 aio_queue_work(ctx);
}
__queue_kicked_iocb只是和队列相关的操作，而重点是aio_queue_work函数：
static void aio_queue_work(struct kioctx * ctx)
{
         unsigned long timeout;
         smp_mb();
         if (waitqueue_active(&ctx->wait))
                 timeout = 1;
         else
                 timeout = HZ/10;
         queue_delayed_work(aio_wq, &ctx->wq, timeout);
}
queue_delayed_work应该很熟悉了，就是往前面系统初始化的时候建立的工作队列aio_wq里排入具体的任务ctx->wq那么，真正调度到该工作队列的时候该怎么办呢？答案是调用aio_kick_handler，这需要在某个地方初始化这个工作队列元素：
INIT_WORK(&ctx->wq, aio_kick_handler, ctx)；
下面看一下这个函数：
static void aio_kick_handler(void *data)
{
         struct kioctx *ctx = data;
         mm_segment_t oldfs = get_fs();
         int requeue;
         set_fs(USER_DS);
         use_mm(ctx->mm); //切换到提交aio进程的地址空间，因为O_DIRECT的io不需要内核缓存
         spin_lock_irq(&ctx->ctx_lock);
         requeue =__aio_run_iocbs(ctx);
         unuse_mm(ctx->mm);
         spin_unlock_irq(&ctx->ctx_lock);
         set_fs(oldfs);
         if (requeue)
                 queue_work(aio_wq, &ctx->wq);
}
__aio_run_iocbs(ctx)是真正完成任务的操作
static int __aio_run_iocbs(struct kioctx *ctx)
{
         struct kiocb *iocb;
         LIST_HEAD(run_list);
         assert_spin_locked(&ctx->ctx_lock);
         list_splice_init(&ctx->run_list, &run_list);
         while (!list_empty(&run_list)) {
                 iocb = list_entry(run_list.next, struct kiocb,
                         ki_run_list);
                 list_del(&iocb->ki_run_list);               
                 iocb->ki_users++;       /* grab extra reference */
                 aio_run_iocb(iocb);
                 if (__aio_put_req(ctx, iocb))  /* drop extra ref */
                         put_ioctx(ctx);
         }
         if (!list_empty(&ctx->run_list))
                 return 1;
         return 0;
}
进一步 aio_run_iocb：
static ssize_t aio_run_iocb(struct kiocb *iocb)
{
         struct kioctx   *ctx = iocb->ki_ctx;
         ssize_t (*retry)(struct kiocb *);
         ssize_t ret;
         if (iocb->ki_retried++ > 1024*1024) {
…               

         }
         if (!(iocb->ki_retried & 0xff))

…

         if (!(retry = iocb->ki_retry))

…

         kiocbClearKicked(iocb);
         iocb->ki_run_list.next = iocb->ki_run_list.prev = NULL;
         spin_unlock_irq(&ctx->ctx_lock);
         if (kiocbIsCancelled(iocb)) {
                 ret = -EINTR;
                 aio_complete(iocb, ret, 0);
                 goto out;
         }
         BUG_ON(current->io_wait != NULL);
         current->io_wait = &iocb->ki_wait;
         ret = retry(iocb);
         current->io_wait = NULL;
         if (ret != -EIOCBRETRY && ret != -EIOCBQUEUED) {
                 BUG_ON(!list_empty(&iocb->ki_wait.task_list));
                 aio_complete(iocb, ret, 0);
         }
out:
         spin_lock_irq(&ctx->ctx_lock);
         if (-EIOCBRETRY == ret) {
                 INIT_LIST_HEAD(&iocb->ki_run_list);
                 if (kiocbIsKicked(iocb)) {
                         __queue_kicked_iocb(iocb);
                         aio_queue_work(ctx);
                 }
         }
         return ret;
}
这个函数本质上就是执行retry回调函数，完成没有完成的工作。异步io和普通io的区别就是在于发生组赛的时候并不睡眠或发生调度，而是直接返回一个错误代码，(在AIO下，进程并不会sleep，而是简单地将一个含有特殊func的wait_queue_t挂入等待队列就返回了)。这里应用工作队列的方式令人赞叹。 那么还是看看普通的唤醒函数比较心安：
int default_wake_function(wait_queue_t *curr, unsigned mode, int sync,
                           void *key)
{
         task_t *p = curr->private;
         return try_to_wake_up(p, mode, sync);
}
就是这么简单，简简单单唤醒进程。在信号的传递结束以后，调用
int fastcall wake_up_process(task_t *p)
{
         return try_to_wake_up(p, TASK_STOPPED | TASK_TRACED |
                                  TASK_INTERRUPTIBLE | TASK_UNINTERRUPTIBLE, 0);
}
唤醒接受信号的进程。下面看一下关于等待队列的事情：
static inline void init_waitqueue_head(wait_queue_head_t *q)
{
         spin_lock_init(&q->lock);
         INIT_LIST_HEAD(&q->task_list);
}
static inline void INIT_LIST_HEAD(struct list_head *list)
{
         list->next = list;
         list->prev = list;
}
********************
void fastcall __sched sleep_on(wait_queue_head_t *q)
{
         SLEEP_ON_VAR //初始化一个wait_queue_t
         current->state = TASK_UNINTERRUPTIBLE;
         SLEEP_ON_HEAD //将这个wait_queue_t加入睡眠队列
         schedule();
         SLEEP_ON_TAIL //睡眠完毕，从睡眠队列清除
}
就是这么简单。