queue: rework kevent cancellation handling

The kevent() function as implemented by libkqueue does not
support thread cancellation the same way a real kevent() (on
FreeBSD) appears to.  So pretend no implementation of kevent()
is cancelable and handle cancellation ourselves using
pthread_testcancel().  This allows us to support any platform
where kevent() may work, since it's unclear if other *BSDs
implement kevent() as a cancellation point.

1 files changed, 22 insertions, 6 deletions

diff --git a/thrpool.c b/thrpool.c
index 3d05e9f..dfffc56 100644
--- a/thrpool.c
+++ b/thrpool.c
@@ -9,7 +9,7 @@
  * We also use syslog() and *printf() functions which take a lot of
  * stack under glibc, so we'll add BUFSIZ (8192 on glibc) to that
  */
-#if defined(LIBKQUEUE) && (LIBKQUEUE == 1)
+#if MOG_LIBKQUEUE /* libkqueue uses quite a bit of stack */
 #  define MOG_THR_STACK_SIZE (0)
 #elif defined(__GLIBC__) || defined(__FreeBSD__)
 #  define MOG_THR_STACK_SIZE ((16 * 1024) + MAX(8192,BUFSIZ))
@@ -49,13 +49,29 @@ mog_thrpool_start(struct mog_thrpool *tp, size_t n,
         }
 }
 
-void mog_thrpool_quit(struct mog_thrpool *tp)
+void mog_thrpool_quit(struct mog_thrpool *tp, struct mog_queue *q)
 {
-        pthread_t *thr = tp->threads;
+        size_t i;
+        int err;
+        uintptr_t ident = 0;
+
+        for (i = 0; i < tp->n_threads; i++)
+                CHECK(int, 0, pthread_cancel(tp->threads[i]));
 
-        for (; tp->n_threads--; thr++) {
-                CHECK(int, 0, pthread_cancel(*thr));
-                CHECK(int, 0, pthread_join(*thr, NULL));
+        for (i = 0; i < tp->n_threads; i++) {
+                if (q) {
+                        /*
+                         * if we can't rely on cancellation, keep poking
+                         * the thread until it wakes up from cancellation.
+                         */
+                        do {
+                                mog_idleq_poke(q, ident++);
+                                err = pthread_kill(tp->threads[i], 0);
+                        } while (err == 0);
+                        assert(err == ESRCH && "pthread_kill() usage bug");
+                }
+                CHECK(int, 0, pthread_join(tp->threads[i], NULL));
         }
+
         mog_free_and_null(&tp->threads);
 }