国产一级a片免费看高清,亚洲熟女中文字幕在线视频,黄三级高清在线播放,免费黄色视频在线看

打開APP
userphoto
未登錄

開通VIP，暢享免費電子書等14項超值服

開通VIP

首頁

好書

留言交流

下載APP

聯(lián)系客服
使用工作隊列 - qinzhonghello的專欄

使用工作隊列

 

      我們先來看一下默認的events任務隊列，然后再看看創(chuàng)建新的工作者線程。

     1.創(chuàng)建推后的工作

      首先要做的是實際創(chuàng)建一些需要推后完成的工作?？梢酝ㄟ^DECLARE_WORK在編譯時靜態(tài)地創(chuàng)建該結構體：

  1. 在<workqueue.h>中
  2. #define DECLARE_WORK(n, f)                  
  3.     struct work_struct n = __WORK_INITIALIZER(n, f)
  4. #define __WORK_INITIALIZER(n, f) {              
  5.     .data = WORK_DATA_INIT(0),              \
  6.         .entry  = { &(n).entry, &(n).entry },           \
  7.     .func = (f),                        \
  8.     }

也可以在運行時通過指針創(chuàng)建一個工作：

  1. #define INIT_WORK(_work, _func)                 
  2.     do {                            \
  3.         (_work)->data = (atomic_long_t) WORK_DATA_INIT(0);  \
  4.         INIT_LIST_HEAD(&(_work)->entry);        \
  5.         PREPARE_WORK((_work), (_func));         \
  6.     } while (0)
  8. /*
  9.  * initialize a work item's function pointer
  10.  */
  11. #define PREPARE_WORK(_work, _func)              
  12.     do {                            \
  13.         (_work)->func = (_func);            \
  14.     } while (0)

2.工作隊列處理函數(shù)

 

       工作隊列對立函數(shù)的原型是：

         void work_handler(void *data)

       這個函數(shù)會由一個工作者線程執(zhí)行，因此，函數(shù)會運行在進程上下文中。默認情況下，運行響應中斷，并且不持有任何鎖。如果需要，函數(shù)可以睡眠。需要注意的是，盡管操作處理函數(shù)運行在進程上下文中，但它不能訪問用戶空間，因為內(nèi)核線程在用戶空間沒有相關的內(nèi)存映射。通常在系統(tǒng)調用發(fā)生時，內(nèi)核會代表用戶空間的進程運行，此時它才能訪問用戶空間，也只有在此時它才會映射用戶空間的內(nèi)存。

 

3.對工作進行調度

 

    要把給定工作的處理函數(shù)提交給默認的events工作線程，只須調用

    schedule_work(&work);

  1. /**
  2.  * schedule_work - put work task in global workqueue
  3.  * @work: job to be done
  4.  *
  5.  * This puts a job in the kernel-global workqueue.
  6.  */
  7. int fastcall schedule_work(struct work_struct *work)
  8. {
  9.     return queue_work(keventd_wq, work);
  10. }
  12. /**
  13.  * queue_work - queue work on a workqueue
  14.  * @wq: workqueue to use
  15.  * @work: work to queue
  16.  *
  17.  * Returns 0 if @work was already on a queue, non-zero otherwise.
  18.  *
  19.  * We queue the work to the CPU it was submitted, but there is no
  20.  * guarantee that it will be processed by that CPU.
  21.  */
  22. int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
  23. {
  24.     int ret = 0, cpu = get_cpu();
  26.     if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  27.         if (unlikely(is_single_threaded(wq)))
  28.             cpu = singlethread_cpu;
  29.         BUG_ON(!list_empty(&work->entry));
  30.         __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
  31.         ret = 1;
  32.     }
  33.     put_cpu();
  34.     return ret;
  35. }

   

  1. /* Preempt must be disabled. */
  2. static void __queue_work(struct cpu_workqueue_struct *cwq,
  3.              struct work_struct *work)
  4. {
  5.     unsigned long flags;
  7.     spin_lock_irqsave(&cwq->lock, flags);
  8.     set_wq_data(work, cwq);
  9.     list_add_tail(&work->entry, &cwq->worklist);
  10.     cwq->insert_sequence++;
  11.     wake_up(&cwq->more_work);
  12.     spin_unlock_irqrestore(&cwq->lock, flags);
  13. }

    work馬上就會被調度，一旦其所在的處理器上的工作者線程被喚醒，它就會被執(zhí)行。如不希望工作馬上被執(zhí)行，延遲一段時間之后再執(zhí)行，可以調度它在指定的時間執(zhí)行：

    schedule_delayed_work(&work,delay)；

  1. /**
  2.  * schedule_delayed_work - put work task in global workqueue after delay
  3.  * @dwork: job to be done
  4.  * @delay: number of jiffies to wait or 0 for immediate execution
  5.  *
  6.  * After waiting for a given time this puts a job in the kernel-global
  7.  * workqueue.
  8.  */
  9. int fastcall schedule_delayed_work(struct delayed_work *dwork,
  10.                     unsigned long delay)
  11. {
  12.     timer_stats_timer_set_start_info(&dwork->timer);
  13.     return queue_delayed_work(keventd_wq, dwork, delay);
  14. }
  16. /**
  17.  * queue_delayed_work - queue work on a workqueue after delay
  18.  * @wq: workqueue to use
  19.  * @dwork: delayable work to queue
  20.  * @delay: number of jiffies to wait before queueing
  21.  *
  22.  * Returns 0 if @work was already on a queue, non-zero otherwise.
  23.  */
  24. int fastcall queue_delayed_work(struct workqueue_struct *wq,
  25.             struct delayed_work *dwork, unsigned long delay)
  26. {
  27.     int ret = 0;
  28.     struct timer_list *timer = &dwork->timer;
  29.     struct work_struct *work = &dwork->work;
  31.     timer_stats_timer_set_start_info(timer);
  32.     if (delay == 0)
  33.         return queue_work(wq, work);
  35.     if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  36.         BUG_ON(timer_pending(timer));
  37.         BUG_ON(!list_empty(&work->entry));
  39.         /* This stores wq for the moment, for the timer_fn */
  40.         set_wq_data(work, wq);
  41.         timer->expires = jiffies + delay;
  42.         timer->data = (unsigned long)dwork;
  43.         timer->function = delayed_work_timer_fn;
  44.         add_timer(timer);
  45.         ret = 1;
  46.     }
  47.     return ret;
  48. }
  50. struct delayed_work {
  51.     struct work_struct work;
  52.     struct timer_list timer;
  53. };

 

4.刷新操作

 

   排入隊列的工作會在工作者線程下一次被喚醒的時候執(zhí)行。有時，在繼續(xù)下一步工作之前，你必須保證一些操作已經(jīng)執(zhí)行完畢了。這一點對模塊來說很重要，在卸載之前，它就有可能需要調用下面的函數(shù)；而在內(nèi)核的其他部分，為了防止競爭條件的出現(xiàn)，也可能需要確保不在有待處理的工作。

   出于以上目的，內(nèi)核準備了一個用于刷新指定工作隊列的函數(shù)：

    void flush_scheduled_work(void);

  1. void flush_scheduled_work(void)
  2. {
  3.     flush_workqueue(keventd_wq);
  4. }
  6. /**
  7.  * flush_workqueue - ensure that any scheduled work has run to completion.
  8.  * @wq: workqueue to flush
  9.  *
  10.  * Forces execution of the workqueue and blocks until its completion.
  11.  * This is typically used in driver shutdown handlers.
  12.  *
  13.  * This function will sample each workqueue's current insert_sequence number and
  14.  * will sleep until the head sequence is greater than or equal to that.  This
  15.  * means that we sleep until all works which were queued on entry have been
  16.  * handled, but we are not livelocked by new incoming ones.
  17.  *
  18.  * This function used to run the workqueues itself.  Now we just wait for the
  19.  * helper threads to do it.
  20.  */
  21. void fastcall flush_workqueue(struct workqueue_struct *wq)
  22. {
  23.     might_sleep();
  25.     if (is_single_threaded(wq)) {
  26.         /* Always use first cpu's area. */
  27.         flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, singlethread_cpu));
  28.     } else {
  29.         int cpu;
  31.         mutex_lock(&workqueue_mutex);
  32.         for_each_online_cpu(cpu)
  33.             flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
  34.         mutex_unlock(&workqueue_mutex);
  35.     }
  36. }

    函數(shù)會一直等待，直到隊列中所有對象都被執(zhí)行以后才返回。在等待所有待處理的工作執(zhí)行的時候，該函數(shù)會進入休眠狀態(tài)，所以只能在進程上下文中使用它。

    注意，該函數(shù)并不取消任何延遲執(zhí)行的工作。就是說，任何通過schedule_delayed_work()調度的工作，如果其延遲時間未結束，它并不會因為調用flush_scheduled_work()而被刷新掉。

      取消延遲執(zhí)行的工作應該調用：

 int cancle_delayed_work(sruct work_struc work);

       這個函數(shù)可以取消任何與work_struct相關的掛起工作。

  1. 在<workqueue.h>中
  2. /*
  3.  * Kill off a pending schedule_delayed_work().  Note that the work callback
  4.  * function may still be running on return from cancel_delayed_work().  Run
  5.  * flush_scheduled_work() to wait on it.
  6.  */
  7. static inline int cancel_delayed_work(struct delayed_work *work)
  8. {
  9.     int ret;
  11.     ret = del_timer_sync(&work->timer);
  12.     if (ret)
  13.         work_release(&work->work);
  14.     return ret;
  15. }
  16. /**
  17.  * work_release - Release a work item under execution
  18.  * @work: The work item to release
  19.  *
  20.  * This is used to release a work item that has been initialised with automatic
  21.  * release mode disabled (WORK_STRUCT_NOAUTOREL is set).  This gives the work
  22.  * function the opportunity to grab auxiliary data from the container of the
  23.  * work_struct before clearing the pending bit as the work_struct may be
  24.  * subject to deallocation the moment the pending bit is cleared.
  25.  *
  26.  * In such a case, this should be called in the work function after it has
  27.  * fetched any data it may require from the containter of the work_struct.
  28.  * After this function has been called, the work_struct may be scheduled for
  29.  * further execution or it may be deallocated unless other precautions are
  30.  * taken.
  31.  *
  32.  * This should also be used to release a delayed work item.
  33.  */
  34. #define work_release(work) 
  35.     clear_bit(WORK_STRUCT_PENDING, work_data_bits(work))

5.創(chuàng)建新的工作隊列

 

         如果默認的隊列不能滿足需要，可以創(chuàng)建一個新的工作對列和與之相應的工作者線程。

         創(chuàng)建一個新的任務隊列和與之相關的工作者線程，只須調用一個簡單的函數(shù)：

         struct workqueue_struct *create_workqueue(const char *name)；

  1. 在<workqueue.h>中
  2. #define create_workqueue(name) __create_workqueue((name), 0, 0)
  4. 在<workqueue.c>中
  6. struct workqueue_struct *__create_workqueue(const char *name,
  7.                         int singlethread, int freezeable)
  8. {
  9.     int cpu, destroy = 0;
  10.     struct workqueue_struct *wq;
  11.     struct task_struct *p;
  13.     wq = kzalloc(sizeof(*wq), GFP_KERNEL);
  14.     if (!wq)
  15.         return NULL;
  17.     wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
  18.     if (!wq->cpu_wq) {
  19.         kfree(wq);
  20.         return NULL;
  21.     }
  23.     wq->name = name;
  24.     mutex_lock(&workqueue_mutex);
  25.     if (singlethread) {
  26.         INIT_LIST_HEAD(&wq->list);
  27.         p = create_workqueue_thread(wq, singlethread_cpu, freezeable);
  28.         if (!p)
  29.             destroy = 1;
  30.         else
  31.             wake_up_process(p);
  32.     } else {
  33.         list_add(&wq->list, &workqueues);
  34.         for_each_online_cpu(cpu) {
  35.             p = create_workqueue_thread(wq, cpu, freezeable);
  36.             if (p) {
  37.                 kthread_bind(p, cpu);
  38.                 wake_up_process(p);
  39.             } else
  40.                 destroy = 1;
  41.         }
  42.     }
  43.     mutex_unlock(&workqueue_mutex);
  45.     /*
  46.      * Was there any error during startup? If yes then clean up:
  47.      */
  48.     if (destroy) {
  49.         destroy_workqueue(wq);
  50.         wq = NULL;
  51.     }
  52.     return wq;
  53. }

     這個函數(shù)會創(chuàng)建所有的工作者線程（系統(tǒng)中的每個處理器都有一個），并且做好所有開始處理工作之前的準備工作。

     創(chuàng)建一個工作的時候無須考慮工作隊列的類型?？梢允褂孟铝泻瘮?shù)對給定工作而不是默認的event隊列進行操作。

int queue_work(struct workqueue_struct *wq, struct work_struct*work);

int queue_delayed_work(struct workqueue_struct *wq, structwork_struct *work, unsigned long delay);

flush_workqueue(struct workqueue_struct *wq);

  1. /**
  2.  * queue_work - queue work on a workqueue
  3.  * @wq: workqueue to use
  4.  * @work: work to queue
  5.  *
  6.  * Returns 0 if @work was already on a queue, non-zero otherwise.
  7.  *
  8.  * We queue the work to the CPU it was submitted, but there is no
  9.  * guarantee that it will be processed by that CPU.
  10.  */
  11. int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
  12. {
  13.     int ret = 0, cpu = get_cpu();
  15.     if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  16.         if (unlikely(is_single_threaded(wq)))
  17.             cpu = singlethread_cpu;
  18.         BUG_ON(!list_empty(&work->entry));
  19.         __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
  20.         ret = 1;
  21.     }
  22.     put_cpu();
  23.     return ret;
  24. }
  26. /**
  27.  * queue_delayed_work - queue work on a workqueue after delay
  28.  * @wq: workqueue to use
  29.  * @dwork: delayable work to queue
  30.  * @delay: number of jiffies to wait before queueing
  31.  *
  32.  * Returns 0 if @work was already on a queue, non-zero otherwise.
  33.  */
  34. int fastcall queue_delayed_work(struct workqueue_struct *wq,
  35.             struct delayed_work *dwork, unsigned long delay)
  36. {
  37.     int ret = 0;
  38.     struct timer_list *timer = &dwork->timer;
  39.     struct work_struct *work = &dwork->work;
  41.     timer_stats_timer_set_start_info(timer);
  42.     if (delay == 0)
  43.         return queue_work(wq, work);
  45.     if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  46.         BUG_ON(timer_pending(timer));
  47.         BUG_ON(!list_empty(&work->entry));
  49.         /* This stores wq for the moment, for the timer_fn */
  50.         set_wq_data(work, wq);
  51.         timer->expires = jiffies + delay;
  52.         timer->data = (unsigned long)dwork;
  53.         timer->function = delayed_work_timer_fn;
  54.         add_timer(timer);
  55.         ret = 1;
  56.     }
  57.     return ret;
  58. }
  60. /**
  61.  * flush_workqueue - ensure that any scheduled work has run to completion.
  62.  * @wq: workqueue to flush
  63.  *
  64.  * Forces execution of the workqueue and blocks until its completion.
  65.  * This is typically used in driver shutdown handlers.
  66.  *
  67.  * This function will sample each workqueue's current insert_sequence number and
  68.  * will sleep until the head sequence is greater than or equal to that.  This
  69.  * means that we sleep until all works which were queued on entry have been
  70.  * handled, but we are not livelocked by new incoming ones.
  71.  *
  72.  * This function used to run the workqueues itself.  Now we just wait for the
  73.  * helper threads to do it.
  74.  */
  75. void fastcall flush_workqueue(struct workqueue_struct *wq)
  76. {
  77.     might_sleep();
  79.     if (is_single_threaded(wq)) {
  80.         /* Always use first cpu's area. */
  81.         flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, singlethread_cpu));
  82.     } else {
  83.         int cpu;
  85.         mutex_lock(&workqueue_mutex);
  86.         for_each_online_cpu(cpu)
  87.             flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
  88.         mutex_unlock(&workqueue_mutex);
  89.     }
  90. }
  2. static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
  3. {
  4.     if (cwq->thread == current) {
  5.         /*
  6.          * Probably keventd trying to flush its own queue. So simply run
  7.          * it by hand rather than deadlocking.
  8.          */
  9.         run_workqueue(cwq);
  10.     } else {
  11.         DEFINE_WAIT(wait);
  12.         long sequence_needed;
  14.         spin_lock_irq(&cwq->lock);
  15.         sequence_needed = cwq->insert_sequence;
  17.         while (sequence_needed - cwq->remove_sequence > 0) {
  18.             prepare_to_wait(&cwq->work_done, &wait,
  19.                     TASK_UNINTERRUPTIBLE);
  20.             spin_unlock_irq(&cwq->lock);
  21.             schedule();
  22.             spin_lock_irq(&cwq->lock);
  23.         }
  24.         finish_wait(&cwq->work_done, &wait);
  25.         spin_unlock_irq(&cwq->lock);
  26.     }
  27. }
  2. static void run_workqueue(struct cpu_workqueue_struct *cwq)
  3. {
  4.     unsigned long flags;
  6.     /*
  7.      * Keep taking off work from the queue until
  8.      * done.
  9.      */
  10.     spin_lock_irqsave(&cwq->lock, flags);
  11.     cwq->run_depth++;
  12.     if (cwq->run_depth > 3) {
  13.         /* morton gets to eat his hat */
  14.         printk("%s: recursion depth exceeded: %d\n",
  15.             __FUNCTION__, cwq->run_depth);
  16.         dump_stack();
  17.     }
  18.     while (!list_empty(&cwq->worklist)) {
  19.         struct work_struct *work = list_entry(cwq->worklist.next,
  20.                         struct work_struct, entry);
  21.         work_func_t f = work->func;
  23.         list_del_init(cwq->worklist.next);
  24.         spin_unlock_irqrestore(&cwq->lock, flags);
  26.         BUG_ON(get_wq_data(work) != cwq);
  27.         if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
  28.             work_release(work);
  29.         f(work);
  31.         if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
  32.             printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
  33.                     "%s/0x%08x/%d\n",
  34.                     current->comm, preempt_count(),
  35.                         current->pid);
  36.             printk(KERN_ERR "    last function: ");
  37.             print_symbol("%s\n", (unsigned long)f);
  38.             debug_show_held_locks(current);
  39.             dump_stack();
  40.         }
  42.         spin_lock_irqsave(&cwq->lock, flags);
  43.         cwq->remove_sequence++;
  44.         wake_up(&cwq->work_done);
  45.     }
  46.     cwq->run_depth--;
  47.     spin_unlock_irqrestore(&cwq->lock, flags);
  48. }
本站僅提供存儲服務，所有內(nèi)容均由用戶發(fā)布，如發(fā)現(xiàn)有害或侵權內(nèi)容，請點擊舉報。
打開APP，閱讀全文并永久保存 查看更多類似文章
猜你喜歡
類似文章
Linux工作隊列實現(xiàn)機制
Linux kernel 中的work queue原理
工作隊列(workqueue)
linux內(nèi)核分析筆記------上半部與下半部(下)
workqueue
linux 內(nèi)核work工作隊列實現(xiàn)
更多類似文章 >>
生活服務
分享 收藏 導長圖 關注 下載文章
綁定賬號成功
后續(xù)可登錄賬號暢享VIP特權！
如果VIP功能使用有故障，
可點擊這里聯(lián)系客服！

聯(lián)系客服