nio-cvs

nio-cvs@common-lisp.net

118 discussions

[nio-cvs] r85 - in branches/home/psmith/restructure/src: io nio-logger utils
by psmith＠common-lisp.net 11 Feb '07

11 Feb '07

Author: psmith Date: Sun Feb 11 18:53:09 2007 New Revision: 85 Modified: branches/home/psmith/restructure/src/io/async-fd.lisp branches/home/psmith/restructure/src/io/async-socket.lisp branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/io/nio.asd branches/home/psmith/restructure/src/io/nodes.lisp branches/home/psmith/restructure/src/nio-logger/nio-logger.asd branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp branches/home/psmith/restructure/src/utils/nio-utils-package.lisp Log: Reconnect working Modified: branches/home/psmith/restructure/src/io/async-fd.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/async-fd.lisp (original) +++ branches/home/psmith/restructure/src/io/async-fd.lisp Sun Feb 11 18:53:09 2007 @@ -45,14 +45,16 @@ :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))) + :accessor socket + :documentation "The remote node we are talking to"))) (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))) + (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)) @@ -120,6 +122,8 @@ (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 write bytebuffer" @@ -144,7 +148,7 @@ (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) + (close-fd (write-fd async-fd)) (error err-cond)))) ;;update buffers (if (eql (remaining foreign-write-buffer) 0) @@ -158,17 +162,6 @@ (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) Modified: branches/home/psmith/restructure/src/io/async-socket.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/async-socket.lisp (original) +++ branches/home/psmith/restructure/src/io/async-socket.lisp Sun Feb 11 18:53:09 2007 @@ -132,10 +132,10 @@ t nil))) -(defun connect-inet-socket (socket-fd addr port) +(defun connect-inet-socket (socket-fd node) + (format-log t "async-socket:connect-inet-socket ccalled with ~A, and ~A~%" socket-fd node) (with-foreign-object (sa 'sockaddr-in) - (init-inet-socket sa port addr) - + (init-inet-socket sa (remote-port node) (remote-host node)) (let ((res (%connect socket-fd sa +sockaddr-in-len+))) (format-log t "async-socket:connect-inet-socket library connect call returned ~A, and errno ~A~%" res (get-errno)) (if (= res -1) @@ -171,13 +171,6 @@ ;;;; SOCKET I/O -(defclass async-socket-fd () - ((family :initform :unknown :initarg :family) - (remote-host :initform nil :initarg :remote-host) - (remote-port :initform nil :initarg :remote-port))) - - - (defun socket-accept (socket-fd connection-type) "Accept connection from SOCKET-FD. Allocates and returns socket structure denoting the connection." @@ -200,16 +193,18 @@ (let ((len (foreign-alloc :unsigned-long :initial-element +sockaddr-in6-len+))) ;; accept connection +#+nio-debug (format-log t "async-socket::socket-accept - calling %accept~%") (let* ((res (%accept socket-fd addr len)) -;; (async-socket-fd (make-instance 'async-socket-fd :read-fd res :write-fd res))) - (async-socket-fd (create-state-machine connection-type res res (make-instance 'async-socket-fd)))) + (async-fd (create-state-machine connection-type res res (node nil nil)))) + +#+nio-debug (format-log t "async-socket::socket-accept - create async-fd ~A~%" async-fd) (unless (< res 0) (let ((len-value (mem-ref len :unsigned-int))) ;; parse sockaddr struct for remote client info - (with-slots (family remote-host remote-port) (socket async-socket-fd) + (with-slots (family remote-host remote-port) (socket async-fd) (cond ((= len-value +sockaddr-in6-len+) @@ -224,12 +219,5 @@ (foreign-free len) - (if (>= res 0) async-socket-fd nil) + (if (>= res 0) async-fd nil) ))))))) - - -(defun remote-info (async-socket-fd) - "Return FAMILY, REMOTE-HOST and REMOTE-PORT in list." - (with-slots (family remote-host remote-port) async-socket-fd - (list family remote-host remote-port))) - Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Sun Feb 11 18:53:09 2007 @@ -41,7 +41,7 @@ (defun process-async-fds (client-hash) (let ((removals nil)) (maphash #'(lambda (k async-fd) -#+nio-debug2 (format-log t "Dealing with ~a => ~a~%" k async-fd) +#+nio-debug (format-log t "Dealing with ~a => ~a~%" k async-fd) ;process reads (when (read-ready async-fd) (read-more async-fd)) @@ -49,20 +49,22 @@ (process-read async-fd)) ;process-writes - (process-write async-fd) - (when (and (write-ready async-fd) - (> (buffer-position (foreign-write-buffer async-fd)) 0)) - (write-more async-fd)) - - (when (close-pending async-fd) - (write-more async-fd) - (push async-fd removals))) + (handler-case + (progn + (process-write async-fd) + (when (and (write-ready async-fd) + (> (buffer-position (foreign-write-buffer async-fd)) 0)) + (write-more async-fd)) + + (when (close-pending async-fd) + (write-more async-fd) + (push async-fd removals))) + (write-error (we) (push async-fd removals)))) client-hash) (dolist (async-fd removals) + (format-log t "nio-server:process-async-fds processing remove for ~a~%" async-fd) + (setf (active-conn (socket async-fd)) nil) (remhash (async-fd-read-fd async-fd) client-hash)))) -; (format t "client-hash list ~A~%"client-hash ) - - (defun start-server (connection-type @@ -121,9 +123,7 @@ (format t "Error setting socket non-blocking: ") (perror))) (setf (gethash (async-fd-read-fd async-fd) client-hash) async-fd) - (add-async-fd event-queue async-fd :read-write) -; (add-async-fd event-queue async-fd :write) - ) + (add-async-fd event-queue async-fd :read-write)) ;; no accept, close (t @@ -149,14 +149,16 @@ (when (read-event-p event) (setf (read-ready async-fd) t)) (when (write-event-p event) (setf (write-ready async-fd) t))))))))) - ;add outgoing sockets to event queue +;add outgoing sockets to event queue #+nio-debug2 (format-log t "nio-server:start-server - Processing new connections queue ~A~%" +connected-sockets-queue+) (loop for node = (nio-compat:take +connected-sockets-queue+ :blocking-call nil) until (null node) do #+nio-debug (format-log t "nio-server:start-server - adding node to nodes-list ~A~%" node) (push node *nodes-list*)) + (with-connect-ready-nodes (a-node) #+nio-debug (format-log t "nio-server:start-server - attempting connection to node ~A~%" a-node) - (let ((new-fd (connect (host a-node) (port a-node) connection-type))) + (let ((new-fd (connect a-node connection-type))) +#+nio-debug (format-log t "nio-server:start-server - connect returned async-fd ~A~%" new-fd) (update-last-connect-attempt a-node) (when new-fd #+nio-debug (format-log t "nio-server:start-server - adding connection to nio thread ~A~%" new-fd) @@ -171,19 +173,17 @@ (close-fd sock)))) -(defun connect(host port connection-type +(defun connect(node connection-type &key (protocol :inet)) - (format-log t "nio-server:connect - Called with: ~A:~A:~A ~%" protocol host port) + (format-log t "nio-server:connect - Called with: ~A ~A~%" protocol node) (let ((sock nil)) (setq sock (ecase protocol (:inet (make-inet-socket)) (:inet6 (make-inet6-socket)))) - (if (connect-inet-socket sock host port) - (let ((sm (create-state-machine connection-type sock sock sock))) -; (nio-compat:add +connected-sockets-queue+ sm) -; (format-log t "nio-server:connect - Socket enqueued: ~A~%" +connected-sockets-queue+) + (if (connect-inet-socket sock node) + (let ((sm (create-state-machine connection-type sock sock node))) (return-from connect sm)) (format t "Connect failed!!~A ~%" (get-errno))))) Modified: branches/home/psmith/restructure/src/io/nio.asd ============================================================================== --- branches/home/psmith/restructure/src/io/nio.asd (original) +++ branches/home/psmith/restructure/src/io/nio.asd Sun Feb 11 18:53:09 2007 @@ -8,8 +8,8 @@ (:file "fd-helper" :depends-on ("nio-package")) (:file "packet" :depends-on ("nio-package")) (:file "async-fd" :depends-on ("fd-helper")) - (:file "async-socket" :depends-on ("async-fd")) (:file "nodes" :depends-on ("nio-package")) + (:file "async-socket" :depends-on ("async-fd" "nodes")) (:file "nio-server" :depends-on ("async-socket" "nodes")) (:file "ip-authorisation" :depends-on ("nio-package")) ) Modified: branches/home/psmith/restructure/src/io/nodes.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nodes.lisp (original) +++ branches/home/psmith/restructure/src/io/nodes.lisp Sun Feb 11 18:53:09 2007 @@ -28,24 +28,28 @@ (declaim (optimize (debug 3) (speed 3) (space 0))) -;;concept of a remote socket + +;;concept of a remote socket with properties e.g. stats, connection attempts etc (defclass node() - ((host :initarg :host - :reader host) - (port :initarg :port - :reader port) + ((family :initform :unknown :initarg :family) + (remote-host :initarg :remote-host + :initform nil + :accessor remote-host) + (remote-port :initarg :remote-port + :initform nil + :accessor remote-port) (last-connect-attempt :initform nil :accessor last-connect-attempt :documentation "Time we last attempted a connection") - (retry-delay :initform 600 + (retry-delay :initform 60 :accessor retry-delay - :documentation "The delay to wait after the last-connection-attempt before trying to connect again") + :documentation "The delay to wait (in secs) after the last-connection-attempt before trying to connect again (10 mins)") (active-conn :initform nil :accessor active-conn :documentation "If we are connected to this remote socket this is set to the SM"))) (defun node(host port) - (make-instance 'node :host host :port port)) + (make-instance 'node :remote-host host :remote-port port)) ;(node-from-socket-repn "192.168.1.1:1234") (defun node-from-socket-repn(socket) @@ -56,8 +60,9 @@ (defmethod print-object ((a-node node) stream) - (with-slots (host port last-connect-attempt retry-delay active-conn) a-node - (format stream "#<NODE :HOST ~A :port ~A :last-connect-attempt ~A :retry-delay ~A :active-conn ~A>" host port last-connect-attempt retry-delay active-conn))) + (with-slots (remote-host remote-port last-connect-attempt retry-delay active-conn) a-node + (format stream "#<NODE :remote-host ~A :remote-port ~A :last-connect-attempt ~A :retry-delay ~A :active-conn ~A>" + remote-host remote-port last-connect-attempt retry-delay active-conn))) (defparameter *nodes-list* nil Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.asd ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.asd (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.asd Sun Feb 11 18:53:09 2007 @@ -8,5 +8,5 @@ (:file "nio-logger" :depends-on ("nio-logger-package")) ) - :depends-on (:nio-yarpc :nio-utils :cl-base64)) + :depends-on (:nio-yarpc :nio-utils :cl-base64 :sb-posix)) Modified: branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp Sun Feb 11 18:53:09 2007 @@ -25,7 +25,7 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |# -;(push :nio-debug *features*) +(push :nio-debug *features*) (require :asdf) (require :nio-logger) @@ -33,7 +33,20 @@ (let ((log-file (second sb-ext:*posix-argv*)) (ip (third sb-ext:*posix-argv*))) (format t "Starting logging client with ~A ~A~%" log-file ip) - (sb-thread:make-thread #'(lambda()(nio-logger:tail-log log-file ip)) :name "nio-server") + (sb-thread:make-thread #'(lambda()(nio:start-server 'nio-yarpc:yarpc-client-state-machine)) :name "nio-server") +; (nio:add-connection (nio:node ip 16323)) +; (sleep 60) - ;;shouldn't be listenting on the client hence nil for accept SM to start-server - (nio:start-server 'nio-yarpc:yarpc-client-state-machine)) + (setf sb-ext:*invoke-debugger-hook* + (lambda (condition hook) + (declare (ignore hook)) + (with-open-file (out (format nil "error-from-pid~A-thread-~A" (sb-posix:getpid) (sb-thread:thread-name sb-thread:*current-thread*)) + :direction :output :external-format :utf-8 :if-exists :append :if-does-not-exist :create) + (format out "Toplevel catch (~A):~%" (nio-utils:get-readable-time)) + (format out "~A - ~A~%" (type-of condition) condition) + (sb-debug:backtrace 20 out)) + (quit))) + + + (nio-logger:tail-log log-file ip) +) Modified: branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp Sun Feb 11 18:53:09 2007 @@ -25,7 +25,7 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |# -;(push :nio-debug *features*) +(push :nio-debug *features*) (require :asdf) (require :nio-logger) Modified: branches/home/psmith/restructure/src/utils/nio-utils-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/utils/nio-utils-package.lisp (original) +++ branches/home/psmith/restructure/src/utils/nio-utils-package.lisp Sun Feb 11 18:53:09 2007 @@ -29,5 +29,5 @@ (:export ;;utils - format-log get-universal-high-res + format-log get-universal-high-res get-readable-time ))

1 0

[nio-cvs] r84 - in branches/home/psmith/restructure/src: event io nio-logger protocol/yarpc statemachine
by psmith＠common-lisp.net 11 Feb '07

11 Feb '07

Author: psmith Date: Sat Feb 10 20:11:03 2007 New Revision: 84 Modified: branches/home/psmith/restructure/src/event/epoll.lisp branches/home/psmith/restructure/src/io/async-fd.lisp branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp branches/home/psmith/restructure/src/statemachine/state-machine.lisp Log: cleanup of debug msgs Modified: branches/home/psmith/restructure/src/event/epoll.lisp ============================================================================== --- branches/home/psmith/restructure/src/event/epoll.lisp (original) +++ branches/home/psmith/restructure/src/event/epoll.lisp Sat Feb 10 20:11:03 2007 @@ -84,10 +84,12 @@ (case res (-1 (let ((errno (get-errno))) - (format t "-1 returned from epoll-wait, errno ~A~%" errno) (if (eql errno 4) ;EINTR - interrupted by a system call - (return nil) - (error 'poll-error)))) + (return nil) + (progn + (format t "-1 returned from epoll-wait, errno:") + (perror) + (error 'poll-error))))) (return nil) (t (let ((idents nil)) Modified: branches/home/psmith/restructure/src/io/async-fd.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/async-fd.lisp (original) +++ branches/home/psmith/restructure/src/io/async-fd.lisp Sat Feb 10 20:11:03 2007 @@ -123,7 +123,7 @@ (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) +#+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) Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Sat Feb 10 20:11:03 2007 @@ -134,7 +134,7 @@ ;; socket i/o available (t (let ((async-fd (gethash fd client-hash))) - (format-log t "IO event ~A on ~A~%" event async-fd) +#+nio-debug (format-log t "nio-server::start-server - IO event ~A on ~A~%" event async-fd) (unless (null async-fd) (catch 'error-exit (handler-bind ((read-error #'(lambda (x) Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Sat Feb 10 20:11:03 2007 @@ -42,7 +42,8 @@ (defun callback(result) - (nio-utils:format-log t "Result of remote-log ~A~%" result)) +#+nio-debug (nio-utils:format-log t "Result of remote-log ~A~%" result) +) ;;Tail the given log and write to remote logger @@ -52,9 +53,9 @@ (nio:add-connection (nio:node ip-address 16323)) (with-line-from-tailed-file (text filename 1) (let ((rpc (format nil "(nio-logger:remote-log \"~A\")" (cl-base64:string-to-base64-string text)))) - (nio-utils:format-log t "Toplevel Submitting job~A~%" rpc) +#+nio-debug (nio-utils:format-log t "nio-logger::tail-log Submitting job~A~%" rpc) (nio:with-connected-nodes (node) - (nio-utils:format-log t "Toplevel sending ~A to ~A~%" rpc node) +#+nio-debug (nio-utils:format-log t "Toplevel sending ~A to ~A~%" rpc node) (nio-yarpc:remote-execute (nio:active-conn node) rpc #'callback))))) ;Runs a multithreaded system with an IO thread dealing with IO only and a 'job' thread taking and executing jobs Modified: branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp Sat Feb 10 20:11:03 2007 @@ -25,7 +25,7 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |# -(push :nio-debug *features*) +;(push :nio-debug *features*) (require :asdf) (require :nio-logger) Modified: branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/run-logging-server.lisp Sat Feb 10 20:11:03 2007 @@ -25,7 +25,7 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |# -(push :nio-debug *features*) +;(push :nio-debug *features*) (require :asdf) (require :nio-logger) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Sat Feb 10 20:11:03 2007 @@ -76,22 +76,22 @@ #+nio-debug (format-log t "yarpc-client-state-machine:process-outgoing-packet called, polling the job-queue ~%") (let ((ttd (nio-compat:take (job-queue sm) :blocking-call nil))) (when ttd - (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" ttd) +#+nio-debug (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" ttd) (destructuring-bind (job call-string) ttd (setf (gethash (incf +request-id+) (request-map sm)) job) (make-instance 'call-method-packet :call-string call-string :request-id +request-id+))))) (defmethod process-incoming-packet ((sm yarpc-client-state-machine) (response method-response-packet)) - (format-log t "yarpc-client-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm response) +#+nio-debug (format-log t "yarpc-client-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm response) (let* ((*package* (find-package :nio-yarpc)) (result (read-from-string (response response))) (request-id (request-id response))) - (format-log t "yarpc-client-state-machine:process-incoming-packet :result ~A :request-id ~A~%" result request-id) - (maphash #'(lambda (k v) (format t "~a -> ~a~%" k v)) (request-map sm)) +#+nio-debug (format-log t "yarpc-client-state-machine:process-incoming-packet :result ~A :request-id ~A~%" result request-id) +; (maphash #'(lambda (k v) (format t "~a -> ~a~%" k v)) (request-map sm)) (let ((remote-job (gethash request-id (request-map sm)))) (funcall (callback remote-job) result)))) ;Execute the call-string on the remote node and call callback with the result (defmethod remote-execute ((sm yarpc-client-state-machine) call-string callback) - (format-log t "yarpc-client-state-machine:remote-execute called :sm ~A :call-string ~A :callback ~A~%" sm call-string callback) +#+nio-debug (format-log t "yarpc-client-state-machine:remote-execute called :sm ~A :call-string ~A :callback ~A~%" sm call-string callback) (nio-compat:add (job-queue sm) (list (remote-job callback) call-string))) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp Sat Feb 10 20:11:03 2007 @@ -53,8 +53,12 @@ (if (<= (- packet-length +yarpc-packet-header-size+) (remaining buf)) ;is the whole packet available in the buffer? (let* ((packet-request-id (bytebuffer-read-32 buf)) (ret-packet (ecase packet-id - (0 (progn (format-log t "yarpc-packet-factory:get-packet - got CALL-METHOD-PACKET-ID~%") (call-method-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+ +PACKET-REQUEST-ID-SIZE+)) :request-id packet-request-id))) - (1 (progn (format-log t "yarpc-packet-factory:get-packet - got METHOD-RESPONSE-PACKET-ID~%") (method-response-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+ +PACKET-REQUEST-ID-SIZE+)) :request-id packet-request-id)))))) + (0 (progn +#+nio-debug (format-log t "yarpc-packet-factory:get-packet - got CALL-METHOD-PACKET-ID~%") + (call-method-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+ +PACKET-REQUEST-ID-SIZE+)) :request-id packet-request-id))) + (1 (progn +#+nio-debug (format-log t "yarpc-packet-factory:get-packet - got METHOD-RESPONSE-PACKET-ID~%") + (method-response-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+ +PACKET-REQUEST-ID-SIZE+)) :request-id packet-request-id)))))) (compact buf) #+nio-debug (format-log t "yarpc-packet-factory:get-packet - after compact ~%~A~%" buf) #+nio-debug (format-log t "yarpc-packet-factory:get-packet - retuirning packet ~A~%" ret-packet) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Sat Feb 10 20:11:03 2007 @@ -59,25 +59,25 @@ (defun run-job(&key (blocking t)) - (format-log t "yarpc-state-machine:run-job - Server toplevel waiting for job~%") +#+nio-debug (format-log t "yarpc-state-machine:run-job - Server toplevel waiting for job~%") (let ((server-job (nio-compat:take nio-yarpc:job-queue :blocking-call blocking))) (when server-job (destructuring-bind (job request-id result-queue) server-job - (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job) +#+nio-debug (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job) (nio-compat:add result-queue (list request-id (nio-yarpc:execute-call job))))))) (defmethod process-outgoing-packet((sm yarpc-state-machine)) - (format-log t "yarpc-state-machine:process-outgoing-packet - called, polling the results-queue ~%" ) +#+nio-debug (format-log t "yarpc-state-machine:process-outgoing-packet - called, polling the results-queue ~%" ) (let ((server-job (nio-compat:take (result-queue sm) :blocking-call nil))) (when server-job (destructuring-bind (request-id result) server-job - (format-log t "yarpc-state-machine:process-outgoing-packet - got :request-id ~A result ~A ~%" request-id result) +#+nio-debug (format-log t "yarpc-state-machine:process-outgoing-packet - got :request-id ~A result ~A ~%" request-id result) (method-response-packet result :request-id request-id))))) ;Process a call method packet by placing it in the job-queue (defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet)) - (format-log t "yarpc-state-machine:process-incoming-packet - called :sm ~A :packet ~A~%" sm call) +#+nio-debug (format-log t "yarpc-state-machine:process-incoming-packet - called :sm ~A :packet ~A~%" sm call) (nio-compat:add job-queue (list (call-string call) (request-id call) (result-queue sm))) (when +process-jobs-inline+ (run-job :blocking nil))) @@ -107,7 +107,7 @@ (handler-case (let* ((rpc-call-list (read-from-string call-string )) (fn (member (symbol-function (first rpc-call-list)) *remote-fns* ))) - (format-log t "yarpc-state-machine:execute-call - fn ~A authorised? : ~A~%" (symbol-function (first rpc-call-list)) fn) +#+nio-debug (format-log t "yarpc-state-machine:execute-call - fn ~A authorised? : ~A~%" (symbol-function (first rpc-call-list)) fn) (if fn (apply (first rpc-call-list) (rest rpc-call-list)) (error 'authorization-error))) Modified: branches/home/psmith/restructure/src/statemachine/state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/statemachine/state-machine.lisp (original) +++ branches/home/psmith/restructure/src/statemachine/state-machine.lisp Sat Feb 10 20:11:03 2007 @@ -61,7 +61,9 @@ (get-packet (get-packet-factory sm) foreign-read-buffer) (buffer-too-small-error (read-err) (if (recommend-buffer-size sm :read (recommended-size read-err)) + (progn (format-log t "resized incomming buffer ~A~%"foreign-read-buffer) + nil) (error 'not-implemented-yet-read-resize-failure)))))) (format-log t "state-machine::process-read - incoming packet: ~A~%" incoming-packet) (when incoming-packet

1 0

[nio-cvs] r83 - branches/home/psmith/restructure/src/protocol/yarpc
by psmith＠common-lisp.net 11 Feb '07

11 Feb '07

Author: psmith Date: Sat Feb 10 19:39:23 2007 New Revision: 83 Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Log: first working multiplexing rpc Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Sat Feb 10 19:39:23 2007 @@ -78,14 +78,18 @@ (when ttd (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" ttd) (destructuring-bind (job call-string) ttd - (setf (gethash (1+ +request-id+) (request-map sm)) job) + (setf (gethash (incf +request-id+) (request-map sm)) job) (make-instance 'call-method-packet :call-string call-string :request-id +request-id+))))) (defmethod process-incoming-packet ((sm yarpc-client-state-machine) (response method-response-packet)) (format-log t "yarpc-client-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm response) (let* ((*package* (find-package :nio-yarpc)) - (result (read-from-string (response response)))) - (nio-compat:add (result-queue sm) result))) + (result (read-from-string (response response))) + (request-id (request-id response))) + (format-log t "yarpc-client-state-machine:process-incoming-packet :result ~A :request-id ~A~%" result request-id) + (maphash #'(lambda (k v) (format t "~a -> ~a~%" k v)) (request-map sm)) + (let ((remote-job (gethash request-id (request-map sm)))) + (funcall (callback remote-job) result)))) ;Execute the call-string on the remote node and call callback with the result (defmethod remote-execute ((sm yarpc-client-state-machine) call-string callback)

1 0

[nio-cvs] r82 - in branches/home/psmith/restructure/src: io nio-logger protocol/yarpc
by psmith＠common-lisp.net 10 Feb '07

10 Feb '07

Author: psmith Date: Sat Feb 10 18:52:32 2007 New Revision: 82 Modified: branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Log: getting there... Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Sat Feb 10 18:52:32 2007 @@ -68,7 +68,7 @@ (defun start-server (connection-type &key (protocol :inet) - (port (+ (random 60000) 1024)) + (port 0) ;//if set then listen (host "127.0.0.1") (accept-connection #'trivial-accept)) (let (sock @@ -76,7 +76,7 @@ (client-hash (make-hash-table :test 'eql)) ) - (when (not (null connection-type)) + (when (not (eql port 0)) (format t "Binding to ~A:~A~%" host port) (setq sock (ecase protocol (:inet (make-inet-socket)) @@ -161,6 +161,7 @@ (when new-fd #+nio-debug (format-log t "nio-server:start-server - adding connection to nio thread ~A~%" new-fd) (setf (gethash (async-fd-read-fd new-fd) client-hash) new-fd) + (setf (active-conn a-node) new-fd) (add-async-fd event-queue new-fd :read-write)))) ;loop over async-fd's processing where necessary Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Sat Feb 10 18:52:32 2007 @@ -54,6 +54,7 @@ (let ((rpc (format nil "(nio-logger:remote-log \"~A\")" (cl-base64:string-to-base64-string text)))) (nio-utils:format-log t "Toplevel Submitting job~A~%" rpc) (nio:with-connected-nodes (node) + (nio-utils:format-log t "Toplevel sending ~A to ~A~%" rpc node) (nio-yarpc:remote-execute (nio:active-conn node) rpc #'callback))))) ;Runs a multithreaded system with an IO thread dealing with IO only and a 'job' thread taking and executing jobs @@ -64,7 +65,7 @@ (setf nio-yarpc:+process-jobs-inline+ nil) (setf +log-file-name+ out-file) (nio:load-ips allowed-ips) - (sb-thread:make-thread #'(lambda()(nio:start-server 'nio-yarpc:yarpc-state-machine :host listen-ip :accept-connection 'nio:check-ip)) :name "nio-server") + (sb-thread:make-thread #'(lambda()(nio:start-server 'nio-yarpc:yarpc-state-machine :host listen-ip :port 16323 :accept-connection 'nio:check-ip)) :name "nio-server") (loop ;;block waiting for jobs (nio-yarpc:run-job))) Modified: branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp Sat Feb 10 18:52:32 2007 @@ -36,4 +36,4 @@ (sb-thread:make-thread #'(lambda()(nio-logger:tail-log log-file ip)) :name "nio-server") ;;shouldn't be listenting on the client hence nil for accept SM to start-server - (nio:start-server nil)) + (nio:start-server 'nio-yarpc:yarpc-client-state-machine)) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Sat Feb 10 18:52:32 2007 @@ -41,7 +41,8 @@ :documentation "A map from request-id (a unique id for this request) to remote-job"))) (defclass remote-job() - ((callback :accessor callback + ((callback :initarg :callback + :accessor callback :documentation "A function accepting one argument to call with the result of the remote operation") (start-time :initform (get-universal-high-res) :reader start-time @@ -72,7 +73,7 @@ (defparameter +request-id+ 0) (defmethod process-outgoing-packet((sm yarpc-client-state-machine)) -#+nio-debug2 (format-log t "yarpc-client-state-machine:process-outgoing-packet called, polling the job-queue ~%") +#+nio-debug (format-log t "yarpc-client-state-machine:process-outgoing-packet called, polling the job-queue ~%") (let ((ttd (nio-compat:take (job-queue sm) :blocking-call nil))) (when ttd (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" ttd) @@ -88,4 +89,5 @@ ;Execute the call-string on the remote node and call callback with the result (defmethod remote-execute ((sm yarpc-client-state-machine) call-string callback) - (nio-compat:add (job-queue sm) '((remote-job callback) call-string))) + (format-log t "yarpc-client-state-machine:remote-execute called :sm ~A :call-string ~A :callback ~A~%" sm call-string callback) + (nio-compat:add (job-queue sm) (list (remote-job callback) call-string))) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp Sat Feb 10 18:52:32 2007 @@ -40,7 +40,7 @@ (defconstant +PACKET-ID-SIZE+ 1) (defconstant +PACKET-LENGTH-SIZE+ 4) -;(defconstant +PACKET-REQUEST-ID+ 4) +(defconstant +PACKET-REQUEST-ID-SIZE+ 4) (defconstant +yarpc-packet-header-size+ (+ +PACKET-ID-SIZE+ +PACKET-LENGTH-SIZE+)) @@ -53,8 +53,8 @@ (if (<= (- packet-length +yarpc-packet-header-size+) (remaining buf)) ;is the whole packet available in the buffer? (let* ((packet-request-id (bytebuffer-read-32 buf)) (ret-packet (ecase packet-id - (0 (progn (format-log t "yarpc-packet-factory:get-packet - got CALL-METHOD-PACKET-ID~%") (call-method-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+)) :request-id packet-request-id))) - (1 (progn (format-log t "yarpc-packet-factory:get-packet - got METHOD-RESPONSE-PACKET-ID~%") (method-response-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+)) :request-id packet-request-id)))))) + (0 (progn (format-log t "yarpc-packet-factory:get-packet - got CALL-METHOD-PACKET-ID~%") (call-method-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+ +PACKET-REQUEST-ID-SIZE+)) :request-id packet-request-id))) + (1 (progn (format-log t "yarpc-packet-factory:get-packet - got METHOD-RESPONSE-PACKET-ID~%") (method-response-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+ +PACKET-REQUEST-ID-SIZE+)) :request-id packet-request-id)))))) (compact buf) #+nio-debug (format-log t "yarpc-packet-factory:get-packet - after compact ~%~A~%" buf) #+nio-debug (format-log t "yarpc-packet-factory:get-packet - retuirning packet ~A~%" ret-packet) @@ -70,10 +70,12 @@ ((request-id :initarg :request-id :reader request-id))) -(defclass call-method-packet (yarpc-packet)((call-string :initarg :call-string - :accessor call-string))) -(defun call-method-packet (call-string) - (make-instance 'call-method-packet :call-string call-string)) +(defclass call-method-packet (yarpc-packet) + ((call-string :initarg :call-string + :accessor call-string))) + +(defun call-method-packet (call-string &key request-id) + (make-instance 'call-method-packet :call-string call-string :request-id request-id)) (defmethod print-object ((packet call-method-packet) stream) (format stream "#<CALL-METHOD-PACKET ~A >" (call-string packet))) @@ -103,8 +105,8 @@ ((response :initarg :response :accessor response))) -(defun method-response-packet (response) - (make-instance 'method-response-packet :response response)) +(defun method-response-packet (response &key request-id) + (make-instance 'method-response-packet :response response :request-id request-id)) (defmethod print-object ((packet method-response-packet) stream) (format stream "#<METHID-RESPONSE-PACKET ~A >" (response packet))) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Sat Feb 10 18:52:32 2007 @@ -60,17 +60,20 @@ (defun run-job(&key (blocking t)) (format-log t "yarpc-state-machine:run-job - Server toplevel waiting for job~%") - (destructuring-bind (job request-id result-queue) (nio-compat:take nio-yarpc:job-queue :blocking-call blocking) - (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job) - (nio-compat:add result-queue (list request-id (nio-yarpc:execute-call job))))) + (let ((server-job (nio-compat:take nio-yarpc:job-queue :blocking-call blocking))) + (when server-job + (destructuring-bind (job request-id result-queue) server-job + (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job) + (nio-compat:add result-queue (list request-id (nio-yarpc:execute-call job))))))) (defmethod process-outgoing-packet((sm yarpc-state-machine)) (format-log t "yarpc-state-machine:process-outgoing-packet - called, polling the results-queue ~%" ) - (destructuring-bind (request-id result) (nio-compat:take (result-queue sm) :blocking-call nil) - (format-log t "yarpc-state-machine:process-outgoing-packet - got :request-id ~A result ~A ~%" request-id result) - (when result - (method-response-packet result :request-id request-id)))) + (let ((server-job (nio-compat:take (result-queue sm) :blocking-call nil))) + (when server-job + (destructuring-bind (request-id result) server-job + (format-log t "yarpc-state-machine:process-outgoing-packet - got :request-id ~A result ~A ~%" request-id result) + (method-response-packet result :request-id request-id))))) ;Process a call method packet by placing it in the job-queue (defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet))

1 0

[nio-cvs] r81 - in branches/home/psmith/restructure/src: io nio-logger protocol/yarpc
by psmith＠common-lisp.net 10 Feb '07

10 Feb '07

Author: psmith Date: Sat Feb 10 15:36:43 2007 New Revision: 81 Modified: branches/home/psmith/restructure/src/io/nio-package.lisp branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/io/nio.asd branches/home/psmith/restructure/src/io/nodes.lisp branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Log: First stab at rpc multiplexing Modified: branches/home/psmith/restructure/src/io/nio-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-package.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-package.lisp Sat Feb 10 15:36:43 2007 @@ -42,4 +42,7 @@ ;;ip-authorisation check-ip load-ips + + ;;nodes + node with-connected-nodes active-conn )) Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Sat Feb 10 15:36:43 2007 @@ -35,7 +35,7 @@ ;TODO thread safety (defparameter +connected-sockets-queue+ (nio-compat:concurrent-queue) - "List of sockets that have been connected and are awaiting addition to the event-notification system") + "List of node objects that are to be connected to") ;loop over hashtable (defun process-async-fds (client-hash) @@ -150,12 +150,18 @@ (when (write-event-p event) (setf (write-ready async-fd) t))))))))) ;add outgoing sockets to event queue -#+nio-debug2 (format-log t "nio-server:start-server - Processing client add ~A~%" +connected-sockets-queue+) - - (loop for new-fd = (nio-compat:take +connected-sockets-queue+ :blocking-call nil) until (null new-fd) do +#+nio-debug2 (format-log t "nio-server:start-server - Processing new connections queue ~A~%" +connected-sockets-queue+) + (loop for node = (nio-compat:take +connected-sockets-queue+ :blocking-call nil) until (null node) do +#+nio-debug (format-log t "nio-server:start-server - adding node to nodes-list ~A~%" node) + (push node *nodes-list*)) + (with-connect-ready-nodes (a-node) +#+nio-debug (format-log t "nio-server:start-server - attempting connection to node ~A~%" a-node) + (let ((new-fd (connect (host a-node) (port a-node) connection-type))) + (update-last-connect-attempt a-node) + (when new-fd #+nio-debug (format-log t "nio-server:start-server - adding connection to nio thread ~A~%" new-fd) - (setf (gethash (async-fd-read-fd new-fd) client-hash) new-fd) - (add-async-fd event-queue new-fd :read-write)) + (setf (gethash (async-fd-read-fd new-fd) client-hash) new-fd) + (add-async-fd event-queue new-fd :read-write)))) ;loop over async-fd's processing where necessary (process-async-fds client-hash) @@ -164,10 +170,10 @@ (close-fd sock)))) -(defun add-connection (host port connection-type - &key - (protocol :inet)) - (format-log t "nio-server:add-connection - Called with: ~A:~A:~A ~%" protocol host port) +(defun connect(host port connection-type + &key + (protocol :inet)) + (format-log t "nio-server:connect - Called with: ~A:~A:~A ~%" protocol host port) (let ((sock nil)) (setq sock (ecase protocol (:inet (make-inet-socket)) @@ -175,8 +181,10 @@ (if (connect-inet-socket sock host port) (let ((sm (create-state-machine connection-type sock sock sock))) - (nio-compat:add +connected-sockets-queue+ sm) - (format-log t "nio-server:add-connection - Socket enqueued: ~A~%" +connected-sockets-queue+) - (return-from add-connection sm)) +; (nio-compat:add +connected-sockets-queue+ sm) +; (format-log t "nio-server:connect - Socket enqueued: ~A~%" +connected-sockets-queue+) + (return-from connect sm)) (format t "Connect failed!!~A ~%" (get-errno))))) - \ No newline at end of file + +(defun add-connection(node) + (nio-compat:add +connected-sockets-queue+ node)) \ No newline at end of file Modified: branches/home/psmith/restructure/src/io/nio.asd ============================================================================== --- branches/home/psmith/restructure/src/io/nio.asd (original) +++ branches/home/psmith/restructure/src/io/nio.asd Sat Feb 10 15:36:43 2007 @@ -9,9 +9,9 @@ (:file "packet" :depends-on ("nio-package")) (:file "async-fd" :depends-on ("fd-helper")) (:file "async-socket" :depends-on ("async-fd")) - (:file "nio-server" :depends-on ("async-socket")) + (:file "nodes" :depends-on ("nio-package")) + (:file "nio-server" :depends-on ("async-socket" "nodes")) (:file "ip-authorisation" :depends-on ("nio-package")) - (:file "nodes" :depends-on ("nio-package")) ) :depends-on (:cffi :event-notification :nio-buffer :nio-compat :nio-utils)) Modified: branches/home/psmith/restructure/src/io/nodes.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nodes.lisp (original) +++ branches/home/psmith/restructure/src/io/nodes.lisp Sat Feb 10 15:36:43 2007 @@ -75,6 +75,21 @@ (get-universal-high-res) (+ (last-connect-attempt node) (retry-delay node)))) +(defun allowed-to-connect(node) + (if (null (last-connect-attempt node)) + t + (and (not (active-conn node)) (< (+ (last-connect-attempt node) (retry-delay node)) (get-universal-high-res))))) + (defun update-last-connect-attempt(node) (setf (last-connect-attempt node) (get-universal-high-res))) +;;iterates over the nodes list looking for nodes that are ready to be connected to +;;i.e. the SM is null and the next-allowed-connect time has expired +(defmacro with-connect-ready-nodes ((node) &rest body) + `(dolist (,node *nodes-list*) + (when (allowed-to-connect ,node) ,@body))) + + +(defmacro with-connected-nodes ((node) &rest body) + `(dolist (,node *nodes-list*) + (when (active-conn ,node) ,@body))) Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Sat Feb 10 15:36:43 2007 @@ -41,16 +41,20 @@ (sleep ,delay)))))) +(defun callback(result) + (nio-utils:format-log t "Result of remote-log ~A~%" result)) + + ;;Tail the given log and write to remote logger ;;e.g. (tail-log "/var/log/httpd/access_log" "192.168.1.1") (defun tail-log(filename ip-address) (sleep 4) - (let ((sm (nio:add-connection ip-address 16323 'nio-yarpc:yarpc-client-state-machine))) - (nio-utils:format-log t "toplevel adding conn ~A to ~A~%" sm ip-address) - (with-line-from-tailed-file (text filename 1) - (let ((rpc (format nil "(nio-logger:remote-log \"~A\")" (cl-base64:string-to-base64-string text)))) - (nio-utils:format-log t "Toplevel Submitting job~A~%" rpc) - (nio-utils:format-log t "Result of remote-log ~A~%" (nio-yarpc:remote-execute sm rpc)))))) + (nio:add-connection (nio:node ip-address 16323)) + (with-line-from-tailed-file (text filename 1) + (let ((rpc (format nil "(nio-logger:remote-log \"~A\")" (cl-base64:string-to-base64-string text)))) + (nio-utils:format-log t "Toplevel Submitting job~A~%" rpc) + (nio:with-connected-nodes (node) + (nio-yarpc:remote-execute (nio:active-conn node) rpc #'callback))))) ;Runs a multithreaded system with an IO thread dealing with IO only and a 'job' thread taking and executing jobs Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Sat Feb 10 15:36:43 2007 @@ -36,9 +36,23 @@ ((job-queue :initform (nio-compat:concurrent-queue) :accessor job-queue :documentation "The queue used to hand off work from an external thread to the io thread") - (result-queue :initform (nio-compat:concurrent-queue) - :accessor result-queue - :documentation "The queue used to return results from the io thread to an external thread"))) + (request-map :initform (make-hash-table) + :reader request-map + :documentation "A map from request-id (a unique id for this request) to remote-job"))) + +(defclass remote-job() + ((callback :accessor callback + :documentation "A function accepting one argument to call with the result of the remote operation") + (start-time :initform (get-universal-high-res) + :reader start-time + :documentation "The (floating point) start time") + (timeout :initarg :timeout + :initform 1.5 + :documentation "The time in seconds before a timeout should occur, abviously we dont guarantee that this will be honored, it depends on other processing but should be close."))) + +(defun remote-job(callback) + (make-instance 'remote-job :callback callback)) + (defun yarpc-client-state-machine () (make-instance 'yarpc-client-state-machine)) @@ -55,25 +69,23 @@ (defconstant STATE-INITIALISED 0) (defconstant STATE-SENT-REQUEST 1) +(defparameter +request-id+ 0) + (defmethod process-outgoing-packet((sm yarpc-client-state-machine)) #+nio-debug2 (format-log t "yarpc-client-state-machine:process-outgoing-packet called, polling the job-queue ~%") - (let ((packet (nio-compat:take (job-queue sm) :blocking-call nil))) - (when packet - (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" packet) - (setf (state sm) STATE-SENT-REQUEST)) - packet)) + (let ((ttd (nio-compat:take (job-queue sm) :blocking-call nil))) + (when ttd + (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" ttd) + (destructuring-bind (job call-string) ttd + (setf (gethash (1+ +request-id+) (request-map sm)) job) + (make-instance 'call-method-packet :call-string call-string :request-id +request-id+))))) (defmethod process-incoming-packet ((sm yarpc-client-state-machine) (response method-response-packet)) - (assert (eql (state sm) STATE-SENT-REQUEST)) (format-log t "yarpc-client-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm response) (let* ((*package* (find-package :nio-yarpc)) (result (read-from-string (response response)))) - (setf (state sm) STATE-INITIALISED) (nio-compat:add (result-queue sm) result))) - -;Called from an external thread i.e. *not* the nio thread -;Blocks calling thread on the remote m/c's response -(defmethod remote-execute ((sm yarpc-client-state-machine) call-string) - (assert (eql (state sm) STATE-INITIALISED)) - (nio-compat:add (job-queue sm) (make-instance 'call-method-packet :call-string call-string)) - (nio-compat:take (result-queue sm))) \ No newline at end of file + +;Execute the call-string on the remote node and call callback with the result +(defmethod remote-execute ((sm yarpc-client-state-machine) call-string callback) + (nio-compat:add (job-queue sm) '((remote-job callback) call-string))) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp Sat Feb 10 15:36:43 2007 @@ -40,19 +40,21 @@ (defconstant +PACKET-ID-SIZE+ 1) (defconstant +PACKET-LENGTH-SIZE+ 4) +;(defconstant +PACKET-REQUEST-ID+ 4) (defconstant +yarpc-packet-header-size+ (+ +PACKET-ID-SIZE+ +PACKET-LENGTH-SIZE+)) (defmethod get-packet ((pf yarpc-packet-factory) buf) (flip buf) - (if (>= (remaining buf) +yarpc-packet-header-size+) ;; First byte denotes packet ID ;;bytes 2,3,4,5 denote packet size + (if (>= (remaining buf) +yarpc-packet-header-size+) ;; First byte denotes packet ID ;;bytes 2,3,4,5 denote packet size ;; 6,7,8,9 request-id (let ((packet-id (bytebuffer-read-8 buf)) (packet-length (bytebuffer-read-32 buf))) (if (<= (- packet-length +yarpc-packet-header-size+) (remaining buf)) ;is the whole packet available in the buffer? - (let ((ret-packet (ecase packet-id - (0 (progn (format-log t "yarpc-packet-factory:get-packet - got CALL-METHOD-PACKET-ID~%") (call-method-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+))))) - (1 (progn (format-log t "yarpc-packet-factory:get-packet - got METHOD-RESPONSE-PACKET-ID~%") (method-response-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+)))))))) + (let* ((packet-request-id (bytebuffer-read-32 buf)) + (ret-packet (ecase packet-id + (0 (progn (format-log t "yarpc-packet-factory:get-packet - got CALL-METHOD-PACKET-ID~%") (call-method-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+)) :request-id packet-request-id))) + (1 (progn (format-log t "yarpc-packet-factory:get-packet - got METHOD-RESPONSE-PACKET-ID~%") (method-response-packet (bytebuffer-read-string buf (- packet-length +yarpc-packet-header-size+)) :request-id packet-request-id)))))) (compact buf) #+nio-debug (format-log t "yarpc-packet-factory:get-packet - after compact ~%~A~%" buf) #+nio-debug (format-log t "yarpc-packet-factory:get-packet - retuirning packet ~A~%" ret-packet) @@ -64,7 +66,11 @@ -(defclass call-method-packet (packet)((call-string :initarg :call-string +(defclass yarpc-packet(packet) + ((request-id :initarg :request-id + :reader request-id))) + +(defclass call-method-packet (yarpc-packet)((call-string :initarg :call-string :accessor call-string))) (defun call-method-packet (call-string) (make-instance 'call-method-packet :call-string call-string)) @@ -79,6 +85,7 @@ (progn (nio-buffer:bytebuffer-write-8 buf +CALL-METHOD-PACKET-ID+) (nio-buffer:bytebuffer-write-32 buf 0) ; come back and write length later + (nio-buffer:bytebuffer-write-32 buf (request-id packet)) (nio-buffer:bytebuffer-write-string buf (call-string packet)) (nio-buffer:bytebuffer-insert-32 buf (buffer-position buf) 1) #+nio-debug (format-log t "yarpc-packet-factory:write-bytes(call-method-packet) - written ~%~A ~%" buf) @@ -92,7 +99,7 @@ (+ +yarpc-packet-header-size+ (length (sb-ext:string-to-octets (write-to-string (call-string packet)))))) -(defclass method-response-packet (packet) +(defclass method-response-packet (yarpc-packet) ((response :initarg :response :accessor response))) @@ -109,6 +116,7 @@ (progn (nio-buffer:bytebuffer-write-8 buf +METHOD-RESPONSE-PACKET-ID+) (nio-buffer:bytebuffer-write-32 buf 0) ; come back and write length later + (nio-buffer:bytebuffer-write-32 buf (request-id packet)) (nio-buffer:bytebuffer-write-string buf (write-to-string (response packet))) (nio-buffer:bytebuffer-insert-32 buf (buffer-position buf) 1) #+nio-debug (format-log t "yarpc-packet-factory:write-bytes - written ~A~%" buf) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Sat Feb 10 15:36:43 2007 @@ -58,26 +58,25 @@ -(defun run-job(&key (wait-on-job-pdw t)) +(defun run-job(&key (blocking t)) (format-log t "yarpc-state-machine:run-job - Server toplevel waiting for job~%") - (destructuring-bind (job result-queue) (nio-compat:take nio-yarpc:job-queue :blocking-call wait-on-job-pdw) + (destructuring-bind (job request-id result-queue) (nio-compat:take nio-yarpc:job-queue :blocking-call blocking) (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job) - (nio-compat:add result-queue (nio-yarpc:execute-call job)))) + (nio-compat:add result-queue (list request-id (nio-yarpc:execute-call job))))) (defmethod process-outgoing-packet((sm yarpc-state-machine)) (format-log t "yarpc-state-machine:process-outgoing-packet - called, polling the results-queue ~%" ) - (let ((result (nio-compat:take (result-queue sm) :blocking-call nil))) - (format-log t "yarpc-state-machine:process-outgoing-packet - got result ~A ~%" result) + (destructuring-bind (request-id result) (nio-compat:take (result-queue sm) :blocking-call nil) + (format-log t "yarpc-state-machine:process-outgoing-packet - got :request-id ~A result ~A ~%" request-id result) (when result - (method-response-packet result)))) + (method-response-packet result :request-id request-id)))) ;Process a call method packet by placing it in the job-queue (defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet)) - (assert (eql (state sm) STATE-INITIALISED)) (format-log t "yarpc-state-machine:process-incoming-packet - called :sm ~A :packet ~A~%" sm call) - (nio-compat:add job-queue (list (call-string call) (result-queue sm))) - (when +process-jobs-inline+ (run-job :wait-on-job-pdw nil))) + (nio-compat:add job-queue (list (call-string call) (request-id call) (result-queue sm))) + (when +process-jobs-inline+ (run-job :blocking nil)))

1 0

[nio-cvs] r80 - in branches/home/psmith/restructure/src: io utils
by psmith＠common-lisp.net 10 Feb '07

10 Feb '07

Author: psmith Date: Fri Feb 9 20:06:30 2007 New Revision: 80 Added: branches/home/psmith/restructure/src/io/nodes.lisp Modified: branches/home/psmith/restructure/src/io/nio.asd branches/home/psmith/restructure/src/utils/nio-utils-package.lisp Log: start of nodes Modified: branches/home/psmith/restructure/src/io/nio.asd ============================================================================== --- branches/home/psmith/restructure/src/io/nio.asd (original) +++ branches/home/psmith/restructure/src/io/nio.asd Fri Feb 9 20:06:30 2007 @@ -11,6 +11,7 @@ (:file "async-socket" :depends-on ("async-fd")) (:file "nio-server" :depends-on ("async-socket")) (:file "ip-authorisation" :depends-on ("nio-package")) + (:file "nodes" :depends-on ("nio-package")) ) :depends-on (:cffi :event-notification :nio-buffer :nio-compat :nio-utils)) Added: branches/home/psmith/restructure/src/io/nodes.lisp ============================================================================== --- (empty file) +++ branches/home/psmith/restructure/src/io/nodes.lisp Fri Feb 9 20:06:30 2007 @@ -0,0 +1,80 @@ +#| +Copyright (c) 2007 +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))) + +;;concept of a remote socket +(defclass node() + ((host :initarg :host + :reader host) + (port :initarg :port + :reader port) + (last-connect-attempt :initform nil + :accessor last-connect-attempt + :documentation "Time we last attempted a connection") + (retry-delay :initform 600 + :accessor retry-delay + :documentation "The delay to wait after the last-connection-attempt before trying to connect again") + (active-conn :initform nil + :accessor active-conn + :documentation "If we are connected to this remote socket this is set to the SM"))) + +(defun node(host port) + (make-instance 'node :host host :port port)) + +;(node-from-socket-repn "192.168.1.1:1234") +(defun node-from-socket-repn(socket) + (let ((colon-idx (search ":" socket))) + (if colon-idx + (node (subseq socket 0 colon-idx) (subseq socket (+ colon-idx 1))) + (error 'parse-error)))) + + +(defmethod print-object ((a-node node) stream) + (with-slots (host port last-connect-attempt retry-delay active-conn) a-node + (format stream "#<NODE :HOST ~A :port ~A :last-connect-attempt ~A :retry-delay ~A :active-conn ~A>" host port last-connect-attempt retry-delay active-conn))) + + +(defparameter *nodes-list* nil + "List of nodes to connect to") + +(defun load-nodes (filename) + (with-open-file (stream filename) + (loop for line = (read-line stream nil nil) do + (push (node-from-socket-repn line) *nodes-list*)))) + + +;;returns floating point (high-res) next allowed connect time +(defun get-next-allowed-connect-time(node) + (if (null (last-connect-attempt node)) + (get-universal-high-res) + (+ (last-connect-attempt node) (retry-delay node)))) + +(defun update-last-connect-attempt(node) + (setf (last-connect-attempt node) (get-universal-high-res))) + Modified: branches/home/psmith/restructure/src/utils/nio-utils-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/utils/nio-utils-package.lisp (original) +++ branches/home/psmith/restructure/src/utils/nio-utils-package.lisp Fri Feb 9 20:06:30 2007 @@ -29,5 +29,5 @@ (:export ;;utils - format-log + format-log get-universal-high-res ))

1 0

[nio-cvs] r79 - branches/home/psmith/restructure/src/nio-logger
by psmith＠common-lisp.net 08 Feb '07

08 Feb '07

Author: psmith Date: Wed Feb 7 19:31:49 2007 New Revision: 79 Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Log: Removed unnecessary logging Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Wed Feb 7 19:31:49 2007 @@ -37,7 +37,7 @@ (if ,line (progn ,@body) (progn - (format t "read nil~%") +; (format t "read nil~%") (sleep ,delay))))))

1 0

[nio-cvs] r78 - in branches/home/psmith/restructure/src: io protocol/yarpc statemachine
by psmith＠common-lisp.net 07 Feb '07

07 Feb '07

Author: psmith Date: Wed Feb 7 01:03:21 2007 New Revision: 78 Modified: branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp branches/home/psmith/restructure/src/statemachine/state-machine.lisp Log: verbosity changes Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Wed Feb 7 01:03:21 2007 @@ -41,7 +41,7 @@ (defun process-async-fds (client-hash) (let ((removals nil)) (maphash #'(lambda (k async-fd) - (format-log t "Dealing with ~a => ~a~%" k async-fd) +#+nio-debug2 (format-log t "Dealing with ~a => ~a~%" k async-fd) ;process reads (when (read-ready async-fd) (read-more async-fd)) @@ -153,7 +153,7 @@ #+nio-debug2 (format-log t "nio-server:start-server - Processing client add ~A~%" +connected-sockets-queue+) (loop for new-fd = (nio-compat:take +connected-sockets-queue+ :blocking-call nil) until (null new-fd) do - (format-log t "nio-server:start-server - Dealing with ~A~%" new-fd) +#+nio-debug (format-log t "nio-server:start-server - adding connection to nio thread ~A~%" new-fd) (setf (gethash (async-fd-read-fd new-fd) client-hash) new-fd) (add-async-fd event-queue new-fd :read-write)) Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Wed Feb 7 01:03:21 2007 @@ -56,12 +56,12 @@ (defconstant STATE-SENT-REQUEST 1) (defmethod process-outgoing-packet((sm yarpc-client-state-machine)) - (format-log t "yarpc-client-state-machine:process-outgoing-packet called, polling the job-queue ~%") +#+nio-debug2 (format-log t "yarpc-client-state-machine:process-outgoing-packet called, polling the job-queue ~%") (let ((packet (nio-compat:take (job-queue sm) :blocking-call nil))) (when packet (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" packet) (setf (state sm) STATE-SENT-REQUEST)) - packet)) + packet)) (defmethod process-incoming-packet ((sm yarpc-client-state-machine) (response method-response-packet)) (assert (eql (state sm) STATE-SENT-REQUEST)) Modified: branches/home/psmith/restructure/src/statemachine/state-machine.lisp ============================================================================== --- branches/home/psmith/restructure/src/statemachine/state-machine.lisp (original) +++ branches/home/psmith/restructure/src/statemachine/state-machine.lisp Wed Feb 7 01:03:21 2007 @@ -72,8 +72,8 @@ (defmethod process-write((sm state-machine)) (with-slots (foreign-write-buffer) sm (let ((outgoing-packet (process-outgoing-packet sm))) - (format-log t "state-machine::process-write - outgoing packet: ~A~%" outgoing-packet) (when outgoing-packet +#+nio-debug (format-log t "state-machine::process-write - outgoing packet: ~A~%" outgoing-packet) (handler-case (write-bytes outgoing-packet foreign-write-buffer) (buffer-too-small-error (write-error1)

1 0

[nio-cvs] r77 - in branches/home/psmith/restructure/src: io nio-logger
by psmith＠common-lisp.net 07 Feb '07

07 Feb '07

Author: psmith Date: Tue Feb 6 23:55:55 2007 New Revision: 77 Modified: branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp Log: nio-server Only listen if necessary Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Tue Feb 6 23:55:55 2007 @@ -65,35 +65,30 @@ -(defun start-server (connection-type +(defun start-server (connection-type &key - (protocol :inet) - (port (+ (random 60000) 1024)) - (host "localhost") + (protocol :inet) + (port (+ (random 60000) 1024)) + (host "127.0.0.1") (accept-connection #'trivial-accept)) - - (let (sock (event-queue (make-event-queue)) (client-hash (make-hash-table :test 'eql)) ) - (setq sock (ecase protocol - (:inet (make-inet-socket)) - (:inet6 (make-inet6-socket)))) - - (unless (ecase protocol - (:inet (bind-inet-socket sock port host)) - (:inet6 (bind-inet6-socket sock port host))) - (error "Can't bind socket!")) - - (set-fd-nonblocking sock) - - (format t "~&Starting server on ~S port ~S.. (socket fd is ~D)~%" host port sock) - - (start-listen sock) - - (add-fd event-queue sock :read :trigger :level) + (when (not (null connection-type)) + (format t "Binding to ~A:~A~%" host port) + (setq sock (ecase protocol + (:inet (make-inet-socket)) + (:inet6 (make-inet6-socket)))) + (unless (ecase protocol + (:inet (bind-inet-socket sock port host)) + (:inet6 (bind-inet6-socket sock port host))) + (error "Can't bind socket!")) + (set-fd-nonblocking sock) + (format t "~&Starting server on ~S port ~S.. (socket fd is ~D)~%" host port sock) + (start-listen sock) + (add-fd event-queue sock :read :trigger :level)) (format t "waiting for events..~%") (force-output) @@ -108,8 +103,10 @@ (loop for (fd . event) in unix-epoll-events do (cond ;; new connection - ((= fd sock) - (let ((async-fd (socket-accept fd connection-type))) + ((and sock (= fd sock)) + (progn +#+nio-debug (format t "start-server - incomming conn") + (let ((async-fd (socket-accept fd connection-type))) #+nio-debug (format t "start-server - New conn: ~A~%" async-fd) (cond ((null async-fd) @@ -131,7 +128,7 @@ ;; no accept, close (t (format-log t "start-server - accept-connection closed~%") - (close-async-fd async-fd))))) + (close-async-fd async-fd)))))) ;; socket i/o available @@ -168,10 +165,9 @@ (defun add-connection (host port connection-type - &key - (protocol :inet) - - ) + &key + (protocol :inet)) + (format-log t "nio-server:add-connection - Called with: ~A:~A:~A ~%" protocol host port) (let ((sock nil)) (setq sock (ecase protocol (:inet (make-inet-socket)) Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Tue Feb 6 23:55:55 2007 @@ -44,8 +44,7 @@ ;;Tail the given log and write to remote logger ;;e.g. (tail-log "/var/log/httpd/access_log" "192.168.1.1") (defun tail-log(filename ip-address) - ;;shouldn't be listenting on the client hence nil for accept SM to start-server - (sb-thread:make-thread #'(lambda()(nio:start-server nil)) :name "nio-server") + (sleep 4) (let ((sm (nio:add-connection ip-address 16323 'nio-yarpc:yarpc-client-state-machine))) (nio-utils:format-log t "toplevel adding conn ~A to ~A~%" sm ip-address) (with-line-from-tailed-file (text filename 1) Modified: branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/run-logging-client.lisp Tue Feb 6 23:55:55 2007 @@ -33,4 +33,7 @@ (let ((log-file (second sb-ext:*posix-argv*)) (ip (third sb-ext:*posix-argv*))) (format t "Starting logging client with ~A ~A~%" log-file ip) - (nio-logger:tail-log log-file ip)) + (sb-thread:make-thread #'(lambda()(nio-logger:tail-log log-file ip)) :name "nio-server") + + ;;shouldn't be listenting on the client hence nil for accept SM to start-server + (nio:start-server nil))

1 0

[nio-cvs] r76 - in branches/home/psmith/restructure: . src/event src/io src/nio-logger
by psmith＠common-lisp.net 07 Feb '07

07 Feb '07

Author: psmith Date: Tue Feb 6 20:48:24 2007 New Revision: 76 Modified: branches/home/psmith/restructure/run-http.lisp branches/home/psmith/restructure/src/event/epoll.lisp branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Log: Thread safe connect; tidy up Modified: branches/home/psmith/restructure/run-http.lisp ============================================================================== --- branches/home/psmith/restructure/run-http.lisp (original) +++ branches/home/psmith/restructure/run-http.lisp Tue Feb 6 20:48:24 2007 @@ -1,4 +1,4 @@ (push :nio-debug *features*) (require :asdf) (require :nio-http) -(nio:start-server 'identity 'identity 'nio-http:http-state-machine :host "127.0.0.1") +(nio:start-server 'nio-http:http-state-machine :host "127.0.0.1") Modified: branches/home/psmith/restructure/src/event/epoll.lisp ============================================================================== --- branches/home/psmith/restructure/src/event/epoll.lisp (original) +++ branches/home/psmith/restructure/src/event/epoll.lisp Tue Feb 6 20:48:24 2007 @@ -73,14 +73,14 @@ (define-condition poll-error (error) ()) (defun poll-events (event-queue) -#+nio-debug (format t "poll-events called with :event-queue ~A~%" event-queue) +#+nio-debug2 (format t "poll-events called with :event-queue ~A~%" event-queue) (with-foreign-object (events 'epoll-event +epoll-size+) (memzero events (* +epoll-event-size+ +epoll-size+)) (loop for res = (%epoll-wait event-queue events +epoll-size+ 100) do (progn -#+nio-debug (format t "poll-events - dealing with ~A~%" res) +#+nio-debug2 (format t "poll-events - dealing with ~A~%" res) (case res (-1 (let ((errno (get-errno))) Modified: branches/home/psmith/restructure/src/io/nio-server.lisp ============================================================================== --- branches/home/psmith/restructure/src/io/nio-server.lisp (original) +++ branches/home/psmith/restructure/src/io/nio-server.lisp Tue Feb 6 20:48:24 2007 @@ -34,7 +34,7 @@ t) ;TODO thread safety -(defparameter +connected-sockets+ nil +(defparameter +connected-sockets-queue+ (nio-compat:concurrent-queue) "List of sockets that have been connected and are awaiting addition to the event-notification system") ;loop over hashtable @@ -59,13 +59,13 @@ (push async-fd removals))) client-hash) (dolist (async-fd removals) - (remhash (async-fd-read-fd async-fd) client-hash)) - (format t "client-hash list ~A~%"client-hash ))) + (remhash (async-fd-read-fd async-fd) client-hash)))) +; (format t "client-hash list ~A~%"client-hash ) -(defun start-server (connection-handler accept-filter connection-type +(defun start-server (connection-type &key (protocol :inet) (port (+ (random 60000) 1024)) @@ -153,14 +153,12 @@ (when (write-event-p event) (setf (write-ready async-fd) t))))))))) ;add outgoing sockets to event queue - (format-log t "nio-server:start-server - Processing client add ~A~%" +connected-sockets+) - (loop for new-fd in +connected-sockets+ do +#+nio-debug2 (format-log t "nio-server:start-server - Processing client add ~A~%" +connected-sockets-queue+) + + (loop for new-fd = (nio-compat:take +connected-sockets-queue+ :blocking-call nil) until (null new-fd) do (format-log t "nio-server:start-server - Dealing with ~A~%" new-fd) (setf (gethash (async-fd-read-fd new-fd) client-hash) new-fd) (add-async-fd event-queue new-fd :read-write)) - - ;TODO thread safety - (setf +connected-sockets+ nil) ;loop over async-fd's processing where necessary (process-async-fds client-hash) @@ -181,8 +179,8 @@ (if (connect-inet-socket sock host port) (let ((sm (create-state-machine connection-type sock sock sock))) - (push sm +connected-sockets+) - (format-log t "nio-server:add-connection - Socket enqueued: ~A~%" +connected-sockets+) + (nio-compat:add +connected-sockets-queue+ sm) + (format-log t "nio-server:add-connection - Socket enqueued: ~A~%" +connected-sockets-queue+) (return-from add-connection sm)) (format t "Connect failed!!~A ~%" (get-errno))))) \ No newline at end of file Modified: branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp ============================================================================== --- branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp (original) +++ branches/home/psmith/restructure/src/nio-logger/nio-logger.lisp Tue Feb 6 20:48:24 2007 @@ -45,8 +45,7 @@ ;;e.g. (tail-log "/var/log/httpd/access_log" "192.168.1.1") (defun tail-log(filename ip-address) ;;shouldn't be listenting on the client hence nil for accept SM to start-server - (sb-thread:make-thread #'(lambda()(nio:start-server 'identity 'identity nil :host "127.0.0.1" :port 9897)) :name "nio-server") - (sleep 4) + (sb-thread:make-thread #'(lambda()(nio:start-server nil)) :name "nio-server") (let ((sm (nio:add-connection ip-address 16323 'nio-yarpc:yarpc-client-state-machine))) (nio-utils:format-log t "toplevel adding conn ~A to ~A~%" sm ip-address) (with-line-from-tailed-file (text filename 1) @@ -62,7 +61,7 @@ (setf nio-yarpc:+process-jobs-inline+ nil) (setf +log-file-name+ out-file) (nio:load-ips allowed-ips) - (sb-thread:make-thread #'(lambda()(nio:start-server 'identity 'identity 'nio-yarpc:yarpc-state-machine :host listen-ip :accept-connection 'nio:check-ip)) :name "nio-server") + (sb-thread:make-thread #'(lambda()(nio:start-server 'nio-yarpc:yarpc-state-machine :host listen-ip :accept-connection 'nio:check-ip)) :name "nio-server") (loop ;;block waiting for jobs (nio-yarpc:run-job)))

1 0

← Newer
1
2
3
4
5
6
7
...
12
Older →

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

2008

2007

2006

nio-cvs