linux內核工作隊列講解和源碼詳細注釋

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;// 降低進程優(yōu)先級， 工作進程不是個很緊急的進程，不和其他進程搶占CPU，通常在系統(tǒng)空閑時運行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);// 進入循環(huán)， 沒明確停止該進程就一直運行while (!kthread_should_stop()) { // 設置more_work等待隊列， 當有新work結構鏈入隊列中時會激發(fā)此等待隊列add_wait_queue(cwq->more_work， wait);if (list_empty(cwq->worklist))

// 工作隊列為空， 睡眠schedule();else // 進行運行狀態(tài)__set_current_state(TASK_RUNNING);// 刪除等待隊列remove_wait_queue(cwq->more_work， wait);// 按鏈表遍歷執(zhí)行工作任務if (!list_empty(cwq->worklist))

run_workqueue(cwq);// 執(zhí)行完工作， 設置進程是可中斷的， 重新循環(huán)等待工作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);// 統(tǒng)計已經遞歸調用了多少次了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節(jié)點的struct work_struct *work = list_entry(cwq->worklist.next，struct work_struct， entry);void (*f) (void *) = work->func;void *data = work->data;// 刪除節(jié)點， 同時節(jié)點中的list參數清空list_del_init(cwq->worklist.next);// 解鎖// 現(xiàn)在在執(zhí)行以下代碼時可以中斷，run_workqueue本身可能會重新被調用， 所以要判斷遞歸深度spin_unlock_irqrestore(cwq->lock， flags);BUG_ON(work->wq_data != cwq);// 工作結構已經不在鏈表中clear_bit(0， work->pending);// 執(zhí)行工作函數f(data);// 重新加鎖spin_lock_irqsave(cwq->lock， flags);// 執(zhí)行完的工作序列號遞增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);