[PATCH 0/5] rcusync: validations + dtor + exclusive

[PATCH 0/5] rcusync: validations + dtor + exclusive

[Oleg Nesterov]

Hello.

On top of Peter's "[PATCH 0/3] Optimize the cpu hotplug locking"
series.

It seems that Paul and Peter agree with 1-3, and they are simple.

4-5 should be ignored without the explicit ack. Peter seems to
dislike the exclusive mode. I disagree but I am not going to argue.
(well, at least right now ;)

Oleg.

 include/linux/rcusync.h |   80 ++++++++++++++++++++------------------
 kernel/cpu.c            |    2 +-
 kernel/rcusync.c        |   98 +++++++++++++++++++++++++++++++++-------------
 3 files changed, 113 insertions(+), 67 deletions(-)

[PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Oleg Nesterov]

Add the new struct rcu_sync_ops which holds sync/call methods, and
turn the function pointers in rcu_sync_struct into the single pointer
to struct rcu_sync_ops.

This simplifies the "init" helpers, and this way it is simpler to add
the new methods we need, especially ifdef'ed.

Signed-off-by: Oleg Nesterov <oleg@redhat.com>
---
 include/linux/rcusync.h |   63 +++++++++++++++++++++--------------------------
 kernel/rcusync.c        |   40 ++++++++++++++---------------
 2 files changed, 47 insertions(+), 56 deletions(-)

diff --git a/include/linux/rcusync.h b/include/linux/rcusync.h
index 7858491..30c6037 100644
--- a/include/linux/rcusync.h
+++ b/include/linux/rcusync.h
@@ -4,6 +4,11 @@
 #include <linux/wait.h>
 #include <linux/rcupdate.h>
 
+struct rcu_sync_ops {
+	void (*sync)(void);
+	void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
+};
+
 struct rcu_sync_struct {
 	int			gp_state;
 	int			gp_count;
@@ -12,43 +17,9 @@ struct rcu_sync_struct {
 	int			cb_state;
 	struct rcu_head		cb_head;
 
-	void (*sync)(void);
-	void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
+	struct rcu_sync_ops	*ops;
 };
 
-#define ___RCU_SYNC_INIT(name)						\
-	.gp_state = 0,							\
-	.gp_count = 0,							\
-	.gp_wait = __WAIT_QUEUE_HEAD_INITIALIZER(name.gp_wait),		\
-	.cb_state = 0
-
-#define __RCU_SCHED_SYNC_INIT(name) {					\
-	___RCU_SYNC_INIT(name),						\
-	.sync = synchronize_sched,					\
-	.call = call_rcu_sched,						\
-}
-
-#define __RCU_BH_SYNC_INIT(name) {					\
-	___RCU_SYNC_INIT(name),						\
-	.sync = synchronize_rcu_bh,					\
-	.call = call_rcu_bh,						\
-}
-
-#define __RCU_SYNC_INIT(name) {						\
-	___RCU_SYNC_INIT(name),						\
-	.sync = synchronize_rcu,					\
-	.call = call_rcu,						\
-}
-
-#define DEFINE_RCU_SCHED_SYNC(name)					\
-	struct rcu_sync_struct name = __RCU_SCHED_SYNC_INIT(name)
-
-#define DEFINE_RCU_BH_SYNC(name)					\
-	struct rcu_sync_struct name = __RCU_BH_SYNC_INIT(name)
-
-#define DEFINE_RCU_SYNC(name)						\
-	struct rcu_sync_struct name = __RCU_SYNC_INIT(name)
-
 static inline bool rcu_sync_is_idle(struct rcu_sync_struct *rss)
 {
 	return !rss->gp_state; /* GP_IDLE */
@@ -60,5 +31,27 @@ extern void rcu_sync_init(struct rcu_sync_struct *, enum rcu_sync_type);
 extern void rcu_sync_enter(struct rcu_sync_struct *);
 extern void rcu_sync_exit(struct rcu_sync_struct *);
 
+extern struct rcu_sync_ops rcu_sync_ops_array[];
+
+#define __RCU_SYNC_INITIALIZER(name, type) {				\
+		.gp_state = 0,						\
+		.gp_count = 0,						\
+		.gp_wait = __WAIT_QUEUE_HEAD_INITIALIZER(name.gp_wait),	\
+		.cb_state = 0,						\
+		.ops = rcu_sync_ops_array + (type),			\
+	}
+
+#define	__DEFINE_RCU_SYNC(name, type)	\
+	struct rcu_sync_struct name = __RCU_SYNC_INITIALIZER(name, type)
+
+#define DEFINE_RCU_SYNC(name)		\
+	__DEFINE_RCU_SYNC(name, RCU_SYNC)
+
+#define DEFINE_RCU_SCHED_SYNC(name)	\
+	__DEFINE_RCU_SYNC(name, RCU_SCHED_SYNC)
+
+#define DEFINE_RCU_BH_SYNC(name)	\
+	__DEFINE_RCU_SYNC(name, RCU_BH_SYNC)
+
 #endif /* _LINUX_RCUSYNC_H_ */
 
diff --git a/kernel/rcusync.c b/kernel/rcusync.c
index f84176a..1cefb83 100644
--- a/kernel/rcusync.c
+++ b/kernel/rcusync.c
@@ -1,4 +1,3 @@
-
 #include <linux/rcusync.h>
 #include <linux/sched.h>
 
@@ -7,27 +6,26 @@ enum { CB_IDLE = 0, CB_PENDING, CB_REPLAY };
 
 #define	rss_lock	gp_wait.lock
 
+struct rcu_sync_ops rcu_sync_ops_array[] = {
+	[RCU_SYNC] = {
+		.sync = synchronize_rcu,
+		.call = call_rcu,
+	},
+	[RCU_SCHED_SYNC] = {
+		.sync = synchronize_sched,
+		.call = call_rcu_sched,
+	},
+	[RCU_BH_SYNC] = {
+		.sync = synchronize_rcu_bh,
+		.call = call_rcu_bh,
+	},
+};
+
 void rcu_sync_init(struct rcu_sync_struct *rss, enum rcu_sync_type type)
 {
 	memset(rss, 0, sizeof(*rss));
 	init_waitqueue_head(&rss->gp_wait);
-
-	switch (type) {
-	case RCU_SYNC:
-		rss->sync = synchronize_rcu;
-		rss->call = call_rcu;
-		break;
-
-	case RCU_SCHED_SYNC:
-		rss->sync = synchronize_sched;
-		rss->call = call_rcu_sched;
-		break;
-
-	case RCU_BH_SYNC:
-		rss->sync = synchronize_rcu_bh;
-		rss->call = call_rcu_bh;
-		break;
-	}
+	rss->ops = rcu_sync_ops_array + type;
 }
 
 void rcu_sync_enter(struct rcu_sync_struct *rss)
@@ -44,7 +42,7 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
 	BUG_ON(need_wait && need_sync);
 
 	if (need_sync) {
-		rss->sync();
+		rss->ops->sync();
 		rss->gp_state = GP_PASSED;
 		wake_up_all(&rss->gp_wait);
 	} else if (need_wait) {
@@ -81,7 +79,7 @@ static void rcu_sync_func(struct rcu_head *rcu)
 		 * to catch a later GP.
 		 */
 		rss->cb_state = CB_PENDING;
-		rss->call(&rss->cb_head, rcu_sync_func);
+		rss->ops->call(&rss->cb_head, rcu_sync_func);
 	} else {
 		/*
 		 * We're at least a GP after rcu_sync_exit(); eveybody will now
@@ -99,7 +97,7 @@ void rcu_sync_exit(struct rcu_sync_struct *rss)
 	if (!--rss->gp_count) {
 		if (rss->cb_state == CB_IDLE) {
 			rss->cb_state = CB_PENDING;
-			rss->call(&rss->cb_head, rcu_sync_func);
+			rss->ops->call(&rss->cb_head, rcu_sync_func);
 		} else if (rss->cb_state == CB_PENDING) {
 			rss->cb_state = CB_REPLAY;
 		}
-- 
1.5.5.1

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Linus Torvalds]

On Fri, Oct 4, 2013 at 11:46 AM, Oleg Nesterov <oleg@redhat.com> wrote:
> Add the new struct rcu_sync_ops which holds sync/call methods, and
> turn the function pointers in rcu_sync_struct into the single pointer
> to struct rcu_sync_ops.
>
> +struct rcu_sync_ops {
> +       void (*sync)(void);
> +       void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
> +};
> +
>  struct rcu_sync_struct {
>         int                     gp_state;
>         int                     gp_count;
> @@ -12,43 +17,9 @@ struct rcu_sync_struct {
>         int                     cb_state;
>         struct rcu_head         cb_head;
>
> -       void (*sync)(void);
> -       void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
> +       struct rcu_sync_ops     *ops;

Ugh.

This interface pretty much guarantees that a compiler can never do
anything clever, like know that "hey, you used a static initializer on
this thing, and the fields are const, so now know statically what the
functions are, and I can just turn the indirect jumps into direct
jumps".

I'm not sure gcc is actually that clever, but by making it this kind
of ops pointer, I *guarantee* that gcc can never do it.

How about you make the rule be:

 - get rid of the stupid "type" enum index thing

 - get rid of the "init" thing that sets pointers in the dynamic data
structures. Get rid of the pointer too.

 - instead, use a "static const" type descriptor for each type (it
approaches being your "rcu_sync_ops" structure). Pass this in as an
argument to all the functions (use a #define per type or something, so
that users don't need to do this by hand)

 - now every single user passes in that type descriptor.

 - together with using a few inline functions, suddenly the "indirect"
jumps through this type descriptor end up actually being nice direct
compile-time constants: iow, they get turned into direct jumps.

Tadaa. You actually get good code generation, and you use *less*
dynamic memory since you don't have to have this pointer to the
descriptor.

                Linus

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Oleg Nesterov]

On 10/04, Linus Torvalds wrote:
>
> On Fri, Oct 4, 2013 at 11:46 AM, Oleg Nesterov <oleg@redhat.com> wrote:
> > Add the new struct rcu_sync_ops which holds sync/call methods, and
> > turn the function pointers in rcu_sync_struct into the single pointer
> > to struct rcu_sync_ops.
> >
> > +struct rcu_sync_ops {
> > +       void (*sync)(void);
> > +       void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
> > +};
> > +
> >  struct rcu_sync_struct {
> >         int                     gp_state;
> >         int                     gp_count;
> > @@ -12,43 +17,9 @@ struct rcu_sync_struct {
> >         int                     cb_state;
> >         struct rcu_head         cb_head;
> >
> > -       void (*sync)(void);
> > -       void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
> > +       struct rcu_sync_ops     *ops;
>
> Ugh.
>
> This interface pretty much guarantees that a compiler can never do
> anything clever, like know that "hey, you used a static initializer on
> this thing, and the fields are const, so now know statically what the
> functions are, and I can just turn the indirect jumps into direct
> jumps".

But we do not care? rcu_sync_struct->ops is only used by the writer
(slow path). In this case the simpler the better, I think.

>  - instead, use a "static const" type descriptor for each type (it
> approaches being your "rcu_sync_ops" structure). Pass this in as an
> argument to all the functions (use a #define per type or something, so
> that users don't need to do this by hand)
>
>  - now every single user passes in that type descriptor.
>
>  - together with using a few inline functions, suddenly the "indirect"
> jumps through this type descriptor end up actually being nice direct
> compile-time constants: iow, they get turned into direct jumps.

Hmm. Can't understand, sorry... Could you spell??

I assume you do not suggest to pass the "type" to, say, rcu_sync_enter?

Oleg.

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Steven Rostedt]

On Fri, 4 Oct 2013 12:12:25 -0700
Linus Torvalds <torvalds@linux-foundation.org> wrote:


>  - together with using a few inline functions, suddenly the "indirect"
> jumps through this type descriptor end up actually being nice direct
> compile-time constants: iow, they get turned into direct jumps.

As all the rcu_synchronization() methods (on non UP) are quite
expensive, I doubt that this optimization is worth anything.

> 
> Tadaa. You actually get good code generation, and you use *less*
> dynamic memory since you don't have to have this pointer to the
> descriptor.

Getting rid of the extra dynamic memory and the pointer business, on the
other hand, does make your suggestion worth doing.

-- Steve

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Linus Torvalds]

On Fri, Oct 4, 2013 at 12:30 PM, Steven Rostedt <rostedt@goodmis.org> wrote:
>
> As all the rcu_synchronization() methods (on non UP) are quite
> expensive, I doubt that this optimization is worth anything.

Maybe. It just annoys me, because afaik, the function that gets called
is always static per callsite.

                  Linus

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Peter Zijlstra]

On Fri, Oct 04, 2013 at 12:38:37PM -0700, Linus Torvalds wrote:
> On Fri, Oct 4, 2013 at 12:30 PM, Steven Rostedt <rostedt@goodmis.org> wrote:
> >
> > As all the rcu_synchronization() methods (on non UP) are quite
> > expensive, I doubt that this optimization is worth anything.
> 
> Maybe. It just annoys me, because afaik, the function that gets called
> is always static per callsite.

Yes, very much so indeed. Worst is that we have no users of the regular
RCU and RCU_BH variants and only included them for completeness since
the general operation is just as valid for those.

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Oleg Nesterov]

On 10/04, Peter Zijlstra wrote:
>
> On Fri, Oct 04, 2013 at 12:38:37PM -0700, Linus Torvalds wrote:
> > On Fri, Oct 4, 2013 at 12:30 PM, Steven Rostedt <rostedt@goodmis.org> wrote:
> > >
> > > As all the rcu_synchronization() methods (on non UP) are quite
> > > expensive, I doubt that this optimization is worth anything.
> >
> > Maybe. It just annoys me, because afaik, the function that gets called
> > is always static per callsite.
>
> Yes, very much so indeed. Worst is that we have no users of the regular
> RCU and RCU_BH variants and only included them for completeness since
> the general operation is just as valid for those.

And personally I think we should keep type/ops for completeness anyway,
even if we do not have RCU and RCU_BH users. But perhaps we can kill
RCU_SYNC and RCU_BH_SYNC enums/entries until we have a user.

Oleg.

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Oleg Nesterov]

On 10/04, Linus Torvalds wrote:
>
> On Fri, Oct 4, 2013 at 12:30 PM, Steven Rostedt <rostedt@goodmis.org> wrote:
> >
> > As all the rcu_synchronization() methods (on non UP) are quite
> > expensive, I doubt that this optimization is worth anything.
>
> Maybe. It just annoys me, because afaik, the function that gets called
> is always static per callsite.

I simply can't understand how we can improve the code generation.
And more importantly, why does this matter at all in this case.

So unless you strongly object, I'd prefer to keep it as is. Except
I forgot to add "const" to _array[], this needs v2.

To me the annoying part is that this patch exports rcu_sync_ops*.
We can add "enum rcu_sync_type rcu_sync_struct->gp_type" instead,
see the patch below. But then the next "add the CONFIG_PROVE_RCU
checks" needs to uninline rcu_sync_is_idle() for CONFIG_PROVE_RCU.

Or do you still think we should do something else?

Oleg.
---

diff --git a/include/linux/rcusync.h b/include/linux/rcusync.h
index 7858491..988ec33 100644
--- a/include/linux/rcusync.h
+++ b/include/linux/rcusync.h
@@ -4,6 +4,8 @@
 #include <linux/wait.h>
 #include <linux/rcupdate.h>
 
+enum rcu_sync_type { RCU_SYNC, RCU_SCHED_SYNC, RCU_BH_SYNC };
+
 struct rcu_sync_struct {
 	int			gp_state;
 	int			gp_count;
@@ -12,53 +14,37 @@ struct rcu_sync_struct {
 	int			cb_state;
 	struct rcu_head		cb_head;
 
-	void (*sync)(void);
-	void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
+	enum rcu_sync_type	gp_type;
 };
 
-#define ___RCU_SYNC_INIT(name)						\
-	.gp_state = 0,							\
-	.gp_count = 0,							\
-	.gp_wait = __WAIT_QUEUE_HEAD_INITIALIZER(name.gp_wait),		\
-	.cb_state = 0
-
-#define __RCU_SCHED_SYNC_INIT(name) {					\
-	___RCU_SYNC_INIT(name),						\
-	.sync = synchronize_sched,					\
-	.call = call_rcu_sched,						\
-}
-
-#define __RCU_BH_SYNC_INIT(name) {					\
-	___RCU_SYNC_INIT(name),						\
-	.sync = synchronize_rcu_bh,					\
-	.call = call_rcu_bh,						\
-}
-
-#define __RCU_SYNC_INIT(name) {						\
-	___RCU_SYNC_INIT(name),						\
-	.sync = synchronize_rcu,					\
-	.call = call_rcu,						\
-}
-
-#define DEFINE_RCU_SCHED_SYNC(name)					\
-	struct rcu_sync_struct name = __RCU_SCHED_SYNC_INIT(name)
-
-#define DEFINE_RCU_BH_SYNC(name)					\
-	struct rcu_sync_struct name = __RCU_BH_SYNC_INIT(name)
-
-#define DEFINE_RCU_SYNC(name)						\
-	struct rcu_sync_struct name = __RCU_SYNC_INIT(name)
-
 static inline bool rcu_sync_is_idle(struct rcu_sync_struct *rss)
 {
 	return !rss->gp_state; /* GP_IDLE */
 }
 
-enum rcu_sync_type { RCU_SYNC, RCU_SCHED_SYNC, RCU_BH_SYNC };
-
 extern void rcu_sync_init(struct rcu_sync_struct *, enum rcu_sync_type);
 extern void rcu_sync_enter(struct rcu_sync_struct *);
 extern void rcu_sync_exit(struct rcu_sync_struct *);
 
+#define __RCU_SYNC_INITIALIZER(name, type) {				\
+		.gp_state = 0,						\
+		.gp_count = 0,						\
+		.gp_wait = __WAIT_QUEUE_HEAD_INITIALIZER(name.gp_wait),	\
+		.cb_state = 0,						\
+		.gp_type = type,					\
+	}
+
+#define	__DEFINE_RCU_SYNC(name, type)	\
+	struct rcu_sync_struct name = __RCU_SYNC_INITIALIZER(name, type)
+
+#define DEFINE_RCU_SYNC(name)		\
+	__DEFINE_RCU_SYNC(name, RCU_SYNC)
+
+#define DEFINE_RCU_SCHED_SYNC(name)	\
+	__DEFINE_RCU_SYNC(name, RCU_SCHED_SYNC)
+
+#define DEFINE_RCU_BH_SYNC(name)	\
+	__DEFINE_RCU_SYNC(name, RCU_BH_SYNC)
+
 #endif /* _LINUX_RCUSYNC_H_ */
 
diff --git a/kernel/rcusync.c b/kernel/rcusync.c
index f84176a..99051b7 100644
--- a/kernel/rcusync.c
+++ b/kernel/rcusync.c
@@ -1,7 +1,24 @@
-
 #include <linux/rcusync.h>
 #include <linux/sched.h>
 
+static const struct {
+	void (*sync)(void);
+	void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
+} gp_ops[] = {
+	[RCU_SYNC] = {
+		.sync = synchronize_rcu,
+		.call = call_rcu,
+	},
+	[RCU_SCHED_SYNC] = {
+		.sync = synchronize_sched,
+		.call = call_rcu_sched,
+	},
+	[RCU_BH_SYNC] = {
+		.sync = synchronize_rcu_bh,
+		.call = call_rcu_bh,
+	},
+};
+
 enum { GP_IDLE = 0, GP_PENDING, GP_PASSED };
 enum { CB_IDLE = 0, CB_PENDING, CB_REPLAY };
 
@@ -11,23 +28,7 @@ void rcu_sync_init(struct rcu_sync_struct *rss, enum rcu_sync_type type)
 {
 	memset(rss, 0, sizeof(*rss));
 	init_waitqueue_head(&rss->gp_wait);
-
-	switch (type) {
-	case RCU_SYNC:
-		rss->sync = synchronize_rcu;
-		rss->call = call_rcu;
-		break;
-
-	case RCU_SCHED_SYNC:
-		rss->sync = synchronize_sched;
-		rss->call = call_rcu_sched;
-		break;
-
-	case RCU_BH_SYNC:
-		rss->sync = synchronize_rcu_bh;
-		rss->call = call_rcu_bh;
-		break;
-	}
+	rss->gp_type = type;
 }
 
 void rcu_sync_enter(struct rcu_sync_struct *rss)
@@ -44,7 +45,7 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
 	BUG_ON(need_wait && need_sync);
 
 	if (need_sync) {
-		rss->sync();
+		gp_ops[rss->gp_type].sync();
 		rss->gp_state = GP_PASSED;
 		wake_up_all(&rss->gp_wait);
 	} else if (need_wait) {
@@ -81,7 +82,7 @@ static void rcu_sync_func(struct rcu_head *rcu)
 		 * to catch a later GP.
 		 */
 		rss->cb_state = CB_PENDING;
-		rss->call(&rss->cb_head, rcu_sync_func);
+		gp_ops[rss->gp_type].call(&rss->cb_head, rcu_sync_func);
 	} else {
 		/*
 		 * We're at least a GP after rcu_sync_exit(); eveybody will now
@@ -99,7 +100,7 @@ void rcu_sync_exit(struct rcu_sync_struct *rss)
 	if (!--rss->gp_count) {
 		if (rss->cb_state == CB_IDLE) {
 			rss->cb_state = CB_PENDING;
-			rss->call(&rss->cb_head, rcu_sync_func);
+			gp_ops[rss->gp_type].call(&rss->cb_head, rcu_sync_func);
 		} else if (rss->cb_state == CB_PENDING) {
 			rss->cb_state = CB_REPLAY;
 		}

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Peter Zijlstra]

On Sat, Oct 05, 2013 at 07:17:46PM +0200, Oleg Nesterov wrote:
> To me the annoying part is that this patch exports rcu_sync_ops*.
> We can add "enum rcu_sync_type rcu_sync_struct->gp_type" instead,
> see the patch below. But then the next "add the CONFIG_PROVE_RCU
> checks" needs to uninline rcu_sync_is_idle() for CONFIG_PROVE_RCU.
> 
> Or do you still think we should do something else?

> @@ -44,7 +45,7 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
>  	BUG_ON(need_wait && need_sync);
>  
>  	if (need_sync) {
> -		rss->sync();
> +		gp_ops[rss->gp_type].sync();
>  		rss->gp_state = GP_PASSED;
>  		wake_up_all(&rss->gp_wait);
>  	} else if (need_wait) {
> @@ -81,7 +82,7 @@ static void rcu_sync_func(struct rcu_head *rcu)
>  		 * to catch a later GP.
>  		 */
>  		rss->cb_state = CB_PENDING;
> -		rss->call(&rss->cb_head, rcu_sync_func);
> +		gp_ops[rss->gp_type].call(&rss->cb_head, rcu_sync_func);
>  	} else {
>  		/*
>  		 * We're at least a GP after rcu_sync_exit(); eveybody will now
> @@ -99,7 +100,7 @@ void rcu_sync_exit(struct rcu_sync_struct *rss)
>  	if (!--rss->gp_count) {
>  		if (rss->cb_state == CB_IDLE) {
>  			rss->cb_state = CB_PENDING;
> -			rss->call(&rss->cb_head, rcu_sync_func);
> +			gp_ops[rss->gp_type].call(&rss->cb_head, rcu_sync_func);
>  		} else if (rss->cb_state == CB_PENDING) {
>  			rss->cb_state = CB_REPLAY;
>  		}
> 

I think Linus meant to have rcu_sync_{enter,exit} as inlines with a
const enum argument for the gp_type.

That said; yes that will generate better code, but also more code, and
like Steven already argued performance isn't really an issue here since
we're going to potentially sleep for a rather long time.

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Oleg Nesterov]

On 10/08, Peter Zijlstra wrote:
>
> I think Linus meant to have rcu_sync_{enter,exit} as inlines with a
> const enum argument for the gp_type.
>
> That said; yes that will generate better code, but also more code, and
> like Steven already argued performance isn't really an issue here since
> we're going to potentially sleep for a rather long time.

Yes, I do not think that we should make them "inline". Plus we need the
non-inline rcu_sync_func() anyway.

Thanks again for the last series you sent.

Oleg.

Re: [PATCH 1/5] rcusync: introduce struct rcu_sync_ops

[Paul E. McKenney]

On Tue, Oct 08, 2013 at 05:33:25PM +0200, Oleg Nesterov wrote:
> On 10/08, Peter Zijlstra wrote:
> >
> > I think Linus meant to have rcu_sync_{enter,exit} as inlines with a
> > const enum argument for the gp_type.
> >
> > That said; yes that will generate better code, but also more code, and
> > like Steven already argued performance isn't really an issue here since
> > we're going to potentially sleep for a rather long time.
> 
> Yes, I do not think that we should make them "inline". Plus we need the
> non-inline rcu_sync_func() anyway.

Yep.  Now if it was rcu_read_lock() rather than synchronize_rcu(),
it would be different.  ;-)

							Thanx, Paul

> Thanks again for the last series you sent.
> 
> Oleg.
> 

[PATCH 2/5] rcusync: add the CONFIG_PROVE_RCU checks

[Oleg Nesterov]

It would be nice to validate that the caller of rcu_sync_is_idle()
holds the corresponding type of RCU read-side lock. Add the new
rcu_sync_ops->held() method and change rcu_sync_is_idle() to
WARN() if it returns false.

This obviously penalizes the readers (fast-path), but only if
CONFIG_PROVE_RCU.

Suggested-by: "Paul E. McKenney" <paulmck@linux.vnet.ibm.com>
Signed-off-by: Oleg Nesterov <oleg@redhat.com>
---
 include/linux/rcusync.h |    6 ++++++
 kernel/rcusync.c        |    9 +++++++++
 2 files changed, 15 insertions(+), 0 deletions(-)

diff --git a/include/linux/rcusync.h b/include/linux/rcusync.h
index 30c6037..ab787c1 100644
--- a/include/linux/rcusync.h
+++ b/include/linux/rcusync.h
@@ -7,6 +7,9 @@
 struct rcu_sync_ops {
 	void (*sync)(void);
 	void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
+#ifdef CONFIG_PROVE_RCU
+	int (*held)(void);
+#endif
 };
 
 struct rcu_sync_struct {
@@ -22,6 +25,9 @@ struct rcu_sync_struct {
 
 static inline bool rcu_sync_is_idle(struct rcu_sync_struct *rss)
 {
+#ifdef CONFIG_PROVE_RCU
+	WARN_ON(!rss->ops->held());
+#endif
 	return !rss->gp_state; /* GP_IDLE */
 }
 
diff --git a/kernel/rcusync.c b/kernel/rcusync.c
index 1cefb83..21cde9b 100644
--- a/kernel/rcusync.c
+++ b/kernel/rcusync.c
@@ -6,18 +6,27 @@ enum { CB_IDLE = 0, CB_PENDING, CB_REPLAY };
 
 #define	rss_lock	gp_wait.lock
 
+#ifdef CONFIG_PROVE_RCU
+#define __INIT_HELD(func)	.held = func,
+#else
+#define __INIT_HELD(func)
+#endif
+
 struct rcu_sync_ops rcu_sync_ops_array[] = {
 	[RCU_SYNC] = {
 		.sync = synchronize_rcu,
 		.call = call_rcu,
+		__INIT_HELD(rcu_read_lock_held)
 	},
 	[RCU_SCHED_SYNC] = {
 		.sync = synchronize_sched,
 		.call = call_rcu_sched,
+		__INIT_HELD(rcu_read_lock_sched_held)
 	},
 	[RCU_BH_SYNC] = {
 		.sync = synchronize_rcu_bh,
 		.call = call_rcu_bh,
+		__INIT_HELD(rcu_read_lock_bh_held)
 	},
 };
 
-- 
1.5.5.1

[PATCH 3/5] rcusync: introduce rcu_sync_dtor()

[Oleg Nesterov]

Add the new rcu_sync_ops->wait() method and the new helper,
rcu_sync_dtor().

It is needed if you are going to, say, kfree(rcu_sync_object).
It simply calls ops->wait() to "flush" the potentially pending
rcu callback.

Signed-off-by: Oleg Nesterov <oleg@redhat.com>
---
 include/linux/rcusync.h |    2 ++
 kernel/rcusync.c        |   21 +++++++++++++++++++++
 2 files changed, 23 insertions(+), 0 deletions(-)

diff --git a/include/linux/rcusync.h b/include/linux/rcusync.h
index ab787c1..33864a0 100644
--- a/include/linux/rcusync.h
+++ b/include/linux/rcusync.h
@@ -7,6 +7,7 @@
 struct rcu_sync_ops {
 	void (*sync)(void);
 	void (*call)(struct rcu_head *, void (*)(struct rcu_head *));
+	void (*wait)(void);
 #ifdef CONFIG_PROVE_RCU
 	int (*held)(void);
 #endif
@@ -36,6 +37,7 @@ enum rcu_sync_type { RCU_SYNC, RCU_SCHED_SYNC, RCU_BH_SYNC };
 extern void rcu_sync_init(struct rcu_sync_struct *, enum rcu_sync_type);
 extern void rcu_sync_enter(struct rcu_sync_struct *);
 extern void rcu_sync_exit(struct rcu_sync_struct *);
+extern void rcu_sync_dtor(struct rcu_sync_struct *);
 
 extern struct rcu_sync_ops rcu_sync_ops_array[];
 
diff --git a/kernel/rcusync.c b/kernel/rcusync.c
index 21cde9b..bb311eb 100644
--- a/kernel/rcusync.c
+++ b/kernel/rcusync.c
@@ -16,16 +16,19 @@ struct rcu_sync_ops rcu_sync_ops_array[] = {
 	[RCU_SYNC] = {
 		.sync = synchronize_rcu,
 		.call = call_rcu,
+		.wait = rcu_barrier,
 		__INIT_HELD(rcu_read_lock_held)
 	},
 	[RCU_SCHED_SYNC] = {
 		.sync = synchronize_sched,
 		.call = call_rcu_sched,
+		.wait = rcu_barrier_sched,
 		__INIT_HELD(rcu_read_lock_sched_held)
 	},
 	[RCU_BH_SYNC] = {
 		.sync = synchronize_rcu_bh,
 		.call = call_rcu_bh,
+		.wait = rcu_barrier_bh,
 		__INIT_HELD(rcu_read_lock_bh_held)
 	},
 };
@@ -113,3 +116,21 @@ void rcu_sync_exit(struct rcu_sync_struct *rss)
 	}
 	spin_unlock_irq(&rss->rss_lock);
 }
+
+void rcu_sync_dtor(struct rcu_sync_struct *rss)
+{
+	int cb_state;
+
+	BUG_ON(rss->gp_count);
+
+	spin_lock_irq(&rss->rss_lock);
+	if (rss->cb_state == CB_REPLAY)
+		rss->cb_state = CB_PENDING;
+	cb_state = rss->cb_state;
+	spin_unlock_irq(&rss->rss_lock);
+
+	if (cb_state != CB_IDLE) {
+		rss->ops->wait();
+		BUG_ON(rss->cb_state != CB_IDLE);
+	}
+}
-- 
1.5.5.1

[PATCH 4/5] rcusync: introduce rcu_sync_struct->exclusive mode

[Oleg Nesterov]

Add rcu_sync_struct->exclusive boolean set by rcu_sync_init(),
it obviously controls the exclusiveness of rcu_sync_enter().
This is what percpu_down_write() actually wants.

We turn ->gp_wait into "struct completion gp_comp", it is used
as a resource counter in "exclusive" mode. Otherwise we only use
its completion->wait member for wait_event/wake_up_all. We never
mix the completion/wait_queue_head_t operations.

Note: it would be more clean to do __complete_locked() under
->rss_lock in rcu_sync_exit() in the "else" branch, but we don't
have this trivial helper.

Signed-off-by: Oleg Nesterov <oleg@redhat.com>
---
 include/linux/rcusync.h |   29 ++++++++++++++++-------------
 kernel/cpu.c            |    2 +-
 kernel/rcusync.c        |   22 +++++++++++++++++-----
 3 files changed, 34 insertions(+), 19 deletions(-)

diff --git a/include/linux/rcusync.h b/include/linux/rcusync.h
index 33864a0..5689f24 100644
--- a/include/linux/rcusync.h
+++ b/include/linux/rcusync.h
@@ -1,8 +1,8 @@
 #ifndef _LINUX_RCUSYNC_H_
 #define _LINUX_RCUSYNC_H_
 
-#include <linux/wait.h>
 #include <linux/rcupdate.h>
+#include <linux/completion.h>
 
 struct rcu_sync_ops {
 	void (*sync)(void);
@@ -16,11 +16,12 @@ struct rcu_sync_ops {
 struct rcu_sync_struct {
 	int			gp_state;
 	int			gp_count;
-	wait_queue_head_t	gp_wait;
+	struct completion	gp_comp;
 
 	int			cb_state;
 	struct rcu_head		cb_head;
 
+	bool			exclusive;
 	struct rcu_sync_ops	*ops;
 };
 
@@ -34,32 +35,34 @@ static inline bool rcu_sync_is_idle(struct rcu_sync_struct *rss)
 
 enum rcu_sync_type { RCU_SYNC, RCU_SCHED_SYNC, RCU_BH_SYNC };
 
-extern void rcu_sync_init(struct rcu_sync_struct *, enum rcu_sync_type);
+extern void rcu_sync_init(struct rcu_sync_struct *,
+				enum rcu_sync_type, bool excl);
 extern void rcu_sync_enter(struct rcu_sync_struct *);
 extern void rcu_sync_exit(struct rcu_sync_struct *);
 extern void rcu_sync_dtor(struct rcu_sync_struct *);
 
 extern struct rcu_sync_ops rcu_sync_ops_array[];
 
-#define __RCU_SYNC_INITIALIZER(name, type) {				\
+#define __RCU_SYNC_INITIALIZER(name, type, excl) {			\
 		.gp_state = 0,						\
 		.gp_count = 0,						\
-		.gp_wait = __WAIT_QUEUE_HEAD_INITIALIZER(name.gp_wait),	\
+		.gp_comp = COMPLETION_INITIALIZER(name.gp_comp),	\
 		.cb_state = 0,						\
+		.exclusive = excl,					\
 		.ops = rcu_sync_ops_array + (type),			\
 	}
 
-#define	__DEFINE_RCU_SYNC(name, type)	\
-	struct rcu_sync_struct name = __RCU_SYNC_INITIALIZER(name, type)
+#define	__DEFINE_RCU_SYNC(name, type, excl)	\
+	struct rcu_sync_struct name = __RCU_SYNC_INITIALIZER(name, type, excl)
 
-#define DEFINE_RCU_SYNC(name)		\
-	__DEFINE_RCU_SYNC(name, RCU_SYNC)
+#define DEFINE_RCU_SYNC(name, excl)		\
+	__DEFINE_RCU_SYNC(name, RCU_SYNC, excl)
 
-#define DEFINE_RCU_SCHED_SYNC(name)	\
-	__DEFINE_RCU_SYNC(name, RCU_SCHED_SYNC)
+#define DEFINE_RCU_SCHED_SYNC(name, excl)	\
+	__DEFINE_RCU_SYNC(name, RCU_SCHED_SYNC, excl)
 
-#define DEFINE_RCU_BH_SYNC(name)	\
-	__DEFINE_RCU_SYNC(name, RCU_BH_SYNC)
+#define DEFINE_RCU_BH_SYNC(name, excl)		\
+	__DEFINE_RCU_SYNC(name, RCU_BH_SYNC, excl)
 
 #endif /* _LINUX_RCUSYNC_H_ */
 
diff --git a/kernel/cpu.c b/kernel/cpu.c
index d5f475a..fb1bdf0 100644
--- a/kernel/cpu.c
+++ b/kernel/cpu.c
@@ -51,7 +51,7 @@ static int cpu_hotplug_disabled;
 
 enum { readers_slow, readers_block };
 
-DEFINE_RCU_SCHED_SYNC(__cpuhp_rss);
+DEFINE_RCU_SCHED_SYNC(__cpuhp_rss, false);
 EXPORT_SYMBOL_GPL(__cpuhp_rss);
 
 DEFINE_PER_CPU(unsigned int, __cpuhp_refcount);
diff --git a/kernel/rcusync.c b/kernel/rcusync.c
index bb311eb..667eb7d 100644
--- a/kernel/rcusync.c
+++ b/kernel/rcusync.c
@@ -4,7 +4,7 @@
 enum { GP_IDLE = 0, GP_PENDING, GP_PASSED };
 enum { CB_IDLE = 0, CB_PENDING, CB_REPLAY };
 
-#define	rss_lock	gp_wait.lock
+#define	rss_lock	gp_comp.wait.lock
 
 #ifdef CONFIG_PROVE_RCU
 #define __INIT_HELD(func)	.held = func,
@@ -33,11 +33,13 @@ struct rcu_sync_ops rcu_sync_ops_array[] = {
 	},
 };
 
-void rcu_sync_init(struct rcu_sync_struct *rss, enum rcu_sync_type type)
+void rcu_sync_init(struct rcu_sync_struct *rss,
+			enum rcu_sync_type type, bool excl)
 {
 	memset(rss, 0, sizeof(*rss));
-	init_waitqueue_head(&rss->gp_wait);
+	init_completion(&rss->gp_comp);
 	rss->ops = rcu_sync_ops_array + type;
+	rss->exclusive = excl;
 }
 
 void rcu_sync_enter(struct rcu_sync_struct *rss)
@@ -56,9 +58,13 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
 	if (need_sync) {
 		rss->ops->sync();
 		rss->gp_state = GP_PASSED;
-		wake_up_all(&rss->gp_wait);
+		if (!rss->exclusive)
+			wake_up_all(&rss->gp_comp.wait);
 	} else if (need_wait) {
-		wait_event(rss->gp_wait, rss->gp_state == GP_PASSED);
+		if (!rss->exclusive)
+			wait_event(rss->gp_comp.wait, rss->gp_state == GP_PASSED);
+		else
+			wait_for_completion(&rss->gp_comp);
 	} else {
 		/*
 		 * Possible when there's a pending CB from a rcu_sync_exit().
@@ -105,6 +111,8 @@ static void rcu_sync_func(struct rcu_head *rcu)
 
 void rcu_sync_exit(struct rcu_sync_struct *rss)
 {
+	bool wakeup_excl = rss->exclusive;
+
 	spin_lock_irq(&rss->rss_lock);
 	if (!--rss->gp_count) {
 		if (rss->cb_state == CB_IDLE) {
@@ -113,8 +121,12 @@ void rcu_sync_exit(struct rcu_sync_struct *rss)
 		} else if (rss->cb_state == CB_PENDING) {
 			rss->cb_state = CB_REPLAY;
 		}
+		wakeup_excl = false;
 	}
 	spin_unlock_irq(&rss->rss_lock);
+
+	if (wakeup_excl)
+		complete(&rss->gp_comp);
 }
 
 void rcu_sync_dtor(struct rcu_sync_struct *rss)
-- 
1.5.5.1

Re: [PATCH 4/5] rcusync: introduce rcu_sync_struct->exclusive mode

[Peter Zijlstra]

On Fri, Oct 04, 2013 at 08:46:40PM +0200, Oleg Nesterov wrote:
> Add rcu_sync_struct->exclusive boolean set by rcu_sync_init(),
> it obviously controls the exclusiveness of rcu_sync_enter().
> This is what percpu_down_write() actually wants.
> 
> We turn ->gp_wait into "struct completion gp_comp", it is used
> as a resource counter in "exclusive" mode. Otherwise we only use
> its completion->wait member for wait_event/wake_up_all. We never
> mix the completion/wait_queue_head_t operations.
> 
> Note: it would be more clean to do __complete_locked() under
> ->rss_lock in rcu_sync_exit() in the "else" branch, but we don't
> have this trivial helper.

Something equivalent in available functions would be:

	rss->gp_comp.done++;
	__wake_up_locked_key(&rss->gp_comp.wait, TASK_NORMAL, NULL);


>  struct rcu_sync_struct {
>  	int			gp_state;
>  	int			gp_count;
> -	wait_queue_head_t	gp_wait;
> +	struct completion	gp_comp;
>  
>  	int			cb_state;
>  	struct rcu_head		cb_head;
>  
> +	bool			exclusive;
>  	struct rcu_sync_ops	*ops;
>  };

I suppose we have a hole before or after cb_state to fit exclusive in.,
now it looks like we're going to create another hole before the *ops
pointer.

> @@ -4,7 +4,7 @@
>  enum { GP_IDLE = 0, GP_PENDING, GP_PASSED };
>  enum { CB_IDLE = 0, CB_PENDING, CB_REPLAY };
>  
> -#define	rss_lock	gp_wait.lock
> +#define	rss_lock	gp_comp.wait.lock

Should we, for convenience, also do:

#define rss_wait	gp_comp.wait

>  #ifdef CONFIG_PROVE_RCU
>  #define __INIT_HELD(func)	.held = func,
> @@ -33,11 +33,13 @@ struct rcu_sync_ops rcu_sync_ops_array[] = {
>  	},
>  };
>  
> -void rcu_sync_init(struct rcu_sync_struct *rss, enum rcu_sync_type type)
> +void rcu_sync_init(struct rcu_sync_struct *rss,
> +			enum rcu_sync_type type, bool excl)
>  {
>  	memset(rss, 0, sizeof(*rss));
> -	init_waitqueue_head(&rss->gp_wait);
> +	init_completion(&rss->gp_comp);
>  	rss->ops = rcu_sync_ops_array + type;
> +	rss->exclusive = excl;
>  }
>  
>  void rcu_sync_enter(struct rcu_sync_struct *rss)
> @@ -56,9 +58,13 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
>  	if (need_sync) {
>  		rss->ops->sync();
>  		rss->gp_state = GP_PASSED;
> -		wake_up_all(&rss->gp_wait);
> +		if (!rss->exclusive)
> +			wake_up_all(&rss->gp_comp.wait);
>  	} else if (need_wait) {
> -		wait_event(rss->gp_wait, rss->gp_state == GP_PASSED);
> +		if (!rss->exclusive)
> +			wait_event(rss->gp_comp.wait, rss->gp_state == GP_PASSED);
> +		else
> +			wait_for_completion(&rss->gp_comp);

I'm still not entirely sure why we need the completion; we already have
the gp_count variable and a waitqueue; together those should be able to
implement the condition/semaphore variable, no?

wait_for_completion:

	spin_lock_irq(&rss->rss_lock);
	if (rss->gp_count > 0) {
		__wait_event_locked(rss->gp_wait, (rss->gp_count > 0),
					TASK_UNINTERRUPTIBLE, 1, 0);
	}
	spin_unlock_irq(&rss->rss_lock);


complete:

	bool excl = rss->excl;

	spin_lock_irq(&rss->rss_lock);
	if (!--rss_gp_count) {
		__wake_up_locked_key(&rss->gp_comp.wait, TASK_NORMAL, NULL);
	}
	spin_unlock_irq(&rss->rss_lock);


All we would need would be something like:

--- a/include/linux/wait.h
+++ b/include/linux/wait.h
@@ -422,7 +422,7 @@ do {									\
 })
 
 
-#define __wait_event_interruptible_locked(wq, condition, exclusive, irq) \
+#define __wait_event_locked(wq, condition, state, exclusive, irq) 	\
 ({									\
 	int __ret = 0;							\
 	DEFINE_WAIT(__wait);						\
@@ -431,8 +431,8 @@ do {									\
 	do {								\
 		if (likely(list_empty(&__wait.task_list)))		\
 			__add_wait_queue_tail(&(wq), &__wait);		\
-		set_current_state(TASK_INTERRUPTIBLE);			\
-		if (signal_pending(current)) {				\
+		set_current_state(state);				\
+		if (___wait_signal_pending(state)) {			\
 			__ret = -ERESTARTSYS;				\
 			break;						\
 		}							\
@@ -451,6 +451,8 @@ do {									\
 	__ret;								\
 })
 
+#define __wait_event_interruptible_locked(wq, condition, exclusive, irq)\
+	__wait_event_locked(wq, condition, TASK_INTERRUPTIBLE, exclusive, irq)
 
 /**
  * wait_event_interruptible_locked - sleep until a condition gets true

Re: [PATCH 4/5] rcusync: introduce rcu_sync_struct->exclusive mode

[Oleg Nesterov]

On 10/04, Peter Zijlstra wrote:
>
> On Fri, Oct 04, 2013 at 08:46:40PM +0200, Oleg Nesterov wrote:
>
> > Note: it would be more clean to do __complete_locked() under
> > ->rss_lock in rcu_sync_exit() in the "else" branch, but we don't
> > have this trivial helper.
>
> Something equivalent in available functions would be:
>
> 	rss->gp_comp.done++;
> 	__wake_up_locked_key(&rss->gp_comp.wait, TASK_NORMAL, NULL);

Or __wake_up_locked(&rss->gp_comp.wait, TASK_NORMAL, 1).

Sure, this is what I had in mind. Just I thought that you also dislike
the idea to use/add the new helper ;) (and I think it would be better
to add the new helper even if we are not going to export it).

> >  struct rcu_sync_struct {
> >  	int			gp_state;
> >  	int			gp_count;
> > -	wait_queue_head_t	gp_wait;
> > +	struct completion	gp_comp;
> >
> >  	int			cb_state;
> >  	struct rcu_head		cb_head;
> >
> > +	bool			exclusive;
> >  	struct rcu_sync_ops	*ops;
> >  };
>
> I suppose we have a hole before or after cb_state to fit exclusive in.,
> now it looks like we're going to create another hole before the *ops
> pointer.

Yes, it probably makes sense to rearrange the members. And, for example,
gp_state and cb_state can be "char" and packed together.

> > @@ -4,7 +4,7 @@
> >  enum { GP_IDLE = 0, GP_PENDING, GP_PASSED };
> >  enum { CB_IDLE = 0, CB_PENDING, CB_REPLAY };
> >
> > -#define	rss_lock	gp_wait.lock
> > +#define	rss_lock	gp_comp.wait.lock
>
> Should we, for convenience, also do:
>
> #define rss_wait	gp_comp.wait

Yes, I considered this too. OK, will do.

> >  void rcu_sync_enter(struct rcu_sync_struct *rss)
> > @@ -56,9 +58,13 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
> >  	if (need_sync) {
> >  		rss->ops->sync();
> >  		rss->gp_state = GP_PASSED;
> > -		wake_up_all(&rss->gp_wait);
> > +		if (!rss->exclusive)
> > +			wake_up_all(&rss->gp_comp.wait);
> >  	} else if (need_wait) {
> > -		wait_event(rss->gp_wait, rss->gp_state == GP_PASSED);
> > +		if (!rss->exclusive)
> > +			wait_event(rss->gp_comp.wait, rss->gp_state == GP_PASSED);
> > +		else
> > +			wait_for_completion(&rss->gp_comp);
>
> I'm still not entirely sure why we need the completion; we already have
> the gp_count variable and a waitqueue;

and we also need the resource counter (like completion->done).

> together those should be able to
> implement the condition/semaphore variable, no?
>
> wait_for_completion:
>
> 	spin_lock_irq(&rss->rss_lock);
> 	if (rss->gp_count > 0) {
> 		__wait_event_locked(rss->gp_wait, (rss->gp_count > 0),

How? I do not even understand what did you mean ;) both conditions
are "gp_count > 0".

We simply can not define the CONDITION for wait_event() here, without
the additional accounting.

Hmm. perhaps you meant that this should be done before rcu_sync_enter()
increments ->gp_count. Perhaps this can work, but the code will be more
complex and this way rcu_sync_exit() will always schedule the callback?
And again, we do want to increment ->gp_count asap to disable this cb
if it is already pending.

Oleg.

Re: [PATCH 4/5] rcusync: introduce rcu_sync_struct->exclusive mode

[Peter Zijlstra]

On Fri, Oct 04, 2013 at 09:56:23PM +0200, Oleg Nesterov wrote:

> Hmm. perhaps you meant that this should be done before rcu_sync_enter()
> increments ->gp_count. Perhaps this can work, but the code will be more
> complex and this way rcu_sync_exit() will always schedule the callback?

Yah, however see below.

> And again, we do want to increment ->gp_count asap to disable this cb
> if it is already pending.

Ah indeed, I forgot about that. We'd have to wait until we'd get
scheduled to increment gp_count again. However I think we can fix this
and the above problem by introduction of rcu_sync_busy() which checks
for eiter gp_count or pending waiters.

But yes, slightly more complex code :/

That would yield something like so I suppose:

void rcu_sync_enter(struct rcu_sync_struct *rss)
{
	bool need_wait, need_sync;

	spin_lock_irq(&rss->rss_lock);
	if (rss->exclusive && rss->gp_count) {
		__wait_event_locked(rss->gp_wait, rss->gp_count);
		rss->gp_count++;
		need_wait = need_sync = false;
	} else {
		need_wait = rss->gp_count++;
		need_sync = rss->gp_state == GP_IDLE;
		if (need_sync)
			rss->gp_state = GP_PENDING;
	}
	spin_unlock_irq(&rss->lock);

	if (need_sync) {
		rss->sync();
		rss->gp_state = GP_PASSED;
		wake_up_all(&rss->gp_wait);
	} else if (need_wait) {
		wait_event(rss->gp_wait, rss->gp_state == GP_PASSED);
	} else {
		BUG_ON(rss->gp_state != GP_PASSED);
	}
}

static bool rcu_sync_busy(struct rcu_sync_struct *rss)
{
	return rss->gp_count ||
		(rss->exclusive && waitqueue_active(&rss->gp_wait));
}

static void rcu_sync_func(struct rcu_head *rcu)
{
	struct rcu_sync_struct *rss =
		container_of(rcu, struct rcu_sync_struct, cb_head);
	unsigned long flags;

	BUG_ON(rss->gp_state != GP_PASSED);
	BUG_ON(rss->cb_state == CB_IDLE);

	spin_lock_irqsave(&rss->rss_lock, flags);
	if (rcu_sync_busy(rss)) {
		/*
		 * A new rcu_sync_begin() has happened; drop the callback.
		 */
		rss->cb_state = CB_IDLE;
	} else if (rss->cb_state == CB_REPLAY) {
		/*
		 * A new rcu_sync_exit() has happened; requeue the callback
		 * to catch a later GP.
		 */
		rss->cb_state = CB_PENDING;
		rss->call(&rss->cb_head, rcu_sync_func);
	} else {
		/*
		 * We're at least a GP after rcu_sync_exit(); eveybody will now
		 * have observed the write side critical section. Let 'em rip!.
		 */
		rss->cb_state = CB_IDLE;
		rss->gp_state = GP_IDLE;
	}
	spin_unlock_irqrestore(&rss->rss_lock, flags);
}

void rcu_sync_exit(struct rcu_sync_struct *rss)
{
	spin_lock_irq(&rss->rss_lock);
	if (!--rss->gp_count) {
		if (!rcu_sync_busy(rss)) {
			if (rss->cb_state == CB_IDLE) {
				rss->cb_state = CB_PENDING;
				rss->call(&rss->cb_head, rcu_sync_func);
			} else if (rss->cb_state == CB_PENDING) {
				rss->cb_state = CB_REPLAY;
			}
		} else {
			__wake_up_locked(&rss->gp_wait, TASK_NORMAL, 1);
		}
	}
	spin_unlock_irq(&rss->rss_lock);
}

Re: [PATCH 4/5] rcusync: introduce rcu_sync_struct->exclusive mode

[Oleg Nesterov]

On 10/04, Peter Zijlstra wrote:
>
> On Fri, Oct 04, 2013 at 09:56:23PM +0200, Oleg Nesterov wrote:
>
> But yes, slightly more complex code :/

Yes. rcu_sync_busy() adds more obscurity and we need to implement
the logic which wait_for_completion already does.

> That would yield something like so I suppose:
>
> void rcu_sync_enter(struct rcu_sync_struct *rss)
> {
> 	bool need_wait, need_sync;
>
> 	spin_lock_irq(&rss->rss_lock);
> 	if (rss->exclusive && rss->gp_count) {
> 		__wait_event_locked(rss->gp_wait, rss->gp_count);
                                                  ^^^^^^^^^^^^^
I guess you meant !rss->gp_count.

> 		rss->gp_count++;
> 		need_wait = need_sync = false;
> 	} else {
> 		need_wait = rss->gp_count++;
> 		need_sync = rss->gp_state == GP_IDLE;
> 		if (need_sync)
> 			rss->gp_state = GP_PENDING;
> 	}
> 	spin_unlock_irq(&rss->lock);
>
> 	if (need_sync) {
> 		rss->sync();
> 		rss->gp_state = GP_PASSED;
> 		wake_up_all(&rss->gp_wait);
> 	} else if (need_wait) {
> 		wait_event(rss->gp_wait, rss->gp_state == GP_PASSED);
> 	} else {
> 		BUG_ON(rss->gp_state != GP_PASSED);
> 	}
> }

I am obviously biased, but imho the code looks worse this way.
I like the current simple "need_wait" and "gp_count != 0" logic.

And afaics this is racy,

> static bool rcu_sync_busy(struct rcu_sync_struct *rss)
> {
> 	return rss->gp_count ||
> 		(rss->exclusive && waitqueue_active(&rss->gp_wait));
> }
>
> static void rcu_sync_func(struct rcu_head *rcu)
> {
> 	struct rcu_sync_struct *rss =
> 		container_of(rcu, struct rcu_sync_struct, cb_head);
> 	unsigned long flags;
>
> 	BUG_ON(rss->gp_state != GP_PASSED);
> 	BUG_ON(rss->cb_state == CB_IDLE);
>
> 	spin_lock_irqsave(&rss->rss_lock, flags);
> 	if (rcu_sync_busy(rss)) {
> 		/*
> 		 * A new rcu_sync_begin() has happened; drop the callback.
> 		 */
> 		rss->cb_state = CB_IDLE;

Yes, but if rcu_sync_exit() does __wake_up_locked(), then
autoremove_wake_function() makes waitqueue_active() == F. If the pending
rcu_sync_func() takes ->rss_lock first we have a problem.

Easy to fix, but needs more complications.

Or we can simply ignore the fact that rcu_sync_func() can race with
wakeup. This can lead to unnecessary sched_sync() but this case is
unlikely. IOW,

	spin_lock_irq(&rss->rss_lock);
	if (rss->exclusive)
		wait_event_locked(rss->gp_wait, !rss->gp_count);
	need_wait = rss->gp_count++;
	need_sync = rss->gp_state == GP_IDLE;
	if (need_sync)
		rss->gp_state = GP_PENDING;
	spin_unlock_irq(&rss->lock);

But still I don't like the (imho) unnecessary complications. And the
fact we can race with rcu_sync_func() even if this is very unlikely,
this just doesn't look good.

Oleg.

Re: [PATCH 4/5] rcusync: introduce rcu_sync_struct->exclusive mode

[Peter Zijlstra]

On Sun, Oct 06, 2013 at 03:22:40PM +0200, Oleg Nesterov wrote:
> On 10/04, Peter Zijlstra wrote:
> > That would yield something like so I suppose:
> >
> > void rcu_sync_enter(struct rcu_sync_struct *rss)
> > {
> > 	bool need_wait, need_sync;
> >
> > 	spin_lock_irq(&rss->rss_lock);
> > 	if (rss->exclusive && rss->gp_count) {
> > 		__wait_event_locked(rss->gp_wait, rss->gp_count);
>                                                   ^^^^^^^^^^^^^
> I guess you meant !rss->gp_count.

Uh yes, obviously :-)

> I am obviously biased, but imho the code looks worse this way.
> I like the current simple "need_wait" and "gp_count != 0" logic.

Yeah, I know.. but it doesn't add the extra variable and doesn't play
games with the completion implementation.

> And afaics this is racy,
> 
> Yes, but if rcu_sync_exit() does __wake_up_locked(), then
> autoremove_wake_function() makes waitqueue_active() == F. If the pending
> rcu_sync_func() takes ->rss_lock first we have a problem.

Ah indeed, it seems I got confused between DECLARE_WAITQUEUE and
DEFINE_WAIT; there's too damn many variants there :/

> Easy to fix, but needs more complications.
> 
> Or we can simply ignore the fact that rcu_sync_func() can race with
> wakeup. This can lead to unnecessary sched_sync() but this case is
> unlikely. IOW,
> 
> 	spin_lock_irq(&rss->rss_lock);
> 	if (rss->exclusive)
> 		wait_event_locked(rss->gp_wait, !rss->gp_count);
> 	need_wait = rss->gp_count++;
> 	need_sync = rss->gp_state == GP_IDLE;
> 	if (need_sync)
> 		rss->gp_state = GP_PENDING;
> 	spin_unlock_irq(&rss->lock);
> 
> But still I don't like the (imho) unnecessary complications. And the
> fact we can race with rcu_sync_func() even if this is very unlikely,
> this just doesn't look good.

OK.. I'll give up trying to wreck this stuff ;-)

[PATCH 5/5] rcusync: make rcu_sync_enter() return "bool"

[Oleg Nesterov]

Change rcu_sync_enter() to return "need_sync" to let the caller
know whether we did the FAST -> SLOW transition or not.

This is particularly useful in exclusive mode, for example
percpu_down_write() can avoid clear_fast_ctr() if rcu_sync_enter()
returns F.

Signed-off-by: Oleg Nesterov <oleg@redhat.com>
---
 include/linux/rcusync.h |    2 +-
 kernel/rcusync.c        |    4 +++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/include/linux/rcusync.h b/include/linux/rcusync.h
index 5689f24..6f4c75b 100644
--- a/include/linux/rcusync.h
+++ b/include/linux/rcusync.h
@@ -37,7 +37,7 @@ enum rcu_sync_type { RCU_SYNC, RCU_SCHED_SYNC, RCU_BH_SYNC };
 
 extern void rcu_sync_init(struct rcu_sync_struct *,
 				enum rcu_sync_type, bool excl);
-extern void rcu_sync_enter(struct rcu_sync_struct *);
+extern bool rcu_sync_enter(struct rcu_sync_struct *);
 extern void rcu_sync_exit(struct rcu_sync_struct *);
 extern void rcu_sync_dtor(struct rcu_sync_struct *);
 
diff --git a/kernel/rcusync.c b/kernel/rcusync.c
index 667eb7d..d03d8e5 100644
--- a/kernel/rcusync.c
+++ b/kernel/rcusync.c
@@ -42,7 +42,7 @@ void rcu_sync_init(struct rcu_sync_struct *rss,
 	rss->exclusive = excl;
 }
 
-void rcu_sync_enter(struct rcu_sync_struct *rss)
+bool rcu_sync_enter(struct rcu_sync_struct *rss)
 {
 	bool need_wait, need_sync;
 
@@ -73,6 +73,8 @@ void rcu_sync_enter(struct rcu_sync_struct *rss)
 		 */
 		BUG_ON(rss->gp_state != GP_PASSED);
 	}
+
+	return need_sync;
 }
 
 static void rcu_sync_func(struct rcu_head *rcu)
-- 
1.5.5.1

Re: [PATCH 0/5] rcusync: validations + dtor + exclusive

[Peter Zijlstra]

On Fri, Oct 04, 2013 at 08:46:14PM +0200, Oleg Nesterov wrote:
> Hello.
> 
> On top of Peter's "[PATCH 0/3] Optimize the cpu hotplug locking"
> series.
> 
> It seems that Paul and Peter agree with 1-3, and they are simple.

Yeah, ACK on those.. shall I pick them up and stuff them through tip
with the rest of the lot?

> 4-5 should be ignored without the explicit ack. Peter seems to
> dislike the exclusive mode. I disagree but I am not going to argue.
> (well, at least right now ;)

Lets have another few rounds of email on that one; maybe I'll get used
to it :-)

Re: [PATCH 0/5] rcusync: validations + dtor + exclusive

[Paul E. McKenney]

On Fri, Oct 04, 2013 at 09:32:03PM +0200, Peter Zijlstra wrote:
> On Fri, Oct 04, 2013 at 08:46:14PM +0200, Oleg Nesterov wrote:
> > Hello.
> > 
> > On top of Peter's "[PATCH 0/3] Optimize the cpu hotplug locking"
> > series.
> > 
> > It seems that Paul and Peter agree with 1-3, and they are simple.
> 
> Yeah, ACK on those.. shall I pick them up and stuff them through tip
> with the rest of the lot?

I am good with that.

> > 4-5 should be ignored without the explicit ack. Peter seems to
> > dislike the exclusive mode. I disagree but I am not going to argue.
> > (well, at least right now ;)
> 
> Lets have another few rounds of email on that one; maybe I'll get used
> to it :-)

;-) ;-) ;-)

							Thanx, Paul

Re: [PATCH 0/5] rcusync: validations + dtor + exclusive

[Oleg Nesterov]

On 10/04, Peter Zijlstra wrote:
>
> shall I pick them up and stuff them through tip
> with the rest of the lot?

This would be great, thanks.

> > 4-5 should be ignored without the explicit ack. Peter seems to
> > dislike the exclusive mode. I disagree but I am not going to argue.
> > (well, at least right now ;)
>
> Lets have another few rounds of email on that one; maybe I'll get used
> to it :-)

OK ;) I'll send v2 anyway.

Oleg.