;; epoll

  Copyright ( C ) 2016   <  >
  Copyright ( C ) 2022   <  >
  Copyright ( C ) 2022   Flaqué <  >
;;;;
;;;; This library is free software; you can redistribute it and/or
;;;; modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation ; either
  version 3 of the License , or ( at your option ) any later version .
;;;;
;;;; This library 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
;;;; Lesser General Public License for more details.
;;;;
  You should have received a copy of the GNU Lesser General Public License
;;;; along with this program.  If not, see </>.
;;;;

(define-module (fibers events-impl)
  #:use-module ((ice-9 binary-ports) #:select (get-u8 put-u8))
  #:use-module (ice-9 atomic)
  #:use-module (ice-9 control)
  #:use-module (ice-9 match)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-9 gnu)
  #:use-module (rnrs bytevectors)
  #:use-module (fibers config)
  #:export (events-impl-create
            events-impl-destroy
            events-impl?
            events-impl-add!
            events-impl-wake!
            events-impl-fd-finalizer
            events-impl-run

            EVENTS_IMPL_READ EVENTS_IMPL_WRITE EVENTS_IMPL_CLOSED_OR_ERROR))

(dynamic-call "init_fibers_epoll"
              (dynamic-link (extension-library "fibers-epoll")))

(when (defined? 'EPOLLRDHUP)
  (export EPOLLRDHUP))
(when (defined? 'EPOLLONESHOT)
  (export EPOLLONESHOT))

(define (make-wake-pipe)
  (let ((pair (pipe2 (logior O_NONBLOCK O_CLOEXEC))))
    (match pair
      ((read-pipe . write-pipe)
       (setvbuf write-pipe 'none)
       (values read-pipe write-pipe)))))

(define-record-type <epoll>
  (make-epoll fd eventsv maxevents state wake-read-pipe wake-write-pipe)
  epoll?
  (fd epoll-fd set-epoll-fd!)
  (eventsv epoll-eventsv set-epoll-eventsv!)
  (maxevents epoll-maxevents set-epoll-maxevents!)
  ;; atomic box of either 'waiting, 'not-waiting or 'dead
  (state epoll-state)
  (wake-read-pipe epoll-wake-read-pipe)
  (wake-write-pipe epoll-wake-write-pipe))

(define-syntax events-offset
  (lambda (x)
    (syntax-case x ()
      ((_ n)
       #`(* n %sizeof-struct-epoll-event)))))

(define-syntax fd-offset
  (lambda (x)
    (syntax-case x ()
      ((_ n)
       #`(+ (* n %sizeof-struct-epoll-event)
            %offsetof-struct-epoll-event-fd)))))

(define epoll-guardian (make-guardian))
(define (pump-epoll-guardian)
  (let ((epoll (epoll-guardian)))
    (when epoll
      (epoll-destroy epoll)
      (pump-epoll-guardian))))
(add-hook! after-gc-hook pump-epoll-guardian)

(define* (epoll-create #:key (close-on-exec? #t) (maxevents 8))
  (call-with-values (lambda () (make-wake-pipe))
    (lambda (read-pipe write-pipe)
      (let* ((state (make-atomic-box 'not-waiting))
             (epoll (make-epoll (primitive-epoll-create close-on-exec?)
                                #f maxevents state read-pipe write-pipe)))
        (epoll-guardian epoll)
        (epoll-add! epoll (fileno read-pipe) EPOLLIN)
        epoll))))

(define (epoll-destroy epoll)
  (atomic-box-set! (epoll-state epoll) 'dead)
  (when (epoll-fd epoll)
    (close-port (epoll-wake-read-pipe epoll))
    ;; FIXME: ignore errors flushing output
    (close-port (epoll-wake-write-pipe epoll))
    (close-fdes (epoll-fd epoll))
    (set-epoll-fd! epoll #f)))

(define (events-impl? impl)
  (epoll? impl))

(define (epoll-add! epoll fd events)
  (primitive-epoll-ctl (epoll-fd epoll) EPOLL_CTL_ADD fd events))

(define (epoll-modify! epoll fd events)
  (primitive-epoll-ctl (epoll-fd epoll) EPOLL_CTL_MOD fd events))

(define (epoll-add*! epoll fd events)
  (catch 'system-error
    (lambda () (epoll-modify! epoll fd events))
    (lambda _
      (epoll-add! epoll fd events))))

(define (epoll-remove! epoll fd)
  (primitive-epoll-ctl (epoll-fd epoll) EPOLL_CTL_DEL fd))

(define (epoll-wake! epoll)
  "Run after modifying the shared state used by a thread that might be
waiting on this epoll descriptor, to break that thread out of the
epoll wait (if appropriate)."
  (match (atomic-box-ref (epoll-state epoll))
    ;; It is always correct to wake an epoll via the pipe.  However we
  can avoid it if the epoll is guaranteed to see that the
    ;; runqueue is not empty before it goes to poll next time.
    ('waiting
     (primitive-epoll-wake (fileno (epoll-wake-write-pipe epoll))))
    ('not-waiting #t)
    ;; This can happen if a fiber was waiting on a condition and
    ;; run-fibers completes before the fiber completes and afterwards
    ;; the condition is signalled.  In that case, we don't have to
    ;; resurrect the fiber or something, we can just do nothing.
    ;; (Bug report: )
    ('dead #t)))

(define (epoll-default-folder fd events seed)
  (acons fd events seed))

(define (ensure-epoll-eventsv epoll maxevents)
  (let ((prev (epoll-eventsv epoll)))
    (if (and prev
             (or (not maxevents)
                 (= (events-offset maxevents) (bytevector-length prev))))
        prev
        (let ((v (make-bytevector (events-offset (or maxevents 8)))))
          (set-epoll-eventsv! epoll v)
          v))))

(define* (epoll epoll #:key (expiry #f)
                (update-expiry (lambda (expiry) expiry))
                (folder epoll-default-folder) (seed '()))
  (define (expiry->timeout expiry)
    (cond
     ((not expiry) -1)
     (else
      (let ((now (get-internal-real-time)))
        (cond
         ((< expiry now) 0)
         (else (- expiry now)))))))
  (let* ((maxevents (epoll-maxevents epoll))
         (eventsv (ensure-epoll-eventsv epoll maxevents))
         (write-pipe-fd (fileno (epoll-wake-write-pipe epoll)))
         (read-pipe-fd (fileno (epoll-wake-read-pipe epoll))))
    (atomic-box-set! (epoll-state epoll) 'waiting)
    ;; Note: update-expiry call must take place after epoll-state is
    ;; set to waiting.
    (let* ((timeout (expiry->timeout (update-expiry expiry)))
           (n (primitive-epoll-wait (epoll-fd epoll)
                                    write-pipe-fd read-pipe-fd
                                    eventsv timeout)))
      (atomic-box-set! (epoll-state epoll) 'not-waiting)
      ;; If we received `maxevents' events, it means that probably there
      ;; are more active fd's in the queue that we were unable to
      ;; receive.  Expand our event buffer in that case.
      (when (= n maxevents)
        (set-epoll-maxevents! epoll (* maxevents 2)))
      (let lp ((seed seed) (i 0))
        (if (< i n)
            (let ((fd (bytevector-s32-native-ref eventsv (fd-offset i)))
                  (events (bytevector-u32-native-ref eventsv (events-offset i))))
              (lp (folder fd events seed) (1+ i)))
            seed)))))

(define EVENTS_IMPL_READ (logior EPOLLIN EPOLLRDHUP))
(define EVENTS_IMPL_WRITE EPOLLOUT)
(define EVENTS_IMPL_CLOSED_OR_ERROR (logior EPOLLHUP EPOLLERR))

(define events-impl-create epoll-create)

(define events-impl-destroy epoll-destroy)

(define (events-impl? impl)
  (epoll? impl))

(define (events-impl-add! impl fd events)
  (epoll-add*! impl fd (logior events EPOLLONESHOT)))

(define events-impl-wake! epoll-wake!)

(define (events-impl-fd-finalizer impl fd-waiters)
  (lambda (fd)
    ;; When a file port is closed, clear out the list of
  waiting tasks so that when / if this FD is re - used , we
    ;; don't resume stale tasks. Note that we don't need to
  remove the FD from the epoll set , as the kernel manages
    ;; that for us.
    ;;
    ;; FIXME: Is there a way to wake all tasks in a thread-safe
    ;; way?  Note that this function may be invoked from a
    ;; finalizer thread.
    (set-cdr! fd-waiters '())
    (set-car! fd-waiters #f)))

(define events-impl-run epoll)

 epoll

 This library is free software; you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public
 either

 This library 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
 Lesser General Public License for more details.

 along with this program.  If not, see </>.

 atomic box of either 'waiting, 'not-waiting or 'dead
 FIXME: ignore errors flushing output
 It is always correct to wake an epoll via the pipe.  However we
 runqueue is not empty before it goes to poll next time.
 This can happen if a fiber was waiting on a condition and
 run-fibers completes before the fiber completes and afterwards
 the condition is signalled.  In that case, we don't have to
 resurrect the fiber or something, we can just do nothing.
 (Bug report: )
 Note: update-expiry call must take place after epoll-state is
 set to waiting.
 If we received `maxevents' events, it means that probably there
 are more active fd's in the queue that we were unable to
 receive.  Expand our event buffer in that case.
 When a file port is closed, clear out the list of
 don't resume stale tasks. Note that we don't need to
 that for us.

 FIXME: Is there a way to wake all tasks in a thread-safe
 way?  Note that this function may be invoked from a
 finalizer thread.

  Copyright ( C ) 2016   <  >
  Copyright ( C ) 2022   <  >
  Copyright ( C ) 2022   Flaqué <  >
  version 3 of the License , or ( at your option ) any later version .
  You should have received a copy of the GNU Lesser General Public License

(define-module (fibers events-impl)
  #:use-module ((ice-9 binary-ports) #:select (get-u8 put-u8))
  #:use-module (ice-9 atomic)
  #:use-module (ice-9 control)
  #:use-module (ice-9 match)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-9 gnu)
  #:use-module (rnrs bytevectors)
  #:use-module (fibers config)
  #:export (events-impl-create
            events-impl-destroy
            events-impl?
            events-impl-add!
            events-impl-wake!
            events-impl-fd-finalizer
            events-impl-run

            EVENTS_IMPL_READ EVENTS_IMPL_WRITE EVENTS_IMPL_CLOSED_OR_ERROR))

(dynamic-call "init_fibers_epoll"
              (dynamic-link (extension-library "fibers-epoll")))

(when (defined? 'EPOLLRDHUP)
  (export EPOLLRDHUP))
(when (defined? 'EPOLLONESHOT)
  (export EPOLLONESHOT))

(define (make-wake-pipe)
  (let ((pair (pipe2 (logior O_NONBLOCK O_CLOEXEC))))
    (match pair
      ((read-pipe . write-pipe)
       (setvbuf write-pipe 'none)
       (values read-pipe write-pipe)))))

(define-record-type <epoll>
  (make-epoll fd eventsv maxevents state wake-read-pipe wake-write-pipe)
  epoll?
  (fd epoll-fd set-epoll-fd!)
  (eventsv epoll-eventsv set-epoll-eventsv!)
  (maxevents epoll-maxevents set-epoll-maxevents!)
  (state epoll-state)
  (wake-read-pipe epoll-wake-read-pipe)
  (wake-write-pipe epoll-wake-write-pipe))

(define-syntax events-offset
  (lambda (x)
    (syntax-case x ()
      ((_ n)
       #`(* n %sizeof-struct-epoll-event)))))

(define-syntax fd-offset
  (lambda (x)
    (syntax-case x ()
      ((_ n)
       #`(+ (* n %sizeof-struct-epoll-event)
            %offsetof-struct-epoll-event-fd)))))

(define epoll-guardian (make-guardian))
(define (pump-epoll-guardian)
  (let ((epoll (epoll-guardian)))
    (when epoll
      (epoll-destroy epoll)
      (pump-epoll-guardian))))
(add-hook! after-gc-hook pump-epoll-guardian)

(define* (epoll-create #:key (close-on-exec? #t) (maxevents 8))
  (call-with-values (lambda () (make-wake-pipe))
    (lambda (read-pipe write-pipe)
      (let* ((state (make-atomic-box 'not-waiting))
             (epoll (make-epoll (primitive-epoll-create close-on-exec?)
                                #f maxevents state read-pipe write-pipe)))
        (epoll-guardian epoll)
        (epoll-add! epoll (fileno read-pipe) EPOLLIN)
        epoll))))

(define (epoll-destroy epoll)
  (atomic-box-set! (epoll-state epoll) 'dead)
  (when (epoll-fd epoll)
    (close-port (epoll-wake-read-pipe epoll))
    (close-port (epoll-wake-write-pipe epoll))
    (close-fdes (epoll-fd epoll))
    (set-epoll-fd! epoll #f)))

(define (events-impl? impl)
  (epoll? impl))

(define (epoll-add! epoll fd events)
  (primitive-epoll-ctl (epoll-fd epoll) EPOLL_CTL_ADD fd events))

(define (epoll-modify! epoll fd events)
  (primitive-epoll-ctl (epoll-fd epoll) EPOLL_CTL_MOD fd events))

(define (epoll-add*! epoll fd events)
  (catch 'system-error
    (lambda () (epoll-modify! epoll fd events))
    (lambda _
      (epoll-add! epoll fd events))))

(define (epoll-remove! epoll fd)
  (primitive-epoll-ctl (epoll-fd epoll) EPOLL_CTL_DEL fd))

(define (epoll-wake! epoll)
  "Run after modifying the shared state used by a thread that might be
waiting on this epoll descriptor, to break that thread out of the
epoll wait (if appropriate)."
  (match (atomic-box-ref (epoll-state epoll))
  can avoid it if the epoll is guaranteed to see that the
    ('waiting
     (primitive-epoll-wake (fileno (epoll-wake-write-pipe epoll))))
    ('not-waiting #t)
    ('dead #t)))

(define (epoll-default-folder fd events seed)
  (acons fd events seed))

(define (ensure-epoll-eventsv epoll maxevents)
  (let ((prev (epoll-eventsv epoll)))
    (if (and prev
             (or (not maxevents)
                 (= (events-offset maxevents) (bytevector-length prev))))
        prev
        (let ((v (make-bytevector (events-offset (or maxevents 8)))))
          (set-epoll-eventsv! epoll v)
          v))))

(define* (epoll epoll #:key (expiry #f)
                (update-expiry (lambda (expiry) expiry))
                (folder epoll-default-folder) (seed '()))
  (define (expiry->timeout expiry)
    (cond
     ((not expiry) -1)
     (else
      (let ((now (get-internal-real-time)))
        (cond
         ((< expiry now) 0)
         (else (- expiry now)))))))
  (let* ((maxevents (epoll-maxevents epoll))
         (eventsv (ensure-epoll-eventsv epoll maxevents))
         (write-pipe-fd (fileno (epoll-wake-write-pipe epoll)))
         (read-pipe-fd (fileno (epoll-wake-read-pipe epoll))))
    (atomic-box-set! (epoll-state epoll) 'waiting)
    (let* ((timeout (expiry->timeout (update-expiry expiry)))
           (n (primitive-epoll-wait (epoll-fd epoll)
                                    write-pipe-fd read-pipe-fd
                                    eventsv timeout)))
      (atomic-box-set! (epoll-state epoll) 'not-waiting)
      (when (= n maxevents)
        (set-epoll-maxevents! epoll (* maxevents 2)))
      (let lp ((seed seed) (i 0))
        (if (< i n)
            (let ((fd (bytevector-s32-native-ref eventsv (fd-offset i)))
                  (events (bytevector-u32-native-ref eventsv (events-offset i))))
              (lp (folder fd events seed) (1+ i)))
            seed)))))

(define EVENTS_IMPL_READ (logior EPOLLIN EPOLLRDHUP))
(define EVENTS_IMPL_WRITE EPOLLOUT)
(define EVENTS_IMPL_CLOSED_OR_ERROR (logior EPOLLHUP EPOLLERR))

(define events-impl-create epoll-create)

(define events-impl-destroy epoll-destroy)

(define (events-impl? impl)
  (epoll? impl))

(define (events-impl-add! impl fd events)
  (epoll-add*! impl fd (logior events EPOLLONESHOT)))

(define events-impl-wake! epoll-wake!)

(define (events-impl-fd-finalizer impl fd-waiters)
  (lambda (fd)
  waiting tasks so that when / if this FD is re - used , we
  remove the FD from the epoll set , as the kernel manages
    (set-cdr! fd-waiters '())
    (set-car! fd-waiters #f)))

(define events-impl-run epoll)
