8 files changed, 450 insertions, 35 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 0f10d2ac..4b09bedc 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -51,6 +51,8 @@ set (NNG_SOURCES
     core/protocol.h
     core/socket.c
     core/socket.h
+    core/thread.c
+    core/thread.h
     core/transport.c
     core/transport.h
 
diff --git a/src/core/nng_impl.h b/src/core/nng_impl.h
index 46341863..4c00a522 100644
--- a/src/core/nng_impl.h
+++ b/src/core/nng_impl.h
@@ -31,6 +31,7 @@
 #include "core/panic.h"
 #include "core/platform.h"
 #include "core/protocol.h"
+#include "core/thread.h"
 #include "core/transport.h"
 
 // These have to come after the others - particularly transport.h
diff --git a/src/core/platform.h b/src/core/platform.h
index 21793086..bdec8bb6 100644
--- a/src/core/platform.h
+++ b/src/core/platform.h
@@ -63,41 +63,53 @@ extern void nni_free(void *, size_t);
 typedef struct nni_mutex	nni_mutex;
 typedef struct nni_cond		nni_cond;
 
+typedef struct nni_plat_mtx nni_plat_mtx;
+typedef struct nni_plat_cv nni_plat_cv;
+typedef struct nni_plat_thr nni_plat_thr;
+
 // Mutex handling.
 
 // nni_mutex_init initializes a mutex structure.  This may require dynamic
 // allocation, depending on the platform.  It can return NNG_ENOMEM if that
 // fails.
+extern int nni_plat_mtx_init(nni_plat_mtx *);
 extern int nni_mutex_init(nni_mutex *);
 
 // nni_mutex_fini destroys the mutex and releases any resources allocated for
 // it's use.
 extern void nni_mutex_fini(nni_mutex *);
+extern void nni_plat_mtx_fini(nni_plat_mtx *);
 
 // nni_mutex_enter locks the mutex.  This is not recursive -- a mutex can only
 // be entered once.
 extern void nni_mutex_enter(nni_mutex *);
+extern void nni_plat_mtx_lock(nni_plat_mtx *);
 
 // nni_mutex_exit unlocks the mutex.  This can only be performed by the thread
 // that owned the mutex.
 extern void nni_mutex_exit(nni_mutex *);
+extern void nni_plat_mtx_unlock(nni_plat_mtx *);
 
 // nni_mutex_tryenter tries to lock the mutex.  If it can't, it may return
 // NNG_EBUSY.
 extern int nni_mutex_tryenter(nni_mutex *);
+extern int nni_plat_mtx_trylock(nni_plat_mtx *);
 
 // nni_cond_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.)  This may require
 // dynamic allocation, and if so this operation may fail with NNG_ENOMEM.
 extern int nni_cond_init(nni_cond *, nni_mutex *);
+extern int nni_plat_cv_init(nni_plat_cv *, nni_plat_mtx *);
 
 // nni_cond_fini releases all resources associated with condition variable.
 extern void nni_cond_fini(nni_cond *);
+extern void nni_plat_cv_fini(nni_plat_cv *);
 
 // nni_cond_broadcast wakes all waiters on the condition.  This should be
 // called with the lock held.
 extern void nni_cond_broadcast(nni_cond *);
+extern void nni_plat_cv_wake(nni_plat_cv *);
 
 // nni_cond_signal wakes a signal waiter.
 extern void nni_cond_signal(nni_cond *);
@@ -106,23 +118,27 @@ extern void nni_cond_signal(nni_cond *);
 // associated lock is atomically released and reacquired upon wake up.
 // Callers can be spuriously woken.  The associated lock must be held.
 extern void nni_cond_wait(nni_cond *);
+extern void nni_plat_cv_wait(nni_plat_cv *);
 
 // nni_cond_waituntil waits for a wakeup on the condition variable, or
 // until the system time reaches the specified absolute time.  (It is an
 // absolute form of nni_cond_timedwait.)  Early wakeups are possible, so
 // check the condition.  It will return either NNG_ETIMEDOUT, or 0.
 extern int nni_cond_waituntil(nni_cond *, nni_time);
+extern int nni_plat_cv_until(nni_plat_cv *, nni_time);
 
 typedef struct nni_thread   nni_thread;
 
 // nni_thread_creates a thread that runs the given function. The thread
 // receives a single argument.
 extern int nni_thread_create(nni_thread **, void (*fn)(void *), void *);
+extern int nni_plat_thr_init(nni_plat_thr *, void (*)(void *), void *);
 
 // nni_thread_reap waits for the thread to exit, and then releases any
 // resources associated with the thread.  After this returns, it
 // is an error to reference the thread in any further way.
 extern void nni_thread_reap(nni_thread *);
+extern void nni_plat_thr_fini(nni_plat_thr *);
 
 // nn_clock returns a number of microseconds since some arbitrary time
 // in the past.  The values returned by nni_clock must use the same base
diff --git a/src/core/thread.c b/src/core/thread.c
new file mode 100644
index 00000000..4f714341
--- /dev/null
+++ b/src/core/thread.c
@@ -0,0 +1,150 @@
+//
+// Copyright 2016 Garrett D'Amore <garrett@damore.org>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include "core/nng_impl.h"
+
+int
+nni_mtx_init(nni_mtx *mtx)
+{
+	return (nni_plat_mtx_init(&mtx->mtx));
+}
+
+void
+nni_mtx_fini(nni_mtx *mtx)
+{
+	nni_plat_mtx_fini(&mtx->mtx);
+}
+
+void
+nni_mtx_lock(nni_mtx *mtx)
+{
+	nni_plat_mtx_lock(&mtx->mtx);
+}
+
+void
+nni_mtx_unlock(nni_mtx *mtx)
+{
+	nni_plat_mtx_unlock(&mtx->mtx);
+}
+
+int
+nni_mtx_trylock(nni_mtx *mtx)
+{
+	return (nni_plat_mtx_trylock(&mtx->mtx));
+}
+
+int
+nni_cv_init(nni_cv *cv, nni_mtx *mtx)
+{
+	return (nni_plat_cv_init(&cv->cv, &mtx->mtx));
+}
+
+void
+nni_cv_fini(nni_cv *cv)
+{
+	nni_plat_cv_fini(&cv->cv);
+}
+
+void
+nni_cv_wait(nni_cv *cv)
+{
+	nni_plat_cv_wait(&cv->cv);
+}
+
+int
+nni_cv_until(nni_cv *cv, nni_time until)
+{
+	// Some special cases for times.  Catching these here means that
+	// platforms can assume a valid time is presented to them.
+	if (until == NNI_TIME_NEVER) {
+		nni_plat_cv_wait(&cv->cv);
+		return (0);
+	}
+	if (until == NNI_TIME_ZERO) {
+		return (NNG_EAGAIN);
+	}
+
+	return (nni_plat_cv_until(&cv->cv, until));
+}
+
+void
+nni_cv_wake(nni_cv *cv)
+{
+	return (nni_plat_cv_wake(&cv->cv));
+}
+
+static void
+nni_thr_wrap(void *arg)
+{
+	nni_thr *thr = arg;
+	int stop;
+
+	nni_plat_mtx_lock(&thr->mtx);
+	while (((stop = thr->stop) == 0) && (thr->start == 0)) {
+		nni_plat_cv_wait(&thr->cv);
+	}
+	nni_plat_mtx_unlock(&thr->mtx);
+	if (!stop) {
+		thr->fn(thr->arg);
+	}
+	nni_plat_mtx_lock(&thr->mtx);
+	thr->done = 1;
+	nni_plat_cv_wake(&thr->cv);
+	nni_plat_mtx_unlock(&thr->mtx);
+}
+
+int
+nni_thr_init(nni_thr *thr, nni_thr_func fn, void *arg)
+{
+	int rv;
+
+	thr->done = 0;
+	thr->start = 0;
+	thr->stop = 0;
+	thr->fn = fn;
+	thr->arg = arg;
+
+	if ((rv = nni_plat_mtx_init(&thr->mtx)) != 0) {
+		return (rv);
+	}
+	if ((rv = nni_plat_cv_init(&thr->cv, &thr->mtx)) != 0) {
+		nni_plat_mtx_fini(&thr->mtx);
+		return (rv);
+	}
+	if ((rv = nni_plat_thr_init(&thr->thr, nni_thr_wrap, thr)) != 0) {
+		nni_plat_cv_fini(&thr->cv);
+		nni_plat_mtx_fini(&thr->mtx);
+		return (rv);
+	}
+	return (0);
+}
+
+void
+nni_thr_run(nni_thr *thr)
+{
+	nni_plat_mtx_lock(&thr->mtx);
+	thr->start = 1;
+	nni_plat_cv_wake(&thr->cv);
+	nni_plat_mtx_unlock(&thr->mtx);
+}
+
+void
+nni_thr_fini(nni_thr *thr)
+{
+	nni_plat_mtx_lock(&thr->mtx);
+	thr->stop = 1;
+	nni_plat_cv_wake(&thr->cv);
+	while (!thr->done) {
+		nni_plat_cv_wait(&thr->cv);
+	}
+	nni_plat_mtx_unlock(&thr->mtx);
+	nni_plat_thr_fini(&thr->thr);
+	nni_plat_cv_fini(&thr->cv);
+	nni_plat_mtx_fini(&thr->mtx);
+}
\ No newline at end of file
diff --git a/src/core/thread.h b/src/core/thread.h
new file mode 100644
index 00000000..062941a1
--- /dev/null
+++ b/src/core/thread.h
@@ -0,0 +1,52 @@
+//
+// Copyright 2016 Garrett D'Amore <garrett@damore.org>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#ifndef CORE_THREAD_H
+#define CORE_THREAD_H
+
+#include "core/nng_impl.h"
+
+typedef struct {
+	nni_plat_mtx mtx;
+} nni_mtx;
+
+typedef struct {
+	nni_plat_cv cv;
+} nni_cv;
+
+typedef void (*nni_thr_func)(void *);
+
+typedef struct {
+	nni_plat_thr	thr;
+	nni_plat_mtx	mtx;
+	nni_plat_cv	cv;
+	nni_thr_func	fn;
+	void		*arg;
+	int		start;
+	int		stop;
+	int		done;
+} nni_thr;
+
+extern int nni_mtx_init(nni_mtx *mtx);
+extern void nni_mtx_fini(nni_mtx *mtx);
+extern void nni_mtx_lock(nni_mtx *mtx);
+extern void nni_mtx_unlock(nni_mtx *mtx);
+extern int nni_mtx_trylock(nni_mtx *mtx);
+
+extern int nni_cv_init(nni_cv *cv, nni_mtx *);
+extern void nni_cv_fini(nni_cv *cv);
+extern void nni_cv_wake(nni_cv *cv);
+extern void nni_cv_wait(nni_cv *cv);
+extern int nni_cv_until(nni_cv *cv, nni_time when);
+
+extern int nni_thr_init(nni_thr *thr, nni_thr_func fn, void *arg);
+extern void nni_thr_fini(nni_thr *thr);
+extern void nni_thr_run(nni_thr *thr);
+
+#endif CORE_THREAD_H
\ No newline at end of file
diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h
index 2df96c71..5ca7812f 100644
--- a/src/platform/posix/posix_impl.h
+++ b/src/platform/posix/posix_impl.h
@@ -41,6 +41,23 @@ struct nni_cond {
 	pthread_cond_t		cv;
 	pthread_mutex_t *	mx;
 };
+
+// These types are provided for here, to permit them to be directly inlined
+// elsewhere.
+
+struct nni_plat_mtx {
+	pthread_mutex_t mtx;
+};
+
+struct nni_plat_thr {
+	pthread_t tid;
+};
+
+struct nni_plat_cv {
+	pthread_cond_t	cv;
+	pthread_mutex_t *mtx;
+};
+
 #endif
 
 #endif // PLATFORM_POSIX_IMPL_H
diff --git a/src/platform/posix/posix_thread.c b/src/platform/posix/posix_thread.c
index f5cad7a9..0c928d84 100644
--- a/src/platform/posix/posix_thread.c
+++ b/src/platform/posix/posix_thread.c
@@ -31,6 +31,9 @@ static int nni_plat_inited = 0;
 static int nni_plat_forked = 0;
 static int nni_plat_next = 0;
 
+pthread_condattr_t nni_cvattr;
+pthread_mutexattr_t nni_mxattr;
+
 uint32_t
 nni_plat_nextid(void)
 {
@@ -86,6 +89,172 @@ nni_thread_reap(nni_thread *thr)
 }
 
 
+int
+nni_plat_mtx_init(nni_plat_mtx *mtx)
+{
+	int rv;
+
+	if ((rv = pthread_mutex_init(&mtx->mtx, &nni_mxattr)) != 0) {
+		switch (rv) {
+		case EAGAIN:
+		case ENOMEM:
+			return (NNG_ENOMEM);
+
+		default:
+			nni_panic("pthread_mutex_init: %s", strerror(rv));
+		}
+	}
+	return (0);
+}
+
+
+void
+nni_plat_mtx_fini(nni_plat_mtx *mtx)
+{
+	int rv;
+
+	if ((rv = pthread_mutex_destroy(&mtx->mtx)) != 0) {
+		nni_panic("pthread_mutex_fini: %s", strerror(rv));
+	}
+}
+
+
+void
+nni_plat_mtx_lock(nni_plat_mtx *mtx)
+{
+	int rv;
+
+	if ((rv = pthread_mutex_lock(&mtx->mtx)) != 0) {
+		nni_panic("pthread_mutex_lock: %s", strerror(rv));
+	}
+}
+
+
+void
+nni_plat_mtx_unlock(nni_plat_mtx *mtx)
+{
+	int rv;
+
+	if ((rv = pthread_mutex_unlock(&mtx->mtx)) != 0) {
+		nni_panic("pthread_mutex_unlock: %s", strerror(rv));
+	}
+}
+
+
+int
+nni_plat_mtx_trylock(nni_plat_mtx *mtx)
+{
+	int rv;
+
+	if ((rv = pthread_mutex_trylock(&mtx->mtx)) == EBUSY) {
+		return (NNG_EBUSY);
+	}
+	if (rv != 0) {
+		nni_panic("pthread_mutex_trylock: %s", strerror(rv));
+	}
+	return (0);
+}
+
+
+int
+nni_plat_cv_init(nni_plat_cv *cv, nni_plat_mtx *mtx)
+{
+	int rv;
+
+	if ((rv = pthread_cond_init(&cv->cv, &nni_cvattr)) != 0) {
+		switch (rv) {
+		case ENOMEM:
+		case EAGAIN:
+			return (NNG_ENOMEM);
+
+		default:
+			nni_panic("pthread_cond_init: %s", strerror(rv));
+		}
+	}
+	cv->mtx = &mtx->mtx;
+	return (0);
+}
+
+void
+nni_plat_cv_wake(nni_plat_cv *cv)
+{
+	int rv;
+
+	if ((rv = pthread_cond_broadcast(&cv->cv)) != 0) {
+		nni_panic("pthread_cond_broadcast: %s", strerror(rv));
+	}
+}
+
+void
+nni_plat_cv_wait(nni_plat_cv *cv)
+{
+	int rv;
+
+	if ((rv = pthread_cond_wait(&cv->cv, cv->mtx)) != 0) {
+		nni_panic("pthread_cond_wait: %s", strerror(rv));
+	}
+}
+
+int
+nni_plat_cv_until(nni_plat_cv *cv, nni_time until)
+{
+	struct timespec ts;
+	int rv;
+
+	// Our caller has already guaranteed a sane value for until.
+	ts.tv_sec = until / 1000000;
+	ts.tv_nsec = (until % 1000000) * 1000;
+
+	rv = pthread_cond_timedwait(&cv->cv, cv->mtx, &ts);
+	if (rv == ETIMEDOUT) {
+		if (nni_clock() < until) {
+			// Buggy pthreads implementation!!  Seen with
+			// CLOCK_MONOTONIC on macOS Sierra.
+			nni_panic("nni_plat_cv_until: Premature wake up!");
+		}
+		return (NNG_ETIMEDOUT);
+	} else if (rv != 0) {
+		nni_panic("pthread_cond_timedwait: %d", rv);
+	}
+	return (0);
+
+}
+
+void
+nni_plat_cv_fini(nni_plat_cv *cv)
+{
+	int rv;
+
+	if ((rv = pthread_cond_destroy(&cv->cv)) != 0) {
+		nni_panic("pthread_cond_destroy: %s", strerror(rv));
+	}
+}
+
+int
+nni_plat_thr_init(nni_plat_thr *thr, void (*fn)(void *), void *arg)
+{
+	int rv;
+
+	// POSIX wants functions to return a void *, but we don't care.
+	if ((rv = pthread_create(&thr->tid, NULL, (void *) fn, arg)) != 0) {
+		//nni_printf("pthread_create: %s", strerror(rv));
+		return (NNG_ENOMEM);
+	}
+	return (0);
+}
+
+
+void
+nni_plat_thr_fini(nni_plat_thr *thr)
+{
+	int rv;
+
+	if ((rv = pthread_join(thr->tid, NULL))) {
+		nni_panic("pthread_join: %s", strerror(rv));
+	}
+}
+
+
 void
 nni_atfork_child(void)
 {
@@ -93,8 +262,6 @@ nni_atfork_child(void)
 }
 
 
-pthread_condattr_t nni_cvattr;
-pthread_mutexattr_t nni_mxattr;
 
 int
 nni_plat_init(int (*helper)(void))
diff --git a/tests/platform.c b/tests/platform.c
index 59d8cd6d..eed5adab 100644
--- a/tests/platform.c
+++ b/tests/platform.c
@@ -50,8 +50,8 @@ add(void *arg)
 struct notifyarg {
 	int did;
 	int when;
-	nni_mutex mx;
-	nni_cond cv;
+	nni_mtx mx;
+	nni_cv cv;
 };
 
 void
@@ -60,10 +60,10 @@ notifyafter(void *arg)
 	struct notifyarg *na = arg;
 
 	nni_usleep(na->when);
-	nni_mutex_enter(&na->mx);
+	nni_mtx_lock(&na->mx);
 	na->did = 1;
-	nni_cond_signal(&na->cv);
-	nni_mutex_exit(&na->mx);
+	nni_cv_wake(&na->cv);
+	nni_mtx_unlock(&na->mx);
 }
 
 TestMain("Platform Operations", {
@@ -101,90 +101,100 @@ TestMain("Platform Operations", {
 		})
 	})
 	Convey("Mutexes work", {
-		nni_mutex mx;
+		nni_mtx mx;
 		int rv;
 
-		rv = nni_mutex_init(&mx);
+		rv = nni_mtx_init(&mx);
 		So(rv == 0);
 
 		Convey("We can lock a mutex", {
-			nni_mutex_enter(&mx);
+			nni_mtx_lock(&mx);
 			So(1);
 			Convey("And cannot recursively lock", {
-				rv = nni_mutex_tryenter(&mx);
+				rv = nni_mtx_trylock(&mx);
 				So(rv != 0);
 			})
 			Convey("And we can unlock it", {
-				nni_mutex_exit(&mx);
+				nni_mtx_unlock(&mx);
 				So(1);
 				Convey("And then lock it again", {
-					rv = nni_mutex_tryenter(&mx);
+					rv = nni_mtx_trylock(&mx);
 					So(rv == 0);
 				})
 			})
 		})
 		Convey("We can finalize it", {
-			nni_mutex_fini(&mx);
+			nni_mtx_fini(&mx);
 		})
 	})
 
 	Convey("Threads work", {
-		nni_thread *thr;
+		nni_thr thr;
 		int val = 0;
 		int rv;
 
 		Convey("We can create threads", {
-			rv = nni_thread_create(&thr, add, &val);
+			rv = nni_thr_init(&thr, add, &val);
 			So(rv == 0);
-			So(thr != NULL);
+			nni_thr_run(&thr);
 
 			Convey("It ran", {
 				nni_usleep(50000);	// for context switch
 				So(val == 1);
 			})
 			Convey("We can reap it", {
-				nni_thread_reap(thr);
+				nni_thr_fini(&thr);
 			})
 		})
 	})
 	Convey("Condition variables work", {
 		struct notifyarg arg;
-		nni_thread *thr = NULL;
+		nni_thr thr;
+
+		So(nni_mtx_init(&arg.mx) == 0);
+		So(nni_cv_init(&arg.cv, &arg.mx) == 0);
+		So(nni_thr_init(&thr, notifyafter, &arg) == 0);
 
-		So(nni_mutex_init(&arg.mx) == 0);
-		So(nni_cond_init(&arg.cv, &arg.mx) == 0);
 		Reset({
-			if (thr != NULL) {
-				nni_thread_reap(thr);
-				thr = NULL;
-			}
-			nni_cond_fini(&arg.cv);
-			nni_mutex_fini(&arg.mx);
+			nni_thr_fini(&thr);
+			nni_cv_fini(&arg.cv);
+			nni_mtx_fini(&arg.mx);
 		});
 
 		Convey("Notification works", {
 			arg.did = 0;
 			arg.when = 10000;
-			So(nni_thread_create(&thr, notifyafter, &arg) == 0);
+			nni_thr_run(&thr);
 
-			nni_mutex_enter(&arg.mx);
+			nni_mtx_lock(&arg.mx);
 			if (!arg.did) {
-				nni_cond_wait(&arg.cv);
+				nni_cv_wait(&arg.cv);
 			}
-			nni_mutex_exit(&arg.mx);
+			nni_mtx_unlock(&arg.mx);
 			So(arg.did == 1);
 		})
 
 		Convey("Timeout works", {
 			arg.did = 0;
 			arg.when = 200000;
-			So(nni_thread_create(&thr, notifyafter, &arg) == 0);
-			nni_mutex_enter(&arg.mx);
+			nni_thr_run(&thr);
+			nni_mtx_lock(&arg.mx);
+			if (!arg.did) {
+				nni_cv_until(&arg.cv, nni_clock() + 10000);
+			}
+			So(arg.did == 0);
+			nni_mtx_unlock(&arg.mx);
+		})
+
+		Convey("Not running works", {
+			arg.did = 0;
+			arg.when = 1;
+			nni_mtx_lock(&arg.mx);
 			if (!arg.did) {
-				nni_cond_waituntil(&arg.cv, nni_clock() + 10000);
+				nni_cv_until(&arg.cv, nni_clock() + 10000);
 			}
 			So(arg.did == 0);
-			nni_mutex_exit(&arg.mx);
+			nni_mtx_unlock(&arg.mx);
 		})
 	})
 })
\ No newline at end of file