Try and ensure that the worker threads don't disappear

Add some top level exception handling to guard against errors anywhere.

1 files changed, 72 insertions, 61 deletions

diff --git a/guix-build-coordinator/utils/fibers.scm b/guix-build-coordinator/utils/fibers.scm
index 947697d..0ce6da0 100644
--- a/guix-build-coordinator/utils/fibers.scm
+++ b/guix-build-coordinator/utils/fibers.scm
@@ -85,6 +85,68 @@ arguments of the worker thread procedure."
             (sleep 5)
             (destructor/safe args)))))
 
+  (define (process channel args)
+    (let loop ((current-lifetime lifetime))
+      (let ((exception?
+             (match (get-message channel)
+               (((? channel? reply) sent-time (? procedure? proc))
+                (let ((time-delay
+                       (- (get-internal-real-time)
+                          sent-time)))
+                  (delay-logger (/ time-delay
+                                   internal-time-units-per-second))
+
+                  (let* ((start-time (get-internal-real-time))
+                         (response
+                          (with-exception-handler
+                              (lambda (exn)
+                                (list 'worker-thread-error
+                                      (/ (- (get-internal-real-time)
+                                            start-time)
+                                         internal-time-units-per-second)
+                                      exn))
+                            (lambda ()
+                              (with-throw-handler #t
+                                (lambda ()
+                                  (call-with-values
+                                      (lambda ()
+                                        (apply proc args))
+                                    (lambda vals
+                                      (cons (/ (- (get-internal-real-time)
+                                                  start-time)
+                                               internal-time-units-per-second)
+                                            vals))))
+                                (lambda args
+                                  (when (match args
+                                          (('%exception exn)
+                                           (log-exception? exn))
+                                          (_ #t))
+                                    (simple-format
+                                     (current-error-port)
+                                     "worker-thread: exception: ~A\n" args)
+                                    (backtrace)))))
+                            #:unwind? #t)))
+                    (put-message reply
+                                 response)
+
+                    (match response
+                      (('worker-thread-error duration _)
+                       (when duration-logger
+                         (duration-logger duration proc))
+                       #t)
+                      ((duration . _)
+                       (when duration-logger
+                         (duration-logger duration proc))
+                       #f))))))))
+        (unless (and expire-on-exception?
+                     exception?)
+          (if (number? current-lifetime)
+              (unless (< current-lifetime 0)
+                (loop (if current-lifetime
+                          (- current-lifetime 1)
+                          #f)))
+              (loop #f))))))
+
   (let ((channel (make-channel)))
     (for-each
      (lambda (thread-index)
@@ -99,67 +161,16 @@ arguments of the worker thread procedure."
             (const #t))
 
           (let init ((args (initializer/safe)))
-            (parameterize ((%worker-thread-args args))
-              (let loop ((current-lifetime lifetime))
-                (let ((exception?
-                       (match (get-message channel)
-                         (((? channel? reply) sent-time (? procedure? proc))
-                          (let ((time-delay
-                                 (- (get-internal-real-time)
-                                    sent-time)))
-                            (delay-logger (/ time-delay
-                                             internal-time-units-per-second))
-
-                            (let* ((start-time (get-internal-real-time))
-                                   (response
-                                    (with-exception-handler
-                                        (lambda (exn)
-                                          (list 'worker-thread-error
-                                                (/ (- (get-internal-real-time)
-                                                      start-time)
-                                                   internal-time-units-per-second)
-                                                exn))
-                                      (lambda ()
-                                        (with-throw-handler #t
-                                          (lambda ()
-                                            (call-with-values
-                                                (lambda ()
-                                                  (apply proc args))
-                                              (lambda vals
-                                                (cons (/ (- (get-internal-real-time)
-                                                            start-time)
-                                                         internal-time-units-per-second)
-                                                      vals))))
-                                          (lambda args
-                                            (when (match args
-                                                    (('%exception exn)
-                                                     (log-exception? exn))
-                                                    (_ #t))
-                                              (simple-format
-                                               (current-error-port)
-                                               "worker-thread: exception: ~A\n" args)
-                                              (backtrace)))))
-                                      #:unwind? #t)))
-                              (put-message reply
-                                           response)
-
-                              (match response
-                                (('worker-thread-error duration _)
-                                 (when duration-logger
-                                   (duration-logger duration proc))
-                                 #t)
-                                ((duration . _)
-                                 (when duration-logger
-                                   (duration-logger duration proc))
-                                 #f))))))))
-                  (unless (and expire-on-exception?
-                               exception?)
-                    (if (number? current-lifetime)
-                        (unless (< current-lifetime 0)
-                          (loop (if current-lifetime
-                                    (- current-lifetime 1)
-                                    #f)))
-                        (loop #f))))))
+            (with-exception-handler
+                (lambda (exn)
+                  (simple-format
+                   (current-error-port)
+                   "worker-thread-channel: exception: ~A\n" exn))
+              (lambda ()
+                (parameterize ((%worker-thread-args args))
+                  (process channel args)))
+              #:unwind? #t)
+
             (when destructor
               (destructor/safe args))