Re: thread cancellation, take 2

[Ludovic Courtès]

Hi Julian,

I went through your patch again and installed a slightly modified
version, mostly formatting issues (attached).  I guess you're now more
familiar than I am with specific pthread issues, so I could have missed
a few points.

You changed `all-threads' so that it hides the signal-delivery thread.
It seems harmless and reasonable, so I installed it too.  If others can
think of reasons not to do it, please speak up.

Thanks!

Ludovic.

--- orig/ChangeLog
+++ mod/ChangeLog
@@ -1,3 +1,7 @@
+2007-10-20  Julian Graham  <address@hidden>
+
+       * NEWS: Mention thread cancellation and cleanup API.
+
 2007-10-17  Ludovic Courtès  <address@hidden>
 
        * NEWS: Mention reader bug-fix.
--- orig/NEWS
+++ mod/NEWS
@@ -26,6 +26,9 @@ be used for efficiently implementing a S
 ** Duplicate bindings among used modules are resolved lazily.
 This slightly improves program startup times.
 
+** New thread cancellation and thread cleanup API
+See `cancel-thread', `set-thread-cleanup!', and `thread-cleanup'.
+
 * Changes to the C interface
 
 ** Functions for handling `scm_option' now no longer require an argument
--- orig/libguile/ChangeLog
+++ mod/libguile/ChangeLog
@@ -1,3 +1,42 @@
+2007-10-20  Julian Graham  <address@hidden>
+
+       Add support for thread cancellation and user-defined thread
+       cleanup handlers.  Small rework by Ludovic Courtès.
+
+       * null-threads.h (scm_i_pthread_cancel,
+       scm_i_pthread_cleanup_push, scm_i_pthread_cleanup_pop): New.
+       * pthread-threads.h (scm_i_pthread_cancel,
+       scm_i_pthread_cleanup_push, scm_i_pthread_cleanup_pop): New.
+       * scmsigs.c (scm_i_signal_delivery_thread,
+       signal_delivery_thread_mutex): New.
+       (signal_delivery_thread): Leave when `read_without_guile ()'
+       returns zero.
+       (start_signal_delivery_thread): Acquire SIGNAL_DELIVERY_THREAD
+       before spawning the thread.  Initialize
+       SCM_I_SIGNAL_DELIVERY_THREAD.
+       (ensure_signal_delivery_thread): Renamed to...
+       (scm_i_ensure_signal_delivery_thread): this.
+       (scm_i_close_signal_pipe): New.
+       * scmsigs.h: Updated.
+       * threads.c (thread_mark): Mark `t->cleanup_handler'.
+       (guilify_self_1): Initialize `t->cleanup_handler' and
+       `t->canceled'.
+       (do_thread_exit): Invoke `t->cleanup_handler'.
+       (on_thread_exit): Call `scm_i_ensure_signal_delivery_thread ()'.
+       Call `scm_i_close_signal_pipe ()' when the next-to-last thread
+       vanishes.
+       (scm_leave_guile_cleanup): New.
+       (scm_i_with_guile_and_parent): Use `scm_i_pthread_cleanup_push ()' 
+       and `scm_leave_guile_cleanup ()' to leave guile mode, rather
+       than call `scm_leave_guile ()' after FUNC.
+       (scm_cancel_thread, scm_set_thread_cleanup_x,
+       scm_threads_cleanup): New.
+       (scm_all_threads): Remove SCM_I_SIGNAL_DELIVERY_THREAD from the
+       returned list.
+       * threads.h (scm_i_thread)[cleanup_handler, canceled]: New
+       fields.
+       Add declarations of new functions.
+       
 2007-10-17  Ludovic Courtès  <address@hidden>
 
        * read.c (CHAR_IS_BLANK_): Add `\r' (ASCII 0x0d).  This fixes a
--- orig/libguile/null-threads.h
+++ mod/libguile/null-threads.h
@@ -41,6 +41,9 @@
 #define scm_i_pthread_create(t,a,f,d)       (*(t)=0, (void)(f), ENOSYS)
 #define scm_i_pthread_detach(t)             do { } while (0)
 #define scm_i_pthread_exit(v)               exit(0)
+#define scm_i_pthread_cancel(t)             0
+#define scm_i_pthread_cleanup_push(t,v)     0
+#define scm_i_pthread_cleanup_pop(e)        0
 #define scm_i_sched_yield()                 0
 
 /* Signals
--- orig/libguile/pthread-threads.h
+++ mod/libguile/pthread-threads.h
@@ -35,6 +35,9 @@
 #define scm_i_pthread_create                pthread_create
 #define scm_i_pthread_detach                pthread_detach
 #define scm_i_pthread_exit                  pthread_exit
+#define scm_i_pthread_cancel                pthread_cancel
+#define scm_i_pthread_cleanup_push          pthread_cleanup_push
+#define scm_i_pthread_cleanup_pop           pthread_cleanup_pop
 #define scm_i_sched_yield                   sched_yield
 
 /* Signals
--- orig/libguile/scmsigs.c
+++ mod/libguile/scmsigs.c
@@ -33,6 +33,7 @@
 #include "libguile/eval.h"
 #include "libguile/root.h"
 #include "libguile/vectors.h"
+#include "libguile/threads.h"
 
 #include "libguile/validate.h"
 #include "libguile/scmsigs.h"
@@ -99,6 +100,14 @@ static SCM *signal_handlers;
 static SCM signal_handler_asyncs;
 static SCM signal_handler_threads;
 
+/* The signal delivery thread.  */
+scm_i_thread *scm_i_signal_delivery_thread = NULL;
+
+/* The mutex held when launching the signal delivery thread.  */
+static scm_i_pthread_mutex_t signal_delivery_thread_mutex =
+  SCM_I_PTHREAD_MUTEX_INITIALIZER;
+
+
 /* saves the original C handlers, when a new handler is installed.
    set to SIG_ERR if the original handler is installed.  */
 #ifdef HAVE_SIGACTION
@@ -185,24 +194,34 @@ signal_delivery_thread (void *data)
          if (scm_is_true (h))
            scm_system_async_mark_for_thread (h, t);
        }
+      else if (n == 0)
+       break; /* the signal pipe was closed. */
       else if (n < 0 && errno != EINTR)
        perror ("error in signal delivery thread");
     }
 
-  return SCM_UNSPECIFIED;      /* not reached */
+  return SCM_UNSPECIFIED; /* not reached unless all other threads exited */
 }
 
 static void
 start_signal_delivery_thread (void)
 {
+  SCM signal_thread;
+
+  scm_i_pthread_mutex_lock (&signal_delivery_thread_mutex);
+
   if (pipe (signal_pipe) != 0)
     scm_syserror (NULL);
-  scm_spawn_thread (signal_delivery_thread, NULL,
-                   scm_handle_by_message, "signal delivery thread");
+  signal_thread = scm_spawn_thread (signal_delivery_thread, NULL,
+                                   scm_handle_by_message,
+                                   "signal delivery thread");
+  scm_i_signal_delivery_thread = SCM_I_THREAD_DATA (signal_thread);
+
+  scm_i_pthread_mutex_unlock (&signal_delivery_thread_mutex);
 }
 
-static void
-ensure_signal_delivery_thread ()
+void
+scm_i_ensure_signal_delivery_thread ()
 {
   static scm_i_pthread_once_t once = SCM_I_PTHREAD_ONCE_INIT;
   scm_i_pthread_once (&once, start_signal_delivery_thread);
@@ -228,8 +247,8 @@ take_signal (int signum)
 #endif
 }
 
-static void
-ensure_signal_delivery_thread ()
+void
+scm_i_ensure_signal_delivery_thread ()
 {
   return;
 }
@@ -332,7 +351,7 @@ SCM_DEFINE (scm_sigaction_for_thread, "s
        SCM_MISC_ERROR ("thread has already exited", SCM_EOL);
     }
 
-  ensure_signal_delivery_thread ();
+  scm_i_ensure_signal_delivery_thread ();
 
   SCM_CRITICAL_SECTION_START;
   old_handler = SCM_SIMPLE_VECTOR_REF (*signal_handlers, csig);
@@ -653,6 +672,21 @@ SCM_DEFINE (scm_raise, "raise", 1, 0, 0,
 
 
 void
+scm_i_close_signal_pipe()
+{
+  /* SIGNAL_DELIVERY_THREAD_MUTEX is only locked while the signal delivery
+     thread is being launched.  The thread that calls this function is
+     already holding the thread admin mutex, so if the delivery thread hasn't
+     been launched at this point, it never will be before shutdown.  */
+  scm_i_pthread_mutex_lock (&signal_delivery_thread_mutex);
+
+  if (scm_i_signal_delivery_thread != NULL)
+    close (signal_pipe[1]);
+
+  scm_i_pthread_mutex_unlock (&signal_delivery_thread_mutex);
+}
+
+void
 scm_init_scmsigs ()
 {
   int i;
--- orig/libguile/scmsigs.h
+++ mod/libguile/scmsigs.h
@@ -3,7 +3,7 @@
 #ifndef SCM_SCMSIGS_H
 #define SCM_SCMSIGS_H
 
-/* Copyright (C) 1995,1996,1997,1998,2000, 2002, 2006 Free Software 
Foundation, Inc.
+/* Copyright (C) 1995,1996,1997,1998,2000, 2002, 2006, 2007 Free Software 
Foundation, Inc.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -23,6 +23,7 @@
 
 
 #include "libguile/__scm.h"
+#include "libguile/threads.h"
 
 
 
@@ -41,6 +42,11 @@ SCM_API SCM scm_usleep (SCM i);
 SCM_API SCM scm_raise (SCM sig);
 SCM_API void scm_init_scmsigs (void);
 
+SCM_API void scm_i_close_signal_pipe (void);
+SCM_API void scm_i_ensure_signal_delivery_thread (void);
+
+SCM_API scm_i_thread *scm_i_signal_delivery_thread;
+
 #endif  /* SCM_SCMSIGS_H */
 
 /*
--- orig/libguile/threads.c
+++ mod/libguile/threads.c
@@ -1,4 +1,4 @@
-/* Copyright (C) 1995,1996,1997,1998,2000,2001, 2002, 2003, 2004, 2005, 2006 
Free Software Foundation, Inc.
+/* Copyright (C) 1995,1996,1997,1998,2000,2001, 2002, 2003, 2004, 2005, 2006, 
2007 Free Software Foundation, Inc.
  * 
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -48,6 +48,7 @@
 #include "libguile/continuations.h"
 #include "libguile/gc.h"
 #include "libguile/init.h"
+#include "libguile/scmsigs.h"
 
 #ifdef __MINGW32__
 #ifndef ETIMEDOUT
@@ -131,6 +132,7 @@ thread_mark (SCM obj)
 {
   scm_i_thread *t = SCM_I_THREAD_DATA (obj);
   scm_gc_mark (t->result);
+  scm_gc_mark (t->cleanup_handler);
   scm_gc_mark (t->join_queue);
   scm_gc_mark (t->dynwinds);
   scm_gc_mark (t->active_asyncs);
@@ -415,6 +417,7 @@ guilify_self_1 (SCM_STACKITEM *base)
   t->pthread = scm_i_pthread_self ();
   t->handle = SCM_BOOL_F;
   t->result = SCM_BOOL_F;
+  t->cleanup_handler = SCM_BOOL_F;
   t->join_queue = SCM_EOL;
   t->dynamic_state = SCM_BOOL_F;
   t->dynwinds = SCM_EOL;
@@ -434,6 +437,7 @@ guilify_self_1 (SCM_STACKITEM *base)
   scm_i_pthread_mutex_init (&t->heap_mutex, NULL);
   t->clear_freelists_p = 0;
   t->gc_running_p = 0;
+  t->canceled = 0;
   t->exited = 0;
 
   t->freelist = SCM_EOL;
@@ -478,7 +482,17 @@ guilify_self_2 (SCM parent)
 static void *
 do_thread_exit (void *v)
 {
-  scm_i_thread *t = (scm_i_thread *)v;
+  scm_i_thread *t = (scm_i_thread *) v;
+
+  if (!scm_is_false (t->cleanup_handler))
+    {
+      SCM ptr = t->cleanup_handler;
+
+      t->cleanup_handler = SCM_BOOL_F;
+      t->result = scm_internal_catch (SCM_BOOL_T,
+                                     (scm_t_catch_body) scm_call_0, ptr,
+                                     scm_handle_by_message_noexit, NULL);
+    }
 
   scm_i_scm_pthread_mutex_lock (&thread_admin_mutex);
 
@@ -489,6 +503,7 @@ do_thread_exit (void *v)
     ;
 
   scm_i_pthread_mutex_unlock (&thread_admin_mutex);
+
   return NULL;
 }
 
@@ -496,10 +511,14 @@ static void
 on_thread_exit (void *v)
 {
   /* This handler is executed in non-guile mode.  */
-  scm_i_thread *t = (scm_i_thread *)v, **tp;
+  scm_i_thread *t = (scm_i_thread *) v, **tp;
 
   scm_i_pthread_setspecific (scm_i_thread_key, v);
 
+  /* Ensure the signal handling thread has been launched, because we might be
+     shutting it down.  */
+  scm_i_ensure_signal_delivery_thread ();
+
   /* Unblocking the joining threads needs to happen in guile mode
      since the queue is a SCM data structure.  */
   scm_with_guile (do_thread_exit, v);
@@ -515,6 +534,14 @@ on_thread_exit (void *v)
        break;
       }
   thread_count--;
+
+  /* If there's only one other thread, it could be the signal delivery
+     thread, so we need to notify it to shut down by closing its read pipe.
+     If it's not the signal delivery thread, then closing the read pipe isn't
+     going to hurt.  */
+  if (thread_count <= 1)
+    scm_i_close_signal_pipe ();
+
   scm_i_pthread_mutex_unlock (&thread_admin_mutex);
 
   scm_i_pthread_setspecific (scm_i_thread_key, NULL);
@@ -684,17 +711,30 @@ scm_with_guile (void *(*func)(void *), v
                                      scm_i_default_dynamic_state);
 }
 
+static void
+scm_leave_guile_cleanup (void *x)
+{
+  scm_leave_guile ();
+}
+
 void *
-scm_i_with_guile_and_parent (void *(*func)(void *), void *data,
-                            SCM parent)
+scm_i_with_guile_and_parent (void *(*func)(void *), void *data, SCM parent)
 {
   void *res;
   int really_entered;
   SCM_STACKITEM base_item;
+
   really_entered = scm_i_init_thread_for_guile (&base_item, parent);
-  res = scm_c_with_continuation_barrier (func, data);
   if (really_entered)
-    scm_leave_guile ();
+    {
+      scm_i_pthread_cleanup_push (scm_leave_guile_cleanup, NULL);
+      res = scm_c_with_continuation_barrier (func, data);
+      scm_i_pthread_cleanup_pop (0);
+      scm_leave_guile ();
+    }
+  else 
+    res = scm_c_with_continuation_barrier (func, data);
+
   return res;
 }
 
@@ -880,6 +920,74 @@ SCM_DEFINE (scm_yield, "yield", 0, 0, 0,
 }
 #undef FUNC_NAME
 
+SCM_DEFINE (scm_cancel_thread, "cancel-thread", 1, 0, 0,
+           (SCM thread),
+"Asynchronously force the target @var{thread} to terminate. @var{thread} "
+"cannot be the current thread, and if @var{thread} has already terminated or "
+"been signaled to terminate, this function is a no-op.")
+#define FUNC_NAME s_scm_cancel_thread
+{
+  scm_i_thread *t = NULL;
+
+  SCM_VALIDATE_THREAD (1, thread);
+  t = SCM_I_THREAD_DATA (thread);
+  scm_i_scm_pthread_mutex_lock (&thread_admin_mutex);
+  if (!t->canceled)
+    {
+      t->canceled = 1;
+      scm_i_pthread_mutex_unlock (&thread_admin_mutex);
+      scm_i_pthread_cancel (t->pthread);
+    }
+  else
+    scm_i_pthread_mutex_unlock (&thread_admin_mutex);
+
+  return SCM_UNSPECIFIED;
+}
+#undef FUNC_NAME
+
+SCM_DEFINE (scm_set_thread_cleanup_x, "set-thread-cleanup!", 2, 0, 0,
+           (SCM thread, SCM proc),
+"Set the thunk @var{proc} as the cleanup handler for the thread @var{thread}. "
+"This handler will be called when the thread exits.")
+#define FUNC_NAME s_scm_set_thread_cleanup_x
+{
+  scm_i_thread *t;
+
+  SCM_VALIDATE_THREAD (1, thread);
+  if (!scm_is_false (proc))
+    SCM_VALIDATE_THUNK (2, proc);
+
+  scm_i_pthread_mutex_lock (&thread_admin_mutex);
+
+  t = SCM_I_THREAD_DATA (thread);
+  if (!(t->exited || t->canceled))
+    t->cleanup_handler = proc;
+
+  scm_i_pthread_mutex_unlock (&thread_admin_mutex);
+
+  return SCM_UNSPECIFIED;
+}
+#undef FUNC_NAME
+
+SCM_DEFINE (scm_thread_cleanup, "thread-cleanup", 1, 0, 0,
+           (SCM thread),
+"Return the cleanup handler installed for the thread @var{thread}.")
+#define FUNC_NAME s_scm_thread_cleanup
+{
+  scm_i_thread *t;
+  SCM ret;
+
+  SCM_VALIDATE_THREAD (1, thread);
+
+  scm_i_pthread_mutex_lock (&thread_admin_mutex);
+  t = SCM_I_THREAD_DATA (thread);
+  ret = (t->exited || t->canceled) ? SCM_BOOL_F : t->cleanup_handler;
+  scm_i_pthread_mutex_unlock (&thread_admin_mutex);
+
+  return ret;
+}
+#undef FUNC_NAME
+
 SCM_DEFINE (scm_join_thread, "join-thread", 1, 0, 0,
            (SCM thread),
 "Suspend execution of the calling thread until the target @var{thread} "
@@ -891,7 +999,7 @@ SCM_DEFINE (scm_join_thread, "join-threa
 
   SCM_VALIDATE_THREAD (1, thread);
   if (scm_is_eq (scm_current_thread (), thread))
-    SCM_MISC_ERROR ("can not join the current thread", SCM_EOL);
+    SCM_MISC_ERROR ("cannot join the current thread", SCM_EOL);
 
   scm_i_scm_pthread_mutex_lock (&thread_admin_mutex);
 
@@ -911,10 +1019,13 @@ SCM_DEFINE (scm_join_thread, "join-threa
   res = t->result;
 
   scm_i_pthread_mutex_unlock (&thread_admin_mutex);
+
   return res;
 }
 #undef FUNC_NAME
 
+
+
 /*** Fat mutexes */
 
 /* We implement our own mutex type since we want them to be 'fair', we
@@ -1537,8 +1648,11 @@ SCM_DEFINE (scm_all_threads, "all-thread
   l = &list;
   for (t = all_threads; t && n > 0; t = t->next_thread)
     {
-      SCM_SETCAR (*l, t->handle);
-      l = SCM_CDRLOC (*l);
+      if (t != scm_i_signal_delivery_thread)
+       {
+         SCM_SETCAR (*l, t->handle);
+         l = SCM_CDRLOC (*l);
+       }
       n--;
     }
   *l = SCM_EOL;
--- orig/libguile/threads.h
+++ mod/libguile/threads.h
@@ -3,7 +3,7 @@
 #ifndef SCM_THREADS_H
 #define SCM_THREADS_H
 
-/* Copyright (C) 1996,1997,1998,2000,2001, 2002, 2003, 2004, 2006 Free 
Software Foundation, Inc.
+/* Copyright (C) 1996,1997,1998,2000,2001, 2002, 2003, 2004, 2006, 2007 Free 
Software Foundation, Inc.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -49,9 +49,11 @@ typedef struct scm_i_thread {
 
   SCM handle;
   scm_i_pthread_t pthread;
-  
+
+  SCM cleanup_handler;
   SCM join_queue;
   SCM result;
+  int canceled;
   int exited;
 
   SCM sleep_object;
@@ -153,6 +155,9 @@ do { \
 
 SCM_API SCM scm_call_with_new_thread (SCM thunk, SCM handler);
 SCM_API SCM scm_yield (void);
+SCM_API SCM scm_cancel_thread (SCM t);
+SCM_API SCM scm_set_thread_cleanup_x (SCM thread, SCM proc);
+SCM_API SCM scm_thread_cleanup (SCM thread);
 SCM_API SCM scm_join_thread (SCM t);
 
 SCM_API SCM scm_make_mutex (void);
--- orig/test-suite/ChangeLog
+++ mod/test-suite/ChangeLog
@@ -1,3 +1,10 @@
+2007-10-20  Julian Graham  <address@hidden>
+
+       * tests/threads.test: Use proper `define-module'.
+       (cancel-thread, handler result passed to join, can cancel self,
+       handler supplants final expr, remove handler by setting false,
+       initial handler is false): New tests.
+
 2007-10-17  Ludovic Courtès  <address@hidden>
 
        * tests/reader.test (reading)[CR recognized as a token
--- orig/test-suite/tests/threads.test
+++ mod/test-suite/tests/threads.test
@@ -1,6 +1,6 @@
 ;;;; threads.test --- Tests for Guile threading.    -*- scheme -*-
 ;;;;
-;;;; Copyright 2003, 2006 Free Software Foundation, Inc.
+;;;; Copyright 2003, 2006, 2007 Free Software Foundation, Inc.
 ;;;;
 ;;;; This program is free software; you can redistribute it and/or modify
 ;;;; it under the terms of the GNU General Public License as published by
@@ -17,8 +17,10 @@
 ;;;; the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
 ;;;; Boston, MA 02110-1301 USA
 
-(use-modules (ice-9 threads)
-            (test-suite lib))
+(define-module (test-threads)
+  :use-module (ice-9 threads)
+  :use-module (test-suite lib))
+
 
 (if (provided? 'threads)
     (begin
@@ -133,4 +135,54 @@
                                (lambda (n) (set! result (cons n result)))
                                (lambda (n) (* 2 n))
                                '(0 1 2 3 4 5))
-           (equal? result '(10 8 6 4 2 0)))))))
+           (equal? result '(10 8 6 4 2 0)))))
+
+      ;;
+      ;; thread cancellation
+      ;;
+
+      (with-test-prefix "cancel-thread"
+
+        (pass-if "cancel succeeds"
+         (let ((m (make-mutex)))
+           (lock-mutex m)
+           (let ((t (begin-thread (begin (lock-mutex m) 'foo))))
+             (cancel-thread t)
+             (join-thread t)
+             #t)))
+
+       (pass-if "handler result passed to join"
+         (let ((m (make-mutex)))
+           (lock-mutex m)
+           (let ((t (begin-thread (lock-mutex m))))
+             (set-thread-cleanup! t (lambda () 'foo))
+             (cancel-thread t)
+             (eq? (join-thread t) 'foo))))
+
+       (pass-if "can cancel self"
+         (let ((m (make-mutex)))
+           (lock-mutex m)
+           (let ((t (begin-thread (begin
+                                    (set-thread-cleanup! (current-thread)
+                                                         (lambda () 'foo))
+                                    (cancel-thread (current-thread))
+                                    (lock-mutex m)))))
+             (eq? (join-thread t) 'foo))))
+
+       (pass-if "handler supplants final expr"
+         (let ((t (begin-thread (begin (set-thread-cleanup! (current-thread)
+                                                            (lambda () 'bar))
+                                       'foo))))
+           (eq? (join-thread t) 'bar)))
+
+       (pass-if "remove handler by setting false"
+         (let ((m (make-mutex)))
+           (lock-mutex m)
+           (let ((t (begin-thread (lock-mutex m) 'bar)))
+             (set-thread-cleanup! t (lambda () 'foo))
+             (set-thread-cleanup! t #f)
+             (unlock-mutex m)
+             (eq? (join-thread t) 'bar))))
+
+       (pass-if "initial handler is false"
+         (not (thread-cleanup (current-thread)))))))