Process revisions in parallel when queuing builds

1 files changed, 87 insertions, 40 deletions

diff --git a/scripts/guix-build-coordinator-queue-builds-from-guix-data-service.in b/scripts/guix-build-coordinator-queue-builds-from-guix-data-service.in
index 7069dec..1b4913a 100644
--- a/scripts/guix-build-coordinator-queue-builds-from-guix-data-service.in
+++ b/scripts/guix-build-coordinator-queue-builds-from-guix-data-service.in
@@ -32,6 +32,9 @@
              (ice-9 threads)
              (ice-9 textual-ports)
              (rnrs bytevectors)
+             (oop goops)
+             (logging logger)
+             (logging port-log)
              (json)
              (web uri)
              (web client)
@@ -68,7 +71,7 @@
      (let-values (((response body) (http-get uri)))
        (json-string->scm (utf8->string body))))
    #:times 6
-   #:delay 30))
+   #:delay (+ 15 (random 30))))
 
 (define (all-repository-ids guix-data-service)
   (let ((data (guix-data-service-request guix-data-service
@@ -107,11 +110,12 @@
                  (assoc-ref data "revisions")))))
 
 (define (record-revision-as-processed processed-commits-file commit)
-  (let ((port (open-file processed-commits-file "a")))
-    (simple-format port "~A\n" commit)
-    (close-port port))
+  (monitor
+   (let ((port (open-file processed-commits-file "a")))
+     (simple-format port "~A\n" commit)
+     (close-port port))
 
-  (hash-set! processed-commits-hash commit #t))
+   (hash-set! processed-commits-hash commit #t)))
 
 (define (channel-instance-derivations-for-commit guix-data-service
                                                  commit
@@ -150,7 +154,8 @@
                           #t))
              derivations)))
 
-(define* (submit-build coordinator guix-data-service derivation #:key (priority 0))
+(define* (submit-build coordinator guix-data-service derivation
+                       #:key (priority 0) (log-prefix ""))
   (retry-on-error
    (lambda ()
      (let ((response
@@ -167,10 +172,9 @@
        (let ((no-build-submitted-response
               (assoc-ref response "no-build-submitted")))
          (if no-build-submitted-response
-             (simple-format #t "skipped: ~A\n"
-                            no-build-submitted-response)
-             (simple-format #t "build submitted as ~A\n"
-                            (assoc-ref response "build-submitted"))))))
+             (log-msg 'DEBUG log-prefix "skipped: " no-build-submitted-response)
+             (log-msg 'DEBUG log-prefix "build submitted as "
+                      (assoc-ref response "build-submitted"))))))
    ;; The TTL Guix uses for transient failures fetching substitutes is 10
    ;; minutes, so we need to retry for longer than that
    #:times 30
@@ -183,12 +187,14 @@
                                      priority-for-derivation
                                      #:key (submit-builds-for-channel-instances?
                                             #t))
-  (simple-format #t "looking at revision ~A\n" commit)
+  (log-msg 'INFO "looking at revision " commit)
   (par-for-each
    (match-lambda
      ((system . target)
       (when (string=? target "none")
         (when submit-builds-for-channel-instances?
+          (log-msg 'INFO "requesting channel instance derivations for "
+                   system " (" commit ")")
           (for-each (lambda (derivation)
                       (submit-build coordinator
                                     guix-data-service
@@ -196,7 +202,11 @@
                                     #:priority
                                     (priority-for-derivation 'channel-instance
                                                              system
-                                                             target)))
+                                                             target)
+                                    #:log-prefix
+                                    (simple-format #f "channel instance (~A): ~A: "
+                                                   system
+                                                   derivation)))
                     (channel-instance-derivations-for-commit guix-data-service
                                                              commit
                                                              system))))
@@ -207,11 +217,15 @@
                      ;; Only request derivations in one thread at a time, just
                      ;; in cause doing this in parallel could lead to timeouts
                      (monitor
+                      (log-msg 'INFO "requesting package derivations for "
+                               system "=>" target " (" commit ")")
                       (package-derivations-for-commit guix-data-service
                                                       commit
                                                       #:system system
                                                       #:target target)))))
 
+        (log-msg 'INFO "submitting package builds for "
+                 system "=>" target " (" commit ")")
         (for-each (lambda (derivation)
                     (submit-build coordinator
                                   guix-data-service
@@ -219,7 +233,10 @@
                                   #:priority
                                   (priority-for-derivation 'package
                                                            system
-                                                           target)))
+                                                           target)
+                                  #:log-prefix
+                                  (simple-format #f "package (~A=>~A): ~A: "
+                                                 system target derivation)))
                   unprocessed-package-derivations)
         (record-derivations-as-processed unprocessed-package-derivations))))
    systems-and-targets))
@@ -255,6 +272,11 @@
                   (alist-cons 'guix-data-service
                               arg
                               (alist-delete 'guix-data-service result))))
+        (option '("threads") #t #f
+                (lambda (opt name arg result)
+                  (alist-cons 'threads
+                              (string->number arg)
+                              (alist-delete 'threads result))))
         (option '("processed-commits-file") #t #f
                 (lambda (opt name arg result)
                   (alist-cons 'processed-commits-file
@@ -265,7 +287,8 @@
   '((coordinator                        . "http://127.0.0.1:8746")
     (guix-data-service                  . "https://data.guix.gnu.org")
     (processed-commits-file             . "processed-revisions")
-    (fetch-channel-instance-derivations . #t)))
+    (fetch-channel-instance-derivations . #t)
+    (threads                            . 4)))
 
 (define (parse-options options defaults args)
   (args-fold
@@ -291,7 +314,17 @@
          (guix-data-service
           (assq-ref opts 'guix-data-service))
          (processed-commits-file
-          (assq-ref opts 'processed-commits-file)))
+          (assq-ref opts 'processed-commits-file))
+
+         (lgr      (make <logger>))
+         (port-log (make <port-log>
+                     #:port (current-output-port)
+                     #:formatter
+                     (lambda (lvl time str)
+                       (format #f "~a (~5a): ~a~%"
+                               (strftime "%F %H:%M:%S" (localtime time))
+                               lvl
+                               str)))))
 
     (define (priority-for-derivation type system target)
       (cond
@@ -313,6 +346,10 @@
        (else
         0)))
 
+    (add-handler! lgr port-log)
+    (open-log! lgr)
+    (set-default-logger! lgr)
+
     (unless systems-and-targets
       (simple-format (current-error-port)
                      "error: you must specify at least one system to fetch builds for\n")
@@ -343,30 +380,40 @@
                                      #t)))
                       commits)))))
 
-    (while #t
-      (for-each
-       (lambda (repository-id)
-         (for-each
-          (lambda (branch)
-            (for-each
-             (lambda (commit)
-               (submit-builds-for-revision
-                (assq-ref opts 'coordinator)
-                guix-data-service
-                commit
-                systems-and-targets
-                priority-for-derivation
-                #:submit-builds-for-channel-instances?
-                (assq-ref opts 'fetch-channel-instance-derivations))
-               (record-revision-as-processed processed-commits-file commit))
-             (unseen-revisions guix-data-service
-                               repository-id
-                               branch)))
-          (all-repository-branches guix-data-service
-                                   repository-id)))
-       (all-repository-ids guix-data-service))
-
-      (simple-format #t "waiting before checking for new revisions...\n")
-      (sleep 60))))
+    (let*-values
+        (((process-job-with-queue count-jobs count-threads list-jobs)
+          (create-work-queue
+           (assq-ref opts 'threads)
+           (lambda (commit)
+             (submit-builds-for-revision
+              (assq-ref opts 'coordinator)
+              guix-data-service
+              commit
+              systems-and-targets
+              priority-for-derivation
+              #:submit-builds-for-channel-instances?
+              (assq-ref opts 'fetch-channel-instance-derivations))
+             (record-revision-as-processed processed-commits-file commit)))))
+
+      (while #t
+        (for-each
+         (lambda (repository-id)
+           (for-each
+            (lambda (branch)
+              (for-each
+               process-job-with-queue
+               (lset-difference
+                string=?
+                (unseen-revisions guix-data-service
+                                  repository-id
+                                  branch)
+                (map car (list-jobs)))))
+            (all-repository-branches guix-data-service
+                                     repository-id)))
+         (all-repository-ids guix-data-service))
+
+        (log-msg 'INFO "jobs: " (count-jobs) " threads: " (count-threads))
+        (log-msg 'INFO "waiting before checking for new revisions...")
+        (sleep 60)))))
 
 (main)