History

workqueue: implement worker states

Implement worker states. After created, a worker is STARTED. While a worker isn't processing a work, it's IDLE and chained on gcwq->idle_list. While processing a work, a worker is BUSY and chained on gcwq->busy_hash. Also, gcwq now counts the number of all workers and idle ones. worker_thread() is restructured to reflect state transitions. cwq->more_work is removed and waking up a worker makes it check for events. A worker is killed by setting DIE flag while it's IDLE and waking it up. This gives gcwq better visibility of what's going on and allows it to find out whether a work is executing quickly which is necessary to have multiple workers processing the same cwq.

Tejun Heo committed 16 years ago

Commit c8e55f31 parent 8b03ae31 file · +173 −41Browse files

Showing the changes to kernel/workqueue.c (1 of 1 file). Show all

Unified · Split

kernel/workqueue.c+173−41
@@ -35,6 +35,17 @@ · 34 unmodified lines
3535 #include <linux/lockdep.h>
3636 #include <linux/idr.h>
3737
38+enum {
39+ /* worker flags */
40+ WORKER_STARTED = 1 << 0, /* started */
41+ WORKER_DIE = 1 << 1, /* die die die */
42+ WORKER_IDLE = 1 << 2, /* is idle */
43+
44+ BUSY_WORKER_HASH_ORDER = 6, /* 64 pointers */
45+ BUSY_WORKER_HASH_SIZE = 1 << BUSY_WORKER_HASH_ORDER,
46+ BUSY_WORKER_HASH_MASK = BUSY_WORKER_HASH_SIZE - 1,
47+};
48+
3849 /*
3950 * Structure fields follow one of the following exclusion rules.
4051 *
@@ -51,11 +62,18 @@ struct global_cwq; · 10 unmodified lines
5162 struct cpu_workqueue_struct;
5263
5364 struct worker {
65+ /* on idle list while idle, on busy hash table while busy */
66+ union {
67+ struct list_head entry; /* L: while idle */
68+ struct hlist_node hentry; /* L: while busy */
69+ };
70+
5471 struct work_struct *current_work; /* L: work being processed */
5572 struct list_head scheduled; /* L: scheduled works */
5673 struct task_struct *task; /* I: worker task */
5774 struct global_cwq *gcwq; /* I: the associated gcwq */
5875 struct cpu_workqueue_struct *cwq; /* I: the associated cwq */
76+ unsigned int flags; /* L: flags */
5977 int id; /* I: worker id */
6078 };
6179
@@ -65,6 +83,15 @@ struct worker { · 3 unmodified lines
6583 struct global_cwq {
6684 spinlock_t lock; /* the gcwq lock */
6785 unsigned int cpu; /* I: the associated cpu */
86+
87+ int nr_workers; /* L: total number of workers */
88+ int nr_idle; /* L: currently idle ones */
89+
90+ /* workers are chained either in the idle_list or busy_hash */
91+ struct list_head idle_list; /* L: list of idle workers */
92+ struct hlist_head busy_hash[BUSY_WORKER_HASH_SIZE];
93+ /* L: hash of busy workers */
94+
6895 struct ida worker_ida; /* L: for worker IDs */
6996 } ____cacheline_aligned_in_smp;
7097
@@ -77,7 +104,6 @@ struct global_cwq { · 6 unmodified lines
77104 struct cpu_workqueue_struct {
78105 struct global_cwq *gcwq; /* I: the associated gcwq */
79106 struct list_head worklist;
80− wait_queue_head_t more_work;
81107 struct worker *worker;
82108 struct workqueue_struct *wq; /* I: the owning workqueue */
83109 int work_color; /* L: current color */
@@ -306,6 +332,33 @@ static inline struct cpu_workqueue_struct *get_wq_data(struct work_struct *work) · 222 unmodified lines
306332 WORK_STRUCT_WQ_DATA_MASK);
307333 }
308334
335+/**
336+ * busy_worker_head - return the busy hash head for a work
337+ * @gcwq: gcwq of interest
338+ * @work: work to be hashed
339+ *
340+ * Return hash head of @gcwq for @work.
341+ *
342+ * CONTEXT:
343+ * spin_lock_irq(gcwq->lock).
344+ *
345+ * RETURNS:
346+ * Pointer to the hash head.
347+ */
348+static struct hlist_head *busy_worker_head(struct global_cwq *gcwq,
349+ struct work_struct *work)
350+{
351+ const int base_shift = ilog2(sizeof(struct work_struct));
352+ unsigned long v = (unsigned long)work;
353+
354+ /* simple shift and fold hash, do we need something better? */
355+ v >>= base_shift;
356+ v += v >> BUSY_WORKER_HASH_ORDER;
357+ v &= BUSY_WORKER_HASH_MASK;
358+
359+ return &gcwq->busy_hash[v];
360+}
361+
309362 /**
310363 * insert_work - insert a work into cwq
311364 * @cwq: cwq @work belongs to
@@ -332,7 +385,7 @@ static void insert_work(struct cpu_workqueue_struct *cwq, · 20 unmodified lines
332385 smp_wmb();
333386
334387 list_add_tail(&work->entry, head);
335− wake_up(&cwq->more_work);
388+ wake_up_process(cwq->worker->task);
336389 }
337390
338391 static void __queue_work(unsigned int cpu, struct workqueue_struct *wq,
@@ -470,13 +523,59 @@ int queue_delayed_work_on(int cpu, struct workqueue_struct *wq, · 131 unmodified lines
470523 }
471524 EXPORT_SYMBOL_GPL(queue_delayed_work_on);
472525
526+/**
527+ * worker_enter_idle - enter idle state
528+ * @worker: worker which is entering idle state
529+ *
530+ * @worker is entering idle state. Update stats and idle timer if
531+ * necessary.
532+ *
533+ * LOCKING:
534+ * spin_lock_irq(gcwq->lock).
535+ */
536+static void worker_enter_idle(struct worker *worker)
537+{
538+ struct global_cwq *gcwq = worker->gcwq;
539+
540+ BUG_ON(worker->flags & WORKER_IDLE);
541+ BUG_ON(!list_empty(&worker->entry) &&
542+ (worker->hentry.next || worker->hentry.pprev));
543+
544+ worker->flags |= WORKER_IDLE;
545+ gcwq->nr_idle++;
546+
547+ /* idle_list is LIFO */
548+ list_add(&worker->entry, &gcwq->idle_list);
549+}
550+
551+/**
552+ * worker_leave_idle - leave idle state
553+ * @worker: worker which is leaving idle state
554+ *
555+ * @worker is leaving idle state. Update stats.
556+ *
557+ * LOCKING:
558+ * spin_lock_irq(gcwq->lock).
559+ */
560+static void worker_leave_idle(struct worker *worker)
561+{
562+ struct global_cwq *gcwq = worker->gcwq;
563+
564+ BUG_ON(!(worker->flags & WORKER_IDLE));
565+ worker->flags &= ~WORKER_IDLE;
566+ gcwq->nr_idle--;
567+ list_del_init(&worker->entry);
568+}
569+
473570 static struct worker *alloc_worker(void)
474571 {
475572 struct worker *worker;
476573
477574 worker = kzalloc(sizeof(*worker), GFP_KERNEL);
478− if (worker)
575+ if (worker) {
576+ INIT_LIST_HEAD(&worker->entry);
479577 INIT_LIST_HEAD(&worker->scheduled);
578+ }
480579 return worker;
481580 }
482581
@@ -541,13 +640,16 @@ fail: · 58 unmodified lines
541640 * start_worker - start a newly created worker
542641 * @worker: worker to start
543642 *
544− * Start @worker.
643+ * Make the gcwq aware of @worker and start it.
545644 *
546645 * CONTEXT:
547646 * spin_lock_irq(gcwq->lock).
548647 */
549648 static void start_worker(struct worker *worker)
550649 {
650+ worker->flags |= WORKER_STARTED;
651+ worker->gcwq->nr_workers++;
652+ worker_enter_idle(worker);
551653 wake_up_process(worker->task);
552654 }
553655
@@ -555,7 +657,10 @@ static void start_worker(struct worker *worker) · 1 unmodified line
555657 * destroy_worker - destroy a workqueue worker
556658 * @worker: worker to be destroyed
557659 *
558− * Destroy @worker.
660+ * Destroy @worker and adjust @gcwq stats accordingly.
661+ *
662+ * CONTEXT:
663+ * spin_lock_irq(gcwq->lock) which is released and regrabbed.
559664 */
560665 static void destroy_worker(struct worker *worker)
561666 {
@@ -566,12 +671,21 @@ static void destroy_worker(struct worker *worker) · 4 unmodified lines
566671 BUG_ON(worker->current_work);
567672 BUG_ON(!list_empty(&worker->scheduled));
568673
674+ if (worker->flags & WORKER_STARTED)
675+ gcwq->nr_workers--;
676+ if (worker->flags & WORKER_IDLE)
677+ gcwq->nr_idle--;
678+
679+ list_del_init(&worker->entry);
680+ worker->flags |= WORKER_DIE;
681+
682+ spin_unlock_irq(&gcwq->lock);
683+
569684 kthread_stop(worker->task);
570685 kfree(worker);
571686
572687 spin_lock_irq(&gcwq->lock);
573688 ida_remove(&gcwq->worker_ida, id);
574− spin_unlock_irq(&gcwq->lock);
575689 }
576690
577691 /**
@@ -686,6 +800,7 @@ static void process_one_work(struct worker *worker, struct work_struct *work) · 108 unmodified lines
686800 {
687801 struct cpu_workqueue_struct *cwq = worker->cwq;
688802 struct global_cwq *gcwq = cwq->gcwq;
803+ struct hlist_head *bwh = busy_worker_head(gcwq, work);
689804 work_func_t f = work->func;
690805 int work_color;
691806 #ifdef CONFIG_LOCKDEP
@@ -700,6 +815,7 @@ static void process_one_work(struct worker *worker, struct work_struct *work) · 8 unmodified lines
700815 #endif
701816 /* claim and process */
702817 debug_work_deactivate(work);
818+ hlist_add_head(&worker->hentry, bwh);
703819 worker->current_work = work;
704820 work_color = get_work_color(work);
705821 list_del_init(&work->entry);
@@ -727,6 +843,7 @@ static void process_one_work(struct worker *worker, struct work_struct *work) · 21 unmodified lines
727843 spin_lock_irq(&gcwq->lock);
728844
729845 /* we're done with it, release */
846+ hlist_del_init(&worker->hentry);
730847 worker->current_work = NULL;
731848 cwq_dec_nr_in_flight(cwq, work_color);
732849 }
@@ -763,47 +880,56 @@ static int worker_thread(void *__worker) · 30 unmodified lines
763880 struct worker *worker = __worker;
764881 struct global_cwq *gcwq = worker->gcwq;
765882 struct cpu_workqueue_struct *cwq = worker->cwq;
766− DEFINE_WAIT(wait);
767883
768− for (;;) {
769− prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
770− if (!kthread_should_stop() &&
771− list_empty(&cwq->worklist))
772− schedule();
773− finish_wait(&cwq->more_work, &wait);
884+woke_up:
885+ if (unlikely(!cpumask_equal(&worker->task->cpus_allowed,
886+ get_cpu_mask(gcwq->cpu))))
887+ set_cpus_allowed_ptr(worker->task, get_cpu_mask(gcwq->cpu));
774888
775− if (kthread_should_stop())
776− break;
889+ spin_lock_irq(&gcwq->lock);
777890
778− if (unlikely(!cpumask_equal(&worker->task->cpus_allowed,
779− get_cpu_mask(gcwq->cpu))))
780− set_cpus_allowed_ptr(worker->task,
781− get_cpu_mask(gcwq->cpu));
891+ /* DIE can be set only while we're idle, checking here is enough */
892+ if (worker->flags & WORKER_DIE) {
893+ spin_unlock_irq(&gcwq->lock);
894+ return 0;
895+ }
782896
783− spin_lock_irq(&gcwq->lock);
897+ worker_leave_idle(worker);
784898
785− while (!list_empty(&cwq->worklist)) {
786− struct work_struct *work =
787− list_first_entry(&cwq->worklist,
788− struct work_struct, entry);
789−
790− if (likely(!(*work_data_bits(work) &
791− WORK_STRUCT_LINKED))) {
792− /* optimization path, not strictly necessary */
793− process_one_work(worker, work);
794− if (unlikely(!list_empty(&worker->scheduled)))
795− process_scheduled_works(worker);
796− } else {
797− move_linked_works(work, &worker->scheduled,
798− NULL);
899+ /*
900+ * ->scheduled list can only be filled while a worker is
901+ * preparing to process a work or actually processing it.
902+ * Make sure nobody diddled with it while I was sleeping.
903+ */
904+ BUG_ON(!list_empty(&worker->scheduled));
905+
906+ while (!list_empty(&cwq->worklist)) {
907+ struct work_struct *work =
908+ list_first_entry(&cwq->worklist,
909+ struct work_struct, entry);
910+
911+ if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
912+ /* optimization path, not strictly necessary */
913+ process_one_work(worker, work);
914+ if (unlikely(!list_empty(&worker->scheduled)))
799915 process_scheduled_works(worker);
800− }
916+ } else {
917+ move_linked_works(work, &worker->scheduled, NULL);
918+ process_scheduled_works(worker);
801919 }
802−
803− spin_unlock_irq(&gcwq->lock);
804920 }
805921
806− return 0;
922+ /*
923+ * gcwq->lock is held and there's no work to process, sleep.
924+ * Workers are woken up only while holding gcwq->lock, so
925+ * setting the current state before releasing gcwq->lock is
926+ * enough to prevent losing any event.
927+ */
928+ worker_enter_idle(worker);
929+ __set_current_state(TASK_INTERRUPTIBLE);
930+ spin_unlock_irq(&gcwq->lock);
931+ schedule();
932+ goto woke_up;
807933 }
808934
809935 struct wq_barrier {
@@ -1600,7 +1726,6 @@ struct workqueue_struct *__create_workqueue_key(const char *name, · 790 unmodified lines
16001726 cwq->max_active = max_active;
16011727 INIT_LIST_HEAD(&cwq->worklist);
16021728 INIT_LIST_HEAD(&cwq->delayed_works);
1603− init_waitqueue_head(&cwq->more_work);
16041729
16051730 if (failed)
16061731 continue;
@@ -1651,7 +1776,7 @@ EXPORT_SYMBOL_GPL(__create_workqueue_key); · 44 unmodified lines
16511776 */
16521777 void destroy_workqueue(struct workqueue_struct *wq)
16531778 {
1654− int cpu;
1779+ unsigned int cpu;
16551780
16561781 flush_workqueue(wq);
16571782
@@ -1670,8 +1795,10 @@ void destroy_workqueue(struct workqueue_struct *wq) · 12 unmodified lines
16701795 int i;
16711796
16721797 if (cwq->worker) {
1798+ spin_lock_irq(&cwq->gcwq->lock);
16731799 destroy_worker(cwq->worker);
16741800 cwq->worker = NULL;
1801+ spin_unlock_irq(&cwq->gcwq->lock);
16751802 }
16761803
16771804 for (i = 0; i < WORK_NR_COLORS; i++)
@@ -1881,7 +2008,7 @@ void thaw_workqueues(void) · 203 unmodified lines
18812008 cwq->nr_active < cwq->max_active)
18822009 cwq_activate_first_delayed(cwq);
18832010
1884− wake_up(&cwq->more_work);
2011+ wake_up_process(cwq->worker->task);
18852012 }
18862013
18872014 spin_unlock_irq(&gcwq->lock);
@@ -1896,6 +2023,7 @@ out_unlock: · 8 unmodified lines
18962023 void __init init_workqueues(void)
18972024 {
18982025 unsigned int cpu;
2026+ int i;
18992027
19002028 singlethread_cpu = cpumask_first(cpu_possible_mask);
19012029 hotcpu_notifier(workqueue_cpu_callback, 0);
@@ -1907,6 +2035,10 @@ void __init init_workqueues(void) · 5 unmodified lines
19072035 spin_lock_init(&gcwq->lock);
19082036 gcwq->cpu = cpu;
19092037
2038+ INIT_LIST_HEAD(&gcwq->idle_list);
2039+ for (i = 0; i < BUSY_WORKER_HASH_SIZE; i++)
2040+ INIT_HLIST_HEAD(&gcwq->busy_hash[i]);
2041+
19102042 ida_init(&gcwq->worker_ida);
19112043 }
19122044