PATCH powerpc Merge asm-ppc*/rwsem.h

PATCH powerpc Merge asm-ppc*/rwsem.h

[Jon Loeliger]

Merge asm-ppc*/rwsem.h into include/asm-powerpc.
Removed smp_*mb() memory barriers from the ppc32 code
as they are now burried in the atomic_*() functions as
suggested by Paul, implemented by Arnd, and pushed out
by Becky.  I am not the droid you are looking for.

This patch depends on Becky's atomic.h merge patch.

Signed-off-by: Jon Loeliger <jdl@freescale.com>
Signed-off-by: Kumar Gala <kumar.gala@freescale.com>
---

 include/asm-powerpc/rwsem.h |  163 +++++++++++++++++++++++++++++++++++++++++
 include/asm-ppc/rwsem.h     |  172 -------------------------------------------
 include/asm-ppc64/rwsem.h   |  167 ------------------------------------------
 3 files changed, 163 insertions(+), 339 deletions(-)


diff --git a/include/asm-powerpc/rwsem.h b/include/asm-powerpc/rwsem.h
new file mode 100644
--- /dev/null
+++ b/include/asm-powerpc/rwsem.h
@@ -0,0 +1,163 @@
+#ifndef _ASM_POWERPC_RWSEM_H
+#define _ASM_POWERPC_RWSEM_H
+
+#ifdef __KERNEL__
+
+/*
+ * include/asm-ppc64/rwsem.h: R/W semaphores for PPC using the stuff
+ * in lib/rwsem.c.  Adapted largely from include/asm-i386/rwsem.h
+ * by Paul Mackerras <paulus@samba.org>.
+ */
+
+#include <linux/list.h>
+#include <linux/spinlock.h>
+#include <asm/atomic.h>
+#include <asm/system.h>
+
+/*
+ * the semaphore definition
+ */
+struct rw_semaphore {
+	/* XXX this should be able to be an atomic_t  -- paulus */
+	signed int		count;
+#define RWSEM_UNLOCKED_VALUE		0x00000000
+#define RWSEM_ACTIVE_BIAS		0x00000001
+#define RWSEM_ACTIVE_MASK		0x0000ffff
+#define RWSEM_WAITING_BIAS		(-0x00010000)
+#define RWSEM_ACTIVE_READ_BIAS		RWSEM_ACTIVE_BIAS
+#define RWSEM_ACTIVE_WRITE_BIAS		(RWSEM_WAITING_BIAS + RWSEM_ACTIVE_BIAS)
+	spinlock_t		wait_lock;
+	struct list_head	wait_list;
+#if RWSEM_DEBUG
+	int			debug;
+#endif
+};
+
+/*
+ * initialisation
+ */
+#if RWSEM_DEBUG
+#define __RWSEM_DEBUG_INIT      , 0
+#else
+#define __RWSEM_DEBUG_INIT	/* */
+#endif
+
+#define __RWSEM_INITIALIZER(name) \
+	{ RWSEM_UNLOCKED_VALUE, SPIN_LOCK_UNLOCKED, \
+	  LIST_HEAD_INIT((name).wait_list) \
+	  __RWSEM_DEBUG_INIT }
+
+#define DECLARE_RWSEM(name)		\
+	struct rw_semaphore name = __RWSEM_INITIALIZER(name)
+
+extern struct rw_semaphore *rwsem_down_read_failed(struct rw_semaphore *sem);
+extern struct rw_semaphore *rwsem_down_write_failed(struct rw_semaphore *sem);
+extern struct rw_semaphore *rwsem_wake(struct rw_semaphore *sem);
+extern struct rw_semaphore *rwsem_downgrade_wake(struct rw_semaphore *sem);
+
+static inline void init_rwsem(struct rw_semaphore *sem)
+{
+	sem->count = RWSEM_UNLOCKED_VALUE;
+	spin_lock_init(&sem->wait_lock);
+	INIT_LIST_HEAD(&sem->wait_list);
+#if RWSEM_DEBUG
+	sem->debug = 0;
+#endif
+}
+
+/*
+ * lock for reading
+ */
+static inline void __down_read(struct rw_semaphore *sem)
+{
+	if (unlikely(atomic_inc_return((atomic_t *)(&sem->count)) <= 0))
+		rwsem_down_read_failed(sem);
+}
+
+static inline int __down_read_trylock(struct rw_semaphore *sem)
+{
+	int tmp;
+
+	while ((tmp = sem->count) >= 0) {
+		if (tmp == cmpxchg(&sem->count, tmp,
+				   tmp + RWSEM_ACTIVE_READ_BIAS)) {
+			return 1;
+		}
+	}
+	return 0;
+}
+
+/*
+ * lock for writing
+ */
+static inline void __down_write(struct rw_semaphore *sem)
+{
+	int tmp;
+
+	tmp = atomic_add_return(RWSEM_ACTIVE_WRITE_BIAS,
+				(atomic_t *)(&sem->count));
+	if (unlikely(tmp != RWSEM_ACTIVE_WRITE_BIAS))
+		rwsem_down_write_failed(sem);
+}
+
+static inline int __down_write_trylock(struct rw_semaphore *sem)
+{
+	int tmp;
+
+	tmp = cmpxchg(&sem->count, RWSEM_UNLOCKED_VALUE,
+		      RWSEM_ACTIVE_WRITE_BIAS);
+	return tmp == RWSEM_UNLOCKED_VALUE;
+}
+
+/*
+ * unlock after reading
+ */
+static inline void __up_read(struct rw_semaphore *sem)
+{
+	int tmp;
+
+	tmp = atomic_dec_return((atomic_t *)(&sem->count));
+	if (unlikely(tmp < -1 && (tmp & RWSEM_ACTIVE_MASK) == 0))
+		rwsem_wake(sem);
+}
+
+/*
+ * unlock after writing
+ */
+static inline void __up_write(struct rw_semaphore *sem)
+{
+	if (unlikely(atomic_sub_return(RWSEM_ACTIVE_WRITE_BIAS,
+			      (atomic_t *)(&sem->count)) < 0))
+		rwsem_wake(sem);
+}
+
+/*
+ * implement atomic add functionality
+ */
+static inline void rwsem_atomic_add(int delta, struct rw_semaphore *sem)
+{
+	atomic_add(delta, (atomic_t *)(&sem->count));
+}
+
+/*
+ * downgrade write lock to read lock
+ */
+static inline void __downgrade_write(struct rw_semaphore *sem)
+{
+	int tmp;
+
+	tmp = atomic_add_return(-RWSEM_WAITING_BIAS, (atomic_t *)(&sem->count));
+	if (tmp < 0)
+		rwsem_downgrade_wake(sem);
+}
+
+/*
+ * implement exchange and add functionality
+ */
+static inline int rwsem_atomic_update(int delta, struct rw_semaphore *sem)
+{
+	return atomic_add_return(delta, (atomic_t *)(&sem->count));
+}
+
+#endif	/* __KERNEL__ */
+#endif	/* _ASM_POWERPC_RWSEM_H */
diff --git a/include/asm-ppc/rwsem.h b/include/asm-ppc/rwsem.h
deleted file mode 100644
--- a/include/asm-ppc/rwsem.h
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * include/asm-ppc/rwsem.h: R/W semaphores for PPC using the stuff
- * in lib/rwsem.c.  Adapted largely from include/asm-i386/rwsem.h
- * by Paul Mackerras <paulus@samba.org>.
- */
-
-#ifndef _PPC_RWSEM_H
-#define _PPC_RWSEM_H
-
-#ifdef __KERNEL__
-#include <linux/list.h>
-#include <linux/spinlock.h>
-#include <asm/atomic.h>
-#include <asm/system.h>
-
-/*
- * the semaphore definition
- */
-struct rw_semaphore {
-	/* XXX this should be able to be an atomic_t  -- paulus */
-	signed long		count;
-#define RWSEM_UNLOCKED_VALUE		0x00000000
-#define RWSEM_ACTIVE_BIAS		0x00000001
-#define RWSEM_ACTIVE_MASK		0x0000ffff
-#define RWSEM_WAITING_BIAS		(-0x00010000)
-#define RWSEM_ACTIVE_READ_BIAS		RWSEM_ACTIVE_BIAS
-#define RWSEM_ACTIVE_WRITE_BIAS		(RWSEM_WAITING_BIAS + RWSEM_ACTIVE_BIAS)
-	spinlock_t		wait_lock;
-	struct list_head	wait_list;
-#if RWSEM_DEBUG
-	int			debug;
-#endif
-};
-
-/*
- * initialisation
- */
-#if RWSEM_DEBUG
-#define __RWSEM_DEBUG_INIT      , 0
-#else
-#define __RWSEM_DEBUG_INIT	/* */
-#endif
-
-#define __RWSEM_INITIALIZER(name) \
-	{ RWSEM_UNLOCKED_VALUE, SPIN_LOCK_UNLOCKED, \
-	  LIST_HEAD_INIT((name).wait_list) \
-	  __RWSEM_DEBUG_INIT }
-
-#define DECLARE_RWSEM(name)		\
-	struct rw_semaphore name = __RWSEM_INITIALIZER(name)
-
-extern struct rw_semaphore *rwsem_down_read_failed(struct rw_semaphore *sem);
-extern struct rw_semaphore *rwsem_down_write_failed(struct rw_semaphore *sem);
-extern struct rw_semaphore *rwsem_wake(struct rw_semaphore *sem);
-extern struct rw_semaphore *rwsem_downgrade_wake(struct rw_semaphore *sem);
-
-static inline void init_rwsem(struct rw_semaphore *sem)
-{
-	sem->count = RWSEM_UNLOCKED_VALUE;
-	spin_lock_init(&sem->wait_lock);
-	INIT_LIST_HEAD(&sem->wait_list);
-#if RWSEM_DEBUG
-	sem->debug = 0;
-#endif
-}
-
-/*
- * lock for reading
- */
-static inline void __down_read(struct rw_semaphore *sem)
-{
-	if (atomic_inc_return((atomic_t *)(&sem->count)) > 0)
-		smp_wmb();
-	else
-		rwsem_down_read_failed(sem);
-}
-
-static inline int __down_read_trylock(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	while ((tmp = sem->count) >= 0) {
-		if (tmp == cmpxchg(&sem->count, tmp,
-				   tmp + RWSEM_ACTIVE_READ_BIAS)) {
-			smp_wmb();
-			return 1;
-		}
-	}
-	return 0;
-}
-
-/*
- * lock for writing
- */
-static inline void __down_write(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	tmp = atomic_add_return(RWSEM_ACTIVE_WRITE_BIAS,
-				(atomic_t *)(&sem->count));
-	if (tmp == RWSEM_ACTIVE_WRITE_BIAS)
-		smp_wmb();
-	else
-		rwsem_down_write_failed(sem);
-}
-
-static inline int __down_write_trylock(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	tmp = cmpxchg(&sem->count, RWSEM_UNLOCKED_VALUE,
-		      RWSEM_ACTIVE_WRITE_BIAS);
-	smp_wmb();
-	return tmp == RWSEM_UNLOCKED_VALUE;
-}
-
-/*
- * unlock after reading
- */
-static inline void __up_read(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	smp_wmb();
-	tmp = atomic_dec_return((atomic_t *)(&sem->count));
-	if (tmp < -1 && (tmp & RWSEM_ACTIVE_MASK) == 0)
-		rwsem_wake(sem);
-}
-
-/*
- * unlock after writing
- */
-static inline void __up_write(struct rw_semaphore *sem)
-{
-	smp_wmb();
-	if (atomic_sub_return(RWSEM_ACTIVE_WRITE_BIAS,
-			      (atomic_t *)(&sem->count)) < 0)
-		rwsem_wake(sem);
-}
-
-/*
- * implement atomic add functionality
- */
-static inline void rwsem_atomic_add(int delta, struct rw_semaphore *sem)
-{
-	atomic_add(delta, (atomic_t *)(&sem->count));
-}
-
-/*
- * downgrade write lock to read lock
- */
-static inline void __downgrade_write(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	smp_wmb();
-	tmp = atomic_add_return(-RWSEM_WAITING_BIAS, (atomic_t *)(&sem->count));
-	if (tmp < 0)
-		rwsem_downgrade_wake(sem);
-}
-
-/*
- * implement exchange and add functionality
- */
-static inline int rwsem_atomic_update(int delta, struct rw_semaphore *sem)
-{
-	smp_mb();
-	return atomic_add_return(delta, (atomic_t *)(&sem->count));
-}
-
-#endif /* __KERNEL__ */
-#endif /* _PPC_RWSEM_XADD_H */
diff --git a/include/asm-ppc64/rwsem.h b/include/asm-ppc64/rwsem.h
deleted file mode 100644
--- a/include/asm-ppc64/rwsem.h
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * include/asm-ppc64/rwsem.h: R/W semaphores for PPC using the stuff
- * in lib/rwsem.c.  Adapted largely from include/asm-i386/rwsem.h
- * by Paul Mackerras <paulus@samba.org>.
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; either version
- * 2 of the License, or (at your option) any later version.
- */
-
-#ifndef _PPC64_RWSEM_H
-#define _PPC64_RWSEM_H
-
-#ifdef __KERNEL__
-#include <linux/list.h>
-#include <linux/spinlock.h>
-#include <asm/atomic.h>
-#include <asm/system.h>
-
-/*
- * the semaphore definition
- */
-struct rw_semaphore {
-	/* XXX this should be able to be an atomic_t  -- paulus */
-	signed int		count;
-#define RWSEM_UNLOCKED_VALUE		0x00000000
-#define RWSEM_ACTIVE_BIAS		0x00000001
-#define RWSEM_ACTIVE_MASK		0x0000ffff
-#define RWSEM_WAITING_BIAS		(-0x00010000)
-#define RWSEM_ACTIVE_READ_BIAS		RWSEM_ACTIVE_BIAS
-#define RWSEM_ACTIVE_WRITE_BIAS		(RWSEM_WAITING_BIAS + RWSEM_ACTIVE_BIAS)
-	spinlock_t		wait_lock;
-	struct list_head	wait_list;
-#if RWSEM_DEBUG
-	int			debug;
-#endif
-};
-
-/*
- * initialisation
- */
-#if RWSEM_DEBUG
-#define __RWSEM_DEBUG_INIT      , 0
-#else
-#define __RWSEM_DEBUG_INIT	/* */
-#endif
-
-#define __RWSEM_INITIALIZER(name) \
-	{ RWSEM_UNLOCKED_VALUE, SPIN_LOCK_UNLOCKED, \
-	  LIST_HEAD_INIT((name).wait_list) \
-	  __RWSEM_DEBUG_INIT }
-
-#define DECLARE_RWSEM(name)		\
-	struct rw_semaphore name = __RWSEM_INITIALIZER(name)
-
-extern struct rw_semaphore *rwsem_down_read_failed(struct rw_semaphore *sem);
-extern struct rw_semaphore *rwsem_down_write_failed(struct rw_semaphore *sem);
-extern struct rw_semaphore *rwsem_wake(struct rw_semaphore *sem);
-extern struct rw_semaphore *rwsem_downgrade_wake(struct rw_semaphore *sem);
-
-static inline void init_rwsem(struct rw_semaphore *sem)
-{
-	sem->count = RWSEM_UNLOCKED_VALUE;
-	spin_lock_init(&sem->wait_lock);
-	INIT_LIST_HEAD(&sem->wait_list);
-#if RWSEM_DEBUG
-	sem->debug = 0;
-#endif
-}
-
-/*
- * lock for reading
- */
-static inline void __down_read(struct rw_semaphore *sem)
-{
-	if (unlikely(atomic_inc_return((atomic_t *)(&sem->count)) <= 0))
-		rwsem_down_read_failed(sem);
-}
-
-static inline int __down_read_trylock(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	while ((tmp = sem->count) >= 0) {
-		if (tmp == cmpxchg(&sem->count, tmp,
-				   tmp + RWSEM_ACTIVE_READ_BIAS)) {
-			return 1;
-		}
-	}
-	return 0;
-}
-
-/*
- * lock for writing
- */
-static inline void __down_write(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	tmp = atomic_add_return(RWSEM_ACTIVE_WRITE_BIAS,
-				(atomic_t *)(&sem->count));
-	if (unlikely(tmp != RWSEM_ACTIVE_WRITE_BIAS))
-		rwsem_down_write_failed(sem);
-}
-
-static inline int __down_write_trylock(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	tmp = cmpxchg(&sem->count, RWSEM_UNLOCKED_VALUE,
-		      RWSEM_ACTIVE_WRITE_BIAS);
-	return tmp == RWSEM_UNLOCKED_VALUE;
-}
-
-/*
- * unlock after reading
- */
-static inline void __up_read(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	tmp = atomic_dec_return((atomic_t *)(&sem->count));
-	if (unlikely(tmp < -1 && (tmp & RWSEM_ACTIVE_MASK) == 0))
-		rwsem_wake(sem);
-}
-
-/*
- * unlock after writing
- */
-static inline void __up_write(struct rw_semaphore *sem)
-{
-	if (unlikely(atomic_sub_return(RWSEM_ACTIVE_WRITE_BIAS,
-			      (atomic_t *)(&sem->count)) < 0))
-		rwsem_wake(sem);
-}
-
-/*
- * implement atomic add functionality
- */
-static inline void rwsem_atomic_add(int delta, struct rw_semaphore *sem)
-{
-	atomic_add(delta, (atomic_t *)(&sem->count));
-}
-
-/*
- * downgrade write lock to read lock
- */
-static inline void __downgrade_write(struct rw_semaphore *sem)
-{
-	int tmp;
-
-	tmp = atomic_add_return(-RWSEM_WAITING_BIAS, (atomic_t *)(&sem->count));
-	if (tmp < 0)
-		rwsem_downgrade_wake(sem);
-}
-
-/*
- * implement exchange and add functionality
- */
-static inline int rwsem_atomic_update(int delta, struct rw_semaphore *sem)
-{
-	return atomic_add_return(delta, (atomic_t *)(&sem->count));
-}
-
-#endif /* __KERNEL__ */
-#endif /* _PPC_RWSEM_XADD_H */

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[David Howells]

Jon Loeliger <jdl@freescale.com> wrote:

> Merge asm-ppc*/rwsem.h into include/asm-powerpc.
> Removed smp_*mb() memory barriers from the ppc32 code
> as they are now burried in the atomic_*() functions as
> suggested by Paul, implemented by Arnd, and pushed out
> by Becky.  I am not the droid you are looking for.

rwsems on ppc64 should really be using a 64-bit counter, not a 32-bit counter,
otherwise you limit the maximum number of processes to 32K-ish.

The counter should be "signed long" really.

David

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[Anton Blanchard]

> rwsems on ppc64 should really be using a 64-bit counter, not a 32-bit counter,
> otherwise you limit the maximum number of processes to 32K-ish.
> 
> The counter should be "signed long" really.

Agreed, we should move to a 64bit count.

Anton

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[David Howells]

Anton Blanchard <anton@samba.org> wrote:

> > The counter should be "signed long" really.
> 
> Agreed, we should move to a 64bit count.

With a 64-bit counter, the constants should be:

#define RWSEM_UNLOCKED_VALUE		  0x0000000000000000L
#define RWSEM_ACTIVE_BIAS		  0x0000000000000001L
#define RWSEM_ACTIVE_MASK		  0x00000000ffffffffL
#define RWSEM_WAITING_BIAS		(-0x0000000100000000L)
#define RWSEM_ACTIVE_READ_BIAS		RWSEM_ACTIVE_BIAS
#define RWSEM_ACTIVE_WRITE_BIAS		(RWSEM_WAITING_BIAS + RWSEM_ACTIVE_BIAS)

Just like in the asm-s390/rwsem.h.

David

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[Jon Loeliger]

So, like, the other day David Howells mumbled:
> 
> rwsems on ppc64 should really be using a 64-bit counter, not a 32-bit counter
> ,
> otherwise you limit the maximum number of processes to 32K-ish.
> 
> The counter should be "signed long" really.

No problem.  I _can_ resubmit the patch with this fix.
However, I am not certain that I should yet.

But what do you wan to do with ppc32 land then?
Leaving it a "signed long" will limit ppc32 land but
not ppc64 folks.  (No problem.)  Or we can have it
be "singed long long" for ppc32 to also get 64 bits.
(Again, no problem.)

Also, this begs the question of the comment from Paul:

    struct rw_semaphore {
	/* XXX this should be able to be an atomic_t  -- paulus */
	signed int		count;

Which, of course is:

    typedef struct { volatile int counter; } atomic_t;

Changing the size of counter will cause bad sizes
due to the actual treatment of count as an atomic_t.

So.  Should we, in fact, convert it to be an atomic_t
and further dispense with the casts?:

    static inline void __down_write(struct rw_semaphore *sem)
    {
	int tmp;

	tmp = atomic_add_return(RWSEM_ACTIVE_WRITE_BIAS,
				(atomic_t *)(&sem->count));

And if we _do_ convert it to be an atomic_t, should _that_
be where the real type for count gets established?

And finally, I've been working on merging header files
under the vague guideline of "merge maintaining existing
functionality/breakage".  I've been trying NOT to introduce
simultaneous "improvements" at the risk of breaking something.

To that end, I ask if the request to make 'count' be 64-bits
should be submitted as a follow on patch that stands on its
own and can clean up around it as necessary?  Or do you want
it mixed in with this "merge" patch too?

Just need to know the preferences.

Thanks,
jdl

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[David Howells]

Jon Loeliger <jdl@freescale.com> wrote:

> No problem.  I _can_ resubmit the patch with this fix.
> However, I am not certain that I should yet.

I'd suggest that you wait until the merge is complete, I think.

> But what do you wan to do with ppc32 land then?
> Leaving it a "signed long" will limit ppc32 land but
> not ppc64 folks.  (No problem.)

I'd suggest "signed long" in both cases. A maximum of 32K processes on ppc32
is probably reasonable.

> Also, this begs the question of the comment from Paul:
> 
>     struct rw_semaphore {
> 	/* XXX this should be able to be an atomic_t  -- paulus */
> 	signed int		count;

Paul can be wrong sometimes:-)

Changing to atomic_t would leave the 32K process limit in place.

> Changing the size of counter will cause bad sizes
> due to the actual treatment of count as an atomic_t.

You will not be able to use the standard atomic ops unless you increase them
to 64-bits on ppc64.

> And if we _do_ convert it to be an atomic_t, should _that_
> be where the real type for count gets established?

You should not do that unless you increase atomic_t to 64-bits on ppc64.

> And finally, I've been working on merging header files
> under the vague guideline of "merge maintaining existing
> functionality/breakage".  I've been trying NOT to introduce
> simultaneous "improvements" at the risk of breaking something.

Sounds reasonable.

> To that end, I ask if the request to make 'count' be 64-bits
> should be submitted as a follow on patch that stands on its
> own and can clean up around it as necessary?  Or do you want
> it mixed in with this "merge" patch too?

Follow-on is fine.

David

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[Paul Mackerras]

David Howells writes:

> rwsems on ppc64 should really be using a 64-bit counter, not a 32-bit counter,
> otherwise you limit the maximum number of processes to 32K-ish.
> 
> The counter should be "signed long" really.

It has long annoyed me that we waste half the bits in the rwsem
counter, just because you assume a lowest-common-denominator set of
atomic ops.  IIRC your implementation replaced the earlier ppc
implementation which had a 32-bit counter and didn't have the 32k
process limit.

I'd have to study it in more detail, but I strongly suspect that with
an atomic operation that did something like

	*p = max(*p, limit) + inc

atomically (or alternatively a min), we could increase the limit to at
least 1G processes with a 32-bit counter, without needing to change
your common slow-path implementation.

Such an atomic op is easy to implement with load-and-reserve /
store-conditional instructions.  Look at __sem_update_count in
arch/ppc64/kernel/semaphore.c for an example.

Paul.

Re: PATCH powerpc Merge asm-ppc*/rwsem.h

[David Howells]

Paul Mackerras <paulus@samba.org> wrote:

>
> > rwsems on ppc64 should really be using a 64-bit counter, not a 32-bit
> > counter, otherwise you limit the maximum number of processes to 32K-ish.
> >
> > The counter should be "signed long" really.
>
> It has long annoyed me that we waste half the bits in the rwsem
> counter, just because you assume a lowest-common-denominator set of
> atomic ops.  IIRC your implementation replaced the earlier ppc
> implementation which had a 32-bit counter and didn't have the 32k
> process limit.

And also didn't work, at least not on i386.

> I'd have to study it in more detail, but I strongly suspect that with
> an atomic operation that did something like
>
> 	*p = max(*p, limit) + inc

There is no requirement for an arch to use the XADD-based algorithm for doing
this - that's optimised for use with the i386/x86_64 XADD instruction. That
can be emulated by cmpxchg or load-locked/store-conditional type constructs
like MIPS, Alpha and ppc have.

> atomically (or alternatively a min), we could increase the limit to at
> least 1G processes with a 32-bit counter, without needing to change
> your common slow-path implementation.

I think you'd probably need a different slow-path for what you want. My XADD
optimised slow patch assumes that certain values represent certain states, and
if you're using different values, then it won't necessarily work. This is why
the thing is contingent on CONFIG_RWSEM_XCHGADD_ALGORITHM.

> Such an atomic op is easy to implement with load-and-reserve /
> store-conditional instructions.  Look at __sem_update_count in
> arch/ppc64/kernel/semaphore.c for an example.

So that does (I think):

	do {
		int old_count = load_reserve(&sem->count);

		int tmp = old_count >> 31;   // not sure about this: 31 or 1?

		tmp = old_count & ~tmp;

		tmp = tmp + incr;

	} while (!conditionally_assign(&sem->count, tmp));

PPC asm makes my head hurt; in particular the description of the SRAWI
instruction I've got is a delight to behold, and is inconsistent with the
description of the SRAW instruction:-/

I'm guessing it does a shift by 31 to leave tmp filled with copies of bit 0
(the sign bit) of old_count. The manual suggests the shift is 32-N (or 1 in
this case) which would seem odd.

I see how it works, then. What's "limit" in this formula:

> 	*p = max(*p, limit) + inc

Is it the maximum number of processes that may hold readlocks? Or maybe the
negated quantity thereof.

It's hard to see immediately how this will work. The counter has to represent
several things:

 (1) The number of active readers (max N) or the number of active writers (max
     1).

 (2) Whether or not there is potential contention.

I do this with my XADD algorithm by splitting the word into two halves:

 MSH = #active_writers + #failed_writers + #sleepers

 LSH = #active_readers + #active_writers + #failed_contenders

Basically, the counter is a lock additional to the spinlock.

Perhaps you could do something like this:

	COUNT			MEANING
	=======================	==========================================
	0			Not in use
	0x00000001 - 0x7fffffff	N readers
	0x80000000		last active reader or writer done, contention
	0x80000001 - 0xffffffff	N active readers or writers, contention

So up_read() and up_write() would decrement the count. If the result reaches
0x80000000 then you have to perform contention resolution.

down_read() would increment the count if >= 0, or attempt to sleep otherwise.

down_write() would just set the sign bit of the count and sleep if != 0, or set
the count to 0x80000001 and return.

We could also divide the space in two, and use the second most significant bit
(0x40000000) to record the fact that there are sleeping contenders or to record
which type of lock is actually active.

However, I think there's a problem with doing that, and that is:

	COUNT	 PROCESS 0		PROCESS 1
	======== ======================	================================
	00000000
		 down_read()
	00000001
					-->down_write()
					lwarx
					stwcx [set sign bit]
	80000001
					-->down_write_slow()
		 -->up_read()
		 lwarx
		 stwcx [dec]
	80000000
		 -->up_read_slow()
		 -->lock sem->wait_lock
		 <--lock sem->wait_lock
					-->lock sem->wait_lock
		 count indicated contention
		  but nothing on sem->wait_list
		  - do we leave the count unaltered?
		  - do we clear the count?
		  - do we set the count to 0x80000001?

If we leave the count set to 0x80000000, then the down_write_slow() when it
runs won't be able to tell whether it got sem->wait_lock before or after
down_read_slow() did. In one case it should go to sleep and in the other it
should claim the lock.

If we set the count to 0x80000001, then the case where _two_ CPUs are
contending with the releaser may _both_ think they have the lock, and in any
case can't tell the difference between that and a sleeping process just having
been woken up and given a writelock.

If we clear the count, then other processes may be able to queue jump (which
isn't necessarily a problem since spinlocks aren't fair anyway, though they
could be made so). down_write_slow() would have to attempt to take the lock
again, though this time it would already have the spinlock. But what might
happen is that:

	COUNT	 PROCESS 0		PROCESS 1	PROCESS 2
	======== ======================	===============	===============
	80000000
					-->lock sem->wait_lock
		 count indicated contention
		  but nothing on sem->wait_list
		  - do we clear the count?
	00000000
							down_write()
	80000001
		 unlock sem->wait_lock
		 <--up_read_slow()
		 <--up_read()
					<--lock sem->wait_lock
					lwarx
					stwcx [set sign bit]
	80000001
							-->up_write()
							lwarx
							stwcx [dec]
	80000000
							-->up_write_slow()
							-->lock sem->wait_lock
					list_add_tail()
		 			unlock sem->wait_lock
					sleep
							<--lock sem->wait_lock
							dequeue process 1
							set count
	80000001
							unlock sem->wait_lock
							<--up_write_slow()
							<--up_write()
					wake
					<--up_write_slow()
					<--up_write()

Ummm... that does seem to work. I think their may be a multi-process race in
there somewhere, but I can't work it out if there is; just as long as
up_write() always does contention resolution.

The down side of this is that you always have to get the spinlock in up_write,
unlike in the XADD implementation where the counter includes a contention
counter.

I think downgrade_write() should simply be a matter of: whilst holding the
spinlock, set the counter to the number of outstanding reads, and only set the
sign bit if there's a sleeping writer left at the front of the queue.

Should interruptible rwsems arrive, then I think just dequeuing the aborted
down op should be enough. Leave the contention counter with the sign bit set as
it'll sort itself out later.

David