工作队列分析

爱被打了一巴掌 2022-09-19 04:29 152阅读 0赞

* 转载地址链接：http://www.360doc.com/content/13/0411/01/11645800\_277493188.shtml
 
 <table> 
 <tbody> 
 <tr> 
 <td>&nbsp;</td> 
 </tr> 
 <tr> 
 <td> 
 <table> 
 <tbody> 
 <tr> 
 <td> 
 <table> 
 <tbody> 
 <tr> 
 <td>一、用法&nbsp; structcpu_workqueue_struct {&nbsp; spinlock_t lock;&nbsp; long remove_sequence; /* Least-recently added (next to run) */&nbsp; long insert_sequence; /* Next to add */&nbsp; struct list_head worklist;&nbsp; wait_queue_head_t more_work;&nbsp; wait_queue_head_t work_done;&nbsp; struct workqueue_struct *wq;&nbsp; struct task_struct *thread;&nbsp; int run_depth; /* Detect run_workqueue() recursion depth */&nbsp; } ____cacheline_aligned;&nbsp; /*&nbsp; * The externally visible workqueue abstraction is an array of&nbsp; * per-CPU workqueues:&nbsp; */&nbsp; struct workqueue_struct {&nbsp; struct cpu_workqueue_struct *cpu_wq;&nbsp; const char *name;&nbsp; struct list_head list;&nbsp;&nbsp;/* Empty if single thread */&nbsp; };&nbsp; 工作队列的使用很简单。&nbsp; 1.首先是建立一个工作队列：&nbsp; struct workqueue_struct *keventd_wq;&nbsp; keventd_wq = create_workqueue("events");&nbsp; 2.然后就是在这个队列中insert你所要做的“工作”：&nbsp; DECLARE_WORK(work, func, data)&nbsp; queue_work(keventd_wq, work);&nbsp; struct work_struct {&nbsp; unsigned long pending;&nbsp; struct list_head entry;&nbsp; void (*func)(void *);&nbsp; void *data;&nbsp; void *wq_data;&nbsp; struct timer_list timer;&nbsp; };&nbsp; 初始化有两种方法。&nbsp; 一种为静态方法：&nbsp; #define __WORK_INITIALIZER(n, f, d) { \&nbsp; &nbsp; &nbsp;&nbsp; &nbsp;&nbsp;&nbsp;.entry = { &amp;(n).entry, &amp;(n).entry }, \&nbsp; .func = (f), \&nbsp; .data = (d), \&nbsp; .timer = TIMER_INITIALIZER(NULL, 0, 0), \&nbsp; }&nbsp; #define DECLARE_WORK(n, f, d) \&nbsp; struct work_struct n = __WORK_INITIALIZER(n, f, d)&nbsp; 另一种为动态方法：&nbsp; /*&nbsp; * initialize all of a work-struct:&nbsp; */&nbsp; #define INIT_WORK(_work, _func, _data) \&nbsp; do { \&nbsp; INIT_LIST_HEAD(&amp;(_work)-&gt;entry); \&nbsp; (_work)-&gt;pending = 0; \&nbsp; PREPARE_WORK((_work), (_func), (_data)); \&nbsp; init_timer(&amp;(_work)-&gt;timer); \&nbsp; } while (0)&nbsp; 二、执行过程&nbsp; create_workqueue() -&gt; __create_workqueue()&nbsp; struct workqueue_struct *__create_workqueue(const char *name,&nbsp; &nbsp; &nbsp; int singlethread)&nbsp; {&nbsp; int cpu, destroy = 0;&nbsp; struct workqueue_struct *wq;&nbsp; struct task_struct *p;&nbsp; wq = kzalloc(sizeof(*wq), GFP_KERNEL);&nbsp; //为每个CPU建立一个结构&nbsp; wq-&gt;cpu_wq = alloc_percpu(struct cpu_workqueue_struct);&nbsp; ...&nbsp; wq-&gt;name = name;&nbsp; mutex_lock(&amp;workqueue_mutex);&nbsp; if (singlethread) {&nbsp; ...&nbsp; } else {&nbsp; list_add(&amp;wq-&gt;list, &amp;workqueues);&nbsp; for_each_online_cpu(cpu) {&nbsp; //为每个CPU创建一个线程&nbsp; p = create_workqueue_thread(wq, cpu);&nbsp; if (p) {&nbsp; kthread_bind(p, cpu);&nbsp; //唤醒这个线程执行工作&nbsp; wake_up_process(p);&nbsp; } else&nbsp; destroy = 1;&nbsp; }&nbsp; }&nbsp; mutex_unlock(&amp;workqueue_mutex);&nbsp; ...&nbsp; return wq;&nbsp; }&nbsp; create_workqueue() -&gt; __create_workqueue() -&gt; create_workqueue_thread()&nbsp; static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq,&nbsp; &nbsp; &nbsp;int cpu)&nbsp; {&nbsp; struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq-&gt;cpu_wq, cpu);&nbsp; struct task_struct *p;&nbsp; spin_lock_init(&amp;cwq-&gt;lock);&nbsp; cwq-&gt;wq = wq;&nbsp; cwq-&gt;thread = NULL;&nbsp; cwq-&gt;insert_sequence = 0;&nbsp; cwq-&gt;remove_sequence = 0;&nbsp; INIT_LIST_HEAD(&amp;cwq-&gt;worklist);&nbsp; init_waitqueue_head(&amp;cwq-&gt;more_work);&nbsp; init_waitqueue_head(&amp;cwq-&gt;work_done);&nbsp; if (is_single_threaded(wq))&nbsp; p = kthread_create(worker_thread, cwq, "%s", wq-&gt;name);&nbsp; else&nbsp; //创建一个线程，这个线程以cwq为数据执行worker_thread这个函数&nbsp; p = kthread_create(worker_thread, cwq, "%s/%d", wq-&gt;name, cpu);&nbsp; if (IS_ERR(p))&nbsp; return NULL;&nbsp; cwq-&gt;thread = p;&nbsp;&nbsp;//&nbsp; return p;&nbsp; }&nbsp; create_workqueue() -&gt; __create_workqueue() -&gt; create_workqueue_thread() -&gt; worker_thread()&nbsp; //本函数在一个死循环等待工作的到来，这一般在睡眠状态中，等待被唤醒执行工作&nbsp; //当有工作到来时queue_work()会将这个线程唤醒&nbsp; static int worker_thread(void *__cwq)&nbsp; {&nbsp; struct cpu_workqueue_struct *cwq = __cwq;&nbsp; DECLARE_WAITQUEUE(wait, current);&nbsp; ...&nbsp; current-&gt;flags |= PF_NOFREEZE;&nbsp; //设置优先级&nbsp; set_user_nice(current, -5);&nbsp; ...&nbsp; set_current_state(TASK_INTERRUPTIBLE);&nbsp; while (!kthread_should_stop()) {&nbsp; //将本线程加入睡眠队列，用于睡眠后可以被唤醒&nbsp; add_wait_queue(&amp;cwq-&gt;more_work, &amp;wait);&nbsp; //如果没用被执行的“工作”，则将自己切换出去，进入睡眠状态&nbsp; if (list_empty(&amp;cwq-&gt;worklist))&nbsp; schedule();&nbsp; else //否则或是被唤醒&nbsp; __set_current_state(TASK_RUNNING);&nbsp; remove_wait_queue(&amp;cwq-&gt;more_work, &amp;wait);&nbsp; //工作队列非空，执行工作&nbsp; if (!list_empty(&amp;cwq-&gt;worklist))&nbsp; run_workqueue(cwq);&nbsp; set_current_state(TASK_INTERRUPTIBLE);&nbsp; }&nbsp; __set_current_state(TASK_RUNNING);&nbsp; return 0;&nbsp; }&nbsp; create_workqueue() -&gt; __create_workqueue() -&gt; create_workqueue_thread()&nbsp;&nbsp; -&gt; worker_thread() -&gt; run_workqueue()&nbsp; //该函数执行真正的工作&nbsp; static void run_workqueue(struct cpu_workqueue_struct *cwq)&nbsp; {&nbsp; unsigned long flags;&nbsp; spin_lock_irqsave(&amp;cwq-&gt;lock, flags);&nbsp; ...&nbsp; //顺次执行队列中的所有工作&nbsp; while (!list_empty(&amp;cwq-&gt;worklist)) {&nbsp; struct work_struct *work = list_entry(cwq-&gt;worklist.next,&nbsp; struct work_struct, entry);&nbsp; void (*f) (void *) = work-&gt;func;&nbsp; void *data = work-&gt;data;&nbsp; //从队列中删除待执行的任务&nbsp; list_del_init(cwq-&gt;worklist.next);&nbsp; spin_unlock_irqrestore(&amp;cwq-&gt;lock, flags);&nbsp; BUG_ON(work-&gt;wq_data != cwq);&nbsp; clear_bit(0, &amp;work-&gt;pending);&nbsp; //执行“工作”&nbsp; f(data);&nbsp; spin_lock_irqsave(&amp;cwq-&gt;lock, flags);&nbsp; cwq-&gt;remove_sequence++;&nbsp; wake_up(&amp;cwq-&gt;work_done);&nbsp;&nbsp;//&nbsp; }&nbsp; cwq-&gt;run_depth--;&nbsp; spin_unlock_irqrestore(&amp;cwq-&gt;lock, flags);&nbsp; }&nbsp; 三、工作线程创建的详细过程&nbsp; &nbsp; &nbsp;create_workqueue() -&gt; __create_workqueue() -&gt; create_workqueue_thread()&nbsp;&nbsp; -&gt; kthread_create()&nbsp; struct task_struct *kthread_create(int (*threadfn)(void *data),&nbsp; &nbsp; &nbsp;void *data,&nbsp; &nbsp; &nbsp;const char namefmt[],&nbsp; &nbsp; &nbsp;...)&nbsp; {&nbsp; //初始化用于创建线程的辅助结构&nbsp; struct kthread_create_info create;&nbsp; DECLARE_WORK(work, keventd_create_kthread, &amp;create);&nbsp; create.threadfn = threadfn;&nbsp; create.data = data;&nbsp; init_completion(&amp;create.started);&nbsp; init_completion(&amp;create.done);&nbsp; if (!helper_wq) //首先创建辅助工作队列&nbsp; work.func(work.data);&nbsp; else {&nbsp; //注意，“创建一个工作队列”这个工作本身又是属于helper_wq工作队列&nbsp; //的一项工作，所以，将这个工作加入的辅助工作队列中等待执行。&nbsp; queue_work(helper_wq, &amp;work);&nbsp; wait_for_completion(&amp;create.done);&nbsp; }&nbsp; ...&nbsp; return create.result;&nbsp; }&nbsp; create_workqueue() -&gt; __create_workqueue() -&gt; create_workqueue_thread()&nbsp;&nbsp; -&gt; kthread_create()-&gt; keventd_create_kthread()&nbsp; //最终会调用kernel_thread为每个工作队列创建一个线程&nbsp; //这样，被创建的线程会以create为数据执行kthread(如下),而kthread中则执行create中的threadfn(data)，&nbsp; //即为create_workqueue_thread中的worker_thread(cwq)，即为我们工作队列要执行的函数了。&nbsp; static void keventd_create_kthread(void *_create)&nbsp; {&nbsp; struct kthread_create_info *create = _create;&nbsp; int pid;&nbsp; /* We want our own signal handler (we take no signals by default). */&nbsp; pid = kernel_thread(kthread, create, CLONE_FS | CLONE_FILES | SIGCHLD);&nbsp; if (pid result = ERR_PTR(pid);&nbsp; } else {&nbsp; wait_for_completion(&amp;create-&gt;started);&nbsp; read_lock(&amp;tasklist_lock);&nbsp; create-&gt;result = find_task_by_pid(pid);&nbsp; read_unlock(&amp;tasklist_lock);&nbsp; }&nbsp; complete(&amp;create-&gt;done);&nbsp; }&nbsp; static int kthread(void *_create)&nbsp; {&nbsp; struct kthread_create_info *create = _create;&nbsp; int (*threadfn)(void *data);&nbsp; void *data;&nbsp; ...&nbsp; threadfn = create-&gt;threadfn;&nbsp; data = create-&gt;data;&nbsp; ...&nbsp; if (!kthread_should_stop())&nbsp; ret = threadfn(data);&nbsp; ...&nbsp; return 0;&nbsp; }&nbsp; 四、插入“工作”&nbsp; /* Preempt must be disabled. */&nbsp; static void __queue_work(struct cpu_workqueue_struct *cwq,&nbsp; struct work_struct *work)&nbsp; {&nbsp; unsigned long flags;&nbsp; spin_lock_irqsave(&amp;cwq-&gt;lock, flags);&nbsp; work-&gt;wq_data = cwq;&nbsp; //将当前工作插入到工作队列中待待执行&nbsp; list_add_tail(&amp;work-&gt;entry, &amp;cwq-&gt;worklist);&nbsp; cwq-&gt;insert_sequence++;&nbsp; wake_up(&amp;cwq-&gt;more_work);&nbsp;&nbsp;//唤醒相应线程&nbsp; spin_unlock_irqrestore(&amp;cwq-&gt;lock, flags);&nbsp; }&nbsp; int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)&nbsp; {&nbsp; int ret = 0, cpu = get_cpu();&nbsp; //如里当前工作未在队列中才插入&nbsp; if (!test_and_set_bit(0, &amp;work-&gt;pending)) {&nbsp; ...&nbsp; __queue_work(per_cpu_ptr(wq-&gt;cpu_wq, cpu), work);&nbsp; ret = 1;&nbsp; }&nbsp; put_cpu();&nbsp; return ret;&nbsp; }</td> 
 </tr> 
 </tbody> 
 </table></td> 
 </tr> 
 </tbody> 
 </table></td> 
 </tr> 
 </tbody> 
 </table>