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;} 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;// 降低进程优先级， 工作进程不是个很紧急的进程，不和其他进程抢占CPU，通常在系统空闲时运行set_user_nice(current， -5);/* Block and flush all signals */ // 阻塞所有信号sigfillset(blocked);sigprocmask(SIG_BLOCK， blocked， NULL);flush_signals(current);/* * We inherited MPOL_INTERLEAVE from the booting kernel. * Set MPOL_DEFAULT to insure node local allocations. */ numa_default_policy();/* SIG_IGN makes children autoreap： see do_notify_parent()。 */ // 信号处理都是忽略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()) { // 设置more_work等待队列， 当有新work结构链入队列中时会激发此等待队列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;}

// 运行工作结构static void run_workqueue(struct cpu_workqueue_struct *cwq)

{ unsigned long flags;/* * Keep taking off work from the queue until * done. */ // 加锁spin_lock_irqsave(cwq->lock， flags);// 统计已经递归调用了多少次了cwq->run_depth++;if (cwq->run_depth > 3) { // 递归调用此时太多/* morton gets to eat his hat */ printk(%s： recursion depth exceeded： %dn，__FUNCTION__， cwq->run_depth);dump_stack();} // 遍历工作链表while (!list_empty(cwq->worklist)) { // 获取的是next节点的struct work_struct *work = list_entry(cwq->worklist.next，struct work_struct， entry);void (*f) (void *) = work->func;void *data = work->data;// 删除节点， 同时节点中的list参数清空list_del_init(cwq->worklist.next);// 解锁// 现在在执行以下代码时可以中断，run_workqueue本身可能会重新被调用， 所以要判断递归深度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);}

4.2 释放工作队列

/** * destroy_workqueue - safely terminate a workqueue * @wq： target workqueue * * Safely destroy a workqueue. All work currently pending will be done first. */ void destroy_workqueue(struct workqueue_struct *wq)

{ int cpu;// 清除当前工作队列中的所有工作flush_workqueue(wq);/* We don't need the distraction of CPUs appearing and vanishing. */ mutex_lock(workqueue_mutex);// 结束该工作队列的线程if (is_single_threaded(wq))

cleanup_workqueue_thread(wq， singlethread_cpu);else { for_each_online_cpu(cpu)

cleanup_workqueue_thread(wq， cpu);list_del(wq->list);} mutex_unlock(workqueue_mutex);// 释放工作队列中对应每个CPU的工作队列数据free_percpu(wq->cpu_wq);kfree(wq);} EXPORT_SYMBOL_GPL(destroy_workqueue);

/** * flush_workqueue - ensure that any scheduled work has run to completion. * @wq： workqueue to flush * * Forces execution of the workqueue and blocks until its completion. * This is typically used in driver shutdown handlers. * * This function will sample each workqueue's current insert_sequence number and * will sleep until the head sequence is greater than or equal to that. This * means that we sleep until all works which were queued on entry have been * handled， but we are not livelocked by new incoming ones. * * This function used to run the workqueues itself. Now we just wait for the * helper threads to do it. */ void fastcall flush_workqueue(struct workqueue_struct *wq)

{ // 该进程可以睡眠might_sleep();// 清空每个CPU上的工作队列if (is_single_threaded(wq)) { /* Always use first cpu's area. */ flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq， singlethread_cpu));} else { int cpu;mutex_lock(workqueue_mutex);for_each_online_cpu(cpu)

flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq， cpu));mutex_unlock(workqueue_mutex);} EXPORT_SYMBOL_GPL(flush_workqueue);