Use a thread pool to process notification if sigev_notify_attributes

is default and caller does not require dedicated thread. timer needs
a dedicated thread to maintain overrun count correctly in notification
context. mqueue and aio can use thread pool to do notification
concurrently, the thread pool has lifecycle control, some threads will
exit if they have idled for a while.

1 files changed, 306 insertions, 48 deletions

diff --git a/lib/librt/sigev_thread.c b/lib/librt/sigev_thread.c
index f942e57..ea86c60 100644
--- a/lib/librt/sigev_thread.c
+++ b/lib/librt/sigev_thread.c
@@ -31,9 +31,12 @@
 #include <machine/atomic.h>
 
 #include "namespace.h"
+#include <assert.h>
 #include <err.h>
+#include <errno.h>
 #include <ucontext.h>
 #include <sys/thr.h>
+#include <sys/time.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -43,23 +46,60 @@
 
 #include "sigev_thread.h"
 
+/* Lowest number of worker threads should be kept. */
+#define SIGEV_WORKER_LOW		0
+
+/* Highest number of worker threads can be created. */
+#define	SIGEV_WORKER_HIGH		20
+
+/* How long an idle worker thread should stay. */
+#define SIGEV_WORKER_IDLE		10
+
+struct sigev_worker {
+	LIST_ENTRY(sigev_worker)	sw_link;
+	pthread_cond_t		sw_cv;
+	struct sigev_node	*sw_sn;
+	int			sw_flags;
+	int			*sw_readyptr;
+};
+
+#define	SWF_READYQ		1
+
 LIST_HEAD(sigev_list_head, sigev_node);
 #define HASH_QUEUES		17
 #define	HASH(t, id)		((((id) << 3) + (t)) % HASH_QUEUES)
+
 static struct sigev_list_head	sigev_hash[HASH_QUEUES];
 static struct sigev_list_head	sigev_all;
+static TAILQ_HEAD(, sigev_node)	sigev_actq;
+static TAILQ_HEAD(, sigev_thread_node)	sigev_threads;
 static int			sigev_generation;
 static pthread_mutex_t		*sigev_list_mtx;
-static TAILQ_HEAD(,sigev_thread_node)	sigev_threads;
+static pthread_once_t		sigev_once = PTHREAD_ONCE_INIT;
+static pthread_once_t		sigev_once_default = PTHREAD_ONCE_INIT;
 static pthread_mutex_t		*sigev_threads_mtx;
+static pthread_cond_t		*sigev_threads_cv;
+static pthread_cond_t		*sigev_actq_cv;
+static struct sigev_thread_node	*sigev_default_thread;
+static struct sigev_thread_attr	sigev_default_sna;
 static pthread_attr_t		sigev_default_attr;
-static pthread_once_t		sigev_once = PTHREAD_ONCE_INIT;
+static int			atfork_registered;
+static LIST_HEAD(,sigev_worker)	sigev_worker_ready;
+static int			sigev_worker_count;
+static int			sigev_worker_start;
+static int			sigev_worker_high;
+static int			sigev_worker_low;
+static pthread_cond_t		*sigev_worker_init_cv;
 
 static void __sigev_fork_prepare(void);
 static void __sigev_fork_parent(void);
 static void __sigev_fork_child(void);
-static struct sigev_thread_node *sigev_thread_create(pthread_attr_t *);
+static struct sigev_thread_node *sigev_thread_create(pthread_attr_t *,
+	struct sigev_thread_node *, int);
 static void *sigev_service_loop(void *);
+static void *sigev_worker_routine(void *);
+static void sigev_put(struct sigev_node *);
+static void worker_cleanup(void *arg);
 
 #pragma weak pthread_create
 #pragma weak pthread_attr_getschedpolicy
@@ -87,7 +127,6 @@ attr2sna(pthread_attr_t *attr, struct sigev_thread_attr *sna)
 	pthread_attr_getinheritsched(attr, &sna->sna_inherit);
 	pthread_attr_getschedparam(attr, &sched_param);
 	sna->sna_prio = sched_param.sched_priority;
-	pthread_attr_getscope(attr, &sna->sna_scope);
 	pthread_attr_getstacksize(attr, &sna->sna_stacksize);
 	pthread_attr_getstackaddr(attr, &sna->sna_stackaddr);
 	pthread_attr_getguardsize(attr, &sna->sna_guardsize);
@@ -102,32 +141,53 @@ sna_eq(const struct sigev_thread_attr *a, const struct sigev_thread_attr *b)
 static __inline int
 have_threads(void)
 {
-	return (pthread_create != NULL);
+	return (&pthread_create != NULL);
 }
 
 void
 __sigev_thread_init(void)
 {
-	static int notfirst = 0;
+	static int inited = 0;
 	int i;
 
 	sigev_list_mtx = malloc(sizeof(pthread_mutex_t));
 	_pthread_mutex_init(sigev_list_mtx, NULL);
 	sigev_threads_mtx = malloc(sizeof(pthread_mutex_t));
 	_pthread_mutex_init(sigev_threads_mtx, NULL);
+	sigev_threads_cv = malloc(sizeof(pthread_cond_t));
+	_pthread_cond_init(sigev_threads_cv, NULL);
+	sigev_actq_cv = malloc(sizeof(pthread_cond_t));
+	_pthread_cond_init(sigev_actq_cv, NULL);
 	for (i = 0; i < HASH_QUEUES; ++i)
 		LIST_INIT(&sigev_hash[i]);
 	LIST_INIT(&sigev_all);
 	TAILQ_INIT(&sigev_threads);
-	if (!notfirst) {
+	TAILQ_INIT(&sigev_actq);
+	sigev_default_thread = NULL;
+	sigev_worker_count = 0;
+	sigev_worker_start = 0;
+	LIST_INIT(&sigev_worker_ready);
+	sigev_worker_high = SIGEV_WORKER_HIGH;
+	sigev_worker_low = SIGEV_WORKER_LOW;
+	sigev_worker_init_cv = malloc(sizeof(pthread_cond_t));
+	_pthread_cond_init(sigev_worker_init_cv, NULL);
+	if (atfork_registered == 0) {
+		pthread_atfork(
+			__sigev_fork_prepare,
+			__sigev_fork_parent,
+			__sigev_fork_child);
+		atfork_registered = 1;
+	}
+	if (!inited) {
 		pthread_attr_init(&sigev_default_attr);
-		pthread_attr_setscope(&sigev_default_attr, PTHREAD_SCOPE_SYSTEM);
+		attr2sna(&sigev_default_attr, &sigev_default_sna);
+		pthread_attr_setscope(&sigev_default_attr,
+			PTHREAD_SCOPE_SYSTEM);
 		pthread_attr_setdetachstate(&sigev_default_attr,
 			PTHREAD_CREATE_DETACHED);
-		pthread_atfork(__sigev_fork_prepare, __sigev_fork_parent,
-			__sigev_fork_child);
-		notfirst = 1;
+		inited = 1;
 	}
+	sigev_default_thread = sigev_thread_create(NULL, NULL, 0);
 }
 
 int
@@ -137,24 +197,29 @@ __sigev_check_init(void)
 		return (-1);
 
 	_pthread_once(&sigev_once, __sigev_thread_init);
-	return (0);
+	return (sigev_default_thread != NULL) ? 0 : -1;
 }
 
-void
+static void
 __sigev_fork_prepare(void)
 {
-	__sigev_thread_list_lock();
 }
 
-void
+static void
 __sigev_fork_parent(void)
 {
-	__sigev_thread_list_unlock();
 }
 
-void
+static void
 __sigev_fork_child(void)
 {
+	/*
+	 * This is a hack, the thread libraries really should
+	 * check if the handlers were already registered in
+	 * pthread_atfork().
+	 */
+	atfork_registered = 1;
+	memcpy(&sigev_once, &sigev_once_default, sizeof(sigev_once));
 	__sigev_thread_init();
 }
 
@@ -183,7 +248,8 @@ __sigev_thread_list_unlock(void)
 }
 
 struct sigev_node *
-__sigev_alloc(int type, const struct sigevent *evp)
+__sigev_alloc(int type, const struct sigevent *evp, struct sigev_node *prev,
+	int usethreadpool)
 {
 	struct sigev_node *sn;
 
@@ -193,7 +259,10 @@ __sigev_alloc(int type, const struct sigevent *evp)
 		sn->sn_func = evp->sigev_notify_function;
 		sn->sn_gen = atomic_fetchadd_int(&sigev_generation, 1);
 		sn->sn_type = type;
-		sn->sn_tn = sigev_thread_create(evp->sigev_notify_attributes);
+		if (usethreadpool)
+			sn->sn_flags |= SNF_THREADPOOL;
+		sn->sn_tn = sigev_thread_create(evp->sigev_notify_attributes,
+			prev ? prev->sn_tn : NULL, usethreadpool);
 		if (sn->sn_tn == NULL) {
 			free(sn);
 			sn = NULL;
@@ -262,10 +331,19 @@ __sigev_delete_node(struct sigev_node *sn)
 	LIST_REMOVE(sn, sn_link);
 	LIST_REMOVE(sn, sn_allist);
 
+	__sigev_thread_list_lock();
+	if (--sn->sn_tn->tn_refcount == 0)
+		if (!(sn->sn_flags & SNF_THREADPOOL))
+			pthread_kill(sn->sn_tn->tn_thread, SIGEV_SIGSERVICE);
+	__sigev_thread_list_unlock();
 	if (sn->sn_flags & SNF_WORKING)
 		sn->sn_flags |= SNF_REMOVED;
-	else
+	else {
+		if (sn->sn_flags & SNF_ACTQ) {
+			TAILQ_REMOVE(&sigev_actq, sn, sn_actq);
+		}
 		__sigev_free(sn);
+	}
 	return (0);
 }
 
@@ -280,15 +358,13 @@ sigev_get_id(siginfo_t *si)
 		return (si->si_mqd);
 	case SI_ASYNCIO:
 		return (sigev_id_t)si->si_value.sival_ptr;
-	default:
-		warnx("%s %s : unknown si_code %d\n", __FILE__, __func__,
-			si->si_code);
 	}
 	return (-1);
 }
 
 static struct sigev_thread_node *
-sigev_thread_create(pthread_attr_t *pattr)
+sigev_thread_create(pthread_attr_t *pattr, struct sigev_thread_node *prev,
+	int usepool)
 {
 	struct sigev_thread_node *tn;
 	struct sigev_thread_attr sna;
@@ -305,43 +381,54 @@ sigev_thread_create(pthread_attr_t *pattr)
 	attr2sna(pattr, &sna);
 
 	__sigev_thread_list_lock();
-	/* Search a thread matching the required pthread_attr. */
-	TAILQ_FOREACH(tn, &sigev_threads, tn_link) {
-		if (sna.sna_stackaddr == NULL) {
-			if (sna_eq(&tn->tn_sna, &sna))
-				break;
-		} else {
-			/*
-			 * Reuse the thread if it has same stack address,
-			 * because two threads can not run on same stack.
-			 */
+
+	if (prev != NULL && sna_eq(&prev->tn_sna, &sna)) {
+		prev->tn_refcount++;
+		__sigev_thread_list_unlock();
+		return (prev);
+	}
+
+	if (sna_eq(&sna, &sigev_default_sna) && usepool &&
+	    sigev_default_thread != NULL) {
+		sigev_default_thread->tn_refcount++;
+		__sigev_thread_list_unlock();
+		return (sigev_default_thread);
+	}
+
+	tn = NULL;
+	/* Search a thread matching the required stack address */
+	if (sna.sna_stackaddr != NULL) {
+		TAILQ_FOREACH(tn, &sigev_threads, tn_link) {
 			if (sna.sna_stackaddr == tn->tn_sna.sna_stackaddr)
 				break;
 		}
 	}
+
 	if (tn != NULL) {
+		tn->tn_refcount++;
 		__sigev_thread_list_unlock();
 		return (tn);
 	}
 	tn = malloc(sizeof(*tn));
 	tn->tn_sna = sna;
 	tn->tn_cur = NULL;
+	tn->tn_lwpid = -1;
+	tn->tn_refcount = 1;
 	TAILQ_INSERT_TAIL(&sigev_threads, tn, tn_link);
 	sigemptyset(&set);
 	sigaddset(&set, SIGEV_SIGSERVICE);
 	_sigprocmask(SIG_BLOCK, &set, NULL);
-	_pthread_cond_init(&tn->tn_cv, NULL);
 	ret = pthread_create(&tn->tn_thread, pattr, sigev_service_loop, tn);
 	_sigprocmask(SIG_UNBLOCK, &set, NULL);
 	if (ret != 0) {
 		TAILQ_REMOVE(&sigev_threads, tn, tn_link);
 		__sigev_thread_list_unlock();
-		_pthread_cond_destroy(&tn->tn_cv);
 		free(tn);
 		tn = NULL;
 	} else {
 		/* wait the thread to get its lwpid */
-		_pthread_cond_wait(&tn->tn_cv, sigev_threads_mtx);
+		while (tn->tn_lwpid == -1)
+			_pthread_cond_wait(sigev_threads_cv, sigev_threads_mtx);
 		__sigev_thread_list_unlock();
 	}
 	return (tn);
@@ -362,6 +449,7 @@ after_dispatch(struct sigev_thread_node *tn)
 		tn->tn_cur = NULL;
 		__sigev_list_unlock();
 	}
+	tn->tn_cur = NULL;
 }
 
 /*
@@ -380,6 +468,11 @@ thread_cleanup(void *arg)
 	abort();
 }
 
+/*
+ * Main notification dispatch function, the function either
+ * run user callback by itself or hand off the notifications
+ * to worker threads depend on flags.
+ */
 static void *
 sigev_service_loop(void *arg)
 {
@@ -388,12 +481,13 @@ sigev_service_loop(void *arg)
 	struct sigev_thread_node *tn;
 	struct sigev_node *sn;
 	sigev_id_t id;
+	int ret;
 
 	tn = arg;
 	thr_self(&tn->tn_lwpid);
-	__sigev_list_lock();
-	_pthread_cond_broadcast(&tn->tn_cv);
-	__sigev_list_unlock();
+	__sigev_thread_list_lock();
+	_pthread_cond_broadcast(sigev_threads_cv);
+	__sigev_thread_list_unlock();
 
 	/*
 	 * Service thread should not be killed by callback, if user
@@ -405,19 +499,33 @@ sigev_service_loop(void *arg)
 	sigaddset(&set, SIGEV_SIGSERVICE);
 	pthread_cleanup_push(thread_cleanup, tn);
 	for (;;) {
-		if (__predict_false(sigwaitinfo(&set, &si) == -1))
+		ret = sigwaitinfo(&set, &si);
+		__sigev_thread_list_lock();
+		if (tn->tn_refcount == 0) {
+			TAILQ_REMOVE(&sigev_threads, tn, tn_link);
+			__sigev_thread_list_unlock();
+			free(tn);
+			break;
+		}
+		__sigev_thread_list_unlock();
+		if (ret == -1)
 			continue;
-
 		id = sigev_get_id(&si);
 		__sigev_list_lock();
 		sn = __sigev_find(si.si_code, id);
 		if (sn != NULL) {
-			tn->tn_cur = sn;
-			sn->sn_flags |= SNF_WORKING;
-			__sigev_list_unlock();
-			sn->sn_dispatch(sn, &si);
-			after_dispatch(tn);
-		} else  {
+			sn->sn_info = si;
+			if (!(sn->sn_flags & SNF_THREADPOOL)) {
+				tn->tn_cur = sn;
+				sn->sn_flags |= SNF_WORKING;
+				__sigev_list_unlock();
+				sn->sn_dispatch(sn);
+				after_dispatch(tn);
+			} else {
+				assert(!(sn->sn_flags & SNF_ACTQ));
+				sigev_put(sn);
+			}
+		} else {
 			tn->tn_cur = NULL;
 			__sigev_list_unlock();
 		}
@@ -425,3 +533,153 @@ sigev_service_loop(void *arg)
 	pthread_cleanup_pop(0);
 	return (0);
 }
+
+/*
+ * Hand off notifications to worker threads.
+ *
+ * prerequist: sigev list locked.
+ */
+static void 
+sigev_put(struct sigev_node *sn)
+{
+	struct sigev_worker *worker;
+	pthread_t td;
+	int ret, ready;
+
+	TAILQ_INSERT_TAIL(&sigev_actq, sn, sn_actq);
+	sn->sn_flags |= SNF_ACTQ;
+	/*
+	 * check if we should add more worker threads unless quota is hit.
+	 */
+	if (LIST_EMPTY(&sigev_worker_ready) &&
+	    sigev_worker_count + sigev_worker_start < sigev_worker_high) {
+		sigev_worker_start++;
+		__sigev_list_unlock();
+		worker = malloc(sizeof(*worker));
+		_pthread_cond_init(&worker->sw_cv, NULL);
+		worker->sw_flags = 0;
+		worker->sw_sn = 0;
+		worker->sw_readyptr = &ready;
+		ready = 0;
+		ret = pthread_create(&td, &sigev_default_attr,
+			sigev_worker_routine, worker);
+		if (ret) {
+			warnc(ret, "%s:%s can not create worker thread",
+				__FILE__, __func__);
+			__sigev_list_lock();
+			sigev_worker_start--;
+			__sigev_list_unlock();
+		} else {
+			__sigev_list_lock();
+			while (ready == 0) {
+				_pthread_cond_wait(sigev_worker_init_cv,
+					sigev_list_mtx);	
+			}
+			__sigev_list_unlock();
+		}
+	} else {
+		worker = LIST_FIRST(&sigev_worker_ready);
+		if (worker) {
+			LIST_REMOVE(worker, sw_link);
+			worker->sw_flags &= ~SWF_READYQ;
+			_pthread_cond_broadcast(&worker->sw_cv);
+			__sigev_list_unlock();
+		}
+	}
+}
+
+/*
+ * Background thread to dispatch notification to user code.
+ * These threads are not bound to any realtime objects.
+ */
+static void *
+sigev_worker_routine(void *arg)
+{
+	struct sigev_worker *worker;
+	struct sigev_node *sn;
+	struct timespec ts;
+	int ret;
+
+	worker = arg;
+	__sigev_list_lock();
+	sigev_worker_count++;
+	sigev_worker_start--;
+	(*(worker->sw_readyptr))++;
+	LIST_INSERT_HEAD(&sigev_worker_ready, worker, sw_link);
+	worker->sw_flags |= SWF_READYQ;
+	_pthread_cond_broadcast(sigev_worker_init_cv);
+
+	for (;;) {
+		if (worker->sw_flags & SWF_READYQ) {
+			LIST_REMOVE(worker, sw_link);
+			worker->sw_flags &= ~SWF_READYQ;
+		}
+
+		sn = TAILQ_FIRST(&sigev_actq);
+		if (sn != NULL) {
+			TAILQ_REMOVE(&sigev_actq, sn, sn_actq);
+			sn->sn_flags &= ~SNF_ACTQ;
+			sn->sn_flags |= SNF_WORKING;
+			__sigev_list_unlock();
+
+			worker->sw_sn = sn;
+			pthread_cleanup_push(worker_cleanup, worker);
+			sn->sn_dispatch(sn);
+			pthread_cleanup_pop(0);
+			worker->sw_sn = NULL;
+
+			__sigev_list_lock();
+			sn->sn_flags &= ~SNF_WORKING;
+			if (sn->sn_flags & SNF_REMOVED)
+				__sigev_free(sn);
+			else if (sn->sn_flags & SNF_ONESHOT)
+				__sigev_delete_node(sn);
+		} else {
+			LIST_INSERT_HEAD(&sigev_worker_ready, worker, sw_link);
+			worker->sw_flags |= SWF_READYQ;
+			clock_gettime(CLOCK_REALTIME, &ts);
+			ts.tv_sec += SIGEV_WORKER_IDLE;
+			ret = _pthread_cond_timedwait(&worker->sw_cv,
+				sigev_list_mtx, &ts);
+			if (ret == ETIMEDOUT) {
+				/*
+				 * If we were timeouted and there is nothing
+				 * to do, exit the thread.
+				 */
+				if (TAILQ_EMPTY(&sigev_actq) &&
+				    (worker->sw_flags & SWF_READYQ) &&
+				    sigev_worker_count > sigev_worker_low)
+					goto out;
+			}
+		}
+	}
+out:
+	if (worker->sw_flags & SWF_READYQ) {
+		LIST_REMOVE(worker, sw_link);
+		worker->sw_flags &= ~SWF_READYQ;
+	}
+	sigev_worker_count--;
+	__sigev_list_unlock();
+	_pthread_cond_destroy(&worker->sw_cv);
+	free(worker);
+	return (0);
+}
+
+static void
+worker_cleanup(void *arg)
+{
+	struct sigev_worker *worker = arg;
+	struct sigev_node *sn;
+
+	sn = worker->sw_sn;
+	__sigev_list_lock();
+	sn->sn_flags &= ~SNF_WORKING;
+	if (sn->sn_flags & SNF_REMOVED)
+		__sigev_free(sn);
+	else if (sn->sn_flags & SNF_ONESHOT)
+		__sigev_delete_node(sn);
+	sigev_worker_count--;
+	__sigev_list_unlock();
+	_pthread_cond_destroy(&worker->sw_cv);
+	free(worker);
+}