cbaines/guile-knots
1
Fork 0
Code Issues Pull requests Activity Actions

Use a queue for the resource pool waiters
All checks were successful
/ test (push) Successful in 8s
Details

Browse source 
As this will maybe improve performance.
This commit is contained in:
Christopher Baines 2025-06-30 22:57:08 +01:00
parent 7709ffe1d3
commit ce1b710bcf
1 changed files with 141 additions and 136 deletions
Download patch file Download diff file Expand all files Collapse all files

277
knots/resource-pool.scm
View file

@ -22,6 +22,7 @@
#:use-module (srfi srfi-9) #:use-module (srfi srfi-9)
#:use-module (srfi srfi-9 gnu) #:use-module (srfi srfi-9 gnu)
#:use-module (srfi srfi-71) #:use-module (srfi srfi-71)
#:use-module (ice-9 q)
#:use-module (ice-9 match) #:use-module (ice-9 match)
#:use-module (ice-9 exceptions) #:use-module (ice-9 exceptions)
#:use-module (fibers) #:use-module (fibers)
@ -267,13 +268,13 @@
(define (main-loop) (define (main-loop)
(let loop ((resources resources) (let loop ((resources resources)
(available resources) (available resources)
(waiters '())) (waiters (make-q)))
(match (get-message channel) (match (get-message channel)
(('checkout reply timeout-time max-waiters) (('checkout reply timeout-time max-waiters)
(if (null? available) (if (null? available)
(let ((waiters-count (let ((waiters-count
(length waiters))) (q-length waiters)))
(if (and max-waiters (if (and max-waiters
(>= waiters-count (>= waiters-count
max-waiters)) max-waiters))
@ -301,8 +302,7 @@
waiters)) waiters))
(loop resources (loop resources
available available
(cons (cons reply timeout-time) (enq! waiters (cons reply timeout-time)))))
waiters))))
(if timeout-time (if timeout-time
(let ((current-internal-time (let ((current-internal-time
@ -345,44 +345,46 @@
(set! checkout-failure-count (set! checkout-failure-count
(+ 1 checkout-failure-count))) (+ 1 checkout-failure-count)))
(if (null? waiters) (if (q-empty? waiters)
(loop resources (loop resources
(cons resource available) (cons resource available)
waiters) waiters)
(let* ((current-internal-time (get-internal-real-time)) (let ((current-internal-time
(alive-waiters (get-internal-real-time)))
dead-waiters (with-exception-handler
(partition! (lambda (exn)
(match-lambda (if (eq? (exception-kind exn) 'q-empty)
((reply . timeout) (loop resources
(or (not timeout) (cons resource available)
(> timeout current-internal-time)))) waiters)
waiters))) (raise-exception exn)))
(if (null? alive-waiters) (lambda ()
(loop resources (let waiter-loop ((waiter (deq! waiters)))
(cons resource available) (match waiter
'()) ((reply . timeout)
(match (last alive-waiters) (if (and timeout
((waiter-channel . waiter-timeout) (< timeout current-internal-time))
(if waiter-timeout (waiter-loop (deq! waiters))
(let ((reply-timeout (begin
(/ (- waiter-timeout (if timeout
current-internal-time) (let ((reply-timeout
internal-time-units-per-second))) (/ (- timeout
;; Don't sleep in this fiber, so spawn a current-internal-time)
;; new fiber to handle handing over the internal-time-units-per-second)))
;; resource, and returning it if there's a ;; Don't sleep in this fiber, so spawn a
;; timeout ;; new fiber to handle handing over the
(spawn-fiber-for-checkout waiter-channel ;; resource, and returning it if there's
reply-timeout ;; a timeout
resource)) (spawn-fiber-for-checkout reply
(put-message waiter-channel (cons 'success reply-timeout
resource))) resource))
(put-message reply (cons 'success
(loop resources resource)))
available (loop resources
(drop-right! alive-waiters 1)))))))) available
waiters)))))))
#:unwind? #t))))
(('list-resources reply) (('list-resources reply)
(spawn-fiber (spawn-fiber
@ -397,7 +399,7 @@
(let ((stats (let ((stats
`((resources . ,(length resources)) `((resources . ,(length resources))
(available . ,(length available)) (available . ,(length available))
(waiters . ,(length waiters)) (waiters . ,(q-length waiters))
(checkout-failure-count . ,checkout-failure-count)))) (checkout-failure-count . ,checkout-failure-count))))
(spawn-fiber (spawn-fiber
@ -420,7 +422,7 @@
(('destroy) (('destroy)
(if (and (null? resources) (if (and (null? resources)
(null? waiters)) (q-empty? waiters))
(signal-condition! (signal-condition!
destroy-condition) destroy-condition)
@ -448,7 +450,7 @@
internal-time-units-per-second)) internal-time-units-per-second))
(const #f))) (const #f)))
op)))))))) op))))))))
waiters) (car waiters))
(if destructor (if destructor
(begin (begin
@ -747,7 +749,7 @@
(define (main-loop) (define (main-loop)
(let loop ((resources '()) (let loop ((resources '())
(available '()) (available '())
(waiters '()) (waiters (make-q))
(resources-last-used '())) (resources-last-used '()))
(match (get-message channel) (match (get-message channel)
@ -769,50 +771,52 @@
(cons (get-internal-real-time) (cons (get-internal-real-time)
resources-last-used)))) resources-last-used))))
(if (null? waiters) (if (q-empty? waiters)
(loop (cons resource resources) (loop (cons resource resources)
(cons resource available) (cons resource available)
waiters waiters
(cons (get-internal-real-time) (cons (get-internal-real-time)
resources-last-used)) resources-last-used))
(let* ((current-internal-time (get-internal-real-time)) (let ((current-internal-time
(alive-waiters (get-internal-real-time)))
dead-waiters (with-exception-handler
(partition! (lambda (exn)
(match-lambda (if (eq? (exception-kind exn) 'q-empty)
((reply . timeout) (loop (cons resource resources)
(or (not timeout) (cons resource available)
(> timeout current-internal-time)))) waiters
waiters))) (cons current-internal-time
(if (null? alive-waiters) resources-last-used))
(loop (cons resource resources) (raise-exception exn)))
(cons resource available) (lambda ()
'() (let waiter-loop ((waiter (deq! waiters)))
(cons (get-internal-real-time) (match waiter
resources-last-used)) ((reply . timeout)
(match (last alive-waiters) (if (and timeout
((waiter-channel . waiter-timeout) (< timeout current-internal-time))
(if waiter-timeout (waiter-loop (deq! waiters))
(let ((reply-timeout (begin
(/ (- waiter-timeout (if timeout
current-internal-time) (let ((reply-timeout
internal-time-units-per-second))) (/ (- timeout
;; Don't sleep in this fiber, so spawn current-internal-time)
;; a new fiber to handle handing over internal-time-units-per-second)))
;; the resource, and returning it if ;; Don't sleep in this fiber, so spawn a
;; there's a timeout ;; new fiber to handle handing over the
(spawn-fiber-for-checkout waiter-channel ;; resource, and returning it if there's
reply-timeout ;; a timeout
resource)) (spawn-fiber-for-checkout reply
(put-message waiter-channel (cons 'success reply-timeout
resource))) resource))
(put-message reply (cons 'success
(loop (cons resource resources) resource)))
available (loop (cons resource resources)
(drop-right! alive-waiters 1) available
(cons (get-internal-real-time) waiters
resources-last-used))))))))) (cons current-internal-time
resources-last-used)))))))
#:unwind? #t))))))
(('checkout reply timeout-time max-waiters) (('checkout reply timeout-time max-waiters)
(if (null? available) (if (null? available)
@ -821,7 +825,7 @@
(spawn-fiber-to-return-new-resource)) (spawn-fiber-to-return-new-resource))
(let ((waiters-count (let ((waiters-count
(length waiters))) (q-length waiters)))
(if (and max-waiters (if (and max-waiters
(>= waiters-count (>= waiters-count
max-waiters)) max-waiters))
@ -850,8 +854,7 @@
resources-last-used)) resources-last-used))
(loop resources (loop resources
available available
(cons (cons reply timeout-time) (enq! waiters (cons reply timeout-time))
waiters)
resources-last-used)))) resources-last-used))))
(if timeout-time (if timeout-time
@ -898,7 +901,7 @@
(set! checkout-failure-count (set! checkout-failure-count
(+ 1 checkout-failure-count))) (+ 1 checkout-failure-count)))
(if (null? waiters) (if (q-empty? waiters)
(loop resources (loop resources
(cons resource available) (cons resource available)
waiters waiters
@ -911,56 +914,58 @@
(get-internal-real-time)) (get-internal-real-time))
resources-last-used)) resources-last-used))
(let* ((current-internal-time (get-internal-real-time)) (let ((current-internal-time
(alive-waiters (get-internal-real-time)))
dead-waiters (with-exception-handler
(partition! (lambda (exn)
(match-lambda (if (eq? (exception-kind exn) 'q-empty)
((reply . timeout) (loop resources
(or (not timeout) (cons resource available)
(> timeout current-internal-time)))) waiters
waiters))) (begin
(if (null? alive-waiters) (when (eq? return-type 'return)
(loop resources (list-set!
(cons resource available) resources-last-used
'() (list-index (lambda (x)
(begin (eq? x resource))
(when (eq? return-type 'return) resources)
(list-set! current-internal-time))
resources-last-used resources-last-used))
(list-index (lambda (x) (raise-exception exn)))
(eq? x resource)) (lambda ()
resources) (let waiter-loop ((waiter (deq! waiters)))
(get-internal-real-time))) (match waiter
resources-last-used)) ((reply . timeout)
(match (last alive-waiters) (if (and timeout
((waiter-channel . waiter-timeout) (< timeout current-internal-time))
(if waiter-timeout (waiter-loop (deq! waiters))
(let ((reply-timeout
(/ (- waiter-timeout
current-internal-time)
internal-time-units-per-second)))
;; Don't sleep in this fiber, so spawn a
;; new fiber to handle handing over the
;; resource, and returning it if there's a
;; timeout
(spawn-fiber-for-checkout waiter-channel
reply-timeout
resource))
(put-message waiter-channel (cons 'success
resource)))
(loop resources
available
(drop-right! alive-waiters 1)
(begin (begin
(list-set! (if timeout
resources-last-used (let ((reply-timeout
(list-index (lambda (x) (/ (- timeout
(eq? x resource)) current-internal-time)
resources) internal-time-units-per-second)))
(get-internal-real-time)) ;; Don't sleep in this fiber, so spawn a
resources-last-used)))))))) ;; new fiber to handle handing over the
;; resource, and returning it if there's
;; a timeout
(spawn-fiber-for-checkout reply
reply-timeout
resource))
(put-message reply (cons 'success
resource)))
(loop resources
available
waiters
(begin
(list-set!
resources-last-used
(list-index (lambda (x)
(eq? x resource))
resources)
current-internal-time)
resources-last-used))))))))
#:unwind? #t))))
(('remove resource) (('remove resource)
(let ((index (let ((index
@ -1003,7 +1008,7 @@
(let ((stats (let ((stats
`((resources . ,(length resources)) `((resources . ,(length resources))
(available . ,(length available)) (available . ,(length available))
(waiters . ,(length waiters)) (waiters . ,(q-length waiters))
(checkout-failure-count . ,checkout-failure-count)))) (checkout-failure-count . ,checkout-failure-count))))
(spawn-fiber (spawn-fiber
@ -1067,7 +1072,7 @@
(('destroy) (('destroy)
(if (and (null? resources) (if (and (null? resources)
(null? waiters)) (q-empty? waiters))
(signal-condition! (signal-condition!
destroy-condition) destroy-condition)
@ -1095,7 +1100,7 @@
internal-time-units-per-second)) internal-time-units-per-second))
(const #f))) (const #f)))
op)))))))) op))))))))
waiters) (car waiters))
(if destructor (if destructor
(begin (begin