io.streams/zeta/buffer.lisp

   1 ;;;; -*- Mode: Lisp; Syntax: ANSI-Common-Lisp; indent-tabs-mode: nil -*-
   2 ;;;
   3 ;;; --- Device buffers.
   4 ;;;
   5
   6 (in-package :io.zeta-streams)
   7
   8 ;;;-------------------------------------------------------------------------
   9 ;;; Classes and Types
  10 ;;;-------------------------------------------------------------------------
  11
  12 (defclass buffer ()
  13   ((synchronized :initarg :synchronized
  14                  :reader synchronizedp)
  15    (device :initform nil
  16            :initarg :device
  17            :accessor device-of)
  18    (input-iobuf :initarg :input-buffer
  19                 :accessor input-iobuf-of)
  20    (output-iobuf :initarg :output-buffer
  21                  :accessor output-iobuf-of))
  22   (:default-initargs :synchronized nil))
  23
  24 (defclass single-channel-buffer (buffer)
  25   ((last-io-op :initform nil :accessor last-io-op-of)))
  26
  27 (defclass dual-channel-buffer (buffer) ())
  28 \f
  29
  30 ;;;-------------------------------------------------------------------------
  31 ;;; Generic Functions
  32 ;;;-------------------------------------------------------------------------
  33
  34 (defgeneric buffer-fill (buffer &key timeout))
  35
  36 (defgeneric buffer-flush (buffer &key timeout))
  37
  38 (defgeneric buffer-wait-until-flushable (buffer &key timeout))
  39
  40 (defgeneric buffer-clear-input (buffer))
  41
  42 (defgeneric buffer-clear-output (buffer))
  43
  44 ;;; Internal functions
  45
  46 (defgeneric %buffer-read-vector (buffer vector start end timeout))
  47
  48 (defgeneric %buffer-write-vector (buffer vector start end timeout))
  49
  50 (defgeneric %buffer-fill (buffer timeout))
  51
  52 (defgeneric %buffer-flush (buffer timeout))
  53
  54 (defgeneric %buffer-clear-input (buffer))
  55
  56 (defgeneric %buffer-clear-output (buffer))
  57 \f
  58
  59 ;;;-------------------------------------------------------------------------
  60 ;;; Helper macros
  61 ;;;-------------------------------------------------------------------------
  62
  63 (defmacro with-synchronized-buffer ((buffer &optional direction) &body body)
  64   (with-gensyms (body-fun)
  65     (labels ((make-locks (body direction)
  66                (ecase direction
  67                  (:input
  68                   `(bt:with-lock-held
  69                        ((iobuf-lock (input-iobuf-of ,buffer)))
  70                      ,body))
  71                  (:output
  72                   `(bt:with-lock-held
  73                        ((iobuf-lock (output-iobuf-of ,buffer)))
  74                      ,body))
  75                  (:io
  76                   (make-locks (make-locks body :output) :input)))))
  77       `(flet ((,body-fun () ,@body))
  78          (if (synchronizedp ,buffer)
  79              ,(make-locks `(,body-fun) direction)
  80              (,body-fun))))))
  81 \f
  82
  83 ;;;-------------------------------------------------------------------------
  84 ;;; Constructors
  85 ;;;-------------------------------------------------------------------------
  86
  87 (defmethod initialize-instance :after
  88     ((buffer single-channel-buffer) &key data size)
  89   (with-accessors ((device device-of)
  90                    (input-iobuf input-iobuf-of)
  91                    (output-iobuf output-iobuf-of))
  92       buffer
  93     (check-type device device)
  94     (check-type data (or null iobuf))
  95     (setf input-iobuf (or data (make-iobuf size))
  96           output-iobuf input-iobuf)))
  97
  98 (defmethod initialize-instance :after
  99     ((buffer dual-channel-buffer)
 100      &key input-data output-data input-size output-size)
 101   (with-accessors ((device device-of)
 102                    (input-iobuf input-iobuf-of)
 103                    (output-iobuf output-iobuf-of))
 104       buffer
 105     (check-type device device)
 106     (check-type input-data (or null iobuf))
 107     (check-type output-data (or null iobuf))
 108     (setf input-iobuf (or input-data (make-iobuf input-size)))
 109     (setf output-iobuf (or output-data (make-iobuf output-size)))))
 110 \f
 111
 112 ;;;-------------------------------------------------------------------------
 113 ;;; RELINQUISH
 114 ;;;-------------------------------------------------------------------------
 115
 116 (defmethod relinquish ((buffer single-channel-buffer) &key abort)
 117   (with-accessors ((device device-of))
 118       buffer
 119     (with-synchronized-buffer (buffer :input)
 120       (unless (or abort (eql :read (last-io-op-of buffer)))
 121         (%buffer-flush buffer 0))
 122       (relinquish device)))
 123   (values buffer))
 124
 125 (defmethod relinquish ((buffer dual-channel-buffer) &key abort)
 126   (with-accessors ((device device-of))
 127       buffer
 128     (with-synchronized-buffer (buffer :io)
 129       (unless abort
 130         (%buffer-flush buffer 0))
 131       (relinquish device)))
 132   (values buffer))
 133 \f
 134
 135 ;;;-------------------------------------------------------------------------
 136 ;;; DEVICE-READ
 137 ;;;-------------------------------------------------------------------------
 138
 139 (defmethod device-read :around ((buffer buffer) vector &key
 140                                 (start 0) end timeout)
 141   (check-bounds vector start end)
 142   (if (= start end)
 143       0
 144       (call-next-method buffer vector :start start
 145                         :end end :timeout timeout)))
 146
 147 (defmethod device-read ((buffer single-channel-buffer) vector
 148                         &key start end timeout)
 149   (with-synchronized-buffer (buffer :input)
 150     (%buffer-read-vector buffer vector start end timeout)))
 151
 152 (defmethod device-read ((buffer dual-channel-buffer) vector
 153                         &key start end timeout)
 154   (with-synchronized-buffer (buffer :input)
 155     (%buffer-read-vector buffer vector start end timeout)))
 156
 157 (defmethod %buffer-read-vector ((buffer buffer) vector start end timeout)
 158   (with-accessors ((input-iobuf input-iobuf-of)
 159                    (output-iobuf output-iobuf-of))
 160       buffer
 161     (cond
 162       ((iobuf-empty-p input-iobuf)
 163        (let ((nbytes (%buffer-fill buffer timeout)))
 164          (if (iobuf-empty-p input-iobuf)
 165              (if (eql :eof nbytes) :eof 0)
 166              (iobuf->vector input-iobuf vector start end))))
 167       (t
 168        (iobuf->vector input-iobuf vector start end)))))
 169 \f
 170
 171 ;;;-------------------------------------------------------------------------
 172 ;;; DEVICE-WRITE
 173 ;;;-------------------------------------------------------------------------
 174
 175 (defmethod device-write :around ((buffer buffer) vector
 176                                  &key (start 0) end timeout)
 177   (check-bounds vector start end)
 178   (if (= start end)
 179       0
 180       (call-next-method buffer vector :start start
 181                         :end end :timeout timeout)))
 182
 183 (defmethod device-write ((buffer single-channel-buffer) vector
 184                          &key start end timeout)
 185   (with-synchronized-buffer (buffer :output)
 186     ;; If the previous operation was a read, flush the read buffer
 187     ;; and reposition the file offset accordingly
 188     (%buffer-clear-input buffer)
 189     (%buffer-write-vector buffer vector start end timeout)))
 190
 191 (defmethod device-write ((buffer dual-channel-buffer) vector
 192                          &key start end timeout)
 193   (with-synchronized-buffer (buffer :output)
 194     (%buffer-write-vector buffer vector start end timeout)))
 195
 196 (defmethod %buffer-write-vector ((buffer buffer) vector start end timeout)
 197   (with-accessors ((output-iobuf output-iobuf-of))
 198       buffer
 199     (multiple-value-prog1
 200         (vector->iobuf output-iobuf vector start end)
 201       (setf (last-io-op-of buffer) :write)
 202       (when (iobuf-full-p output-iobuf)
 203         (%buffer-flush buffer timeout)))))
 204 \f
 205
 206 ;;;-------------------------------------------------------------------------
 207 ;;; DEVICE-POSITION
 208 ;;;-------------------------------------------------------------------------
 209
 210 (defmethod device-position ((buffer single-channel-buffer))
 211   (with-synchronized-buffer (buffer :input)
 212     (%buffer-position buffer)))
 213
 214 (defun %buffer-position (buffer)
 215   (let ((position (device-position (device-of buffer))))
 216     (ecase (last-io-op-of buffer)
 217       (:read
 218        (- position (iobuf-available-octets (input-iobuf-of buffer))))
 219       (:write
 220        (+ position (iobuf-available-octets (output-iobuf-of buffer)))))))
 221
 222 (defmethod (setf device-position)
 223     (position (buffer single-channel-buffer) &optional (from :start))
 224   (setf (%buffer-position buffer from) position))
 225
 226 (defun (setf %buffer-position) (position buffer from)
 227   (setf (device-position (device-of buffer) from) position))
 228 \f
 229
 230 ;;;-------------------------------------------------------------------------
 231 ;;; CLEAR-INPUT
 232 ;;;-------------------------------------------------------------------------
 233
 234 (defmethod buffer-clear-input ((buffer single-channel-buffer))
 235   (with-synchronized-buffer (buffer :input)
 236     (%buffer-clear-input buffer)))
 237
 238 (defmethod %buffer-clear-input ((buffer single-channel-buffer))
 239   (when (eql :read (last-io-op-of buffer))
 240     (let ((nbytes (iobuf-available-octets (input-iobuf-of buffer))))
 241       (unless (zerop nbytes)
 242         (setf (%buffer-position buffer :current) (- nbytes)))
 243       (iobuf-reset (input-iobuf-of buffer)))))
 244
 245 (defmethod buffer-clear-input ((buffer buffer))
 246   (with-synchronized-buffer (buffer :input)
 247     (%buffer-clear-input buffer)))
 248
 249 (defmethod %buffer-clear-input ((buffer dual-channel-buffer))
 250   (iobuf-reset (input-iobuf-of buffer)))
 251 \f
 252
 253 ;;;-------------------------------------------------------------------------
 254 ;;; CLEAR-OUTPUT
 255 ;;;-------------------------------------------------------------------------
 256
 257 (defmethod buffer-clear-output ((buffer single-channel-buffer))
 258   (with-synchronized-buffer (buffer :output)
 259     (%buffer-clear-output buffer)))
 260
 261 (defmethod %buffer-clear-output ((buffer single-channel-buffer))
 262   (when (eql :write (last-io-op-of buffer))
 263     (iobuf-reset (output-iobuf-of buffer))))
 264
 265 (defmethod buffer-clear-output ((buffer dual-channel-buffer))
 266   (with-synchronized-buffer (buffer :output)
 267     (iobuf-reset (output-iobuf-of buffer))))
 268 \f
 269
 270 ;;;-------------------------------------------------------------------------
 271 ;;; FILL-INPUT
 272 ;;;-------------------------------------------------------------------------
 273
 274 (defmethod buffer-fill ((buffer single-channel-buffer) &key timeout)
 275   (with-synchronized-buffer (buffer :input)
 276     (%buffer-clear-output buffer)
 277     (%buffer-fill buffer timeout)))
 278
 279 (defmethod buffer-fill ((buffer dual-channel-buffer) &key timeout)
 280   (with-synchronized-buffer (buffer :input)
 281     (%buffer-fill buffer timeout)))
 282
 283 (defmethod %buffer-fill ((buffer buffer) timeout)
 284   (with-accessors ((device device-of)
 285                    (input-iobuf input-iobuf-of))
 286       buffer
 287     (multiple-value-bind (data start end)
 288         (iobuf-next-empty-zone input-iobuf)
 289       (let ((nbytes
 290              (device-read device data :start start
 291                           :end end :timeout timeout)))
 292         (setf (iobuf-end input-iobuf) (+ start nbytes))
 293         (setf (last-io-op-of buffer) :read)
 294         (values nbytes)))))
 295 \f
 296
 297 ;;;-------------------------------------------------------------------------
 298 ;;; FLUSH-OUTPUT
 299 ;;;-------------------------------------------------------------------------
 300
 301 (defmethod buffer-flush ((buffer single-channel-buffer) &key timeout)
 302   (with-synchronized-buffer (buffer :output)
 303     (when (eql :write (last-io-op-of buffer))
 304       (%buffer-flush buffer timeout))))
 305
 306 (defmethod buffer-flush ((buffer dual-channel-buffer) &key timeout)
 307   (with-synchronized-buffer (buffer :output)
 308     (%buffer-flush buffer timeout)))
 309
 310 (defmethod %buffer-flush ((buffer buffer) timeout)
 311   (with-accessors ((device device-of)
 312                    (output-iobuf output-iobuf-of))
 313       buffer
 314     (multiple-value-bind (data start end)
 315         (iobuf-next-data-zone output-iobuf)
 316       (let ((nbytes
 317              (device-write device data :start start
 318                            :end end :timeout timeout)))
 319         (setf (iobuf-start output-iobuf) (+ start nbytes))
 320         (setf (last-io-op-of buffer) :write)
 321         (iobuf-available-octets output-iobuf)))))
 322 \f
 323
 324 ;;;-------------------------------------------------------------------------
 325 ;;; I/O WAIT
 326 ;;;-------------------------------------------------------------------------
 327
 328 (defmethod buffer-wait-until-flushable ((buffer buffer) &key timeout)
 329   (device-poll-output (device-of buffer) :timeout timeout))