path: root/guix-data-service/utils.scm

;;; Guix Data Service -- Information about Guix over time
;;; Copyright © 2020 Christopher Baines <mail@cbaines.net>
;;;
;;; This program is free software: you can redistribute it and/or
;;; modify it under the terms of the GNU Affero General Public License
;;; as published by the Free Software Foundation, either version 3 of
;;; the License, or (at your option) any later version.
;;;
;;; This program is distributed in the hope that it will be useful,
;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;;; Affero General Public License for more details.
;;;
;;; You should have received a copy of the GNU Affero General Public
;;; License along with this program.  If not, see
;;; <http://www.gnu.org/licenses/>.

(define-module (guix-data-service utils)
  #:use-module (srfi srfi-1)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-11)
  #:use-module (srfi srfi-71)
  #:use-module (ice-9 q)
  #:use-module (ice-9 ftw)
  #:use-module (ice-9 match)
  #:use-module (ice-9 atomic)
  #:use-module (ice-9 format)
  #:use-module (ice-9 threads)
  #:use-module (ice-9 exceptions)
  #:use-module (ice-9 ports internal)
  #:use-module (ice-9 suspendable-ports)
  #:use-module (lzlib)
  #:use-module ((guix build syscalls)
                #:select (set-thread-name))
  #:use-module (fibers)
  #:use-module (fibers channels)
  #:use-module (fibers operations)
  #:use-module (fibers timers)
  #:use-module (fibers conditions)
  #:use-module (fibers scheduler)
  #:use-module (prometheus)
  #:export (call-with-time-logging
            with-time-logging
            prevent-inlining-for-tests

            resource-pool-default-timeout
            resource-pool-retry-checkout-timeout
            %resource-pool-timeout-handler
            resource-pool-timeout-error?
            make-resource-pool
            destroy-resource-pool
            call-with-resource-from-pool
            with-resource-from-pool
            resource-pool-stats

            call-with-default-io-waiters
            make-worker-thread-channel
            %worker-thread-default-timeout
            call-with-worker-thread
            worker-thread-timeout-error?

            fiberize

            fibers-delay
            fibers-force
            fibers-promise-reset

            fibers-batch-for-each
            fibers-for-each
            fibers-batch-map
            fibers-map

            parallel-via-fibers
            par-map&
            letpar&
            fibers-map-with-progress

            chunk
            chunk!
            chunk-for-each!

            delete-duplicates/sort!

            get-guix-metrics-updater

            call-with-sigint
            run-server/patched

            spawn-port-monitoring-fiber

            make-queueing-channel))

(define (call-with-time-logging action thunk)
  (simple-format #t "debug: Starting ~A\n" action)
  (let ((start-time (current-time)))
    (let-values
        ((result (thunk)))
      (let ((time-taken (- (current-time) start-time)))
        (simple-format #t "debug: Finished ~A, took ~A seconds\n"
                       action time-taken))
      (apply values result))))

(define-syntax-rule (with-time-logging action exp ...)
  "Log under NAME the time taken to evaluate EXP."
  (call-with-time-logging action (lambda () exp ...)))

(define-syntax-rule (prevent-inlining-for-tests var)
  (set! var var))

(define-record-type <resource-pool>
  (make-resource-pool-record name channel)
  resource-pool?
  (name    resource-pool-name)
  (channel resource-pool-channel))

(define* (make-resource-pool initializer max-size
                             #:key (min-size max-size)
                             (idle-seconds #f)
                             (delay-logger (const #f))
                             (duration-logger (const #f))
                             destructor
                             lifetime
                             scheduler
                             (name "unnamed")
                             assume-reliable-waiters?)
  (define (initializer/safe)
    (with-exception-handler
        (lambda (exn)
          (simple-format
           (current-error-port)
           "exception running ~A resource pool initializer: ~A:\n  ~A\n"
           name
           initializer
           exn)
          #f)
      (lambda ()
        (with-throw-handler #t
          initializer
          (lambda args
            (backtrace))))
      #:unwind? #t))

  (define (destructor/safe args)
    (let ((success?
           (with-exception-handler
               (lambda (exn)
                 (simple-format
                  (current-error-port)
                  "exception running resource pool destructor (~A): ~A:\n  ~A\n"
                  name
                  destructor
                  exn)
                 #f)
             (lambda ()
               (with-throw-handler #t
                 (lambda ()
                   (destructor args)
                   #t)
                 (lambda _
                   (backtrace))))
             #:unwind? #t)))

      (or success?
          #t
          (begin
            (sleep 5)
            (destructor/safe args)))))

  (let ((channel (make-channel))
        (checkout-failure-count 0))
    (spawn-fiber
     (lambda ()
       (when idle-seconds
         (spawn-fiber
          (lambda ()
            (while #t
              (sleep idle-seconds)
              (put-message channel '(check-for-idle-resources))))))

       (while #t
         (with-exception-handler
             (lambda (exn)
               (simple-format
                (current-error-port)
                "exception in the ~A pool fiber: ~A\n"
                name
                exn))
           (lambda ()
             (let loop ((resources '())
                        (available '())
                        (waiters '())
                        (resources-last-used '()))

               (match (get-message channel)
                 (('checkout reply)
                  (if (null? available)
                      (if (= (length resources) max-size)
                          (loop resources
                                available
                                (cons reply waiters)
                                resources-last-used)
                          (let ((new-resource (initializer/safe)))
                            (if new-resource
                                (let ((checkout-success?
                                       (perform-operation
                                        (choice-operation
                                         (wrap-operation
                                          (put-operation reply new-resource)
                                          (const #t))
                                         (wrap-operation (sleep-operation 1)
                                                         (const #f))))))
                                  (unless checkout-success?
                                    (set! checkout-failure-count
                                          (+ 1 checkout-failure-count)))

                                  (loop (cons new-resource resources)
                                        (if checkout-success?
                                            available
                                            (cons new-resource available))
                                        waiters
                                        (cons (get-internal-real-time)
                                              resources-last-used)))
                                (loop resources
                                      available
                                      (cons reply waiters)
                                      resources-last-used))))
                      (let ((checkout-success?
                             (perform-operation
                              (choice-operation
                               (wrap-operation
                                (put-operation reply (car available))
                                (const #t))
                               (wrap-operation (sleep-operation 1)
                                               (const #f))))))
                        (unless checkout-success?
                          (set! checkout-failure-count
                                (+ 1 checkout-failure-count)))

                        (if checkout-success?
                            (loop resources
                                  (cdr available)
                                  waiters
                                  resources-last-used)
                            (loop resources
                                  available
                                  waiters
                                  resources-last-used)))))
                 (('return resource)
                  (if (and assume-reliable-waiters?
                           (not (null? waiters)))
                      (let ((checkout-success?
                             (perform-operation
                              (choice-operation
                               (wrap-operation
                                (put-operation (last waiters)
                                               resource)
                                (const #t))
                               (wrap-operation (sleep-operation 1)
                                               (const #f))))))
                        (unless checkout-success?
                          (set! checkout-failure-count
                                (+ 1 checkout-failure-count)))

                        (if checkout-success?
                            (loop resources
                                  available
                                  (drop-right! waiters 1)
                                  (begin
                                    (list-set!
                                     resources-last-used
                                     (list-index (lambda (x)
                                                   (eq? x resource))
                                                 resources)
                                     (get-internal-real-time))
                                    resources-last-used))
                            (begin
                              (for-each
                               (lambda (waiter)
                                 (spawn-fiber
                                  (lambda ()
                                    (perform-operation
                                     (choice-operation
                                      (put-operation waiter 'resource-pool-retry-checkout)
                                      (sleep-operation 0.2))))))
                               waiters)

                              (loop resources
                                    (cons resource available)
                                    '()
                                    (begin
                                      (list-set!
                                       resources-last-used
                                       (list-index (lambda (x)
                                                     (eq? x resource))
                                                   resources)
                                       (get-internal-real-time))
                                      resources-last-used)))))
                      (begin
                        ;; When a resource is returned, prompt all the waiters
                        ;; to request again.  This is to avoid the pool waiting
                        ;; on channels that may be dead.
                        (for-each
                         (lambda (waiter)
                           (spawn-fiber
                            (lambda ()
                              (perform-operation
                               (choice-operation
                                (put-operation waiter 'resource-pool-retry-checkout)
                                (sleep-operation 0.2))))))
                         waiters)

                        (loop resources
                              (cons resource available)
                              ;; clear waiters, as they've been notified
                              '()
                              (begin
                                (list-set!
                                 resources-last-used
                                 (list-index (lambda (x)
                                               (eq? x resource))
                                             resources)
                                 (get-internal-real-time))
                                resources-last-used)))))
                 (('stats reply)
                  (let ((stats
                         `((resources              . ,(length resources))
                           (available              . ,(length available))
                           (waiters                . ,(length waiters))
                           (checkout-failure-count . ,checkout-failure-count))))

                    (spawn-fiber
                     (lambda ()
                       (perform-operation
                        (choice-operation
                         (wrap-operation
                          (put-operation reply stats)
                          (const #t))
                         (wrap-operation (sleep-operation 1)
                                         (const #f)))))))

                  (loop resources
                        available
                        waiters
                        resources-last-used))
                 (('check-for-idle-resources)
                  (let* ((resources-last-used-seconds
                          (map
                           (lambda (internal-time)
                             (/ (- (get-internal-real-time) internal-time)
                                internal-time-units-per-second))
                           resources-last-used))
                         (resources-to-destroy
                          (filter-map
                           (lambda (resource last-used-seconds)
                             (if (and (member resource available)
                                      (> last-used-seconds idle-seconds))
                                 resource
                                 #f))
                           resources
                           resources-last-used-seconds)))

                    (for-each
                     (lambda (resource)
                       (destructor/safe resource))
                     resources-to-destroy)

                    (loop (lset-difference eq? resources resources-to-destroy)
                          (lset-difference eq? available resources-to-destroy)
                          waiters
                          (filter-map
                           (lambda (resource last-used)
                             (if (memq resource resources-to-destroy)
                                 #f
                                 last-used))
                           resources
                           resources-last-used))))
                 (('destroy reply)
                  (if (= (length resources) (length available))
                      (begin
                        (for-each
                         (lambda (resource)
                           (destructor/safe resource))
                         resources)
                        (put-message reply 'destroy-success))
                      (begin
                        (spawn-fiber
                         (lambda ()
                           (perform-operation
                            (choice-operation
                             (put-operation reply 'resource-pool-destroy-failed)
                             (sleep-operation 10)))))
                        (loop resources
                              available
                              waiters
                              resources-last-used))))
                 (unknown
                  (simple-format
                   (current-error-port)
                   "unrecognised message to ~A resource pool channel: ~A\n"
                   name
                   unknown)
                  (loop resources
                        available
                        waiters
                        resources-last-used)))))
           #:unwind? #t)))
     (or scheduler
         (current-scheduler)))

    (make-resource-pool-record name channel)))

(define (destroy-resource-pool pool)
  (let ((reply (make-channel)))
    (put-message (resource-pool-channel pool)
                 (list 'destroy reply))
    (let ((msg (get-message reply)))
      (unless (eq? msg 'destroy-success)
        (error msg)))))

(define resource-pool-default-timeout
  (make-parameter #f))

(define resource-pool-retry-checkout-timeout
  (make-parameter 5))

(define &resource-pool-timeout
  (make-exception-type '&recource-pool-timeout
                       &error
                       '(name)))

(define make-resource-pool-timeout-error
  (record-constructor &resource-pool-timeout))

(define resource-pool-timeout-error?
  (record-predicate &resource-pool-timeout))

(define %resource-pool-timeout-handler
  (make-parameter #f))

(define* (call-with-resource-from-pool pool proc #:key (timeout 'default)
                                       (timeout-handler (%resource-pool-timeout-handler)))
  "Call PROC with a resource from POOL, blocking until a resource becomes
available.  Return the resource once PROC has returned."

  (define retry-timeout
    (resource-pool-retry-checkout-timeout))

  (define timeout-or-default
    (if (eq? timeout 'default)
        (resource-pool-default-timeout)
        timeout))

  (let ((resource
         (let ((reply (make-channel)))
           (let loop ((start-time (get-internal-real-time)))
             (let ((request-success?
                    (perform-operation
                     (choice-operation
                      (wrap-operation
                       (put-operation (resource-pool-channel pool)
                                      `(checkout ,reply))
                       (const #t))
                      (wrap-operation (sleep-operation (or timeout-or-default
                                                           retry-timeout))
                                      (const #f))))))
               (if request-success?
                   (let ((time-remaining
                          (- (or timeout-or-default
                                 retry-timeout)
                             (/ (- (get-internal-real-time)
                                   start-time)
                                internal-time-units-per-second))))
                     (if (> time-remaining 0)
                         (let ((response
                                (perform-operation
                                 (choice-operation
                                  (get-operation reply)
                                  (wrap-operation (sleep-operation time-remaining)
                                                  (const #f))))))
                           (if (or (not response)
                                   (eq? response 'resource-pool-retry-checkout))
                               (if (> (- (or timeout-or-default
                                             retry-timeout)
                                         (/ (- (get-internal-real-time)
                                               start-time)
                                            internal-time-units-per-second))
                                      0)
                                   (loop start-time)
                                   (if (eq? timeout-or-default #f)
                                       (loop (get-internal-real-time))
                                       #f))
                               response))
                         (if (eq? timeout-or-default #f)
                             (loop (get-internal-real-time))
                             #f)))
                   (if (eq? timeout-or-default #f)
                       (loop (get-internal-real-time))
                       #f)))))))

    (when (or (not resource)
              (eq? resource 'resource-pool-retry-checkout))
      (when timeout-handler
        (timeout-handler pool proc timeout))

      (raise-exception
       (make-resource-pool-timeout-error (resource-pool-name pool))))

    (with-exception-handler
        (lambda (exception)
          (put-message (resource-pool-channel pool)
                       `(return ,resource))
          (raise-exception exception))
      (lambda ()
        (call-with-values
            (lambda ()
              (with-throw-handler #t
                (lambda ()
                  (proc resource))
                (lambda _
                  (backtrace))))
          (lambda vals
            (put-message (resource-pool-channel pool)
                         `(return ,resource))
            (apply values vals))))
      #:unwind? #t)))

(define-syntax-rule (with-resource-from-pool pool resource exp ...)
  (call-with-resource-from-pool
      pool
    (lambda (resource) exp ...)))

(define* (resource-pool-stats pool #:key (timeout 5))
  (let ((reply (make-channel))
        (start-time (get-internal-real-time)))
    (perform-operation
     (choice-operation
      (wrap-operation
       (put-operation (resource-pool-channel pool)
                      `(stats ,reply))
       (const #t))
      (wrap-operation (sleep-operation timeout)
                      (lambda _
                        (raise-exception
                         (make-resource-pool-timeout-error))))))

    (let ((time-remaining
           (- timeout
              (/ (- (get-internal-real-time)
                    start-time)
                 internal-time-units-per-second))))
      (if (> time-remaining 0)
          (perform-operation
           (choice-operation
            (get-operation reply)
            (wrap-operation (sleep-operation time-remaining)
                            (lambda _
                              (raise-exception
                               (make-resource-pool-timeout-error))))))
          (raise-exception
           (make-resource-pool-timeout-error))))))

(define (call-with-default-io-waiters thunk)
  (parameterize
      ((current-read-waiter (@@ (ice-9 suspendable-ports)
                                default-read-waiter))
       (current-write-waiter (@@ (ice-9 suspendable-ports)
                                 default-write-waiter)))
    (thunk)))

(define %worker-thread-args
  (make-parameter #f))

(define* (make-worker-thread-channel initializer
                                     #:key (parallelism 1)
                                     (delay-logger (lambda _ #f))
                                     (duration-logger (const #f))
                                     destructor
                                     lifetime
                                     (log-exception? (const #t))
                                     (expire-on-exception? #f)
                                     (name "unnamed"))
  "Return a channel used to offload work to a dedicated thread.  ARGS are the
arguments of the worker thread procedure."
  (define thread-proc-vector
    (make-vector parallelism #f))

  (define (initializer/safe)
    (let ((args
           (with-exception-handler
               (lambda (exn)
                 (simple-format
                  (current-error-port)
                  "exception running initializer in worker thread (~A): ~A:\n  ~A\n"
                  name
                  initializer
                  exn)
                 #f)
             (lambda ()
               (with-throw-handler #t
                 initializer
                 (lambda args
                   (backtrace))))
             #:unwind? #t)))

      (if args
          args
          ;; never give up, just keep retrying
          (begin
            (sleep 1)
            (initializer/safe)))))

  (define (destructor/safe args)
    (let ((success?
           (with-exception-handler
               (lambda (exn)
                 (simple-format
                  (current-error-port)
                  "exception running destructor in worker thread (~A): ~A:\n  ~A\n"
                  name
                  destructor
                  exn)
                 #f)
             (lambda ()
               (with-throw-handler #t
                 (lambda ()
                   (apply destructor args)
                   #t)
                 (lambda _
                   (backtrace))))
             #:unwind? #t)))

      (or success?
          #t
          (begin
            (sleep 1)
            (destructor/safe args)))))

  (define (process thread-index channel args)
    (let loop ((current-lifetime lifetime))
      (let ((exception?
             (match (get-message channel)
               (((? channel? reply) sent-time (? procedure? proc))
                (let ((time-delay
                       (- (get-internal-real-time)
                          sent-time)))
                  (delay-logger (/ time-delay
                                   internal-time-units-per-second))

                  (let* ((start-time (get-internal-real-time))
                         (response
                          (with-exception-handler
                              (lambda (exn)
                                (list 'worker-thread-error
                                      (/ (- (get-internal-real-time)
                                            start-time)
                                         internal-time-units-per-second)
                                      exn))
                            (lambda ()
                              (vector-set! thread-proc-vector
                                           thread-index
                                           proc)
                              (with-throw-handler #t
                                (lambda ()
                                  (call-with-values
                                      (lambda ()
                                        (start-stack
                                         'worker-thread
                                         (apply proc args)))
                                    (lambda vals
                                      (cons (/ (- (get-internal-real-time)
                                                  start-time)
                                               internal-time-units-per-second)
                                            vals))))
                                (lambda args
                                  (when (match args
                                          (('%exception exn)
                                           (log-exception? exn))
                                          (_ #t))
                                    (simple-format
                                     (current-error-port)
                                     "worker-thread: exception: ~A\n" args)
                                    (backtrace)))))
                            #:unwind? #t)))
                    (put-message reply
                                 response)

                    (vector-set! thread-proc-vector
                                 thread-index
                                 #f)

                    (match response
                      (('worker-thread-error duration _)
                       (when duration-logger
                         (duration-logger duration proc))
                       #t)
                      ((duration . _)
                       (when duration-logger
                         (duration-logger duration proc))
                       #f))))))))
        (unless (and expire-on-exception?
                     exception?)
          (if (number? current-lifetime)
              (unless (< current-lifetime 0)
                (loop (if current-lifetime
                          (- current-lifetime 1)
                          #f)))
              (loop #f))))))

  (let ((channel (make-channel)))
    (for-each
     (lambda (thread-index)
       (call-with-new-thread
        (lambda ()
          (catch 'system-error
            (lambda ()
              (set-thread-name
               (string-append
                name " w t "
                (number->string thread-index))))
            (const #t))

          (let init ((args (initializer/safe)))
            (with-exception-handler
                (lambda (exn)
                  (simple-format
                   (current-error-port)
                   "worker-thread-channel: exception: ~A\n" exn))
              (lambda ()
                (parameterize ((%worker-thread-args args))
                  (process thread-index channel args)))
              #:unwind? #t)

            (when destructor
              (destructor/safe args))

            (init (initializer/safe))))))
     (iota parallelism))

    (values channel
            thread-proc-vector)))

(define &worker-thread-timeout
  (make-exception-type '&worker-thread-timeout
                       &error
                       '()))

(define make-worker-thread-timeout-error
  (record-constructor &worker-thread-timeout))

(define worker-thread-timeout-error?
  (record-predicate &worker-thread-timeout))

(define %worker-thread-default-timeout
  (make-parameter 30))

(define* (call-with-worker-thread channel proc #:key duration-logger
                                  (timeout (%worker-thread-default-timeout)))
  "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))
               (operation-success?
                (perform-operation
                 (let ((put
                        (wrap-operation
                         (put-operation channel
                                        (list reply
                                              (get-internal-real-time)
                                              proc))
                         (const #t))))

                   (if timeout
                       (choice-operation
                        put
                        (wrap-operation (sleep-operation timeout)
                                        (const #f)))
                       put)))))

          (unless operation-success?
            (raise-exception
             (make-worker-thread-timeout-error)))

          (match (get-message reply)
            (('worker-thread-error duration exn)
             (when duration-logger
               (duration-logger duration))
             (raise-exception exn))
            ((duration . result)
             (when duration-logger
               (duration-logger duration))
             (apply values result)))))))

(define* (fiberize proc #:key (parallelism 1))
  (let ((channel (make-channel)))
    (for-each
     (lambda _
       (spawn-fiber
        (lambda ()
          (while #t
            (let ((reply-channel args (car+cdr
                                       (get-message channel))))
              (put-message
               reply-channel
               (with-exception-handler
                   (lambda (exn)
                     (cons 'exception exn))
                 (lambda ()
                   (with-throw-handler #t
                     (lambda ()
                       (call-with-values
                           (lambda ()
                             (apply proc args))
                         (lambda vals
                           (cons 'result vals))))
                     (lambda _
                       (backtrace))))
                 #:unwind? #t)))))
        #:parallel? #t))
     (iota parallelism))

    (lambda args
      (let ((reply-channel (make-channel)))
        (put-message channel (cons reply-channel args))
        (match (get-message reply-channel)
          (('result . vals) (apply values vals))
          (('exception . exn) (raise-exception exn)))))))

(define-record-type <fibers-promise>
  (make-fibers-promise thunk values-box evaluated-condition)
  fibers-promise?
  (thunk                fibers-promise-thunk)
  (values-box           fibers-promise-values-box)
  (evaluated-condition  fibers-promise-evaluated-condition))

(define (fibers-delay thunk)
  (make-fibers-promise
   thunk
   (make-atomic-box #f)
   (make-condition)))

(define (fibers-force fp)
  (let ((res (atomic-box-compare-and-swap!
              (fibers-promise-values-box fp)
              #f
              'started)))
    (if (eq? #f res)
        (call-with-values
            (lambda ()
              (with-exception-handler
                  (lambda (exn)
                    (atomic-box-set! (fibers-promise-values-box fp)
                                     exn)
                    (signal-condition!
                     (fibers-promise-evaluated-condition fp))
                    (raise-exception exn))
                (fibers-promise-thunk fp)
                #:unwind? #t))
          (lambda vals
            (atomic-box-set! (fibers-promise-values-box fp)
                             vals)
            (signal-condition!
             (fibers-promise-evaluated-condition fp))
            (apply values vals)))
        (if (eq? res 'started)
            (begin
              (wait (fibers-promise-evaluated-condition fp))
              (let ((result (atomic-box-ref (fibers-promise-values-box fp))))
                (if (exception? result)
                    (raise-exception result)
                    (apply values result))))
            (if (exception? res)
                (raise-exception res)
                (apply values res))))))

(define (fibers-promise-reset fp)
  (atomic-box-set! (fibers-promise-values-box fp)
                   #f))

;; 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 batch-size 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))))

(define (defer-to-parallel-fiber thunk)
  (let ((reply (make-channel)))
    (spawn-fiber
     (lambda ()
       (with-exception-handler
           (lambda (exn)
             (put-message reply (cons 'exception exn)))
         (lambda ()
           (call-with-values
               (lambda ()
                 (with-throw-handler #t
                   thunk
                   (lambda _
                     (backtrace))))
             (lambda vals
               (put-message reply vals))))
         #:unwind? #t))
     #:parallel? #t)
    reply))

(define (fetch-result-of-defered-thunks . reply-channels)
  (let ((responses (map get-message
                        reply-channels)))
    (map
     (match-lambda
       (('exception . exn)
        (raise-exception exn))
       (result
        (apply values result)))
     responses)))

(define (fibers-batch-map proc batch-size . lists)
  (let loop ((lists lists)
             (result '()))
    (let ((batch
           rest
           (get-batch batch-size lists)))
      (if (any null? batch)
          result
          (let ((response-channels
                 (apply map
                        (lambda args
                          (defer-to-parallel-fiber
                            (lambda ()
                              (apply proc args))))
                        batch)))
            (loop rest
                  (append! result
                           (apply fetch-result-of-defered-thunks
                                  response-channels))))))))

(define (fibers-map proc . lists)
  (apply fibers-batch-map proc 20 lists))

(define (fibers-batch-for-each proc batch-size . lists)
  (let loop ((lists lists))
    (let ((batch
           rest
           (get-batch batch-size lists)))
      (if (any null? batch)
          *unspecified*
          (let ((response-channels
                 (apply map
                        (lambda args
                          (defer-to-parallel-fiber
                            (lambda ()
                              (apply proc args))))
                        batch)))
            (apply fetch-result-of-defered-thunks
                   response-channels)
            (loop rest))))))

(define (fibers-for-each proc . lists)
  (apply fibers-batch-for-each proc 20 lists))

(define-syntax parallel-via-fibers
  (lambda (x)
    (syntax-case x ()
      ((_ e0 ...)
       (with-syntax (((tmp0 ...) (generate-temporaries (syntax (e0 ...)))))
         #'(let ((tmp0 (defer-to-parallel-fiber
                         (lambda ()
                           e0)))
                 ...)
             (apply values (fetch-result-of-defered-thunks tmp0 ...))))))))

(define-syntax-rule (letpar& ((v e) ...) b0 b1 ...)
  (call-with-values
      (lambda () (parallel-via-fibers e ...))
    (lambda (v ...)
      b0 b1 ...)))

(define (par-mapper' mapper cons)
  (lambda (proc . lists)
    (apply
     fetch-result-of-defered-thunks
     (let loop ((lists lists))
       (match lists
         (((heads tails ...) ...)
          (let ((tail (loop tails))
                (head (defer-to-parallel-fiber
                        (lambda ()
                          (apply proc heads)))))
            (cons head tail)))
         (_
          '()))))))

(define par-map& (par-mapper' map cons))

(define* (fibers-map-with-progress proc lists #:key report)
  (let loop ((channels-to-results
              (apply map
                     (lambda args
                       (cons (defer-to-parallel-fiber
                               (lambda ()
                                 (apply proc args)))
                             #f))
                     lists)))
    (let ((active-channels
           (filter-map car channels-to-results)))
      (when report
        (report (apply map
                       list
                       (map cdr channels-to-results)
                       lists)))
      (if (null? active-channels)
          (map
           (match-lambda
             ((#f . ('exception . exn))
              (raise-exception exn))
             ((#f . ('result . val))
              val))
           channels-to-results)
          (loop
           (perform-operation
            (apply
             choice-operation
             (filter-map
              (lambda (p)
                (match p
                  ((channel . _)
                   (if channel
                       (wrap-operation
                        (get-operation channel)
                        (lambda (result)
                          (map (match-lambda
                                 ((c . r)
                                  (if (eq? channel c)
                                      (cons #f
                                            (match result
                                              (('exception . exn)
                                               result)
                                              (_
                                               (cons 'result result))))
                                      (cons c r))))
                               channels-to-results)))
                       #f))))
              channels-to-results))))))))

(define (chunk lst max-length)
  (if (> (length lst)
         max-length)
      (call-with-values (lambda ()
                          (split-at lst max-length))
        (lambda (first-lst rest)
          (cons first-lst
                (chunk rest max-length))))
      (list lst)))

(define (chunk! lst max-length)
  (if (> (length lst)
         max-length)
      (call-with-values (lambda ()
                          (split-at! lst max-length))
        (lambda (first-lst rest)
          (cons first-lst
                (chunk! rest max-length))))
      (list lst)))

(define* (chunk-for-each! proc chunk-size #:rest lsts)
  (define (do-one-iteration lsts)
    (if (> (length (car lsts))
           chunk-size)
        (let ((chunks-and-rest
               (map (lambda (lst)
                      (call-with-values (lambda ()
                                          (split-at! lst chunk-size))
                        (lambda (first-lst rest)
                          (cons first-lst
                                rest))))
                    lsts)))
          (apply proc
                 (map car chunks-and-rest))
          (do-one-iteration
           (map cdr chunks-and-rest)))
        (apply proc lsts)))

  (let ((list-lengths (map length lsts)))
    (unless (eq? 1 (length (delete-duplicates list-lengths)))
      (error "lists not equal length"))

    (unless (eq? 0 (first list-lengths))
      (do-one-iteration lsts)))

  #t)

(define* (delete-duplicates/sort! unsorted-lst less #:optional (equal? equal?))
  (if (null? unsorted-lst)
      unsorted-lst
      (let ((sorted-lst (sort! unsorted-lst less)))

        (let loop ((lst (cdr sorted-lst))
                   (last-element (car sorted-lst))
                   (result (list (car sorted-lst))))
          (if (null? lst)
              result
              (let ((current-element (car lst)))
                (if (equal? current-element last-element)
                    (loop (cdr lst)
                          last-element
                          result)
                    (loop (cdr lst)
                          current-element
                          (cons current-element
                                result)))))))))

(define (get-guix-metrics-updater registry)
  (define guix-db "/var/guix/db/db.sqlite")
  (define guix-db-wal (string-append guix-db "-wal"))

  (let ((guix-db-bytes-metric
         (make-gauge-metric registry "guix_db_bytes"))
        (guix-db-wal-bytes-metric
         (make-gauge-metric registry "guix_db_wal_bytes")))
    (lambda ()
      (with-exception-handler
          (lambda _
            #f)
        (lambda ()
          (metric-set guix-db-bytes-metric (stat:size (stat guix-db)))
          (metric-set guix-db-wal-bytes-metric
                      (if (file-exists? guix-db-wal)
                          (stat:size (stat guix-db-wal))
                          0)))
        #:unwind? #t))))

;; This variant of run-server from the fibers library supports running
;; multiple servers within one process.
(define run-server/patched
  (let ((fibers-web-server-module
         (resolve-module '(fibers web server))))

    (define set-nonblocking!
      (module-ref fibers-web-server-module 'set-nonblocking!))

    (define make-default-socket
      (module-ref fibers-web-server-module 'make-default-socket))

    (define socket-loop
      (module-ref fibers-web-server-module 'socket-loop))

    (lambda* (handler
              #:key
              (host #f)
              (family AF_INET)
              (addr (if host
                        (inet-pton family host)
                        INADDR_LOOPBACK))
              (port 8080)
              (socket (make-default-socket family addr port)))
      ;; We use a large backlog by default.  If the server is suddenly hit
      ;; with a number of connections on a small backlog, clients won't
      ;; receive confirmation for their SYN, leading them to retry --
      ;; probably successfully, but with a large latency.
      (listen socket 1024)
      (set-nonblocking! socket)
      (sigaction SIGPIPE SIG_IGN)
      (spawn-fiber (lambda () (socket-loop socket handler))))))

(define &port-timeout
  (make-exception-type '&port-timeout
                       &external-error
                       '(port)))

(define make-port-timeout-error
  (record-constructor &port-timeout))

(define port-timeout-error?
  (record-predicate &port-timeout))

(define &port-read-timeout
  (make-exception-type '&port-read-timeout
                       &port-timeout
                       '()))

(define make-port-read-timeout-error
  (record-constructor &port-read-timeout))

(define port-read-timeout-error?
  (record-predicate &port-read-timeout))

(define &port-write-timeout
  (make-exception-type '&port-write-timeout
                       &port-timeout
                       '()))

(define make-port-write-timeout-error
  (record-constructor &port-write-timeout))

(define port-write-timeout-error?
  (record-predicate &port-write-timeout))

;; These procedure are subject to spurious wakeups.

(define (readable? port)
  "Test if PORT is writable."
  (match (select (vector port) #() #() 0)
    ((#() #() #()) #f)
    ((#(_) #() #()) #t)))

(define (writable? port)
  "Test if PORT is writable."
  (match (select #() (vector port) #() 0)
    ((#() #() #()) #f)
    ((#() #(_) #()) #t)))

(define (make-wait-operation ready? schedule-when-ready port
                             port-ready-fd this-procedure)
  (make-base-operation
   #f
   (lambda _
     (and (ready? (port-ready-fd port)) values))
   (lambda (flag sched resume)
     (define (commit)
       (match (atomic-box-compare-and-swap! flag 'W 'S)
         ('W (resume values))
         ('C (commit))
         ('S #f)))
     (schedule-when-ready
      sched (port-ready-fd port) commit))))

(define (wait-until-port-readable-operation port)
  "Make an operation that will succeed when PORT is readable."
  (unless (input-port? port)
    (error "refusing to wait forever for input on non-input port"))
  (make-wait-operation readable? schedule-task-when-fd-readable port
                       port-read-wait-fd
                       wait-until-port-readable-operation))

(define (wait-until-port-writable-operation port)
  "Make an operation that will succeed when PORT is writable."
  (unless (output-port? port)
    (error "refusing to wait forever for output on non-output port"))
  (make-wait-operation writable? schedule-task-when-fd-writable port
                       port-write-wait-fd
                       wait-until-port-writable-operation))

(define* (with-fibers-port-timeouts thunk
                                    #:key timeout
                                    (read-timeout timeout)
                                    (write-timeout timeout))
  (define (no-fibers-wait port mode timeout)
    (define poll-timeout-ms 200)

    ;; When the GC runs, it restarts the poll syscall, but the timeout
    ;; remains unchanged! When the timeout is longer than the time
    ;; between the syscall restarting, I think this renders the
    ;; timeout useless. Therefore, this code uses a short timeout, and
    ;; repeatedly calls poll while watching the clock to see if it has
    ;; timed out overall.
    (let ((timeout-internal
           (+ (get-internal-real-time)
              (* internal-time-units-per-second
                 (/ timeout 1000)))))
      (let loop ((poll-value
                  (port-poll port mode poll-timeout-ms)))
        (if (= poll-value 0)
            (if (> (get-internal-real-time)
                   timeout-internal)
                (raise-exception
                 (if (string=? mode "r")
                     (make-port-read-timeout-error port)
                     (make-port-write-timeout-error port)))
                (loop (port-poll port mode poll-timeout-ms)))
            poll-value))))

  (parameterize
      ((current-read-waiter
        (lambda (port)
          (if (current-scheduler)
              (perform-operation
               (choice-operation
                (wait-until-port-readable-operation port)
                (wrap-operation
                 (sleep-operation read-timeout)
                 (lambda ()
                   (raise-exception
                    (make-port-read-timeout-error thunk port))))))
              (no-fibers-wait port "r" read-timeout))))
       (current-write-waiter
        (lambda (port)
          (if (current-scheduler)
              (perform-operation
               (choice-operation
                (wait-until-port-writable-operation port)
                (wrap-operation
                 (sleep-operation write-timeout)
                 (lambda ()
                   (raise-exception
                    (make-port-write-timeout-error thunk port))))))
              (no-fibers-wait port "w" write-timeout)))))
    (thunk)))

(define (spawn-port-monitoring-fiber port error-condition)
  (spawn-fiber
   (lambda ()
     (while #t
       (sleep 20)
       (with-exception-handler
           (lambda (exn)
             (simple-format (current-error-port)
                            "port monitoring fiber failed to connect to ~A: ~A\n"
                            port exn)
             (signal-condition! error-condition))
         (lambda ()
           (with-fibers-port-timeouts
            (lambda ()
              (let ((sock (socket PF_INET SOCK_STREAM 0)))
                (connect sock AF_INET INADDR_LOOPBACK port)
                (close-port sock)))
            #:timeout 20))
         #:unwind? #t)))))

;; Copied from (fibers web server)
(define (call-with-sigint thunk cvar)
  (let ((handler #f))
    (dynamic-wind
      (lambda ()
        (set! handler
          (sigaction SIGINT (lambda (sig) (signal-condition! cvar)))))
      thunk
      (lambda ()
        (if handler
            ;; restore Scheme handler, SIG_IGN or SIG_DFL.
            (sigaction SIGINT (car handler) (cdr handler))
            ;; restore original C handler.
            (sigaction SIGINT #f))))))

(define (make-queueing-channel channel)
  (define queue (make-q))

  (let ((queue-channel (make-channel)))
    (spawn-fiber
     (lambda ()
       (while #t
         (if (q-empty? queue)
             (enq! queue
                   (perform-operation
                    (get-operation queue-channel)))
             (let ((front (q-front queue)))
               (perform-operation
                (choice-operation
                 (wrap-operation (get-operation queue-channel)
                                 (lambda (val)
                                   (enq! queue val)))
                 (wrap-operation (put-operation channel front)
                                 (lambda _
                                   (q-pop! queue))))))))))
    queue-channel))