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