From 5a17584c6a0dcf2fb05817c53ed005b6ee5b5306 Mon Sep 17 00:00:00 2001
From: Christopher Baines <mail@cbaines.net>
Date: Tue, 2 May 2023 15:10:05 +0200
Subject: 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.
---
 guix-build-coordinator/utils/fibers.scm | 41 ++++++++++++++++++++++++++++++---
 1 file changed, 38 insertions(+), 3 deletions(-)

(limited to 'guix-build-coordinator/utils')

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
-- 
cgit v1.2.3