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