nio-cvs February 2007

nio-cvs@common-lisp.net

1 participants
41 discussions

[nio-cvs] r102 - branches/home/psmith/restructure/src/io
by psmith＠common-lisp.net 28 Feb '07

28 Feb '07

Author: psmith Date: Wed Feb 28 18:24:31 2007 New Revision: 102 Modified: branches/home/psmith/restructure/src/io/nio-server.lisp Log: Set outgoing connections to be non-blocking 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 28 18:24:31 2007 @@ -166,6 +166,7 @@ (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) + (set-fd-nonblocking (async-fd-read-fd 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))))

1 0

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

28 Feb '07

Author: psmith Date: Tue Feb 27 22:42:29 2007 New Revision: 101 Modified: branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp Log: Allow simulated remote calls for testing Modified: branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp Tue Feb 27 22:42:29 2007 @@ -37,5 +37,5 @@ test-rpc test-rpc-list test-rpc-string execute-call defremote ;;yarpc-client-state-machine - yarpc-client-state-machine remote-execute + yarpc-client-state-machine remote-execute simulate-connection )) 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 Tue Feb 27 22:42:29 2007 @@ -91,7 +91,19 @@ (let ((remote-job (gethash request-id (request-map sm)))) (funcall (callback remote-job) result)))) +(defparameter *simulate-calls* nil) + ;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-debug (format-log t "yarpc-client-state-machine:remote-execute called :sm ~A :call-string ~A :callback ~A~%" sm call-string callback) - (nio-utils:add (job-queue sm) (list (remote-job callback) call-string))) + (if *simulate-calls* + (funcall callback (execute-call call-string)) + (nio-utils:add (job-queue sm) (list (remote-job callback) call-string)))) + + +(defun simulate-connection() + (setf *simulate-calls* t) + (let* ((node (nio:node "127.0.0.1" 9999))) + (setf (nio:active-conn node) (nio::create-state-machine 'yarpc-client-state-machine 1 1 6)) + (push node nio::*nodes-list*))) +

1 0

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

27 Feb '07

Author: psmith Date: Mon Feb 26 22:00:16 2007 New Revision: 100 Modified: branches/home/psmith/restructure/src/buffer/buffer.lisp branches/home/psmith/restructure/src/buffer/nio-buffer-package.lisp branches/home/psmith/restructure/src/io/async-fd.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp branches/home/psmith/restructure/src/statemachine/state-machine.lisp Log: Fixed large packet problems: calculated header size correctly; use buffer-pointer to take into account buffer-position on write and moved external format to UTF-8 Modified: branches/home/psmith/restructure/src/buffer/buffer.lisp ============================================================================== --- branches/home/psmith/restructure/src/buffer/buffer.lisp (original) +++ branches/home/psmith/restructure/src/buffer/buffer.lisp Mon Feb 26 22:00:16 2007 @@ -197,7 +197,7 @@ ; Read bytes from bytebuffer abd return a string using the supplied decoding ;TODO move octets-to-string into nio-compat -(defmethod bytebuffer-read-string((bb byte-buffer) &optional (num-bytes-to-read (remaining bb)) (external-format :ascii)) +(defmethod bytebuffer-read-string((bb byte-buffer) &optional (num-bytes-to-read (remaining bb)) (external-format :utf-8)) (sb-ext:octets-to-string (bytebuffer-read-vector bb num-bytes-to-read) :external-format external-format)) ; Read a byte from bytebuffer and return it incrementing the byte-buffers position @@ -246,7 +246,7 @@ ;; Writes data from string str to bytebuffer using specified encoding ;TODO move string-to-octets into nio-compat -(defmethod bytebuffer-write-string((bb byte-buffer) str &optional (external-format :ascii)) +(defmethod bytebuffer-write-string((bb byte-buffer) str &optional (external-format :utf-8)) :documentation "Returns number of bytes written to bytebuffer" (bytebuffer-write-vector bb (sb-ext:string-to-octets str :external-format external-format))) Modified: branches/home/psmith/restructure/src/buffer/nio-buffer-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/buffer/nio-buffer-package.lisp (original) +++ branches/home/psmith/restructure/src/buffer/nio-buffer-package.lisp Mon Feb 26 22:00:16 2007 @@ -31,6 +31,6 @@ bytebuffer-write-vector bytebuffer-write-string bytebuffer-read-vector bytebuffer-read-string bytebuffer-read-8 bytebuffer-read-32 bytebuffer-write-8 bytebuffer-write-32 bytebuffer-insert-8 bytebuffer-insert-32 - flip unflip clear buffer-position copy-buffer buffer-capacity compact mark reset + flip unflip clear buffer-position buffer-limit copy-buffer buffer-capacity compact mark reset buffer-too-small-error recommended-size buffer-pointer )) 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 Mon Feb 26 22:00:16 2007 @@ -106,6 +106,7 @@ (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))))))) @@ -138,7 +139,7 @@ (do ((total-written 0)) ((or (eql now-written -1) (eql (remaining foreign-write-buffer) 0)) total-written) (progn - (setf now-written (%write write-fd (buffer-buf foreign-write-buffer) (remaining foreign-write-buffer))) + (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))) @@ -166,7 +167,7 @@ (defmacro realloc-buffer(async-fd accessor size) `(let ((buffer (,accessor ,async-fd))) - (if (>= (buffer-capacity buffer) size) + (if (>= (buffer-capacity buffer) ,size) t (let ((new-buffer (byte-buffer ,size))) (copy-buffer buffer new-buffer) @@ -181,7 +182,6 @@ (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) @@ -208,3 +208,12 @@ (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)))))) + 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 Mon Feb 26 22:00:16 2007 @@ -45,6 +45,8 @@ (defconstant +yarpc-packet-header-size+ (+ +PACKET-ID-SIZE+ +PACKET-LENGTH-SIZE+)) +(defconstant +yarpc-rpc-packet-header-size+ (+ +yarpc-packet-header-size+ +PACKET-REQUEST-ID-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 ;; 6,7,8,9 request-id @@ -92,7 +94,7 @@ (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-write-string buf (call-string packet) :utf-8) (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) ) @@ -102,8 +104,8 @@ (defmethod get-packet-size ((packet call-method-packet)) - (+ +yarpc-packet-header-size+ - (length (sb-ext:string-to-octets (write-to-string (call-string packet)))))) + (+ +yarpc-rpc-packet-header-size+ + (length (sb-ext:string-to-octets (call-string packet) :external-format :utf-8)))) (defclass method-response-packet (yarpc-packet) ((response :initarg :response @@ -131,8 +133,9 @@ ) (buffer-too-small-error (err) (nio-buffer:reset buf) +#+nio-debug (format-log t "yarpc-packet-factory:write-bytes - buffer too small caught, reset to ~A~%" buf) (error err)))) (defmethod get-packet-size ((packet method-response-packet)) - (+ +yarpc-packet-header-size+ + (+ +yarpc-rpc-packet-header-size+ (length (sb-ext:string-to-octets (funcall +serialise-packet-fn+ (response packet)) :external-format :utf-8)))) 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 Mon Feb 26 22:00:16 2007 @@ -79,11 +79,13 @@ (handler-case (write-bytes outgoing-packet foreign-write-buffer) (buffer-too-small-error (write-error1) - (if (recommend-buffer-size sm :write (get-packet-size outgoing-packet)) + (let ((new-size (get-packet-size outgoing-packet))) + (format-log t "state-machine::process-write - write-error1 trying resize to ~A" new-size) + (if (recommend-buffer-size sm :write new-size) (handler-case (write-bytes outgoing-packet foreign-write-buffer) - (buffer-too-small-error (write-error1) (format t "Failed to write packet after resize (something already in write buffer?, dropping packet ~A~% out buffer:~%~A~%" outgoing-packet foreign-write-buffer))) - (format t "Failed to resize io buffer, dropping packet: ~A~%" outgoing-packet)))))))) + (buffer-too-small-error (write-error2) (format t "Failed to write packet after resize (something already in write buffer?, dropping packet ~A~% out buffer:~%~A~%" outgoing-packet foreign-write-buffer))) + (format t "Failed to resize io buffer, dropping packet: ~A~%" outgoing-packet)))))))))

1 0

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

26 Feb '07

Author: psmith Date: Mon Feb 26 18:13:58 2007 New Revision: 99 Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp Log: Changed method-response-packet to write utf-8 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 Mon Feb 26 18:13:58 2007 @@ -125,7 +125,7 @@ (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 (funcall +serialise-packet-fn+ (response packet))) + (nio-buffer:bytebuffer-write-string buf (funcall +serialise-packet-fn+ (response packet)) :utf-8) (nio-buffer:bytebuffer-insert-32 buf (buffer-position buf) 1) #+nio-debug (format-log t "yarpc-packet-factory:write-bytes - written ~A~%" buf) ) @@ -135,4 +135,4 @@ (defmethod get-packet-size ((packet method-response-packet)) (+ +yarpc-packet-header-size+ - (length (funcall +serialise-packet-fn+ (response packet))))) + (length (sb-ext:string-to-octets (funcall +serialise-packet-fn+ (response packet)) :external-format :utf-8))))

1 0

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

26 Feb '07

Author: psmith Date: Mon Feb 26 17:57:14 2007 New Revision: 98 Modified: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp Log: Use +serialise-packaet-fn+ to calculate size of packet when resizing 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 Mon Feb 26 17:57:14 2007 @@ -135,4 +135,4 @@ (defmethod get-packet-size ((packet method-response-packet)) (+ +yarpc-packet-header-size+ - (length (sb-ext:string-to-octets (write-to-string (response packet)))))) + (length (funcall +serialise-packet-fn+ (response packet)))))

1 0

[nio-cvs] r97 - branches/home/psmith/restructure/src/io
by psmith＠common-lisp.net 25 Feb '07

25 Feb '07

Author: psmith Date: Sun Feb 25 12:59:02 2007 New Revision: 97 Modified: branches/home/psmith/restructure/src/io/nio-package.lisp branches/home/psmith/restructure/src/io/nodes.lisp Log: Added nio util for parsing sockets into nodes 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 Sun Feb 25 12:59:02 2007 @@ -44,5 +44,5 @@ check-ip load-ips ;;nodes - node with-connected-nodes active-conn + node with-connected-nodes active-conn node-from-socket-repn )) 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 25 12:59:02 2007 @@ -55,7 +55,7 @@ (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))) + (node (subseq socket 0 colon-idx) (parse-integer (subseq socket (+ colon-idx 1)))) (error 'parse-error))))

1 0

[nio-cvs] r96 - branches/home/psmith/restructure/src/compat
by psmith＠common-lisp.net 23 Feb '07

23 Feb '07

Author: psmith Date: Fri Feb 23 11:58:29 2007 New Revision: 96 Added: branches/home/psmith/restructure/src/compat/threading.lisp Log: Added missing file Added: branches/home/psmith/restructure/src/compat/threading.lisp ============================================================================== --- (empty file) +++ branches/home/psmith/restructure/src/compat/threading.lisp Fri Feb 23 11:58:29 2007 @@ -0,0 +1,40 @@ +#| +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-compat) + + +(defun make-mutex (name) + (sb-thread:make-mutex :name name)) + +(defmacro with-mutex ((mutex) &body body) + `(sb-thread:with-mutex (,mutex) + ,@body)) + + + +

1 0

[nio-cvs] r95 - branches/home/psmith/restructure/src/utils
by psmith＠common-lisp.net 23 Feb '07

23 Feb '07

Author: psmith Date: Thu Feb 22 19:18:57 2007 New Revision: 95 Modified: branches/home/psmith/restructure/src/utils/concurrent-queue.lisp Log: Fixed problem with NIL being read from queue and then waiting when we should have returned Modified: branches/home/psmith/restructure/src/utils/concurrent-queue.lisp ============================================================================== --- branches/home/psmith/restructure/src/utils/concurrent-queue.lisp (original) +++ branches/home/psmith/restructure/src/utils/concurrent-queue.lisp Thu Feb 22 19:18:57 2007 @@ -47,51 +47,42 @@ (make-instance 'concurrent-queue)) - -(defmacro pop-elt(a-buffer loc) - `(if ,a-buffer - (let ((head (car ,a-buffer))) - (setf ,a-buffer (cdr ,a-buffer)) -#+nio-debug (format-threadsafe t "concurent-queue:take - (~A,~A) read ~A at ~A~%" sb-thread:*current-thread* (length (buffer queue)) head ,loc) - head) - nil)) - - ;Do an (optionally blocking) remove of the element at the head of this queue (defmethod take ((queue concurrent-queue) &key (blocking-call t)) -#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) attempting to obtain mutex ~A~%" sb-thread:*current-thread* (buffer-lock queue)) +#+nio-debug (format-log t "concurent-queue:take - (~A) attempting to obtain mutex ~A~%" sb-thread:*current-thread* (buffer-lock queue)) (sb-thread:with-mutex ((buffer-lock queue)) -#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) aquired mutex mutex ~A~%" sb-thread:*current-thread* (buffer-lock queue)) +#+nio-debug (format-log t "concurent-queue:take - (~A) aquired mutex mutex ~A~%" sb-thread:*current-thread* (buffer-lock queue)) ;if its there, pop it - (let ((ret (pop-elt (buffer queue) "1sttry"))) - (if (or ret (not blocking-call)) - ret - (progn -#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) about to wait on queue~%" sb-thread:*current-thread*) - (sb-thread:condition-wait (buffer-queue queue) (buffer-lock queue)) -#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) notified on queue~%" sb-thread:*current-thread*) - (pop-elt (buffer queue) "2ndtry")))))) + (if (> (length (buffer queue)) 0) + (pop (buffer queue)) + (when blocking-call + (loop +#+nio-debug (format-log t "concurent-queue:take - (~A) about to wait on queue~%" sb-thread:*current-thread*) + (sb-thread:condition-wait (buffer-queue queue) (buffer-lock queue)) +#+nio-debug (format-log t "concurent-queue:take - (~A) notified on queue~%" sb-thread:*current-thread*) + (if (> (length (buffer queue)) 0) + (return-from take (pop (buffer queue))))))))) ;Append the element to the tail of this queue (defmethod add ((queue concurrent-queue) elt) -#+nio-debug (format-threadsafe t "concurent-queue:add - (~A) adding ~A~%" sb-thread:*current-thread* elt) +#+nio-debug (format-log t "concurent-queue:add - (~A) adding ~A~%" sb-thread:*current-thread* elt) (sb-thread:with-mutex ((buffer-lock queue)) (setf (buffer queue) (append (buffer queue) (list elt)) ) - (sb-thread:condition-broadcast (buffer-queue queue)))) + (sb-thread:condition-notify (buffer-queue queue)))) (defun test-writer(queue) (loop for i from 0 to 100 do ; (sleep (random 0.1)) - (format-threadsafe t "Adding ~A~%" i) + (format-log t "Adding ~A~%" i) (add queue i))) (defun test-reader(queue results) - (format-threadsafe t "Started reader ~A~%" sb-thread:*current-thread*) + (format-log t "Started reader ~A~%" sb-thread:*current-thread*) (loop (let ((elt (take queue))) (push elt results) - (format-threadsafe t "reader on ~A got elt ~A~%" + (format-log t "reader on ~A got elt ~A~%" sb-thread:*current-thread* results)))) @@ -105,8 +96,8 @@ (let ((t1 (sb-thread:make-thread #'(lambda()(test-reader queue *results1*))))) ;(t2 (sb-thread:make-thread #'(lambda()(test-reader queue *results2*))))) (sleep 5) ;;wait for it to probably complete - (format-threadsafe t "t1 got: ~A~%" *results1*) - (format-threadsafe t "t2 got: ~A~%" *results2*) + (format-log t "t1 got: ~A~%" *results1*) + (format-log t "t2 got: ~A~%" *results2*) (sb-thread:destroy-thread t1) ; (sb-thread:destroy-thread t2) ) @@ -120,6 +111,6 @@ (sb-thread:make-thread #'(lambda()(test-reader queue))) (sb-thread:make-thread #'(lambda()(test-writer queue))) (sleep 10) - (format-threadsafe t "running asserts") + (format-log t "running asserts") (sb-thread:with-mutex ((buffer-lock queue)) (assert (eql (length (buffer queue)) 0)))))

1 0

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

22 Feb '07

Author: psmith Date: Thu Feb 22 17:50:56 2007 New Revision: 94 Added: branches/home/psmith/restructure/src/utils/concurrent-queue.lisp - copied, changed from r93, branches/home/psmith/restructure/src/compat/concurrent-queue.lisp Removed: branches/home/psmith/restructure/src/compat/concurrent-queue.lisp Modified: branches/home/psmith/restructure/src/compat/nio-compat-package.lisp branches/home/psmith/restructure/src/compat/nio-compat.asd branches/home/psmith/restructure/src/io/nio-server.lisp branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc.asd branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp branches/home/psmith/restructure/src/utils/nio-utils-package.lisp branches/home/psmith/restructure/src/utils/nio-utils.asd branches/home/psmith/restructure/src/utils/utils.lisp Log: moved threadsafe queue and added more tests. Modified: branches/home/psmith/restructure/src/compat/nio-compat-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/compat/nio-compat-package.lisp (original) +++ branches/home/psmith/restructure/src/compat/nio-compat-package.lisp Thu Feb 22 17:50:56 2007 @@ -30,7 +30,6 @@ ;; errno.lisp get-errno +ERRNO_EAGAIN+ perror - - ;;concurrent-queue - concurrent-queue add take + ;;threading + with-mutex make-mutex )) Modified: branches/home/psmith/restructure/src/compat/nio-compat.asd ============================================================================== --- branches/home/psmith/restructure/src/compat/nio-compat.asd (original) +++ branches/home/psmith/restructure/src/compat/nio-compat.asd Thu Feb 22 17:50:56 2007 @@ -6,7 +6,7 @@ :components ((:file "nio-compat-package") (:file "errno" :depends-on ("nio-compat-package")) - (:file "concurrent-queue" :depends-on ("nio-compat-package")) + (:file "threading" :depends-on ("nio-compat-package")) ) :depends-on (:cffi)) 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 Thu Feb 22 17:50:56 2007 @@ -34,7 +34,7 @@ t) ;TODO thread safety -(defparameter +connected-sockets-queue+ (nio-compat:concurrent-queue) +(defparameter +connected-sockets-queue+ (nio-utils:concurrent-queue) "List of node objects that are to be connected to") ;loop over hashtable @@ -155,7 +155,7 @@ ;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 + (loop for node = (nio-utils: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*)) @@ -192,4 +192,4 @@ (format t "Connect failed!!~A ~%" (get-errno))))) (defun add-connection(node) - (nio-compat:add +connected-sockets-queue+ node)) \ No newline at end of file + (nio-utils:add +connected-sockets-queue+ node)) \ No newline at end of file Modified: branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc.asd ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc.asd (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc.asd Thu Feb 22 17:50:56 2007 @@ -10,4 +10,4 @@ (:file "yarpc-client-state-machine" :depends-on ("yarpc-packet-factory")) ) - :depends-on (:nio :nio-sm :nio-compat)) \ No newline at end of file + :depends-on (:nio :nio-sm :nio-utils)) \ No newline at end of file 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 Thu Feb 22 17:50:56 2007 @@ -33,7 +33,7 @@ ;; A client that accepts jobs to be run via a threadsafe queue and then submits them to the remote end for execution ;; (defclass yarpc-client-state-machine (state-machine) - ((job-queue :initform (nio-compat:concurrent-queue) + ((job-queue :initform (nio-utils:concurrent-queue) :accessor job-queue :documentation "The queue used to hand off work from an external thread to the io thread") (request-map :initform (make-hash-table) @@ -74,7 +74,7 @@ (defmethod process-outgoing-packet((sm yarpc-client-state-machine)) #+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))) + (let ((ttd (nio-utils:take (job-queue sm) :blocking-call nil))) (when ttd #+nio-debug (format-log t "yarpc-client-state-machine:process-outgoing-packet got job ~A ~%" ttd) (destructuring-bind (job call-string) ttd @@ -94,4 +94,4 @@ ;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-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))) + (nio-utils:add (job-queue sm) (list (remote-job callback) call-string))) 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 Thu Feb 22 17:50:56 2007 @@ -34,11 +34,11 @@ ;; (defclass yarpc-state-machine (state-machine) ( - (result-queue :initform (nio-compat:concurrent-queue) + (result-queue :initform (nio-utils:concurrent-queue) :accessor result-queue :documentation "The queue used to return results from an external thread to the nio thread"))) -(defparameter job-queue (nio-compat:concurrent-queue) +(defparameter job-queue (nio-utils:concurrent-queue) "The queue used to hand off work from the NIO thread to an external thread for execution") (defparameter yarpc-pf (yarpc-packet-factory)) @@ -60,16 +60,16 @@ (defun run-job(&key (blocking t)) #+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))) + (let ((server-job (nio-utils:take nio-yarpc:job-queue :blocking-call blocking))) (when server-job (destructuring-bind (job request-id result-queue) server-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))))))) + (nio-utils:add result-queue (list request-id (nio-yarpc:execute-call job))))))) (defmethod process-outgoing-packet((sm yarpc-state-machine)) #+nio-debug2 (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))) + (let ((server-job (nio-utils:take (result-queue sm) :blocking-call nil))) (when server-job (destructuring-bind (request-id result) server-job #+nio-debug (format-log t "yarpc-state-machine:process-outgoing-packet - got :request-id ~A result ~A ~%" request-id result) @@ -78,7 +78,7 @@ ;Process a call method packet by placing it in the job-queue (defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet)) #+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))) + (nio-utils:add job-queue (list (call-string call) (request-id call) (result-queue sm))) (when +process-jobs-inline+ (run-job :blocking nil))) Copied: branches/home/psmith/restructure/src/utils/concurrent-queue.lisp (from r93, branches/home/psmith/restructure/src/compat/concurrent-queue.lisp) ============================================================================== --- branches/home/psmith/restructure/src/compat/concurrent-queue.lisp (original) +++ branches/home/psmith/restructure/src/utils/concurrent-queue.lisp Thu Feb 22 17:50:56 2007 @@ -25,16 +25,19 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |# -(in-package :nio-compat) +(in-package :nio-utils) (declaim (optimize (debug 3) (speed 3) (space 0))) ;Implements a threadsafe queue where readers wait for elements of a FIFO queue to appear using a waitqueue ;Modified from sbcl manual example + (defclass concurrent-queue() ((buffer-queue :initform (sb-thread:make-waitqueue) :reader buffer-queue) +; (buffer-queue-mutex :initform (sb-thread:make-mutex :name "buffer queue mutex") +; :reader buffer-queue-mutex) (buffer-lock :initform (sb-thread:make-mutex :name "buffer lock") :reader buffer-lock) (buffer :initform nil @@ -43,47 +46,80 @@ (defun concurrent-queue() (make-instance 'concurrent-queue)) + + (defmacro pop-elt(a-buffer loc) `(if ,a-buffer (let ((head (car ,a-buffer))) (setf ,a-buffer (cdr ,a-buffer)) -#+nio-debug (format t "concurent-queue:take - (~A) read ~A at ~A~%" sb-thread:*current-thread* head ,loc) +#+nio-debug (format-threadsafe t "concurent-queue:take - (~A,~A) read ~A at ~A~%" sb-thread:*current-thread* (length (buffer queue)) head ,loc) head) nil)) + ;Do an (optionally blocking) remove of the element at the head of this queue (defmethod take ((queue concurrent-queue) &key (blocking-call t)) +#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) attempting to obtain mutex ~A~%" sb-thread:*current-thread* (buffer-lock queue)) (sb-thread:with-mutex ((buffer-lock queue)) +#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) aquired mutex mutex ~A~%" sb-thread:*current-thread* (buffer-lock queue)) ;if its there, pop it (let ((ret (pop-elt (buffer queue) "1sttry"))) (if (or ret (not blocking-call)) ret (progn +#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) about to wait on queue~%" sb-thread:*current-thread*) (sb-thread:condition-wait (buffer-queue queue) (buffer-lock queue)) +#+nio-debug (format-threadsafe t "concurent-queue:take - (~A) notified on queue~%" sb-thread:*current-thread*) (pop-elt (buffer queue) "2ndtry")))))) ;Append the element to the tail of this queue (defmethod add ((queue concurrent-queue) elt) -#+nio-debug (format t "concurent-queue:add - (~A) adding ~A~%" sb-thread:*current-thread* elt) +#+nio-debug (format-threadsafe t "concurent-queue:add - (~A) adding ~A~%" sb-thread:*current-thread* elt) (sb-thread:with-mutex ((buffer-lock queue)) (setf (buffer queue) (append (buffer queue) (list elt)) ) - (sb-thread:condition-notify (buffer-queue queue)))) - + (sb-thread:condition-broadcast (buffer-queue queue)))) (defun test-writer(queue) - (loop for i from 0 to 999 do - (sleep 0.1) + (loop for i from 0 to 100 do +; (sleep (random 0.1)) + (format-threadsafe t "Adding ~A~%" i) (add queue i))) -(defun test-reader(queue) +(defun test-reader(queue results) + (format-threadsafe t "Started reader ~A~%" sb-thread:*current-thread*) (loop - (format t "reader on ~A got elt ~A~%" - sb-thread:*current-thread* (take queue)))) + (let ((elt (take queue))) + (push elt results) + (format-threadsafe t "reader on ~A got elt ~A~%" + sb-thread:*current-thread* + results)))) + +(defparameter *results1* (list 999999)) +(defparameter *results2* (list 888888)) (defun test-queue() (let ((queue (make-instance 'concurrent-queue))) (sb-thread:make-thread #'(lambda()(test-writer queue))) - (sleep 10) +; (sleep 10) + (let ((t1 (sb-thread:make-thread #'(lambda()(test-reader queue *results1*))))) + ;(t2 (sb-thread:make-thread #'(lambda()(test-reader queue *results2*))))) + (sleep 5) ;;wait for it to probably complete + (format-threadsafe t "t1 got: ~A~%" *results1*) + (format-threadsafe t "t2 got: ~A~%" *results2*) + (sb-thread:destroy-thread t1) +; (sb-thread:destroy-thread t2) +) + (sb-thread:with-mutex ((buffer-lock queue)) + (assert (eql (length (buffer queue)) 0))))) + +(defun test-queue2() + (let ((queue (make-instance 'concurrent-queue))) + (sb-thread:make-thread #'(lambda()(test-reader queue))) + (sb-thread:make-thread #'(lambda()(test-writer queue))) (sb-thread:make-thread #'(lambda()(test-reader queue))) - (sb-thread:make-thread #'(lambda()(test-reader queue))))) + (sb-thread:make-thread #'(lambda()(test-writer queue))) + (sleep 10) + (format-threadsafe t "running asserts") + (sb-thread:with-mutex ((buffer-lock queue)) + (assert (eql (length (buffer queue)) 0))))) 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 Thu Feb 22 17:50:56 2007 @@ -30,4 +30,9 @@ ;;utils format-log get-universal-high-res get-readable-time + + ;;concurrent-queue + concurrent-queue add take + + )) Modified: branches/home/psmith/restructure/src/utils/nio-utils.asd ============================================================================== --- branches/home/psmith/restructure/src/utils/nio-utils.asd (original) +++ branches/home/psmith/restructure/src/utils/nio-utils.asd Thu Feb 22 17:50:56 2007 @@ -6,7 +6,8 @@ :components ((:file "nio-utils-package") (:file "utils" :depends-on ("nio-utils-package")) + (:file "concurrent-queue" :depends-on ("utils")) ) - :depends-on ()) + :depends-on (:nio-compat)) Modified: branches/home/psmith/restructure/src/utils/utils.lisp ============================================================================== --- branches/home/psmith/restructure/src/utils/utils.lisp (original) +++ branches/home/psmith/restructure/src/utils/utils.lisp Thu Feb 22 17:50:56 2007 @@ -59,6 +59,9 @@ ) +(defparameter *format-mutex* (nio-compat:make-mutex "format lock")) + ;Format the message to destination but prepend a high res time to the message, useful for logging (defmacro format-log (destination control-string &rest format-arguments) - `(format ,destination (concatenate 'string "~A - " ,control-string) (get-readable-high-res-time) ,@format-arguments)) + `(nio-compat:with-mutex (*format-mutex*) + (format ,destination (concatenate 'string "~A - " ,control-string) (get-readable-high-res-time) ,@format-arguments)))

1 0

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

21 Feb '07

Author: psmith Date: Wed Feb 21 01:38:56 2007 New Revision: 93 Modified: branches/home/psmith/restructure/src/io/async-fd.lisp branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-packet-factory.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Log: Eval whole string on rpc, allow user defined writer, increase default buffer size 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 Wed Feb 21 01:38:56 2007 @@ -33,9 +33,9 @@ :accessor write-fd) (read-fd :initarg :read-fd :accessor read-fd) - (foreign-read-buffer :initform (byte-buffer 1024) + (foreign-read-buffer :initform (byte-buffer 2096) :accessor foreign-read-buffer) - (foreign-write-buffer :initform (byte-buffer 1024) + (foreign-write-buffer :initform (byte-buffer 2096) :accessor foreign-write-buffer) (read-ready :initform nil :accessor read-ready Modified: branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp ============================================================================== --- branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp (original) +++ branches/home/psmith/restructure/src/protocol/yarpc/nio-yarpc-package.lisp Wed Feb 21 01:38:56 2007 @@ -32,7 +32,7 @@ yarpc-state-machine-factory get-packet-factory ;; yarpc-state-machine - yarpc-state-machine job-queue run-job +process-jobs-inline+ + yarpc-state-machine job-queue run-job +process-jobs-inline+ +serialise-packet-fn+ ;to be moved test-rpc test-rpc-list test-rpc-string execute-call defremote 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 Wed Feb 21 01:38:56 2007 @@ -115,6 +115,8 @@ (defmethod print-object ((packet method-response-packet) stream) (format stream "#<METHID-RESPONSE-PACKET ~A >" (response packet))) +(defparameter +serialise-packet-fn+ #'(lambda (result)(write-to-string result))) + (defmethod write-bytes((packet method-response-packet) buf) #+nio-debug (format-log t "yarpc-packet-factory:write-bytes - writing ~A to ~A~%" packet buf) (nio-buffer:mark buf) @@ -123,7 +125,7 @@ (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-write-string buf (funcall +serialise-packet-fn+ (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 Wed Feb 21 01:38:56 2007 @@ -109,7 +109,7 @@ (fn (member (symbol-function (first rpc-call-list)) *remote-fns* ))) #+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)) + (eval rpc-call-list) (error 'authorization-error))) (reader-error (re) (format-log t "yarpc-state-machine:execute-call - reader error on call-string ~A ~%" re))))

1 0

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

2008

2007

2006

nio-cvs February 2007