Re-work the fibers scheduling

Use a single thread for receiving and responding to requests, and delegate the
processing of the requests to a separate set of threads.

I'm hoping this will avoid the processing of requests affecting accepting new
connections, or the sending of responses.

1 files changed, 119 insertions, 84 deletions

diff --git a/guix-data-service/web/server.scm b/guix-data-service/web/server.scm
index f1b061d..07f05f8 100644
--- a/guix-data-service/web/server.scm
+++ b/guix-data-service/web/server.scm
@@ -20,12 +20,14 @@
   #:use-module (srfi srfi-1)
   #:use-module (srfi srfi-11)
   #:use-module (ice-9 match)
+  #:use-module (ice-9 threads)
   #:use-module (web http)
   #:use-module (web request)
   #:use-module (web uri)
   #:use-module (system repl error-handling)
   #:use-module (ice-9 atomic)
   #:use-module (fibers)
+  #:use-module (fibers channels)
   #:use-module (fibers scheduler)
   #:use-module (fibers conditions)
   #:use-module ((guix build syscalls)
@@ -98,101 +100,133 @@
 
   (%guix-data-service-metrics-registry registry)
 
-  (let ((finished? (make-condition)))
+  (let ((finished?         (make-condition))
+        (render-metrics    (make-render-metrics registry))
+        (request-scheduler #f))
     (call-with-sigint
      (lambda ()
+       (call-with-new-thread
+        (lambda ()
+          (run-fibers
+           (lambda ()
+             (let* ((current (current-scheduler))
+                    (schedulers
+                     (cons current (scheduler-remote-peers current))))
+
+               (set! request-scheduler current)
+
+               (for-each
+                (lambda (i sched)
+                  (spawn-fiber
+                   (lambda ()
+                     (catch 'system-error
+                       (lambda ()
+                         (set-thread-name
+                          (string-append "rp " (number->string i))))
+                       (const #t)))
+                   sched))
+                (iota (length schedulers))
+                schedulers))
+
+             (wait finished?))
+           #:hz 0
+           #:parallelism 4)))
+
        (run-fibers
         (lambda ()
-          (let* ((current (current-scheduler))
-                 (schedulers
-                  (cons current (scheduler-remote-peers current))))
-            (for-each
-             (lambda (i sched)
-               (spawn-fiber
-                (lambda ()
-                  (catch 'system-error
-                    (lambda ()
-                      (set-thread-name
-                       (string-append "fibers " (number->string i))))
-                    (const #t)))
-                sched))
-             (iota (length schedulers))
-             schedulers))
-
-          (parameterize
-              ((connection-pool
-                (make-resource-pool
-                 (lambda ()
-                   (open-postgresql-connection
-                    "web"
-                    postgresql-statement-timeout))
-                 (floor (/ postgresql-connections 2))
-                 #:idle-seconds 30
-                 #:destructor
-                 (lambda (conn)
-                   (close-postgresql-connection conn "web"))))
-
-               (reserved-connection-pool
-                (make-resource-pool
-                 (lambda ()
-                   (open-postgresql-connection
-                    "web-reserved"
-                    postgresql-statement-timeout))
-                 (floor (/ postgresql-connections 2))
-                 #:idle-seconds 600
-                 #:destructor
-                 (lambda (conn)
-                   (close-postgresql-connection conn "web-reserved"))))
-
-               (resource-pool-default-timeout 5))
-
-            (let ((resource-pool-checkout-failures-metric
-                   (make-counter-metric registry
-                                        "resource_pool_checkout_timeouts_total"
-                                        #:labels '(pool_name))))
-              (%resource-pool-timeout-handler
-               (lambda (pool proc timeout)
-                 (let ((pool-name
-                        (cond
-                         ((eq? pool (connection-pool)) "normal")
-                         ((eq? pool (reserved-connection-pool)) "reserved")
-                         (else #f))))
-                   (when pool-name
-                     (metric-increment
-                      resource-pool-checkout-failures-metric
-                      #:label-values `((pool_name . ,pool-name))))))))
-
-            (spawn-fiber
-             (lambda ()
-               (with-resource-from-pool (connection-pool) conn
-                 (backfill-guix-revision-package-derivation-distribution-counts
-                  conn))))
-
-            (let ((render-metrics
-                   (make-render-metrics registry))
-                  (requests-metric
-                   (make-counter-metric registry "requests_total")))
-
-              (with-exception-handler
-                  (lambda (exn)
-                    (simple-format
-                     (current-error-port)
-                     "\n
+          (catch 'system-error
+            (lambda ()
+              (set-thread-name
+               (string-append "server")))
+            (const #t))
+
+          (while (not request-scheduler)
+            (sleep 0.1))
+
+          (let ((requests-metric
+                 (make-counter-metric registry "requests_total")))
+
+            (with-exception-handler
+                (lambda (exn)
+                  (simple-format
+                   (current-error-port)
+                   "\n
 error: guix-data-service could not start: ~A
 
 Check if it's already running, or whether another process is using that
 port. Also, the port used can be changed by passing the --port option.\n"
-                     exn)
-                    (primitive-exit 1))
-                (lambda ()
+                   exn)
+                  (primitive-exit 1))
+              (lambda ()
+                (parameterize
+                    ((connection-pool
+                      (make-resource-pool
+                       (lambda ()
+                         (open-postgresql-connection
+                          "web"
+                          postgresql-statement-timeout))
+                       (floor (/ postgresql-connections 2))
+                       #:idle-seconds 30
+                       #:destructor
+                       (lambda (conn)
+                         (close-postgresql-connection conn "web"))))
+
+                     (reserved-connection-pool
+                      (make-resource-pool
+                       (lambda ()
+                         (open-postgresql-connection
+                          "web-reserved"
+                          postgresql-statement-timeout))
+                       (floor (/ postgresql-connections 2))
+                       #:idle-seconds 600
+                       #:destructor
+                       (lambda (conn)
+                         (close-postgresql-connection conn "web-reserved"))))
+
+                     (resource-pool-default-timeout 5))
+
+                  (let ((resource-pool-checkout-failures-metric
+                         (make-counter-metric registry
+                                              "resource_pool_checkout_timeouts_total"
+                                              #:labels '(pool_name))))
+                    (%resource-pool-timeout-handler
+                     (lambda (pool proc timeout)
+                       (let ((pool-name
+                              (cond
+                               ((eq? pool (connection-pool)) "normal")
+                               ((eq? pool (reserved-connection-pool)) "reserved")
+                               (else #f))))
+                         (when pool-name
+                           (metric-increment
+                            resource-pool-checkout-failures-metric
+                            #:label-values `((pool_name . ,pool-name))))))))
+
+                  (spawn-fiber
+                   (lambda ()
+                     (with-resource-from-pool (connection-pool) conn
+                       (backfill-guix-revision-package-derivation-distribution-counts
+                        conn)))
+                   request-scheduler)
+
                   (run-server/patched
                    (lambda (request body)
                      (metric-increment requests-metric)
 
-                     (handler request finished? body controller
-                              secret-key-base
-                              startup-completed
-                              render-metrics))
+                     (let ((reply (make-channel)))
+                       (spawn-fiber
+                        (lambda ()
+                          (call-with-values
+                              (lambda ()
+                                (handler request finished? body controller
+                                         secret-key-base
+                                         startup-completed
+                                         render-metrics))
+                            (lambda vals
+                              (put-message reply vals))))
+                        request-scheduler
+                        #:parallel? #t)
+
+                       (apply values (get-message reply))))
                    #:host host
                    #:port port))
                 #:unwind? #t)))
@@ -202,5 +236,6 @@ port. Also, the port used can be changed by passing the --port option.\n"
           (spawn-port-monitoring-fiber port finished?)
 
           (wait finished?))
-        #:parallelism 4))
+        #:hz 5
+        #:parallelism 1))
      finished?)))