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