;;; Guile Knots
;;; Copyright © 2020 Christopher Baines <mail@cbaines.net>
;;;
;;; This file is part of Guile Knots.
;;;
;;; The Guile Knots is free software; you can redistribute it and/or
;;; modify it under the terms of the GNU General Public License as
;;; published by the Free Software Foundation; either version 3 of the
;;; License, or (at your option) any later version.
;;;
;;; The Guile Knots is distributed in the hope that it will be useful,
;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;;; General Public License for more details.
;;;
;;; You should have received a copy of the GNU General Public License
;;; along with the guix-data-service.  If not, see
;;; <http://www.gnu.org/licenses/>.

(define-module (knots thread-pool)
  #:use-module (srfi srfi-1)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-19)
  #:use-module (srfi srfi-71)
  #:use-module (system foreign)
  #:use-module (system base target)
  #:use-module (rnrs bytevectors)
  #:use-module (ice-9 q)
  #:use-module (ice-9 match)
  #:use-module (ice-9 atomic)
  #:use-module (ice-9 threads)
  #:use-module (fibers)
  #:use-module (fibers timers)
  #:use-module (fibers channels)
  #:use-module (fibers operations)
  #:use-module (knots)
  #:use-module (knots resource-pool)
  #:export (set-thread-name
            thread-name

            &thread-pool-timeout-error
            thread-pool-timeout-error-pool
            thread-pool-timeout-error?

            make-thread-pool
            thread-pool?
            thread-pool-resource-pool

            make-fixed-size-thread-pool
            fixed-size-thread-pool?
            fixed-size-thread-pool-channel
            fixed-size-thread-pool-current-procedures

            ;; These procedures work for thread pools and fixed size
            ;; thread pools
            thread-pool-arguments-parameter
            thread-pool-default-checkout-timeout

            destroy-thread-pool

            call-with-thread))

(define* (syscall->procedure return-type name argument-types
                             #:key library)
  "Return a procedure that wraps the C function NAME using the dynamic FFI,
and that returns two values: NAME's return value, and errno.  When LIBRARY is
specified, look up NAME in that library rather than in the global symbol name
space.

If an error occurs while creating the binding, defer the error report until
the returned procedure is called."
  (catch #t
    (lambda ()
      ;; Note: When #:library is set, try it first and fall back to libc
      ;; proper.  This is because libraries like libutil.so have been subsumed
      ;; by libc.so with glibc >= 2.34.
      (let ((ptr (dynamic-func name
                               (if library
                                   (or (false-if-exception
                                        (dynamic-link library))
                                       (dynamic-link))
                                   (dynamic-link)))))
        ;; The #:return-errno? facility was introduced in Guile 2.0.12.
        (pointer->procedure return-type ptr argument-types
                            #:return-errno? #t)))
    (lambda args
      (lambda _
        (throw 'system-error name  "~A" (list (strerror ENOSYS))
               (list ENOSYS))))))

(define %prctl
  ;; Should it win the API contest against 'ioctl'?  You tell us!
  (syscall->procedure int "prctl"
                      (list int unsigned-long unsigned-long
                            unsigned-long unsigned-long)))

(define PR_SET_NAME 15)                           ;<linux/prctl.h>
(define PR_GET_NAME 16)
(define PR_SET_CHILD_SUBREAPER 36)

(define (set-child-subreaper!)
  "Set the CHILD_SUBREAPER capability for the current process."
  (%prctl PR_SET_CHILD_SUBREAPER 1 0 0 0))

(define %max-thread-name-length
  ;; Maximum length in bytes of the process name, including the terminating
  ;; zero.
  16)

(define (set-thread-name!/linux name)
  "Set the name of the calling thread to NAME.  NAME is truncated to 15
bytes."
  (let ((ptr (string->pointer name)))
    (let ((ret
           err
           (%prctl PR_SET_NAME
                   (pointer-address ptr) 0 0 0)))
      (unless (zero? ret)
        (throw 'set-process-name "set-process-name"
               "set-process-name: ~A"
               (list (strerror err))
               (list err))))))

(define (bytes->string bytes)
  "Read BYTES, a list of bytes, and return the null-terminated string decoded
from there, or #f if that would be an empty string."
  (match (take-while (negate zero?) bytes)
    (()
     #f)
    (non-zero
     (list->string (map integer->char non-zero)))))

(define (thread-name/linux)
  "Return the name of the calling thread as a string."
  (let ((buf (make-bytevector %max-thread-name-length)))
    (let ((ret
           err
           (%prctl PR_GET_NAME
                   (pointer-address (bytevector->pointer buf))
                   0 0 0)))
      (if (zero? ret)
          (bytes->string (bytevector->u8-list buf))
          (throw 'process-name "process-name"
                 "process-name: ~A"
                 (list (strerror err))
                 (list err))))))

(define set-thread-name
  (if (string-contains %host-type "linux")
      set-thread-name!/linux
      (const #f)))

(define thread-name
  (if (string-contains %host-type "linux")
      thread-name/linux
      (const "")))

(define-record-type <thread-pool>
  (thread-pool resource-pool arguments-parameter)
  thread-pool?
  (resource-pool       thread-pool-resource-pool)
  (arguments-parameter thread-pool-arguments-parameter-accessor))

(define-record-type <fixed-size-thread-pool>
  (fixed-size-thread-pool channel arguments-parameter current-procedures
                          default-checkout-timeout)
  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))

;; Since both thread pool records have this field, use a procedure
;; than handles the appropriate accessor
(define (thread-pool-arguments-parameter pool)
  (if (fixed-size-thread-pool? pool)
      (fixed-size-thread-pool-arguments-parameter pool)
      (thread-pool-arguments-parameter-accessor pool)))

(define (thread-pool-default-checkout-timeout pool)
  (if (fixed-size-thread-pool? pool)
      (fixed-size-thread-pool-default-checkout-timeout pool)
      (assq-ref (resource-pool-configuration
                 (thread-pool-resource-pool pool))
                'default-checkout-timeout)))

(define &thread-pool-timeout
  (make-exception-type '&thread-pool-timeout
                       &error
                       '(pool)))

(define make-thread-pool-timeout-error
  (record-constructor &thread-pool-timeout))

(define thread-pool-timeout-error-pool
  (exception-accessor
   &thread-pool-timeout
   (record-accessor &thread-pool-timeout 'pool)))

(define thread-pool-timeout-error?
  (record-predicate &thread-pool-timeout))

(define* (make-fixed-size-thread-pool size
                                      #:key
                                      thread-initializer
                                      thread-destructor
                                      delay-logger
                                      duration-logger
                                      thread-lifetime
                                      (expire-on-exception? #f)
                                      (name "unnamed")
                                      (use-default-io-waiters? #t)
                                      default-checkout-timeout)
  (define channel
    (make-channel))

  (define param
    (make-parameter #f))

  (define thread-proc-vector
    (make-vector size #f))

  (define (initializer/safe)
    (let ((args
           (with-exception-handler
               (lambda _ #f)
             (lambda ()
               (with-exception-handler
                   (lambda (exn)
                     (simple-format
                      (current-error-port)
                      "exception running initializer in thread pool (~A): ~A\n"
                      name
                      thread-initializer)
                     (print-backtrace-and-exception/knots exn)
                     (raise-exception exn))
                 thread-initializer))
             #:unwind? #t)))

      (if args
          args
          ;; never give up, just keep retrying
          (begin
            (sleep 1)
            (initializer/safe)))))

  (define (destructor/safe args)
    (let ((success?
           (with-exception-handler
               (lambda _ #f)
             (lambda ()
               (with-exception-handler
                   (lambda (exn)
                     (simple-format
                      (current-error-port)
                      "exception running destructor in thread pool (~A): ~A\n"
                      name
                      thread-destructor)
                     (print-backtrace-and-exception/knots exn)
                     (raise-exception exn))
                 (lambda ()
                   (apply thread-destructor args)
                   #t)))
             #:unwind? #t)))

      (or success?
          #t
          (begin
            (sleep 1)
            (destructor/safe args)))))

  (define (process channel args)
    (let loop ()
      (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)))

         (let* ((start-time (get-internal-real-time))
                (response
                 (with-exception-handler
                     (lambda (exn)
                       (list 'thread-pool-error
                             (/ (- (get-internal-real-time)
                                   start-time)
                                internal-time-units-per-second)
                             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 ()
                         (call-with-values
                             (lambda ()
                               (start-stack
                                #t
                                (apply proc args)))
                           (lambda vals
                             (cons (/ (- (get-internal-real-time)
                                         start-time)
                                      internal-time-units-per-second)
                                   vals))))))
                   #:unwind? #t)))

           (put-message reply
                        response)

           (let ((exception?
                  (match response
                    (('thread-pool-error duration _)
                     (when duration-logger
                       (duration-logger duration proc))
                     #t)
                    ((duration . _)
                     (when duration-logger
                       (duration-logger duration proc))
                     #f))))
             (if (and exception?
                      expire-on-exception?)
                 #t
                 (loop))))))))

  (define (start-thread index channel)
    (call-with-new-thread
     (lambda ()
       (catch 'system-error
         (lambda ()
           (set-thread-name
            (string-append
             name " w t " (number->string index))))
         (const #t))

       (let init ((args (if thread-initializer
                            (initializer/safe)
                            '())))
         (let ((continue?
                (with-exception-handler
                    (lambda (exn)
                      (simple-format
                       (current-error-port)
                       "knots: thread-pool: internal exception: ~A\n" exn))
                  (lambda ()
                    (parameterize ((param args))
                      (process channel args)))
                  #:unwind? #t)))

           (when thread-destructor
             (destructor/safe args))

           (when continue?
             (init (if thread-initializer
                       (initializer/safe)
                       '()))))))))

  (for-each
   (lambda (i)
     (if use-default-io-waiters?
         (call-with-default-io-waiters
          (lambda ()
            (start-thread i channel)))
         (start-thread i channel)))
   (iota size))

  (fixed-size-thread-pool channel
                          param
                          thread-proc-vector
                          default-checkout-timeout))

(define* (make-thread-pool max-size
                           #:key
                           (min-size max-size)
                           scheduler
                           thread-initializer
                           thread-destructor
                           (delay-logger (lambda _ #f))
                           (duration-logger (const #f))
                           thread-lifetime
                           (expire-on-exception? #f)
                           (name "unnamed")
                           (use-default-io-waiters? #t)
                           default-checkout-timeout)
  "Return a channel used to offload work to a dedicated thread.  ARGS are the
arguments of the thread pool procedure."
  (define param
    (make-parameter #f))

  (let ((resource-pool
         (make-resource-pool
          (lambda ()
            (make-fixed-size-thread-pool
             1
             #:thread-initializer thread-initializer
             #:thread-destructor thread-destructor
             #:thread-lifetime thread-lifetime
             #:expire-on-exception? expire-on-exception?
             #:name name
             #:use-default-io-waiters? use-default-io-waiters?))
          max-size
          #:destructor destroy-thread-pool
          #:min-size min-size
          #:delay-logger delay-logger
          #:scheduler scheduler
          #:duration-logger duration-logger
          #:default-checkout-timeout default-checkout-timeout)))

    (thread-pool resource-pool
                 param)))

(define* (call-with-thread thread-pool
                           proc
                           #:key
                           duration-logger
                           checkout-timeout
                           channel
                           destroy-thread-on-exception?
                           (max-waiters 'default))
  "Send PROC to the thread pool through CHANNEL.  Return the result of PROC.
If already in the thread pool, call PROC immediately."
  (define (handle-proc fixed-size-thread-pool
                       reply-channel
                       start-time
                       timeout)
    (let* ((request-channel
            (or channel
                (fixed-size-thread-pool-channel
                 fixed-size-thread-pool)))
           (operation-success?
            (perform-operation
             (let ((put
                    (wrap-operation
                     (put-operation request-channel
                                    (list reply-channel
                                          start-time
                                          proc))
                     (const #t))))

               (if timeout
                   (choice-operation
                    put
                    (wrap-operation (sleep-operation timeout)
                                    (const #f)))
                   put)))))

    (unless operation-success?
      (raise-exception
       (make-thread-pool-timeout-error)))

    (let ((reply (get-message reply-channel)))
      (match reply
        (('thread-pool-error duration exn)
         (when duration-logger
           (duration-logger duration))
         (raise-exception exn))
        ((duration . result)
         (when duration-logger
           (duration-logger duration))
         (apply values result))))))

  (let ((args ((thread-pool-arguments-parameter thread-pool))))
    (if args
        (apply proc args)
        (let ((start-time (get-internal-real-time))
              (reply-channel (make-channel)))
          (if (fixed-size-thread-pool? thread-pool)
              (handle-proc thread-pool
                           reply-channel
                           start-time
                           checkout-timeout)
              (with-exception-handler
                  (lambda (exn)
                    (if (and (resource-pool-timeout-error? exn)
                             (eq? (resource-pool-timeout-error-pool exn)
                                  (thread-pool-resource-pool thread-pool)))
                        (raise-exception
                         (make-thread-pool-timeout-error thread-pool))
                        (raise-exception exn)))
                (lambda ()
                  (call-with-resource-from-pool (thread-pool-resource-pool
                                                 thread-pool)
                    (lambda (fixed-size-thread-pool)
                      (if checkout-timeout
                          (let ((remaining-time
                                 (/ (- (get-internal-real-time) start-time)
                                    internal-time-units-per-second)))
                            (if (< remaining-time checkout-timeout)
                                (handle-proc fixed-size-thread-pool
                                             reply-channel
                                             start-time
                                             remaining-time)
                                (raise-exception
                                 (make-thread-pool-timeout-error thread-pool))))
                          (handle-proc fixed-size-thread-pool
                                       reply-channel
                                       start-time
                                       #f)))
                    #:max-waiters max-waiters
                    #:timeout checkout-timeout
                    #:destroy-resource-on-exception?
                    destroy-thread-on-exception?))))))))

(define (destroy-thread-pool pool)
  (if (fixed-size-thread-pool? pool)
      (put-message
       (fixed-size-thread-pool-channel pool)
       'destroy)
      (destroy-resource-pool
       (thread-pool-resource-pool pool))))