guile-knots/knots/timeout.scm

;;; 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 timeout)
  #:use-module (srfi srfi-71)
  #:use-module (ice-9 match)
  #:use-module (ice-9 atomic)
  #:use-module (ice-9 exceptions)
  #:use-module (ice-9 ports internal)
  #:use-module (ice-9 suspendable-ports)
  #:use-module (fibers)
  #:use-module (fibers timers)
  #:use-module (fibers channels)
  #:use-module (fibers scheduler)
  #:use-module (fibers operations)
  #:export (with-fibers-timeout

            wait-until-port-readable-operation
            wait-until-port-writable-operation

            &port-timeout
            port-timeout-error?

            &port-read-timeout
            port-read-timeout-error

            &port-write-timeout
            port-write-timeout-error?

            with-port-timeouts))

(define* (with-fibers-timeout thunk #:key timeout on-timeout)
  (let ((channel (make-channel)))
    (spawn-fiber
     (lambda ()
       (with-exception-handler
           (lambda (exn)
             (perform-operation
              (choice-operation
               (put-operation channel (cons 'exception exn))
               (sleep-operation timeout))))
         (lambda ()
           (call-with-values thunk
             (lambda vals
               (perform-operation
                (choice-operation
                 (put-operation channel vals)
                 (sleep-operation timeout))))))
         #:unwind? #t)))

    (match (perform-operation
            (choice-operation
             (get-operation channel)
             (wrap-operation (sleep-operation timeout)
                             (const 'timeout))))
      ('timeout
       (on-timeout))
      (('exception . exn)
       (raise-exception exn))
      (vals
       (apply values vals)))))

(define &port-timeout
  (make-exception-type '&port-timeout
                       &external-error
                       '(thunk port)))

(define make-port-timeout-error
  (record-constructor &port-timeout))

(define port-timeout-error?
  (record-predicate &port-timeout))

(define &port-read-timeout
  (make-exception-type '&port-read-timeout
                       &port-timeout
                       '()))

(define make-port-read-timeout-error
  (record-constructor &port-read-timeout))

(define port-read-timeout-error?
  (record-predicate &port-read-timeout))

(define &port-write-timeout
  (make-exception-type '&port-write-timeout
                       &port-timeout
                       '()))

(define make-port-write-timeout-error
  (record-constructor &port-write-timeout))

(define port-write-timeout-error?
  (record-predicate &port-write-timeout))

(define (readable? port)
  "Test if PORT is writable."
  (= 1 (port-poll port "r" 0)))

(define (writable? port)
  "Test if PORT is writable."
  (= 1 (port-poll port "w" 0)))

(define (make-wait-operation ready? schedule-when-ready port
                             port-ready-fd this-procedure)
  ((module-ref (resolve-interface '(fibers operations))
               'make-base-operation)
   #f
   (lambda _
     (and (ready? port) values))
   (lambda (flag sched resume)
     (define (commit)
       (match (atomic-box-compare-and-swap! flag 'W 'S)
         ('W (resume values))
         ('C (commit))
         ('S #f)))
     (schedule-when-ready
      sched (port-ready-fd port) commit))))

(define (wait-until-port-readable-operation port)
  "Make an operation that will succeed when PORT is readable."
  (unless (input-port? port)
    (error "refusing to wait forever for input on non-input port"))
  (make-wait-operation readable? schedule-task-when-fd-readable port
                       port-read-wait-fd
                       wait-until-port-readable-operation))

(define (wait-until-port-writable-operation port)
  "Make an operation that will succeed when PORT is writable."
  (unless (output-port? port)
    (error "refusing to wait forever for output on non-output port"))
  (make-wait-operation writable? schedule-task-when-fd-writable port
                       port-write-wait-fd
                       wait-until-port-writable-operation))

(define* (with-port-timeouts thunk
                             #:key timeout
                             (read-timeout timeout)
                             (write-timeout timeout))
  (define (no-fibers-wait thunk port mode timeout)
    (define poll-timeout-ms 200)

    ;; When the GC runs, it restarts the poll syscall, but the timeout
    ;; remains unchanged! When the timeout is longer than the time
    ;; between the syscall restarting, I think this renders the
    ;; timeout useless. Therefore, this code uses a short timeout, and
    ;; repeatedly calls poll while watching the clock to see if it has
    ;; timed out overall.
    (let ((timeout-internal
           (+ (get-internal-real-time)
              (* internal-time-units-per-second timeout))))
      (let loop ((poll-value
                  (port-poll port mode poll-timeout-ms)))
        (if (= poll-value 0)
            (if (> (get-internal-real-time)
                   timeout-internal)
                (raise-exception
                 (if (string=? mode "r")
                     (make-port-read-timeout-error thunk port)
                     (make-port-write-timeout-error thunk port)))
                (loop (port-poll port mode poll-timeout-ms)))
            poll-value))))

  (parameterize
      ((current-read-waiter
        (lambda (port)
          (if (current-scheduler)
              (perform-operation
               (choice-operation
                (wait-until-port-readable-operation port)
                (wrap-operation
                 (sleep-operation read-timeout)
                 (lambda ()
                   (raise-exception
                    (make-port-read-timeout-error thunk port))))))
              (no-fibers-wait thunk port "r" read-timeout))))
       (current-write-waiter
        (lambda (port)
          (if (current-scheduler)
              (perform-operation
               (choice-operation
                (wait-until-port-writable-operation port)
                (wrap-operation
                 (sleep-operation write-timeout)
                 (lambda ()
                   (raise-exception
                    (make-port-write-timeout-error thunk port))))))
              (no-fibers-wait thunk port "w" write-timeout)))))
    (thunk)))