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-11)
  #:use-module (ice-9 match)
  #:use-module (ice-9 format)
  #:use-module (ice-9 threads)
  #:use-module (fibers)
  #:use-module (fibers channels)
  #:use-module (fibers operations)
  #:use-module (fibers timers)
  #:use-module (fibers conditions)
  #:use-module (prometheus)
  #:export (call-with-time-logging
            with-time-logging
            prevent-inlining-for-tests

            resource-pool-default-timeout
            make-resource-pool
            call-with-resource-from-pool
            with-resource-from-pool
            resource-pool-stats

            parallel-via-fibers
            par-map&
            letpar&

            chunk
            chunk!
            chunk-for-each!

            delete-duplicates/sort!

            get-gc-metrics-updater

            call-with-sigint
            run-server/patched))

(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* (make-resource-pool initializer max-size
                             #:key (min-size max-size)
                             (idle-duration #f)
                             (delay-logger (const #f))
                             (duration-logger (const #f))
                             destructor
                             lifetime
                             (name "unnamed"))
  (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))

  (let ((channel (make-channel)))
    (spawn-fiber
     (lambda ()
       (let loop ((resources '())
                  (available '())
                  (waiters '()))

         (match (get-message channel)
           (('checkout reply)
            (if (null? available)
                (if (= (length resources) max-size)
                    (loop resources
                          available
                          (cons reply waiters))
                    (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 0.2)
                                                   (const #f))))))
                            (loop (cons new-resource resources)
                                  (if checkout-success?
                                      available
                                      (cons new-resource available))
                                  waiters))
                          (loop resources
                                available
                                (cons reply waiters)))))
                (let ((checkout-success?
                       (perform-operation
                        (choice-operation
                         (wrap-operation
                          (put-operation reply (car available))
                          (const #t))
                         (wrap-operation (sleep-operation 0.2)
                                         (const #f))))))
                  (if checkout-success?
                      (loop resources
                            (cdr available)
                            waiters)
                      (loop resources
                            available
                            waiters)))))
           (('return resource)
            ;; 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
                  '()))
           (('stats reply)
            (let ((stats
                   `((resources . ,(length resources))
                     (available . ,(length available))
                     (waiters   . ,(length waiters)))))

              (perform-operation
               (choice-operation
                (wrap-operation
                 (put-operation reply stats)
                 (const #t))
                (wrap-operation (sleep-operation 0.2)
                                (const #f)))))

            (loop resources
                  available
                  waiters))
           (unknown
            (simple-format
             (current-error-port)
             "unrecognised message to ~A resource pool channel: ~A\n"
             name
             unknown)
            (loop resources
                  available
                  waiters))))))

    channel))

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

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

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

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

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

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

  (let ((resource
         (let ((reply (make-channel)))
           (if timeout-or-default
               (let loop ((start-time (get-internal-real-time)))
                 (perform-operation
                  (choice-operation
                   (wrap-operation
                    (put-operation pool `(checkout ,reply))
                    (const #t))
                   (wrap-operation (sleep-operation timeout-or-default)
                                   (const #f))))

                 (let ((time-remaining
                        (- timeout-or-default
                           (/ (- (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 (> (- timeout-or-default
                                       (/ (- (get-internal-real-time)
                                             start-time)
                                          internal-time-units-per-second))
                                    0)
                                 (loop start-time)
                                 #f)
                             response))
                       #f)))
               (begin
                 (put-message pool `(checkout ,reply))
                 (get-message reply))))))

    (when (or (not resource)
              (eq? resource 'resource-pool-retry-checkout))
      (raise-exception
       (make-resource-pool-timeout-error)))

    (with-exception-handler
        (lambda (exception)
          (put-message pool `(return ,resource))
          (raise-exception exception))
      (lambda ()
        (call-with-values
            (lambda ()
             (proc resource))
          (lambda vals
            (put-message 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 pool `(stats ,reply))
       (const #t))
      (wrap-operation (sleep-operation timeout)
                      (const #f))))

    (let ((time-remaining
           (- 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))))))
            response)
          (raise-exception
           (make-resource-pool-timeout-error))))))

(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 thunk
             (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-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)
    (let loop ((lists lists))
      (match lists
        (((heads tails ...) ...)
         (let ((tail (loop tails))
               (head (defer-to-parallel-fiber
                       (lambda ()
                         (apply proc heads)))))
           (cons (fetch-result-of-defered-thunks head) tail)))
        (_
         '())))))

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

(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)
  (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 (eq? current-element last-element)
                    (loop (cdr lst)
                          last-element
                          result)
                    (loop (cdr lst)
                          current-element
                          (cons current-element
                                result)))))))))

(define (get-gc-metrics-updater registry)
  (define metrics
    `((gc-time-taken
       . ,(make-gauge-metric registry "guile_gc_time_taken"))
      (heap-size
       . ,(make-gauge-metric registry "guile_heap_size"))
      (heap-free-size
       . ,(make-gauge-metric registry "guile_heap_free_size"))
      (heap-total-allocated
       . ,(make-gauge-metric registry "guile_heap_total_allocated"))
      (heap-allocated-since-gc
       . ,(make-gauge-metric registry "guile_allocated_since_gc"))
      (protected-objects
       . ,(make-gauge-metric registry "guile_gc_protected_objects"))
      (gc-times
       . ,(make-gauge-metric registry "guile_gc_times"))))

  (lambda ()
    (let ((stats (gc-stats)))
      (for-each
       (match-lambda
         ((name . metric)
          (let ((value (assq-ref stats name)))
            (metric-set metric value))))
       metrics))))

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

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