Rewrite deleting unreferenced derivations

Use fibers more, leaning in on the non-blocking use of Squee for parallelism.

guix-data-service/data-deletion.scm

@@ -22,6 +22,7 @@
   #:use-module (ice-9 threads)
   #:use-module (squee)
   #:use-module (fibers)
+  #:use-module (fibers channels)
   #:use-module (guix-data-service utils)
   #:use-module (guix-data-service database)
   #:use-module (guix-data-service model git-branch)
@@ -538,7 +539,10 @@ DELETE FROM derivations WHERE id = $1"
 
        1)))
 
-  (define (delete-batch conn connection-pool)
+  (define deleted-count 0)
+  (define channel (make-channel))
+
+  (define (delete-batch conn)
     (let* ((derivations
             (with-time-logging "fetching batch of derivations"
               (map car
@@ -566,77 +570,85 @@ WHERE NOT EXISTS (
 ) LIMIT $1"
                     (list (number->string batch-size))))))
            (derivations-count (length derivations)))
-      (let ((deleted-count 0))
-        (with-time-logging
-            (simple-format #f
-                           "Looking at ~A derivations"
-                           derivations-count)
-          (n-par-for-each
-           8
-           (lambda (derivation-id)
-             (unless (string->number derivation-id)
-               (error
-                (simple-format #f "derivation-id: ~A is not a number"
-                               derivation-id)))
 
-             (let ((val
+      (with-time-logging
-                    (call-with-resource-from-pool connection-pool
+          (simple-format #f "Looking at ~A derivations" derivations-count)
-                      (lambda (conn)
-                        (catch 'psql-query-error
-                          (lambda ()
-                            (with-postgresql-transaction
-                             conn
-                             (lambda (conn)
-                               (exec-query
-                                conn
-                                "
-SET CONSTRAINTS derivations_by_output_details_set_derivation_id_fkey DEFERRED")
 
-                               (exec-query conn "SET LOCAL lock_timeout = '5s';")
+        (set! deleted-count 0)
+        (for-each
+         (lambda (derivation-id)
+           (put-message channel derivation-id))
+         derivations))
 
-                               (maybe-delete-derivation conn
+      (simple-format (current-error-port)
-                                                        derivation-id))))
+                     "Deleted ~A derivations\n"
-                          (lambda (key . args)
+                     deleted-count)
-                            (simple-format
+      deleted-count))
-                             (current-error-port)
-                             "error when attempting to delete derivation: ~A ~A\n"
-                             key args)
-
-                            0))))))
-               (monitor
-                (set! deleted-count
-                      (+ val deleted-count)))))
-           derivations))
-
-        (simple-format (current-error-port)
-                       "Deleted ~A derivations\n"
-                       deleted-count)
-        deleted-count)))
 
   (run-fibers
    (lambda ()
-     (let* ((connection-pool
+     ;; First spawn some fibers to delete the derivations
-             (make-resource-pool
+     (for-each
-              (lambda ()
+      (lambda _
-                (open-postgresql-connection "data-deletion" #f))
+        (spawn-fiber
-              8)))
+         (lambda ()
+           (with-postgresql-connection
+            "data-deletion"
+            (lambda (conn)
+              (let loop ((derivation-id (get-message channel)))
+                (unless (string->number derivation-id)
+                  (error
+                   (simple-format #f "derivation-id: ~A is not a number"
+                                  derivation-id)))
 
-       (with-postgresql-connection
+                (let ((val
-        "data-deletion"
+                       (catch 'psql-query-error
-        (lambda (conn)
+                         (lambda ()
-          (obtain-advisory-transaction-lock
+                           (with-postgresql-transaction
-           conn
+                            conn
-           'delete-unreferenced-derivations)
+                            (lambda (conn)
+                              (exec-query
+                               conn
+                               "
+SET CONSTRAINTS derivations_by_output_details_set_derivation_id_fkey DEFERRED")
 
-          (let loop ((total-deleted 0))
+                              (exec-query conn "SET LOCAL lock_timeout = '5s';")
-            (let ((batch-deleted-count (delete-batch conn connection-pool)))
+
-              (if (eq? 0 batch-deleted-count)
+                              (maybe-delete-derivation conn
-                  (begin
+                                                       derivation-id))))
-                    (with-time-logging
+                         (lambda (key . args)
-                        "Deleting unused derivation_source_files entries"
+                           (simple-format
-                      (delete-unreferenced-derivations-source-files conn))
+                            (current-error-port)
-                    (simple-format
+                            "error when attempting to delete derivation: ~A ~A\n"
-                     (current-output-port)
+                            key args)
-                     "Finished deleting derivations, deleted ~A in total\n"
+
-                     total-deleted))
+                           0))))
-                  (loop (+ total-deleted batch-deleted-count)))))))))))
+
+                  ;; This is safe as all fibers are in the same
+                  ;; thread and cooperative.
+                  (set! deleted-count
+                        (+ val deleted-count)))
+                (loop (get-message channel))))))))
+      (iota 12))
+
+     (with-postgresql-connection
+      "data-deletion"
+      (lambda (conn)
+        (obtain-advisory-transaction-lock
+         conn
+         'delete-unreferenced-derivations)
+
+        (let loop ((total-deleted 0))
+          (let ((batch-deleted-count (delete-batch conn)))
+            (if (eq? 0 batch-deleted-count)
+                (begin
+                  (with-time-logging
+                   "Deleting unused derivation_source_files entries"
+                   (delete-unreferenced-derivations-source-files conn))
+                  (simple-format
+                   (current-output-port)
+                   "Finished deleting derivations, deleted ~A in total\n"
+                   total-deleted))
+                (loop (+ total-deleted batch-deleted-count))))))))
+   #:hz 0
+   #:parallelism 1))