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