rcu_pending: cursor; combined __rcu_pending_enqueue()rcu_pending

 - go back to previous sequence number scheme
 - expose rcu_get_seq()

4 files changed, 128 insertions, 103 deletions

diff --git a/include/linux/rcu_pending.h b/include/linux/rcu_pending.h
index 5ef6392ce180..588c423ee8f9 100644
--- a/include/linux/rcu_pending.h
+++ b/include/linux/rcu_pending.h
@@ -10,6 +10,7 @@ struct rcu_pending_pcpu;
 struct rcu_pending {
 	struct rcu_pending_pcpu __percpu *p;
 	struct srcu_struct		*srcu;
+	unsigned long			*seq;
 	rcu_pending_process_fn		process;
 };
 
diff --git a/kernel/rcu/pending.c b/kernel/rcu/pending.c
index 2f9ad9ac6637..e26b1ee74acd 100644
--- a/kernel/rcu/pending.c
+++ b/kernel/rcu/pending.c
@@ -8,6 +8,8 @@
 #include <linux/srcu.h>
 #include <linux/vmalloc.h>
 
+#include "rcu.h"
+
 enum rcu_pending_special {
 	RCU_PENDING_KFREE	= 1,
 	RCU_PENDING_VFREE	= 2,
@@ -61,9 +63,9 @@ struct rcu_pending_seq {
 	 */
 	GENRADIX(struct rcu_head *)	objs;
 	size_t				nr;
+	struct rcu_head			**cursor;
 	/* Overflow list, if radix tree allocation fails */
 	struct rcu_head			*list;
-	unsigned long			seq;
 };
 
 struct rcu_pending_pcpu {
@@ -71,7 +73,8 @@ struct rcu_pending_pcpu {
 	spinlock_t			lock;
 	int				cpu:31;
 	bool				rcu_armed:1;
-	struct rcu_pending_seq		objs[2];
+	unsigned long			seq;
+	struct rcu_pending_seq		objs[NUM_ACTIVE_RCU_POLL_OLDSTATE];
 	struct rcu_head			rcu;
 	struct work_struct		work;
 };
@@ -104,6 +107,7 @@ static void rcu_pending_rcu_cb(struct rcu_head *rcu)
 static noinline void __rcu_pending_arm_cb(struct rcu_pending *pending, struct rcu_pending_pcpu *p)
 {
 	p->rcu_armed = true;
+	/* XXX: enqueue cb to when oldest seq completes */
 	__call_rcu(pending->srcu, &p->rcu, rcu_pending_rcu_cb);
 }
 
@@ -116,8 +120,7 @@ static inline void rcu_pending_arm_cb(struct rcu_pending *pending, struct rcu_pe
 static noinline void __process_finished_items(struct rcu_pending *pending,
 					      struct rcu_pending_pcpu *p,
 					      struct rcu_pending_seq *objs_p,
-					      unsigned long flags,
-					      bool locked)
+					      unsigned long flags)
 {
 	write_lock(&objs_p->objs.tree.free_lock);
 	struct rcu_pending_seq objs = (struct rcu_pending_seq) {
@@ -132,10 +135,9 @@ static noinline void __process_finished_items(struct rcu_pending *pending,
 	};
 	write_unlock(&objs_p->objs.tree.free_lock);
 	objs_p->nr	= 0;
+	objs_p->cursor	= NULL;
 	objs_p->list	= NULL;
-	if (locked)
-		spin_unlock(&p->lock);
-	local_irq_restore(flags);
+	spin_unlock_irqrestore(&p->lock, flags);
 
 	switch ((ulong) pending->process) {
 	case RCU_PENDING_KFREE:
@@ -180,27 +182,24 @@ static noinline void __process_finished_items(struct rcu_pending *pending,
 	}
 }
 
-static inline struct rcu_pending_seq *
-get_finished_items(struct srcu_struct *srcu,
-		   struct rcu_pending_pcpu *p)
+static bool process_finished_items(struct rcu_pending *pending,
+				   struct srcu_struct *srcu,
+				   struct rcu_pending_pcpu *p,
+				   unsigned long new_seq,
+				   unsigned long flags)
 {
-	for_each_object_list(p, objs)
-		if (!objs_empty(objs) &&
-		    __poll_state_synchronize_rcu(srcu, objs->seq))
-			return objs;
-	return NULL;
-}
+	unsigned long old_seq = p->seq;
 
-static inline bool process_finished_items(struct rcu_pending *pending,
-					  struct srcu_struct *srcu,
-					  struct rcu_pending_pcpu *p,
-					  unsigned long flags,
-					  bool locked)
-{
-	struct rcu_pending_seq *finished = get_finished_items(srcu, p);
-	if (unlikely(finished))
-		__process_finished_items(pending, p, finished, flags, locked);
-	return finished != NULL;
+	int keep = max(0, 2 - (int) ((new_seq - old_seq) >> RCU_SEQ_CTR_SHIFT));
+
+	for (struct rcu_pending_seq *objs = p->objs + ARRAY_SIZE(p->objs) - 1;
+	     objs >= p->objs + keep;
+	     --objs)
+		if (!objs_empty(objs)) {
+			__process_finished_items(pending, p, objs, flags);
+			return true;
+		}
+	return false;
 }
 
 static void rcu_pending_work(struct work_struct *work)
@@ -208,11 +207,12 @@ static void rcu_pending_work(struct work_struct *work)
 	struct rcu_pending_pcpu *p =
 		container_of(work, struct rcu_pending_pcpu, work);
 	struct rcu_pending *pending = p->parent;
-	unsigned long flags;
+	unsigned long seq, flags;
 
 	do {
 		spin_lock_irqsave(&p->lock, flags);
-	} while (process_finished_items(pending, pending->srcu, p, flags, true));
+		seq = *pending->seq & ~((ulong) RCU_SEQ_STATE_MASK);
+	} while (process_finished_items(pending, pending->srcu, p, seq, flags));
 
 	BUG_ON(!p->rcu_armed);
 	p->rcu_armed = false;
@@ -222,81 +222,121 @@ static void rcu_pending_work(struct work_struct *work)
 	spin_unlock_irqrestore(&p->lock, flags);
 }
 
+static noinline struct rcu_pending_seq *
+start_new_object_list(struct srcu_struct *srcu, struct rcu_pending_pcpu *p)
+{
+	BUG_ON(!objs_empty(&p->objs[1]));
+
+	/* start a new grace period */
+	__start_poll_synchronize_rcu(srcu);
+
+	/* we can't do a straight object copy because of the
+	 * lock in the genradix */
+	p->objs[1].objs.tree	= p->objs[0].objs.tree;
+	p->objs[1].nr		= p->objs[0].nr;
+	p->objs[1].cursor	= p->objs[0].cursor;
+	p->objs[1].list		= p->objs[0].list;
+
+	p->objs[0].objs.tree.root = NULL;
+	p->objs[0].nr		= 0;
+	p->objs[0].list		= NULL;
+	p->objs[0].cursor	= NULL;
+	return p->objs;
+}
+
 static __always_inline struct rcu_pending_seq *
 get_object_list(struct rcu_pending *pending, struct srcu_struct *srcu,
-		struct rcu_pending_pcpu **pp, unsigned long *flags,
-		bool locked)
+		struct rcu_pending_pcpu **pp,
+		unsigned long seq, unsigned long *flags)
 {
 	struct rcu_pending_pcpu *p;
-
 relock:
 	local_irq_save(*flags);
 	p = *pp = this_cpu_ptr(pending->p);
-	if (locked)
-		spin_lock(&p->lock);
-process_finished:
-	if (process_finished_items(pending, srcu, p, *flags, locked))
-		goto relock;
-
-	unsigned long seq = __get_state_synchronize_rcu(srcu);
-	for_each_object_list(p, objs)
-		if (!objs_empty(objs) && objs->seq == seq)
-			return objs;
+	spin_lock(&p->lock);
+
+	if (unlikely(objs_empty(p->objs) || seq > p->seq)) {
+		if (process_finished_items(pending, srcu, p, seq, *flags))
+			goto relock;
+		start_new_object_list(srcu, p);
+		p->seq = seq;
+		return p->objs;
+	}
 
-	seq = __start_poll_synchronize_rcu(srcu);
-	for_each_object_list(p, objs)
-		if (objs_empty(objs)) {
-			objs->seq = seq;
-			return objs;
-		}
+	ulong idx = ((long) (p->seq - seq)) >> 2;
+	if (likely(idx < ARRAY_SIZE(p->objs)))
+		return p->objs + idx;
 
-	/*
-	 * start_poll_synchronize_rcu() raced with our previous
-	 * process_finished_items(), and we now have another set of objects that
-	 * are ready to be processed
-	 */
-	goto process_finished;
+	/* seq must be expired */
+	return NULL;
 }
 
-void rcu_pending_enqueue(struct rcu_pending *pending, struct rcu_head *obj)
+static void __rcu_pending_enqueue(struct rcu_pending *pending, struct rcu_head *head, void *ptr)
 {
+
 	struct rcu_pending_pcpu *p;
 	struct rcu_pending_seq *objs;
-	struct rcu_head **entry;
-	unsigned long flags;
-	bool may_sleep = true;
+	unsigned long seq = *pending->seq & ~((ulong) RCU_SEQ_STATE_MASK), flags;
+	bool may_sleep = !ptr || !head;
 relock:
-	objs = get_object_list(pending, pending->srcu, &p, &flags, true);
-	entry = genradix_ptr_alloc_inlined(&objs->objs, objs->nr, GFP_ATOMIC|__GFP_NOWARN);
-	if (likely(entry)) {
-		*entry = obj;
-		objs->nr++;
-	} else if (may_sleep) {
+	objs = get_object_list(pending, NULL, &p, seq, &flags);
+	if (unlikely(!objs)) {
 		spin_unlock_irqrestore(&p->lock, flags);
-		if (!genradix_ptr_alloc(&objs->objs, objs->nr, GFP_KERNEL))
-			may_sleep = false;
-		goto relock;
-	} else {
-		obj->next = objs->list;
-		objs->list = obj;
+		if (ptr)
+			kvfree(ptr);
+		else
+			pending->process(pending, head);
+		return;
 	}
 
+	if (unlikely(!objs->cursor)) {
+		objs->cursor = genradix_ptr_alloc_inlined(&objs->objs, objs->nr,
+							  GFP_ATOMIC|__GFP_NOWARN);
+		if (unlikely(!objs->cursor)) {
+			if (may_sleep) {
+				spin_unlock_irqrestore(&p->lock, flags);
+
+				gfp_t gfp = GFP_KERNEL;
+				if (!head)
+					gfp |= __GFP_NOFAIL;
+
+				may_sleep = genradix_ptr_alloc(&objs->objs, objs->nr, gfp) != NULL;
+				goto relock;
+			} else {
+				if (ptr)
+					head->func = ptr;
+				head->next = objs->list;
+				objs->list = head;
+				goto done;
+			}
+		}
+	}
+
+	*objs->cursor++ = ptr;
+	if (!(((ulong) objs->cursor) & (GENRADIX_NODE_SIZE - 1)))
+		objs->cursor = NULL;
+	objs->nr++;
+done:
 	rcu_pending_arm_cb(pending, p);
 	spin_unlock_irqrestore(&p->lock, flags);
 }
 
+void rcu_pending_enqueue(struct rcu_pending *pending, struct rcu_head *obj)
+{
+	__rcu_pending_enqueue(pending, obj, NULL);
+}
+
 static struct rcu_head *rcu_pending_pcpu_dequeue(struct rcu_pending_pcpu *p)
 {
 	struct rcu_head *ret = NULL;
 
 	spin_lock_irq(&p->lock);
-	unsigned idx = p->objs[1].seq > p->objs[0].seq;
-
-	for (unsigned i = 0; i < 2; i++, idx ^= 1) {
-		struct rcu_pending_seq *objs = p->objs + idx;
-
+	for (struct rcu_pending_seq *objs = p->objs + ARRAY_SIZE(p->objs) - 1;
+	     objs >= p->objs;
+	     --objs) {
 		if (objs->nr) {
 			ret = *genradix_ptr(&objs->objs, --objs->nr);
+			objs->cursor = NULL;
 			break;
 		}
 
@@ -394,6 +434,7 @@ int rcu_pending_init(struct rcu_pending *pending,
 	}
 
 	pending->srcu = srcu;
+	pending->seq	= rcu_get_seq();
 	pending->process = process;
 
 	return 0;
@@ -402,44 +443,20 @@ int rcu_pending_init(struct rcu_pending *pending,
 #ifndef CONFIG_TINY_RCU
 /* kvfree_rcu */
 
-static struct rcu_pending kfree_rcu_pending;
-static struct rcu_pending vfree_rcu_pending;
+static struct rcu_pending kvfree_rcu_pending[2];
 
 void kvfree_call_rcu(struct rcu_head *head, void *ptr)
 {
-	struct rcu_pending *pending = is_vmalloc_addr_inlined(ptr)
-		? &vfree_rcu_pending
-		: &kfree_rcu_pending;
+	BUG_ON(!ptr);
 
-	struct rcu_pending_pcpu *p;
-	struct rcu_pending_seq *objs;
-	struct rcu_head **entry;
-	unsigned long flags;
-reget:
-	objs = get_object_list(pending, NULL, &p, &flags, false);
-	entry = genradix_ptr_alloc_inlined(&objs->objs, objs->nr, GFP_ATOMIC|__GFP_NOWARN);
-	if (likely(entry)) {
-		*entry = ptr;
-		objs->nr++;
-	} else if (head) {
-		head->func = ptr;
-		head->next = objs->list;
-		objs->list = head;
-	} else {
-		local_irq_restore(flags);
-		genradix_ptr_alloc(&objs->objs, objs->nr, GFP_KERNEL|__GFP_NOFAIL);
-		goto reget;
-	}
-
-	rcu_pending_arm_cb(pending, p);
-	local_irq_restore(flags);
+	__rcu_pending_enqueue(&kvfree_rcu_pending[is_vmalloc_addr_inlined(ptr)], head, ptr);
 }
 EXPORT_SYMBOL_GPL(kvfree_call_rcu);
 
 void __init kvfree_rcu_pending_init(void)
 {
-	if (rcu_pending_init(&kfree_rcu_pending, NULL, RCU_PENDING_KFREE_FN) ?:
-	    rcu_pending_init(&vfree_rcu_pending, NULL, RCU_PENDING_VFREE_FN))
+	if (rcu_pending_init(&kvfree_rcu_pending[0], NULL, RCU_PENDING_KFREE_FN) ?:
+	    rcu_pending_init(&kvfree_rcu_pending[1], NULL, RCU_PENDING_VFREE_FN))
 		panic("%s failed\n", __func__);
 }
 #endif
diff --git a/kernel/rcu/rcu.h b/kernel/rcu/rcu.h
index 38238e595a61..3834124f1ac6 100644
--- a/kernel/rcu/rcu.h
+++ b/kernel/rcu/rcu.h
@@ -207,6 +207,8 @@ static inline unsigned long rcu_seq_diff(unsigned long new, unsigned long old)
 	return ((rnd_diff - RCU_SEQ_STATE_MASK - 1) >> RCU_SEQ_CTR_SHIFT) + 2;
 }
 
+unsigned long *rcu_get_seq(void);
+
 /*
  * debug_rcu_head_queue()/debug_rcu_head_unqueue() are used internally
  * by call_rcu() and rcu callback execution, and are therefore not part
diff --git a/kernel/rcu/tree.c b/kernel/rcu/tree.c
index 98633610c6c3..ac294d7e7e4c 100644
--- a/kernel/rcu/tree.c
+++ b/kernel/rcu/tree.c
@@ -3348,6 +3348,11 @@ unsigned long get_state_synchronize_rcu(void)
 }
 EXPORT_SYMBOL_GPL(get_state_synchronize_rcu);
 
+unsigned long *rcu_get_seq(void)
+{
+	return &rcu_state.gp_seq_polled;
+}
+
 /**
  * get_state_synchronize_rcu_full - Snapshot RCU state, both normal and expedited
  * @rgosp: location to place combined normal/expedited grace-period state