fixes nanomsg#1409 reader/writer lock desired

This provides the initial implementation, and converts the
transport lookup routines to use it.  This is probably of limited
performance benefit, but rwlock's may be useful in further future work.

src/core/platform.h

@@ -1,5 +1,5 @@
 //
-// Copyright 2020 Staysail Systems, Inc. <[email protected]>
+// Copyright 2021 Staysail Systems, Inc. <[email protected]>
 // Copyright 2018 Capitar IT Group BV <[email protected]>
 // Copyright 2018 Devolutions <[email protected]>
 //
@@ -88,9 +88,11 @@ extern void *nni_zalloc(size_t);
 // Most implementations can just call free() here.
 extern void nni_free(void *, size_t);
 
-typedef struct nni_plat_mtx nni_plat_mtx;
-typedef struct nni_plat_cv  nni_plat_cv;
-typedef struct nni_plat_thr nni_plat_thr;
+typedef struct nni_plat_mtx    nni_plat_mtx;
+typedef struct nni_plat_rwlock nni_plat_rwlock;
+typedef struct nni_plat_cv     nni_plat_cv;
+typedef struct nni_plat_thr    nni_plat_thr;
+typedef struct nni_rwlock      nni_rwlock;
 
 //
 // Threading & Synchronization Support
@@ -112,6 +114,15 @@ extern void nni_plat_mtx_lock(nni_plat_mtx *);
 // thread that owned the mutex.
 extern void nni_plat_mtx_unlock(nni_plat_mtx *);
 
+// read/write locks - these work like mutexes except that multiple readers
+// can acquire the lock.  These are not safe for recursive use, and it is
+// unspecified whether any measures are provided to prevent starvation.
+extern void nni_rwlock_init(nni_rwlock *);
+extern void nni_rwlock_fini(nni_rwlock *);
+extern void nni_rwlock_rdlock(nni_rwlock *);
+extern void nni_rwlock_wrlock(nni_rwlock *);
+extern void nni_rwlock_unlock(nni_rwlock *);
+
 // nni_plat_cv_init initializes a condition variable.  We require a mutex be
 // supplied with it, and that mutex must always be held when performing any
 // operations on the condition variable (other than fini.)  As with mutexes, an
@@ -161,6 +172,7 @@ extern bool nni_plat_thr_is_self(nni_plat_thr *);
 // should be a short ASCII string.  It may or may not be supported --
 // this is intended to facilitate debugging.
 extern void nni_plat_thr_set_name(nni_plat_thr *, const char *);
+
 //
 // Atomics support.  This will evolve over time.
 //

src/platform/posix/posix_impl.h

@@ -1,5 +1,5 @@
 //
-// Copyright 2020 Staysail Systems, Inc. <[email protected]>
+// Copyright 2021 Staysail Systems, Inc. <[email protected]>
 // Copyright 2018 Capitar IT Group BV <[email protected]>
 //
 // This software is supplied under the terms of the MIT License, a
@@ -58,6 +58,10 @@ struct nni_plat_mtx {
 	pthread_mutex_t mtx;
 };
 
+struct nni_rwlock {
+	pthread_rwlock_t rwl;
+};
+
 struct nni_plat_cv {
 	pthread_cond_t cv;
 	nni_plat_mtx * mtx;

src/platform/posix/posix_thread.c

@@ -1,5 +1,5 @@
 //
-// Copyright 2020 Staysail Systems, Inc. <[email protected]>
+// Copyright 2021 Staysail Systems, Inc. <[email protected]>
 // Copyright 2018 Capitar IT Group BV <[email protected]>
 //
 // This software is supplied under the terms of the MIT License, a
@@ -81,7 +81,6 @@ static void
 nni_pthread_mutex_lock(pthread_mutex_t *m)
 {
 	int rv;
-
 	if ((rv = pthread_mutex_lock(m)) != 0) {
 		nni_panic("pthread_mutex_lock: %s", strerror(rv));
 	}
@@ -91,7 +90,6 @@ static void
 nni_pthread_mutex_unlock(pthread_mutex_t *m)
 {
 	int rv;
-
 	if ((rv = pthread_mutex_unlock(m)) != 0) {
 		nni_panic("pthread_mutex_unlock: %s", strerror(rv));
 	}
@@ -101,7 +99,6 @@ static void
 nni_pthread_cond_broadcast(pthread_cond_t *c)
 {
 	int rv;
-
 	if ((rv = pthread_cond_broadcast(c)) != 0) {
 		nni_panic("pthread_cond_broadcast: %s", strerror(rv));
 	}
@@ -155,6 +152,53 @@ nni_plat_mtx_unlock(nni_plat_mtx *mtx)
 	nni_pthread_mutex_unlock(&mtx->mtx);
 }
 
+void
+nni_rwlock_init(nni_rwlock *rwl)
+{
+	while (pthread_rwlock_init(&rwl->rwl, NULL) != 0) {
+		// We must have memory exhaustion -- ENOMEM, or
+		// in some cases EAGAIN.  Wait a bit before we try to
+		// give things a chance to settle down.
+		nni_msleep(10);
+	}
+}
+
+void
+nni_rwlock_fini(nni_rwlock *rwl)
+{
+	int rv;
+	if ((rv = pthread_rwlock_destroy(&rwl->rwl)) != 0) {
+		nni_panic("pthread_rwlock_destroy: %s", strerror(rv));
+	}
+}
+
+void
+nni_rwlock_rdlock(nni_rwlock *rwl)
+{
+	int rv;
+	if ((rv = pthread_rwlock_rdlock(&rwl->rwl)) != 0) {
+		nni_panic("pthread_rwlock_rdlock: %s", strerror(rv));
+	}
+}
+
+void
+nni_rwlock_wrlock(nni_rwlock *rwl)
+{
+	int rv;
+	if ((rv = pthread_rwlock_wrlock(&rwl->rwl)) != 0) {
+		nni_panic("pthread_rwlock_wrlock: %s", strerror(rv));
+	}
+}
+
+void
+nni_rwlock_unlock(nni_rwlock *rwl)
+{
+	int rv;
+	if ((rv = pthread_rwlock_unlock(&rwl->rwl)) != 0) {
+		nni_panic("pthread_rwlock_unlock: %s", strerror(rv));
+	}
+}
+
 void
 nni_plat_cv_init(nni_plat_cv *cv, nni_plat_mtx *mtx)
 {
@@ -264,7 +308,7 @@ nni_plat_thr_set_name(nni_plat_thr *thr, const char *name)
 #if defined(__APPLE__)
 	// Darwin is weird, it can only set the name of pthread_self.
 	if ((thr == NULL) || (pthread_self() == thr->tid)) {
-        	pthread_setname_np(name);
+		pthread_setname_np(name);
 	}
 #elif defined(__NetBSD__)
 	if (thr == NULL) {
@@ -283,7 +327,7 @@ nni_plat_thr_set_name(nni_plat_thr *thr, const char *name)
 	if (thr == NULL) {
 		pthread_set_name_np(pthread_self(), name);
 	} else {
-        	pthread_set_name_np(thr->tid, name);
+		pthread_set_name_np(thr->tid, name);
 	}
 #endif
 }

src/platform/windows/win_impl.h

@@ -41,6 +41,11 @@ struct nni_plat_mtx {
 	int     init;
 };
 
+struct nni_rwlock {
+	SRWLOCK rwl;
+	BOOLEAN exclusive;
+};
+
 struct nni_plat_cv {
 	CONDITION_VARIABLE cv;
 	PSRWLOCK           srl;

src/platform/windows/win_thread.c

@@ -76,6 +76,42 @@ nni_plat_mtx_unlock(nni_plat_mtx *mtx)
 	ReleaseSRWLockExclusive(&mtx->srl);
 }
 
+void
+nni_rwlock_init(nni_rwlock *rwl)
+{
+	InitializeSRWLock(&rwl->rwl);
+}
+
+void
+nni_rwlock_fini(nni_rwlock *rwl)
+{
+	rwl->exclusive = FALSE;
+}
+
+void
+nni_rwlock_rdlock(nni_rwlock *rwl)
+{
+	AcquireSRWLockShared(&rwl->rwl);
+}
+
+void
+nni_rwlock_wrlock(nni_rwlock *rwl)
+{
+	AcquireSRWLockExclusive(&rwl->rwl);
+	rwl->exclusive = TRUE;
+}
+
+void
+nni_rwlock_unlock(nni_rwlock *rwl)
+{
+	if (rwl->exclusive) {
+		rwl->exclusive = FALSE;
+		ReleaseSRWLockExclusive(&rwl->rwl);
+	} else {
+		ReleaseSRWLockShared(&rwl->rwl);
+	}
+}
+
 void
 nni_plat_cv_init(nni_plat_cv *cv, nni_plat_mtx *mtx)
 {
@@ -112,7 +148,7 @@ nni_plat_cv_until(nni_plat_cv *cv, nni_time until)
 	if (now > until) {
 		msec = 0;
 	} else {
-		msec = (DWORD)(until - now);
+		msec = (DWORD) (until - now);
 	}
 
 	ok = SleepConditionVariableSRW(&cv->cv, cv->srl, msec, 0);
@@ -178,7 +214,7 @@ uint64_t
 nni_atomic_get64(nni_atomic_u64 *v)
 {
 
-	return ((uint64_t)(InterlockedExchangeAdd64(&v->v, 0)));
+	return ((uint64_t) (InterlockedExchangeAdd64(&v->v, 0)));
 }
 
 void
@@ -190,7 +226,7 @@ nni_atomic_set64(nni_atomic_u64 *v, uint64_t u)
 uint64_t
 nni_atomic_swap64(nni_atomic_u64 *v, uint64_t u)
 {
-	return ((uint64_t)(InterlockedExchange64(&v->v, (LONGLONG) u)));
+	return ((uint64_t) (InterlockedExchange64(&v->v, (LONGLONG) u)));
 }
 
 void
@@ -213,9 +249,9 @@ uint64_t
 nni_atomic_dec64_nv(nni_atomic_u64 *v)
 {
 #ifdef _WIN64
-	return ((uint64_t)(InterlockedDecrementRelease64(&v->v)));
+	return ((uint64_t) (InterlockedDecrementRelease64(&v->v)));
 #else
-	return ((uint64_t)(InterlockedDecrement64(&v->v)));
+	return ((uint64_t) (InterlockedDecrement64(&v->v)));
 #endif
 }
 
@@ -331,17 +367,17 @@ void
 nni_plat_thr_set_name(nni_plat_thr *thr, const char *name)
 {
 	if (set_thread_desc != NULL) {
-                wchar_t *wcs;
-                size_t len;
-		HANDLE h;
+		wchar_t *wcs;
+		size_t   len;
+		HANDLE   h;
 
 		if (thr == NULL) {
 			h = GetCurrentThread();
 		} else {
 			h = thr->handle;
 		}
 
-                len = strlen(name) + 1;
+		len = strlen(name) + 1;
 		if ((wcs = nni_alloc(len * 2)) == NULL) {
 			return;
 		}
@@ -372,18 +408,18 @@ nni_plat_init(int (*helper)(void))
 		return (0); // fast path
 	}
 
-
-        AcquireSRWLockExclusive(&lock);
+	AcquireSRWLockExclusive(&lock);
 
 	if (!plat_inited) {
-                // Let's look up the function to set thread descriptions.
-                hKernel32 = LoadLibrary(TEXT("kernel32.dll"));
-                if (hKernel32 != NULL) {
-                        set_thread_desc = (pfnSetThreadDescription)
-                            GetProcAddress(hKernel32, "SetThreadDescription");
-                }
-
-                if (((rv = nni_win_io_sysinit()) != 0) ||
+		// Let's look up the function to set thread descriptions.
+		hKernel32 = LoadLibrary(TEXT("kernel32.dll"));
+		if (hKernel32 != NULL) {
+			set_thread_desc =
+			    (pfnSetThreadDescription) GetProcAddress(
+			        hKernel32, "SetThreadDescription");
+		}
+
+		if (((rv = nni_win_io_sysinit()) != 0) ||
 		    ((rv = nni_win_ipc_sysinit()) != 0) ||
 		    ((rv = nni_win_tcp_sysinit()) != 0) ||
 		    ((rv = nni_win_udp_sysinit()) != 0) ||

src/sp/transport.c

@@ -43,9 +43,9 @@ typedef struct nni_sp_transport {
 	nni_list_node t_node;
 } nni_sp_transport;
 
-static nni_list nni_sp_tran_list;
-static nni_mtx  nni_sp_tran_lk;
-static int      nni_sp_tran_inited;
+static nni_list   nni_sp_tran_list;
+static nni_rwlock nni_sp_tran_lk;
+static int        nni_sp_tran_inited;
 
 int
 nni_sp_tran_register(const nni_sp_tran *tran)
@@ -65,32 +65,32 @@ nni_sp_tran_register(const nni_sp_tran *tran)
 		return (NNG_ENOTSUP);
 	}
 
-	nni_mtx_lock(&nni_sp_tran_lk);
+	nni_rwlock_wrlock(&nni_sp_tran_lk);
 	// Check to see if the transport is already registered...
 	NNI_LIST_FOREACH (&nni_sp_tran_list, t) {
 		if (strcmp(tran->tran_scheme, t->t_tran.tran_scheme) == 0) {
 			if (tran->tran_init == t->t_tran.tran_init) {
 				// duplicate.
-				nni_mtx_unlock(&nni_sp_tran_lk);
+				nni_rwlock_unlock(&nni_sp_tran_lk);
 				return (0);
 			}
-			nni_mtx_unlock(&nni_sp_tran_lk);
+			nni_rwlock_unlock(&nni_sp_tran_lk);
 			return (NNG_ESTATE);
 		}
 	}
 	if ((t = NNI_ALLOC_STRUCT(t)) == NULL) {
-		nni_mtx_unlock(&nni_sp_tran_lk);
+		nni_rwlock_unlock(&nni_sp_tran_lk);
 		return (NNG_ENOMEM);
 	}
 
 	t->t_tran = *tran;
 	if ((rv = t->t_tran.tran_init()) != 0) {
-		nni_mtx_unlock(&nni_sp_tran_lk);
+		nni_rwlock_unlock(&nni_sp_tran_lk);
 		NNI_FREE_STRUCT(t);
 		return (rv);
 	}
 	nni_list_append(&nni_sp_tran_list, t);
-	nni_mtx_unlock(&nni_sp_tran_lk);
+	nni_rwlock_unlock(&nni_sp_tran_lk);
 	return (0);
 }
 
@@ -100,14 +100,14 @@ nni_sp_tran_find(nni_url *url)
 	// address is of the form "<scheme>://blah..."
 	nni_sp_transport *t;
 
-	nni_mtx_lock(&nni_sp_tran_lk);
+	nni_rwlock_rdlock(&nni_sp_tran_lk);
 	NNI_LIST_FOREACH (&nni_sp_tran_list, t) {
 		if (strcmp(url->u_scheme, t->t_tran.tran_scheme) == 0) {
-			nni_mtx_unlock(&nni_sp_tran_lk);
+			nni_rwlock_unlock(&nni_sp_tran_lk);
 			return (&t->t_tran);
 		}
 	}
-	nni_mtx_unlock(&nni_sp_tran_lk);
+	nni_rwlock_unlock(&nni_sp_tran_lk);
 	return (NULL);
 }
 
@@ -150,7 +150,7 @@ nni_sp_tran_sys_init(void)
 
 	nni_sp_tran_inited = 1;
 	NNI_LIST_INIT(&nni_sp_tran_list, nni_sp_transport, t_node);
-	nni_mtx_init(&nni_sp_tran_lk);
+	nni_rwlock_init(&nni_sp_tran_lk);
 
 	for (i = 0; nni_sp_tran_ctors[i] != NULL; i++) {
 		int rv;
@@ -174,6 +174,6 @@ nni_sp_tran_sys_fini(void)
 		t->t_tran.tran_fini();
 		NNI_FREE_STRUCT(t);
 	}
-	nni_mtx_fini(&nni_sp_tran_lk);
+	nni_rwlock_fini(&nni_sp_tran_lk);
 	nni_sp_tran_inited = 0;
 }