TT#14008 use pthread_cancel semantics for shutdown

Instead of having each thread sleep only a little while and then
periodically check for the shutdown flag, make them sleep longer and use
pthread_cancel() to interrupt the sleep during a shutdown in the
designated break points.

Change-Id: I13f1872a0176697e064ceef4062db6ca6ccf7a0e

daemon/aux.c

@@ -133,12 +133,12 @@ static gint thread_equal(gconstpointer a, gconstpointer b) {
 	return !pthread_equal(*x, *y);
 }
 
-void threads_join_all(bool wait) {
+void threads_join_all(bool cancel) {
 	pthread_t *t;
 	GList *l;
 
 	while (1) {
-		if (wait) {
+		if (cancel) {
 			mutex_lock(&thread_wakers_lock);
 			for (l = thread_wakers; l; l = l->next) {
 				struct thread_waker *wk = l->data;
@@ -147,6 +147,13 @@ void threads_join_all(bool wait) {
 				mutex_unlock(wk->lock);
 			}
 			mutex_unlock(&thread_wakers_lock);
+
+			mutex_lock(&threads_lists_lock);
+			for (l = threads_running; l; l = l->next) {
+				t = l->data;
+				pthread_cancel(*t);
+			}
+			mutex_unlock(&threads_lists_lock);
 		}
 
 		mutex_lock(&threads_lists_lock);
@@ -164,7 +171,7 @@ void threads_join_all(bool wait) {
 			g_slice_free1(sizeof(*t), t);
 		}
 
-		if (!wait || !threads_running) {
+		if ((!cancel && rtpe_shutdown) || (cancel && !threads_running)) {
 			mutex_unlock(&threads_lists_lock);
 			break;
 		}
@@ -184,10 +191,18 @@ void thread_waker_del(struct thread_waker *wk) {
 	mutex_unlock(&thread_wakers_lock);
 }
 
+static void thread_detach_cleanup(void *dtp) {
+	struct detach_thread *dt = dtp;
+	g_slice_free1(sizeof(*dt), dt);
+	thread_join_me();
+}
+
 static void *thread_detach_func(void *d) {
 	struct detach_thread *dt = d;
 	pthread_t *t;
 
+	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
+
 	t = g_slice_alloc(sizeof(*t));
 	*t = pthread_self();
 	mutex_lock(&threads_lists_lock);
@@ -231,9 +246,10 @@ static void *thread_detach_func(void *d) {
 					dt->priority, strerror(errno));
 	}
 
+	pthread_cleanup_push(thread_detach_cleanup, dt);
 	dt->func(dt->data);
-	g_slice_free1(sizeof(*dt), dt);
-	thread_join_me();
+	pthread_cleanup_pop(true);
+
 	return NULL;
 }
 

daemon/load.c

@@ -15,6 +15,10 @@ int cpu_usage; // percent times 100 (0 - 9999)
 static long used_last, idle_last;
 
 void load_thread(void *dummy) {
+	// anything to do?
+	if (!rtpe_config.load_limit && !rtpe_config.cpu_limit)
+		return;
+
 	while (!rtpe_shutdown) {
 		if (rtpe_config.load_limit) {
 			double loadavg;
@@ -48,6 +52,8 @@ void load_thread(void *dummy) {
 			}
 		}
 
+		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 		usleep(500000);
+		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 	}
 }

daemon/main.c

@@ -116,11 +116,14 @@ static void sighandler(gpointer x) {
 	sigaddset(&ss, SIGUSR1);
 	sigaddset(&ss, SIGUSR2);
 
-	ts.tv_sec = 0;
-	ts.tv_nsec = 100000000; /* 0.1 sec */
+	ts.tv_sec = 10;
+	ts.tv_nsec = 0;
 
 	while (!rtpe_shutdown) {
+		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 		ret = sigtimedwait(&ss, NULL, &ts);
+		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
+
 		if (ret == -1) {
 			if (errno == EAGAIN || errno == EINTR)
 				continue;
@@ -1310,10 +1313,8 @@ int main(int argc, char **argv) {
 	}
 
 
-	while (!rtpe_shutdown) {
-		usleep(100000);
-		threads_join_all(false);
-	}
+	// reap threads as they shut down during run time
+	threads_join_all(false);
 
         // free libevent
 #if LIBEVENT_VERSION_NUMBER >= 0x02010100

daemon/poller.c

@@ -386,7 +386,9 @@ int poller_poll(struct poller *p, int timeout) {
 
 	mutex_unlock(&p->lock);
 	errno = 0;
+	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 	ret = epoll_wait(p->fd, evs, sizeof(evs) / sizeof(*evs), timeout);
+	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 	mutex_lock(&p->lock);
 
 	if (errno == EINTR)
@@ -567,17 +569,23 @@ int poller_add_timer(struct poller *p, void (*f)(void *), struct obj *o) {
 /* run in thread separate from poller_poll() */
 void poller_timer_loop(void *d) {
 	struct poller *p = d;
-	struct timeval tv;
-	int wt;
 
 	while (!rtpe_shutdown) {
-		gettimeofday(&tv, NULL);
-		if (tv.tv_sec != rtpe_now.tv_sec)
+		// run once a second on top of each second
+		struct timeval now;
+		gettimeofday(&now, NULL);
+		struct timeval next = { rtpe_now.tv_sec + 1, 0 };
+		if (now.tv_sec >= next.tv_sec)
 			goto now;
 
-		wt = 1000000 - tv.tv_usec;
-		wt = MIN(wt, 100000);
-		usleep(wt);
+		long long sleeptime = timeval_diff(&next, &now);
+		if (sleeptime <= 0)
+			goto now;
+
+		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
+		usleep(sleeptime);
+		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
+
 		continue;
 
 now:
@@ -605,7 +613,7 @@ void poller_loop2(void *d) {
 	struct poller *p = d;
 
 	while (!rtpe_shutdown) {
-		int ret = poller_poll(p, 100);
+		int ret = poller_poll(p, 10000);
 		if (ret < 0)
 			usleep(20 * 1000);
 	}

include/aux.h

@@ -343,7 +343,7 @@ struct thread_waker {
 
 void thread_waker_add(struct thread_waker *);
 void thread_waker_del(struct thread_waker *);
-void threads_join_all(bool);
+void threads_join_all(bool cancel);
 void thread_create_detach_prio(void (*)(void *), void *, const char *, int, const char *);
 INLINE void thread_create_detach(void (*f)(void *), void *a, const char *name) {
 	thread_create_detach_prio(f, a, NULL, 0, name);