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 &optional timeout))
  34
  35 (defgeneric buffer-flush-output (buffer &optional timeout))
  36
  37 (defgeneric buffer-wait-until-flushable (buffer &optional 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 ((iobuf-lock (input-iobuf-of ,buffer)))
  64                      ,body))
  65                  (:output
  66                   `(bt:with-lock-held ((iobuf-lock (output-iobuf-of ,buffer)))
  67                      ,body))
  68                  (:io
  69                   (make-locks (make-locks body :output) :input)))))
  70       `(flet ((,body-fun () ,@body))
  71          (if (synchronizedp ,buffer)
  72              ,(make-locks `(,body-fun) direction)
  73              (,body-fun))))))
  74 \f
  75
  76 ;;;-----------------------------------------------------------------------------
  77 ;;; Buffer Constructors
  78 ;;;-----------------------------------------------------------------------------
  79
  80 (defmethod initialize-instance :after
  81     ((buffer single-channel-buffer) &key data size)
  82   (with-accessors ((input-iobuf input-iobuf-of)
  83                    (output-iobuf output-iobuf-of))
  84       buffer
  85     (check-type data (or null iobuf))
  86     (setf input-iobuf (or data (make-iobuf size))
  87           output-iobuf input-iobuf)))
  88
  89 (defmethod initialize-instance :after
  90     ((buffer dual-channel-buffer)
  91      &key input-data output-data input-size output-size)
  92   (with-accessors ((input-iobuf input-iobuf-of)
  93                    (output-iobuf output-iobuf-of))
  94       buffer
  95     (check-type input-data (or null iobuf))
  96     (check-type output-data (or null iobuf))
  97     (setf input-iobuf (or input-data (make-iobuf input-size)))
  98     (setf output-iobuf (or output-data (make-iobuf output-size)))))
  99 \f
 100
 101 ;;;-----------------------------------------------------------------------------
 102 ;;; Buffer DEVICE-CLOSE
 103 ;;;-----------------------------------------------------------------------------
 104
 105 (defmethod relinquish ((buffer single-channel-buffer) &key abort)
 106   (with-accessors ((device device-of))
 107       buffer
 108     (with-synchronized-buffer (buffer :input)
 109       (unless (or abort (eql :read (last-io-op-of buffer)))
 110         (%buffer-flush-output buffer 0))
 111       (relinquish device)))
 112   (values buffer))
 113
 114 (defmethod relinquish ((buffer buffer) &key abort)
 115   (with-accessors ((device device-of))
 116       buffer
 117     (with-synchronized-buffer (buffer :io)
 118       (unless abort
 119         (%buffer-flush-output buffer 0))
 120       (relinquish device)))
 121   (values buffer))
 122 \f
 123
 124 ;;;-----------------------------------------------------------------------------
 125 ;;; Buffer DEVICE-READ
 126 ;;;-----------------------------------------------------------------------------
 127
 128 (defmethod device-read ((buffer single-channel-buffer) vector start end
 129                         &optional timeout)
 130   (with-synchronized-buffer (buffer :input)
 131     (%buffer-read-vector buffer vector start end timeout)))
 132
 133 (defmethod device-read ((buffer dual-channel-buffer) vector start end
 134                         &optional timeout)
 135   (with-synchronized-buffer (buffer :input)
 136     (%buffer-read-vector buffer vector start end timeout)))
 137
 138 (defmethod %buffer-read-vector ((buffer buffer) vector start end timeout)
 139   (with-accessors ((input-iobuf input-iobuf-of)
 140                    (output-iobuf output-iobuf-of))
 141       buffer
 142     (cond
 143       ((iobuf-empty-p input-iobuf)
 144        (let ((nbytes
 145               (%buffer-fill-input buffer timeout)))
 146          (if (iobuf-empty-p input-iobuf)
 147              (if (eql :eof nbytes) :eof 0)
 148              (iobuf->vector input-iobuf vector start end))))
 149       (t
 150        (iobuf->vector input-iobuf vector start end)))))
 151 \f
 152
 153 ;;;-----------------------------------------------------------------------------
 154 ;;; Buffer DEVICE-WRITE
 155 ;;;-----------------------------------------------------------------------------
 156
 157 (defmethod device-write ((buffer single-channel-buffer) vector start end
 158                          &optional timeout)
 159   (with-synchronized-buffer (buffer :output)
 160     ;; If the previous operation was a read, flush the read buffer
 161     ;; and reposition the file offset accordingly
 162     (%buffer-clear-input buffer)
 163     (%buffer-write-vector buffer vector start end timeout)))
 164
 165 (defmethod device-write ((buffer dual-channel-buffer) vector start end
 166                          &optional timeout)
 167   (with-synchronized-buffer (buffer :output)
 168     (%buffer-write-vector buffer vector start end timeout)))
 169
 170 (defmethod %buffer-write-vector ((buffer buffer) vector start end timeout)
 171   (with-accessors ((output-iobuf output-iobuf-of))
 172       buffer
 173     (multiple-value-prog1
 174         (vector->iobuf output-iobuf vector start end)
 175       (setf (last-io-op-of buffer) :write)
 176       (when (iobuf-full-p output-iobuf)
 177         (%buffer-flush-output buffer timeout)))))
 178 \f
 179
 180 ;;;-----------------------------------------------------------------------------
 181 ;;; Buffer DEVICE-POSITION
 182 ;;;-----------------------------------------------------------------------------
 183
 184 (defmethod device-position ((buffer single-channel-buffer))
 185   (with-synchronized-buffer (buffer :input)
 186     (%buffer-position buffer)))
 187
 188 (defun %buffer-position (buffer)
 189   (let ((position (device-position (device-of buffer))))
 190     (ecase (last-io-op-of buffer)
 191       (:read
 192        (- position (iobuf-available-octets (input-iobuf-of buffer))))
 193       (:write
 194        (+ position (iobuf-available-octets (output-iobuf-of buffer)))))))
 195
 196 (defmethod (setf device-position) (position (buffer single-channel-buffer) &optional (from :start))
 197   (setf (%buffer-position buffer from) position))
 198
 199 (defun (setf %buffer-position) (position buffer from)
 200   (setf (device-position (device-of buffer) from) position))
 201 \f
 202
 203 ;;;-----------------------------------------------------------------------------
 204 ;;; Buffer CLEAR-INPUT
 205 ;;;-----------------------------------------------------------------------------
 206
 207 (defmethod buffer-clear-input ((buffer single-channel-buffer))
 208   (with-synchronized-buffer (buffer :input)
 209     (%buffer-clear-input buffer)))
 210
 211 (defmethod %buffer-clear-input ((buffer single-channel-buffer))
 212   (when (eql :read (last-io-op-of buffer))
 213     (let ((nbytes (iobuf-available-octets (input-iobuf-of buffer))))
 214       (unless (zerop nbytes)
 215         (setf (%buffer-position buffer :current) (- nbytes)))
 216       (iobuf-reset (input-iobuf-of buffer)))))
 217
 218 (defmethod buffer-clear-input ((buffer buffer))
 219   (with-synchronized-buffer (buffer :input)
 220     (%buffer-clear-input buffer)))
 221
 222 (defmethod %buffer-clear-input ((buffer dual-channel-buffer))
 223   (iobuf-reset (input-iobuf-of buffer)))
 224 \f
 225
 226 ;;;-----------------------------------------------------------------------------
 227 ;;; Buffer CLEAR-OUTPUT
 228 ;;;-----------------------------------------------------------------------------
 229
 230 (defmethod buffer-clear-output ((buffer single-channel-buffer))
 231   (with-synchronized-buffer (buffer :output)
 232     (when (eql :write (last-io-op-of buffer))
 233       (iobuf-reset (output-iobuf-of buffer)))))
 234
 235 (defmethod %buffer-clear-output ((buffer single-channel-buffer))
 236   (when (eql :write (last-io-op-of buffer))
 237     (iobuf-reset (output-iobuf-of buffer))))
 238
 239 (defmethod buffer-clear-output ((buffer dual-channel-buffer))
 240   (with-synchronized-buffer (buffer :output)
 241     (iobuf-reset (output-iobuf-of buffer))))
 242 \f
 243
 244 ;;;-----------------------------------------------------------------------------
 245 ;;; Buffer FILL-INPUT
 246 ;;;-----------------------------------------------------------------------------
 247
 248 (defmethod buffer-fill-input ((buffer single-channel-buffer) &optional timeout)
 249   (with-synchronized-buffer (buffer :input)
 250     (%buffer-clear-output buffer)
 251     (%buffer-fill-input buffer timeout)))
 252
 253 (defmethod buffer-fill-input ((buffer dual-channel-buffer) &optional timeout)
 254   (with-synchronized-buffer (buffer :input)
 255     (%buffer-fill-input buffer timeout)))
 256
 257 (defmethod %buffer-fill-input ((buffer buffer) timeout)
 258   (with-accessors ((device device-of)
 259                    (input-iobuf input-iobuf-of))
 260       buffer
 261     (multiple-value-bind (data start end)
 262         (iobuf-next-empty-zone input-iobuf)
 263       (let ((nbytes
 264              (device-read device data start end timeout)))
 265         (setf (iobuf-end input-iobuf) (+ start nbytes))
 266         (setf (last-io-op-of buffer) :read)
 267         (values nbytes)))))
 268 \f
 269
 270 ;;;-----------------------------------------------------------------------------
 271 ;;; Buffer FLUSH-OUTPUT
 272 ;;;-----------------------------------------------------------------------------
 273
 274 (defmethod buffer-flush-output ((buffer single-channel-buffer) &optional timeout)
 275   (with-synchronized-buffer (buffer :output)
 276     (when (eql :write (last-io-op-of buffer))
 277       (%buffer-flush-output buffer timeout))))
 278
 279 (defmethod buffer-flush-output ((buffer dual-channel-buffer) &optional timeout)
 280   (with-synchronized-buffer (buffer :output)
 281     (%buffer-flush-output buffer timeout)))
 282
 283 (defmethod %buffer-flush-output ((buffer dual-channel-buffer) timeout)
 284   (with-accessors ((device device-of)
 285                    (output-iobuf output-iobuf-of))
 286       buffer
 287     (multiple-value-bind (data start end)
 288         (iobuf-next-data-zone output-iobuf)
 289       (let ((nbytes
 290              (device-write device data start end timeout)))
 291         (setf (iobuf-start output-iobuf) (+ start nbytes))
 292         (setf (last-io-op-of buffer) :write)
 293         (iobuf-available-octets output-iobuf)))))
 294 \f
 295
 296 ;;;-----------------------------------------------------------------------------
 297 ;;; I/O WAIT
 298 ;;;-----------------------------------------------------------------------------
 299
 300 (defmethod buffer-wait-until-flushable ((buffer buffer) &optional timeout)
 301   (device-poll-output (device-of buffer) timeout))