diff options
| -rw-r--r-- | guix-data-service/utils.scm | 51 |
1 files changed, 39 insertions, 12 deletions
diff --git a/guix-data-service/utils.scm b/guix-data-service/utils.scm index c5c89a4..b7124d5 100644 --- a/guix-data-service/utils.scm +++ b/guix-data-service/utils.scm @@ -32,6 +32,7 @@ prevent-inlining-for-tests thread-pool-channel + thread-pool-request-timeout make-thread-pool-channel parallel-via-thread-pool-channel par-map& @@ -129,30 +130,56 @@ (iota threads)) channel)) +(define &thread-pool-request-timeout + (make-exception-type '&thread-pool-request-timeout + &error + '())) + +(define make-thread-pool-request-timeout-error + (record-constructor &thread-pool-request-timeout)) + +(define thread-pool-request-timeout-error? + (record-predicate &thread-pool-request-timeout)) + (define thread-pool-channel (make-parameter #f)) +(define thread-pool-request-timeout + (make-parameter #f)) + (define (defer-to-thread-pool-channel thunk) (let ((reply (make-channel))) (spawn-fiber (lambda () - (put-message (thread-pool-channel) - (list reply - (get-internal-real-time) - thunk)))) + (let ((val + (perform-operation + (let ((put + (wrap-operation + (put-operation (thread-pool-channel) + (list reply + (get-internal-real-time) + thunk)) + (const 'success)))) + (or + (and=> (thread-pool-request-timeout) + (lambda (timeout) + (choice-operation + put + (wrap-operation (sleep-operation timeout) + (const 'request-timeout))))) + put))))) + (when (eq? val 'request-timeout) + (put-message reply val))))) reply)) -(define (fetch-result-of-defered-thunk reply-channel) - (match (get-message reply-channel) - (('worker-thread-error . exn) - (raise-exception exn)) - (result - (apply values result)))) - (define (fetch-result-of-defered-thunks . reply-channels) - (let ((responses (map get-message reply-channels))) + (let ((responses (map get-message + reply-channels))) (map (match-lambda + ('request-timeout + (raise-exception + (make-thread-pool-request-timeout-error))) (('worker-thread-error . exn) (raise-exception exn)) (result |