Log the current state of the reader threads

1 files changed, 135 insertions, 111 deletions

diff --git a/guix-build-coordinator/datastore/sqlite.scm b/guix-build-coordinator/datastore/sqlite.scm
index dda6935..3e20e6f 100644
--- a/guix-build-coordinator/datastore/sqlite.scm
+++ b/guix-build-coordinator/datastore/sqlite.scm
@@ -107,7 +107,9 @@
 (define-class <sqlite-datastore> (<abstract-datastore>)
   database-file
   worker-reader-thread-channel
+  worker-reader-thread-proc-vector
   worker-writer-thread-channel
+  worker-writer-thread-proc-vector
   metrics-registry)
 
 (define* (sqlite-datastore database-uri
@@ -137,118 +139,129 @@
     (slot-set! datastore 'database-file database-file)
     (slot-set! datastore 'metrics-registry metrics-registry)
 
-    (slot-set!
-     datastore
-     'worker-writer-thread-channel
-     (make-worker-thread-channel
-      (lambda ()
-        (let ((db
-               (db-open database-file)))
-          (sqlite-exec db "PRAGMA busy_timeout = 0;")
-          (sqlite-exec db "PRAGMA synchronous = NORMAL;")
-          (sqlite-exec db "PRAGMA temp_store = MEMORY;")
-          (sqlite-exec db "PRAGMA foreign_keys = ON;")
-
-          (list db)))
-      #:name "ds write"
-      #:destructor
-      (let ((writer-thread-destructor-counter
-             (make-gauge-metric metrics-registry
-                                "datastore_writer_thread_close_total")))
-        (lambda (db)
-          (db-optimize db
-                       database-file
-                       metrics-registry
-                       #:maybe-truncate-wal? #f)
-
-          (metric-increment writer-thread-destructor-counter)
-          (sqlite-close db)))
-      #:lifetime 500
-      #:expire-on-exception? #t
-
-      ;; SQLite doesn't support parallel writes
-      #:parallelism 1
-      #:delay-logger (let ((delay-metric
-                            (make-histogram-metric
-                             metrics-registry
-                             "datastore_write_delay_seconds")))
-                       (lambda (seconds-delayed)
-                         (metric-observe delay-metric
-                                         ;; TODO exact->inexact to work around
-                                         ;; a bug in guile-prometheus where
-                                         ;; the metric sum will output in the
-                                         ;; exact form including the /q
-                                         (exact->inexact seconds-delayed))
-                         (log-delay "datastore write" seconds-delayed)
-                         (when (> seconds-delayed 1)
-                           (format
-                            (current-error-port)
-                            "warning: database write delayed by ~1,2f seconds~%"
-                            seconds-delayed))))
-      #:duration-logger
-      (lambda (duration proc)
-        (when (> duration 10)
-          (format
-           (current-error-port)
-           "warning: database write took ~1,2f seconds (~a)~%"
-           duration
-           proc)))
-      #:log-exception? worker-thread-log-exception?))
-
-    (slot-set!
-     datastore
-     'worker-reader-thread-channel
-     (make-worker-thread-channel
-      (lambda ()
-        (let ((db
-               (db-open database-file #:write? #f)))
-          (sqlite-exec db "PRAGMA temp_store = MEMORY;")
-          (sqlite-exec db "PRAGMA busy_timeout = 0;")
-          (sqlite-exec db "PRAGMA cache_size = -16000;")
-
-          (list db)))
-      #:name "ds read"
-      #:destructor
-      (let ((reader-thread-destructor-counter
-             (make-gauge-metric metrics-registry
-                                "datastore_reader_thread_close_total")))
-        (lambda (db)
-          (metric-increment reader-thread-destructor-counter)
-          (sqlite-close db)))
-      #:lifetime 50000
-      #:expire-on-exception? #t
-
-      ;; Use a minimum of 8 and a maximum of 16 threads
-      #:parallelism
-      (min (max (current-processor-count)
-                8)
-           16)
-      #:delay-logger (let ((delay-metric
-                            (make-histogram-metric
+    (let ((channel
+           proc-vector
+           (make-worker-thread-channel
+            (lambda ()
+              (let ((db
+                     (db-open database-file)))
+                (sqlite-exec db "PRAGMA busy_timeout = 0;")
+                (sqlite-exec db "PRAGMA synchronous = NORMAL;")
+                (sqlite-exec db "PRAGMA temp_store = MEMORY;")
+                (sqlite-exec db "PRAGMA foreign_keys = ON;")
+
+                (list db)))
+            #:name "ds write"
+            #:destructor
+            (let ((writer-thread-destructor-counter
+                   (make-gauge-metric metrics-registry
+                                      "datastore_writer_thread_close_total")))
+              (lambda (db)
+                (db-optimize db
+                             database-file
                              metrics-registry
-                             "datastore_read_delay_seconds")))
-                       (lambda (seconds-delayed)
-                         (metric-observe delay-metric
-                                         ;; TODO exact->inexact to work around
-                                         ;; a bug in guile-prometheus where
-                                         ;; the metric sum will output in the
-                                         ;; exact form including the /q
-                                         (exact->inexact seconds-delayed))
-                         (log-delay "datastore read" seconds-delayed)
-                         (when (> seconds-delayed 1)
-                           (format
-                            (current-error-port)
-                            "warning: database read delayed by ~1,2f seconds~%"
-                            seconds-delayed))))
-      #:duration-logger
-      (lambda (duration proc)
-        (when (> duration 30)
-          (format
-           (current-error-port)
-           "warning: database read took ~1,2f seconds (~a)~%"
-           duration
-           proc)))
-      #:log-exception? worker-thread-log-exception?))
+                             #:maybe-truncate-wal? #f)
+
+                (metric-increment writer-thread-destructor-counter)
+                (sqlite-close db)))
+            #:lifetime 500
+            #:expire-on-exception? #t
+
+            ;; SQLite doesn't support parallel writes
+            #:parallelism 1
+            #:delay-logger (let ((delay-metric
+                                  (make-histogram-metric
+                                   metrics-registry
+                                   "datastore_write_delay_seconds")))
+                             (lambda (seconds-delayed)
+                               (metric-observe delay-metric
+                                               ;; TODO exact->inexact to work around
+                                               ;; a bug in guile-prometheus where
+                                               ;; the metric sum will output in the
+                                               ;; exact form including the /q
+                                               (exact->inexact seconds-delayed))
+                               (log-delay "datastore write" seconds-delayed)
+                               (when (> seconds-delayed 1)
+                                 (format
+                                  (current-error-port)
+                                  "warning: database write delayed by ~1,2f seconds~%"
+                                  seconds-delayed))))
+            #:duration-logger
+            (lambda (duration proc)
+              (when (> duration 10)
+                (format
+                 (current-error-port)
+                 "warning: database write took ~1,2f seconds (~a)~%"
+                 duration
+                 proc)))
+            #:log-exception? worker-thread-log-exception?)))
+
+      (slot-set! datastore
+                 'worker-writer-thread-channel
+                 channel)
+      (slot-set! datastore
+                 'worker-writer-thread-proc-vector
+                 proc-vector))
+
+    (let ((channel
+           proc-vector
+           (make-worker-thread-channel
+            (lambda ()
+              (let ((db
+                     (db-open database-file #:write? #f)))
+                (sqlite-exec db "PRAGMA temp_store = MEMORY;")
+                (sqlite-exec db "PRAGMA busy_timeout = 0;")
+                (sqlite-exec db "PRAGMA cache_size = -16000;")
+
+                (list db)))
+            #:name "ds read"
+            #:destructor
+            (let ((reader-thread-destructor-counter
+                   (make-gauge-metric metrics-registry
+                                      "datastore_reader_thread_close_total")))
+              (lambda (db)
+                (metric-increment reader-thread-destructor-counter)
+                (sqlite-close db)))
+            #:lifetime 50000
+            #:expire-on-exception? #t
+
+            ;; Use a minimum of 8 and a maximum of 16 threads
+            #:parallelism
+            (min (max (current-processor-count)
+                      8)
+                 16)
+            #:delay-logger (let ((delay-metric
+                                  (make-histogram-metric
+                                   metrics-registry
+                                   "datastore_read_delay_seconds")))
+                             (lambda (seconds-delayed)
+                               (metric-observe delay-metric
+                                               ;; TODO exact->inexact to work around
+                                               ;; a bug in guile-prometheus where
+                                               ;; the metric sum will output in the
+                                               ;; exact form including the /q
+                                               (exact->inexact seconds-delayed))
+                               (log-delay "datastore read" seconds-delayed)
+                               (when (> seconds-delayed 1)
+                                 (format
+                                  (current-error-port)
+                                  "warning: database read delayed by ~1,2f seconds~%"
+                                  seconds-delayed))))
+            #:duration-logger
+            (lambda (duration proc)
+              (when (> duration 30)
+                (format
+                 (current-error-port)
+                 "warning: database read took ~1,2f seconds (~a)~%"
+                 duration
+                 proc)))
+            #:log-exception? worker-thread-log-exception?)))
+      (slot-set! datastore
+                 'worker-reader-thread-channel
+                 channel)
+      (slot-set! datastore
+                 'worker-reader-thread-proc-vector
+                 proc-vector))
 
     datastore))
 
@@ -392,6 +405,17 @@ SELECT * FROM dbstat WHERE aggregate = true"
   (spawn-fiber
    (lambda ()
      (while #t
+       (sleep 20)
+       (vector-for-each
+        (lambda (i proc)
+          (simple-format (current-error-port)
+                         "reader thread ~A running: ~A\n"
+                         i proc))
+        (slot-ref datastore 'worker-reader-thread-proc-vector)))))
+
+  (spawn-fiber
+   (lambda ()
+     (while #t
        (sleep (* 60 10))                ; 10 minutes
        (with-exception-handler
            (lambda (exn)