Adjust the delay and duration loggers for thread pools

Based on the changes in resource pools.

knots/resource-pool.scm

@@ -60,6 +60,9 @@
             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
@@ -103,6 +106,14 @@
  '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)
@@ -138,7 +149,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
@@ -164,8 +175,8 @@
 
 (define* (make-fixed-size-resource-pool resources-list-or-vector
                                         #:key
-                                        (delay-logger (const #f))
-                                        (duration-logger (const #f))
+                                        (delay-logger #f)
+                                        (duration-logger #f)
                                         scheduler
                                         (name "unnamed")
                                         default-checkout-timeout
@@ -193,6 +204,19 @@ 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.
@@ -547,8 +571,8 @@ to the current scheduler.
 (define* (make-resource-pool return-new-resource max-size
                              #:key (min-size 0)
                              (idle-seconds #f)
-                             (delay-logger (const #f))
-                             (duration-logger (const #f))
+                             (delay-logger #f)
+                             (duration-logger #f)
                              destructor
                              lifetime
                              scheduler
@@ -605,6 +629,19 @@ 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.
@@ -1374,9 +1411,20 @@ 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))
+          (destroy-resource-on-exception? #f)
+          (delay-logger (resource-pool-delay-logger pool))
+          (duration-logger (resource-pool-duration-logger pool)))
   "Call PROC with a resource from POOL, blocking until a resource becomes
-available.  Return the resource once PROC has returned."
+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."
 
   (define timeout-or-default
     (if (eq? timeout 'default)
@@ -1390,6 +1438,26 @@ available.  Return the resource once PROC has returned."
                   '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)))
@@ -1462,6 +1530,12 @@ available.  Return the resource once PROC has returned."
        (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
@@ -1469,6 +1543,10 @@ available.  Return the resource once PROC has returned."
                      ;; 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?
@@ -1496,9 +1574,13 @@ available.  Return the resource once PROC has returned."
                        (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)))))))
+             (apply values vals))))))))
 
 (define-syntax-rule (with-resource-from-pool pool resource exp ...)
   "Evaluate EXP ... with RESOURCE bound to a resource checked out from

knots/thread-pool.scm

@@ -55,6 +55,8 @@
             ;; thread pools
             thread-pool-arguments-parameter
             thread-pool-default-checkout-timeout
+            thread-pool-delay-logger
+            thread-pool-duration-logger
 
             destroy-thread-pool
 
@@ -171,12 +173,15 @@ 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 threads)
+                          default-checkout-timeout delay-logger
+                          duration-logger 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?))
@@ -207,6 +212,20 @@ 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
@@ -277,12 +296,12 @@ Seconds to wait for a free thread slot before raising
 forever).
 
 @item #:delay-logger
-Called as @code{(delay-logger seconds proc)} with the time spent
-waiting for a thread to become available.
+Called as @code{(delay-logger seconds)} with the time spent waiting
+for a thread to become available.
 
 @item #:duration-logger
-Called as @code{(duration-logger seconds proc)} after each procedure
-completes.
+Called as @code{(duration-logger seconds)} after each procedure
+completes, whether it returned normally or raised an exception.
 @end table"
   (define channel
     (make-channel))
@@ -346,15 +365,7 @@ completes.
     (let loop ((lifetime thread-lifetime))
       (match (get-message channel)
         ('destroy #f)
-        ((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)))
-
+        ((reply proc)
          (let* ((start-time (get-internal-real-time))
                 (response
                  (with-exception-handler
@@ -406,11 +417,11 @@ completes.
                   (match response
                     (('thread-pool-error duration _)
                      (when duration-logger
-                       (duration-logger duration proc))
+                       (duration-logger duration))
                      #t)
                     ((duration . _)
                      (when duration-logger
-                       (duration-logger duration proc))
+                       (duration-logger duration))
                      #f))))
              (if (and exception?
                       expire-on-exception?)
@@ -466,6 +477,8 @@ completes.
                           param
                           thread-proc-vector
                           default-checkout-timeout
+                          delay-logger
+                          duration-logger
                           threads))
 
 (define* (make-thread-pool max-size
@@ -474,8 +487,8 @@ completes.
                            scheduler
                            thread-initializer
                            thread-destructor
-                           (delay-logger (lambda _ #f))
-                           (duration-logger (const #f))
+                           delay-logger
+                           duration-logger
                            thread-lifetime
                            (expire-on-exception? #f)
                            (name "unnamed")
@@ -539,7 +552,10 @@ Maximum number of fibers that may queue waiting for a thread.  Raises
 (define* (call-with-thread thread-pool
                            proc
                            #:key
-                           duration-logger
+                           (delay-logger
+                            (thread-pool-delay-logger thread-pool))
+                           (duration-logger
+                            (thread-pool-duration-logger thread-pool))
                            checkout-timeout
                            channel
                            destroy-thread-on-exception?
@@ -565,9 +581,15 @@ 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).
+(whether or not it raised an exception).  Defaults to the pool's
+@code{#:duration-logger} if not specified.
 
 @item #:channel
 Override the channel used to communicate with the thread.
@@ -575,7 +597,8 @@ Override the channel used to communicate with the thread.
   (define (handle-proc fixed-size-thread-pool
                        reply-channel
                        start-time
-                       timeout)
+                       timeout
+                       delay-logger)
     (let* ((request-channel
             (or channel
                 (fixed-size-thread-pool-channel
@@ -586,7 +609,6 @@ Override the channel used to communicate with the thread.
                     (wrap-operation
                      (put-operation request-channel
                                     (list reply-channel
-                                          start-time
                                           proc))
                      (const #t))))
 
@@ -601,6 +623,11 @@ 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)
@@ -621,7 +648,8 @@ Override the channel used to communicate with the thread.
               (handle-proc thread-pool
                            reply-channel
                            start-time
-                           checkout-timeout)
+                           checkout-timeout
+                           delay-logger)
               (with-exception-handler
                   (lambda (exn)
                     (if (and (resource-pool-timeout-error? exn)
@@ -642,13 +670,17 @@ Override the channel used to communicate with the thread.
                                 (handle-proc fixed-size-thread-pool
                                              reply-channel
                                              start-time
-                                             remaining-time)
+                                             remaining-time
+                                             #f)
                                 (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?

tests/resource-pool.scm

@@ -282,4 +282,50 @@
 
      (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")

tests/thread-pool.scm

@@ -172,4 +172,54 @@
   (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")