Add a timeout for using the worker threads

Otherwise operations can build up while the thread pool is busy, and this'll
hopefully make it clearer when there are issues with the thread pools.

1 files changed, 38 insertions, 3 deletions

diff --git a/guix-build-coordinator/utils/fibers.scm b/guix-build-coordinator/utils/fibers.scm
index 921da8d..4a7572d 100644
--- a/guix-build-coordinator/utils/fibers.scm
+++ b/guix-build-coordinator/utils/fibers.scm
@@ -2,11 +2,14 @@
   #:use-module (ice-9 match)
   #:use-module (ice-9 threads)
   #:use-module (fibers)
+  #:use-module (fibers timers)
   #:use-module (fibers channels)
+  #:use-module (fibers operations)
   #:use-module (fibers conditions)
   #:use-module (guix-build-coordinator utils)
   #:export (make-worker-thread-channel
             call-with-worker-thread
+            worker-thread-timeout-error?
 
             call-with-sigint
 
@@ -100,14 +103,46 @@ arguments of the worker thread procedure."
      (iota parallelism))
     channel))
 
-(define* (call-with-worker-thread channel proc #:key duration-logger)
+(define &worker-thread-timeout
+  (make-exception-type '&worker-thread-timeout
+                       &error
+                       '()))
+
+(define make-worker-thread-timeout-error
+  (record-constructor &worker-thread-timeout))
+
+(define worker-thread-timeout-error?
+  (record-predicate &worker-thread-timeout))
+
+(define* (call-with-worker-thread channel proc #:key duration-logger
+                                  (timeout 30))
   "Send PROC to the worker thread through CHANNEL.  Return the result of PROC.
 If already in the worker thread, call PROC immediately."
   (let ((args (%worker-thread-args)))
     (if args
         (call-with-delay-logging proc #:args args)
-        (let ((reply (make-channel)))
-          (put-message channel (list reply (get-internal-real-time) proc))
+        (let* ((reply (make-channel))
+               (operation-success?
+                (perform-operation
+                 (let ((put
+                        (wrap-operation
+                         (put-operation channel
+                                        (list reply
+                                              (get-internal-real-time)
+                                              proc))
+                         (const #t))))
+
+                   (if timeout
+                       (choice-operation
+                        put
+                        (wrap-operation (sleep-operation timeout)
+                                        (const #f)))
+                       put)))))
+
+          (unless operation-success?
+            (raise-exception
+             (make-worker-thread-timeout-error)))
+
           (match (get-message reply)
             (('worker-thread-error duration exn)
              (when duration-logger