io.multiplex/event-loop.lisp

   1 ;;;; -*- Mode: Lisp; Syntax: ANSI-Common-Lisp; indent-tabs-mode: nil -*-
   2 ;;;
   3 ;;; --- Main event loop.
   4 ;;;
   5
   6 (in-package :io.multiplex)
   7
   8 \f
   9 ;;;; EVENT-BASE
  10
  11 (defclass event-base ()
  12   ((mux :initarg :mux
  13         :reader mux-of)
  14    (fds :initform (make-hash-table :test 'eql)
  15         :reader fds-of)
  16    (timers :initform (make-priority-queue :key #'%timer-expire-time)
  17            :reader timers-of)
  18    (fd-timers :initform (make-priority-queue :key #'%timer-expire-time)
  19               :reader fd-timers-of)
  20    (expired-events :initform nil
  21                    :accessor expired-events-of)
  22    (exit :initform nil
  23          :accessor exit-p)
  24    (exit-when-empty :initarg :exit-when-empty
  25                     :accessor exit-when-empty-p))
  26   (:default-initargs :mux (make-instance *default-multiplexer*)
  27                      :exit-when-empty nil)
  28   (:documentation "An event base ..."))
  29
  30 (defmacro with-event-base ((var &rest initargs) &body body)
  31   "Binds VAR to a new EVENT-BASE, instantiated with INITARGS,
  32 within the extent of BODY.  Closes VAR."
  33   `(let ((,var (make-instance 'event-base ,@initargs)))
  34      (unwind-protect
  35           (locally ,@body)
  36        (when ,var (close ,var)))))
  37
  38 (defmethod print-object ((base event-base) stream)
  39   (print-unreadable-object (base stream :type nil :identity t)
  40     (if (fds-of base)
  41         (format stream "event base, ~A FDs monitored, using: ~A"
  42                 (hash-table-count (fds-of base)) (mux-of base))
  43         (format stream "event base, closed"))))
  44
  45 (defmethod initialize-instance :after ((base event-base) &key)
  46   (with-slots (mux) base
  47     (when (symbolp mux)
  48       (setf mux (make-instance mux)))))
  49
  50 ;;; KLUDGE: CLOSE is for streams. --luis
  51 ;;;
  52 ;;; Also, we might want to close FDs here.  Or have a version/argument
  53 ;;; that handles that.  Or... add finalizers to the fd streams.
  54 (defmethod close ((event-base event-base) &key abort)
  55   (declare (ignore abort))
  56   (with-accessors ((mux mux-of)) event-base
  57     (close-multiplexer mux)
  58     (dolist (slot '(mux fds timers fd-timers expired-events))
  59       (setf (slot-value event-base slot) nil))
  60     (values event-base)))
  61
  62 (defgeneric add-fd (base fd event-type function &key timeout one-shot)
  63   (:documentation ""))
  64
  65 (defgeneric add-timer (event-base function timeout &key one-shot)
  66   (:documentation ""))
  67
  68 (defgeneric remove-event (event-base event)
  69   (:documentation ""))
  70
  71 (defgeneric remove-fd (event-base fd)
  72   (:documentation ""))
  73
  74 (defgeneric event-dispatch (event-base &key one-shot timeout &allow-other-keys)
  75   (:documentation ""))
  76
  77 (defgeneric exit-event-loop (event-base &key delay)
  78   (:documentation "")
  79   (:method ((event-base event-base) &key (delay 0))
  80     (add-timer event-base
  81                (lambda () (setf (exit-p event-base) t))
  82                delay :one-shot t)))
  83
  84 (defgeneric event-base-empty-p (event-base)
  85   (:documentation "Return T if no FD event or timeout is registered with EVENT-BASE.")
  86   (:method ((event-base event-base))
  87     (and (zerop (hash-table-count (fds-of event-base)))
  88          (priority-queue-empty-p (timers-of event-base)))))
  89 \f
  90 ;;;;;;;;;;;;;;;;;
  91 ;;; Utilities ;;;
  92 ;;;;;;;;;;;;;;;;;
  93
  94 (defun fd-entry-of (event-base fd)
  95   "Return the FD-ENTRY of FD in EVENT-BASE."
  96   (gethash fd (fds-of event-base)))
  97
  98 (defun (setf fd-entry-of) (fd-entry event-base fd)
  99   "Return the FD-ENTRY of FD in EVENT-BASE."
 100   (setf (gethash fd (fds-of event-base)) fd-entry))
 101
 102 (defun remove-fd-entry (event-base fd)
 103   "Remove the FD-ENTRY of FD from EVENT-BASE."
 104   (remhash fd (fds-of event-base)))
 105 \f
 106 ;;;;;;;;;;;;;;;;;
 107 ;;; Internals ;;;
 108 ;;;;;;;;;;;;;;;;;
 109
 110 (defun expire-event (event-base event)
 111   (push event (expired-events-of event-base)))
 112
 113 (defun %add-fd-timer (event-base timer)
 114   (schedule-timer (fd-timers-of event-base) timer))
 115
 116 (defun %add-fd (event-base event fd-entry timeout)
 117   (with-accessors ((fd-timers fd-timers-of)) event-base
 118     (let ((fd (fd-event-fd event)))
 119       (when timeout
 120         (let ((timer (make-timer (lambda () (expire-event event-base event))
 121                                  timeout)))
 122           (setf (fd-event-timer event) timer)
 123           (%add-fd-timer event-base timer)))
 124       (setf (fd-entry-event fd-entry (fd-event-type event)) event)
 125       (setf (fd-entry-of event-base fd) fd-entry)
 126       (values event))))
 127
 128 (defmethod add-fd :before ((event-base event-base) fd event-type function
 129                            &key timeout one-shot)
 130   (declare (ignore timeout))
 131   (check-type fd unsigned-byte)
 132   (check-type event-type fd-event-type)
 133   (check-type function (or symbol function))
 134   (check-type one-shot boolean)
 135   (let ((fd-limit (fd-limit-of (mux-of event-base))))
 136     (when (and fd-limit (> fd fd-limit))
 137       (error "Cannot add such a large FD: ~A" fd))))
 138
 139 (defmethod add-fd ((event-base event-base) fd event-type function
 140                    &key timeout one-shot)
 141   (let ((current-entry (fd-entry-of event-base fd))
 142         (event (make-event fd event-type function one-shot)))
 143     (cond (current-entry
 144            (assert (null (fd-entry-event current-entry event-type))
 145                    ((fd-entry-event current-entry event-type))
 146                    "FD ~A is already monitored for event ~A" fd event-type)
 147            (%add-fd event-base event current-entry timeout)
 148            (update-fd (mux-of event-base) current-entry event-type :add))
 149           (t
 150            (let ((new-fd-entry (make-fd-entry fd)))
 151              (%add-fd event-base event new-fd-entry timeout)
 152              (monitor-fd (mux-of event-base) new-fd-entry))))
 153     (values event)))
 154
 155 (defun %add-timer (event-base timer)
 156   (schedule-timer (timers-of event-base) timer))
 157
 158 (defmethod add-timer ((event-base event-base) function
 159                       timeout &key one-shot)
 160   (check-type function (or symbol function))
 161   (check-type one-shot boolean)
 162   (%add-timer event-base (make-timer function timeout :one-shot one-shot)))
 163
 164 (defun %remove-fd-timer (event-base timer)
 165   (unschedule-timer (fd-timers-of event-base) timer))
 166
 167 (defun %remove-fd (event-base event)
 168   (with-accessors ((timers timers-of)) event-base
 169     (let* ((fd (fd-event-fd event))
 170            (fd-entry (fd-entry-of event-base fd)))
 171       (assert fd-entry (fd-entry) "FD ~A does not have an FD-ENTRY" fd)
 172       (setf (fd-entry-event fd-entry (fd-event-type event)) nil)
 173       (when-let (timer (fd-event-timer event))
 174         (%remove-fd-timer event-base timer))
 175       (when (fd-entry-empty-p fd-entry)
 176         (remove-fd-entry event-base fd))
 177       (values event))))
 178
 179 (defun %remove-fd-event (event-base event)
 180   (let* ((fd (fd-event-fd event))
 181          (current-entry (fd-entry-of event-base fd)))
 182     (cond (current-entry
 183            (%remove-fd event-base event)
 184            (if (fd-entry-empty-p current-entry)
 185                (unmonitor-fd (mux-of event-base) current-entry)
 186                (update-fd (mux-of event-base) current-entry
 187                           (fd-event-type event) :del)))
 188           (t
 189            (%remove-fd event-base event)))))
 190
 191 (defun %remove-timer (event-base timer)
 192   (unschedule-timer (timers-of event-base) timer))
 193
 194 (defmethod remove-event ((event-base event-base) event)
 195   (etypecase event
 196     (timer    (%remove-timer event-base event))
 197     (fd-event (%remove-fd-event event-base event)))
 198   (values event-base))
 199
 200 (defun remove-events (event-base event-list)
 201   (dolist (ev event-list)
 202     (remove-event event-base ev)))
 203
 204 (defmethod remove-fd ((event-base event-base) fd)
 205   (let ((entry (fd-entry-of event-base fd)))
 206     (symbol-macrolet ((rev (fd-entry-read-event entry))
 207                       (wev (fd-entry-write-event entry))
 208                       (eev (fd-entry-error-event entry)))
 209       (labels ((maybe-remove-timer (event)
 210                  (when (and event (fd-event-timer event))
 211                    (%remove-fd-timer event-base (fd-event-timer event))))
 212                (maybe-remove-all-timers ()
 213                  (maybe-remove-timer rev)
 214                  (maybe-remove-timer wev)
 215                  (maybe-remove-timer eev)))
 216         (cond (entry
 217                (maybe-remove-all-timers)
 218                (unmonitor-fd (mux-of event-base) fd)
 219                (remove-fd-entry event-base fd))
 220               (t (warn "Trying to remove an unmonitored FD.")))))))
 221
 222 (defvar *maximum-event-loop-timeout* 1)
 223
 224 (defmethod event-dispatch :around ((event-base event-base)
 225                                    &key timeout one-shot)
 226   (declare (ignore one-shot))
 227   (setf (exit-p event-base) nil)
 228   (when timeout
 229     (exit-event-loop event-base :delay timeout))
 230   (call-next-method))
 231
 232 (defmethod event-dispatch ((event-base event-base) &key one-shot timeout
 233                            (max-timeout *maximum-event-loop-timeout*))
 234   (declare (ignore timeout))
 235   (with-accessors ((mux mux-of) (fds fds-of) (exit-p exit-p)
 236                    (exit-when-empty exit-when-empty-p)
 237                    (timers timers-of) (fd-timers fd-timers-of)
 238                    (expired-events expired-events-of))
 239       event-base
 240     (flet ((poll-timeout ()
 241              (min-timeout (time-to-next-timer timers)
 242                           (time-to-next-timer fd-timers)
 243                           max-timeout)))
 244       (do ((deletion-list () ())
 245            (got-fd-events-p nil nil)
 246            (got-fd-timeouts-p nil nil)
 247            (got-timers-p nil nil)
 248            (poll-timeout (poll-timeout) (poll-timeout))
 249            (now (osicat-sys:get-monotonic-time) (osicat-sys:get-monotonic-time)))
 250           ((or exit-p (and exit-when-empty (event-base-empty-p event-base))))
 251         (setf (expired-events-of event-base) nil)
 252         (setf (values got-fd-events-p deletion-list)
 253               (dispatch-fd-events-once event-base poll-timeout now))
 254         (remove-events event-base deletion-list)
 255         (setf got-fd-timeouts-p (expire-pending-timers fd-timers now))
 256         (dispatch-fd-timeouts expired-events)
 257         (setf got-timers-p (expire-pending-timers timers now))
 258         (when (and (or got-fd-events-p got-fd-timeouts-p got-timers-p)
 259                    one-shot)
 260           (setf exit-p t))))))
 261
 262 ;;; Waits for events and dispatches them.  Returns T if some events
 263 ;;; have been received, NIL otherwise.
 264 (defun dispatch-fd-events-once (event-base timeout now)
 265   (with-accessors ((mux mux-of) (fds fds-of) (fd-timers fd-timers-of))
 266       event-base
 267     (let ((deletion-list ())
 268           (fd-events (harvest-events mux timeout)))
 269       (dolist (ev fd-events)
 270         (destructuring-bind (fd ev-types) ev
 271           (let* ((fd-entry (fd-entry-of event-base fd))
 272                  (errorp (and fd-entry (member :error ev-types))))
 273             (labels ((append-events (events)
 274                        (nconcf deletion-list events))
 275                      (do-error ()
 276                        (%dispatch-event fd-entry :error now)
 277                        (append-events (fd-entry-all-events fd-entry)))
 278                      (do-read ()
 279                        (let ((events (%dispatch-event fd-entry :read now)))
 280                          (or errorp (append-events events))))
 281                      (do-write ()
 282                        (let ((events (%dispatch-event fd-entry :write now)))
 283                          (or errorp (append-events events)))))
 284               (cond (fd-entry
 285                      (when errorp (do-error))
 286                      (when (member :read ev-types) (do-read))
 287                      (when (member :write ev-types) (do-write)))
 288                     (t
 289                      (warn "Got spurious event for non-monitored FD: ~A" fd)))))))
 290       (priority-queue-reorder fd-timers)
 291       (values (consp fd-events) deletion-list))))
 292
 293 (defun %dispatch-event (fd-entry event-type now)
 294   (let ((deletion-list ())
 295         (ev (fd-entry-event fd-entry event-type)))
 296     (funcall (fd-event-handler ev) (fd-entry-fd fd-entry) event-type)
 297     (when-let (timer (fd-event-timer ev))
 298       (reschedule-timer-relative-to-now timer now))
 299     (when (fd-event-one-shot-p ev) (push ev deletion-list))
 300     (values deletion-list)))
 301
 302 (defun dispatch-fd-timeouts (events)
 303   (dolist (ev events)
 304     (funcall (fd-event-handler ev)
 305              (fd-event-fd ev)
 306              :timeout)))