Replace userland rwlock with a pure kernel based rwlock, the new

implementation does not switch pointers when it resumes waiters.

Asked by: jeff

4 files changed, 199 insertions, 223 deletions

diff --git a/lib/libthr/thread/thr_private.h b/lib/libthr/thread/thr_private.h
index 9b306a0..711ae5e 100644
--- a/lib/libthr/thread/thr_private.h
+++ b/lib/libthr/thread/thr_private.h
@@ -272,12 +272,7 @@ struct pthread_rwlockattr {
 };
 
 struct pthread_rwlock {
-	pthread_mutex_t	lock;	/* monitor lock */
-	pthread_cond_t	read_signal;
-	pthread_cond_t	write_signal;
-	volatile int32_t	state;
-	int		blocked_writers;
-	int		blocked_readers;
+	struct urwlock 	lock;
 	struct pthread	*owner;
 };
 
diff --git a/lib/libthr/thread/thr_rwlock.c b/lib/libthr/thread/thr_rwlock.c
index d52454f..2ed8af8 100644
--- a/lib/libthr/thread/thr_rwlock.c
+++ b/lib/libthr/thread/thr_rwlock.c
@@ -32,15 +32,10 @@
 
 #include "namespace.h"
 #include <pthread.h>
+#include <pthread_np.h>
 #include "un-namespace.h"
 #include "thr_private.h"
 
-#define RWLOCK_WRITE_OWNER	0x80000000U
-#define RWLOCK_WRITE_WAITERS	0x40000000U
-#define RWLOCK_READ_WAITERS	0x20000000U
-#define RWLOCK_MAX_READERS	0x1fffffffU
-#define RWLOCK_READER_COUNT(c)	((c) & RWLOCK_MAX_READERS)
-
 __weak_reference(_pthread_rwlock_destroy, pthread_rwlock_destroy);
 __weak_reference(_pthread_rwlock_init, pthread_rwlock_init);
 __weak_reference(_pthread_rwlock_rdlock, pthread_rwlock_rdlock);
@@ -59,44 +54,12 @@ static int
 rwlock_init(pthread_rwlock_t *rwlock, const pthread_rwlockattr_t *attr __unused)
 {
 	pthread_rwlock_t prwlock;
-	int ret;
-
-	/* allocate rwlock object */
-	prwlock = (pthread_rwlock_t)malloc(sizeof(struct pthread_rwlock));
 
+	prwlock = (pthread_rwlock_t)calloc(1, sizeof(struct pthread_rwlock));
 	if (prwlock == NULL)
 		return (ENOMEM);
-
-	/* initialize the lock */
-	if ((ret = _pthread_mutex_init(&prwlock->lock, NULL)) != 0)
-		free(prwlock);
-	else {
-		/* initialize the read condition signal */
-		ret = _pthread_cond_init(&prwlock->read_signal, NULL);
-
-		if (ret != 0) {
-			_pthread_mutex_destroy(&prwlock->lock);
-			free(prwlock);
-		} else {
-			/* initialize the write condition signal */
-			ret = _pthread_cond_init(&prwlock->write_signal, NULL);
-
-			if (ret != 0) {
-				_pthread_cond_destroy(&prwlock->read_signal);
-				_pthread_mutex_destroy(&prwlock->lock);
-				free(prwlock);
-			} else {
-				/* success */
-				prwlock->state = 0;
-				prwlock->blocked_readers = 0;
-				prwlock->blocked_writers = 0;
-				prwlock->owner = NULL;
-				*rwlock = prwlock;
-			}
-		}
-	}
-
-	return (ret);
+	*rwlock = prwlock;
+	return (0);
 }
 
 int
@@ -112,11 +75,7 @@ _pthread_rwlock_destroy (pthread_rwlock_t *rwlock)
 		prwlock = *rwlock;
 		*rwlock = NULL;
 
-		_pthread_mutex_destroy(&prwlock->lock);
-		_pthread_cond_destroy(&prwlock->read_signal);
-		_pthread_cond_destroy(&prwlock->write_signal);
 		free(prwlock);
-
 		ret = 0;
 	}
 	return (ret);
@@ -146,37 +105,14 @@ _pthread_rwlock_init (pthread_rwlock_t *rwlock, const pthread_rwlockattr_t *attr
 	return (rwlock_init(rwlock, attr));
 }
 
-static inline int
-rwlock_tryrdlock(struct pthread_rwlock *prwlock, int prefer_reader)
-{
-	int32_t state;
-	int32_t wrflags;
- 
-	if (prefer_reader)
-		wrflags = RWLOCK_WRITE_OWNER;
-	else
-		wrflags = RWLOCK_WRITE_OWNER | RWLOCK_WRITE_WAITERS;
-	state = prwlock->state;
-        while (!(state & wrflags)) {
-		if (__predict_false(RWLOCK_READER_COUNT(state) == RWLOCK_MAX_READERS))
-			return (EAGAIN);
-		if (atomic_cmpset_acq_32(&prwlock->state, state, state + 1))
-			return (0);
-		CPU_SPINWAIT;
-		state = prwlock->state;
-	}
-
-	return (EBUSY);
-}
-
 static int
 rwlock_rdlock_common(pthread_rwlock_t *rwlock, const struct timespec *abstime)
 {
 	struct pthread *curthread = _get_curthread();
-	const int prefer_read = curthread->rdlock_count > 0;
 	pthread_rwlock_t prwlock;
-	int ret, wrflags, old;
-	int32_t state;
+	struct timespec ts, ts2, *tsp;
+	int flags;
+	int ret;
 
 	if (__predict_false(rwlock == NULL))
 		return (EINVAL);
@@ -191,23 +127,7 @@ rwlock_rdlock_common(pthread_rwlock_t *rwlock, const struct timespec *abstime)
 		prwlock = *rwlock;
 	}
 
-	/*
-	 * POSIX said the validity of the abstimeout parameter need
-	 * not be checked if the lock can be immediately acquired.
-	 */
-	ret = rwlock_tryrdlock(prwlock, prefer_read);
-	if (ret == 0) {
-		curthread->rdlock_count++;
-		return (ret);
-	}
-	if (__predict_false(ret == EAGAIN))
-		return (ret);
-
-	if (__predict_false(abstime && 
-		(abstime->tv_nsec >= 1000000000 || abstime->tv_nsec < 0)))
-		return (EINVAL);
-
-	if (prefer_read) {
+	if (curthread->rdlock_count) {
 		/*
 		 * To avoid having to track all the rdlocks held by
 		 * a thread or all of the threads that hold a rdlock,
@@ -220,50 +140,47 @@ rwlock_rdlock_common(pthread_rwlock_t *rwlock, const struct timespec *abstime)
 		 * when it already has one or more rdlocks avoids the
 		 * deadlock.  I hope the reader can follow that logic ;-)
 		 */
+		flags = URWLOCK_PREFER_READER;
+	} else {
+		flags = 0;
+	}
 
-		wrflags = RWLOCK_WRITE_OWNER;
-	} else
-		wrflags = RWLOCK_WRITE_OWNER | RWLOCK_WRITE_WAITERS;
+	/*
+	 * POSIX said the validity of the abstimeout parameter need
+	 * not be checked if the lock can be immediately acquired.
+	 */
+	ret = _thr_rwlock_tryrdlock(&prwlock->lock, flags);
+	if (ret == 0) {
+		curthread->rdlock_count++;
+		return (ret);
+	}
 
-	/* reset to zero */
-	ret = 0;
-	for (;;) {
-		_pthread_mutex_lock(&prwlock->lock);
-		state = prwlock->state;
-		/* set read contention bit */
-		while ((state & wrflags) && !(state & RWLOCK_READ_WAITERS)) {
-			if (atomic_cmpset_acq_32(&prwlock->state, state, state | RWLOCK_READ_WAITERS))
-				break;
-			CPU_SPINWAIT;
-			state = prwlock->state;
-		}
- 
-		atomic_add_32(&prwlock->blocked_readers, 1);
-		if (state & wrflags) {
-			ret = _pthread_cond_wait_unlocked(&prwlock->read_signal, &prwlock->lock, abstime);
-			old = atomic_fetchadd_32(&prwlock->blocked_readers, -1);
-			if (old == 1)
-				_pthread_mutex_lock(&prwlock->lock);
-			else
-				goto try_it;
-		} else {
-			atomic_subtract_32(&prwlock->blocked_readers, 1);
-		}
+	if (__predict_false(abstime && 
+		(abstime->tv_nsec >= 1000000000 || abstime->tv_nsec < 0)))
+		return (EINVAL);
 
-		if (prwlock->blocked_readers == 0)
-			atomic_clear_32(&prwlock->state, RWLOCK_READ_WAITERS);
-		_pthread_mutex_unlock(&prwlock->lock);
+	for (;;) {
+		if (abstime) {
+			clock_gettime(CLOCK_REALTIME, &ts);
+			TIMESPEC_SUB(&ts2, abstime, &ts);
+			if (ts2.tv_sec < 0 || 
+			    (ts2.tv_sec == 0 && ts2.tv_nsec <= 0))
+				return (ETIMEDOUT);
+			tsp = &ts2;
+		} else
+			tsp = NULL;
+
+		/* goto kernel and lock it */
+		ret = __thr_rwlock_rdlock(&prwlock->lock, flags, tsp);
+		if (ret != EINTR)
+			break;
 
-try_it:
-		/* try to lock it again. */
-		if (rwlock_tryrdlock(prwlock, prefer_read) == 0) {
-			curthread->rdlock_count++;
+		/* if interrupted, try to lock it in userland again. */
+		if (_thr_rwlock_tryrdlock(&prwlock->lock, flags) == 0) {
 			ret = 0;
+			curthread->rdlock_count++;
 			break;
 		}
-
-		if (ret)
-			break;
 	}
 	return (ret);
 }
@@ -286,6 +203,7 @@ _pthread_rwlock_tryrdlock (pthread_rwlock_t *rwlock)
 {
 	struct pthread *curthread = _get_curthread();
 	pthread_rwlock_t prwlock;
+	int flags;
 	int ret;
 
 	if (__predict_false(rwlock == NULL))
@@ -301,27 +219,30 @@ _pthread_rwlock_tryrdlock (pthread_rwlock_t *rwlock)
 		prwlock = *rwlock;
 	}
 
-	ret = rwlock_tryrdlock(prwlock, curthread->rdlock_count > 0);
+	if (curthread->rdlock_count) {
+		/*
+		 * To avoid having to track all the rdlocks held by
+		 * a thread or all of the threads that hold a rdlock,
+		 * we keep a simple count of all the rdlocks held by
+		 * a thread.  If a thread holds any rdlocks it is
+		 * possible that it is attempting to take a recursive
+		 * rdlock.  If there are blocked writers and precedence
+		 * is given to them, then that would result in the thread
+		 * deadlocking.  So allowing a thread to take the rdlock
+		 * when it already has one or more rdlocks avoids the
+		 * deadlock.  I hope the reader can follow that logic ;-)
+		 */
+		flags = URWLOCK_PREFER_READER;
+	} else {
+		flags = 0;
+	}
+
+	ret = _thr_rwlock_tryrdlock(&prwlock->lock, flags);
 	if (ret == 0)
 		curthread->rdlock_count++;
 	return (ret);
 }
 
-static inline int
-rwlock_trywrlock(struct pthread_rwlock *prwlock)
-{
-	int32_t state;
-
-	state = prwlock->state;
-	while (!(state & RWLOCK_WRITE_OWNER) && RWLOCK_READER_COUNT(state) == 0) {
-		if (atomic_cmpset_acq_32(&prwlock->state, state, state | RWLOCK_WRITE_OWNER))
-			return (0);
-		CPU_SPINWAIT;
-		state = prwlock->state;
-	}
-	return (EBUSY);
-}
-
 int
 _pthread_rwlock_trywrlock (pthread_rwlock_t *rwlock)
 {
@@ -342,7 +263,7 @@ _pthread_rwlock_trywrlock (pthread_rwlock_t *rwlock)
 		prwlock = *rwlock;
 	}
 
-	ret = rwlock_trywrlock(prwlock);
+	ret = _thr_rwlock_trywrlock(&prwlock->lock);
 	if (ret == 0)
 		prwlock->owner = curthread;
 	return (ret);
@@ -353,8 +274,8 @@ rwlock_wrlock_common (pthread_rwlock_t *rwlock, const struct timespec *abstime)
 {
 	struct pthread *curthread = _get_curthread();
 	pthread_rwlock_t prwlock;
+	struct timespec ts, ts2, *tsp;
 	int ret;
-	int32_t state;
 
 	if (__predict_false(rwlock == NULL))
 		return (EINVAL);
@@ -373,9 +294,7 @@ rwlock_wrlock_common (pthread_rwlock_t *rwlock, const struct timespec *abstime)
 	 * POSIX said the validity of the abstimeout parameter need
 	 * not be checked if the lock can be immediately acquired.
 	 */
-
-	/* try to lock it in userland */
-	ret = rwlock_trywrlock(prwlock);
+	ret = _thr_rwlock_trywrlock(&prwlock->lock);
 	if (ret == 0) {
 		prwlock->owner = curthread;
 		return (ret);
@@ -385,46 +304,33 @@ rwlock_wrlock_common (pthread_rwlock_t *rwlock, const struct timespec *abstime)
 		(abstime->tv_nsec >= 1000000000 || abstime->tv_nsec < 0)))
 		return (EINVAL);
 
-	/* reset to zero */
-	ret = 0;
-
 	for (;;) {
-		_pthread_mutex_lock(&prwlock->lock);
-		state = prwlock->state;
-		while (((state & RWLOCK_WRITE_OWNER) || RWLOCK_READER_COUNT(state) != 0) &&
-			(state & RWLOCK_WRITE_WAITERS) == 0) {
-			if (atomic_cmpset_acq_32(&prwlock->state, state, state | RWLOCK_WRITE_WAITERS))
-				break;
-			CPU_SPINWAIT;
-			state = prwlock->state;
-		}
-
-		prwlock->blocked_writers++;
-
-		while ((state & RWLOCK_WRITE_OWNER) || RWLOCK_READER_COUNT(state) != 0) {
-			if (abstime == NULL)
-				ret = _pthread_cond_wait(&prwlock->write_signal, &prwlock->lock);
-			else
-				ret = _pthread_cond_timedwait(&prwlock->write_signal, &prwlock->lock, abstime);
-
-			if (ret)
-				break;
-			state = prwlock->state;
+		if (abstime != NULL) {
+			clock_gettime(CLOCK_REALTIME, &ts);
+			TIMESPEC_SUB(&ts2, abstime, &ts);
+			if (ts2.tv_sec < 0 || 
+			    (ts2.tv_sec == 0 && ts2.tv_nsec <= 0))
+				return (ETIMEDOUT);
+			tsp = &ts2;
+		} else
+			tsp = NULL;
+
+		/* goto kernel and lock it */
+		ret = __thr_rwlock_wrlock(&prwlock->lock, tsp);
+		if (ret == 0) {
+			prwlock->owner = curthread;
+			break;
 		}
 
-		prwlock->blocked_writers--;
-		if (prwlock->blocked_writers == 0)
-			atomic_clear_32(&prwlock->state, RWLOCK_WRITE_WAITERS);
-		_pthread_mutex_unlock(&prwlock->lock);
+		if (ret != EINTR)
+			break;
 
-		if (rwlock_trywrlock(prwlock) == 0) {
-			prwlock->owner = curthread;
+		/* if interrupted, try to lock it in userland again. */
+		if (_thr_rwlock_trywrlock(&prwlock->lock) == 0) {
 			ret = 0;
+			prwlock->owner = curthread;
 			break;
 		}
-
-		if (ret)
-			break;
 	}
 	return (ret);
 }
@@ -447,6 +353,7 @@ _pthread_rwlock_unlock (pthread_rwlock_t *rwlock)
 {
 	struct pthread *curthread = _get_curthread();
 	pthread_rwlock_t prwlock;
+	int ret;
 	int32_t state;
 
 	if (__predict_false(rwlock == NULL))
@@ -457,47 +364,16 @@ _pthread_rwlock_unlock (pthread_rwlock_t *rwlock)
 	if (__predict_false(prwlock == NULL))
 		return (EINVAL);
 
-	state = prwlock->state;
-
-	if (state & RWLOCK_WRITE_OWNER) {
+	state = prwlock->lock.rw_state;
+	if (state & URWLOCK_WRITE_OWNER) {
 		if (__predict_false(prwlock->owner != curthread))
 			return (EPERM);
 		prwlock->owner = NULL;
-		while (!atomic_cmpset_rel_32(&prwlock->state, state, state & ~RWLOCK_WRITE_OWNER)) {
-			CPU_SPINWAIT;
-			state = prwlock->state;
-		}
-	} else if (RWLOCK_READER_COUNT(state) != 0) {
-		while (!atomic_cmpset_rel_32(&prwlock->state, state, state - 1)) {
-			CPU_SPINWAIT;
-			state = prwlock->state;
-			if (RWLOCK_READER_COUNT(state) == 0)
-				return (EPERM);
-		}
-		curthread->rdlock_count--;
-        } else {
-		return (EPERM);
 	}
 
-#if 1
-	if (state & RWLOCK_WRITE_WAITERS) {
-		_pthread_mutex_lock(&prwlock->lock);
-		_pthread_cond_signal(&prwlock->write_signal);
-		_pthread_mutex_unlock(&prwlock->lock);
-	} else if (state & RWLOCK_READ_WAITERS) {
-		_pthread_mutex_lock(&prwlock->lock);
-		_pthread_cond_broadcast(&prwlock->read_signal);
-		_pthread_mutex_unlock(&prwlock->lock);
-	}
-#else
-
-	if (state & RWLOCK_WRITE_WAITERS) {
-		_pthread_mutex_lock(&prwlock->lock);
-		_pthread_cond_broadcast_unlock(&prwlock->write_signal, &prwlock->lock, 0);
-	} else if (state & RWLOCK_READ_WAITERS) {
-		_pthread_mutex_lock(&prwlock->lock);
-		_pthread_cond_broadcast_unlock(&prwlock->write_signal, &prwlock->lock, 1);
-	}
-#endif
-	return (0);
+	ret = _thr_rwlock_unlock(&prwlock->lock);
+	if (ret == 0 && (state & URWLOCK_WRITE_OWNER) == 0)
+		curthread->rdlock_count--;
+
+	return (ret);
 }
diff --git a/lib/libthr/thread/thr_umtx.c b/lib/libthr/thread/thr_umtx.c
index 058205b..d0a225d 100644
--- a/lib/libthr/thread/thr_umtx.c
+++ b/lib/libthr/thread/thr_umtx.c
@@ -159,3 +159,27 @@ _thr_ucond_broadcast(struct ucond *cv)
 		return (0);
 	return (errno);
 }
+
+int
+__thr_rwlock_rdlock(struct urwlock *rwlock, int flags, struct timespec *tsp)
+{
+	if (_umtx_op(rwlock, UMTX_OP_RW_RDLOCK, flags, NULL, tsp) != -1)
+		return (0);
+	return (errno);
+}
+
+int
+__thr_rwlock_wrlock(struct urwlock *rwlock, struct timespec *tsp)
+{
+	if (_umtx_op(rwlock, UMTX_OP_RW_WRLOCK, 0, NULL, tsp) != -1)
+		return (0);
+	return (errno);
+}
+
+int
+__thr_rwlock_unlock(struct urwlock *rwlock)
+{
+	if (_umtx_op(rwlock, UMTX_OP_RW_UNLOCK, 0, NULL, NULL) != -1)
+		return (0);
+	return (errno);
+}
diff --git a/lib/libthr/thread/thr_umtx.h b/lib/libthr/thread/thr_umtx.h
index 65d00f2..752e7b2 100644
--- a/lib/libthr/thread/thr_umtx.h
+++ b/lib/libthr/thread/thr_umtx.h
@@ -54,6 +54,10 @@ void _thr_ucond_init(struct ucond *cv) __hidden;
 int _thr_ucond_signal(struct ucond *cv) __hidden;
 int _thr_ucond_broadcast(struct ucond *cv) __hidden;
 
+int __thr_rwlock_rdlock(struct urwlock *rwlock, int flags, struct timespec *tsp) __hidden;
+int __thr_rwlock_wrlock(struct urwlock *rwlock, struct timespec *tsp) __hidden;
+int __thr_rwlock_unlock(struct urwlock *rwlock) __hidden;
+
 static inline int
 _thr_umutex_trylock(struct umutex *mtx, uint32_t id)
 {
@@ -97,4 +101,81 @@ _thr_umutex_unlock(struct umutex *mtx, uint32_t id)
     return (__thr_umutex_unlock(mtx));
 }
 
+static inline int
+_thr_rwlock_tryrdlock(struct urwlock *rwlock, int flags)
+{
+	int32_t state;
+	int32_t wrflags;
+
+	if (flags & URWLOCK_PREFER_READER || rwlock->rw_flags & URWLOCK_PREFER_READER)
+		wrflags = URWLOCK_WRITE_OWNER;
+	else
+		wrflags = URWLOCK_WRITE_OWNER | URWLOCK_WRITE_WAITERS;
+	state = rwlock->rw_state;
+	while (!(state & wrflags)) {
+		if (__predict_false(URWLOCK_READER_COUNT(state) == URWLOCK_MAX_READERS))
+			return (EAGAIN);
+		if (atomic_cmpset_acq_32(&rwlock->rw_state, state, state + 1))
+			return (0);
+		state = rwlock->rw_state;
+	}
+
+	return (EBUSY);
+}
+
+static inline int
+_thr_rwlock_trywrlock(struct urwlock *rwlock)
+{
+	int32_t state;
+
+	state = rwlock->rw_state;
+	while (!(state & URWLOCK_WRITE_OWNER) && URWLOCK_READER_COUNT(state) == 0) {
+		if (atomic_cmpset_acq_32(&rwlock->rw_state, state, state | URWLOCK_WRITE_OWNER))
+			return (0);
+		state = rwlock->rw_state;
+	}
+
+	return (EBUSY);
+}
+
+static inline int
+_thr_rwlock_rdlock(struct urwlock *rwlock, int flags, struct timespec *tsp)
+{
+	if (_thr_rwlock_tryrdlock(rwlock, flags) == 0)
+		return (0);
+	return (__thr_rwlock_rdlock(rwlock, flags, tsp));
+}
+
+static inline int
+_thr_rwlock_wrlock(struct urwlock *rwlock, struct timespec *tsp)
+{
+	if (_thr_rwlock_trywrlock(rwlock) == 0)
+		return (0);
+	return (__thr_rwlock_wrlock(rwlock, tsp));
+}
+
+static inline int
+_thr_rwlock_unlock(struct urwlock *rwlock)
+{
+	int32_t state;
+
+	state = rwlock->rw_state;
+	if (state & URWLOCK_WRITE_OWNER) {
+		if (atomic_cmpset_rel_32(&rwlock->rw_state, URWLOCK_WRITE_OWNER, 0))
+			return (0);
+	} else {
+		for (;;) {
+			if (__predict_false(URWLOCK_READER_COUNT(state) == 0))
+				return (EPERM);
+			if (!((state & URWLOCK_WRITE_WAITERS) && URWLOCK_READER_COUNT(state) == 1)) {
+				if (atomic_cmpset_rel_32(&rwlock->rw_state, state, state-1))
+					return (0);
+				state = rwlock->rw_state;
+			} else {
+				break;
+			}
+		}
+    	}
+    	return (__thr_rwlock_unlock(rwlock));
+}
 #endif