linux内核工作队列讲解和源码详细注释

ee(wq);return NULL;} wq->name = name;mutex_lock(workqueue_mutex);if (singlethread) { // 使用create_workqueue宏时该参数始终为0 // 如果是单一线程模式， 在单线程中调用各个工作队列// 建立一个的工作队列内核线程INIT_LIST_HEAD(wq->list);// 建立工作队列的线程p = create_workqueue_thread(wq， singlethread_cpu);if (!p)

destroy = 1;else // 唤醒该线程wake_up_process(p);} else { // 链表模式， 将工作队列添加到工作队列链表list_add(wq->list， workqueues);// 为每个CPU建立一个工作队列线程for_each_online_cpu(cpu) { p = create_workqueue_thread(wq， cpu);if (p) { // 绑定CPU kthread_bind(p， cpu);// 唤醒线程wake_up_process(p);} else destroy = 1;} mutex_unlock(workqueue_mutex);/* * Was there any error during startup? If yes then clean up：*/ if (destroy) { // 建立线程失败， 释放工作队列destroy_workqueue(wq);wq = NULL;} return wq;} EXPORT_SYMBOL_GPL(__create_workqueue);

// 创建工作队列线程static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq，int cpu)

{ // 每个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);// 初始化等待队列more_work， 该队列处理要执行的工作结构init_waitqueue_head(cwq->more_work);// 初始化等待队列work_done， 该队列处理执行完的工作结构init_waitqueue_head(cwq->work_done);// 建立内核线程work_thread 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;} 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 d