convert mmap_sem to a scalable rw_mutex

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

---
 include/linux/rwmutex.h |  103 +++++++++++++++++++++++++++++++++++++
 kernel/Makefile         |    3 -
 kernel/rwmutex.c        |  132 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 237 insertions(+), 1 deletion(-)

Index: linux-2.6/include/linux/rwmutex.h
===================================================================

include/linux/rwmutex.h new file +22 -0

@@ -0,0 +1,103 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * This file contains the public data structure and API definitions.
+ */
+#ifndef _LINUX_RWMUTEX_H
+#define _LINUX_RWMUTEX_H
+
+#include <linux/preempt.h>
+#include <linux/wait.h>
+#include <linux/percpu_counter.h>
+#include <linux/lockdep.h>
+#include <linux/mutex.h>
+#include <asm/atomic.h>
+
+struct rw_mutex {
+	/* Read mostly global */
+	struct percpu_counter	readers;
+	unsigned int		status;
+

include/linux/rwmutex.h new file +10 -0

+#define rw_mutex_init(rw_mutex)					\
+	do {							\
+		static struct lock_class_key __key;		\
+		__rw_mutex_init((rw_mutex), #rw_mutex, &__key);	\
+	} while (0)
+
+extern void __rw_mutex_read_lock(struct rw_mutex *rw_mutex);
+
+extern void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass);
+extern void rw_mutex_write_unlock(struct rw_mutex *rw_mutex);

include/linux/rwmutex.h new file +11 -0

+static inline int rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	int ret = __rw_mutex_read_trylock(rw_mutex);
+	if (ret)
+		rwsem_acquire_read(&rw_mutex->dep_map, 0, 1, _RET_IP_);
+	return ret;
+}
+
+static inline void rw_mutex_read_lock(struct rw_mutex *rw_mutex)
+{
+	int ret;

include/linux/rwmutex.h new file +22 -0

+
+void rw_mutex_destroy(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_destroy(&rw_mutex->readers);
+	mutex_destroy(&rw_mutex->read_mutex);
+	mutex_destroy(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL_GPL(rw_mutex_destroy);
+
+void __rw_mutex_read_lock(struct rw_mutex *rw_mutex)
+{
+	/*
+	 * read lock slow path;
+	 * count the number of readers waiting on the read_mutex
+	 */
+	atomic_inc(&rw_mutex->read_waiters);
+	mutex_lock(&rw_mutex->read_mutex);
+	/*
+	 * rw_mutex->state is only set while the read_mutex is held
+	 * so by serialising on this lock, we're sure its free.
+	 */
+	BUG_ON(rw_mutex->status);

yipes. percpu_counter_sum() is expensive.

Quoted 1 line · Andrew Morton

yipes. percpu_counter_sum() is expensive.

Capable of triggering NMI watchdog on 4096+ processors?

Quoted 4 lines · On Fri, 11 May 2007 10:07:17 -0700 (PDT) Christoph Lameter <[email protected]>…

Quoted 1 line · Andrew Morton

yipes. percpu_counter_sum() is expensive.

Quoted 1 line

Capable of triggering NMI watchdog on 4096+ processors?

Well. That would be a millisecond per cpu which sounds improbable. And we'd need to be calling it under local_irq_save() which we presently don't. And nobody has reported any problems against the existing callsites.

But it's no speed demon, that's for sure.

Quoted 13 lines · Andrew Morton

Quoted 5 lines · On Fri, 11 May 2007 10:07:17 -0700 (PDT) Christoph Lameter <[email protected]>…

Quoted 1 line · Andrew Morton

yipes. percpu_counter_sum() is expensive.

Capable of triggering NMI watchdog on 4096+ processors?

Well. That would be a millisecond per cpu which sounds improbable. And we'd need to be calling it under local_irq_save() which we presently don't. And nobody has reported any problems against the existing callsites.

Quoted 1 line

But it's no speed demon, that's for sure.

There is one possible optimization for this I did some time ago. You don't really need to sum all over the possible map, but only all CPUs that were ever online. But this only helps on systems where the possible map is bigger than online map in the common case. But that shouldn't be the case anymore on x86 -- it just used to be. If it's true on some other architectures it might be still worth it.

We could put a cpumask in percpu_counter, initialise it to cpu_possible_map. Then, those callsites which have hotplug notifiers can call into new percpu_counter functions which clear and set bits in that cpumask and which drain percpu_counter.counts[cpu] into percpu_counter.count.

And percpu_counter_sum() gets taught to do for_each_cpu_mask(fbc->cpumask).

Quoted 15 lines · On 12 May 2007 20:55:28 +0200 Andi Kleen <[email protected]> wrote:

Quoted 13 lines · Andrew Morton

Quoted 5 lines · On Fri, 11 May 2007 10:07:17 -0700 (PDT) Christoph Lameter <[email protected]>…

Quoted 1 line · Andrew Morton

yipes. percpu_counter_sum() is expensive.

Capable of triggering NMI watchdog on 4096+ processors?

Well. That would be a millisecond per cpu which sounds improbable. And we'd need to be calling it under local_irq_save() which we presently don't. And nobody has reported any problems against the existing callsites.

Quoted 9 lines

Quoted 1 line

But it's no speed demon, that's for sure.

There is one possible optimization for this I did some time ago. You don't really need to sum all over the possible map, but only all CPUs that were ever online. But this only helps on systems where the possible map is bigger than online map in the common case. But that shouldn't be the case anymore on x86 -- it just used to be. If it's true on some other architectures it might be still worth it.

hm, yeah.

We could put a cpumask in percpu_counter, initialise it to cpu_possible_map. Then, those callsites which have hotplug notifiers can call into new percpu_counter functions which clear and set bits in that cpumask and which drain percpu_counter.counts[cpu] into percpu_counter.count.

And percpu_counter_sum() gets taught to do for_each_cpu_mask(fbc->cpumask).

(I hope. Might have race windows in which the percpu_counter_sum() count is inaccurate?)

Quoted 2 lines · Andrew Morton

(I hope. Might have race windows in which the percpu_counter_sum() count is inaccurate?)

The question is how do these race windows affect the locking scheme?

kernel/cpu.c +4 -0

@@ -26,6 +26,10 @@ static __cpuinitdata RAW_NOTIFIER_HEAD(c
  */
 static int cpu_hotplug_disabled;
 
+/* Contains any CPUs that were ever online at some point.
+   No guarantee they were fully initialized though */
+cpumask_t cpu_everonline_map;
+
 #ifdef CONFIG_HOTPLUG_CPU
 
 /* Crappy recursive lock-takers in cpufreq! Complain loudly about idiots */

This also allows us to de-uglify workqueue.c a little bit, it uses a home-grown cpu_populated_map.

Quoted 3 lines

This also allows us to de-uglify workqueue.c a little bit, it uses a home-grown cpu_populated_map.

It might be obsolete iff more and more architecture don't use NR_CPUS filled possible_map anymore (haven't checked them all to know if it's true or not)

If not there are a couple of more optimizations that can be done, e.g. in networking by converting more code to hotplug notifier.

Quoted 6 lines · On Fri, 11 May 2007 15:15:42 +0200 Peter Zijlstra <[email protected]>…

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

It looks .... surprisingly sane, given the history of these things ;)

Quoted 40 lines

---
 include/linux/rwmutex.h |  103 +++++++++++++++++++++++++++++++++++++
 kernel/Makefile         |    3 -
 kernel/rwmutex.c        |  132 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 237 insertions(+), 1 deletion(-)

Index: linux-2.6/include/linux/rwmutex.h
===================================================================

include/linux/rwmutex.h new file +29 -0

@@ -0,0 +1,103 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * This file contains the public data structure and API definitions.
+ */
+#ifndef _LINUX_RWMUTEX_H
+#define _LINUX_RWMUTEX_H
+
+#include <linux/preempt.h>
+#include <linux/wait.h>
+#include <linux/percpu_counter.h>
+#include <linux/lockdep.h>
+#include <linux/mutex.h>
+#include <asm/atomic.h>
+
+struct rw_mutex {
+	/* Read mostly global */
+	struct percpu_counter	readers;
+	unsigned int		status;
+
+	/* The following variables are only for the slowpath */
+	struct mutex		read_mutex;	/* r -> w waiting */
+	struct mutex		write_mutex;	/* w -> w waiting */
+	wait_queue_head_t	wait_queue;	/* w -> r waiting */
+	atomic_t		read_waiters;
+
+#ifdef CONFIG_DEBUG_LOCK_ALLOC

include/linux/rwmutex.h line 32

Quoted 4 lines

include/linux/rwmutex.h new file +3 -0

+	struct lockdep_map dep_map;
+#endif
+};

A nice comment describing the overall design and the runtime dynamics and the lock's characteristics would be useful. It should include a prominent description of the lock's storage requirements, which are considerable.

include/linux/rwmutex.h line 36

Quoted 3 lines

include/linux/rwmutex.h new file +3 -0

+extern void __rw_mutex_init(struct rw_mutex *rw_mutex, const char * name,
+		struct lock_class_key *key);
+extern void rw_mutex_destroy(struct rw_mutex *rw_mutex);

Sometimes you use `extern'.

Quoted 14 lines

include/linux/rwmutex.h new file +14 -0

+#define rw_mutex_init(rw_mutex)					\
+	do {							\
+		static struct lock_class_key __key;		\
+		__rw_mutex_init((rw_mutex), #rw_mutex, &__key);	\
+	} while (0)
+
+extern void __rw_mutex_read_lock(struct rw_mutex *rw_mutex);
+
+extern void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass);
+extern void rw_mutex_write_unlock(struct rw_mutex *rw_mutex);
+
+static inline unsigned int __rw_mutex_reader_slow(struct rw_mutex *rw_mutex)
+{
+	unsigned int ret;

include/linux/rwmutex.h line 57

Quoted 6 lines

include/linux/rwmutex.h new file +6 -0

+
+	smp_rmb();
+	ret = rw_mutex->status;
+
+	return ret;
+}

An undocumented barrier!

Quoted 5 lines

include/linux/rwmutex.h new file +5 -0

+static inline int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	preempt_disable();
+	if (likely(!__rw_mutex_reader_slow(rw_mutex))) {
+		percpu_counter_mod(&rw_mutex->readers, 1);

include/linux/rwmutex.h line 69

Quoted 6 lines

include/linux/rwmutex.h new file +6 -0

+		preempt_enable();
+		return 1;
+	}
+	preempt_enable();
+	return 0;
+}

What does the preempt_disable() do?

include/linux/rwmutex.h line 67

Quoted 1 line

include/linux/rwmutex.h new file +1 -0

+EXPORT_SYMBOL_GPL(__rw_mutex_init);

down_foo(mmap_sem) was previously accessible to non-gpl modules, so the GPL export might be a problem.

Quoted 25 lines

include/linux/rwmutex.h new file +25 -0

+void rw_mutex_destroy(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_destroy(&rw_mutex->readers);
+	mutex_destroy(&rw_mutex->read_mutex);
+	mutex_destroy(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL_GPL(rw_mutex_destroy);
+
+void __rw_mutex_read_lock(struct rw_mutex *rw_mutex)
+{
+	/*
+	 * read lock slow path;
+	 * count the number of readers waiting on the read_mutex
+	 */
+	atomic_inc(&rw_mutex->read_waiters);
+	mutex_lock(&rw_mutex->read_mutex);
+	/*
+	 * rw_mutex->state is only set while the read_mutex is held
+	 * so by serialising on this lock, we're sure its free.
+	 */
+	BUG_ON(rw_mutex->status);
+	/*
+	 * take the read reference, and drop the read_waiters count
+	 * and nudge all those waiting on the read_waiters count.
+	 */

include/linux/rwmutex.h line 98

Quoted 6 lines

include/linux/rwmutex.h new file +6 -0

+	percpu_counter_mod(&rw_mutex->readers, 1);
+	atomic_dec(&rw_mutex->read_waiters);
+	wake_up_all(&rw_mutex->wait_queue);
+	mutex_unlock(&rw_mutex->read_mutex);
+}
+EXPORT_SYMBOL_GPL(__rw_mutex_read_lock);

hm, I'm surprised that any foo_lock() would ever wake anyone up.

include/linux/rwmutex.h line 103

Quoted 5 lines

include/linux/rwmutex.h new file +5 -0

+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	percpu_counter_mod(&rw_mutex->readers, -1);

percpu_counter_dec()?

include/linux/rwmutex.h

Quoted 5 lines

include/linux/rwmutex.h new file +5 -0

+	if (unlikely(__rw_mutex_reader_slow(rw_mutex)) &&
+			percpu_counter_sum(&rw_mutex->readers) == 0)
+		wake_up_all(&rw_mutex->wait_queue);
+}
+EXPORT_SYMBOL_GPL(rw_mutex_read_unlock);

yipes. percpu_counter_sum() is expensive.

kernel/rwmutex.c new file +10 -0

+		percpu_counter_mod(&rw_mutex->readers, 1);
+		preempt_enable();
+		return 1;
+	}
+	preempt_enable();
+	return 0;
+}
 
[...snip...]
 
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
[...snip...]
+

kernel/rwmutex.c new file +33 -0

+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	might_sleep();
+	rwsem_acquire(&rw_mutex->dep_map, subclass, 0, _RET_IP_);
+
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+
+	/*
+	 * block new readers
+	 */
+	__rw_mutex_status_set(rw_mutex, RW_MUTEX_READER_SLOW);
+	/*
+	 * wait for all readers to go away
+	 */
+	wait_event(rw_mutex->wait_queue,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}
+
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex)
+{
+	int waiters;
+
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	/*
+	 * let the readers rip
+	 */
+	__rw_mutex_status_set(rw_mutex, RW_MUTEX_READER_FAST);
+	waiters = atomic_read(&rw_mutex->read_waiters);
+	mutex_unlock(&rw_mutex->read_mutex);
+	/*
+	 * wait for at least 1 reader to get through

include/linux/rwmutex.h line 62

Quoted 5 lines · Peter Zijlstra

include/linux/rwmutex.h new file +4 -0

+static inline int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	preempt_disable();
+	if (likely(!__rw_mutex_reader_slow(rw_mutex))) {

	--- WINDOW ---

include/linux/rwmutex.h line 69

Quoted 7 lines

include/linux/rwmutex.h new file +7 -0

+		percpu_counter_mod(&rw_mutex->readers, 1);
+		preempt_enable();
+		return 1;
+	}
+	preempt_enable();
+	return 0;
+}

Quoted 11 lines

[...snip...]

kernel/rwmutex.c new file +8 -0

+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
[...snip...]
+
+	/*
+	 * block new readers
+	 */
+	__rw_mutex_status_set(rw_mutex, RW_MUTEX_READER_SLOW);
+	/*

kernel/rwmutex.c line 105

Quoted 5 lines

kernel/rwmutex.c new file +5 -0

+	 * wait for all readers to go away
+	 */
+	wait_event(rw_mutex->wait_queue,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}

This look a bit suspicious, can't mutex_write_lock() set RW_MUTEX_READER_SLOW and find percpu_counter_sum() == 0 in that WINDOW above?

kernel/rwmutex.c line 73

Quoted 7 lines

kernel/rwmutex.c new file +7 -0

+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+     rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+     percpu_counter_mod(&rw_mutex->readers, -1);
+     if (unlikely(__rw_mutex_reader_slow(rw_mutex)) &&
+                     percpu_counter_sum(&rw_mutex->readers) == 0)

Review

Quoted 2 lines

(quoted diff) +2 -0

+             wake_up_all(&rw_mutex->wait_queue);
+}

The same. __rw_mutex_status_set()->wmb() in rw_mutex_write_lock below is not enough. percpu_counter_mod() doesn't take fbc->lock if < FBC_BATCH, so we don't have a proper serialization.

write_lock() sets RW_MUTEX_READER_SLOW, finds percpu_counter_sum() != 0, and sleeps. rw_mutex_read_unlock() decrements cpu-local var, does not see RW_MUTEX_READER_SLOW and skips wake_up_all().

Quoted 39 lines

kernel/rwmutex.c new file +39 -0

+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	might_sleep();
+	rwsem_acquire(&rw_mutex->dep_map, subclass, 0, _RET_IP_);
+
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+
+	/*
+	 * block new readers
+	 */
+	__rw_mutex_status_set(rw_mutex, RW_MUTEX_READER_SLOW);
+	/*
+	 * wait for all readers to go away
+	 */
+	wait_event(rw_mutex->wait_queue,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}
+
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex)
+{
+	int waiters;
+
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	/*
+	 * let the readers rip
+	 */
+	__rw_mutex_status_set(rw_mutex, RW_MUTEX_READER_FAST);
+	waiters = atomic_read(&rw_mutex->read_waiters);
+	mutex_unlock(&rw_mutex->read_mutex);
+	/*
+	 * wait for at least 1 reader to get through
+	 */
+	if (waiters) {
+		wait_event(rw_mutex->wait_queue,
+			(atomic_read(&rw_mutex->read_waiters) < waiters));
+	}
+	/*

kernel/rwmutex.c line 131

Quoted 4 lines

kernel/rwmutex.c new file +4 -0

+	 * before we let the writers rip
+	 */
+	mutex_unlock(&rw_mutex->write_mutex);
+}

Looks like we can have only one task on rw_mutex->wait_queue, and it holds ->write_mutex. Can't we use just a "task_struct *write_waiter" instead of ->wait_queue ? This makes rw_mutex smaller.

kernel/rwmutex.c new file +19 -0

+static inline void rw_mutex_readers_dec(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_dec(&rw_mutex->readers);
+	smp_wmb();
+}
 
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	rw_mutex_readers_dec(rw_mutex);
+	/*
+	 * on the slow path;
+	 * nudge the writer waiting for the last reader to go away
+	 */
+	if (unlikely(rw_mutex_reader_slow(rw_mutex)))
+		rw_mutex_writer_wake(rw_mutex);
+}
 
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+

Quoted 22 lines · Peter Zijlstra

kernel/rwmutex.c new file +19 -0

+static inline void rw_mutex_readers_dec(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_dec(&rw_mutex->readers);
+	smp_wmb();
+}
 
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	rw_mutex_readers_dec(rw_mutex);
+	/*
+	 * on the slow path;
+	 * nudge the writer waiting for the last reader to go away
+	 */
+	if (unlikely(rw_mutex_reader_slow(rw_mutex)))
+		rw_mutex_writer_wake(rw_mutex);
+}
 
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+

kernel/rwmutex.c line 226

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	/*
+	 * block new readers
+	 */
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+	rw_mutex_status_set(rw_mutex, RW_MUTEX_READER_SLOW);
+	/*
+	 * and wait for all current readers to go away
+	 */
+	rw_mutex_writer_wait(rw_mutex, (rw_mutex_readers(rw_mutex) == 0));
+}

I think this is still not right, but when it comes to barriers we need a true expert (Paul cc-ed).

this code roughly does (the only reader does unlock)

	READER			WRITER

	readers = 0;		state = 1;
	wmb();			wmb();
	CHECK(state != 0)	CHECK(readers == 0)

We need to ensure that we can't miss both CHECKs. Either reader should see RW_MUTEX_READER_SLOW, o writer sees "readers == 0" and does not sleep.

In that case both barriers should be converted to smp_mb(). There was a _long_ discussion about STORE-MB-LOAD behaviour, and experts seem to believe everething is ok.

Another question. Isn't it possible to kill rw_mutex->status ?

I have a vague feeling you can change the code so that

	rw_mutex_reader_slow() <=> "->waiter != NULL"

, but I am not sure.

Quoted 9 lines · Oleg Nesterov

this code roughly does (the only reader does unlock)

	READER			WRITER

	readers = 0;		state = 1;
	wmb();			wmb();
	CHECK(state != 0)	CHECK(readers == 0)

Quoted 7 lines

We need to ensure that we can't miss both CHECKs. Either reader should see RW_MUTEX_READER_SLOW, o writer sees "readers == 0" and does not sleep.

In that case both barriers should be converted to smp_mb(). There was a _long_ discussion about STORE-MB-LOAD behaviour, and experts seem to believe everething is ok.

Ah, but note that both those CHECK()s have a rmb(), so that ends up being:

	READER				WRITER

	readers = 0;			state = 1;
	wmb();				wmb();

	rmb();				rmb();		
	if (state != 0)			if (readers == 0)

and a wmb+rmb is a full mb, right?

Changes include:

- wmb+rmb != mb
- ->state folded into ->waiter

---
Subject: scalable rw_mutex

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

kernel/rwmutex.c new file +49 -0

+#define rw_mutex_writer_wait(rw_mutex, condition)		\
+do {								\
+	struct task_struct *tsk = (rw_mutex)->waiter;		\
+	BUG_ON(tsk != current);					\
+								\
+	set_task_state(tsk, TASK_UNINTERRUPTIBLE);		\
+	while (!(condition)) {					\
+		schedule();					\
+		set_task_state(tsk, TASK_UNINTERRUPTIBLE);	\
+	}							\
+	tsk->state = TASK_RUNNING;				\
+} while (0)
+
+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	/*
+	 * read lock slow path;
+	 * count the number of readers waiting on the read_mutex
+	 */
+	atomic_inc(&rw_mutex->read_waiters);
+	mutex_lock(&rw_mutex->read_mutex);
+
+	percpu_counter_inc(&rw_mutex->readers);
+
+	/*
+	 * wake up a possible write unlock; waiting for at least a single
+	 * reader to pass before letting a new writer through.
+	 */
+	atomic_dec(&rw_mutex->read_waiters);
+	tsk = rw_mutex->waiter;
+	if (tsk)
+		wake_up_process(tsk);
+	mutex_unlock(&rw_mutex->read_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_read_lock_slow);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	percpu_counter_inc(&rw_mutex->readers);
+	/*
+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */
+	smp_mb();
+	tsk = rw_mutex->waiter;

kernel/rwmutex.c new file +65 -0

+		return 0;
+	}
+	return 1;
+}
+EXPORT_SYMBOL(__rw_mutex_read_trylock);
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	percpu_counter_dec(&rw_mutex->readers);
+	/*
+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */
+	smp_mb();
+	tsk = rw_mutex->waiter;
+	if (unlikely(tsk)) {
+		/*
+		 * on the slow path; nudge the writer waiting for the last
+		 * reader to go away
+		 */
+		wake_up_process(tsk);
+	}
+}
+EXPORT_SYMBOL(rw_mutex_read_unlock);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	might_sleep();
+	rwsem_acquire(&rw_mutex->dep_map, subclass, 0, _RET_IP_);
+
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+	BUG_ON(rw_mutex->waiter);
+
+	/*
+	 * block new readers
+	 */
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+	rw_mutex->waiter = current;
+	/*
+	 * full barrier to sequence the store of ->waiter
+	 * and the load of ->readers
+	 */
+	smp_mb();
+	/*
+	 * and wait for all current readers to go away
+	 */
+	rw_mutex_writer_wait(rw_mutex,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}
+EXPORT_SYMBOL(rw_mutex_write_lock_nested);
+
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex)
+{
+	int waiters;
+
+	might_sleep();
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	/*
+	 * let the readers rip
+	 */

Changes include:

- wmb+rmb != mb
- ->state folded into ->waiter

---
Subject: scalable rw_mutex

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

Signed-off-by Peter Zijlstra <[email protected]>

---
 include/linux/rwmutex.h |   82 ++++++++++++++++
 kernel/Makefile         |    3 
 kernel/rwmutex.c        |  232 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 316 insertions(+), 1 deletion(-)

Index: linux-2.6/include/linux/rwmutex.h
===================================================================

include/linux/rwmutex.h new file +15 -0

@@ -0,0 +1,82 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * This file contains the public data structure and API definitions.
+ */
+#ifndef _LINUX_RWMUTEX_H
+#define _LINUX_RWMUTEX_H
+
+#include <linux/preempt.h>
+#include <linux/wait.h>
+#include <linux/percpu_counter.h>
+#include <linux/lockdep.h>
+#include <linux/mutex.h>

include/linux/rwmutex.h new file +57 -0

+	atomic_t		read_waiters;
+
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+	struct lockdep_map dep_map;
+#endif
+};
+
+void __rw_mutex_init(struct rw_mutex *rw_mutex, const char * name,
+		struct lock_class_key *key);
+void rw_mutex_destroy(struct rw_mutex *rw_mutex);
+
+#define rw_mutex_init(rw_mutex)					\
+	do {							\
+		static struct lock_class_key __key;		\
+		__rw_mutex_init((rw_mutex), #rw_mutex, &__key);	\
+	} while (0)
+
+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass);
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex);
+
+static inline int rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	int ret = __rw_mutex_read_trylock(rw_mutex);
+	if (ret)
+		rwsem_acquire_read(&rw_mutex->dep_map, 0, 1, _RET_IP_);
+	return ret;
+}
+
+static inline void rw_mutex_read_lock(struct rw_mutex *rw_mutex)
+{
+	int ret;
+
+	might_sleep();
+	rwsem_acquire_read(&rw_mutex->dep_map, 0, 0, _RET_IP_);
+
+	ret = __rw_mutex_read_trylock(rw_mutex);
+	if (!ret)
+		rw_mutex_read_lock_slow(rw_mutex);
+}
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex);
+
+static inline int rw_mutex_is_locked(struct rw_mutex *rw_mutex)
+{
+	return mutex_is_locked(&rw_mutex->write_mutex);
+}
+
+static inline void rw_mutex_write_lock(struct rw_mutex *rw_mutex)
+{
+	rw_mutex_write_lock_nested(rw_mutex, 0);
+}
+
+#endif /* _LINUX_RWMUTEX_H */

Index: linux-2.6/kernel/rwmutex.c
===================================================================

kernel/rwmutex.c new file +82 -0

@@ -0,0 +1,229 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * Its scalable in that the read count is a percpu counter and the reader fast
+ * path does not write to a shared cache-line.
+ *
+ * Its not FIFO fair, but starvation proof by alternating readers and writers.
+ */
+#include <linux/sched.h>
+#include <linux/rwmutex.h>
+#include <linux/debug_locks.h>
+#include <linux/module.h>
+
+/*
+ * rw mutex - oxymoron when we take mutex to stand for 'MUTual EXlusion'
+ *
+ * However in this context we take mutex to mean a sleeping lock, with the
+ * property that it must be released by the same context that acquired it.
+ *
+ * design goals:
+ *
+ * A sleeping reader writer lock with a scalable read side, to avoid bouncing
+ * cache-lines.
+ *
+ * dynamics:
+ *
+ * The reader fast path is modification of a percpu_counter and a read of a
+ * shared cache-line.
+ *
+ * The write side is quite heavy; it takes two mutexes, a writer mutex and a
+ * readers mutex. The writer mutex is for w <-> w interaction, the read mutex
+ * for r -> w. The read side is forced into the slow path by setting the
+ * status bit. Then it waits for all current readers to disappear.
+ *
+ * The read lock slow path; taken when the status bit is set; takes the read
+ * mutex. Because the write side also takes this mutex, the new readers are
+ * blocked. The read unlock slow path tickles the writer every time a read
+ * lock is released.
+ *
+ * Write unlock clears the status bit, and drops the read mutex; allowing new
+ * readers. It then waits for at least one waiting reader to get a lock (if
+ * there were any readers waiting) before releasing the write mutex which will
+ * allow possible other writers to come in an stop new readers, thus avoiding
+ * starvation by alternating between readers and writers
+ *
+ * considerations:
+ *
+ * The lock's space footprint is quite large (on x86_64):
+ *
+ *   88 bytes				[struct rw_mutex]
+ *    8 bytes per cpu NR_CPUS		[void *]
+ *   32 bytes per cpu (nr_cpu_ids)	[smallest slab]
+ *
+ *  408 bytes for x86_64 defconfig (NR_CPUS = 32) on a 2-way box.
+ *
+ * The write side is quite heavy; this lock is best suited for situations
+ * where the read side vastly dominates the write side.
+ */
+
+void __rw_mutex_init(struct rw_mutex *rw_mutex, const char *name,
+		struct lock_class_key *key)
+{
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+	debug_check_no_locks_freed((void *)rw_mutex, sizeof(*rw_mutex));
+	lockdep_init_map(&rw_mutex->dep_map, name, key, 0);
+#endif
+
+	percpu_counter_init(&rw_mutex->readers, 0);
+	rw_mutex->waiter = NULL;
+	mutex_init(&rw_mutex->read_mutex);
+	mutex_init(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL(__rw_mutex_init);
+
+void rw_mutex_destroy(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_destroy(&rw_mutex->readers);
+	mutex_destroy(&rw_mutex->read_mutex);
+	mutex_destroy(&rw_mutex->write_mutex);
+}

kernel/rwmutex.c new file +34 -0

+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	/*
+	 * read lock slow path;
+	 * count the number of readers waiting on the read_mutex
+	 */
+	atomic_inc(&rw_mutex->read_waiters);
+	mutex_lock(&rw_mutex->read_mutex);
+
+	percpu_counter_inc(&rw_mutex->readers);
+
+	/*
+	 * wake up a possible write unlock; waiting for at least a single
+	 * reader to pass before letting a new writer through.
+	 */
+	atomic_dec(&rw_mutex->read_waiters);
+	tsk = rw_mutex->waiter;
+	if (tsk)
+		wake_up_process(tsk);
+	mutex_unlock(&rw_mutex->read_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_read_lock_slow);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	percpu_counter_inc(&rw_mutex->readers);
+	/*
+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */

kernel/rwmutex.c new file +15 -0

+		wake_up_process(tsk);
+		return 0;
+	}
+	return 1;
+}
+EXPORT_SYMBOL(__rw_mutex_read_trylock);
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	percpu_counter_dec(&rw_mutex->readers);
+	/*

kernel/rwmutex.c new file +40 -0

+		wake_up_process(tsk);
+	}
+}
+EXPORT_SYMBOL(rw_mutex_read_unlock);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	might_sleep();
+	rwsem_acquire(&rw_mutex->dep_map, subclass, 0, _RET_IP_);
+
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+	BUG_ON(rw_mutex->waiter);
+
+	/*
+	 * block new readers
+	 */
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+	rw_mutex->waiter = current;
+	/*
+	 * full barrier to sequence the store of ->waiter
+	 * and the load of ->readers
+	 */
+	smp_mb();
+	/*
+	 * and wait for all current readers to go away
+	 */
+	rw_mutex_writer_wait(rw_mutex,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}
+EXPORT_SYMBOL(rw_mutex_write_lock_nested);
+
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex)
+{
+	int waiters;
+
+	might_sleep();
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	/*
+	 * let the readers rip

kernel/rwmutex.c new file +7 -0

+	rw_mutex->waiter = NULL;
+	/*
+	 * before we let the writers rip
+	 */
+	mutex_unlock(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_write_unlock);

Index: linux-2.6/kernel/Makefile
===================================================================

Quoted 8 lines · Peter Zijlstra

Changes include:

- wmb+rmb != mb
- ->state folded into ->waiter

---
Subject: scalable rw_mutex

Quoted 6 lines

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

Hmmm... brlock reincarnates, but as sleeplock. ;-)

I believe that there are a few severe problems in this code, search for "!!!" to quickly find the specific areas that concern me.

Thanx, Paul

Quoted 6 lines

Signed-off-by Peter Zijlstra <[email protected]>

---
 include/linux/rwmutex.h |   82 ++++++++++++++++
 kernel/Makefile         |    3 
 kernel/rwmutex.c        |  232 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 316 insertions(+), 1 deletion(-)

List of races that must be resolved:

1. Read acquire vs. write acquire.

rw_mutex_read_lock() invokes __rw_mutex_read_trylock(), if
this fails, invokes rw_mutex_read_lock_slow().

__rw_mutex_read_trylock() increments the per-CPU counter,
does smp_mb(), picks up ->waiter:
if non-NULL decrements the per-CPU counter, does a barrier(), does wake_up_process() on the task fetched from ->waiter. Return failure.

Otherwise, return success.

rw_mutex_read_lock_slow() increments ->read_waiters,
acquires ->read_mutex, increments the ->readers counter, and decrements the ->read_waiters counter. It then fetches ->waiter, and, if non-NULL, wakes up the tasks.
Either way, releases ->read_mutex.

rw_mutex_write_lock_nested(): acquires ->write_mutex, which
prevents any writer-writer races. Acquires ->read_mutex, which does -not- prevent readers from continuing to acquire. Sets ->waiter to current, which -does- (eventually) stop readers. smp_mb(), then invokes rw_mutex_writer_wait() for the sum of the per-CPU counters to go to zero.

!In principle, this could be indefinitely postponed, but in practice would require an infinite number of reading tasks, so probably OK to ignore. ;-) This can occur because the readers unconditionally increment their per-CPU counters, and decrement it only later.

The smp_mb()s currently in the reader and the writer code forms a Dekker-algorithm-like barrier, preventing both the reader and writer from entering their critical section, as required.

2. Read acquire vs. write release (need to avoid reader sleeping

	forever, even in the case where no one ever uses the lock again).

	rw_mutex_read_lock() invokes __rw_mutex_read_trylock(), if
		this fails, invokes rw_mutex_read_lock_slow().

		__rw_mutex_read_trylock() increments the per-CPU counter,
			does smp_mb(), picks up ->waiter:
				if non-NULL decrements the per-CPU
				counter, does a barrier(), does
				wake_up_process() on the task fetched
				from ->waiter.  Return failure.

				Otherwise, return success.

		rw_mutex_read_lock_slow() increments ->read_waiters,
			acquires ->read_mutex, increments the ->readers
			counter, and decrements the ->read_waiters
			counter.  It then fetches ->waiter, and, if
			non-NULL, wakes up the tasks.
			Either way, releases ->read_mutex.

	rw_mutex_write_unlock(): pick up ->read_waiters, release
		->read_mutex, if copy of ->read_waiters was non-NULL
		do slow path (but refetches ->read_waiters??? why???
		[Ah -- refetched each time through the loop in the
		rw_mutex_writer_wait() macro), NULL out ->waiter,
		then release ->write_mutex.

		Slow path: Pick up ->waiter, make sure it is us,
			set state to uninterruptible, loop while
			->read_waiters less than the value fetched
			earlier from ->read_waiters, scheduling each time
			through, set state back to running.

		(!!!This is subject to indefinite postponement by a
		flurry of readers, see the commentary for 
		rw_mutex_write_unlock() interspersed below.)

	However, the fact that ->read_mutex is unconditionally released
	by the writer prevents the readers from being starved.

3. Read release vs. write acquire (similar to #2).

rw_mutex_read_unlock(): decrement the per-CPU counter, smb_mb(),
pick up ->waiter, if non-NULL, wake it up.

rw_mutex_write_lock_nested(): acquires ->write_mutex, which
prevents any writer-writer races. Acquires ->read_mutex, which does -not- prevent readers from continuing to acquire. Sets ->waiter to current, which -does- (eventually) stop readers. smp_mb(), then invokes rw_mutex_writer_wait() for the sum of the per-CPU counters to go to zero.

As written, the writer never really blocks, so omitting the wakeup is OK. (Failing to really block is -not- OK given realtime processes, but more on that later.)

4. Read release vs. write release -- presumably this one cannot
happen, but wouldn't want to fall prey to complacency. ;-)

Strangely enough, this -can- happen!!! When the readers
indefinitely postpone writer release, the readers will also be read-releasing... So the readers will be needlessly waking up the task that is trying to finish releasing the write lock. :-/ Not a big deal in this case, but just shows to go that when reviewing locking algorithms, you should -never- restrict yourself to considering only things that seem possible. ;-)

Quoted 20 lines

Index: linux-2.6/include/linux/rwmutex.h
===================================================================

include/linux/rwmutex.h new file +15 -0

@@ -0,0 +1,82 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * This file contains the public data structure and API definitions.
+ */
+#ifndef _LINUX_RWMUTEX_H
+#define _LINUX_RWMUTEX_H
+
+#include <linux/preempt.h>
+#include <linux/wait.h>
+#include <linux/percpu_counter.h>
+#include <linux/lockdep.h>
+#include <linux/mutex.h>

include/linux/rwmutex.h line 25

Quoted 10 lines

include/linux/rwmutex.h new file +10 -0

+#include <asm/atomic.h>
+
+struct rw_mutex {
+	/* Read mostly global */
+	struct percpu_counter	readers;
+
+	/* The following variables are only for the slowpath */
+	struct task_struct	*waiter;	/* w -> r waiting */
+	struct mutex		read_mutex;	/* r -> w waiting */
+	struct mutex		write_mutex;	/* w -> w waiting */

Priority-inheritance relationship? Seems like this would be tough to arrange while still avoiding deadlock...

Readers currently do boost the writer via ->read_mutex.

!!!

Writers currently do -not- boost readers. In fact, the identities of the readers are not tracked, so there is no way for the writer to tell what to boost. Admittedly, write-to-read priority boosting is quite the can of worms if you allow more than one reader, but something will be needed for realtime kernels.

A brlock-like implementation can allow boosting in both directions, but has other issues (such as write-side performance even in the case where there are no readers).

Quoted 144 lines

include/linux/rwmutex.h new file +57 -0

+	atomic_t		read_waiters;
+
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+	struct lockdep_map dep_map;
+#endif
+};
+
+void __rw_mutex_init(struct rw_mutex *rw_mutex, const char * name,
+		struct lock_class_key *key);
+void rw_mutex_destroy(struct rw_mutex *rw_mutex);
+
+#define rw_mutex_init(rw_mutex)					\
+	do {							\
+		static struct lock_class_key __key;		\
+		__rw_mutex_init((rw_mutex), #rw_mutex, &__key);	\
+	} while (0)
+
+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass);
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex);
+
+static inline int rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	int ret = __rw_mutex_read_trylock(rw_mutex);
+	if (ret)
+		rwsem_acquire_read(&rw_mutex->dep_map, 0, 1, _RET_IP_);
+	return ret;
+}
+
+static inline void rw_mutex_read_lock(struct rw_mutex *rw_mutex)
+{
+	int ret;
+
+	might_sleep();
+	rwsem_acquire_read(&rw_mutex->dep_map, 0, 0, _RET_IP_);
+
+	ret = __rw_mutex_read_trylock(rw_mutex);
+	if (!ret)
+		rw_mutex_read_lock_slow(rw_mutex);
+}
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex);
+
+static inline int rw_mutex_is_locked(struct rw_mutex *rw_mutex)
+{
+	return mutex_is_locked(&rw_mutex->write_mutex);
+}
+
+static inline void rw_mutex_write_lock(struct rw_mutex *rw_mutex)
+{
+	rw_mutex_write_lock_nested(rw_mutex, 0);
+}
+
+#endif /* _LINUX_RWMUTEX_H */

Index: linux-2.6/kernel/rwmutex.c
===================================================================

kernel/rwmutex.c new file +82 -0

@@ -0,0 +1,229 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * Its scalable in that the read count is a percpu counter and the reader fast
+ * path does not write to a shared cache-line.
+ *
+ * Its not FIFO fair, but starvation proof by alternating readers and writers.
+ */
+#include <linux/sched.h>
+#include <linux/rwmutex.h>
+#include <linux/debug_locks.h>
+#include <linux/module.h>
+
+/*
+ * rw mutex - oxymoron when we take mutex to stand for 'MUTual EXlusion'
+ *
+ * However in this context we take mutex to mean a sleeping lock, with the
+ * property that it must be released by the same context that acquired it.
+ *
+ * design goals:
+ *
+ * A sleeping reader writer lock with a scalable read side, to avoid bouncing
+ * cache-lines.
+ *
+ * dynamics:
+ *
+ * The reader fast path is modification of a percpu_counter and a read of a
+ * shared cache-line.
+ *
+ * The write side is quite heavy; it takes two mutexes, a writer mutex and a
+ * readers mutex. The writer mutex is for w <-> w interaction, the read mutex
+ * for r -> w. The read side is forced into the slow path by setting the
+ * status bit. Then it waits for all current readers to disappear.
+ *
+ * The read lock slow path; taken when the status bit is set; takes the read
+ * mutex. Because the write side also takes this mutex, the new readers are
+ * blocked. The read unlock slow path tickles the writer every time a read
+ * lock is released.
+ *
+ * Write unlock clears the status bit, and drops the read mutex; allowing new
+ * readers. It then waits for at least one waiting reader to get a lock (if
+ * there were any readers waiting) before releasing the write mutex which will
+ * allow possible other writers to come in an stop new readers, thus avoiding
+ * starvation by alternating between readers and writers
+ *
+ * considerations:
+ *
+ * The lock's space footprint is quite large (on x86_64):
+ *
+ *   88 bytes				[struct rw_mutex]
+ *    8 bytes per cpu NR_CPUS		[void *]
+ *   32 bytes per cpu (nr_cpu_ids)	[smallest slab]
+ *
+ *  408 bytes for x86_64 defconfig (NR_CPUS = 32) on a 2-way box.
+ *
+ * The write side is quite heavy; this lock is best suited for situations
+ * where the read side vastly dominates the write side.
+ */
+
+void __rw_mutex_init(struct rw_mutex *rw_mutex, const char *name,
+		struct lock_class_key *key)
+{
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+	debug_check_no_locks_freed((void *)rw_mutex, sizeof(*rw_mutex));
+	lockdep_init_map(&rw_mutex->dep_map, name, key, 0);
+#endif
+
+	percpu_counter_init(&rw_mutex->readers, 0);
+	rw_mutex->waiter = NULL;
+	mutex_init(&rw_mutex->read_mutex);
+	mutex_init(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL(__rw_mutex_init);
+
+void rw_mutex_destroy(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_destroy(&rw_mutex->readers);
+	mutex_destroy(&rw_mutex->read_mutex);
+	mutex_destroy(&rw_mutex->write_mutex);
+}

kernel/rwmutex.c line 92

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+EXPORT_SYMBOL(rw_mutex_destroy);
+
+#define rw_mutex_writer_wait(rw_mutex, condition)		\
+do {								\
+	struct task_struct *tsk = (rw_mutex)->waiter;		\
+	BUG_ON(tsk != current);					\
+								\
+	set_task_state(tsk, TASK_UNINTERRUPTIBLE);		\
+	while (!(condition)) {					\
+		schedule();					\

!!!

If there are at least as many of these scalable reader-writer locks as there are CPUs, and each such lock has a realtime-priority writer executing, you can have an infinite loop starving all the non-realtime readers, who are then unable to ever read-release the lock, preventing the writers from making progress. Or am I missing something subtle here?

kernel/rwmutex.c line 96

Quoted 4 lines

kernel/rwmutex.c new file +4 -0

+		set_task_state(tsk, TASK_UNINTERRUPTIBLE);	\
+	}							\
+	tsk->state = TASK_RUNNING;				\
+} while (0)

Can't something like __wait_event() be used here?

Quoted 34 lines

kernel/rwmutex.c new file +34 -0

+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	/*
+	 * read lock slow path;
+	 * count the number of readers waiting on the read_mutex
+	 */
+	atomic_inc(&rw_mutex->read_waiters);
+	mutex_lock(&rw_mutex->read_mutex);
+
+	percpu_counter_inc(&rw_mutex->readers);
+
+	/*
+	 * wake up a possible write unlock; waiting for at least a single
+	 * reader to pass before letting a new writer through.
+	 */
+	atomic_dec(&rw_mutex->read_waiters);
+	tsk = rw_mutex->waiter;
+	if (tsk)
+		wake_up_process(tsk);
+	mutex_unlock(&rw_mutex->read_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_read_lock_slow);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	percpu_counter_inc(&rw_mutex->readers);
+	/*
+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */

kernel/rwmutex.c line 141

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	smp_mb();
+	tsk = rw_mutex->waiter;
+	if (unlikely(tsk)) {
+		percpu_counter_dec(&rw_mutex->readers);
+		/*
+		 * ensure the ->readers store has taken place before we issue
+		 * the wake_up
+		 *
+		 * XXX: or does this require an smp_wmb() and the waiter to do
+		 *   (smp_rmb(), percpu_counter(&rw_mutex->readers) == 0)

I don't think that -anything- is needed here as written. The "sleeping" writers don't really block, they instead simply loop on schedule(). Of course that in itself is a -really- bad idea if case of realtime-priority writers...

So the writers need to really block, which will make the wakeup code much more ticklish. In that case, it seems to me that smp_mb() will be needed here, but much will depend on the exact structure of the block/wakeup scheme chosen.

kernel/rwmutex.c line 147

Quoted 5 lines

kernel/rwmutex.c new file +5 -0

+		barrier();
+		/*
+		 * possibly wake up a writer waiting for this reference to
+		 * disappear
+		 */

!!!

Suppose we get delayed here (preempted, interrupted, whatever) and the task referenced by tsk has exited and cleaned up before we get a chance to wake it up? We might well be "waking up" some entirely different data structure in that case, not?

Quoted 15 lines

kernel/rwmutex.c new file +15 -0

+		wake_up_process(tsk);
+		return 0;
+	}
+	return 1;
+}
+EXPORT_SYMBOL(__rw_mutex_read_trylock);
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	percpu_counter_dec(&rw_mutex->readers);
+	/*

kernel/rwmutex.c line 172

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */
+	smp_mb();
+	tsk = rw_mutex->waiter;
+	if (unlikely(tsk)) {
+		/*
+		 * on the slow path; nudge the writer waiting for the last
+		 * reader to go away
+		 */

!!!

What if we are delayed (interrupted, preempted, whatever) here, so that the task referenced by tsk has exited and been cleaned up before we can execute the following? We could end up "waking up" the freelist, not?

Quoted 40 lines

kernel/rwmutex.c new file +40 -0

+		wake_up_process(tsk);
+	}
+}
+EXPORT_SYMBOL(rw_mutex_read_unlock);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	might_sleep();
+	rwsem_acquire(&rw_mutex->dep_map, subclass, 0, _RET_IP_);
+
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+	BUG_ON(rw_mutex->waiter);
+
+	/*
+	 * block new readers
+	 */
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+	rw_mutex->waiter = current;
+	/*
+	 * full barrier to sequence the store of ->waiter
+	 * and the load of ->readers
+	 */
+	smp_mb();
+	/*
+	 * and wait for all current readers to go away
+	 */
+	rw_mutex_writer_wait(rw_mutex,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}
+EXPORT_SYMBOL(rw_mutex_write_lock_nested);
+
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex)
+{
+	int waiters;
+
+	might_sleep();
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	/*
+	 * let the readers rip

kernel/rwmutex.c line 222

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	 */
+	waiters = atomic_read(&rw_mutex->read_waiters);
+	mutex_unlock(&rw_mutex->read_mutex);
+	/*
+	 * wait for at least 1 reader to get through
+	 */
+	if (waiters) {
+		rw_mutex_writer_wait(rw_mutex,
+			(atomic_read(&rw_mutex->read_waiters) < waiters));
+	}

!!!

Readers can indefinitely postpone the write-unlock at this point. Suppose that there is one reader waiting when we fetch ->read_waiters above. Suppose that as soon as the mutex is released, a large number of readers arrive. Because we have not yet NULLed ->waiter, all of these readers will take the slow path, incrementing the ->read_waiters counter, acquiring the ->read_mutex, and so on. The ->read_mutex lock acquisition is likely to be the bottleneck for large systems, which would result in the count remaining 2 or higher indefinitely. The readers would make progress (albeit quite inefficiently), but the writer would never get out of the loop in the rw_mutex_writer_wait() macro.

Consider instead using a generation number that just increments every time a reader successfully acquires the lock and is never decremented. That way, you can unambiguously determine when at least one reader has made progress. An additional counter required, but might be able to multiplex with something else. Or get rid of the current ->read_waiters in favor of the generation number? This latter seems like it should be possible, at least at first glance... Just change the name of the field, get rid of the decrement, and change the comparison to "!=", right?

Quoted 19 lines

kernel/rwmutex.c new file +7 -0

+	rw_mutex->waiter = NULL;
+	/*
+	 * before we let the writers rip
+	 */
+	mutex_unlock(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_write_unlock);

Index: linux-2.6/kernel/Makefile
===================================================================

kernel/Makefile +2 -1

@@ -8,7 +8,8 @@ obj-y     = sched.o fork.o exec_domain.o
 	    signal.o sys.o kmod.o workqueue.o pid.o \
 	    rcupdate.o extable.o params.o posix-timers.o \
 	    kthread.o wait.o kfifo.o sys_ni.o posix-cpu-timers.o mutex.o \
-	    hrtimer.o rwsem.o latency.o nsproxy.o srcu.o die_notifier.o
+	    hrtimer.o rwsem.o latency.o nsproxy.o srcu.o die_notifier.o \
+	    rwmutex.o
 

kernel/Makefile obj-y = sched.o fork.o exec_domain.o · line 15

Quoted 4 lines

kernel/Makefile +0 -0

 obj-$(CONFIG_STACKTRACE) += stacktrace.o
 obj-y += time/

Quoted 9 lines · Paul E. McKenney

Quoted 8 lines · Peter Zijlstra

Changes include:

- wmb+rmb != mb
- ->state folded into ->waiter

---
Subject: scalable rw_mutex

Quoted 8 lines

Quoted 6 lines

Scalable reader/writer lock.

Its scalable in that the read count is a percpu counter and the reader fast path does not write to a shared cache-line.

Its not FIFO fair, but starvation proof by alternating readers and writers.

Hmmm... brlock reincarnates, but as sleeplock. ;-)

/me googles... ooh, yeah, quite similar.

Quoted 2 lines

I believe that there are a few severe problems in this code, search for "!!!" to quickly find the specific areas that concern me.

Thanks for the feedback, replies at the !!! sites.

Quoted 166 lines

Thanx, Paul

Quoted 6 lines

Signed-off-by Peter Zijlstra <[email protected]>

---
 include/linux/rwmutex.h |   82 ++++++++++++++++
 kernel/Makefile         |    3 
 kernel/rwmutex.c        |  232 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 316 insertions(+), 1 deletion(-)

List of races that must be resolved:

1. Read acquire vs. write acquire.

rw_mutex_read_lock() invokes __rw_mutex_read_trylock(), if
this fails, invokes rw_mutex_read_lock_slow().

__rw_mutex_read_trylock() increments the per-CPU counter,
does smp_mb(), picks up ->waiter:
if non-NULL decrements the per-CPU counter, does a barrier(), does wake_up_process() on the task fetched from ->waiter. Return failure.

Otherwise, return success.

rw_mutex_read_lock_slow() increments ->read_waiters,
acquires ->read_mutex, increments the ->readers counter, and decrements the ->read_waiters counter. It then fetches ->waiter, and, if non-NULL, wakes up the tasks.
Either way, releases ->read_mutex.

rw_mutex_write_lock_nested(): acquires ->write_mutex, which
prevents any writer-writer races. Acquires ->read_mutex, which does -not- prevent readers from continuing to acquire. Sets ->waiter to current, which -does- (eventually) stop readers. smp_mb(), then invokes rw_mutex_writer_wait() for the sum of the per-CPU counters to go to zero.

!In principle, this could be indefinitely postponed, but in practice would require an infinite number of reading tasks, so probably OK to ignore. ;-) This can occur because the readers unconditionally increment their per-CPU counters, and decrement it only later.

The smp_mb()s currently in the reader and the writer code forms a Dekker-algorithm-like barrier, preventing both the reader and writer from entering their critical section, as required.

2. Read acquire vs. write release (need to avoid reader sleeping

	forever, even in the case where no one ever uses the lock again).

	rw_mutex_read_lock() invokes __rw_mutex_read_trylock(), if
		this fails, invokes rw_mutex_read_lock_slow().

		__rw_mutex_read_trylock() increments the per-CPU counter,
			does smp_mb(), picks up ->waiter:
				if non-NULL decrements the per-CPU
				counter, does a barrier(), does
				wake_up_process() on the task fetched
				from ->waiter.  Return failure.

				Otherwise, return success.

		rw_mutex_read_lock_slow() increments ->read_waiters,
			acquires ->read_mutex, increments the ->readers
			counter, and decrements the ->read_waiters
			counter.  It then fetches ->waiter, and, if
			non-NULL, wakes up the tasks.
			Either way, releases ->read_mutex.

	rw_mutex_write_unlock(): pick up ->read_waiters, release
		->read_mutex, if copy of ->read_waiters was non-NULL
		do slow path (but refetches ->read_waiters??? why???
		[Ah -- refetched each time through the loop in the
		rw_mutex_writer_wait() macro), NULL out ->waiter,
		then release ->write_mutex.

		Slow path: Pick up ->waiter, make sure it is us,
			set state to uninterruptible, loop while
			->read_waiters less than the value fetched
			earlier from ->read_waiters, scheduling each time
			through, set state back to running.

		(!!!This is subject to indefinite postponement by a
		flurry of readers, see the commentary for 
		rw_mutex_write_unlock() interspersed below.)

	However, the fact that ->read_mutex is unconditionally released
	by the writer prevents the readers from being starved.

3. Read release vs. write acquire (similar to #2).

rw_mutex_read_unlock(): decrement the per-CPU counter, smb_mb(),
pick up ->waiter, if non-NULL, wake it up.

rw_mutex_write_lock_nested(): acquires ->write_mutex, which
prevents any writer-writer races. Acquires ->read_mutex, which does -not- prevent readers from continuing to acquire. Sets ->waiter to current, which -does- (eventually) stop readers. smp_mb(), then invokes rw_mutex_writer_wait() for the sum of the per-CPU counters to go to zero.

As written, the writer never really blocks, so omitting the wakeup is OK. (Failing to really block is -not- OK given realtime processes, but more on that later.)

4. Read release vs. write release -- presumably this one cannot
happen, but wouldn't want to fall prey to complacency. ;-)

Strangely enough, this -can- happen!!! When the readers
indefinitely postpone writer release, the readers will also be read-releasing... So the readers will be needlessly waking up the task that is trying to finish releasing the write lock. :-/ Not a big deal in this case, but just shows to go that when reviewing locking algorithms, you should -never- restrict yourself to considering only things that seem possible. ;-)

Quoted 20 lines

Index: linux-2.6/include/linux/rwmutex.h
===================================================================

include/linux/rwmutex.h new file +15 -0

@@ -0,0 +1,82 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * This file contains the public data structure and API definitions.
+ */
+#ifndef _LINUX_RWMUTEX_H
+#define _LINUX_RWMUTEX_H
+
+#include <linux/preempt.h>
+#include <linux/wait.h>
+#include <linux/percpu_counter.h>
+#include <linux/lockdep.h>
+#include <linux/mutex.h>

include/linux/rwmutex.h line 25

Quoted 10 lines

include/linux/rwmutex.h new file +10 -0

+#include <asm/atomic.h>
+
+struct rw_mutex {
+	/* Read mostly global */
+	struct percpu_counter	readers;
+
+	/* The following variables are only for the slowpath */
+	struct task_struct	*waiter;	/* w -> r waiting */
+	struct mutex		read_mutex;	/* r -> w waiting */
+	struct mutex		write_mutex;	/* w -> w waiting */

Priority-inheritance relationship? Seems like this would be tough to arrange while still avoiding deadlock...

Readers currently do boost the writer via ->read_mutex.

!!!

Writers currently do -not- boost readers. In fact, the identities of the readers are not tracked, so there is no way for the writer to tell what to boost. Admittedly, write-to-read priority boosting is quite the can of worms if you allow more than one reader, but something will be needed for realtime kernels.

Quoted 3 lines

A brlock-like implementation can allow boosting in both directions, but has other issues (such as write-side performance even in the case where there are no readers).

You could short circuit the no readers case for a full brlock like affair. But yeah, the write side will be horribly heavy when you have to take nr_cpu_ids() locks.

Quoted 157 lines

Quoted 144 lines

include/linux/rwmutex.h new file +57 -0

+	atomic_t		read_waiters;
+
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+	struct lockdep_map dep_map;
+#endif
+};
+
+void __rw_mutex_init(struct rw_mutex *rw_mutex, const char * name,
+		struct lock_class_key *key);
+void rw_mutex_destroy(struct rw_mutex *rw_mutex);
+
+#define rw_mutex_init(rw_mutex)					\
+	do {							\
+		static struct lock_class_key __key;		\
+		__rw_mutex_init((rw_mutex), #rw_mutex, &__key);	\
+	} while (0)
+
+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass);
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex);
+
+static inline int rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	int ret = __rw_mutex_read_trylock(rw_mutex);
+	if (ret)
+		rwsem_acquire_read(&rw_mutex->dep_map, 0, 1, _RET_IP_);
+	return ret;
+}
+
+static inline void rw_mutex_read_lock(struct rw_mutex *rw_mutex)
+{
+	int ret;
+
+	might_sleep();
+	rwsem_acquire_read(&rw_mutex->dep_map, 0, 0, _RET_IP_);
+
+	ret = __rw_mutex_read_trylock(rw_mutex);
+	if (!ret)
+		rw_mutex_read_lock_slow(rw_mutex);
+}
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex);
+
+static inline int rw_mutex_is_locked(struct rw_mutex *rw_mutex)
+{
+	return mutex_is_locked(&rw_mutex->write_mutex);
+}
+
+static inline void rw_mutex_write_lock(struct rw_mutex *rw_mutex)
+{
+	rw_mutex_write_lock_nested(rw_mutex, 0);
+}
+
+#endif /* _LINUX_RWMUTEX_H */

Index: linux-2.6/kernel/rwmutex.c
===================================================================

kernel/rwmutex.c new file +82 -0

@@ -0,0 +1,229 @@
+/*
+ * Scalable reader/writer lock.
+ *
+ *  Copyright (C) 2007 Red Hat, Inc., Peter Zijlstra <[email protected]>
+ *
+ * Its scalable in that the read count is a percpu counter and the reader fast
+ * path does not write to a shared cache-line.
+ *
+ * Its not FIFO fair, but starvation proof by alternating readers and writers.
+ */
+#include <linux/sched.h>
+#include <linux/rwmutex.h>
+#include <linux/debug_locks.h>
+#include <linux/module.h>
+
+/*
+ * rw mutex - oxymoron when we take mutex to stand for 'MUTual EXlusion'
+ *
+ * However in this context we take mutex to mean a sleeping lock, with the
+ * property that it must be released by the same context that acquired it.
+ *
+ * design goals:
+ *
+ * A sleeping reader writer lock with a scalable read side, to avoid bouncing
+ * cache-lines.
+ *
+ * dynamics:
+ *
+ * The reader fast path is modification of a percpu_counter and a read of a
+ * shared cache-line.
+ *
+ * The write side is quite heavy; it takes two mutexes, a writer mutex and a
+ * readers mutex. The writer mutex is for w <-> w interaction, the read mutex
+ * for r -> w. The read side is forced into the slow path by setting the
+ * status bit. Then it waits for all current readers to disappear.
+ *
+ * The read lock slow path; taken when the status bit is set; takes the read
+ * mutex. Because the write side also takes this mutex, the new readers are
+ * blocked. The read unlock slow path tickles the writer every time a read
+ * lock is released.
+ *
+ * Write unlock clears the status bit, and drops the read mutex; allowing new
+ * readers. It then waits for at least one waiting reader to get a lock (if
+ * there were any readers waiting) before releasing the write mutex which will
+ * allow possible other writers to come in an stop new readers, thus avoiding
+ * starvation by alternating between readers and writers
+ *
+ * considerations:
+ *
+ * The lock's space footprint is quite large (on x86_64):
+ *
+ *   88 bytes				[struct rw_mutex]
+ *    8 bytes per cpu NR_CPUS		[void *]
+ *   32 bytes per cpu (nr_cpu_ids)	[smallest slab]
+ *
+ *  408 bytes for x86_64 defconfig (NR_CPUS = 32) on a 2-way box.
+ *
+ * The write side is quite heavy; this lock is best suited for situations
+ * where the read side vastly dominates the write side.
+ */
+
+void __rw_mutex_init(struct rw_mutex *rw_mutex, const char *name,
+		struct lock_class_key *key)
+{
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+	debug_check_no_locks_freed((void *)rw_mutex, sizeof(*rw_mutex));
+	lockdep_init_map(&rw_mutex->dep_map, name, key, 0);
+#endif
+
+	percpu_counter_init(&rw_mutex->readers, 0);
+	rw_mutex->waiter = NULL;
+	mutex_init(&rw_mutex->read_mutex);
+	mutex_init(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL(__rw_mutex_init);
+
+void rw_mutex_destroy(struct rw_mutex *rw_mutex)
+{
+	percpu_counter_destroy(&rw_mutex->readers);
+	mutex_destroy(&rw_mutex->read_mutex);
+	mutex_destroy(&rw_mutex->write_mutex);
+}

kernel/rwmutex.c line 92

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+EXPORT_SYMBOL(rw_mutex_destroy);
+
+#define rw_mutex_writer_wait(rw_mutex, condition)		\
+do {								\
+	struct task_struct *tsk = (rw_mutex)->waiter;		\
+	BUG_ON(tsk != current);					\
+								\
+	set_task_state(tsk, TASK_UNINTERRUPTIBLE);		\
+	while (!(condition)) {					\
+		schedule();					\

!!!

Quoted 5 lines

If there are at least as many of these scalable reader-writer locks as there are CPUs, and each such lock has a realtime-priority writer executing, you can have an infinite loop starving all the non-realtime readers, who are then unable to ever read-release the lock, preventing the writers from making progress. Or am I missing something subtle here?

It will actually block the task. schedule() will make it sleep when state is uninterruptible.

Quoted 6 lines

kernel/rwmutex.c line 96

Quoted 4 lines

kernel/rwmutex.c new file +4 -0

+		set_task_state(tsk, TASK_UNINTERRUPTIBLE);	\
+	}							\
+	tsk->state = TASK_RUNNING;				\
+} while (0)

Can't something like __wait_event() be used here?

Blame Oleg for this :-)
He suggested I not use waitqueues since I only ever have a single waiter.

Quoted 50 lines

Quoted 34 lines

kernel/rwmutex.c new file +34 -0

+void rw_mutex_read_lock_slow(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	/*
+	 * read lock slow path;
+	 * count the number of readers waiting on the read_mutex
+	 */
+	atomic_inc(&rw_mutex->read_waiters);
+	mutex_lock(&rw_mutex->read_mutex);
+
+	percpu_counter_inc(&rw_mutex->readers);
+
+	/*
+	 * wake up a possible write unlock; waiting for at least a single
+	 * reader to pass before letting a new writer through.
+	 */
+	atomic_dec(&rw_mutex->read_waiters);
+	tsk = rw_mutex->waiter;
+	if (tsk)
+		wake_up_process(tsk);
+	mutex_unlock(&rw_mutex->read_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_read_lock_slow);
+
+int __rw_mutex_read_trylock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	percpu_counter_inc(&rw_mutex->readers);
+	/*
+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */

kernel/rwmutex.c line 141

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	smp_mb();
+	tsk = rw_mutex->waiter;
+	if (unlikely(tsk)) {
+		percpu_counter_dec(&rw_mutex->readers);
+		/*
+		 * ensure the ->readers store has taken place before we issue
+		 * the wake_up
+		 *
+		 * XXX: or does this require an smp_wmb() and the waiter to do
+		 *   (smp_rmb(), percpu_counter(&rw_mutex->readers) == 0)

I don't think that -anything- is needed here as written. The "sleeping" writers don't really block, they instead simply loop on schedule(). Of course that in itself is a -really- bad idea if case of realtime-priority writers...

Quoted 4 lines

So the writers need to really block, which will make the wakeup code much more ticklish. In that case, it seems to me that smp_mb() will be needed here, but much will depend on the exact structure of the block/wakeup scheme chosen.

It does really sleep. And Nick thinks the suggested wmb + rmb will suffice if I read his latest mail correctly.

Quoted 8 lines

kernel/rwmutex.c line 147

Quoted 5 lines

kernel/rwmutex.c new file +5 -0

+		barrier();
+		/*
+		 * possibly wake up a writer waiting for this reference to
+		 * disappear
+		 */

!!!

Quoted 4 lines

Suppose we get delayed here (preempted, interrupted, whatever) and the task referenced by tsk has exited and cleaned up before we get a chance to wake it up? We might well be "waking up" some entirely different data structure in that case, not?

Yes, this is a real problem, I'll borrow a spinlock from one of the mutexes.

Quoted 28 lines

Quoted 15 lines

kernel/rwmutex.c new file +15 -0

+		wake_up_process(tsk);
+		return 0;
+	}
+	return 1;
+}
+EXPORT_SYMBOL(__rw_mutex_read_trylock);
+
+void rw_mutex_read_unlock(struct rw_mutex *rw_mutex)
+{
+	struct task_struct *tsk;
+
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	percpu_counter_dec(&rw_mutex->readers);
+	/*

kernel/rwmutex.c line 172

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	 * ensure the ->readers store and the ->waiter load is properly
+	 * sequenced
+	 */
+	smp_mb();
+	tsk = rw_mutex->waiter;
+	if (unlikely(tsk)) {
+		/*
+		 * on the slow path; nudge the writer waiting for the last
+		 * reader to go away
+		 */

!!!

Quoted 3 lines

What if we are delayed (interrupted, preempted, whatever) here, so that the task referenced by tsk has exited and been cleaned up before we can execute the following? We could end up "waking up" the freelist, not?

idem.

Quoted 53 lines

Quoted 40 lines

kernel/rwmutex.c new file +40 -0

+		wake_up_process(tsk);
+	}
+}
+EXPORT_SYMBOL(rw_mutex_read_unlock);
+
+void rw_mutex_write_lock_nested(struct rw_mutex *rw_mutex, int subclass)
+{
+	might_sleep();
+	rwsem_acquire(&rw_mutex->dep_map, subclass, 0, _RET_IP_);
+
+	mutex_lock_nested(&rw_mutex->write_mutex, subclass);
+	BUG_ON(rw_mutex->waiter);
+
+	/*
+	 * block new readers
+	 */
+	mutex_lock_nested(&rw_mutex->read_mutex, subclass);
+	rw_mutex->waiter = current;
+	/*
+	 * full barrier to sequence the store of ->waiter
+	 * and the load of ->readers
+	 */
+	smp_mb();
+	/*
+	 * and wait for all current readers to go away
+	 */
+	rw_mutex_writer_wait(rw_mutex,
+			(percpu_counter_sum(&rw_mutex->readers) == 0));
+}
+EXPORT_SYMBOL(rw_mutex_write_lock_nested);
+
+void rw_mutex_write_unlock(struct rw_mutex *rw_mutex)
+{
+	int waiters;
+
+	might_sleep();
+	rwsem_release(&rw_mutex->dep_map, 1, _RET_IP_);
+
+	/*
+	 * let the readers rip

kernel/rwmutex.c line 222

Quoted 10 lines

kernel/rwmutex.c new file +10 -0

+	 */
+	waiters = atomic_read(&rw_mutex->read_waiters);
+	mutex_unlock(&rw_mutex->read_mutex);
+	/*
+	 * wait for at least 1 reader to get through
+	 */
+	if (waiters) {
+		rw_mutex_writer_wait(rw_mutex,
+			(atomic_read(&rw_mutex->read_waiters) < waiters));
+	}

!!!

Quoted 11 lines

Readers can indefinitely postpone the write-unlock at this point. Suppose that there is one reader waiting when we fetch ->read_waiters above. Suppose that as soon as the mutex is released, a large number of readers arrive. Because we have not yet NULLed ->waiter, all of these readers will take the slow path, incrementing the ->read_waiters counter, acquiring the ->read_mutex, and so on. The ->read_mutex lock acquisition is likely to be the bottleneck for large systems, which would result in the count remaining 2 or higher indefinitely. The readers would make progress (albeit quite inefficiently), but the writer would never get out of the loop in the rw_mutex_writer_wait() macro.

Ah, yeah. Ouch!

I missed this detail when I got rid of ->state. I used to clear the reader slow path before unlocking the ->read_mutex.

Quoted 8 lines

Consider instead using a generation number that just increments every time a reader successfully acquires the lock and is never decremented. That way, you can unambiguously determine when at least one reader has made progress. An additional counter required, but might be able to multiplex with something else. Or get rid of the current ->read_waiters in favor of the generation number? This latter seems like it should be possible, at least at first glance... Just change the name of the field, get rid of the decrement, and change the comparison to "!=", right?

Yes, this seems like a very good suggestion, I'll give it a shot. Thanks!

Quoted 19 lines

Quoted 19 lines

kernel/rwmutex.c new file +7 -0

+	rw_mutex->waiter = NULL;
+	/*
+	 * before we let the writers rip
+	 */
+	mutex_unlock(&rw_mutex->write_mutex);
+}
+EXPORT_SYMBOL(rw_mutex_write_unlock);

Index: linux-2.6/kernel/Makefile
===================================================================

kernel/Makefile +2 -1

@@ -8,7 +8,8 @@ obj-y     = sched.o fork.o exec_domain.o
 	    signal.o sys.o kmod.o workqueue.o pid.o \
 	    rcupdate.o extable.o params.o posix-timers.o \
 	    kthread.o wait.o kfifo.o sys_ni.o posix-cpu-timers.o mutex.o \
-	    hrtimer.o rwsem.o latency.o nsproxy.o srcu.o die_notifier.o
+	    hrtimer.o rwsem.o latency.o nsproxy.o srcu.o die_notifier.o \
+	    rwmutex.o
 

Quoted 4 lines

kernel/Makefile obj-y = sched.o fork.o exec_domain.o · line 15

Quoted 4 lines

kernel/Makefile +0 -0

 obj-$(CONFIG_STACKTRACE) += stacktrace.o
 obj-y += time/

My concern would be the possibility that someone else also woke the sleeper up just as we were preparing to do so. This might happen in the case where multiple readers executed this code simultaneously. Seems to me that this might be vulnerable to the following sequence of events:

1. Reader-wannabe #1 wakes up the writer.

2. The writer wakes up and sees that the sum of the per-cpu
counters is still non-zero, due to the presence of reader-wannabe #2.

3. Reader-wannabe #2 decrements its counter. Note that
reader-wannabe #2 could do all the memory barriers it wanted -- this vulnerability would still exist.

4. Reader-wannabe #2 wakes up the writer, which has no effect,
since the writer is already awake (but is perhaps preempted, interrupted, or something).

5. The writer goes to sleep, never to awaken again (unless some
other reader comes along, which might or might not happen).

Or am I missing something here?

Quoted 20 lines · Paul E. McKenney

My concern would be the possibility that someone else also woke the sleeper up just as we were preparing to do so. This might happen in the case where multiple readers executed this code simultaneously. Seems to me that this might be vulnerable to the following sequence of events:

1. Reader-wannabe #1 wakes up the writer.

2. The writer wakes up and sees that the sum of the per-cpu
counters is still non-zero, due to the presence of reader-wannabe #2.

3. Reader-wannabe #2 decrements its counter. Note that
reader-wannabe #2 could do all the memory barriers it wanted -- this vulnerability would still exist.

4. Reader-wannabe #2 wakes up the writer, which has no effect,
since the writer is already awake (but is perhaps preempted, interrupted, or something).

Quoted 4 lines

5. The writer goes to sleep, never to awaken again (unless some
other reader comes along, which might or might not happen).

Or am I missing something here?

Ugh, nasty. Will have to ponder this a bit. It looks like a spinlock is needed somewhere.