536 lines
18 KiB
Scheme
536 lines
18 KiB
Scheme
;;; 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 ftw)
|
|
#: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
|
|
get-port-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 ()
|
|
(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 '()))
|
|
|
|
(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)))))
|
|
#:unwind? #t))))
|
|
|
|
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))))
|
|
|
|
(define (get-port-metrics-updater registry)
|
|
(let ((ports-metric
|
|
(make-gauge-metric registry "guile_ports_total"))
|
|
(fds-metric
|
|
(make-gauge-metric registry "file_descriptors_total")))
|
|
(lambda ()
|
|
(let ((count 0))
|
|
(port-for-each
|
|
(lambda _
|
|
(set! count (+ 1 count))))
|
|
|
|
(metric-set ports-metric count))
|
|
|
|
(metric-set
|
|
fds-metric
|
|
(length
|
|
;; In theory 'scandir' cannot return #f, but in practice,
|
|
;; we've seen it before.
|
|
(or (scandir "/proc/self/fd"
|
|
(lambda (file)
|
|
(not (member file '("." "..")))))
|
|
'()))))))
|
|
|
|
;; 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))))))
|