Skip to contentSkip to conversation

rt: workqueue PI support -v2

PATCH In review Peter Zijlstra 2 participants 12 messages Oct 23, 2007

Subsystems (7)
lore
Share & tools
lore
#0 lore raw

cover rt: workqueue PI support -v2

Still not more than boot tested,...

Oleg, do you have workqueue test modules?

Changes since -v1:
- proper plist_head_splice() implementation
- removed the plist_add(, .tail) thing, using prio -1 instead.

(patch against v2.6.23-rt1)

--

#1 lore raw

PATCH 1/5 rt: rename rt_mutex_setprio to task_setprio

With there being multiple non-mutex users of this function its past time it got renamed.

---
 include/linux/sched.h     |    7 ++++++-
 kernel/rcupreempt-boost.c |    4 ++--
 kernel/sched.c            |    8 ++------
 3 files changed, 10 insertions(+), 9 deletions(-)
include/linux/sched.h +6 -1
Index: linux-2.6/include/linux/sched.h
===================================================================
@@ -1655,9 +1655,14 @@ extern unsigned int sysctl_sched_compat_
 extern unsigned int sysctl_sched_child_runs_first;
 extern unsigned int sysctl_sched_features;
 
+extern void task_setprio(struct task_struct *p, int prio);
+
 #ifdef CONFIG_RT_MUTEXES
 extern int rt_mutex_getprio(struct task_struct *p);
-extern void rt_mutex_setprio(struct task_struct *p, int prio);
+static inline void rt_mutex_setprio(struct task_struct *p, int prio)
+{
+	task_setprio(p, prio);
+}
 extern void rt_mutex_adjust_pi(struct task_struct *p);
 #else
 static inline int rt_mutex_getprio(struct task_struct *p)
kernel/sched.c +2 -6
Index: linux-2.6/kernel/sched.c
===================================================================
@@ -4445,10 +4445,8 @@ long __sched sleep_on_timeout(wait_queue
 }
 EXPORT_SYMBOL(sleep_on_timeout);
 
-#ifdef CONFIG_RT_MUTEXES
-
 /*
- * rt_mutex_setprio - set the current priority of a task
+ * task_setprio - set the current priority of a task
  * @p: task
  * @prio: prio value (kernel-internal form)
  *
@@ -4457,7 +4455,7 @@ EXPORT_SYMBOL(sleep_on_timeout);
  *
  * Used by the rt_mutex code to implement priority inheritance logic.
  */
-void rt_mutex_setprio(struct task_struct *p, int prio)
+void task_setprio(struct task_struct *p, int prio)
 {
 	unsigned long flags;
 	int oldprio, prev_resched, on_rq;
@@ -4522,8 +4520,6 @@ out_unlock:
 	task_rq_unlock(rq, &flags);
 }
 
-#endif
-
 void set_user_nice(struct task_struct *p, long nice)
 {
 	int old_prio, delta, on_rq;
kernel/rcupreempt-boost.c +2 -2
Index: linux-2.6/kernel/rcupreempt-boost.c
===================================================================
@@ -233,7 +233,7 @@ static void rcu_boost_task(struct task_s
 
 	if (task->rcu_prio < task->prio) {
 		rcu_trace_boost_task_boosted(RCU_BOOST_ME);
-		rt_mutex_setprio(task, task->rcu_prio);
+		task_setprio(task, task->rcu_prio);
 	}
 }
 
@@ -325,7 +325,7 @@ void __rcu_preempt_unboost(void)
 
 	spin_lock(&curr->pi_lock);
 	prio = rt_mutex_getprio(curr);
-	rt_mutex_setprio(curr, prio);
+	task_setprio(curr, prio);
 
 	curr->rcub_rbdp = NULL;
 

--

#2 lore raw

PATCH 2/5 rt: list_splice2

In reply to Peter Zijlstra

Introduce list_splice2{,_tail}() which will splice a sub-list denoted by two list items instead of the full list.

---
 drivers/usb/host/ehci-q.c |    2 -
 include/linux/list.h      |   66 ++++++++++++++++++++++++++++++++++++++++------
 lib/lock_list.c           |    2 -
 3 files changed, 60 insertions(+), 10 deletions(-)
include/linux/list.h +58 -8
Index: linux-2.6/include/linux/list.h
===================================================================
@@ -320,17 +320,17 @@ static inline int list_empty_careful(con
 }
 
 static inline void __list_splice(struct list_head *list,
-				 struct list_head *head)
+				 struct list_head *prev,
+				 struct list_head *next)
 {
 	struct list_head *first = list->next;
 	struct list_head *last = list->prev;
-	struct list_head *at = head->next;
 
-	first->prev = head;
-	head->next = first;
+	first->prev = prev;
+	prev->next = first;
 
-	last->next = at;
-	at->prev = last;
+	last->next = next;
+	next->prev = last;
 }
 
 /**
@@ -341,7 +341,13 @@ static inline void __list_splice(struct 
 static inline void list_splice(struct list_head *list, struct list_head *head)
 {
 	if (!list_empty(list))
-		__list_splice(list, head);
+		__list_splice(list, head, head->next);
+}
+
+static inline void list_splice_tail(struct list_head *list, struct list_head *head)
+{
+	if (!list_empty(list))
+		__list_splice(list, head->prev, head);
 }
 
 /**
@@ -355,11 +361,55 @@ static inline void list_splice_init(stru
 				    struct list_head *head)
 {
 	if (!list_empty(list)) {
-		__list_splice(list, head);
+		__list_splice(list, head, head->next);
+		INIT_LIST_HEAD(list);
+	}
+}
+
+static inline void list_splice_tail_init(struct list_head *list,
+					 struct list_head *head)
+{
+	if (!list_empty(list)) {
+		__list_splice(list, head->prev, head);
 		INIT_LIST_HEAD(list);
 	}
 }
 
+static inline void __list_splice2(struct list_head *first,
+	       			struct list_head *last,
+				struct list_head *prev,
+				struct list_head *next)
+{
+	first->prev->next = last->next;
+	last->next->prev = first->prev;
+
+	first->prev = prev;
+	prev->next = first;
+
+	last->next = next;
+	next->prev = last;
+}
+
+/**
+ * list_splice2 - join [first, last] to head
+ * @first: list item
+ * @last: list item further on the same list
+ * @head: the place to add it on another list
+ */
+static inline void list_splice2(struct list_head *first,
+				struct list_head *last,
+				struct list_head *head)
+{
+	__list_splice2(first, last, head, head->next);
+}
+
+static inline void list_splice2_tail(struct list_head *first,
+				     struct list_head *last,
+				     struct list_head *head)
+{
+	__list_splice2(first, last, head->prev, head);
+}
+
 /**
  * list_splice_init_rcu - splice an RCU-protected list into an existing list.
  * @list:	the RCU-protected list to splice
drivers/usb/host/ehci-q.c +1 -1
Index: linux-2.6/drivers/usb/host/ehci-q.c
===================================================================
@@ -887,7 +887,7 @@ static struct ehci_qh *qh_append_tds (
 
 			list_del (&qtd->qtd_list);
 			list_add (&dummy->qtd_list, qtd_list);
-			__list_splice (qtd_list, qh->qtd_list.prev);
+			list_splice_tail (qtd_list, &qh->qtd_list);
 
 			ehci_qtd_init(ehci, qtd, qtd->qtd_dma);
 			qh->dummy = qtd;
lib/lock_list.c +1 -1
Index: linux-2.6/lib/lock_list.c
===================================================================
@@ -128,7 +128,7 @@ void lock_list_splice_init(struct lock_l
 	lock = __lock_list_reverse(list);
 	if (!list_empty(&list->head)) {
 		spin_lock_nested(&head->lock, LOCK_LIST_NESTING_NEXT);
-		__list_splice(&list->head, &head->head);
+		__list_splice(&list->head, &head->head, head->head.next);
 		INIT_LIST_HEAD(&list->head);
 		spin_unlock(&head->lock);
 	}

--

#3 lore raw

Re: [RFC/PATCH 2/5] rt: list_splice2

--

Quoted 17 lines · Peter Zijlstra

/**

include/linux/list.h +3 -1
@@ -341,7 +341,13 @@ static inline void __list_splice(struct
 static inline void list_splice(struct list_head *list, struct list_head *head)
 {
 	if (!list_empty(list))
-		__list_splice(list, head);
+		__list_splice(list, head, head->next);
+}
+
 
 

Index: linux-2.6/lib/lock_list.c
===================================================================

lib/lock_list.c void lock_list_splice_init(struct lock_l · line 131

Quoted 5 lines
lib/lock_list.c +1 -1
@@ -128,7 +128,7 @@ void lock_list_splice_init(struct lock_l
 	lock = __lock_list_reverse(list);
 	if (!list_empty(&list->head)) {
 		spin_lock_nested(&head->lock, LOCK_LIST_NESTING_NEXT);
-		__list_splice(&list->head, &head->head);
+		__list_splice(&list->head, &head->head, head->head.next);

Can't this just now be list_splice and not __list_splice?

Yes it tests for list_empty again, but that should (hopefully) be optimized out.

-- Steve

Quoted 8 lines
lib/lock_list.c +0 -1
 		INIT_LIST_HEAD(&list->head);
 		spin_unlock(&head->lock);
 	}
 
--
#4 lore raw

PATCH 3/5 rt: plist_head_splice

In reply to Peter Zijlstra

merge-sort two plists together

---
 include/linux/plist.h |    2 +
 lib/plist.c           |   68 ++++++++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 68 insertions(+), 2 deletions(-)
include/linux/plist.h +2 -0
Index: linux-2.6/include/linux/plist.h
===================================================================
@@ -148,6 +148,8 @@ static inline void plist_node_init(struc
 extern void plist_add(struct plist_node *node, struct plist_head *head);
 extern void plist_del(struct plist_node *node, struct plist_head *head);
 
+extern void plist_head_splice(struct plist_head *src, struct plist_head *dst);
+
 /**
  * plist_for_each - iterate over the plist
  * @pos:	the type * to use as a loop counter
lib/plist.c +66 -2

83 lines not on this page · raw message

--

#5 lore raw

Re: [RFC/PATCH 3/5] rt: plist_head_splice

--

Quoted 16 lines · Peter Zijlstra
lib/plist.c +16 -0
+
+void plist_head_splice(struct plist_head *src, struct plist_head *dst)
+{
+	struct plist_node *src_iter_first, *src_iter_last, *dst_iter;
+	struct plist_node *tail = container_of(dst, struct plist_node, plist);
+
+	dst_iter = next_prio(tail);
+
+	while (!plist_head_empty(src) && dst_iter != tail) {
+		src_iter_first = plist_first(src);
+
+		src_iter_last = next_prio(src_iter_first);
+		src_iter_last = prev_node(src_iter_last);
+
+		WARN_ON(src_iter_first->prio != src_iter_last->prio);
+		WARN_ON(list_empty(&src_iter_first->plist.prio_list));

lib/plist.c void plist_del(struct plist_node *node, · line 169

Quoted 10 lines
lib/plist.c +10 -0
+
+		while (src_iter_first->prio > dst_iter->prio) {
+			dst_iter = next_prio(dst_iter);
+			if (dst_iter == tail)
+				goto tail;
+		}
+
+		list_del_init(&src_iter_first->plist.prio_list);
+
+		if (src_iter_first->prio < dst_iter->prio) {

I may be confused here, but shouldn't we be linking the src_iter_first->prio_list somewhere here? Doesn't all different prios need to be in its separate prio_list? Otherwise two splices in a row can cause the above WARN_ON (prio != prio).

-- Steve

Quoted 20 lines
lib/plist.c +15 -1
+			list_add_tail(&src_iter_first->plist.node_list,
+					&dst_iter->plist.node_list);
+		} else if (src_iter_first->prio == dst_iter->prio) {
+			dst_iter = next_prio(dst_iter);
+		} else BUG();
+
+		list_splice2_tail(&src_iter_first->plist.node_list,
+			       	  &src_iter_last->plist.node_list,
+				  &dst_iter->plist.node_list);
+	}
+
+tail:
+	list_splice_tail_init(&src->prio_list, &dst->prio_list);
+	list_splice_tail_init(&src->node_list, &dst->node_list);
+}
 
--
#6 lore raw

Re: [RFC/PATCH 3/5] rt: plist_head_splice

Quoted 35 lines · Steven Rostedt

--

Quoted 16 lines · Peter Zijlstra
lib/plist.c +16 -0
+
+void plist_head_splice(struct plist_head *src, struct plist_head *dst)
+{
+	struct plist_node *src_iter_first, *src_iter_last, *dst_iter;
+	struct plist_node *tail = container_of(dst, struct plist_node, plist);
+
+	dst_iter = next_prio(tail);
+
+	while (!plist_head_empty(src) && dst_iter != tail) {
+		src_iter_first = plist_first(src);
+
+		src_iter_last = next_prio(src_iter_first);
+		src_iter_last = prev_node(src_iter_last);
+
+		WARN_ON(src_iter_first->prio != src_iter_last->prio);
+		WARN_ON(list_empty(&src_iter_first->plist.prio_list));

lib/plist.c void plist_del(struct plist_node *node, · line 169

Quoted 10 lines
lib/plist.c +10 -0
+
+		while (src_iter_first->prio > dst_iter->prio) {
+			dst_iter = next_prio(dst_iter);
+			if (dst_iter == tail)
+				goto tail;
+		}
+
+		list_del_init(&src_iter_first->plist.prio_list);
+
+		if (src_iter_first->prio < dst_iter->prio) {

I may be confused here, but shouldn't we be linking the src_iter_first->prio_list somewhere here? Doesn't all different prios need to be in its separate prio_list? Otherwise two splices in a row can cause the above WARN_ON (prio != prio).

Quoted 2 lines

Review

Quoted 2 lines
(quoted diff) +2 -0
+			list_add_tail(&src_iter_first->plist.node_list,
+					&dst_iter->plist.node_list);

Uhm, yeah, that was supposed the be prio_list indeed.

Thanks!

Quoted 18 lines
Quoted 18 lines
lib/plist.c +13 -1
+		} else if (src_iter_first->prio == dst_iter->prio) {
+			dst_iter = next_prio(dst_iter);
+		} else BUG();
+
+		list_splice2_tail(&src_iter_first->plist.node_list,
+			       	  &src_iter_last->plist.node_list,
+				  &dst_iter->plist.node_list);
+	}
+
+tail:
+	list_splice_tail_init(&src->prio_list, &dst->prio_list);
+	list_splice_tail_init(&src->node_list, &dst->node_list);
+}
 
--
#7 lore raw

Re: [RFC/PATCH 3/5] rt: plist_head_splice

In reply to Steven Rostedt
lib/plist.c +2 -2
Index: linux-2.6/lib/plist.c
===================================================================
@@ -167,8 +167,8 @@ void plist_head_splice(struct plist_head
 		list_del_init(&src_iter_first->plist.prio_list);
 
 		if (src_iter_first->prio < dst_iter->prio) {
-			list_add_tail(&src_iter_first->plist.node_list,
-					&dst_iter->plist.node_list);
+			list_add_tail(&src_iter_first->plist.prio_list,
+					&dst_iter->plist.prio_list);
 		} else if (src_iter_first->prio == dst_iter->prio) {
 			dst_iter = next_prio(dst_iter);
 		} else BUG();
#8 lore raw

PATCH 4/5 rt: PI-workqueue support

In reply to Peter Zijlstra

Add support for priority queueing and priority inheritance to the workqueue infrastructure. This is done by replacing the linear linked worklist with a priority sorted plist.

The drawback is that this breaks the workqueue barrier, needed to support flush_workqueue() and wait_on_work().

---
 include/linux/workqueue.h |    7 ++++---
 kernel/power/poweroff.c   |    1 +
 kernel/workqueue.c        |   40 +++++++++++++++++++++++++---------------
 3 files changed, 30 insertions(+), 18 deletions(-)
include/linux/workqueue.h +4 -3
Index: linux-2.6/include/linux/workqueue.h
===================================================================
@@ -8,6 +8,7 @@
 #include <linux/timer.h>
 #include <linux/linkage.h>
 #include <linux/bitops.h>
+#include <linux/plist.h>
 #include <asm/atomic.h>
 
 struct workqueue_struct;
@@ -26,7 +27,7 @@ struct work_struct {
 #define WORK_STRUCT_PENDING 0		/* T if work item pending execution */
 #define WORK_STRUCT_FLAG_MASK (3UL)
 #define WORK_STRUCT_WQ_DATA_MASK (~WORK_STRUCT_FLAG_MASK)
-	struct list_head entry;
+	struct plist_node entry;
 	work_func_t func;
 };
 
@@ -43,7 +44,7 @@ struct execute_work {
 
 #define __WORK_INITIALIZER(n, f) {				\
 	.data = WORK_DATA_INIT(),				\
-	.entry	= { &(n).entry, &(n).entry },			\
+	.entry	= PLIST_NODE_INIT(n.entry, MAX_PRIO),		\
 	.func = (f),						\
 	}
 
@@ -79,7 +80,7 @@ struct execute_work {
 #define INIT_WORK(_work, _func)						\
 	do {								\
 		(_work)->data = (atomic_long_t) WORK_DATA_INIT();	\
-		INIT_LIST_HEAD(&(_work)->entry);			\
+		plist_node_init(&(_work)->entry, -1);			\
 		PREPARE_WORK((_work), (_func));				\
 	} while (0)
 
kernel/power/poweroff.c +1 -0
Index: linux-2.6/kernel/power/poweroff.c
===================================================================
@@ -8,6 +8,7 @@
 #include <linux/sysrq.h>
 #include <linux/init.h>
 #include <linux/pm.h>
+#include <linux/sched.h>
 #include <linux/workqueue.h>
 #include <linux/reboot.h>
 
kernel/workqueue.c +25 -15
Index: linux-2.6/kernel/workqueue.c
===================================================================
@@ -44,7 +44,7 @@ struct cpu_workqueue_struct {
 
 	spinlock_t lock;
 
-	struct list_head worklist;
+	struct plist_head worklist;
 	wait_queue_head_t more_work;
 	struct work_struct *current_work;
 
@@ -127,16 +127,19 @@ struct cpu_workqueue_struct *get_wq_data
 static void insert_work(struct cpu_workqueue_struct *cwq,
 				struct work_struct *work, int tail)
 {
+	int prio = current->normal_prio;
+
 	set_wq_data(work, cwq);
 	/*
 	 * Ensure that we get the right work->data if we see the
 	 * result of list_add() below, see try_to_grab_pending().
 	 */
 	smp_wmb();
-	if (tail)
-		list_add_tail(&work->entry, &cwq->worklist);
-	else
-		list_add(&work->entry, &cwq->worklist);
+	plist_node_init(&work->entry, prio);
+	plist_add(&work->entry, &cwq->worklist);
+
+	if (prio < cwq->thread->prio)
+		task_setprio(cwq->thread, prio);
 	wake_up(&cwq->more_work);
 }
 
@@ -168,7 +171,7 @@ int fastcall queue_work(struct workqueue
 	int ret = 0, cpu = raw_smp_processor_id();
 
 	if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
-		BUG_ON(!list_empty(&work->entry));
+		BUG_ON(!plist_node_empty(&work->entry));
 		__queue_work(wq_per_cpu(wq, cpu), work);
 		ret = 1;
 	}
@@ -222,7 +225,7 @@ int queue_delayed_work_on(int cpu, struc
 
 	if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 		BUG_ON(timer_pending(timer));
-		BUG_ON(!list_empty(&work->entry));
+		BUG_ON(!plist_node_empty(&work->entry));
 
 		/* This stores cwq for the moment, for the timer_fn */
 		set_wq_data(work, wq_per_cpu(wq, raw_smp_processor_id()));
@@ -264,13 +267,17 @@ static void run_workqueue(struct cpu_wor
 			__FUNCTION__, cwq->run_depth);
 		dump_stack();
 	}
-	while (!list_empty(&cwq->worklist)) {
-		struct work_struct *work = list_entry(cwq->worklist.next,
+	while (!plist_head_empty(&cwq->worklist)) {
+		struct work_struct *work = plist_first_entry(&cwq->worklist,
 						struct work_struct, entry);
 		work_func_t f = work->func;
 
+		if (likely(cwq->thread->prio != work->entry.prio))
+			task_setprio(cwq->thread, work->entry.prio);
+
 		cwq->current_work = work;
-		list_del_init(cwq->worklist.next);
+		plist_del(&work->entry, &cwq->worklist);
+		plist_node_init(&work->entry, MAX_PRIO);
 		spin_unlock_irq(&cwq->lock);
 
 		BUG_ON(get_wq_data(work) != cwq);
@@ -283,6 +290,7 @@ static void run_workqueue(struct cpu_wor
 		spin_lock_irq(&cwq->lock);
 		cwq->current_work = NULL;
 	}
+	task_setprio(cwq->thread, current->normal_prio);
 	cwq->run_depth--;
 	spin_unlock_irq(&cwq->lock);
 }
@@ -301,7 +309,7 @@ static int worker_thread(void *__cwq)
 		prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
 		if (!freezing(current) &&
 		    !kthread_should_stop() &&
-		    list_empty(&cwq->worklist))
+		    plist_head_empty(&cwq->worklist))
 			schedule();
 		finish_wait(&cwq->more_work, &wait);
 
@@ -354,7 +362,8 @@ static int flush_cpu_workqueue(struct cp
 
 		active = 0;
 		spin_lock_irq(&cwq->lock);
-		if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) {
+		if (!plist_head_empty(&cwq->worklist) ||
+			cwq->current_work != NULL) {
 			insert_wq_barrier(cwq, &barr, 1);
 			active = 1;
 		}
@@ -413,7 +422,7 @@ static int try_to_grab_pending(struct wo
 		return ret;
 
 	spin_lock_irq(&cwq->lock);
-	if (!list_empty(&work->entry)) {
+	if (!plist_node_empty(&work->entry)) {
 		/*
 		 * This work is queued, but perhaps we locked the wrong cwq.
 		 * In that case we must see the new value after rmb(), see
@@ -421,7 +430,8 @@ static int try_to_grab_pending(struct wo
 		 */
 		smp_rmb();
 		if (cwq == get_wq_data(work)) {
-			list_del_init(&work->entry);
+			plist_del(&work->entry, &cwq->worklist);
+			plist_node_init(&work->entry, MAX_PRIO);
 			ret = 1;
 		}
 	}
@@ -747,7 +757,7 @@ init_cpu_workqueue(struct workqueue_stru
 
 	cwq->wq = wq;
 	spin_lock_init(&cwq->lock);
-	INIT_LIST_HEAD(&cwq->worklist);
+	plist_head_init(&cwq->worklist, NULL);
 	init_waitqueue_head(&cwq->more_work);
 
 	return cwq;

--

#9 lore raw

PATCH 5/5 rt: PI-workqueue: fix barriers

In reply to Peter Zijlstra

The plist change to the workqueues left the barrier functionality broken.

The barrier is used for two things:
- wait_on_work(), and
- flush_cpu_workqueue().

wait_on_work() - uses the barrier to wait on the completion of the currently worklet. This was done by inserting a completion barrier at the very head of the worklist. With plist this would be the head of the highest prio.

In order to do that, we extend the priority range to exceed the normal range and enqueue it higher than anything else. Another noteworthy point is that this high prio worklet must not boost the prio further than the waiting task's prio, even though we enqueue it at prio 100.

flush_cpu_workqueue() - is a full ordering barrier, although as the name suggests usually used to wait for the worklist to drain. We'll support the full ordering semantics currently present. This means that:

W10, W22, W65, B, W80, B, W99

[ where Wn is a worklet at prio n, and B a barrier ]

would most likely execute in the following order:

W10@99, W65@99, W22@99, W80@99, W99

[ Wn@m is Wn executed at prio m ]
[ W10 would be first because it can start executing while the others
are being added ]

Whereas without the barriers it would be:

W10@99, W99, W80, W65, W22

The prio ordering of the plist makes it hard to impose an extra order on top. The solution used is to nest plist structures. The example will look like:

W10, B(B(W65, W22), W80), W99

That is, the barrier will splice the worklist into itself, and enqueue itself as the next item to run (very first item, highest prio). The barrier will then run its own plist to completion before 'popping' back to the regular worklist.

To avoid callstack nesting, run_workqueue is taught about this barrier stack.

---
 kernel/workqueue.c |  111 +++++++++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 95 insertions(+), 16 deletions(-)
kernel/workqueue.c +95 -16
Index: linux-2.6/kernel/workqueue.c
===================================================================
@@ -36,6 +36,8 @@
 
 #include <asm/uaccess.h>
 
+struct wq_full_barrier;
+
 /*
  * The per-CPU workqueue (if single thread, we always use the first
  * possible cpu).
@@ -52,6 +54,8 @@ struct cpu_workqueue_struct {
 	struct task_struct *thread;
 
 	int run_depth;		/* Detect run_workqueue() recursion depth */
+
+	struct wq_full_barrier *barrier;
 } ____cacheline_aligned;
 
 /*
@@ -125,10 +129,8 @@ struct cpu_workqueue_struct *get_wq_data
 }
 
 static void insert_work(struct cpu_workqueue_struct *cwq,
-				struct work_struct *work, int tail)
+		struct work_struct *work, int prio, int boost_prio)
 {
-	int prio = current->normal_prio;
-
 	set_wq_data(work, cwq);
 	/*
 	 * Ensure that we get the right work->data if we see the
@@ -138,8 +140,8 @@ static void insert_work(struct cpu_workq
 	plist_node_init(&work->entry, prio);
 	plist_add(&work->entry, &cwq->worklist);
 
-	if (prio < cwq->thread->prio)
-		task_setprio(cwq->thread, prio);
+	if (boost_prio < cwq->thread->prio)
+		task_setprio(cwq->thread, boost_prio);
 	wake_up(&cwq->more_work);
 }
 
@@ -150,7 +152,7 @@ static void __queue_work(struct cpu_work
 	unsigned long flags;
 
 	spin_lock_irqsave(&cwq->lock, flags);
-	insert_work(cwq, work, 1);
+	insert_work(cwq, work, current->normal_prio, current->normal_prio);
 	spin_unlock_irqrestore(&cwq->lock, flags);
 }
 
@@ -257,8 +259,20 @@ static void leak_check(void *func)
 	dump_stack();
 }
 
+struct wq_full_barrier {
+	struct work_struct		work;
+	struct plist_head		worklist;
+	struct wq_full_barrier 		*prev_barrier;
+	int				prev_prio;
+	int				waiter_prio;
+	struct cpu_workqueue_struct 	*cwq;
+	struct completion		done;
+};
+
 static void run_workqueue(struct cpu_workqueue_struct *cwq)
 {
+	struct plist_head *worklist = &cwq->worklist;
+
 	spin_lock_irq(&cwq->lock);
 	cwq->run_depth++;
 	if (cwq->run_depth > 3) {
@@ -267,16 +281,27 @@ static void run_workqueue(struct cpu_wor
 			__FUNCTION__, cwq->run_depth);
 		dump_stack();
 	}
-	while (!plist_head_empty(&cwq->worklist)) {
-		struct work_struct *work = plist_first_entry(&cwq->worklist,
+
+again:
+	while (!plist_head_empty(worklist)) {
+		int prio;
+		struct work_struct *work = plist_first_entry(worklist,
 						struct work_struct, entry);
 		work_func_t f = work->func;
 
-		if (likely(cwq->thread->prio != work->entry.prio))
-			task_setprio(cwq->thread, work->entry.prio);
+		prio = work->entry.prio;
+		if (unlikely(worklist != &cwq->worklist)) {
+			prio = min(prio, cwq->barrier->prev_prio);
+			prio = min(prio, cwq->barrier->waiter_prio);
+			prio = min(prio, plist_first(&cwq->worklist)->prio);
+		}
+		prio = max(prio, 0);
+
+		if (likely(cwq->thread->prio != prio))
+			task_setprio(cwq->thread, prio);
 
 		cwq->current_work = work;
-		plist_del(&work->entry, &cwq->worklist);
+		plist_del(&work->entry, worklist);
 		plist_node_init(&work->entry, MAX_PRIO);
 		spin_unlock_irq(&cwq->lock);
 
@@ -289,7 +314,27 @@ static void run_workqueue(struct cpu_wor
 
 		spin_lock_irq(&cwq->lock);
 		cwq->current_work = NULL;
+
+		if (unlikely(cwq->barrier))
+			worklist = &cwq->barrier->worklist;
+	}
+
+	if (unlikely(worklist != &cwq->worklist)) {
+		struct wq_full_barrier *barrier = cwq->barrier;
+
+		BUG_ON(!barrier);
+		cwq->barrier = barrier->prev_barrier;
+		complete(&barrier->done);
+
+		if (unlikely(cwq->barrier))
+			worklist = &cwq->barrier->worklist;
+		else
+			worklist = &cwq->worklist;
+
+		if (!plist_head_empty(worklist))
+			goto again;
 	}
+
 	task_setprio(cwq->thread, current->normal_prio);
 	cwq->run_depth--;
 	spin_unlock_irq(&cwq->lock);
@@ -336,14 +381,47 @@ static void wq_barrier_func(struct work_
 }
 
 static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
-					struct wq_barrier *barr, int tail)
+					struct wq_barrier *barr, int prio)
 {
 	INIT_WORK(&barr->work, wq_barrier_func);
 	__set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
 
 	init_completion(&barr->done);
 
-	insert_work(cwq, &barr->work, tail);
+	insert_work(cwq, &barr->work, prio, current->prio);
+}
+
+static void wq_full_barrier_func(struct work_struct *work)
+{
+	struct wq_full_barrier *barrier =
+		container_of(work, struct wq_full_barrier, work);
+	struct cpu_workqueue_struct *cwq = barrier->cwq;
+	int prio = MAX_PRIO;
+
+	spin_lock_irq(&cwq->lock);
+	barrier->prev_barrier = cwq->barrier;
+	if (cwq->barrier) {
+		prio = min(prio, cwq->barrier->waiter_prio);
+		prio = min(prio, plist_first(&cwq->barrier->worklist)->prio);
+	}
+	barrier->prev_prio = prio;
+	cwq->barrier = barrier;
+	spin_unlock_irq(&cwq->lock);
+}
+
+static void insert_wq_full_barrier(struct cpu_workqueue_struct *cwq,
+		struct wq_full_barrier *barr)
+{
+	INIT_WORK(&barr->work, wq_full_barrier_func);
+	__set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
+
+	plist_head_init(&barr->worklist, NULL);
+	plist_head_splice(&cwq->worklist, &barr->worklist);
+	barr->cwq = cwq;
+	init_completion(&barr->done);
+	barr->waiter_prio = current->prio;
+
+	insert_work(cwq, &barr->work, 0, current->prio);
 }
 
 static int flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
@@ -358,13 +436,13 @@ static int flush_cpu_workqueue(struct cp
 		run_workqueue(cwq);
 		active = 1;
 	} else {
-		struct wq_barrier barr;
+		struct wq_full_barrier barr;
 
 		active = 0;
 		spin_lock_irq(&cwq->lock);
 		if (!plist_head_empty(&cwq->worklist) ||
 			cwq->current_work != NULL) {
-			insert_wq_barrier(cwq, &barr, 1);
+			insert_wq_full_barrier(cwq, &barr);
 			active = 1;
 		}
 		spin_unlock_irq(&cwq->lock);
@@ -448,7 +526,7 @@ static void wait_on_cpu_work(struct cpu_
 
 	spin_lock_irq(&cwq->lock);
 	if (unlikely(cwq->current_work == work)) {
-		insert_wq_barrier(cwq, &barr, 0);
+		insert_wq_barrier(cwq, &barr, -1);
 		running = 1;
 	}
 	spin_unlock_irq(&cwq->lock);
@@ -759,6 +837,7 @@ init_cpu_workqueue(struct workqueue_stru
 	spin_lock_init(&cwq->lock);
 	plist_head_init(&cwq->worklist, NULL);
 	init_waitqueue_head(&cwq->more_work);
+	cwq->barrier = NULL;
 
 	return cwq;
 }

--

#10 lore raw

[RFC/PATCH 6/5] rt: PI-workqueue: wait_on_work() fixup

In reply to Peter Zijlstra

Subject: rt: PI-workqueue: wait_on_work() fixup

Oleg noticed that the new wait_on_work() barrier does not properly interact with the nesting barrier.

The problem is that a wait_on_work() targeted at a worklet in a nested list will complete too late.

Fix this by using a wait_queue instead.

[ will be folded into the previous patch on next posting ]

---
 kernel/workqueue.c |   74 ++++++++++++++++++++---------------------------------
 1 file changed, 29 insertions(+), 45 deletions(-)
kernel/workqueue.c +29 -45
Index: linux-2.6/kernel/workqueue.c
===================================================================
@@ -33,10 +33,11 @@
 #include <linux/freezer.h>
 #include <linux/kallsyms.h>
 #include <linux/debug_locks.h>
+#include <linux/wait.h>
 
 #include <asm/uaccess.h>
 
-struct wq_full_barrier;
+struct wq_barrier;
 
 /*
  * The per-CPU workqueue (if single thread, we always use the first
@@ -55,7 +56,8 @@ struct cpu_workqueue_struct {
 
 	int run_depth;		/* Detect run_workqueue() recursion depth */
 
-	struct wq_full_barrier *barrier;
+	wait_queue_head_t work_done;
+	struct wq_barrier *barrier;
 } ____cacheline_aligned;
 
 /*
@@ -259,10 +261,10 @@ static void leak_check(void *func)
 	dump_stack();
 }
 
-struct wq_full_barrier {
+struct wq_barrier {
 	struct work_struct		work;
 	struct plist_head		worklist;
-	struct wq_full_barrier 		*prev_barrier;
+	struct wq_barrier 		*prev_barrier;
 	int				prev_prio;
 	int				waiter_prio;
 	struct cpu_workqueue_struct 	*cwq;
@@ -314,13 +316,13 @@ again:
 
 		spin_lock_irq(&cwq->lock);
 		cwq->current_work = NULL;
-
+		wake_up_all(&cwq->work_done);
 		if (unlikely(cwq->barrier))
 			worklist = &cwq->barrier->worklist;
 	}
 
 	if (unlikely(worklist != &cwq->worklist)) {
-		struct wq_full_barrier *barrier = cwq->barrier;
+		struct wq_barrier *barrier = cwq->barrier;
 
 		BUG_ON(!barrier);
 		cwq->barrier = barrier->prev_barrier;
@@ -369,32 +371,10 @@ static int worker_thread(void *__cwq)
 	return 0;
 }
 
-struct wq_barrier {
-	struct work_struct	work;
-	struct completion	done;
-};
-
 static void wq_barrier_func(struct work_struct *work)
 {
-	struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
-	complete(&barr->done);
-}
-
-static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
-					struct wq_barrier *barr, int prio)
-{
-	INIT_WORK(&barr->work, wq_barrier_func);
-	__set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
-
-	init_completion(&barr->done);
-
-	insert_work(cwq, &barr->work, prio, current->prio);
-}
-
-static void wq_full_barrier_func(struct work_struct *work)
-{
-	struct wq_full_barrier *barrier =
-		container_of(work, struct wq_full_barrier, work);
+	struct wq_barrier *barrier =
+		container_of(work, struct wq_barrier, work);
 	struct cpu_workqueue_struct *cwq = barrier->cwq;
 	int prio = MAX_PRIO;
 
@@ -409,10 +389,10 @@ static void wq_full_barrier_func(struct 
 	spin_unlock_irq(&cwq->lock);
 }
 
-static void insert_wq_full_barrier(struct cpu_workqueue_struct *cwq,
-		struct wq_full_barrier *barr)
+static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
+		struct wq_barrier *barr)
 {
-	INIT_WORK(&barr->work, wq_full_barrier_func);
+	INIT_WORK(&barr->work, wq_barrier_func);
 	__set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
 
 	plist_head_init(&barr->worklist, NULL);
@@ -436,13 +416,13 @@ static int flush_cpu_workqueue(struct cp
 		run_workqueue(cwq);
 		active = 1;
 	} else {
-		struct wq_full_barrier barr;
+		struct wq_barrier barr;
 
 		active = 0;
 		spin_lock_irq(&cwq->lock);
 		if (!plist_head_empty(&cwq->worklist) ||
 			cwq->current_work != NULL) {
-			insert_wq_full_barrier(cwq, &barr);
+			insert_wq_barrier(cwq, &barr);
 			active = 1;
 		}
 		spin_unlock_irq(&cwq->lock);
@@ -518,21 +498,24 @@ static int try_to_grab_pending(struct wo
 	return ret;
 }
 
-static void wait_on_cpu_work(struct cpu_workqueue_struct *cwq,
-				struct work_struct *work)
+static inline
+int is_current_work(struct cpu_workqueue_struct *cwq, struct work_struct *work)
 {
-	struct wq_barrier barr;
-	int running = 0;
+	int ret;
 
 	spin_lock_irq(&cwq->lock);
-	if (unlikely(cwq->current_work == work)) {
-		insert_wq_barrier(cwq, &barr, -1);
-		running = 1;
-	}
+	ret = (cwq->current_work == work);
 	spin_unlock_irq(&cwq->lock);
 
-	if (unlikely(running))
-		wait_for_completion(&barr.done);
+	return ret;
+}
+
+static void wait_on_cpu_work(struct cpu_workqueue_struct *cwq,
+				struct work_struct *work)
+{
+	DEFINE_WAIT(wait);
+
+	wait_event(cwq->work_done, is_current_work(cwq, work));
 }
 
 static void wait_on_work(struct work_struct *work)
@@ -838,6 +821,7 @@ init_cpu_workqueue(struct workqueue_stru
 	plist_head_init(&cwq->worklist, NULL);
 	init_waitqueue_head(&cwq->more_work);
 	cwq->barrier = NULL;
+	init_waitqueue_head(&cwq->work_done);
 
 	return cwq;
 }
#11 lore raw

[RFC/PATCH 7/5] rt: PI-workqueue: propagate prio for delayed work

In reply to Peter Zijlstra

Subject: rt: PI-workqueue: propagate prio for delayed work

Delayed work looses its enqueue priority, and will be enqueued on the prio of the softirq thread. Ammend this.

---
 include/linux/workqueue.h |    1 +
 kernel/workqueue.c        |   16 ++++++++++------
 2 files changed, 11 insertions(+), 6 deletions(-)
include/linux/workqueue.h +1 -0
Index: linux-2.6/include/linux/workqueue.h
===================================================================
@@ -36,6 +36,7 @@ struct work_struct {
 struct delayed_work {
 	struct work_struct work;
 	struct timer_list timer;
+	int prio;
 };
 
 struct execute_work {
kernel/workqueue.c +10 -6
Index: linux-2.6/kernel/workqueue.c
===================================================================
@@ -149,12 +149,12 @@ static void insert_work(struct cpu_workq
 
 /* Preempt must be disabled. */
 static void __queue_work(struct cpu_workqueue_struct *cwq,
-			 struct work_struct *work)
+			 struct work_struct *work, int prio)
 {
 	unsigned long flags;
 
 	spin_lock_irqsave(&cwq->lock, flags);
-	insert_work(cwq, work, current->normal_prio, current->normal_prio);
+	insert_work(cwq, work, prio, prio);
 	spin_unlock_irqrestore(&cwq->lock, flags);
 }
 
@@ -176,7 +176,7 @@ int fastcall queue_work(struct workqueue
 
 	if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 		BUG_ON(!plist_node_empty(&work->entry));
-		__queue_work(wq_per_cpu(wq, cpu), work);
+		__queue_work(wq_per_cpu(wq, cpu), work, current->normal_prio);
 		ret = 1;
 	}
 	return ret;
@@ -189,7 +189,8 @@ void delayed_work_timer_fn(unsigned long
 	struct cpu_workqueue_struct *cwq = get_wq_data(&dwork->work);
 	struct workqueue_struct *wq = cwq->wq;
 
-	__queue_work(wq_per_cpu(wq, raw_smp_processor_id()), &dwork->work);
+	__queue_work(wq_per_cpu(wq, raw_smp_processor_id()),
+			&dwork->work, dwork->prio);
 }
 
 /**
@@ -232,6 +233,7 @@ int queue_delayed_work_on(int cpu, struc
 		BUG_ON(!plist_node_empty(&work->entry));
 
 		/* This stores cwq for the moment, for the timer_fn */
+		dwork->prio = current->normal_prio;
 		set_wq_data(work, wq_per_cpu(wq, raw_smp_processor_id()));
 		timer->expires = jiffies + delay;
 		timer->data = (unsigned long)dwork;
@@ -702,7 +704,8 @@ int schedule_on_each_cpu(void (*func)(vo
 		work->info = info;
 		INIT_WORK(&work->work, schedule_on_each_cpu_func);
 		set_bit(WORK_STRUCT_PENDING, work_data_bits(&work->work));
-		__queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), &work->work);
+		__queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu),
+				&work->work, current->normal_prio);
 	}
 	unlock_cpu_hotplug();
 
@@ -749,7 +752,8 @@ int schedule_on_each_cpu_wq(struct workq
 
 		INIT_WORK(work, func);
 		set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
-		__queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
+		__queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work,
+				current->normal_prio);
 	}
 	flush_workqueue(wq);
 	free_percpu(works);