Have the agent fetch builds based on the current number of threads

This means that if the agent is only processing 2 builds at a time, it'll only
fetch up to two builds, rather than whatever maximum it would fetch. This
avoids fetching builds unnecessarily.

2 files changed, 39 insertions, 34 deletions

diff --git a/guix-build-coordinator/agent.scm b/guix-build-coordinator/agent.scm
index efa61f6..d587f67 100644
--- a/guix-build-coordinator/agent.scm
+++ b/guix-build-coordinator/agent.scm
@@ -159,7 +159,7 @@
 
   (log-msg lgr 'INFO "starting agent " uuid)
   (log-msg lgr 'INFO "connecting to coordinator " coordinator-uri)
-  (let-values (((process-job-with-queue count-jobs list-jobs)
+  (let-values (((process-job-with-queue count-jobs count-threads list-jobs)
                 (create-work-queue current-max-builds
                                    process-job
                                    #:thread-start-delay
@@ -174,38 +174,39 @@
        (vector->list (assoc-ref details "builds")))
 
       (while #t
-        (if (< (count-jobs) (current-max-builds))
-            (let* ((queued-build-ids
-                    (map (lambda (job-args)
-                           (assoc-ref (car job-args) "uuid"))
-                         (list-jobs)))
-                   (fetched-builds
-                    (fetch-builds-for-agent coordinator-uri uuid password
-                                            systems
-                                            max-parallel-builds
-                                            #:log (build-log-procedure lgr)))
-                   (new-builds
-                    (remove (lambda (build)
-                              (member (assoc-ref build "uuid")
-                                      queued-build-ids))
-                            fetched-builds)))
-
-              (log-msg lgr 'INFO
-                       "max " max-parallel-builds
-                       " builds, currently allocated "
-                       (length fetched-builds))
-              (log-msg lgr 'INFO
-                       "starting " (length new-builds)
-                       " new "
-                       (if (eq? (length new-builds) 1) "build" "builds"))
-              (for-each
-               (lambda (job-args)
-                 (process-job-with-queue job-args))
-               new-builds)
-
-              (when (null? new-builds)
-                (sleep 30)))
-            (sleep 3))))))
+        (let ((current-threads (count-threads)))
+          (if (< (count-jobs) current-threads)
+              (let* ((queued-build-ids
+                      (map (lambda (job-args)
+                             (assoc-ref (car job-args) "uuid"))
+                           (list-jobs)))
+                     (fetched-builds
+                      (fetch-builds-for-agent coordinator-uri uuid password
+                                              systems
+                                              current-threads
+                                              #:log (build-log-procedure lgr)))
+                     (new-builds
+                      (remove (lambda (build)
+                                (member (assoc-ref build "uuid")
+                                        queued-build-ids))
+                              fetched-builds)))
+
+                (log-msg lgr 'INFO
+                         "running " current-threads
+                         " threads, currently allocated "
+                         (length fetched-builds))
+                (log-msg lgr 'INFO
+                         "starting " (length new-builds)
+                         " new "
+                         (if (eq? (length new-builds) 1) "build" "builds"))
+                (for-each
+                 (lambda (job-args)
+                   (process-job-with-queue job-args))
+                 new-builds)
+
+                (when (null? new-builds)
+                  (sleep 30)))
+              (sleep 3)))))))
 
 (define* (build-log-procedure lgr #:optional build-id)
   (lambda (level . components)
diff --git a/guix-build-coordinator/utils.scm b/guix-build-coordinator/utils.scm
index 8e42d02..9127ce8 100644
--- a/guix-build-coordinator/utils.scm
+++ b/guix-build-coordinator/utils.scm
@@ -589,6 +589,10 @@ References: ~a~%"
         (enq! queue args)
         (signal-condition-variable job-available)))
 
+    (define (count-threads)
+      (with-mutex queue-mutex
+        (hash-count (const #t) running-job-args)))
+
     (define (count-jobs)
       (with-mutex queue-mutex
         (+ (q-length queue)
@@ -705,7 +709,7 @@ References: ~a~%"
            (sleep 15)
            (start-new-threads-if-necessary (get-thread-count))))))
 
-    (values process-job count-jobs list-jobs)))
+    (values process-job count-jobs count-threads list-jobs)))
 
 (define %random-state
   (seed->random-state (+ (ash (cdr (gettimeofday)) 32) (getpid))))