[PATCH 1/4] utils: Change critical section terminology to worker threads.

[PATCH 1/4] utils: Change critical section terminology to worker threads.

[Christopher Baines]

As far as I'm aware, it's necessary to use a separate thread for interacting
with SQLite as one of the threads used for fibers will be blocked while the
SQLite query is running.

This doesn't mean all queries have to be executed one at a time though,
providing the queries are executed outside the threads used by fibers, and a
single connection isn't used in multiple threads.

These changes start to move in this direction, first by just changing the
terminology.

* src/cuirass/base.scm (clear-build-queue, cancel-old-builds): Change
with-db-critical-section to with-db-worker-thread.
* src/cuirass/database.scm (with-db-critical-section): Rename syntax rule to
with-db-worker-thread.
(db-add-input, db-add-checkout, db-add-specification, db-remove-specification,
db-get-inputs, db-get-specification, db-add-evaluation,
db-set-evaluations-done, db-set-evaluation-done, db-add-derivation-output,
db-add-build, db-update-build-status!, db-get-output, db-get-outputs,
db-get-builds-by-search, db-get-builds, db-get-build derivation-or-id,
db-add-event, db-get-events, db-delete-events-with-ids-<=-to,
db-get-pending-derivations, db-get-checkouts, db-get-evaluations,
db-get-evaluations-build-summary, db-get-evaluations-id-max,
db-get-evaluation-summary, db-get-builds-query-min, db-get-builds-query-max,
db-get-builds-min, db-get-builds-max, db-get-evaluation-specification): Change
from using with-db-critical-section to
with-db-worker-thread.
(with-database): Change syntax rule to use make-worker-thread-channel,
renaming from make-critical-section.
* src/cuirass/utils.scm (%critical-section-args): Rename parameter to
%worker-thread-args.
(make-critical-section): Rename to make-worker-thread-channel, and adjust
parameter and docstring.
(call-with-critical-section): Rename to call-with-worker-thread and adjust
parameter.
(with-critical-section): Rename to with-worker-thread, and adjust to call
call-with-worker-thread.
* tests/database.scm (db-init): Use make-worker-thread-channel rather than
make-critical-section.
* tests/http.scm (db-init): Use make-worker-thread-channel rather than
make-critical-section.
---
 src/cuirass/base.scm     |  4 +--
 src/cuirass/database.scm | 74 ++++++++++++++++++++--------------------
 src/cuirass/utils.scm    | 38 +++++++++------------
 tests/database.scm       |  2 +-
 tests/http.scm           |  2 +-
 5 files changed, 58 insertions(+), 62 deletions(-)

diff --git a/src/cuirass/base.scm b/src/cuirass/base.scm
index 143bc2e..2b18dc6 100644
--- a/src/cuirass/base.scm
+++ b/src/cuirass/base.scm
@@ -607,13 +607,13 @@ updating the database accordingly."
   "Reset the status of builds in the database that are marked as \"started\".
 This procedure is meant to be called at startup."
   (log-message "marking stale builds as \"scheduled\"...")
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "UPDATE Builds SET status = -2 WHERE status = -1;")))
 
 (define (cancel-old-builds age)
   "Cancel builds older than AGE seconds."
   (log-message "canceling builds older than ~a seconds..." age)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec
      db "UPDATE Builds SET status = 4 WHERE status = -2 AND timestamp < "
      (- (time-second (current-time time-utc)) age) ";")))
diff --git a/src/cuirass/database.scm b/src/cuirass/database.scm
index 308b5c3..3e93492 100644
--- a/src/cuirass/database.scm
+++ b/src/cuirass/database.scm
@@ -73,7 +73,7 @@
             %db-channel
             %record-events?
             ;; Macros.
-            with-db-critical-section
+            with-db-worker-thread
             with-database))
 
 (define (%sqlite-exec db sql . args)
@@ -172,12 +172,12 @@ specified."
 (define %record-events?
   (make-parameter #f))
 
-(define-syntax-rule (with-db-critical-section db exp ...)
+(define-syntax-rule (with-db-worker-thread db exp ...)
   "Evaluate EXP... in the critical section corresponding to %DB-CHANNEL.
 DB is bound to the argument of that critical section: the database
 connection."
-  (call-with-critical-section (%db-channel)
-                              (lambda (db) exp ...)))
+  (call-with-worker-thread (%db-channel)
+                           (lambda (db) exp ...)))
 
 (define (read-sql-file file-name)
   "Return a list of string containing SQL instructions from FILE-NAME."
@@ -292,7 +292,7 @@ of the list, and returns #f when there is no result."
     (() #f)))
 
 (define (db-add-input spec-name input)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "\
 INSERT OR IGNORE INTO Inputs (specification, name, url, load_path, branch, \
 tag, revision, no_compile_p) VALUES ("
@@ -309,7 +309,7 @@ tag, revision, no_compile_p) VALUES ("
 (define (db-add-checkout spec-name eval-id checkout)
   "Insert CHECKOUT associated with SPEC-NAME and EVAL-ID.  If a checkout with
 the same revision already exists for SPEC-NAME, return #f."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (catch-sqlite-error
      (sqlite-exec db "\
 INSERT INTO Checkouts (specification, revision, evaluation, input,
@@ -329,7 +329,7 @@ directory) VALUES ("
 (define (db-add-specification spec)
   "Store SPEC in database the database.  SPEC inputs are stored in the INPUTS
 table."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "\
 INSERT OR IGNORE INTO Specifications (name, load_path_inputs, \
 package_path_inputs, proc_input, proc_file, proc, proc_args) \
@@ -349,7 +349,7 @@ package_path_inputs, proc_input, proc_file, proc, proc_args) \
 
 (define (db-remove-specification name)
   "Remove the specification matching NAME from the database and its inputs."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "BEGIN TRANSACTION;")
     (sqlite-exec db "\
 DELETE FROM Inputs WHERE specification=" name ";")
@@ -358,7 +358,7 @@ DELETE FROM Specifications WHERE name=" name ";")
     (sqlite-exec db "COMMIT;")))
 
 (define (db-get-inputs spec-name)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let loop ((rows (sqlite-exec
                       db "SELECT * FROM Inputs WHERE specification="
                       spec-name ";"))
@@ -378,7 +378,7 @@ DELETE FROM Specifications WHERE name=" name ";")
                      inputs)))))))
 
 (define (db-get-specifications)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let loop ((rows  (sqlite-exec db "SELECT * FROM Specifications ORDER BY name DESC;"))
                (specs '()))
       (match rows
@@ -402,7 +402,7 @@ DELETE FROM Specifications WHERE name=" name ";")
 (define (db-add-evaluation spec-name checkouts)
   "Add a new evaluation for SPEC-NAME only if one of the CHECKOUTS is new.
 Otherwise, return #f."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "BEGIN TRANSACTION;")
     (sqlite-exec db "INSERT INTO Evaluations (specification, in_progress)
 VALUES (" spec-name ", true);")
@@ -422,11 +422,11 @@ VALUES (" spec-name ", true);")
                  eval-id)))))
 
 (define (db-set-evaluations-done)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "UPDATE Evaluations SET in_progress = false;")))
 
 (define (db-set-evaluation-done eval-id)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec db "UPDATE Evaluations SET in_progress = false
 WHERE id = " eval-id ";")
     (db-add-event 'evaluation
@@ -450,7 +450,7 @@ a critical section that allows database operations to be serialized."
      ;; access blocks on PUT-MESSAGE, which allows the scheduler to schedule
      ;; another fiber.  Also, creating one new handle for each request would
      ;; be costly and may defeat statement caching.
-     (parameterize ((%db-channel (make-critical-section db)))
+     (parameterize ((%db-channel (make-worker-thread-channel db)))
        body ...)
      (db-close db))))
 
@@ -485,7 +485,7 @@ string."
 (define (db-add-output derivation output)
   "Insert OUTPUT associated with DERIVATION.  If an output with the same path
 already exists, return #f."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (catch-sqlite-error
      (match output
        ((name . path)
@@ -502,7 +502,7 @@ INSERT INTO Outputs (derivation, name, path) VALUES ("
 (define (db-add-build build)
   "Store BUILD in database the database only if one of its outputs is new.
 Return #f otherwise.  BUILD outputs are stored in the OUTPUTS table."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (catch-sqlite-error
      (sqlite-exec db "BEGIN TRANSACTION;")
      (sqlite-exec db "
@@ -559,7 +559,7 @@ log file for DRV."
       (,(build-status failed-other)      . "failed (other)")
       (,(build-status canceled)          . "canceled")))
 
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (if (= status (build-status started))
         (begin
           (sqlite-exec db "UPDATE Builds SET starttime=" now ", status="
@@ -591,7 +591,7 @@ log file for DRV."
 
 (define (db-get-output path)
   "Retrieve the OUTPUT for PATH."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     ;; There isn't a unique index on path, but because Cuirass avoids adding
     ;; derivations which introduce the same outputs, there should only be one
     ;; result.
@@ -606,7 +606,7 @@ LIMIT 1;")
 (define (db-get-outputs derivation)
   "Retrieve the OUTPUTS of the build identified by DERIVATION in the
 database."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let loop ((rows
                 (sqlite-exec db "SELECT name, path FROM Outputs
 WHERE derivation =" derivation ";"))
@@ -669,7 +669,7 @@ WHERE derivation =" derivation ";"))
   "Retrieve all builds in the database which are matched by given FILTERS.
 FILTERS is an assoc list whose possible keys are the symbols query,
 border-low-id, border-high-id, and nr."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let* ((stmt-text (format #f "SELECT * FROM (
 SELECT Builds.rowid, Builds.timestamp, Builds.starttime,
 Builds.stoptime, Builds.log, Builds.status, Builds.job_name, Builds.system,
@@ -726,7 +726,7 @@ ORDER BY rowid DESC;"))
   "Retrieve all builds in the database which are matched by given FILTERS.
 FILTERS is an assoc list whose possible keys are 'derivation | 'id | 'jobset |
 'job | 'system | 'nr | 'order | 'status | 'evaluation."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let* ((order (filters->order filters))
            (stmt-text (format #f "SELECT * FROM (
 SELECT Builds.derivation, Builds.rowid, Builds.timestamp, Builds.starttime,
@@ -801,13 +801,13 @@ ORDER BY ~a, rowid ASC;" order))
 
 (define (db-get-build derivation-or-id)
   "Retrieve a build in the database which corresponds to DERIVATION-OR-ID."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((key (if (number? derivation-or-id) 'id 'derivation)))
       (expect-one-row (db-get-builds `((,key . ,derivation-or-id)))))))
 
 (define (db-add-event type timestamp details)
   (when (%record-events?)
-    (with-db-critical-section db
+    (with-db-worker-thread db
       (sqlite-exec db "\
 INSERT INTO Events (type, timestamp, event_json) VALUES ("
                    (symbol->string type) ", "
@@ -817,7 +817,7 @@ INSERT INTO Events (type, timestamp, event_json) VALUES ("
       #t)))
 
 (define (db-get-events filters)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let* ((stmt-text "\
 SELECT Events.id,
        Events.type,
@@ -857,7 +857,7 @@ LIMIT :nr;")
                        events))))))))
 
 (define (db-delete-events-with-ids-<=-to id)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (sqlite-exec
      db
      "DELETE FROM Events WHERE id <= " id ";")))
@@ -865,13 +865,13 @@ LIMIT :nr;")
 (define (db-get-pending-derivations)
   "Return the list of derivation file names corresponding to pending builds in
 the database.  The returned list is guaranteed to not have any duplicates."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (map (match-lambda (#(drv) drv))
          (sqlite-exec db "
 SELECT derivation FROM Builds WHERE Builds.status < 0;"))))
 
 (define (db-get-checkouts eval-id)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let loop ((rows (sqlite-exec
                       db "SELECT revision, input, directory FROM Checkouts
 WHERE evaluation =" eval-id ";"))
@@ -887,7 +887,7 @@ WHERE evaluation =" eval-id ";"))
                      checkouts)))))))
 
 (define (db-get-evaluations limit)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let loop ((rows  (sqlite-exec db "SELECT id, specification, in_progress
 FROM Evaluations ORDER BY id DESC LIMIT " limit ";"))
                (evaluations '()))
@@ -903,7 +903,7 @@ FROM Evaluations ORDER BY id DESC LIMIT " limit ";"))
                      evaluations)))))))
 
 (define (db-get-evaluations-build-summary spec limit border-low border-high)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let loop ((rows (sqlite-exec db "
 SELECT E.id, E.in_progress, B.succeeded, B.failed, B.scheduled
 FROM
@@ -936,7 +936,7 @@ ORDER BY E.id ASC;"))
 
 (define (db-get-evaluations-id-min spec)
   "Return the min id of evaluations for the given specification SPEC."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((rows (sqlite-exec db "
 SELECT MIN(id) FROM Evaluations
 WHERE specification=" spec)))
@@ -944,14 +944,14 @@ WHERE specification=" spec)))
 
 (define (db-get-evaluations-id-max spec)
   "Return the max id of evaluations for the given specification SPEC."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((rows (sqlite-exec db "
 SELECT MAX(id) FROM Evaluations
 WHERE specification=" spec)))
       (and=> (expect-one-row rows) (cut vector-ref <> 0)))))
 
 (define (db-get-evaluation-summary id)
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((rows (sqlite-exec db "
 SELECT E.id, E.in_progress, B.total, B.succeeded, B.failed, B.scheduled
 FROM
@@ -977,7 +977,7 @@ ORDER BY E.id ASC;")))
 
 (define (db-get-builds-query-min query)
   "Return the smallest build row identifier matching QUERY."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let* ((stmt-text "SELECT MIN(Builds.rowid) FROM Builds
 INNER JOIN Evaluations ON Builds.evaluation = Evaluations.id
 INNER JOIN Specifications ON Evaluations.specification = Specifications.name
@@ -996,7 +996,7 @@ AND (:system IS NULL
 
 (define (db-get-builds-query-max query)
   "Return the largest build row identifier matching QUERY."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let* ((stmt-text "SELECT MAX(Builds.rowid) FROM Builds
 INNER JOIN Evaluations ON Builds.evaluation = Evaluations.id
 INNER JOIN Specifications ON Evaluations.specification = Specifications.name
@@ -1016,7 +1016,7 @@ AND (:system IS NULL
 (define (db-get-builds-min eval status)
   "Return the min build (stoptime, rowid) pair for the given evaluation EVAL
 and STATUS."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((rows (sqlite-exec db "
 SELECT stoptime, MIN(rowid) FROM
 (SELECT rowid, stoptime FROM Builds
@@ -1035,7 +1035,7 @@ AND (" status " IS NULL OR (" status " = 'pending'
 (define (db-get-builds-max eval status)
   "Return the max build (stoptime, rowid) pair for the given evaluation EVAL
 and STATUS."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((rows (sqlite-exec db "
 SELECT stoptime, MAX(rowid) FROM
 (SELECT rowid, stoptime FROM Builds
@@ -1053,7 +1053,7 @@ AND (" status " IS NULL OR (" status " = 'pending'
 
 (define (db-get-evaluation-specification eval)
   "Return specification of evaluation with id EVAL."
-  (with-db-critical-section db
+  (with-db-worker-thread db
     (let ((rows (sqlite-exec db "
 SELECT specification FROM Evaluations
 WHERE id = " eval)))
diff --git a/src/cuirass/utils.scm b/src/cuirass/utils.scm
index fe74b69..514899e 100644
--- a/src/cuirass/utils.scm
+++ b/src/cuirass/utils.scm
@@ -35,9 +35,9 @@
             define-enumeration
             unwind-protect
 
-            make-critical-section
-            call-with-critical-section
-            with-critical-section
+            make-worker-thread-channel
+            call-with-worker-thread
+            with-worker-thread
 
             %non-blocking
             non-blocking
@@ -96,21 +96,17 @@ delimited continuations and fibers."
         (conclusion)
         (apply throw args)))))
 
-(define %critical-section-args
+(define %worker-thread-args
   (make-parameter #f))
 
-(define (make-critical-section . args)
-  "Return a channel used to implement a critical section.  That channel can
-then be passed to 'join-critical-section', which will ensure sequential
-ordering.  ARGS are the arguments of the critical section.
-
-Critical sections are implemented by passing the procedure to execute to a
-dedicated thread."
+(define (make-worker-thread-channel . args)
+  "Return a channel used to offload work to a dedicated thread.  ARGS are the
+arguments of the worker thread procedure."
   (parameterize (((@@ (fibers internal) current-fiber) #f))
     (let ((channel (make-channel)))
       (call-with-new-thread
        (lambda ()
-         (parameterize ((%critical-section-args args))
+         (parameterize ((%worker-thread-args args))
            (let loop ()
              (match (get-message channel)
                (((? channel? reply) . (? procedure? proc))
@@ -118,21 +114,21 @@ dedicated thread."
              (loop)))))
       channel)))
 
-(define (call-with-critical-section channel proc)
-  "Send PROC to the critical section through CHANNEL.  Return the result of
-PROC.  If already in the critical section, call PROC immediately."
-  (let ((args (%critical-section-args)))
+(define (call-with-worker-thread channel proc)
+  "Send PROC to the worker thread through CHANNEL.  Return the result of PROC.
+If already in the worker thread, call PROC immediately."
+  (let ((args (%worker-thread-args)))
     (if args
         (apply proc args)
         (let ((reply (make-channel)))
           (put-message channel (cons reply proc))
           (get-message reply)))))
 
-(define-syntax-rule (with-critical-section channel (vars ...) exp ...)
-  "Evaluate EXP... in the critical section corresponding to CHANNEL.
-VARS... are bound to the arguments of the critical section."
-  (call-with-critical-section channel
-                              (lambda (vars ...) exp ...)))
+(define-syntax-rule (with-worker-thread channel (vars ...) exp ...)
+  "Evaluate EXP... in the worker thread corresponding to CHANNEL.
+VARS... are bound to the arguments of the worker thread."
+  (call-with-worker-thread channel
+                           (lambda (vars ...) exp ...)))
 
 (define (%non-blocking thunk)
   (parameterize (((@@ (fibers internal) current-fiber) #f))
diff --git a/tests/database.scm b/tests/database.scm
index d9dfe13..271f166 100644
--- a/tests/database.scm
+++ b/tests/database.scm
@@ -87,7 +87,7 @@
   (test-assert "db-init"
     (begin
       (%db (db-init database-name))
-      (%db-channel (make-critical-section (%db)))
+      (%db-channel (make-worker-thread-channel (%db)))
       #t))
 
   (test-assert "sqlite-exec"
diff --git a/tests/http.scm b/tests/http.scm
index b21fa17..337a775 100644
--- a/tests/http.scm
+++ b/tests/http.scm
@@ -108,7 +108,7 @@
   (test-assert "db-init"
     (begin
       (%db (db-init database-name))
-      (%db-channel (make-critical-section (%db)))
+      (%db-channel (make-worker-thread-channel (%db)))
       #t))
 
   (test-assert "cuirass-run"
-- 
2.24.1

[PATCH 2/4] Adjust make-worker-thread-channel to take an initializer.

[Christopher Baines]

While this is a generic method, and initializer function will give the
flexibility required to create multiple worker threads for performing SQLite
queries, each with it's own database connection (as a result of calling the
initializer once for each thread). Without this change, they'd all have to use
the same connection, which would not work.

* src/cuirass/utils.scm (make-worker-thread-channel): Change procedure to take
an initializer, rather than arguments directly.
* src/cuirass/database.scm (with-database): Adjust to call
make-worker-thread-channel with an initializer.
* tests/database.scm (db-init): Change to use make-worker-thread-channel
initializer.
* tests/http.scm (db-init): Change to use make-worker-thread-channel
initializer.
---
 src/cuirass/database.scm | 25 +++++++------------------
 src/cuirass/utils.scm    | 19 ++++++++++---------
 tests/database.scm       |  4 +++-
 tests/http.scm           |  4 +++-
 4 files changed, 23 insertions(+), 29 deletions(-)

diff --git a/src/cuirass/database.scm b/src/cuirass/database.scm
index 3e93492..0f5e38f 100644
--- a/src/cuirass/database.scm
+++ b/src/cuirass/database.scm
@@ -435,24 +435,13 @@ WHERE id = " eval-id ";")
                     (#:in_progress . #f)))))
 
 (define-syntax-rule (with-database body ...)
-  "Run BODY with %DB-CHANNEL being dynamically bound to a channel implementing
-a critical section that allows database operations to be serialized."
-  ;; XXX: We don't install an unwind handler to play well with delimited
-  ;; continuations and fibers.  But as a consequence, we leak DB when BODY
-  ;; raises an exception.
-  (let ((db (db-open)))
-    (unwind-protect
-     ;; Process database queries sequentially in a thread.  We need this
-     ;; because otherwise we would need to use the SQLite multithreading
-     ;; feature for which it is required to wait until the database is
-     ;; available, and the waiting would happen in non-cooperative and
-     ;; non-resumable code that blocks the fibers scheduler.  Now the database
-     ;; access blocks on PUT-MESSAGE, which allows the scheduler to schedule
-     ;; another fiber.  Also, creating one new handle for each request would
-     ;; be costly and may defeat statement caching.
-     (parameterize ((%db-channel (make-worker-thread-channel db)))
-       body ...)
-     (db-close db))))
+  "Run BODY with %DB-CHANNEL being dynamically bound to a channel providing a
+worker thread that allows database operations to run without intefering with
+fibers."
+  (parameterize ((%db-channel (make-worker-thread-channel
+                               (lambda ()
+                                 (list (db-open))))))
+    body ...))
 
 (define* (read-quoted-string #:optional (port (current-input-port)))
   "Read all of the characters out of PORT and return them as a SQL quoted
diff --git a/src/cuirass/utils.scm b/src/cuirass/utils.scm
index 514899e..dfed4a9 100644
--- a/src/cuirass/utils.scm
+++ b/src/cuirass/utils.scm
@@ -99,19 +99,20 @@ delimited continuations and fibers."
 (define %worker-thread-args
   (make-parameter #f))
 
-(define (make-worker-thread-channel . args)
+(define (make-worker-thread-channel initializer)
   "Return a channel used to offload work to a dedicated thread.  ARGS are the
 arguments of the worker thread procedure."
   (parameterize (((@@ (fibers internal) current-fiber) #f))
     (let ((channel (make-channel)))
-      (call-with-new-thread
-       (lambda ()
-         (parameterize ((%worker-thread-args args))
-           (let loop ()
-             (match (get-message channel)
-               (((? channel? reply) . (? procedure? proc))
-                (put-message reply (apply proc args))))
-             (loop)))))
+      (let ((args (initializer)))
+        (call-with-new-thread
+         (lambda ()
+           (parameterize ((%worker-thread-args args))
+             (let loop ()
+               (match (get-message channel)
+                 (((? channel? reply) . (? procedure? proc))
+                  (put-message reply (apply proc args))))
+               (loop))))))
       channel)))
 
 (define (call-with-worker-thread channel proc)
diff --git a/tests/database.scm b/tests/database.scm
index 271f166..6098465 100644
--- a/tests/database.scm
+++ b/tests/database.scm
@@ -87,7 +87,9 @@
   (test-assert "db-init"
     (begin
       (%db (db-init database-name))
-      (%db-channel (make-worker-thread-channel (%db)))
+      (%db-channel (make-worker-thread-channel
+                    (lambda ()
+                      (list (%db)))))
       #t))
 
   (test-assert "sqlite-exec"
diff --git a/tests/http.scm b/tests/http.scm
index 337a775..d20a3c3 100644
--- a/tests/http.scm
+++ b/tests/http.scm
@@ -108,7 +108,9 @@
   (test-assert "db-init"
     (begin
       (%db (db-init database-name))
-      (%db-channel (make-worker-thread-channel (%db)))
+      (%db-channel (make-worker-thread-channel
+                    (lambda ()
+                      (list (%db)))))
       #t))
 
   (test-assert "cuirass-run"
-- 
2.24.1

[PATCH 3/4] Enable make-worker-thread-channel to create multiple worker threads.

[Christopher Baines]

This will allow running multiple threads, that all listen on the same channel,
enabling processing multiple jobs at one time.

* src/cuirass/utils.scm (make-worker-thread-channel): Add a #:parallelism
argument, and create as many threads as the given parallelism.
---
 src/cuirass/utils.scm | 24 ++++++++++++++----------
 1 file changed, 14 insertions(+), 10 deletions(-)

diff --git a/src/cuirass/utils.scm b/src/cuirass/utils.scm
index dfed4a9..f3ba18d 100644
--- a/src/cuirass/utils.scm
+++ b/src/cuirass/utils.scm
@@ -99,20 +99,24 @@ delimited continuations and fibers."
 (define %worker-thread-args
   (make-parameter #f))
 
-(define (make-worker-thread-channel initializer)
+(define* (make-worker-thread-channel initializer
+                                     #:key (parallelism 1))
   "Return a channel used to offload work to a dedicated thread.  ARGS are the
 arguments of the worker thread procedure."
   (parameterize (((@@ (fibers internal) current-fiber) #f))
     (let ((channel (make-channel)))
-      (let ((args (initializer)))
-        (call-with-new-thread
-         (lambda ()
-           (parameterize ((%worker-thread-args args))
-             (let loop ()
-               (match (get-message channel)
-                 (((? channel? reply) . (? procedure? proc))
-                  (put-message reply (apply proc args))))
-               (loop))))))
+      (for-each
+       (lambda _
+         (let ((args (initializer)))
+           (call-with-new-thread
+            (lambda ()
+              (parameterize ((%worker-thread-args args))
+                (let loop ()
+                  (match (get-message channel)
+                    (((? channel? reply) . (? procedure? proc))
+                     (put-message reply (apply proc args))))
+                  (loop)))))))
+       (iota parallelism))
       channel)))
 
 (define (call-with-worker-thread channel proc)
-- 
2.24.1

[PATCH 4/4] database: Enable running up to 4 database queries at once.

[Christopher Baines]

The number of threads is copied from bin/cuirass.in. When you have at least
two processors, this will allow database queries to be executed in parallel.

With some crude testing using the Apache HTTP server benchmarking tool (ab
from the httpd package), the max request latency does seem to drop when
multiple threads are used, especially when the database queries are slow (I
tested by adding usleep to the worker thread code).

* src/cuirass/database.scm (with-database): Pass #:parallelism to
make-worker-thread-channel.
---
 src/cuirass/database.scm | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/src/cuirass/database.scm b/src/cuirass/database.scm
index 0f5e38f..1585df4 100644
--- a/src/cuirass/database.scm
+++ b/src/cuirass/database.scm
@@ -440,7 +440,9 @@ worker thread that allows database operations to run without intefering with
 fibers."
   (parameterize ((%db-channel (make-worker-thread-channel
                                (lambda ()
-                                 (list (db-open))))))
+                                 (list (db-open)))
+                               #:parallelism
+                               (min (current-processor-count) 4))))
     body ...))
 
 (define* (read-quoted-string #:optional (port (current-input-port)))
-- 
2.24.1

Re: [PATCH 1/4] utils: Change critical section terminology to worker threads.

[Ludovic Courtès]

Hi,

Christopher Baines <mail@cbaines.net> skribis:

> As far as I'm aware, it's necessary to use a separate thread for interacting
> with SQLite as one of the threads used for fibers will be blocked while the
> SQLite query is running.

Indeed.

> This doesn't mean all queries have to be executed one at a time though,
> providing the queries are executed outside the threads used by fibers, and a
> single connection isn't used in multiple threads.

Serialization of database accesses comes from:

  commit 4612a3a70f1e70afa4e0ce00e8cb1a7848dddf58
  Author: Clément Lassieur <clement@lassieur.org>
  Date:   Sun Aug 5 13:14:44 2018 +0200

      database: Serialize all database accesses in a thread.

      Fixes <https://bugs.gnu.org/32234>.

Apparently at the time we weren’t quite sure how SQLite would perform if
we accessed it from several threads, but you seem to suggest that it
works fine, right?

Did you try it on a big database like that of berlin?

> These changes start to move in this direction, first by just changing the
> terminology.

That’s also welcome!

Ludo’.

Re: [PATCH 4/4] database: Enable running up to 4 database queries at once.

[Ludovic Courtès]

The rest of this patch series LGTM!

Ludo’.

Re: [PATCH 1/4] utils: Change critical section terminology to worker threads.

[Christopher Baines]

[-- Attachment #1: Type: text/plain, Size: 1443 bytes --]


Ludovic Courtès <ludo@gnu.org> writes:

>> This doesn't mean all queries have to be executed one at a time though,
>> providing the queries are executed outside the threads used by fibers, and a
>> single connection isn't used in multiple threads.
>
> Serialization of database accesses comes from:
>
>   commit 4612a3a70f1e70afa4e0ce00e8cb1a7848dddf58
>   Author: Clément Lassieur <clement@lassieur.org>
>   Date:   Sun Aug 5 13:14:44 2018 +0200
>
>       database: Serialize all database accesses in a thread.
>
>       Fixes <https://bugs.gnu.org/32234>.
>
> Apparently at the time we weren’t quite sure how SQLite would perform if
> we accessed it from several threads, but you seem to suggest that it
> works fine, right?

Yeah, the documentation suggests it can work [1]. We're even opening the
database in multi-threaded mode already (SQLITE_OPEN_NOMUTEX).

1: https://www.sqlite.org/threadsafe.html

> Did you try it on a big database like that of berlin?

I have an old copy of the berlin database, and it seems to work fine
with that. At least the web interface that is, but if that works,
everything else should work too.

>> These changes start to move in this direction, first by just changing the
>> terminology.
>
> That’s also welcome!

I've gone ahead an pushed these patches now, as it's something I'd like
to deploy prior to the Guix days.

Thanks for taking a look,

Chris

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 962 bytes --]

Re: [PATCH 1/4] utils: Change critical section terminology to worker threads.

[Ludovic Courtès]

Hi,

Christopher Baines <mail@cbaines.net> skribis:

> I've gone ahead an pushed these patches now, as it's something I'd like
> to deploy prior to the Guix days.

Woohoo!  Should we deploy it on berlin?  Let’s synchronize on IRC maybe?

Thanks,
Ludo’.

Re: [PATCH 1/4] utils: Change critical section terminology to worker threads.

[Christopher Baines]

[-- Attachment #1: Type: text/plain, Size: 787 bytes --]


Ludovic Courtès <ludo@gnu.org> writes:

> Christopher Baines <mail@cbaines.net> skribis:
>
>> I've gone ahead an pushed these patches now, as it's something I'd like
>> to deploy prior to the Guix days.
>
> Woohoo!  Should we deploy it on berlin?  Let’s synchronize on IRC maybe?

Yep, it would be good. I think it works fine on Bayfront (ignoring the
issues there with running out of disk space, and the build users thing).

It might already be deployed though, as I've just noticed the /output/
requests work on Berlin ([1] for example). That's really good, as it
means I'll be able to pull builds in to the Guix Data Service much more
easily.

1: https://ci.guix.gnu.org/output/4whab5ys3k3r4yn7knc3pfjpwcxqiqxi-maven-resolver-provider-3.6.1

Thanks,

Chris

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 962 bytes --]

Re: [PATCH 1/4] utils: Change critical section terminology to worker threads.

[Ludovic Courtès]

Hi!

Christopher Baines <mail@cbaines.net> skribis:

> Ludovic Courtès <ludo@gnu.org> writes:
>
>> Christopher Baines <mail@cbaines.net> skribis:
>>
>>> I've gone ahead an pushed these patches now, as it's something I'd like
>>> to deploy prior to the Guix days.
>>
>> Woohoo!  Should we deploy it on berlin?  Let’s synchronize on IRC maybe?
>
> Yep, it would be good. I think it works fine on Bayfront (ignoring the
> issues there with running out of disk space, and the build users thing).
>
> It might already be deployed though, as I've just noticed the /output/
> requests work on Berlin ([1] for example). That's really good, as it
> means I'll be able to pull builds in to the Guix Data Service much more
> easily.
>
> 1: https://ci.guix.gnu.org/output/4whab5ys3k3r4yn7knc3pfjpwcxqiqxi-maven-resolver-provider-3.6.1

Yeah, it was reconfigured on Jan. 26 from a Cuirass commit that was
current at that point; specifically, the config has this uncommitted
change:

+(define cuirass-devel
+  (package
+   (inherit cuirass)
+   (version (string-append (package-version cuirass) "-dev"))
+   (source (git-checkout
+           (url "https://git.savannah.gnu.org/git/guix/guix-cuirass.git")))))

Ludo’.