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 (device)
  13   ((single-channel-p :initarg :single-channel :accessor single-channel-buffer-p)
  14    (last-io-op :initform nil :accessor last-io-op-of)
  15    (input-buffer :initarg :input-buffer :accessor input-buffer-of)
  16    (output-buffer :initarg :output-buffer :accessor output-buffer-of)
  17    (synchronized :initarg :synchronized :reader buffer-synchronized-p))
  18   (:default-initargs :input-buffer nil
  19                      :output-buffer nil
  20                      :synchronized nil))
  21 \f
  22
  23 ;;;-----------------------------------------------------------------------------
  24 ;;; Buffer Constructors
  25 ;;;-----------------------------------------------------------------------------
  26
  27 (defmethod initialize-instance :after ((buffer buffer) &key
  28                                        (single-channel nil single-channel-provided)
  29                                        input-buffer-size output-buffer-size)
  30   (with-accessors ((single-channel-p single-channel-buffer-p)
  31                    (input-handle input-handle-of)
  32                    (input-buffer input-buffer-of)
  33                    (output-buffer output-buffer-of))
  34       buffer
  35     (unless single-channel-provided
  36       (setf single-channel-p (typep input-handle 'single-channel-device)))
  37     (if input-buffer
  38         (check-type input-buffer iobuf)
  39         (setf input-buffer (make-iobuf input-buffer-size)))
  40     (if single-channel
  41         (setf output-buffer input-buffer)
  42         (cond
  43           (output-buffer
  44            (check-type output-buffer iobuf)
  45            (assert (not (eq input-buffer output-buffer))))
  46           (t (setf output-buffer (make-iobuf output-buffer-size)))))))
  47 \f
  48
  49 ;;;-----------------------------------------------------------------------------
  50 ;;; Buffer Generic Functions
  51 ;;;-----------------------------------------------------------------------------
  52
  53 (defgeneric buffer-clear-input (buffer))
  54
  55 (defgeneric buffer-clear-output (buffer))
  56
  57 (defgeneric buffer-flush-output (buffer &optional timeout))
  58 \f
  59
  60 ;;;-----------------------------------------------------------------------------
  61 ;;; Buffer DEVICE-READ
  62 ;;;-----------------------------------------------------------------------------
  63
  64 (defmethod device-read ((device buffer) buffer start end &optional timeout)
  65   (when (= start end) (return-from device-read 0))
  66   (cond
  67     ((buffer-synchronized-p device)
  68      (flet ((%read-octets ()
  69               (bt:with-lock-held ((iobuf-lock (input-buffer-of device)))
  70                 (read-octets/buffered device buffer start end 0))))
  71        (let ((nbytes (%read-octets)))
  72          (cond
  73            ((and (not (eql timeout 0))
  74                  (eql nbytes 0))
  75             (wait-for-input (input-handle-of device) timeout)
  76             (%read-octets))
  77            (t nbytes)))))
  78     (t
  79      (read-octets/buffered device buffer start end timeout))))
  80
  81 (defun read-octets/buffered (device vector start end timeout)
  82   (declare (type buffer device)
  83            (type ub8-simple-vector vector)
  84            (type iobuf-index start end)
  85            (type device-timeout timeout))
  86   (with-accessors ((input-handle input-handle-of)
  87                    (input-buffer input-buffer-of)
  88                    (output-handle output-handle-of)
  89                    (output-buffer output-buffer-of))
  90       device
  91     ;; If the previous operation was a write, try to flush the output buffer.
  92     ;; If the buffer couldn't be flushed entirely, signal an error
  93     (synchronize-input device output-handle output-buffer)
  94     (cond
  95       ((iobuf-empty-p input-buffer)
  96        (let ((nbytes
  97               (fill-input-buffer device input-handle input-buffer timeout)))
  98          (if (iobuf-empty-p input-buffer)
  99              (if (eql :eof nbytes) :eof 0)
 100              (iobuf->vector input-buffer vector start end))))
 101       (t
 102        (iobuf->vector input-buffer vector start end)))))
 103
 104 (defun synchronize-input (device output-handle output-buffer)
 105   (when (and (single-channel-buffer-p device)
 106              (eql :write (last-io-op-of device)))
 107     (if (plusp (flush-output-buffer output-handle output-buffer 0))
 108         (error "Could not flush the entire write buffer !")
 109         (iobuf-reset output-buffer))))
 110
 111 (defun fill-input-buffer (device input-handle input-buffer timeout)
 112   (multiple-value-bind (data start end)
 113       (iobuf-next-empty-zone input-buffer)
 114     (let ((nbytes
 115            (device-read input-handle data start end timeout)))
 116       (setf (iobuf-end input-buffer) (+ start nbytes))
 117       (setf (last-io-op-of device) :read)
 118       (values nbytes))))
 119
 120 (defun flush-input-buffer (input-buffer)
 121   (prog1
 122       (iobuf-available-octets input-buffer)
 123     (iobuf-reset input-buffer)))
 124 \f
 125
 126 ;;;-----------------------------------------------------------------------------
 127 ;;; Buffer DEVICE-WRITE
 128 ;;;-----------------------------------------------------------------------------
 129
 130 (defmethod device-write ((device buffer) buffer start end &optional timeout)
 131   (when (= start end) (return-from device-write 0))
 132   (cond
 133     ((buffer-synchronized-p device)
 134      (flet ((%write-octets ()
 135               (bt:with-lock-held ((iobuf-lock (output-buffer-of device)))
 136                 (write-octets/buffered device buffer start end 0))))
 137        (let ((nbytes (%write-octets)))
 138          (cond
 139            ((and (not (eql timeout 0))
 140                  (eql nbytes 0))
 141             (wait-for-output (output-handle-of device) timeout)
 142             (%write-octets))
 143            (t nbytes)))))
 144     (t
 145      (write-octets/buffered device buffer start end timeout))))
 146
 147 (defun write-octets/buffered (device vector start end timeout)
 148   (declare (type buffer device)
 149            (type ub8-simple-vector vector)
 150            (type iobuf-index start end)
 151            (type device-timeout timeout))
 152   (with-accessors ((output-handle output-handle-of)
 153                    (output-buffer output-buffer-of))
 154       device
 155     ;; If the previous operation was a read, flush the read buffer
 156     ;; and reposition the file offset accordingly
 157     (synchronize-output device)
 158     (prog1
 159         (vector->iobuf output-buffer vector start end)
 160       (setf (last-io-op-of device) :write)
 161       (when (iobuf-full-p output-buffer)
 162         (flush-output-buffer output-handle output-buffer timeout)))))
 163
 164 (defun synchronize-output (device)
 165   (when (and (single-channel-buffer-p device)
 166              (eql :read (last-io-op-of device)))
 167     (let ((nbytes (flush-input-buffer (input-buffer-of device))))
 168       (unless (zerop nbytes)
 169         (setf (device-position device :from :current) (- nbytes))))))
 170
 171 (defun flush-output-buffer (output-handle output-buffer timeout)
 172   (multiple-value-bind (data start end)
 173       (iobuf-next-data-zone output-buffer)
 174     (let ((nbytes
 175            (device-write output-handle data start end timeout)))
 176       (setf (iobuf-start output-buffer) (+ start nbytes))))
 177   (iobuf-available-octets output-buffer))
 178 \f
 179
 180 ;;;-----------------------------------------------------------------------------
 181 ;;; Buffer DEVICE-POSITION
 182 ;;;-----------------------------------------------------------------------------
 183
 184 (defmethod device-position ((device buffer))
 185   (when-let ((handle-position
 186               (device-position (input-handle-of device))))
 187     (ecase (last-io-op-of device)
 188       (:read
 189        (- handle-position (iobuf-available-octets (input-buffer-of device))))
 190       (:write
 191        (+ handle-position (iobuf-available-octets (output-buffer-of device)))))))
 192
 193 (defmethod (setf device-position) (position (device buffer) &key (from :start))
 194   (setf (device-position device :from from) position))
 195 \f
 196
 197 ;;;-----------------------------------------------------------------------------
 198 ;;; Buffer cleaning
 199 ;;;-----------------------------------------------------------------------------
 200
 201 (defmethod buffer-clear-input ((buffer buffer))
 202   (iobuf-reset (input-buffer-of buffer)))
 203
 204 (defmethod buffer-clear-output ((buffer buffer))
 205   (iobuf-reset (output-buffer-of buffer)))
 206
 207 (defmethod buffer-flush-output ((buffer buffer) &optional timeout)
 208   (with-accessors ((output-handle output-handle-of)
 209                    (output-buffer output-buffer-of))
 210       buffer
 211     (flush-output-buffer output-handle output-buffer timeout)))