Use more from knots

6 files changed, 68 insertions, 123 deletions

diff --git a/guix-qa-frontpage/database.scm b/guix-qa-frontpage/database.scm
index c44d83a..3bab909 100644
--- a/guix-qa-frontpage/database.scm
+++ b/guix-qa-frontpage/database.scm
@@ -28,16 +28,15 @@
   #:use-module (sqlite3)
   #:use-module (fibers)
   #:use-module (prometheus)
+  #:use-module (knots queue)
+  #:use-module (knots worker-threads)
   #:use-module (guix narinfo)
   #:use-module (guix derivations)
   #:use-module ((guix-build-coordinator utils)
                 #:select (log-delay
                           call-with-delay-logging))
   #:use-module ((guix-build-coordinator utils fibers)
-                #:select (retry-on-error
-                          make-worker-thread-channel
-                          call-with-worker-thread
-                          make-queueing-channel))
+                #:select (retry-on-error))
   #:use-module (guix-qa-frontpage guix-data-service)
   #:export (setup-database
 
@@ -58,14 +57,16 @@
             delete-create-branch-for-issue-log))
 
 (define-record-type <database>
-  (make-database database-file reader-thread-channel writer-thread-channel
+  (make-database database-file reader-thread-set writer-thread-set
+                 writer-thread-set-channel
                  metrics-registry)
   database?
   (database-file          database-file)
-  (reader-thread-channel  database-reader-thread-channel)
-  (writer-thread-channel  database-writer-thread-channel
-                          set-database-writer-thread-channel!)
-  (metrics-registry       database-metrics-registry))
+  (reader-thread-set  database-reader-thread-set)
+  (writer-thread-set  database-writer-thread-set)
+  (writer-thread-set-channel  database-writer-thread-set-channel
+                          set-database-writer-thread-set-channel!)
+  (metrics-registry   database-metrics-registry))
 
 (define* (db-open database
                   #:key (write? #t))
@@ -145,8 +146,8 @@ CREATE TABLE IF NOT EXISTS create_branch_for_issue_logs (
 
     (sqlite-close db))
 
-  (let ((reader-thread-channel
-         (make-worker-thread-channel
+  (let ((reader-thread-set
+         (make-worker-thread-set
           (lambda ()
             (let ((db
                    (db-open database-file #:write? #f)))
@@ -182,8 +183,8 @@ CREATE TABLE IF NOT EXISTS create_branch_for_issue_logs (
           #:log-exception?
           (lambda (exn)
             (not (guix-data-service-error? exn)))))
-        (writer-thread-channel
-         (make-worker-thread-channel
+        (writer-thread-set
+         (make-worker-thread-set
           (lambda ()
             (let ((db
                    (db-open database-file)))
@@ -220,8 +221,9 @@ CREATE TABLE IF NOT EXISTS create_branch_for_issue_logs (
                                 seconds-delayed)))))))
 
     (make-database database-file
-                   reader-thread-channel
-                   writer-thread-channel
+                   reader-thread-set
+                   writer-thread-set
+                   (worker-thread-set-channel writer-thread-set)
                    metrics-registry)))
 
 (define (db-optimize db db-filename)
@@ -247,7 +249,7 @@ PRAGMA optimize;")))
   (retry-on-error
    (lambda ()
      (call-with-worker-thread
-      (database-writer-thread-channel database)
+      (database-writer-thread-set database)
       (lambda (db)
         (db-optimize
          db
@@ -258,10 +260,11 @@ PRAGMA optimize;")))
 (define (database-spawn-fibers database)
   ;; Queue messages to the writer thread, so that they're handled in a first
   ;; come first served manor
-  (set-database-writer-thread-channel!
+  (set-database-writer-thread-set-channel!
    database
-   (make-queueing-channel
-    (database-writer-thread-channel database)))
+   (spawn-queueing-fiber
+    (worker-thread-set-channel
+     (database-writer-thread-set database))))
 
   (spawn-fiber
    (lambda ()
@@ -324,8 +327,8 @@ PRAGMA optimize;")))
 
   (match (call-with-worker-thread
           ((if readonly?
-               database-reader-thread-channel
-               database-writer-thread-channel)
+               database-reader-thread-set
+               database-writer-thread-set)
            database)
           (lambda (db)
             (let ((start-time (get-internal-real-time)))
@@ -346,7 +349,11 @@ PRAGMA optimize;")))
                         duration-seconds)
                        (current-error-port)))
 
-                    (cons duration-seconds vals)))))))
+                    (cons duration-seconds vals))))))
+          #:channel
+          (if readonly?
+              #f
+              (database-writer-thread-set-channel database)))
     ((duration vals ...)
      (apply values vals))))
 
@@ -431,7 +438,7 @@ DELETE FROM cache WHERE key = :key"
 
   (let ((cached-values
          (call-with-worker-thread
-          (database-reader-thread-channel database)
+          (database-reader-thread-set database)
           (lambda (db)
             (let ((statement
                    (sqlite-prepare
@@ -465,11 +472,16 @@ SELECT data, timestamp FROM cache WHERE key = :key"
         (call-with-values
             (lambda ()
               (call-with-worker-thread
-               (database-reader-thread-channel database)
+               (database-reader-thread-set database)
                (lambda (db)
-                 (call-with-delay-logging
-                  proc
-                  #:args args))))
+                 (with-throw-handler #t
+                   (lambda ()
+                     (call-with-delay-logging
+                      proc
+                      #:args args))
+                   (lambda args
+                     (display (backtrace) (current-error-port))
+                     (newline (current-error-port)))))))
           (lambda vals
             (when (if (procedure? store-computed-value?)
                       (apply store-computed-value? vals)
@@ -558,7 +570,7 @@ WHERE category_name = :name AND category_value = :value"
 
 (define (select-from-builds-to-cancel-later database category-name)
   (call-with-worker-thread
-   (database-reader-thread-channel database)
+   (database-reader-thread-set database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
@@ -614,7 +626,7 @@ VALUES (:issue, :log)"
 
 (define (select-create-branch-for-issue-log database issue)
   (call-with-worker-thread
-   (database-reader-thread-channel database)
+   (database-reader-thread-set database)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
diff --git a/guix-qa-frontpage/guix-data-service.scm b/guix-qa-frontpage/guix-data-service.scm
index 40e2560..8540524 100644
--- a/guix-qa-frontpage/guix-data-service.scm
+++ b/guix-qa-frontpage/guix-data-service.scm
@@ -177,7 +177,7 @@
 (define* (guix-data-service-request url #:key (retry-times 0) (retry-delay 5))
   (define (make-request)
     (let ((port
-           (nonblocking-open-socket-for-uri (string->uri url))))
+           (non-blocking-open-socket-for-uri (string->uri url))))
 
       (let ((response
              body
diff --git a/guix-qa-frontpage/manage-patch-branches.scm b/guix-qa-frontpage/manage-patch-branches.scm
index fc389e1..634cefe 100644
--- a/guix-qa-frontpage/manage-patch-branches.scm
+++ b/guix-qa-frontpage/manage-patch-branches.scm
@@ -17,12 +17,9 @@
   #:use-module (guix sets)
   #:use-module (guix memoization)
   #:use-module (guix build utils)
-  #:use-module ((guix build syscalls)
-                #:select (set-thread-name))
-  #:use-module (guix-build-coordinator utils)
-  #:use-module (guix-build-coordinator utils fibers)
   #:use-module ((guix build download) #:select (http-fetch))
   #:use-module ((guix build utils) #:select (with-directory-excursion))
+  #:use-module (knots worker-threads)
   #:use-module (guix-qa-frontpage mumi)
   #:use-module (guix-qa-frontpage database)
   #:use-module (guix-qa-frontpage git-repository)
diff --git a/guix-qa-frontpage/patchwork.scm b/guix-qa-frontpage/patchwork.scm
index 0ebb248..06d18e4 100644
--- a/guix-qa-frontpage/patchwork.scm
+++ b/guix-qa-frontpage/patchwork.scm
@@ -14,6 +14,8 @@
   #:use-module (web client)
   #:use-module (web request)
   #:use-module (web response)
+  #:use-module (knots timeout)
+  #:use-module (knots non-blocking)
   #:use-module ((guix-build-coordinator utils)
                 #:select (call-with-delay-logging))
   #:use-module ((guix-build-coordinator utils fibers)
@@ -83,7 +85,7 @@
          (retry-on-error
           (lambda ()
             (http-request uri
-                          #:port (nonblocking-open-socket-for-uri uri)
+                          #:port (non-blocking-open-socket-for-uri uri)
                           #:decode-body? #f
                           #:streaming? #t))
           #:times 2
diff --git a/guix-qa-frontpage/server.scm b/guix-qa-frontpage/server.scm
index b222969..8ec745f 100644
--- a/guix-qa-frontpage/server.scm
+++ b/guix-qa-frontpage/server.scm
@@ -32,6 +32,8 @@
   #:use-module (fibers)
   #:use-module (fibers scheduler)
   #:use-module (fibers conditions)
+  #:use-module (knots)
+  #:use-module (knots web-server)
   #:use-module (guix store)
   #:use-module (knots web-server)
   #:use-module ((guix build syscalls)
@@ -44,8 +46,6 @@
   #:use-module ((guix-build-coordinator utils)
                 #:select (with-time-logging
                              call-with-delay-logging))
-  #:use-module ((guix-build-coordinator utils fibers)
-                #:select (call-with-sigint))
   #:use-module (guix-qa-frontpage database)
   #:use-module (guix-qa-frontpage derivation-changes)
   #:use-module (guix-qa-frontpage reproducible-builds)
diff --git a/guix-qa-frontpage/utils.scm b/guix-qa-frontpage/utils.scm
index 60f8bc0..8009b59 100644
--- a/guix-qa-frontpage/utils.scm
+++ b/guix-qa-frontpage/utils.scm
@@ -21,13 +21,13 @@
   #:use-module (srfi srfi-71)
   #:use-module (ice-9 match)
   #:use-module (ice-9 threads)
-  #:use-module (zlib)
   #:use-module (fibers)
   #:use-module (fibers channels)
-  #:use-module (knots timeout)
+  #:use-module (knots)
+  #:use-module (zlib)
   #:export (fiberize
-            call-with-zlib-input-port*
-            non-blocking))
+            non-blocking
+            call-with-zlib-input-port*))
 
 (define* (fiberize proc #:key (parallelism 1))
   (let ((channel (make-channel)))
@@ -64,91 +64,25 @@
           (('result . vals) (apply values vals))
           (('exception . exn) (raise-exception exn)))))))
 
-(define (fibers-map proc . lists)
-  (let ((channels
-         (apply
-          map
-          (lambda args
-            (let ((channel (make-channel)))
-              (spawn-fiber
-               (lambda ()
-                 (put-message
-                  channel
-                  (with-exception-handler
-                      (lambda (exn)
-                        (cons 'exception exn))
-                    (lambda ()
-                      (with-throw-handler #t
-                        (lambda ()
-                          (call-with-values
-                              (lambda ()
-                                (apply proc args))
-                            (lambda val
-                              (cons 'result val))))
-                        (lambda _
-                          (backtrace))))
-                    #:unwind? #t))))
-              channel))
-          lists)))
-    (map
-     (match-lambda
-       (('result . val) val)
-       (('exception . exn) (raise-exception exn)))
-     (map get-message channels))))
-
-(define (fibers-batch-for-each proc batch-size . lists)
-  ;; Like split-at, but don't care about the order of the resulting lists, and
-  ;; don't error if the list is shorter than i elements
-  (define (split-at* lst i)
-    (let lp ((l lst) (n i) (acc '()))
-      (if (or (<= n 0) (null? l))
-          (values (reverse! acc) l)
-          (lp (cdr l) (- n 1) (cons (car l) acc)))))
-
-  ;; As this can be called with lists with tens of thousands of items in them,
-  ;; batch the
-  (define (get-batch lists)
-    (let ((split-lists
-           (map (lambda (lst)
-                  (let ((batch rest (split-at* lst batch-size)))
-                    (cons batch rest)))
-                lists)))
-      (values (map car split-lists)
-              (map cdr split-lists))))
-
-  (let loop ((lists lists))
-    (call-with-values
-        (lambda ()
-          (get-batch lists))
-      (lambda (batch rest)
-        (apply fibers-map proc batch)
-        (unless (null? (car rest))
-          (loop  rest)))))
-  *unspecified*)
-
-(define (fibers-for-each proc . lists)
-  (apply fibers-batch-for-each proc 20 lists))
-
 (define (non-blocking thunk)
   (let ((channel (make-channel)))
-    (call-with-new-thread
+    (call-with-default-io-waiters
      (lambda ()
-       (with-exception-handler
-           (lambda (exn)
-             (put-message channel `(exception ,exn)))
-         (lambda ()
-           (with-throw-handler #t
-             (lambda ()
-               (call-with-values
-                   (lambda ()
-                     ;; This is mostly to set non fibers IO waiters
-                     (with-port-timeouts thunk #:timeout 300))
-                 (lambda values
-                   (put-message channel `(values ,@values)))))
-             (lambda args
-               (display (backtrace) (current-error-port))
-               (newline (current-error-port)))))
-         #:unwind? #t)))
+       (call-with-new-thread
+        (lambda ()
+          (with-exception-handler
+              (lambda (exn)
+                (put-message channel `(exception ,exn)))
+            (lambda ()
+              (with-throw-handler #t
+                (lambda ()
+                  (call-with-values thunk
+                    (lambda values
+                      (put-message channel `(values ,@values)))))
+                (lambda args
+                  (display (backtrace) (current-error-port))
+                  (newline (current-error-port)))))
+            #:unwind? #t)))))
     (match (get-message channel)
       (('values . results)
        (apply values results))