Compare commits
5 commits
trunk
...
actions-te
| Author | SHA1 | Date | |
|---|---|---|---|
| 003c5aa6b0 | |||
| eadfa53b36 | |||
| 81dd3370e6 | |||
| 7f5f05ef2b | |||
| 7c2c6f2de9 |
9 changed files with 313 additions and 917 deletions
|
|
@ -1,7 +1,7 @@
|
|||
on:
|
||||
push:
|
||||
branches:
|
||||
- trunk
|
||||
- actions-test
|
||||
jobs:
|
||||
test:
|
||||
runs-on: host
|
||||
|
|
@ -10,17 +10,13 @@ jobs:
|
|||
- run: git clone --depth=1 https://$FORGEJO_TOKEN@forge.cbaines.net/cbaines/guile-knots.git --branch=pages knots-pages
|
||||
- run: |
|
||||
cd knots-trunk
|
||||
guix shell -D -f guix-dev.scm -- documenta api "knots.scm knots/"
|
||||
guix shell -D -f guix-dev.scm -- documenta api knots
|
||||
guix shell texinfo -- makeinfo --css-ref=https://luis-felipe.gitlab.io/texinfo-css/static/css/texinfo-7.css --no-split --html -c SHOW_TITLE=true -o ../knots-pages/index.html doc/index.texi
|
||||
|
||||
- run: |
|
||||
cd knots-pages
|
||||
git add .
|
||||
if [[ -z "$(git status -s)" ]]; then
|
||||
echo "Nothing to push"
|
||||
else
|
||||
git config user.email ""
|
||||
git config user.name "Automatic website updater"
|
||||
git commit -m "Automatic website update"
|
||||
git push
|
||||
fi
|
||||
|
|
@ -20,9 +20,6 @@
|
|||
(define-module (knots parallelism)
|
||||
#:use-module (srfi srfi-1)
|
||||
#:use-module (srfi srfi-71)
|
||||
#:use-module (srfi srfi-9)
|
||||
#:use-module (srfi srfi-9 gnu)
|
||||
#:use-module (srfi srfi-43)
|
||||
#:use-module (ice-9 match)
|
||||
#:use-module (ice-9 control)
|
||||
#:use-module (ice-9 exceptions)
|
||||
|
|
@ -30,7 +27,6 @@
|
|||
#:use-module (fibers channels)
|
||||
#:use-module (fibers operations)
|
||||
#:use-module (knots)
|
||||
#:use-module (knots resource-pool)
|
||||
#:export (fibers-batch-map
|
||||
fibers-map
|
||||
|
||||
|
|
@ -42,13 +38,7 @@
|
|||
fibers-parallel
|
||||
fibers-let
|
||||
|
||||
fiberize
|
||||
|
||||
make-parallelism-limiter
|
||||
parallelism-limiter?
|
||||
destroy-parallelism-limiter
|
||||
call-with-parallelism-limiter
|
||||
with-parallelism-limiter))
|
||||
fiberize))
|
||||
|
||||
(define (defer-to-parallel-fiber thunk)
|
||||
(let ((reply (make-channel)))
|
||||
|
|
@ -58,7 +48,7 @@
|
|||
(lambda (exn)
|
||||
(put-message
|
||||
reply
|
||||
(cons 'exception exn)))
|
||||
(list 'exception exn)))
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
|
|
@ -79,7 +69,7 @@
|
|||
(lambda ()
|
||||
(start-stack #t (thunk)))
|
||||
(lambda vals
|
||||
(put-message reply (cons 'result vals)))))))
|
||||
(put-message reply vals))))))
|
||||
#:unwind? #t))
|
||||
#:parallel? #t)
|
||||
reply))
|
||||
|
|
@ -89,16 +79,13 @@
|
|||
reply-channels)))
|
||||
(map
|
||||
(match-lambda
|
||||
(('exception . exn)
|
||||
(('exception exn)
|
||||
(raise-exception exn))
|
||||
(('result . vals)
|
||||
(apply values vals)))
|
||||
(result
|
||||
(apply values result)))
|
||||
responses)))
|
||||
|
||||
(define (fibers-batch-map proc parallelism-limit . lists)
|
||||
"Map PROC over LISTS in parallel, with a PARALLELISM-LIMIT. If any of
|
||||
the invocations of PROC raise an exception, this will be raised once
|
||||
all of the calls to PROC have finished."
|
||||
(define vecs (map (lambda (list-or-vec)
|
||||
(if (vector? list-or-vec)
|
||||
list-or-vec
|
||||
|
|
@ -118,18 +105,9 @@ all of the calls to PROC have finished."
|
|||
(channel-indexes '()))
|
||||
(if (and (eq? #f next-to-process-index)
|
||||
(null? channel-indexes))
|
||||
(let ((processed-result-vec
|
||||
(vector-map
|
||||
(lambda (_ result-or-exn)
|
||||
(match result-or-exn
|
||||
(('exception . exn)
|
||||
(raise-exception exn))
|
||||
(('result . vals)
|
||||
(car vals))))
|
||||
result-vec)))
|
||||
(if (vector? (first lists))
|
||||
processed-result-vec
|
||||
(vector->list processed-result-vec)))
|
||||
result-vec
|
||||
(vector->list result-vec))
|
||||
|
||||
(if (or (= (length channel-indexes)
|
||||
(min parallelism-limit vecs-length))
|
||||
|
|
@ -145,13 +123,18 @@ all of the calls to PROC have finished."
|
|||
(get-operation
|
||||
(vector-ref result-vec index))
|
||||
(lambda (result)
|
||||
(match result
|
||||
(('exception exn)
|
||||
(raise-exception exn))
|
||||
(_
|
||||
(vector-set! result-vec
|
||||
index
|
||||
result)
|
||||
(first result))
|
||||
|
||||
(values next-to-process-index
|
||||
(lset-difference =
|
||||
channel-indexes
|
||||
(list index))))))
|
||||
(list index))))))))
|
||||
channel-indexes)))))
|
||||
(loop new-index
|
||||
new-channel-indexes))
|
||||
|
|
@ -174,14 +157,9 @@ all of the calls to PROC have finished."
|
|||
channel-indexes)))))))
|
||||
|
||||
(define (fibers-map proc . lists)
|
||||
"Map PROC over LISTS in parallel, running up to 20 fibers in
|
||||
PARALLEL. If any of the invocations of PROC raise an exception, this
|
||||
will be raised once all of the calls to PROC have finished."
|
||||
(apply fibers-batch-map proc 20 lists))
|
||||
|
||||
(define (fibers-batch-for-each proc parallelism-limit . lists)
|
||||
"Call PROC on LISTS, running up to PARALLELISM-LIMIT fibers in
|
||||
parallel."
|
||||
(apply fibers-batch-map
|
||||
(lambda args
|
||||
(apply proc args)
|
||||
|
|
@ -192,13 +170,10 @@ parallel."
|
|||
*unspecified*)
|
||||
|
||||
(define (fibers-for-each proc . lists)
|
||||
"Call PROC on LISTS, running up to 20 fibers in parallel."
|
||||
(apply fibers-batch-for-each proc 20 lists))
|
||||
|
||||
(define-syntax fibers-parallel
|
||||
(lambda (x)
|
||||
"Run each expression in parallel. If any expression raises an
|
||||
exception, this will be raised after all exceptions have finished."
|
||||
(syntax-case x ()
|
||||
((_ e0 ...)
|
||||
(with-syntax (((tmp0 ...) (generate-temporaries (syntax (e0 ...)))))
|
||||
|
|
@ -209,16 +184,12 @@ parallel."
|
|||
(apply values (fetch-result-of-defered-thunks tmp0 ...))))))))
|
||||
|
||||
(define-syntax-rule (fibers-let ((v e) ...) b0 b1 ...)
|
||||
"Let, but run each binding in a fiber in parallel."
|
||||
(call-with-values
|
||||
(lambda () (fibers-parallel e ...))
|
||||
(lambda (v ...)
|
||||
b0 b1 ...)))
|
||||
|
||||
(define* (fibers-map-with-progress proc lists #:key report)
|
||||
"Map PROC over LISTS, calling #:REPORT if specified after each
|
||||
invocation of PROC finishes. REPORT is passed the results for each
|
||||
element of LISTS, or #f if no result has been received yet."
|
||||
(let loop ((channels-to-results
|
||||
(apply map
|
||||
(lambda args
|
||||
|
|
@ -239,8 +210,8 @@ invocation of PROC finishes. REPORT is passed the results for each
|
|||
(match-lambda
|
||||
((#f . ('exception . exn))
|
||||
(raise-exception exn))
|
||||
((#f . ('result . vals))
|
||||
(car vals)))
|
||||
((#f . ('result . val))
|
||||
val))
|
||||
channels-to-results)
|
||||
(loop
|
||||
(perform-operation
|
||||
|
|
@ -257,7 +228,12 @@ invocation of PROC finishes. REPORT is passed the results for each
|
|||
(map (match-lambda
|
||||
((c . r)
|
||||
(if (eq? channel c)
|
||||
(cons #f result)
|
||||
(cons #f
|
||||
(match result
|
||||
(('exception . exn)
|
||||
result)
|
||||
(_
|
||||
(cons 'result result))))
|
||||
(cons c r))))
|
||||
channels-to-results)))
|
||||
#f))))
|
||||
|
|
@ -278,7 +254,7 @@ invocation of PROC finishes. REPORT is passed the results for each
|
|||
reply-channel
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(cons 'exception exn))
|
||||
(list 'exception exn))
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
|
|
@ -309,32 +285,5 @@ invocation of PROC finishes. REPORT is passed the results for each
|
|||
(put-message input-channel (cons reply-channel args))
|
||||
(match (get-message reply-channel)
|
||||
(('result . vals) (apply values vals))
|
||||
(('exception . exn)
|
||||
(('exception exn)
|
||||
(raise-exception exn))))))
|
||||
|
||||
(define-record-type <parallelism-limiter>
|
||||
(make-parallelism-limiter-record resource-pool)
|
||||
parallelism-limiter?
|
||||
(resource-pool parallelism-limiter-resource-pool))
|
||||
|
||||
(define* (make-parallelism-limiter limit #:key (name "unnamed"))
|
||||
(make-parallelism-limiter-record
|
||||
(make-fixed-size-resource-pool
|
||||
(iota limit)
|
||||
#:name name)))
|
||||
|
||||
(define (destroy-parallelism-limiter parallelism-limiter)
|
||||
(destroy-resource-pool
|
||||
(parallelism-limiter-resource-pool
|
||||
parallelism-limiter)))
|
||||
|
||||
(define* (call-with-parallelism-limiter parallelism-limiter thunk)
|
||||
(call-with-resource-from-pool
|
||||
(parallelism-limiter-resource-pool parallelism-limiter)
|
||||
(lambda _
|
||||
(thunk))))
|
||||
|
||||
(define-syntax-rule (with-parallelism-limiter parallelism-limiter exp ...)
|
||||
(call-with-parallelism-limiter
|
||||
parallelism-limiter
|
||||
(lambda () exp ...)))
|
||||
|
|
|
|||
|
|
@ -22,7 +22,6 @@
|
|||
#:use-module (srfi srfi-9)
|
||||
#:use-module (srfi srfi-9 gnu)
|
||||
#:use-module (srfi srfi-71)
|
||||
#:use-module (ice-9 q)
|
||||
#:use-module (ice-9 match)
|
||||
#:use-module (ice-9 exceptions)
|
||||
#:use-module (fibers)
|
||||
|
|
@ -33,10 +32,9 @@
|
|||
#:use-module (fibers conditions)
|
||||
#:use-module (knots)
|
||||
#:use-module (knots parallelism)
|
||||
#:export (make-fixed-size-resource-pool
|
||||
make-resource-pool
|
||||
#:export (resource-pool?
|
||||
|
||||
resource-pool?
|
||||
make-resource-pool
|
||||
resource-pool-name
|
||||
resource-pool-channel
|
||||
resource-pool-configuration
|
||||
|
|
@ -75,7 +73,7 @@
|
|||
(record-constructor &resource-pool-abort-add-resource))
|
||||
|
||||
(define resource-pool-abort-add-resource-error?
|
||||
(exception-predicate &resource-pool-abort-add-resource))
|
||||
(record-predicate &resource-pool-abort-add-resource))
|
||||
|
||||
(define-record-type <resource-pool>
|
||||
(make-resource-pool-record name channel destroy-condition configuration)
|
||||
|
|
@ -93,429 +91,6 @@
|
|||
(resource-pool-name resource-pool))
|
||||
port)))
|
||||
|
||||
(define (remove-at-index! lst i)
|
||||
(let ((start
|
||||
end
|
||||
(split-at! lst i)))
|
||||
(append
|
||||
start
|
||||
(cdr end))))
|
||||
|
||||
(define* (make-fixed-size-resource-pool resources
|
||||
#:key
|
||||
(delay-logger (const #f))
|
||||
(duration-logger (const #f))
|
||||
destructor
|
||||
scheduler
|
||||
(name "unnamed")
|
||||
default-checkout-timeout
|
||||
default-max-waiters)
|
||||
(define channel (make-channel))
|
||||
(define destroy-condition
|
||||
(make-condition))
|
||||
|
||||
(define pool
|
||||
(make-resource-pool-record
|
||||
name
|
||||
channel
|
||||
destroy-condition
|
||||
`((delay-logger . ,delay-logger)
|
||||
(duration-logger . ,duration-logger)
|
||||
(destructor . ,destructor)
|
||||
(scheduler . ,scheduler)
|
||||
(name . ,name)
|
||||
(default-checkout-timeout . ,default-checkout-timeout)
|
||||
(default-max-waiters . ,default-max-waiters))))
|
||||
|
||||
(define checkout-failure-count 0)
|
||||
|
||||
(define (spawn-fiber-to-destroy-resource resource)
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let loop ()
|
||||
(let ((success?
|
||||
(with-exception-handler
|
||||
(lambda _ #f)
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(simple-format
|
||||
(current-error-port)
|
||||
"exception running resource pool destructor (~A): ~A\n"
|
||||
name
|
||||
destructor)
|
||||
(print-backtrace-and-exception/knots exn)
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(start-stack #t (destructor resource))
|
||||
#t)))
|
||||
#:unwind? #t)))
|
||||
|
||||
(if success?
|
||||
(put-message channel
|
||||
(list 'remove resource))
|
||||
(begin
|
||||
(sleep 5)
|
||||
|
||||
(loop))))))))
|
||||
|
||||
(define (spawn-fiber-for-checkout reply-channel
|
||||
reply-timeout
|
||||
resource)
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((checkout-success?
|
||||
(perform-operation
|
||||
(choice-operation
|
||||
(wrap-operation
|
||||
(put-operation reply-channel
|
||||
(cons 'success resource))
|
||||
(const #t))
|
||||
(wrap-operation (sleep-operation
|
||||
reply-timeout)
|
||||
(const #f))))))
|
||||
(unless checkout-success?
|
||||
(put-message
|
||||
channel
|
||||
(list 'return-failed-checkout resource)))))))
|
||||
|
||||
(define (destroy-loop resources)
|
||||
(let loop ((resources resources))
|
||||
(match (get-message channel)
|
||||
(('checkout reply timeout-time max-waiters)
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation
|
||||
reply
|
||||
(cons 'resource-pool-destroyed
|
||||
#f))))
|
||||
(perform-operation
|
||||
(if timeout-time
|
||||
(choice-operation
|
||||
op
|
||||
(wrap-operation
|
||||
(sleep-operation
|
||||
(/ (- timeout-time
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second))
|
||||
(const #f)))
|
||||
op)))))
|
||||
(loop resources))
|
||||
(((and (or 'return
|
||||
'return-failed-checkout
|
||||
'remove)
|
||||
return-type)
|
||||
resource)
|
||||
(when (and (not (eq? return-type 'remove))
|
||||
destructor)
|
||||
(spawn-fiber-to-destroy-resource resource))
|
||||
|
||||
(let ((index
|
||||
(list-index (lambda (x)
|
||||
(eq? x resource))
|
||||
resources)))
|
||||
(let ((new-resources
|
||||
(if index
|
||||
(remove-at-index! resources index)
|
||||
(begin
|
||||
(simple-format
|
||||
(current-error-port)
|
||||
"resource pool error: unable to remove ~A\n"
|
||||
resource)
|
||||
resources))))
|
||||
(if (null? new-resources)
|
||||
(begin
|
||||
(signal-condition! destroy-condition)
|
||||
|
||||
;; No loop
|
||||
*unspecified*)
|
||||
(loop new-resources)))))
|
||||
|
||||
(('stats reply timeout-time)
|
||||
(let ((stats
|
||||
`((resources . ,(length resources))
|
||||
(available . 0)
|
||||
(waiters . 0)
|
||||
(checkout-failure-count . ,checkout-failure-count))))
|
||||
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation reply stats)))
|
||||
(perform-operation
|
||||
(if timeout-time
|
||||
(choice-operation
|
||||
op
|
||||
(sleep-operation
|
||||
(/ (- timeout-time
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second)))
|
||||
op))))))
|
||||
|
||||
(loop resources))
|
||||
|
||||
(('destroy reply)
|
||||
(loop resources))
|
||||
(unknown
|
||||
(simple-format
|
||||
(current-error-port)
|
||||
"unrecognised message to ~A resource pool channel: ~A\n"
|
||||
name
|
||||
unknown)
|
||||
(loop resources)))))
|
||||
|
||||
(define (main-loop)
|
||||
(let loop ((resources resources)
|
||||
(available resources)
|
||||
(waiters (make-q)))
|
||||
|
||||
(match (get-message channel)
|
||||
(('checkout reply timeout-time max-waiters)
|
||||
(if (null? available)
|
||||
(let ((waiters-count
|
||||
(q-length waiters)))
|
||||
(if (and max-waiters
|
||||
(>= waiters-count
|
||||
max-waiters))
|
||||
(begin
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation
|
||||
reply
|
||||
(cons 'too-many-waiters
|
||||
waiters-count))))
|
||||
(perform-operation
|
||||
(if timeout-time
|
||||
(choice-operation
|
||||
op
|
||||
(wrap-operation
|
||||
(sleep-operation
|
||||
(/ (- timeout-time
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second))
|
||||
(const #f)))
|
||||
op)))))
|
||||
(loop resources
|
||||
available
|
||||
waiters))
|
||||
(loop resources
|
||||
available
|
||||
(enq! waiters (cons reply timeout-time)))))
|
||||
|
||||
(if timeout-time
|
||||
(let ((current-internal-time
|
||||
(get-internal-real-time)))
|
||||
;; If this client is still waiting
|
||||
(if (> timeout-time
|
||||
current-internal-time)
|
||||
(let ((reply-timeout
|
||||
(/ (- timeout-time
|
||||
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 reply
|
||||
reply-timeout
|
||||
(car available))
|
||||
(loop resources
|
||||
(cdr available)
|
||||
waiters))
|
||||
(loop resources
|
||||
available
|
||||
waiters)))
|
||||
(begin
|
||||
(put-message reply (cons 'success
|
||||
(car available)))
|
||||
|
||||
(loop resources
|
||||
(cdr available)
|
||||
waiters)))))
|
||||
|
||||
(((and (or 'return
|
||||
'return-failed-checkout)
|
||||
return-type)
|
||||
resource)
|
||||
|
||||
(when (eq? 'return-failed-checkout
|
||||
return-type)
|
||||
(set! checkout-failure-count
|
||||
(+ 1 checkout-failure-count)))
|
||||
|
||||
(if (q-empty? waiters)
|
||||
(loop resources
|
||||
(cons resource available)
|
||||
waiters)
|
||||
|
||||
(let ((current-internal-time
|
||||
(get-internal-real-time)))
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(if (eq? (exception-kind exn) 'q-empty)
|
||||
(loop resources
|
||||
(cons resource available)
|
||||
waiters)
|
||||
(raise-exception exn)))
|
||||
(lambda ()
|
||||
(let waiter-loop ((waiter (deq! waiters)))
|
||||
(match waiter
|
||||
((reply . timeout)
|
||||
(if (and timeout
|
||||
(< timeout current-internal-time))
|
||||
(waiter-loop (deq! waiters))
|
||||
(if timeout
|
||||
(let ((reply-timeout
|
||||
(/ (- 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 reply
|
||||
reply-timeout
|
||||
resource))
|
||||
(put-message reply (cons 'success
|
||||
resource))))))))
|
||||
#:unwind? #t)
|
||||
(loop resources
|
||||
available
|
||||
waiters))))
|
||||
|
||||
(('list-resources reply)
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(put-message reply (list-copy resources))))
|
||||
|
||||
(loop resources
|
||||
available
|
||||
waiters))
|
||||
|
||||
(('stats reply timeout-time)
|
||||
(let ((stats
|
||||
`((resources . ,(length resources))
|
||||
(available . ,(length available))
|
||||
(waiters . ,(q-length waiters))
|
||||
(checkout-failure-count . ,checkout-failure-count))))
|
||||
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation reply stats)))
|
||||
(perform-operation
|
||||
(if timeout-time
|
||||
(choice-operation
|
||||
op
|
||||
(sleep-operation
|
||||
(/ (- timeout-time
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second)))
|
||||
op))))))
|
||||
|
||||
(loop resources
|
||||
available
|
||||
waiters))
|
||||
|
||||
(('destroy)
|
||||
(if (and (null? resources)
|
||||
(q-empty? waiters))
|
||||
(signal-condition!
|
||||
destroy-condition)
|
||||
|
||||
(let ((current-internal-time (get-internal-real-time)))
|
||||
(for-each
|
||||
(match-lambda
|
||||
((reply . timeout)
|
||||
(when (or (not timeout)
|
||||
(> timeout current-internal-time))
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation
|
||||
reply
|
||||
(cons 'resource-pool-destroyed
|
||||
#f))))
|
||||
(perform-operation
|
||||
(if timeout
|
||||
(choice-operation
|
||||
op
|
||||
(wrap-operation
|
||||
(sleep-operation
|
||||
(/ (- timeout
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second))
|
||||
(const #f)))
|
||||
op))))))))
|
||||
(car waiters))
|
||||
|
||||
(if destructor
|
||||
(begin
|
||||
(for-each
|
||||
(lambda (resource)
|
||||
(spawn-fiber-to-destroy-resource resource))
|
||||
available)
|
||||
(destroy-loop resources))
|
||||
(let dl ((resources resources)
|
||||
(available available))
|
||||
(if (null? available)
|
||||
(if (null? resources)
|
||||
(signal-condition!
|
||||
destroy-condition)
|
||||
(destroy-loop resources))
|
||||
(let ((index
|
||||
(list-index (lambda (x)
|
||||
(eq? x (car available)))
|
||||
resources)))
|
||||
(dl (remove-at-index! resources index)
|
||||
(cdr available)))))))))
|
||||
|
||||
(unknown
|
||||
(simple-format
|
||||
(current-error-port)
|
||||
"unrecognised message to ~A resource pool channel: ~A\n"
|
||||
name
|
||||
unknown)
|
||||
(loop resources
|
||||
available
|
||||
waiters)))))
|
||||
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
#f)
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(let* ((stack (make-stack #t))
|
||||
(error-string
|
||||
(call-with-output-string
|
||||
(lambda (port)
|
||||
(display-backtrace stack port 3)
|
||||
(simple-format
|
||||
port
|
||||
"exception in the ~A pool fiber, " name)
|
||||
(print-exception
|
||||
port
|
||||
(stack-ref stack 3)
|
||||
'%exception
|
||||
(list exn))))))
|
||||
(display error-string
|
||||
(current-error-port)))
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(start-stack
|
||||
#t
|
||||
(main-loop)))))
|
||||
#:unwind? #t))
|
||||
(or scheduler
|
||||
(current-scheduler)))
|
||||
|
||||
pool)
|
||||
|
||||
(define* (make-resource-pool return-new-resource max-size
|
||||
#:key (min-size 0)
|
||||
(idle-seconds #f)
|
||||
|
|
@ -551,33 +126,28 @@
|
|||
|
||||
(define checkout-failure-count 0)
|
||||
|
||||
(define return-new-resource/parallelism-limiter
|
||||
(make-parallelism-limiter
|
||||
(or add-resources-parallelism
|
||||
max-size)
|
||||
#:name
|
||||
(string-append
|
||||
name
|
||||
" resource pool new resource parallelism limiter")))
|
||||
|
||||
(define (spawn-fiber-to-return-new-resource)
|
||||
(spawn-fiber
|
||||
(define spawn-fiber-to-return-new-resource
|
||||
(if add-resources-parallelism
|
||||
(let ((thunk
|
||||
(fiberize
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
;; This can happen if the resource pool is destroyed very
|
||||
;; quickly
|
||||
(unless (resource-pool-destroyed-error? exn)
|
||||
(raise-exception exn)))
|
||||
(lambda ()
|
||||
(with-parallelism-limiter
|
||||
return-new-resource/parallelism-limiter
|
||||
(let ((max-size
|
||||
(assq-ref (resource-pool-configuration pool)
|
||||
'max-size))
|
||||
(size (assq-ref (resource-pool-stats pool #:timeout #f)
|
||||
(size (assq-ref (resource-pool-stats pool)
|
||||
'resources)))
|
||||
(unless (= size max-size)
|
||||
(let ((new-resource
|
||||
(return-new-resource)))
|
||||
(put-message channel
|
||||
(list 'add-resource new-resource))))))
|
||||
#:parallelism add-resources-parallelism)))
|
||||
(lambda ()
|
||||
(spawn-fiber thunk)))
|
||||
(lambda ()
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((new-resource
|
||||
(with-exception-handler
|
||||
(lambda _ #f)
|
||||
(lambda ()
|
||||
|
|
@ -591,12 +161,11 @@
|
|||
(print-backtrace-and-exception/knots exn)
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(let ((new-resource
|
||||
(start-stack #t (return-new-resource))))
|
||||
(start-stack #t (return-new-resource)))))
|
||||
#:unwind? #t)))
|
||||
(when new-resource
|
||||
(put-message channel
|
||||
(list 'add-resource new-resource))))))
|
||||
#:unwind? #t)))))
|
||||
#:unwind? #t))))
|
||||
(list 'add-resource new-resource)))))))))
|
||||
|
||||
(define (spawn-fiber-to-destroy-resource resource)
|
||||
(spawn-fiber
|
||||
|
|
@ -681,14 +250,21 @@
|
|||
'remove)
|
||||
return-type)
|
||||
resource)
|
||||
(when (and (not (eq? return-type 'remove))
|
||||
destructor)
|
||||
(when destructor
|
||||
(spawn-fiber-to-destroy-resource resource))
|
||||
|
||||
(let ((index
|
||||
(list-index (lambda (x)
|
||||
(eq? x resource))
|
||||
resources)))
|
||||
(define (remove-at-index! lst i)
|
||||
(let ((start
|
||||
end
|
||||
(split-at! lst i)))
|
||||
(append
|
||||
start
|
||||
(cdr end))))
|
||||
|
||||
(let ((new-resources
|
||||
(if index
|
||||
(remove-at-index! resources index)
|
||||
|
|
@ -700,16 +276,13 @@
|
|||
resources))))
|
||||
(if (null? new-resources)
|
||||
(begin
|
||||
(and=> return-new-resource/parallelism-limiter
|
||||
destroy-parallelism-limiter)
|
||||
|
||||
(signal-condition! destroy-condition)
|
||||
|
||||
;; No loop
|
||||
*unspecified*)
|
||||
(loop new-resources)))))
|
||||
|
||||
(('stats reply timeout-time)
|
||||
(('stats reply)
|
||||
(let ((stats
|
||||
`((resources . ,(length resources))
|
||||
(available . 0)
|
||||
|
|
@ -718,17 +291,13 @@
|
|||
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation reply stats)))
|
||||
(perform-operation
|
||||
(if timeout-time
|
||||
(choice-operation
|
||||
op
|
||||
(sleep-operation
|
||||
(/ (- timeout-time
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second)))
|
||||
op))))))
|
||||
(wrap-operation
|
||||
(put-operation reply stats)
|
||||
(const #t))
|
||||
(wrap-operation (sleep-operation 5)
|
||||
(const #f)))))))
|
||||
|
||||
(loop resources))
|
||||
|
||||
|
|
@ -748,7 +317,7 @@
|
|||
(define (main-loop)
|
||||
(let loop ((resources '())
|
||||
(available '())
|
||||
(waiters (make-q))
|
||||
(waiters '())
|
||||
(resources-last-used '()))
|
||||
|
||||
(match (get-message channel)
|
||||
|
|
@ -770,51 +339,50 @@
|
|||
(cons (get-internal-real-time)
|
||||
resources-last-used))))
|
||||
|
||||
(if (q-empty? waiters)
|
||||
(if (null? waiters)
|
||||
(loop (cons resource resources)
|
||||
(cons resource available)
|
||||
waiters
|
||||
(cons (get-internal-real-time)
|
||||
resources-last-used))
|
||||
|
||||
(let ((current-internal-time
|
||||
(get-internal-real-time)))
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(if (eq? (exception-kind exn) 'q-empty)
|
||||
(let* ((current-internal-time (get-internal-real-time))
|
||||
(alive-waiters
|
||||
dead-waiters
|
||||
(partition!
|
||||
(match-lambda
|
||||
((reply . timeout)
|
||||
(or (not timeout)
|
||||
(> timeout current-internal-time))))
|
||||
waiters)))
|
||||
(if (null? alive-waiters)
|
||||
(loop (cons resource resources)
|
||||
(cons resource available)
|
||||
waiters
|
||||
(cons current-internal-time
|
||||
'()
|
||||
(cons (get-internal-real-time)
|
||||
resources-last-used))
|
||||
(raise-exception exn)))
|
||||
(lambda ()
|
||||
(let waiter-loop ((waiter (deq! waiters)))
|
||||
(match waiter
|
||||
((reply . timeout)
|
||||
(if (and timeout
|
||||
(< timeout current-internal-time))
|
||||
(waiter-loop (deq! waiters))
|
||||
(if timeout
|
||||
(match (last alive-waiters)
|
||||
((waiter-channel . waiter-timeout)
|
||||
(if waiter-timeout
|
||||
(let ((reply-timeout
|
||||
(/ (- 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 reply
|
||||
;; 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 reply (cons 'success
|
||||
resource))))))))
|
||||
#:unwind? #t)
|
||||
(put-message waiter-channel (cons 'success
|
||||
resource)))
|
||||
|
||||
(loop (cons resource resources)
|
||||
available
|
||||
waiters
|
||||
(cons current-internal-time
|
||||
resources-last-used))))))
|
||||
(drop-right! alive-waiters 1)
|
||||
(cons (get-internal-real-time)
|
||||
resources-last-used)))))))))
|
||||
|
||||
(('checkout reply timeout-time max-waiters)
|
||||
(if (null? available)
|
||||
|
|
@ -823,7 +391,7 @@
|
|||
(spawn-fiber-to-return-new-resource))
|
||||
|
||||
(let ((waiters-count
|
||||
(q-length waiters)))
|
||||
(length waiters)))
|
||||
(if (and max-waiters
|
||||
(>= waiters-count
|
||||
max-waiters))
|
||||
|
|
@ -852,7 +420,8 @@
|
|||
resources-last-used))
|
||||
(loop resources
|
||||
available
|
||||
(enq! waiters (cons reply timeout-time))
|
||||
(cons (cons reply timeout-time)
|
||||
waiters)
|
||||
resources-last-used))))
|
||||
|
||||
(if timeout-time
|
||||
|
|
@ -899,7 +468,7 @@
|
|||
(set! checkout-failure-count
|
||||
(+ 1 checkout-failure-count)))
|
||||
|
||||
(if (q-empty? waiters)
|
||||
(if (null? waiters)
|
||||
(loop resources
|
||||
(cons resource available)
|
||||
waiters
|
||||
|
|
@ -912,14 +481,19 @@
|
|||
(get-internal-real-time))
|
||||
resources-last-used))
|
||||
|
||||
(let ((current-internal-time
|
||||
(get-internal-real-time)))
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(if (eq? (exception-kind exn) 'q-empty)
|
||||
(let* ((current-internal-time (get-internal-real-time))
|
||||
(alive-waiters
|
||||
dead-waiters
|
||||
(partition!
|
||||
(match-lambda
|
||||
((reply . timeout)
|
||||
(or (not timeout)
|
||||
(> timeout current-internal-time))))
|
||||
waiters)))
|
||||
(if (null? alive-waiters)
|
||||
(loop resources
|
||||
(cons resource available)
|
||||
waiters
|
||||
'()
|
||||
(begin
|
||||
(when (eq? return-type 'return)
|
||||
(list-set!
|
||||
|
|
@ -927,48 +501,50 @@
|
|||
(list-index (lambda (x)
|
||||
(eq? x resource))
|
||||
resources)
|
||||
current-internal-time))
|
||||
(get-internal-real-time)))
|
||||
resources-last-used))
|
||||
(raise-exception exn)))
|
||||
(lambda ()
|
||||
(let waiter-loop ((waiter (deq! waiters)))
|
||||
(match waiter
|
||||
((reply . timeout)
|
||||
(if (and timeout
|
||||
(< timeout current-internal-time))
|
||||
(waiter-loop (deq! waiters))
|
||||
(if timeout
|
||||
(match (last alive-waiters)
|
||||
((waiter-channel . waiter-timeout)
|
||||
(if waiter-timeout
|
||||
(let ((reply-timeout
|
||||
(/ (- 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 reply
|
||||
;; resource, and returning it if there's a
|
||||
;; timeout
|
||||
(spawn-fiber-for-checkout waiter-channel
|
||||
reply-timeout
|
||||
resource))
|
||||
(put-message reply (cons 'success
|
||||
resource))))))))
|
||||
#:unwind? #t)
|
||||
(put-message waiter-channel (cons 'success
|
||||
resource)))
|
||||
|
||||
(loop resources
|
||||
available
|
||||
waiters
|
||||
(drop-right! alive-waiters 1)
|
||||
(begin
|
||||
(list-set!
|
||||
resources-last-used
|
||||
(list-index (lambda (x)
|
||||
(eq? x resource))
|
||||
resources)
|
||||
current-internal-time)
|
||||
resources-last-used)))))
|
||||
(get-internal-real-time))
|
||||
resources-last-used))))))))
|
||||
|
||||
(('remove resource)
|
||||
(let ((index
|
||||
(list-index (lambda (x)
|
||||
(eq? x resource))
|
||||
resources)))
|
||||
(define (remove-at-index! lst i)
|
||||
(let ((start
|
||||
end
|
||||
(split-at! lst i)))
|
||||
(append
|
||||
start
|
||||
(cdr end))))
|
||||
|
||||
(loop (if index
|
||||
(remove-at-index! resources index)
|
||||
(begin
|
||||
|
|
@ -1001,26 +577,22 @@
|
|||
waiters
|
||||
resources-last-used))
|
||||
|
||||
(('stats reply timeout-time)
|
||||
(('stats reply)
|
||||
(let ((stats
|
||||
`((resources . ,(length resources))
|
||||
(available . ,(length available))
|
||||
(waiters . ,(q-length waiters))
|
||||
(waiters . ,(length waiters))
|
||||
(checkout-failure-count . ,checkout-failure-count))))
|
||||
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(let ((op
|
||||
(put-operation reply stats)))
|
||||
(perform-operation
|
||||
(if timeout-time
|
||||
(choice-operation
|
||||
op
|
||||
(sleep-operation
|
||||
(/ (- timeout-time
|
||||
(get-internal-real-time))
|
||||
internal-time-units-per-second)))
|
||||
op))))))
|
||||
(wrap-operation
|
||||
(put-operation reply stats)
|
||||
(const #t))
|
||||
(wrap-operation (sleep-operation 5)
|
||||
(const #f)))))))
|
||||
|
||||
(loop resources
|
||||
available
|
||||
|
|
@ -1069,10 +641,22 @@
|
|||
|
||||
(('destroy)
|
||||
(if (and (null? resources)
|
||||
(q-empty? waiters))
|
||||
(null? waiters))
|
||||
(signal-condition!
|
||||
destroy-condition)
|
||||
|
||||
(begin
|
||||
(for-each
|
||||
(lambda (resource)
|
||||
(if destructor
|
||||
(spawn-fiber-to-destroy-resource resource)
|
||||
(spawn-fiber
|
||||
(lambda ()
|
||||
(put-message channel
|
||||
(list 'remove resource)))
|
||||
#:parallel? #t)))
|
||||
available)
|
||||
|
||||
(let ((current-internal-time (get-internal-real-time)))
|
||||
(for-each
|
||||
(match-lambda
|
||||
|
|
@ -1097,28 +681,10 @@
|
|||
internal-time-units-per-second))
|
||||
(const #f)))
|
||||
op))))))))
|
||||
(car waiters))
|
||||
waiters))
|
||||
|
||||
(destroy-loop resources))))
|
||||
|
||||
(if destructor
|
||||
(begin
|
||||
(for-each
|
||||
(lambda (resource)
|
||||
(spawn-fiber-to-destroy-resource resource))
|
||||
available)
|
||||
(destroy-loop resources))
|
||||
(let dl ((resources resources)
|
||||
(available available))
|
||||
(if (null? available)
|
||||
(if (null? resources)
|
||||
(signal-condition!
|
||||
destroy-condition)
|
||||
(destroy-loop resources))
|
||||
(let ((index
|
||||
(list-index (lambda (x)
|
||||
(eq? x (car available)))
|
||||
resources)))
|
||||
(dl (remove-at-index! resources index)
|
||||
(cdr available)))))))))
|
||||
(unknown
|
||||
(simple-format
|
||||
(current-error-port)
|
||||
|
|
@ -1178,8 +744,7 @@
|
|||
(put-operation (resource-pool-channel pool)
|
||||
(list 'destroy))
|
||||
(lambda _
|
||||
(wait
|
||||
(resource-pool-destroy-condition pool))))
|
||||
(wait (resource-pool-destroy-condition pool))))
|
||||
(wait-operation
|
||||
(resource-pool-destroy-condition pool))))
|
||||
#t)
|
||||
|
|
@ -1198,7 +763,7 @@
|
|||
(record-constructor &resource-pool-timeout))
|
||||
|
||||
(define resource-pool-timeout-error?
|
||||
(exception-predicate &resource-pool-timeout))
|
||||
(record-predicate &resource-pool-timeout))
|
||||
|
||||
(define &resource-pool-too-many-waiters
|
||||
(make-exception-type '&recource-pool-too-many-waiters
|
||||
|
|
@ -1219,7 +784,7 @@
|
|||
(record-constructor &resource-pool-too-many-waiters))
|
||||
|
||||
(define resource-pool-too-many-waiters-error?
|
||||
(exception-predicate &resource-pool-too-many-waiters))
|
||||
(record-predicate &resource-pool-too-many-waiters))
|
||||
|
||||
(define &resource-pool-destroyed
|
||||
(make-exception-type '&recource-pool-destroyed
|
||||
|
|
@ -1235,7 +800,7 @@
|
|||
(record-constructor &resource-pool-destroyed))
|
||||
|
||||
(define resource-pool-destroyed-error?
|
||||
(exception-predicate &resource-pool-destroyed))
|
||||
(record-predicate &resource-pool-destroyed))
|
||||
|
||||
(define &resource-pool-destroy-resource
|
||||
(make-exception-type '&recource-pool-destroy-resource
|
||||
|
|
@ -1246,7 +811,7 @@
|
|||
(record-constructor &resource-pool-destroy-resource))
|
||||
|
||||
(define resource-pool-destroy-resource-exception?
|
||||
(exception-predicate &resource-pool-destroy-resource))
|
||||
(record-predicate &resource-pool-destroy-resource))
|
||||
|
||||
(define resource-pool-default-timeout-handler
|
||||
(make-parameter #f))
|
||||
|
|
@ -1314,9 +879,8 @@ available. Return the resource once PROC has returned."
|
|||
start-time)
|
||||
'timeout)
|
||||
response))
|
||||
'timeout))
|
||||
'timeout)))
|
||||
(let ((reply (make-channel)))
|
||||
'timeout)))))
|
||||
(let loop ((reply (make-channel)))
|
||||
(put-message channel
|
||||
(list 'checkout
|
||||
reply
|
||||
|
|
@ -1354,7 +918,8 @@ available. Return the resource once PROC has returned."
|
|||
'destroy
|
||||
'return)
|
||||
resource))
|
||||
(raise-exception exn))
|
||||
(unless (resource-pool-destroy-resource-exception? exn)
|
||||
(raise-exception exn)))
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
|
|
@ -1384,17 +949,13 @@ available. Return the resource once PROC has returned."
|
|||
(lambda (resource) exp ...)))
|
||||
|
||||
(define* (resource-pool-stats pool #:key (timeout 5))
|
||||
(if timeout
|
||||
(let* ((reply (make-channel))
|
||||
(start-time (get-internal-real-time))
|
||||
(timeout-time
|
||||
(+ start-time
|
||||
(* internal-time-units-per-second timeout))))
|
||||
(let ((reply (make-channel))
|
||||
(start-time (get-internal-real-time)))
|
||||
(perform-operation
|
||||
(choice-operation
|
||||
(wrap-operation
|
||||
(put-operation (resource-pool-channel pool)
|
||||
`(stats ,reply ,timeout-time))
|
||||
`(stats ,reply))
|
||||
(const #t))
|
||||
(wrap-operation (sleep-operation timeout)
|
||||
(lambda _
|
||||
|
|
@ -1415,11 +976,7 @@ available. Return the resource once PROC has returned."
|
|||
(raise-exception
|
||||
(make-resource-pool-timeout-error pool))))))
|
||||
(raise-exception
|
||||
(make-resource-pool-timeout-error pool)))))
|
||||
(let ((reply (make-channel)))
|
||||
(put-message (resource-pool-channel pool)
|
||||
`(stats ,reply #f))
|
||||
(get-message reply))))
|
||||
(make-resource-pool-timeout-error pool))))))
|
||||
|
||||
(define (resource-pool-list-resources pool)
|
||||
(let ((reply (make-channel)))
|
||||
|
|
|
|||
|
|
@ -198,7 +198,7 @@ from there, or #f if that would be an empty string."
|
|||
(record-accessor &thread-pool-timeout-error 'pool)))
|
||||
|
||||
(define thread-pool-timeout-error?
|
||||
(exception-predicate &thread-pool-timeout-error))
|
||||
(record-predicate &thread-pool-timeout-error))
|
||||
|
||||
(define* (make-fixed-size-thread-pool size
|
||||
#:key
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@
|
|||
(record-constructor &port-timeout-error))
|
||||
|
||||
(define port-timeout-error?
|
||||
(exception-predicate &port-timeout-error))
|
||||
(record-predicate &port-timeout-error))
|
||||
|
||||
(define &port-read-timeout-error
|
||||
(make-exception-type '&port-read-timeout-error
|
||||
|
|
@ -96,7 +96,7 @@
|
|||
(record-constructor &port-read-timeout-error))
|
||||
|
||||
(define port-read-timeout-error?
|
||||
(exception-predicate &port-read-timeout-error))
|
||||
(record-predicate &port-read-timeout-error))
|
||||
|
||||
(define &port-write-timeout-error
|
||||
(make-exception-type '&port-write-timeout-error
|
||||
|
|
@ -107,7 +107,7 @@
|
|||
(record-constructor &port-write-timeout-error))
|
||||
|
||||
(define port-write-timeout-error?
|
||||
(exception-predicate &port-write-timeout-error))
|
||||
(record-predicate &port-write-timeout-error))
|
||||
|
||||
(define (readable? port)
|
||||
"Test if PORT is writable."
|
||||
|
|
|
|||
|
|
@ -63,14 +63,6 @@
|
|||
(bind sock family addr port)
|
||||
sock))
|
||||
|
||||
(define crlf-bv
|
||||
(string->utf8 "\r\n"))
|
||||
|
||||
(define (chunked-output-port-overhead-bytes write-size)
|
||||
(+ (string-length (number->string write-size 16))
|
||||
(bytevector-length crlf-bv)
|
||||
(bytevector-length crlf-bv)))
|
||||
|
||||
(define* (make-chunked-output-port/knots port #:key (keep-alive? #f)
|
||||
(buffering 1200))
|
||||
"Returns a new port which translates non-encoded data into a HTTP
|
||||
|
|
@ -82,12 +74,10 @@ when done, as it will output the remaining data, and encode the final
|
|||
zero chunk. When the port is closed it will also close PORT, unless
|
||||
KEEP-ALIVE? is true."
|
||||
(define (write! bv start count)
|
||||
(let ((len-string
|
||||
(number->string count 16)))
|
||||
(put-string port len-string))
|
||||
(put-bytevector port crlf-bv 0 2)
|
||||
(put-string port (number->string count 16))
|
||||
(put-string port "\r\n")
|
||||
(put-bytevector port bv start count)
|
||||
(put-bytevector port crlf-bv 0 2)
|
||||
(put-string port "\r\n")
|
||||
(force-output port)
|
||||
count)
|
||||
|
||||
|
|
@ -140,7 +130,7 @@ closes PORT, unless KEEP-ALIVE? is true."
|
|||
(record-constructor &request-body-ended-prematurely))
|
||||
|
||||
(define request-body-ended-prematurely-error?
|
||||
(exception-predicate &request-body-ended-prematurely))
|
||||
(record-predicate &request-body-ended-prematurely))
|
||||
|
||||
(define (request-body-port/knots r)
|
||||
(cond
|
||||
|
|
@ -341,19 +331,15 @@ on the procedure being called at any particular time."
|
|||
(string->utf8
|
||||
"internal server error")))
|
||||
|
||||
(define* (handle-request handler client
|
||||
(define (handle-request handler client
|
||||
read-request-exception-handler
|
||||
write-response-exception-handler
|
||||
buffer-size
|
||||
#:key post-request-hook)
|
||||
write-response-exception-handler)
|
||||
(let ((request
|
||||
(with-exception-handler
|
||||
read-request-exception-handler
|
||||
(lambda ()
|
||||
(read-request client))
|
||||
#:unwind? #t))
|
||||
(read-request-time
|
||||
(get-internal-real-time)))
|
||||
#:unwind? #t)))
|
||||
(let ((response
|
||||
body
|
||||
(cond
|
||||
|
|
@ -402,9 +388,7 @@ on the procedure being called at any particular time."
|
|||
(lambda ()
|
||||
(write-response response client)
|
||||
|
||||
(let ((response-start-time
|
||||
(get-internal-real-time))
|
||||
(body-written?
|
||||
(let ((body-written?
|
||||
(if (procedure? body)
|
||||
(let* ((type (response-content-type response
|
||||
'(text/plain)))
|
||||
|
|
@ -415,11 +399,7 @@ on the procedure being called at any particular time."
|
|||
client
|
||||
(make-chunked-output-port/knots
|
||||
client
|
||||
#:keep-alive? #t
|
||||
#:buffering
|
||||
(- buffer-size
|
||||
(chunked-output-port-overhead-bytes
|
||||
buffer-size))))))
|
||||
#:keep-alive? #t))))
|
||||
(set-port-encoding! body-port charset)
|
||||
(let ((body-written?
|
||||
(with-exception-handler
|
||||
|
|
@ -443,11 +423,6 @@ on the procedure being called at any particular time."
|
|||
(if body-written?
|
||||
(begin
|
||||
(force-output client)
|
||||
(when post-request-hook
|
||||
(post-request-hook request
|
||||
#:read-request-time read-request-time
|
||||
#:response-start-time response-start-time
|
||||
#:response-end-time (get-internal-real-time)))
|
||||
(when (and (procedure? body)
|
||||
(response-content-length response))
|
||||
(set-port-encoding! client "ISO-8859-1"))
|
||||
|
|
@ -459,8 +434,7 @@ on the procedure being called at any particular time."
|
|||
read-request-exception-handler
|
||||
write-response-exception-handler
|
||||
connection-idle-timeout
|
||||
buffer-size
|
||||
post-request-hook)
|
||||
buffer-size)
|
||||
;; Always disable Nagle's algorithm, as we handle buffering
|
||||
;; ourselves; when we force-output, we really want the data to go
|
||||
;; out.
|
||||
|
|
@ -498,29 +472,11 @@ on the procedure being called at any particular time."
|
|||
(else
|
||||
(let ((keep-alive? (handle-request handler client
|
||||
read-request-exception-handler
|
||||
write-response-exception-handler
|
||||
buffer-size
|
||||
#:post-request-hook
|
||||
post-request-hook)))
|
||||
write-response-exception-handler)))
|
||||
(if keep-alive?
|
||||
(loop)
|
||||
(close-port client)))))))
|
||||
|
||||
(define (post-request-hook/safe post-request-hook)
|
||||
(if post-request-hook
|
||||
(lambda args
|
||||
(with-exception-handler
|
||||
(lambda (exn) #f)
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(print-backtrace-and-exception/knots exn)
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(apply post-request-hook args))))
|
||||
#:unwind? #t))
|
||||
#f))
|
||||
|
||||
(define-record-type <web-server>
|
||||
(make-web-server socket port)
|
||||
web-server?
|
||||
|
|
@ -540,8 +496,7 @@ on the procedure being called at any particular time."
|
|||
(write-response-exception-handler
|
||||
default-write-response-exception-handler)
|
||||
(connection-idle-timeout #f)
|
||||
(connection-buffer-size 1024)
|
||||
post-request-hook)
|
||||
(connection-buffer-size 1024))
|
||||
"Run the knots web server.
|
||||
|
||||
HANDLER should be a procedure that takes one argument, the HTTP
|
||||
|
|
@ -577,9 +532,7 @@ before sending back to the client."
|
|||
read-request-exception-handler
|
||||
write-response-exception-handler
|
||||
connection-idle-timeout
|
||||
connection-buffer-size
|
||||
(post-request-hook/safe
|
||||
post-request-hook)))
|
||||
connection-buffer-size))
|
||||
#:parallel? #t)
|
||||
(loop))))))
|
||||
|
||||
|
|
|
|||
12
tests.scm
12
tests.scm
|
|
@ -1,11 +1,10 @@
|
|||
(define-module (tests)
|
||||
#:use-module (ice-9 exceptions)
|
||||
#:use-module (fibers)
|
||||
#:use-module (knots)
|
||||
#:export (run-fibers-for-tests
|
||||
assert-no-heap-growth))
|
||||
|
||||
(define* (run-fibers-for-tests thunk #:key (drain? #t))
|
||||
(define (run-fibers-for-tests thunk)
|
||||
(let ((result
|
||||
(run-fibers
|
||||
(lambda ()
|
||||
|
|
@ -13,18 +12,15 @@
|
|||
(lambda (exn)
|
||||
exn)
|
||||
(lambda ()
|
||||
(simple-format #t "running ~A\n" thunk)
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(print-backtrace-and-exception/knots exn)
|
||||
(backtrace)
|
||||
(raise-exception exn))
|
||||
(lambda ()
|
||||
(start-stack #t (thunk))))
|
||||
thunk)
|
||||
#t)
|
||||
#:unwind? #t))
|
||||
#:hz 0
|
||||
#:parallelism 1
|
||||
#:drain? drain?)))
|
||||
#:parallelism 1)))
|
||||
(if (exception? result)
|
||||
(raise-exception result)
|
||||
result)))
|
||||
|
|
|
|||
|
|
@ -61,24 +61,6 @@
|
|||
identity
|
||||
'(()))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
(unless (and (exception-with-message? exn)
|
||||
(string=? (exception-message exn)
|
||||
"foo"))
|
||||
(raise-exception exn)))
|
||||
(lambda ()
|
||||
(fibers-map-with-progress
|
||||
(lambda _
|
||||
(raise-exception
|
||||
(make-exception-with-message "foo")))
|
||||
'((1)))
|
||||
|
||||
(error 'should-not-reach-here))
|
||||
#:unwind? #t)))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(with-exception-handler
|
||||
|
|
@ -129,16 +111,4 @@
|
|||
|
||||
(assert-equal a 1))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(let ((parallelism-limiter (make-parallelism-limiter 2)))
|
||||
(fibers-for-each
|
||||
(lambda _
|
||||
(with-parallelism-limiter
|
||||
parallelism-limiter
|
||||
#f))
|
||||
(iota 50))
|
||||
|
||||
(destroy-parallelism-limiter parallelism-limiter))))
|
||||
|
||||
(display "parallelism test finished successfully\n")
|
||||
|
|
|
|||
|
|
@ -19,21 +19,7 @@
|
|||
(number?
|
||||
(with-resource-from-pool resource-pool
|
||||
res
|
||||
res)))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
(let ((resource-pool (make-fixed-size-resource-pool
|
||||
(list 1))))
|
||||
(assert-true
|
||||
(number?
|
||||
(with-resource-from-pool resource-pool
|
||||
res
|
||||
res)))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
res))))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
|
|
@ -45,9 +31,7 @@
|
|||
(number?
|
||||
(with-resource-from-pool resource-pool
|
||||
res
|
||||
res)))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
res))))))
|
||||
|
||||
(let* ((error-constructor
|
||||
(record-constructor &resource-pool-timeout))
|
||||
|
|
@ -104,13 +88,10 @@
|
|||
res))
|
||||
(iota 20))
|
||||
|
||||
(let loop ((stats (resource-pool-stats resource-pool
|
||||
#:timeout #f)))
|
||||
(let loop ((stats (resource-pool-stats resource-pool)))
|
||||
(unless (= 0 (assq-ref stats 'resources))
|
||||
(sleep 0.1)
|
||||
(loop (resource-pool-stats resource-pool #:timeout #f))))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
(loop (resource-pool-stats resource-pool)))))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
|
|
@ -134,9 +115,7 @@
|
|||
(set! counter (+ 1 counter))
|
||||
(error "collision detected")))))
|
||||
20
|
||||
(iota 50))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
(iota 50)))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
|
|
@ -150,7 +129,7 @@
|
|||
(error "collision detected")))
|
||||
(new-number))
|
||||
1
|
||||
#:default-checkout-timeout 5)))
|
||||
#:default-checkout-timeout 120)))
|
||||
(fibers-batch-for-each
|
||||
(lambda _
|
||||
(with-resource-from-pool
|
||||
|
|
@ -161,9 +140,7 @@
|
|||
(set! counter (+ 1 counter))
|
||||
(error "collision detected")))))
|
||||
20
|
||||
(iota 50))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
(iota 50)))))
|
||||
|
||||
(run-fibers-for-tests
|
||||
(lambda ()
|
||||
|
|
@ -187,14 +164,14 @@
|
|||
(call-with-resource-from-pool
|
||||
resource-pool
|
||||
(lambda (res)
|
||||
#f)))
|
||||
(error 'should-not-be-reached))))
|
||||
#:unwind? #t)))
|
||||
|
||||
(while (= 0
|
||||
(assq-ref
|
||||
(resource-pool-stats resource-pool #:timeout #f)
|
||||
(resource-pool-stats resource-pool)
|
||||
'waiters))
|
||||
(sleep 0.1))
|
||||
(sleep 0))
|
||||
|
||||
(with-exception-handler
|
||||
(lambda (exn)
|
||||
|
|
@ -207,8 +184,6 @@
|
|||
resource-pool
|
||||
(lambda (res)
|
||||
(error 'should-not-be-reached))))
|
||||
#:unwind? #t)))
|
||||
|
||||
(destroy-resource-pool resource-pool))))
|
||||
#:unwind? #t))))))
|
||||
|
||||
(display "resource-pool test finished successfully\n")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue