source: branches/home/psmith/restructure/src/io/async-fd.lisp @ 68

#|
Copyright (c) 2006 Risto Laakso
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
3. The name of the author may not be used to endorse or promote products
derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|#
(in-package :nio)

(declaim (optimize (debug 3) (speed 3) (space 0)))

(defclass async-fd ()
  ((write-fd :initarg :write-fd
             :accessor write-fd)
   (read-fd :initarg :read-fd
            :accessor read-fd)
   (foreign-read-buffer :initform (byte-buffer 1024)
                        :accessor foreign-read-buffer)
   (foreign-write-buffer :initform (byte-buffer 1024)
                         :accessor foreign-write-buffer)
   (read-ready :initform nil
                 :accessor read-ready
                 :documentation "Have we been notified as read ready and not received EAGAIN from %read?")
   (write-ready :initform nil
                  :accessor write-ready
                  :documentation "Have we been notified as write ready and not received EAGAIN from %write?")
   (close-pending :initform nil)
   (socket :initarg :socket
           :accessor socket)))


(defmethod print-object ((async-fd async-fd) stream)
  (with-slots (socket read-fd write-fd) async-fd
    (format stream "#<ASYNC-FD :socket ~D :read-fd ~D :write-fd ~D.>"
            socket read-fd write-fd)))

;;Implement this in concrete SM for read
(defgeneric process-read (async-fd))

;;Implement this in concrete SM for read
(defgeneric process-write (async-fd))

;;SM factory 
(defun create-state-machine(sm-type read-fd write-fd socket)
  (let ((sm (make-instance sm-type :read-fd read-fd :write-fd write-fd :socket socket)))
    (format-log t "async-fd:create-state-machine - Created ~S~%" sm)
    (nio-buffer:clear (foreign-read-buffer sm))
    (nio-buffer:clear (foreign-write-buffer sm))
    sm))

;;override this in concrete SM for close
(defmethod process-close((async-fd async-fd)reason)())


(defmethod close-sm((async-fd async-fd))
  :documentation "Mark the socket for close to write"
#+nio-debug  (format t "(mark for)close called with ~A~%" async-fd)
  (with-slots (close-pending) async-fd
      (setf close-pending t)))

(define-condition read-error (error) ())

;;  "Read more data from STATE-MACHINE."
(defun read-more (state-machine)
  (with-slots (foreign-read-buffer read-fd) state-machine
#+nio-debug  (format t "read-more called with ~A~%" state-machine)
#+nio-debug      (format t "read-more - calling read() into ~A~%" foreign-read-buffer) 
    (let ((new-bytes (%read read-fd (buffer-pointer foreign-read-buffer) (remaining foreign-read-buffer))))
#+nio-debug      (format t "read-more : Read ~A bytes into ~A~%" new-bytes foreign-read-buffer)
      (cond
       ((< new-bytes 0)
        (progn
;;TODO if ret is -1 and errno is EAGAIN save state and wait for notification
          (format t "read-error - Errno: ~A~%" (get-errno))
          (error 'read-error)))
       ((= new-bytes 0)
        nil);;(throw 'end-of-file nil)
       (t
          ;;Update buffer position
          (inc-position foreign-read-buffer new-bytes)
          (setf (read-ready state-machine) nil))))))

(defun close-async-fd (async-fd)
  "Close ASYNC-FD's fd after everything has been written from write-queue."
#+nio-debug    (format t "close-async-fd called with :async-fd ~A~%" async-fd)
  (with-slots (read-fd write-fd foreign-read-buffer foreign-write-buffer) async-fd
    (nio-buffer:flip foreign-write-buffer)
#+nio-debug    (format t "close-async-fd foreign-write-buffer ~A~%" foreign-write-buffer)
    (assert (eql (remaining foreign-write-buffer) 0))
      ;; if write-queue is emtpy, close now
       (close-fd read-fd)
       (free-buffer foreign-read-buffer)
       (free-buffer foreign-write-buffer)
       (unless (= read-fd write-fd) (close-fd write-fd))))


(define-condition read-error (error) ())

(defun write-more (async-fd)
  "Write data from ASYNC-FD's write bytebuffer"
  (format-log t "async-fd:write-more - called with ~A~%" async-fd)
  (with-slots (write-fd foreign-write-buffer close-pending) async-fd
#+nio-debug  (format t "async-fd:write-more - foreign-write-buffer b4 flip ~A~%" foreign-write-buffer)
    (nio-buffer:flip foreign-write-buffer)
#+nio-debug (format t "async-fd:write-more -foreign-write-buffer after flip ~A~%" foreign-write-buffer)
    (let ((now-written 0))
      (do ((total-written 0))
          ((or (eql now-written -1) (eql (remaining foreign-write-buffer) 0)) total-written)
        (progn
          (setf now-written (%write write-fd (buffer-buf foreign-write-buffer) (remaining foreign-write-buffer)))
          (when (not (eql now-written -1))
            (inc-position foreign-write-buffer now-written)
            (incf total-written now-written)))
#+nio-debug (format t "async-fd:write-more - after write :foreign-write-buffer ~A :now-written ~A :total-written ~A ~%" foreign-write-buffer now-written total-written))
      (if (eql now-written -1)
          ;;Deal with failure
        (let ((err (get-errno)))
          (format t "write-more - write returned -1 :errno ~A~%" err)
          (unless (eql err 11) ;; eagain - failed to write whole buffer need to wait for next notify
            (perror)
            (let ((err-cond (make-instance 'write-error :error err)))
              (close err-cond)
              (error err-cond))))
        ;;update buffers
        (if (eql (remaining foreign-write-buffer) 0)
            (clear foreign-write-buffer)
            (error 'not-implemented-yet))))

#+nio-debug    (format t "write buffer after write :~A~%" foreign-write-buffer)
      (when (eql (remaining foreign-write-buffer) 0)
        (when close-pending (close-async-fd async-fd)))))


(defconstant +MAX-BUFFER-SIZE-BYTES+ (* 1024 1024))



;(let ((buffer (foreign-read-buffer async-fd)))
;          (if (>= (length buffer) size)
;              t
;              (let ((new-buffer (byte-buffer size)))
;                (copy-buffer buffer new-buffer)
;                (free-buffer buffer)
;                (setf (foreign-read-buffer async-fd) new-buffer)))))


(defmacro realloc-buffer(async-fd accessor size)
  `(let ((buffer (,accessor ,async-fd)))
     (if (>= (buffer-capacity buffer) size)
         t
         (let ((new-buffer (byte-buffer ,size)))
           (copy-buffer buffer new-buffer)
           (free-buffer buffer)
           (setf (,accessor ,async-fd) new-buffer)))))



;TODO actually deal with buffer allocation failure
(defmethod recommend-buffer-size((async-fd async-fd) mode size)
  (if (> size +MAX-BUFFER-SIZE-BYTES+) nil
      (ecase mode
        (:read (realloc-buffer async-fd foreign-read-buffer size))
        (:write (realloc-buffer async-fd foreign-write-buffer size)))))
   


(defun force-close-async-fd (async-fd)
  "Drop ASYNC-FD's write-queue and close it."
  (free-buffer (slot-value async-fd 'foreign-write-buffer))
  (close-async-fd async-fd))


(defun add-async-fd (event-queue async-fd mode)
  (ecase mode
    (:read-write (add-fd event-queue (slot-value async-fd 'write-fd) :read-write))))


(defun remove-async-fd (event-queue async-fd mode)
  (ecase mode
    (:read (remove-fd event-queue (slot-value async-fd 'read-fd) :read))
    (:write (remove-fd event-queue (slot-value async-fd 'write-fd) :write))
    (:read-write (remove-fd event-queue (slot-value async-fd 'write-fd) :read-write))))


(defun async-fd-read-fd (async-fd)
  (slot-value async-fd 'read-fd))

(defun async-fd-write-fd (async-fd)
  (slot-value async-fd 'write-fd))