Compare commits
No commits in common. "c2e1cd94d7a8c7cb64d90fd15a29738905cb8295" and "d0ff89023b9fb0ff2617b95b45af90128fe6c8c7" have entirely different histories.
c2e1cd94d7
...
d0ff89023b
4 changed files with 68 additions and 278 deletions
|
|
@ -60,9 +60,6 @@
|
|||
make-resource-pool-destroy-resource-exception
|
||||
resource-pool-destroy-resource-exception?
|
||||
|
||||
resource-pool-delay-logger
|
||||
resource-pool-duration-logger
|
||||
|
||||
resource-pool-default-timeout-handler
|
||||
|
||||
call-with-resource-from-pool
|
||||
|
|
@ -106,14 +103,6 @@
|
|||
'documentation
|
||||
"Return the configuration alist of the resource pool.")
|
||||
|
||||
(define (resource-pool-delay-logger resource-pool)
|
||||
(assq-ref (resource-pool-configuration resource-pool)
|
||||
'delay-logger))
|
||||
|
||||
(define (resource-pool-duration-logger resource-pool)
|
||||
(assq-ref (resource-pool-configuration resource-pool)
|
||||
'duration-logger))
|
||||
|
||||
(set-record-type-printer!
|
||||
<resource-pool>
|
||||
(lambda (resource-pool port)
|
||||
|
|
@ -149,7 +138,7 @@
|
|||
(define-inlinable (decrement-resource-checkout-count! resource)
|
||||
(set-resource-details-checkout-count!
|
||||
resource
|
||||
(1- (resource-details-checkout-count resource))))
|
||||
(1+ (resource-details-checkout-count resource))))
|
||||
|
||||
(define (spawn-fiber-for-checkout channel
|
||||
reply-channel
|
||||
|
|
@ -175,8 +164,8 @@
|
|||
|
||||
(define* (make-fixed-size-resource-pool resources-list-or-vector
|
||||
#:key
|
||||
(delay-logger #f)
|
||||
(duration-logger #f)
|
||||
(delay-logger (const #f))
|
||||
(duration-logger (const #f))
|
||||
scheduler
|
||||
(name "unnamed")
|
||||
default-checkout-timeout
|
||||
|
|
@ -204,19 +193,6 @@ Maximum number of fibers that may queue waiting for a resource. When
|
|||
this limit is exceeded, @code{&resource-pool-too-many-waiters} is
|
||||
raised when a resource is requested. Defaults to @code{#f} (no limit).
|
||||
|
||||
@item #:delay-logger
|
||||
Called as @code{(delay-logger seconds)} with the time spent waiting
|
||||
for a resource to become available. Defaults to @code{#f} (no
|
||||
logging).
|
||||
|
||||
@item #:duration-logger
|
||||
Called as @code{(duration-logger seconds)} after the proc passed to
|
||||
@code{call-with-resource-from-pool} completes, whether it returned
|
||||
normally or raised an exception. Can be overridden per-call via the
|
||||
@code{#:duration-logger} keyword argument to
|
||||
@code{call-with-resource-from-pool}. Defaults to @code{#f} (no
|
||||
logging).
|
||||
|
||||
@item #:scheduler
|
||||
The Fibers scheduler to use for the pool's internal fiber. Defaults
|
||||
to the current scheduler.
|
||||
|
|
@ -571,8 +547,8 @@ to the current scheduler.
|
|||
(define* (make-resource-pool return-new-resource max-size
|
||||
#:key (min-size 0)
|
||||
(idle-seconds #f)
|
||||
(delay-logger #f)
|
||||
(duration-logger #f)
|
||||
(delay-logger (const #f))
|
||||
(duration-logger (const #f))
|
||||
destructor
|
||||
lifetime
|
||||
scheduler
|
||||
|
|
@ -629,19 +605,6 @@ Maximum number of fibers that may queue waiting for a resource. When
|
|||
this limit is exceeded, @code{&resource-pool-too-many-waiters} is
|
||||
raised when a resource is requested. Defaults to @code{#f} (no limit).
|
||||
|
||||
@item #:delay-logger
|
||||
Called as @code{(delay-logger seconds)} with the time spent waiting
|
||||
for a resource to become available. Defaults to @code{#f} (no
|
||||
logging).
|
||||
|
||||
@item #:duration-logger
|
||||
Called as @code{(duration-logger seconds)} after the proc passed to
|
||||
@code{call-with-resource-from-pool} completes, whether it returned
|
||||
normally or raised an exception. Can be overridden per-call via the
|
||||
@code{#:duration-logger} keyword argument to
|
||||
@code{call-with-resource-from-pool}. Defaults to @code{#f} (no
|
||||
logging).
|
||||
|
||||
@item #:scheduler
|
||||
The Fibers scheduler to use for the pool's internal fiber. Defaults
|
||||
to the current scheduler.
|
||||
|
|
@ -1411,20 +1374,9 @@ receive @code{&resource-pool-destroyed}."
|
|||
(timeout-handler (resource-pool-default-timeout-handler))
|
||||
(max-waiters 'default)
|
||||
(channel (resource-pool-channel pool))
|
||||
(destroy-resource-on-exception? #f)
|
||||
(delay-logger (resource-pool-delay-logger pool))
|
||||
(duration-logger (resource-pool-duration-logger pool)))
|
||||
(destroy-resource-on-exception? #f))
|
||||
"Call PROC with a resource from POOL, blocking until a resource becomes
|
||||
available. Return the resource once PROC has returned.
|
||||
|
||||
@code{#:delay-logger} is called as @code{(delay-logger seconds)} with
|
||||
the time spent waiting for a resource to become available. Defaults
|
||||
to the pool's @code{#:delay-logger} if not specified.
|
||||
|
||||
@code{#:duration-logger} is called as @code{(duration-logger seconds)}
|
||||
after PROC completes, whether it returned normally or raised an
|
||||
exception. Defaults to the pool's @code{#:duration-logger} if not
|
||||
specified."
|
||||
available. Return the resource once PROC has returned."
|
||||
|
||||
(define timeout-or-default
|
||||
(if (eq? timeout 'default)
|
||||
|
|
@ -1438,26 +1390,6 @@ specified."
|
|||
'default-max-waiters)
|
||||
max-waiters))
|
||||
|
||||
(define (delay-logger/safe seconds)
|
||||
(with-exception-handler
|
||||
;; Ignore exceptions, since this would break returning the
|
||||
;; resource
|
||||
(lambda (exn) #f)
|
||||
(lambda ()
|
||||
(delay-logger seconds))
|
||||
#:unwind? #t))
|
||||
|
||||
(define (duration-logger/safe seconds)
|
||||
(with-exception-handler
|
||||
;; Ignore exceptions, since this would break returning the
|
||||
;; resource
|
||||
(lambda (exn) #f)
|
||||
(lambda ()
|
||||
(duration-logger seconds))
|
||||
#:unwind? #t))
|
||||
|
||||
(define checkout-start-time (get-internal-real-time))
|
||||
|
||||
(unless channel
|
||||
(raise-exception
|
||||
(make-resource-pool-destroyed-error pool)))
|
||||
|
|
@ -1530,57 +1462,43 @@ specified."
|
|||
(raise-exception
|
||||
(make-resource-pool-destroyed-error pool)))
|
||||
(('success resource-id resource-value)
|
||||
(when delay-logger
|
||||
(delay-logger/safe
|
||||
(/ (- (get-internal-real-time) checkout-start-time)
|
||||
internal-time-units-per-second)))
|
||||
|
||||
(let ((proc-start-time (get-internal-real-time)))
|
||||
(call-with-values
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
;; Unwind the stack before calling put-message, as
|
||||
;; this avoids inconsistent behaviour with
|
||||
;; continuation barriers
|
||||
(when duration-logger
|
||||
(duration-logger/safe
|
||||
(/ (- (get-internal-real-time) proc-start-time)
|
||||
internal-time-units-per-second)))
|
||||
(put-message
|
||||
channel
|
||||
(list (if (or destroy-resource-on-exception?
|
||||
(resource-pool-destroy-resource-exception? exn))
|
||||
'destroy
|
||||
'return)
|
||||
resource-id))
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(let ((stack
|
||||
(match (fluid-ref %stacks)
|
||||
((stack-tag . prompt-tag)
|
||||
(make-stack #t
|
||||
0 prompt-tag
|
||||
0 (and prompt-tag 1)))
|
||||
(_
|
||||
(make-stack #t)))))
|
||||
(raise-exception
|
||||
(make-exception
|
||||
exn
|
||||
(make-knots-exception stack)))))
|
||||
(lambda ()
|
||||
(proc resource-value))))
|
||||
#:unwind? #t))
|
||||
(lambda vals
|
||||
(when duration-logger
|
||||
(duration-logger/safe
|
||||
(/ (- (get-internal-real-time) proc-start-time)
|
||||
internal-time-units-per-second)))
|
||||
(put-message channel
|
||||
`(return ,resource-id))
|
||||
(apply values vals))))))))
|
||||
(call-with-values
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
;; Unwind the stack before calling put-message, as
|
||||
;; this avoids inconsistent behaviour with
|
||||
;; continuation barriers
|
||||
(put-message
|
||||
channel
|
||||
(list (if (or destroy-resource-on-exception?
|
||||
(resource-pool-destroy-resource-exception? exn))
|
||||
'destroy
|
||||
'return)
|
||||
resource-id))
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(let ((stack
|
||||
(match (fluid-ref %stacks)
|
||||
((stack-tag . prompt-tag)
|
||||
(make-stack #t
|
||||
0 prompt-tag
|
||||
0 (and prompt-tag 1)))
|
||||
(_
|
||||
(make-stack #t)))))
|
||||
(raise-exception
|
||||
(make-exception
|
||||
exn
|
||||
(make-knots-exception stack)))))
|
||||
(lambda ()
|
||||
(proc resource-value))))
|
||||
#:unwind? #t))
|
||||
(lambda vals
|
||||
(put-message channel
|
||||
`(return ,resource-id))
|
||||
(apply values vals)))))))
|
||||
|
||||
(define-syntax-rule (with-resource-from-pool pool resource exp ...)
|
||||
"Evaluate EXP ... with RESOURCE bound to a resource checked out from
|
||||
|
|
|
|||
|
|
@ -55,8 +55,6 @@
|
|||
;; thread pools
|
||||
thread-pool-arguments-parameter
|
||||
thread-pool-default-checkout-timeout
|
||||
thread-pool-delay-logger
|
||||
thread-pool-duration-logger
|
||||
|
||||
destroy-thread-pool
|
||||
|
||||
|
|
@ -173,15 +171,12 @@ from there, or #f if that would be an empty string."
|
|||
|
||||
(define-record-type <fixed-size-thread-pool>
|
||||
(fixed-size-thread-pool channel arguments-parameter current-procedures
|
||||
default-checkout-timeout delay-logger
|
||||
duration-logger threads)
|
||||
default-checkout-timeout threads)
|
||||
fixed-size-thread-pool?
|
||||
(channel fixed-size-thread-pool-channel)
|
||||
(arguments-parameter fixed-size-thread-pool-arguments-parameter)
|
||||
(current-procedures fixed-size-thread-pool-current-procedures)
|
||||
(default-checkout-timeout fixed-size-thread-pool-default-checkout-timeout)
|
||||
(delay-logger fixed-size-thread-pool-delay-logger)
|
||||
(duration-logger fixed-size-thread-pool-duration-logger)
|
||||
(threads fixed-size-thread-pool-threads))
|
||||
(set-procedure-property!
|
||||
(macro-transformer (module-ref (current-module) 'fixed-size-thread-pool?))
|
||||
|
|
@ -212,20 +207,6 @@ from there, or #f if that would be an empty string."
|
|||
(thread-pool-resource-pool pool))
|
||||
'default-checkout-timeout)))
|
||||
|
||||
(define (thread-pool-delay-logger pool)
|
||||
"Return the delay logger for POOL, dispatching on pool type."
|
||||
(if (fixed-size-thread-pool? pool)
|
||||
(fixed-size-thread-pool-delay-logger pool)
|
||||
(resource-pool-delay-logger
|
||||
(thread-pool-resource-pool pool))))
|
||||
|
||||
(define (thread-pool-duration-logger pool)
|
||||
"Return the duration logger for POOL, dispatching on pool type."
|
||||
(if (fixed-size-thread-pool? pool)
|
||||
(fixed-size-thread-pool-duration-logger pool)
|
||||
(resource-pool-duration-logger
|
||||
(thread-pool-resource-pool pool))))
|
||||
|
||||
(define &thread-pool-timeout-error
|
||||
(make-exception-type '&thread-pool-timeout-error
|
||||
&error
|
||||
|
|
@ -296,12 +277,12 @@ Seconds to wait for a free thread slot before raising
|
|||
forever).
|
||||
|
||||
@item #:delay-logger
|
||||
Called as @code{(delay-logger seconds)} with the time spent waiting
|
||||
for a thread to become available.
|
||||
Called as @code{(delay-logger seconds proc)} with the time spent
|
||||
waiting for a thread to become available.
|
||||
|
||||
@item #:duration-logger
|
||||
Called as @code{(duration-logger seconds)} after each procedure
|
||||
completes, whether it returned normally or raised an exception.
|
||||
Called as @code{(duration-logger seconds proc)} after each procedure
|
||||
completes.
|
||||
@end table"
|
||||
(define channel
|
||||
(make-channel))
|
||||
|
|
@ -365,7 +346,15 @@ completes, whether it returned normally or raised an exception.
|
|||
(let loop ((lifetime thread-lifetime))
|
||||
(match (get-message channel)
|
||||
('destroy #f)
|
||||
((reply proc)
|
||||
((reply sent-time proc)
|
||||
(when delay-logger
|
||||
(let ((time-delay
|
||||
(- (get-internal-real-time)
|
||||
sent-time)))
|
||||
(delay-logger (/ time-delay
|
||||
internal-time-units-per-second)
|
||||
proc)))
|
||||
|
||||
(let* ((start-time (get-internal-real-time))
|
||||
(response
|
||||
(with-exception-handler
|
||||
|
|
@ -417,11 +406,11 @@ completes, whether it returned normally or raised an exception.
|
|||
(match response
|
||||
(('thread-pool-error duration _)
|
||||
(when duration-logger
|
||||
(duration-logger duration))
|
||||
(duration-logger duration proc))
|
||||
#t)
|
||||
((duration . _)
|
||||
(when duration-logger
|
||||
(duration-logger duration))
|
||||
(duration-logger duration proc))
|
||||
#f))))
|
||||
(if (and exception?
|
||||
expire-on-exception?)
|
||||
|
|
@ -477,8 +466,6 @@ completes, whether it returned normally or raised an exception.
|
|||
param
|
||||
thread-proc-vector
|
||||
default-checkout-timeout
|
||||
delay-logger
|
||||
duration-logger
|
||||
threads))
|
||||
|
||||
(define* (make-thread-pool max-size
|
||||
|
|
@ -487,8 +474,8 @@ completes, whether it returned normally or raised an exception.
|
|||
scheduler
|
||||
thread-initializer
|
||||
thread-destructor
|
||||
delay-logger
|
||||
duration-logger
|
||||
(delay-logger (lambda _ #f))
|
||||
(duration-logger (const #f))
|
||||
thread-lifetime
|
||||
(expire-on-exception? #f)
|
||||
(name "unnamed")
|
||||
|
|
@ -552,10 +539,7 @@ Maximum number of fibers that may queue waiting for a thread. Raises
|
|||
(define* (call-with-thread thread-pool
|
||||
proc
|
||||
#:key
|
||||
(delay-logger
|
||||
(thread-pool-delay-logger thread-pool))
|
||||
(duration-logger
|
||||
(thread-pool-duration-logger thread-pool))
|
||||
duration-logger
|
||||
checkout-timeout
|
||||
channel
|
||||
destroy-thread-on-exception?
|
||||
|
|
@ -581,15 +565,9 @@ When @code{#t}, destroy the thread after PROC raises an exception.
|
|||
Equivalent to per-call @code{#:expire-on-exception?}. Defaults to
|
||||
@code{#f}.
|
||||
|
||||
@item #:delay-logger
|
||||
Called as @code{(delay-logger seconds)} with the time spent waiting
|
||||
for a thread to become available. Defaults to the pool's
|
||||
@code{#:delay-logger} if not specified.
|
||||
|
||||
@item #:duration-logger
|
||||
Called as @code{(duration-logger seconds)} after PROC completes
|
||||
(whether or not it raised an exception). Defaults to the pool's
|
||||
@code{#:duration-logger} if not specified.
|
||||
(whether or not it raised an exception).
|
||||
|
||||
@item #:channel
|
||||
Override the channel used to communicate with the thread.
|
||||
|
|
@ -597,8 +575,7 @@ Override the channel used to communicate with the thread.
|
|||
(define (handle-proc fixed-size-thread-pool
|
||||
reply-channel
|
||||
start-time
|
||||
timeout
|
||||
delay-logger)
|
||||
timeout)
|
||||
(let* ((request-channel
|
||||
(or channel
|
||||
(fixed-size-thread-pool-channel
|
||||
|
|
@ -609,6 +586,7 @@ Override the channel used to communicate with the thread.
|
|||
(wrap-operation
|
||||
(put-operation request-channel
|
||||
(list reply-channel
|
||||
start-time
|
||||
proc))
|
||||
(const #t))))
|
||||
|
||||
|
|
@ -623,11 +601,6 @@ Override the channel used to communicate with the thread.
|
|||
(raise-exception
|
||||
(make-thread-pool-timeout-error)))
|
||||
|
||||
(when delay-logger
|
||||
(delay-logger
|
||||
(/ (- (get-internal-real-time) start-time)
|
||||
internal-time-units-per-second)))
|
||||
|
||||
(let ((reply (get-message reply-channel)))
|
||||
(match reply
|
||||
(('thread-pool-error duration exn)
|
||||
|
|
@ -648,8 +621,7 @@ Override the channel used to communicate with the thread.
|
|||
(handle-proc thread-pool
|
||||
reply-channel
|
||||
start-time
|
||||
checkout-timeout
|
||||
delay-logger)
|
||||
checkout-timeout)
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(if (and (resource-pool-timeout-error? exn)
|
||||
|
|
@ -670,17 +642,13 @@ Override the channel used to communicate with the thread.
|
|||
(handle-proc fixed-size-thread-pool
|
||||
reply-channel
|
||||
start-time
|
||||
remaining-time
|
||||
#f)
|
||||
remaining-time)
|
||||
(raise-exception
|
||||
(make-thread-pool-timeout-error thread-pool))))
|
||||
(handle-proc fixed-size-thread-pool
|
||||
reply-channel
|
||||
start-time
|
||||
#f
|
||||
#f)))
|
||||
#:delay-logger delay-logger
|
||||
#:duration-logger #f
|
||||
#:max-waiters max-waiters
|
||||
#:timeout checkout-timeout
|
||||
#:destroy-resource-on-exception?
|
||||
|
|
|
|||
|
|
@ -282,50 +282,4 @@
|
|||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
|
||||
;; Test delay-logger and duration-logger
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(let* ((logged-delay #f)
|
||||
(logged-duration #f)
|
||||
(resource-pool (make-fixed-size-resource-pool
|
||||
(list 1)
|
||||
#:delay-logger
|
||||
(lambda (seconds)
|
||||
(set! logged-delay seconds))
|
||||
#:duration-logger
|
||||
(lambda (seconds)
|
||||
(set! logged-duration seconds)))))
|
||||
|
||||
(call-with-resource-from-pool resource-pool
|
||||
(lambda (res)
|
||||
(sleep 0.2)))
|
||||
|
||||
(assert-true (number? logged-delay))
|
||||
(assert-true (number? logged-duration))
|
||||
(assert-true (>= logged-duration 0.1))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
|
||||
;; Test per-call duration-logger overrides pool default
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(let* ((pool-logged #f)
|
||||
(call-logged #f)
|
||||
(resource-pool (make-fixed-size-resource-pool
|
||||
(list 1)
|
||||
#:duration-logger
|
||||
(lambda (seconds)
|
||||
(set! pool-logged seconds)))))
|
||||
|
||||
(call-with-resource-from-pool resource-pool
|
||||
(lambda (res) #t)
|
||||
#:duration-logger
|
||||
(lambda (seconds)
|
||||
(set! call-logged seconds)))
|
||||
|
||||
(assert-true (not pool-logged))
|
||||
(assert-true (number? call-logged))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
|
||||
(display "resource-pool test finished successfully\n")
|
||||
|
|
|
|||
|
|
@ -172,54 +172,4 @@
|
|||
(destroy-thread-pool thread-pool)
|
||||
(assert-equal pool-size destructor-count))
|
||||
|
||||
;; Test delay-logger and duration-logger for fixed-size thread pool
|
||||
(let* ((logged-delay #f)
|
||||
(logged-duration #f)
|
||||
(thread-pool
|
||||
(make-fixed-size-thread-pool
|
||||
1
|
||||
#:delay-logger
|
||||
(lambda (seconds)
|
||||
(set! logged-delay seconds))
|
||||
#:duration-logger
|
||||
(lambda (seconds)
|
||||
(set! logged-duration seconds)))))
|
||||
|
||||
(call-with-thread
|
||||
thread-pool
|
||||
(lambda ()
|
||||
(usleep 100000)))
|
||||
|
||||
(assert-true (number? logged-delay))
|
||||
(assert-true (number? logged-duration))
|
||||
(assert-true (>= logged-duration 0.1))
|
||||
|
||||
(destroy-thread-pool thread-pool))
|
||||
|
||||
;; Test delay-logger and duration-logger for dynamic thread pool
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(let* ((logged-delay #f)
|
||||
(logged-duration #f)
|
||||
(thread-pool
|
||||
(make-thread-pool
|
||||
1
|
||||
#:delay-logger
|
||||
(lambda (seconds)
|
||||
(set! logged-delay seconds))
|
||||
#:duration-logger
|
||||
(lambda (seconds)
|
||||
(set! logged-duration seconds)))))
|
||||
|
||||
(call-with-thread
|
||||
thread-pool
|
||||
(lambda ()
|
||||
(usleep 100000)))
|
||||
|
||||
(assert-true (number? logged-delay))
|
||||
(assert-true (number? logged-duration))
|
||||
(assert-true (>= logged-duration 0.1))
|
||||
|
||||
(destroy-thread-pool thread-pool))))
|
||||
|
||||
(display "thread-pool test finished successfully\n")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue