+;;;; CORE OUTPUT FUNCTIONS
+
+;;; Buffer the section of THING delimited by START and END by copying
+;;; to output buffer(s) of stream.
+(defun buffer-output (stream thing start end)
+ (declare (index start end))
+ (when (< end start)
+ (error ":END before :START!"))
+ (when (> end start)
+ ;; Copy bytes from THING to buffers.
+ (flet ((copy-to-buffer (buffer tail count)
+ (declare (buffer buffer) (index tail count))
+ (aver (plusp count))
+ (let ((sap (buffer-sap buffer)))
+ (etypecase thing
+ (system-area-pointer
+ (system-area-ub8-copy thing start sap tail count))
+ ((simple-unboxed-array (*))
+ (copy-ub8-to-system-area thing start sap tail count))))
+ ;; Not INCF! If another thread has moved tail from under
+ ;; us, we don't want to accidentally increment tail
+ ;; beyond buffer-length.
+ (setf (buffer-tail buffer) (+ count tail))
+ (incf start count)))
+ (tagbody
+ ;; First copy is special: the buffer may already contain
+ ;; something, or be even full.
+ (let* ((obuf (fd-stream-obuf stream))
+ (tail (buffer-tail obuf))
+ (space (- (buffer-length obuf) tail)))
+ (when (plusp space)
+ (copy-to-buffer obuf tail (min space (- end start)))
+ (go :more-output-p)))
+ :flush-and-fill
+ ;; Later copies should always have an empty buffer, since
+ ;; they are freshly flushed, but if another thread is
+ ;; stomping on the same buffer that might not be the case.
+ (let* ((obuf (flush-output-buffer stream))
+ (tail (buffer-tail obuf))
+ (space (- (buffer-length obuf) tail)))
+ (copy-to-buffer obuf tail (min space (- end start))))
+ :more-output-p
+ (when (> end start)
+ (go :flush-and-fill))))))
+
+;;; Flush the current output buffer of the stream, ensuring that the
+;;; new buffer is empty. Returns (for convenience) the new output
+;;; buffer -- which may or may not be EQ to the old one. If the is no
+;;; queued output we try to write the buffer immediately -- otherwise
+;;; we queue it for later.
+(defun flush-output-buffer (stream)
+ (let ((obuf (fd-stream-obuf stream)))
+ (when obuf
+ (let ((head (buffer-head obuf))
+ (tail (buffer-tail obuf)))
+ (cond ((eql head tail)
+ ;; Buffer is already empty -- just ensure that is is
+ ;; set to zero as well.
+ (reset-buffer obuf))
+ ((fd-stream-output-queue stream)
+ ;; There is already stuff on the queue -- go directly
+ ;; there.
+ (aver (< head tail))
+ (%queue-and-replace-output-buffer stream))
+ (t
+ ;; Try a non-blocking write, if SERVE-EVENT is allowed, queue
+ ;; whatever is left over. Otherwise wait until we can write.
+ (aver (< head tail))
+ (synchronize-stream-output stream)
+ (loop
+ (let ((length (- tail head)))
+ (multiple-value-bind (count errno)
+ (sb!unix:unix-write (fd-stream-fd stream) (buffer-sap obuf)
+ head length)
+ (flet ((queue-or-wait ()
+ (if (fd-stream-serve-events stream)
+ (return (%queue-and-replace-output-buffer stream))
+ (or (wait-until-fd-usable (fd-stream-fd stream) :output
+ (fd-stream-timeout stream)
+ nil)
+ (signal-timeout 'io-timeout
+ :stream stream
+ :direction :output
+ :seconds (fd-stream-timeout stream))))))
+ (cond ((eql count length)
+ ;; Complete write -- we can use the same buffer.
+ (return (reset-buffer obuf)))
+ (count
+ ;; Partial write -- update buffer status and
+ ;; queue or wait.
+ (incf head count)
+ (setf (buffer-head obuf) head)
+ (queue-or-wait))
+ #!-win32
+ ((eql errno sb!unix:ewouldblock)
+ ;; Blocking, queue or wair.
+ (queue-or-wait))
+ ;; if interrupted on win32, just try again
+ #!+win32 ((eql errno sb!unix:eintr))
+ (t
+ (simple-stream-perror "Couldn't write to ~s"
+ stream errno)))))))))))))
+
+;;; Helper for FLUSH-OUTPUT-BUFFER -- returns the new buffer.
+(defun %queue-and-replace-output-buffer (stream)
+ (aver (fd-stream-serve-events stream))
+ (let ((queue (fd-stream-output-queue stream))
+ (later (list (or (fd-stream-obuf stream) (bug "Missing obuf."))))
+ (new (get-buffer)))
+ ;; Important: before putting the buffer on queue, give the stream
+ ;; a new one. If we get an interrupt and unwind losing the buffer
+ ;; is relatively OK, but having the same buffer in two places
+ ;; would be bad.
+ (setf (fd-stream-obuf stream) new)
+ (cond (queue
+ (nconc queue later))
+ (t
+ (setf (fd-stream-output-queue stream) later)))
+ (unless (fd-stream-handler stream)
+ (setf (fd-stream-handler stream)
+ (add-fd-handler (fd-stream-fd stream)
+ :output
+ (lambda (fd)
+ (declare (ignore fd))
+ (write-output-from-queue stream)))))
+ new))
+
+;;; This is called by the FD-HANDLER for the stream when output is
+;;; possible.
+(defun write-output-from-queue (stream)
+ (aver (fd-stream-serve-events stream))
+ (synchronize-stream-output stream)
+ (let (not-first-p)
+ (tagbody
+ :pop-buffer
+ (let* ((buffer (pop (fd-stream-output-queue stream)))
+ (head (buffer-head buffer))
+ (length (- (buffer-tail buffer) head)))
+ (declare (index head length))
+ (aver (>= length 0))
+ (multiple-value-bind (count errno)
+ (sb!unix:unix-write (fd-stream-fd stream) (buffer-sap buffer)
+ head length)
+ (cond ((eql count length)
+ ;; Complete write, see if we can do another right
+ ;; away, or remove the handler if we're done.
+ (release-buffer buffer)
+ (cond ((fd-stream-output-queue stream)
+ (setf not-first-p t)
+ (go :pop-buffer))
+ (t
+ (let ((handler (fd-stream-handler stream)))
+ (aver handler)
+ (setf (fd-stream-handler stream) nil)
+ (remove-fd-handler handler)))))
+ (count
+ ;; Partial write. Update buffer status and requeue.
+ (aver (< count length))
+ ;; Do not use INCF! Another thread might have moved head.
+ (setf (buffer-head buffer) (+ head count))
+ (push buffer (fd-stream-output-queue stream)))
+ (not-first-p
+ ;; We tried to do multiple writes, and finally our
+ ;; luck ran out. Requeue.
+ (push buffer (fd-stream-output-queue stream)))
+ (t
+ ;; Could not write on the first try at all!
+ #!+win32
+ (simple-stream-perror "Couldn't write to ~S." stream errno)
+ #!-win32
+ (if (= errno sb!unix:ewouldblock)
+ (bug "Unexpected blocking in WRITE-OUTPUT-FROM-QUEUE.")
+ (simple-stream-perror "Couldn't write to ~S"
+ stream errno))))))))
+ nil)
+
+;;; Try to write THING directly to STREAM without buffering, if
+;;; possible. If direct write doesn't happen, buffer.
+(defun write-or-buffer-output (stream thing start end)
+ (declare (index start end))
+ (cond ((fd-stream-output-queue stream)
+ (buffer-output stream thing start end))
+ ((< end start)
+ (error ":END before :START!"))
+ ((> end start)
+ (let ((length (- end start)))
+ (synchronize-stream-output stream)
+ (multiple-value-bind (count errno)
+ (sb!unix:unix-write (fd-stream-fd stream) thing start length)
+ (cond ((eql count length)
+ ;; Complete write -- done!
+ )
+ (count
+ (aver (< count length))
+ ;; Partial write -- buffer the rest.
+ (buffer-output stream thing (+ start count) end))
+ (t
+ ;; Could not write -- buffer or error.
+ #!+win32
+ (simple-stream-perror "couldn't write to ~s" stream errno)
+ #!-win32
+ (if (= errno sb!unix:ewouldblock)
+ (buffer-output stream thing start end)
+ (simple-stream-perror "couldn't write to ~s" stream errno)))))))))
+
+;;; Deprecated -- can go away after 1.1 or so. Deprecated because
+;;; this is not something we want to export. Nikodemus thinks the
+;;; right thing is to support a low-level non-stream like IO layer,
+;;; akin to java.nio.
+(declaim (inline output-raw-bytes))
+(define-deprecated-function :late "1.0.8.16" output-raw-bytes write-sequence
+ (stream thing &optional start end)
+ (write-or-buffer-output stream thing (or start 0) (or end (length thing))))
+\f
projects / sbcl.git / blobdiff