source: trunk/src/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)))

;;; FFI

(defconstant +cmd-get-flags+ 3)
(defconstant +cmd-set-flags+ 4)
(defconstant +arg-nonblock+ #x0004)

(defcfun ("close" %close) :int
  (fd :int))

(defcfun ("fcntl" %fcntl) :int
  (fd :int)
  (cmd :int)
  (arg :int))

(defcfun ("read" %read) :long
  (fd :int)
  (buffer :pointer)
  (nbytes :unsigned-long))

(defcfun ("write" %write) :long
  (fd :int)
  (buffer :pointer)
  (nbytes :unsigned-long))

(defcfun ("memset" %memset) :pointer
  (buffer :pointer)
  (byte :int)
  (len :int))


;;; CLASSES

(defclass async-fd ()
  ((write-fd :initarg :write-fd)
   (write-queue :initform nil)

   (read-fd :initarg :read-fd)

   (foreign-read-buffer :initform (foreign-alloc :uint8 :count 4096))
   (foreign-read-buffer-size :initform 4096)

   (lisp-read-buffer :initform (make-uint8-seq 1024))
   (lisp-read-buffer-write-ptr :initform 0)

   (close-pending :initform nil)

   (accept-filter :initform nil)
   (read-callback :initform nil) 
   ))


(defmethod print-object ((async-fd async-fd) stream)
  (with-slots (read-fd write-fd write-queue) async-fd
    (format stream "#<ASYNC-FD r/w fd: ~D/~D, write queue length: ~D.>"
            read-fd write-fd (length write-queue))))


(defclass packet ()
  ((buffer  :initarg :buffer  :initform nil :documentation "Foreign array")
   (size    :initarg :size    :initform 0)
   (written :initarg :written :initform 0)))


;;; FUNCTIONS


(defun memzero (ptr size)
  (%memset ptr 0 size))


(defun set-fd-nonblocking (unix-fd)
  "Set UNIX-FD to non-blocking mode (O_NONBLOCK)."
  (%fcntl unix-fd +cmd-set-flags+ +arg-nonblock+))


(defun close-fd (unix-fd)
  "Close UNIX-FD."
  (%close unix-fd))


(defun make-uint8-seq (size)
  "Make uint8 sequence."
  (make-sequence '(vector (unsigned-byte 8)) size :initial-element 0))


(defun map-to-foreign (seq start end)
  "Map SEQ to foreign array."
  (let* ((len (- end start))
         (foreign-array (foreign-alloc :uint8 :count len)))

    (loop for i from 0 below len do
         (setf (mem-aref foreign-array :uint8 i) (aref seq (+ start i))))

    foreign-array))


(defun ensure-lisp-read-buffer-size (async-fd size)
  "Ensure that ASYNC-FD's LISP-READ-BUFFER is at least SIZE big."
  (with-slots (lisp-read-buffer lisp-read-buffer-write-ptr) async-fd

    (assert (and (not (null lisp-read-buffer)) (integerp lisp-read-buffer-write-ptr)))
 
    (unless (>= (length lisp-read-buffer) size)

      (let ((new-buf (make-uint8-seq (truncate (* 1.5 size)))))

        ;; copy old
        (loop for i from 0 below lisp-read-buffer-write-ptr do
             (setf (aref new-buf i) (aref lisp-read-buffer i)))

        (setf lisp-read-buffer new-buf)
        t))))


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

(defun read-more (async-fd)
  "Read more data from ASYNC-FD."
  (with-slots (foreign-read-buffer foreign-read-buffer-size) async-fd
    (with-slots (read-fd lisp-read-buffer lisp-read-buffer-write-ptr) async-fd

    (let ((new-bytes (%read read-fd foreign-read-buffer foreign-read-buffer-size)))

      (cond
       ((< new-bytes 0) 
        (error 'read-error))

       ((= new-bytes 0)
        nil);;(throw 'end-of-file nil))

       (t

        (ensure-lisp-read-buffer-size async-fd (+ lisp-read-buffer-write-ptr new-bytes))

        ;; copy data from foreign buffer to input-buffer
        (loop for i from 0 below new-bytes do
              (setf (aref lisp-read-buffer (+ i lisp-read-buffer-write-ptr)) 
                    (mem-aref foreign-read-buffer :uint8 i)))
        (incf lisp-read-buffer-write-ptr new-bytes)
        
        ;; call callback
        (with-slots (accept-filter read-callback) async-fd
          (if accept-filter
              (if (funcall accept-filter lisp-read-buffer lisp-read-buffer-write-ptr)
                  (funcall read-callback async-fd (subseq lisp-read-buffer 0 lisp-read-buffer-write-ptr)))
                  (funcall read-callback async-fd (subseq lisp-read-buffer 0 lisp-read-buffer-write-ptr))))

        (values lisp-read-buffer lisp-read-buffer-write-ptr new-bytes)
        ))))))  


(defun close-async-fd (async-fd)
  "Close ASYNC-FD's fd after everything has been written from write-queue."
  (with-slots (write-queue read-fd write-fd foreign-read-buffer) async-fd
    (cond

      ;; if write-queue is emtpy, close now
      ((null write-queue)
       (close-fd read-fd)
       (foreign-free foreign-read-buffer)
       (unless (= read-fd write-fd) (close-fd write-fd)))

      ;; data in write-queue, mark as closing
      (t
       (setf (slot-value async-fd 'close-pending) t)))))


(defun write-more (async-fd)
  "Write data from ASYNC-FD's write-queue."
  (with-slots (write-fd write-queue) async-fd

    ;; loop for packets in queue
    (loop for packet in write-queue do

          (with-slots (buffer size written) packet

            (assert (and (pointerp buffer) (integerp size) (integerp written)))

            ;; write data from entry to socket
            (let ((now-written (%write write-fd (inc-pointer buffer written) (- size written))))
              ;; update count
              (setf written (+ written now-written))
              ;; unless could write fully, return
              (unless (= written size)
                (return)))))

     ;; remove (and dealloc buffers) of entries that have been fully written
     (let ((unfinished-entries NIL))
       (setf write-queue
           (nreverse
            (dolist (entry write-queue unfinished-entries)
              (with-slots (size written) entry
                (if (= size written)
                    (with-slots (buffer) entry
                      (foreign-free buffer))
                    (push entry unfinished-entries)))))))

    ;; if queue empty and close pending, close fd
    (if (and (null write-queue) (slot-value async-fd 'close-pending))
        (close-async-fd async-fd))

    t))


(defun async-write-seq (async-fd seq &optional (start 0) (end (length seq)))
  "Queue from SEQ between START and END to write-queue." 

  (assert (and (numberp start) (not (null seq))))

  ;; enqueue sequence
  (with-slots (write-queue) async-fd
    (let ((entry (make-instance 'packet :buffer (map-to-foreign seq start end) :size (- end start))))
      (setf write-queue (append write-queue (list entry)))))

  ;; start writing
  (write-more async-fd))





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


(defun add-async-fd (event-queue async-fd mode)
  (ecase mode
    (:read (add-fd event-queue (slot-value async-fd 'read-fd) :read))
    (:write (add-fd event-queue (slot-value async-fd 'write-fd) :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))))


(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 set-accept-filter (async-fd accept-filter)
  (setf (slot-value async-fd 'accept-filter) accept-filter))

(defun set-read-callback (async-fd read-callback)
  (setf (slot-value async-fd 'read-callback) read-callback))