Update Guile Knots

3 files changed, 80 insertions, 78 deletions

diff --git a/guix-dev.scm b/guix-dev.scm
index b2e6da3..8d3151c 100644
--- a/guix-dev.scm
+++ b/guix-dev.scm
@@ -46,7 +46,7 @@
              (srfi srfi-1))
 
 (define guile-knots
-  (let ((commit "dcb56ee2c5ac3e283cb46841766e7282f3c2c52e")
+  (let ((commit "d572f591a3c136bfc7b23160e16381c92588f8d9")
         (revision "1"))
     (package
     (name "guile-knots")
@@ -58,7 +58,7 @@
                     (commit commit)))
               (sha256
                (base32
-                "04z48572canx35hl0kfli3pf3g3m6184zvmnpyg1rbwla6g5z1fk"))
+                "0g85frfniblxb2cl81fg558ic3cxvla7fvml08scjgbbxn8151gv"))
               (file-name (string-append name "-" version "-checkout"))))
     (build-system gnu-build-system)
     (native-inputs
diff --git a/nar-herder/cached-compression.scm b/nar-herder/cached-compression.scm
index 1ac6d96..0bc0a82 100644
--- a/nar-herder/cached-compression.scm
+++ b/nar-herder/cached-compression.scm
@@ -32,8 +32,8 @@
   #:use-module (fibers channels)
   #:use-module (fibers operations)
   #:use-module (knots timeout)
+  #:use-module (knots thread-pool)
   #:use-module (knots non-blocking)
-  #:use-module (knots worker-threads)
   #:use-module (web uri)
   #:use-module (web client)
   #:use-module (web response)
diff --git a/nar-herder/database.scm b/nar-herder/database.scm
index 239a7e7..f41ba21 100644
--- a/nar-herder/database.scm
+++ b/nar-herder/database.scm
@@ -27,7 +27,7 @@
   #:use-module (web uri)
   #:use-module (sqlite3)
   #:use-module (fibers)
-  #:use-module (knots worker-threads)
+  #:use-module (knots thread-pool)
   #:use-module (prometheus)
   #:use-module (guix store)
   #:use-module (guix narinfo)
@@ -78,12 +78,12 @@
             database-insert-scheduled-cached-narinfo-removal))
 
 (define-record-type <database>
-  (make-database database-file reader-thread-set writer-thread-set
+  (make-database database-file reader-thread-pool writer-thread-pool
                  metrics-registry)
   database?
   (database-file          database-file)
-  (reader-thread-set      database-reader-thread-set)
-  (writer-thread-set      database-writer-thread-set)
+  (reader-thread-pool      database-reader-thread-pool)
+  (writer-thread-pool      database-writer-thread-pool)
   (metrics-registry       database-metrics-registry))
 
 (define* (db-open database
@@ -296,8 +296,11 @@ CREATE UNIQUE INDEX IF NOT EXISTS
 
       (sqlite-close db)))
 
-  (let ((reader-thread-set
-         (make-worker-thread-set
+  (let ((reader-thread-pool
+         (make-thread-pool
+          ;; Use a minimum of 2 and a maximum of 8 threads
+          reader-threads
+          #:thread-initializer
           (lambda ()
             (let ((db
                    (db-open database-file #:write? #f)))
@@ -308,15 +311,13 @@ CREATE UNIQUE INDEX IF NOT EXISTS
                  (simple-format #f "PRAGMA mmap_size=~A;"
                                 (number->string mmap-size))))
               (list db)))
-          #:destructor
+          #:thread-destructor
           (lambda (db)
             (sqlite-close db))
-          #:lifetime 50000
+          #:thread-lifetime 50000
           #:expire-on-exception? #t
           #:name "db r"
 
-          ;; Use a minimum of 2 and a maximum of 8 threads
-          #:parallelism reader-threads
           #:delay-logger (let ((delay-metric
                                 (make-histogram-metric
                                  metrics-registry
@@ -342,10 +343,13 @@ CREATE UNIQUE INDEX IF NOT EXISTS
                 proc)
                (current-error-port))))))
 
-        (writer-thread-set
+        (writer-thread-pool
          (if readonly?
              #f
-             (make-worker-thread-set
+             (make-thread-pool
+              ;; SQLite doesn't support parallel writes
+              1
+              #:thread-initializer
               (lambda ()
                 (let ((db
                        (db-open database-file)))
@@ -357,18 +361,16 @@ CREATE UNIQUE INDEX IF NOT EXISTS
                      (simple-format #f "PRAGMA mmap_size=~A;"
                                     (number->string mmap-size))))
                   (list db)))
-              #:destructor
+              #:thread-destructor
               (lambda (db)
                 (db-optimize db
                              database-file)
 
                 (sqlite-close db))
-              #:lifetime 500
+              #:thread-lifetime 500
               #:expire-on-exception? #t
               #:name "db w"
 
-              ;; SQLite doesn't support parallel writes
-              #:parallelism 1
               #:delay-logger (let ((delay-metric
                                     (make-histogram-metric
                                      metrics-registry
@@ -395,8 +397,8 @@ CREATE UNIQUE INDEX IF NOT EXISTS
                    (current-error-port))))))))
 
     (make-database database-file
-                   reader-thread-set
-                   writer-thread-set
+                   reader-thread-pool
+                   writer-thread-pool
                    metrics-registry)))
 
 (define (update-database-metrics! database)
@@ -445,8 +447,8 @@ PRAGMA optimize;")))
 (define (database-optimize database)
   (retry-on-error
    (lambda ()
-     (call-with-worker-thread
-      (database-writer-thread-set database)
+     (call-with-thread
+      (database-writer-thread-pool database)
       (lambda (db)
         (db-optimize
          db
@@ -580,10 +582,10 @@ PRAGMA optimize;")))
 
             (cons duration-seconds vals))))))
 
-  (match (call-with-worker-thread
+  (match (call-with-thread
           ((if readonly?
-               database-reader-thread-set
-               database-writer-thread-set)
+               database-reader-thread-pool
+               database-writer-thread-pool)
            database)
           (lambda (db)
             (if (%current-transaction-proc)
@@ -604,8 +606,8 @@ PRAGMA optimize;")))
 
       (sqlite-close db)))
 
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (sqlite-exec
       db
@@ -958,8 +960,8 @@ DELETE FROM narinfo_tags WHERE narinfo_id = :narinfo_id"
    database
    "select_narinfo"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -990,8 +992,8 @@ WHERE id = :id"
    database
    "select_narinfo_by_hash"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1023,8 +1025,8 @@ WHERE substr(store_path, 12, 32) = :hash"
    database
    "select_narinfo_contents_by_hash"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1045,8 +1047,8 @@ SELECT id, contents FROM narinfos WHERE substr(store_path, 12, 32) = :hash"
              (values #f #f)))))))))
 
 (define (database-count-recent-changes database)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1062,8 +1064,8 @@ SELECT COUNT(*) FROM recent_changes"
          result)))))
 
 (define* (database-select-recent-changes database after-date #:key (limit 8192))
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1092,8 +1094,8 @@ LIMIT :limit"
          result)))))
 
 (define (database-select-latest-recent-change-datetime database)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1110,8 +1112,8 @@ SELECT datetime FROM recent_changes ORDER BY datetime DESC LIMIT 1"
          result)))))
 
 (define (database-get-recent-changes-id-for-deletion database limit)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1132,8 +1134,8 @@ SELECT id FROM recent_changes ORDER BY datetime DESC LIMIT 1 OFFSET :offset"
          result)))))
 
 (define (database-delete-recent-changes-with-id-below database id)
-  (call-with-worker-thread
-   (database-writer-thread-set database)
+  (call-with-thread
+   (database-writer-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1152,8 +1154,8 @@ DELETE FROM recent_changes WHERE id < :id"
        (changes db)))))
 
 (define (database-select-narinfo-for-file database narinfo-file-url)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1187,8 +1189,8 @@ WHERE narinfo_files.url = :url"
    database
    "select_narinfo_files"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1222,8 +1224,8 @@ WHERE substr(narinfos.store_path, 12, 32) = :hash"
    database
    "select_narinfo_files_by_narinfo_id"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1253,8 +1255,8 @@ WHERE narinfos.id = :narinfo_id"
             result)))))))
 
 (define (database-fold-all-narinfo-files database proc init)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1287,8 +1289,8 @@ FROM narinfo_files"
    '()))
 
 (define (database-count-narinfo-files database)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1308,8 +1310,8 @@ SELECT COUNT(*) FROM narinfo_files"
                                              narinfo-id
                                              size
                                              compression)
-  (call-with-worker-thread
-   (database-writer-thread-set database)
+  (call-with-thread
+   (database-writer-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1339,8 +1341,8 @@ INSERT INTO cached_narinfo_files (
    database
    "select_cached_narinfo_file_by_hash"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1376,8 +1378,8 @@ WHERE substr(narinfos.store_path, 12, 32) = :hash
    database
    "select_cached_narinfo_file_by_narinfo_id"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1418,8 +1420,8 @@ WHERE narinfo_id = :narinfo_id"
    database
    "select_cached_narinfo_file_by_narinfo_id_and_compression"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1455,8 +1457,8 @@ WHERE narinfo_id = :narinfo_id
 (define (database-fold-cached-narinfo-files database
                                             proc
                                             init)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1484,8 +1486,8 @@ INNER JOIN narinfos
          result-list)))))
 
 (define (database-remove-cached-narinfo-file database narinfo-id compression)
-  (call-with-worker-thread
-   (database-writer-thread-set database)
+  (call-with-thread
+   (database-writer-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1509,8 +1511,8 @@ WHERE narinfo_id = :narinfo_id
    database
    "select_scheduled_narinfo_removal"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1541,8 +1543,8 @@ WHERE narinfo_id = :narinfo_id"
    database
    "select_scheduled_narinfo_removal"
    (lambda ()
-     (call-with-worker-thread
-      (database-reader-thread-set database)
+     (call-with-thread
+      (database-reader-thread-pool database)
       (lambda (db)
         (let ((statement
                (sqlite-prepare
@@ -1569,8 +1571,8 @@ WHERE cached_narinfo_file_id = :cached_narinfo_file_id"
 
 (define (database-delete-scheduled-cached-narinfo-removal database
                                                           cached-narinfo-file-id)
-  (call-with-worker-thread
-   (database-writer-thread-set database)
+  (call-with-thread
+   (database-writer-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1590,8 +1592,8 @@ RETURNING 1"
          result)))))
 
 (define (database-select-oldest-scheduled-cached-narinfo-removal database)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1630,8 +1632,8 @@ LIMIT 1"
          result)))))
 
 (define (database-count-scheduled-cached-narinfo-removal database)
-  (call-with-worker-thread
-   (database-reader-thread-set database)
+  (call-with-thread
+   (database-reader-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -1650,8 +1652,8 @@ SELECT COUNT(*) FROM scheduled_cached_narinfo_removal"
 (define (database-insert-scheduled-cached-narinfo-removal database
                                                           cached-narinfo-file-id
                                                           removal-datetime)
-  (call-with-worker-thread
-   (database-writer-thread-set database)
+  (call-with-thread
+   (database-writer-thread-pool database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare