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

#|
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 2096)
                        :accessor foreign-read-buffer)
   (foreign-write-buffer :initform (byte-buffer 2096)
                         :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
                  :accessor close-pending)
;TODO this is either an inet-socket if we are client side or a node is we are server side...
   (socket :initarg :socket
           :accessor socket
           :documentation "The remote node we are talking to")))


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

;;Implement this in concrete SM for timeout processing
(defgeneric process-timeout (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) 
  ((errno :initform :errno
             :accessor read-error-errno)))

;;  "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)
        (let ((errno (get-errno)))
          (format-log t "async-fd:read-more - read-error, Errno: ~A~%" errno)
          (cond ((eql errno +ERRNO_EAGAIN+)
                 (setf (read-ready state-machine) nil))
                (t
                 (close-fd (read-fd state-machine))
                 (error 'read-error :errno errno)))))
       ((= new-bytes 0)
        (format-log t "async-fd:read-more - EOF on ~A~%" state-machine)
        (error 'read-error));;(throw 'end-of-file nil)
       (t
          ;;Update buffer position
          (inc-position foreign-read-buffer new-bytes)
#+nio-debug      (format t "read-more : Updated buffer ~A~%" foreign-read-buffer)
          (when (> (remaining foreign-read-buffer) 0)
            (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) ())
(define-condition write-error (error) 
  ((error-number :initarg :error)))

(defun write-more (async-fd)
  "Write data from ASYNC-FD's foreign-write-buffer to the network \
   Leaves foreign-write-buffer in state ready to be written to \
   Sets write-ready appropriatly"
#+nio-debug  (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-pointer 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)
        )
      
      (when (eql now-written -1)
          ;;Deal with failure
        (let ((err (get-errno)))
#+nio-debug       (format t "write-more - write returned -1 :errno ~A~%" err)
          (if (eql err 11) ;; eagain - failed to write whole buffer need to wait for next notify
              (setf (write-ready async-fd) nil)
              (progn
                (perror)
                (let ((err-cond (make-instance 'write-error :error err)))
                  (close-fd (write-fd async-fd)); - deal with in nio-server?
                  (error err-cond))))))
      ;;update buffers
      (if (eql (remaining foreign-write-buffer) 0)
          (clear foreign-write-buffer)
          (compact foreign-write-buffer)))
#+nio-debug    (format t "write buffer after write :~A~%" foreign-write-buffer)
      (when (eql (buffer-position foreign-write-buffer) 0)
        (when close-pending (close-async-fd async-fd)))))


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

(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))


(defun test-realloc()
  (let* ((sm (create-state-machine 'async-fd 1 1 6))
        (pos-b4-resize (bytebuffer-write-string (foreign-read-buffer sm) "this string is OK")))
     (recommend-buffer-size sm :read 4096)
     (assert (eql 4096 (buffer-capacity (foreign-read-buffer sm))))
     (assert (eql 4096 (nio-buffer:buffer-limit (foreign-read-buffer sm))))
     (assert (eql pos-b4-resize (nio-buffer:buffer-position (foreign-read-buffer sm))))))