aboutsummaryrefslogtreecommitdiff
path: root/guix-data-service
diff options
context:
space:
mode:
authorChristopher Baines <mail@cbaines.net>2024-01-18 14:39:39 +0000
committerChristopher Baines <mail@cbaines.net>2024-01-18 14:41:32 +0000
commit5af6233e5b411d1735f704d311ee46c0eec8ab6f (patch)
tree41cdf4a136efaa41f1bf5477ce7c820aed40edfe /guix-data-service
parent15b6dad5a56a48f81d2a0bb45de9224c5e8ed805 (diff)
downloaddata-service-5af6233e5b411d1735f704d311ee46c0eec8ab6f.tar
data-service-5af6233e5b411d1735f704d311ee46c0eec8ab6f.tar.gz
Make it possible to destroy a resource pool
And implement removing idle resources.
Diffstat (limited to 'guix-data-service')
-rw-r--r--guix-data-service/utils.scm141
1 files changed, 127 insertions, 14 deletions
diff --git a/guix-data-service/utils.scm b/guix-data-service/utils.scm
index c91b10b..e1c6d84 100644
--- a/guix-data-service/utils.scm
+++ b/guix-data-service/utils.scm
@@ -34,6 +34,7 @@
resource-pool-default-timeout
make-resource-pool
+ destroy-resource-pool
call-with-resource-from-pool
with-resource-from-pool
resource-pool-stats
@@ -73,7 +74,7 @@
(define* (make-resource-pool initializer max-size
#:key (min-size max-size)
- (idle-duration #f)
+ (idle-seconds #f)
(delay-logger (const #f))
(duration-logger (const #f))
destructor
@@ -96,6 +97,32 @@
(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)))
(spawn-fiber
(lambda ()
@@ -110,15 +137,25 @@
(lambda ()
(let loop ((resources '())
(available '())
- (waiters '()))
-
- (match (get-message channel)
+ (waiters '())
+ (resources-last-used '()))
+
+ (match (if idle-seconds
+ (perform-operation
+ (choice-operation
+ (get-operation channel)
+ (wrap-operation
+ ;; TODO Do something smarter
+ (sleep-operation 10)
+ (const '(check-for-idle-resources)))))
+ (get-message channel))
(('checkout reply)
(if (null? available)
(if (= (length resources) max-size)
(loop resources
available
- (cons reply waiters))
+ (cons reply waiters)
+ resources-last-used)
(let ((new-resource (initializer/safe)))
(if new-resource
(let ((checkout-success?
@@ -133,10 +170,13 @@
(if checkout-success?
available
(cons new-resource available))
- waiters))
+ waiters
+ (cons (get-internal-real-time)
+ resources-last-used)))
(loop resources
available
- (cons reply waiters)))))
+ (cons reply waiters)
+ resources-last-used))))
(let ((checkout-success?
(perform-operation
(choice-operation
@@ -148,10 +188,12 @@
(if checkout-success?
(loop resources
(cdr available)
- waiters)
+ waiters
+ resources-last-used)
(loop resources
available
- waiters)))))
+ waiters
+ resources-last-used)))))
(('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
@@ -169,7 +211,15 @@
(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))
@@ -186,7 +236,59 @@
(loop resources
available
- waiters))
+ 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)
@@ -195,11 +297,19 @@
unknown)
(loop resources
available
- waiters)))))
+ waiters
+ resources-last-used)))))
#:unwind? #t))))
channel))
+(define (destroy-resource-pool pool)
+ (let ((reply (make-channel)))
+ (put-message pool (list 'destroy reply))
+ (let ((msg (get-message reply)))
+ (unless (eq? msg 'destroy-success)
+ (error msg)))))
+
(define resource-pool-default-timeout
(make-parameter #f))
@@ -258,9 +368,12 @@ available. Return the resource once PROC has returned."
#f)
response))
#f)))
- (begin
+ (let loop ()
(put-message pool `(checkout ,reply))
- (get-message reply))))))
+ (let ((response (get-message reply)))
+ (if (eq? response 'resource-pool-retry-checkout)
+ (loop)
+ response)))))))
(when (or (not resource)
(eq? resource 'resource-pool-retry-checkout))