PATCH powerpc Merge asm-ppc*/rwsem.h

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.

---

 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(-)

new file mode 100644
@@ -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 <[email protected]>.
+ */
+
+#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 */

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

Agreed, we should move to a 64bit count.

Anton

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

So, like, the other day David Howells mumbled:

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

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

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

Paul can be wrong sometimes:-)

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

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

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

Sounds reasonable.

Follow-on is fine.

David

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.

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

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.

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.

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:

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