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