Move build allocation complexity out of the datastore

And in to the coordinator module. This will make adding more datastore's
easier.

1 files changed, 112 insertions, 156 deletions

diff --git a/guix-build-coordinator/datastore/sqlite.scm b/guix-build-coordinator/datastore/sqlite.scm
index bc3dd58..8515da8 100644
--- a/guix-build-coordinator/datastore/sqlite.scm
+++ b/guix-build-coordinator/datastore/sqlite.scm
@@ -73,7 +73,10 @@
             datastore-count-allocated-builds
             datastore-agent-requested-systems
             datastore-update-agent-requested-systems
-            datastore-allocate-builds-to-agent
+            datastore-fetch-build-to-allocate
+            datastore-insert-to-allocated-builds
+            datastore-remove-builds-from-plan
+            datastore-select-allocated-builds
             datastore-list-allocation-plan-builds))
 
 (define-class <sqlite-datastore> (<abstract-datastore>)
@@ -2141,17 +2144,17 @@ INSERT INTO build_allocation_agent_requested_systems (agent_id, system) VALUES "
 
          #t))))
 
-(define-method (datastore-allocate-builds-to-agent
+(define-method (datastore-fetch-build-to-allocate
                 (datastore <sqlite-datastore>)
-                agent-id
-                max-builds
-                deprecated-requested-count)
-  (define (fetch-build db)
-    (let ((statement
-           (sqlite-prepare
-            db
-            ;; This needs to guard against the plan being out of date
-            "
+                agent-id)
+  (datastore-call-with-transaction
+   datastore
+   (lambda (db)
+     (let ((statement
+            (sqlite-prepare
+             db
+             ;; This needs to guard against the plan being out of date
+             "
 SELECT builds.uuid, builds.derivation_name
 FROM builds
 INNER JOIN build_allocation_plan
@@ -2166,11 +2169,11 @@ WHERE build_allocation_plan.agent_id = :agent_id
   AND builds.canceled = 0
   AND builds.uuid NOT IN (SELECT build_id FROM allocated_builds)
 ORDER BY build_allocation_plan.ordering DESC"
-            #:cache? #t))
-          (output-conflicts-statement
-           (sqlite-prepare
-            db
-            "
+             #:cache? #t))
+           (output-conflicts-statement
+            (sqlite-prepare
+             db
+             "
 SELECT 1
 FROM derivation_outputs AS build_derivation_outputs
 INNER JOIN allocated_builds
@@ -2183,165 +2186,118 @@ INNER JOIN derivation_outputs AS allocated_builds_derivation_outputs
 WHERE build_derivation_outputs.derivation_name = :derivation_name
   AND build_derivation_outputs.output =
       allocated_builds_derivation_outputs.output"
-            #:cache? #t)))
+             #:cache? #t)))
 
-      (sqlite-bind-arguments
-       statement
-       #:agent_id agent-id)
+       (sqlite-bind-arguments
+        statement
+        #:agent_id agent-id)
 
-      (let ((builds
-             (sqlite-fold
-              (lambda (row result)
-                (cons
-                 (match row
-                   (#f #f)
-                   (#(uuid derivation_name)
-                    `((uuid            . ,uuid)
-                      (derivation-name . ,derivation_name))))
-                 result))
-              '()
-              statement)))
-        (sqlite-reset statement)
+       (let ((builds
+              (sqlite-fold
+               (lambda (row result)
+                 (cons
+                  (match row
+                    (#f #f)
+                    (#(uuid derivation_name)
+                     `((uuid            . ,uuid)
+                       (derivation-name . ,derivation_name))))
+                  result))
+               '()
+               statement)))
+         (sqlite-reset statement)
 
-        (find (lambda (build-details)
-                (sqlite-bind-arguments
-                 output-conflicts-statement
-                 #:derivation_name (assq-ref build-details
-                                             'derivation-name))
+         (find (lambda (build-details)
+                 (sqlite-bind-arguments
+                  output-conflicts-statement
+                  #:derivation_name (assq-ref build-details
+                                              'derivation-name))
 
-                (let ((result (sqlite-step output-conflicts-statement)))
-                  (sqlite-reset output-conflicts-statement)
+                 (let ((result (sqlite-step output-conflicts-statement)))
+                   (sqlite-reset output-conflicts-statement)
 
-                  (match result
-                    (#f #t)
-                    (_ #f))))
-              builds))))
+                   (match result
+                     (#f #t)
+                     (_ #f))))
+               builds))))))
 
-  (define (insert-to-allocated-builds db agent-id build-ids)
-    (sqlite-exec
-     db
-     (string-append
-      "
+(define-method (datastore-insert-to-allocated-builds
+                (datastore <sqlite-datastore>)
+                agent-id
+                build-ids)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-channel)
+   (lambda (db)
+     (sqlite-exec
+      db
+      (string-append
+       "
 INSERT INTO allocated_builds (build_id, agent_id) VALUES "
-      (string-join
-       (map (lambda (build-id)
-              (simple-format
-               #f
-               "('~A', '~A')"
-               build-id
-               agent-id))
-            build-ids)
-       ", ")
-      ";")))
+       (string-join
+        (map (lambda (build-id)
+               (simple-format
+                #f
+                "('~A', '~A')"
+                build-id
+                agent-id))
+             build-ids)
+        ", ")
+       ";")))))
 
-  (define (remove-builds-from-plan db build-ids)
-    (sqlite-exec
-     db
-     (string-append
-      "
+(define-method (datastore-remove-builds-from-plan
+                (datastore <sqlite-datastore>)
+                build-ids)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-channel)
+   (lambda (db)
+     (sqlite-exec
+      db
+      (string-append
+       "
 DELETE FROM build_allocation_plan
 WHERE build_id IN ("
-      (string-join
-       (map (lambda (build-id)
-              (string-append "'" build-id "'"))
-            build-ids)
-       ", ")
-      ")")))
-
-  (define (allocate-one-build db agent-id)
-    (let ((build-details (fetch-build db)))
-      (if build-details
-          (let ((build-id (assq-ref build-details 'uuid)))
-            (insert-to-allocated-builds db agent-id (list build-id))
-            (remove-builds-from-plan db (list build-id))
-            build-details)
-          #f)))
-
-  (define (allocate-several-builds db agent-id count)
-    (let loop ((builds '()))
-      (if (= (length builds) count)
-          builds
-          (let ((build-details (allocate-one-build db agent-id)))
-            (if build-details
-                (loop (cons build-details builds))
-                builds)))))
-
-  (define (select-allocated-builds db agent-id)
-    (let ((statement
-           (sqlite-prepare
-            db
-            "
+       (string-join
+        (map (lambda (build-id)
+               (string-append "'" build-id "'"))
+             build-ids)
+        ", ")
+       ")")))))
+
+(define-method (datastore-select-allocated-builds
+                (datastore <sqlite-datastore>)
+                agent-id)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-reader-channel)
+   (lambda (db)
+     (let ((statement
+            (sqlite-prepare
+             db
+             "
 SELECT builds.uuid, builds.derivation_name
 FROM allocated_builds
 INNER JOIN builds
   ON allocated_builds.build_id = builds.uuid
 WHERE agent_id = :agent_id"
-            #:cache? #t)))
-
-      (sqlite-bind-arguments
-       statement
-       #:agent_id agent-id)
+             #:cache? #t)))
 
-      (let ((result
-             (sqlite-fold
-              (lambda (row result)
-                (cons
-                 (match row
-                   (#(uuid derivation_name)
-                    `((uuid            . ,uuid)
-                      (derivation-name . ,derivation_name))))
-                 result))
-              '()
-              statement)))
+       (sqlite-bind-arguments
+        statement
+        #:agent_id agent-id)
 
-        (sqlite-reset statement)
+       (let ((result
+              (sqlite-fold
+               (lambda (row result)
+                 (cons
+                  (match row
+                    (#(uuid derivation_name)
+                     `((uuid            . ,uuid)
+                       (derivation-name . ,derivation_name))))
+                  result))
+               '()
+               statement)))
 
-        result)))
+         (sqlite-reset statement)
 
-  (datastore-call-with-transaction
-   datastore
-   (lambda (db)
-     (let* ((initially-allocated-builds
-             (select-allocated-builds db agent-id))
-            (start-count
-             (length initially-allocated-builds))
-            (target-count (or max-builds
-                              (+ start-count
-                                 deprecated-requested-count))))
-       (if (< start-count target-count)
-           (let ((new-builds
-                  (allocate-several-builds db
-                                           agent-id
-                                           (- target-count start-count))))
-             (unless (null? new-builds)
-               (let ((allocation-plan-metric
-                      (metrics-registry-fetch-metric
-                       (slot-ref datastore 'metrics-registry)
-                       "build_allocation_plan_total")))
-                 (for-each
-                  (match-lambda
-                    ((agent-id . count)
-                     (metric-set allocation-plan-metric
-                                 count
-                                 #:label-values
-                                 `((agent_id . ,agent-id)))))
-                  (datastore-count-build-allocation-plan-entries datastore))))
-
-             ;; Previously allocate builds just returned newly allocated
-             ;; builds, but if max-builds is provided, return all the
-             ;; builds. This means the agent can handle this in a idempotent
-             ;; manor.
-             (if max-builds
-                 (append initially-allocated-builds
-                         new-builds)
-                 new-builds))
-           ;; Previously allocate builds just returned newly allocated builds,
-           ;; but if max-builds is provided, return all the builds. This means
-           ;; the agent can handle this in a idempotent manor.
-           (if max-builds
-               initially-allocated-builds
-               '()))))
-   #:duration-metric-name "allocate_builds_to_agent"))
+         result)))))
 
 (define-method (datastore-list-allocation-plan-builds
                 (datastore <sqlite-datastore>)