Author: psmith Date: Tue Apr 17 00:26:29 2007 New Revision: 109

Modified: branches/home/psmith/restructure/src/io/async-fd.lisp branches/home/psmith/restructure/src/io/nio-package.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/protocol/yarpc/nio-yarpc-package.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-client-state-machine.lisp branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp Log: Remove +process-jobs-inline+ as can't work like this. Added timeout mechanism

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 Tue Apr 17 00:26:29 2007 @@ -62,6 +62,9 @@ ;;Implement this in concrete SM for read (defgeneric process-write (async-fd))

+;;Implement this in concrete SM for timeout processing +(defgeneric process-timeout (async-fd)) + ;;SM factory (defun create-state-machine(sm-type read-fd write-fd socket) (let ((sm (make-instance sm-type :read-fd read-fd :write-fd write-fd :socket socket)))

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 Tue Apr 17 00:26:29 2007 @@ -30,7 +30,7 @@

;; async-fd.lisp async-fd process-read process-write foreign-read-buffer foreign-write-buffer close-sm - recommend-buffer-size close-pending + recommend-buffer-size close-pending process-timeout

;; async-socket.lisp

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 Apr 17 00:26:29 2007 @@ -46,6 +46,7 @@ ;process reads (handler-case (progn + (process-timeout async-fd) (when (read-ready async-fd) (read-more async-fd)) (when (> (buffer-position (foreign-read-buffer async-fd)) 0) (process-read async-fd)) @@ -63,7 +64,8 @@ (write-more async-fd) (push async-fd removals))) (read-error (re) (push async-fd removals)) - (write-error (we) (push async-fd removals)))) + (write-error (we) (push async-fd removals)) + (timeout (to) (push async-fd removals)))) client-hash) (dolist (async-fd removals) (format-log t "nio-server:process-async-fds processing remove for ~a~%" async-fd)

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 Apr 17 00:26:29 2007 @@ -63,7 +63,6 @@ (defparameter +log-file-name+ "./out")

(defun run-logging-server(listen-ip out-file &optional (allowed-ips "ips.txt")) - (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 :port 16323 :accept-connection 'nio:check-ip)) :name "nio-server")

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 Apr 17 00:26:29 2007 @@ -32,9 +32,9 @@ yarpc-state-machine-factory get-packet-factory

;; yarpc-state-machine - yarpc-state-machine job-queue run-job +process-jobs-inline+ +serialise-packet-fn+ + yarpc-state-machine job-queue run-job +process-jobs-inline+ +serialise-packet-fn+ process-timeout ;to be moved - test-rpc test-rpc-list test-rpc-string execute-call defremote + test-rpc test-rpc-list test-rpc-string execute-call defremote process-timeout ;;yarpc-client-state-machine 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 Apr 17 00:26:29 2007 @@ -48,11 +48,12 @@ :reader start-time :documentation "The (floating point) start time") (timeout :initarg :timeout - :initform 1.5 + :initform 15 + :reader timeout :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 remote-job(callback &key (timeout 15)) + (make-instance 'remote-job :callback callback :timeout timeout))

(defun yarpc-client-state-machine () @@ -67,11 +68,37 @@ (defmethod print-object ((sm yarpc-client-state-machine) stream) (format stream "#<YARPC-CLIENT-STATE-MACHINE ~A >" (call-next-method sm nil)))

+(defmethod print-object ((job remote-job) stream) + (format stream "#<REMOTE-JOB :start-time ~A :timeout ~A>" (start-time job) (timeout job))) + + (defconstant STATE-INITIALISED 0) (defconstant STATE-SENT-REQUEST 1)

(defparameter +request-id+ 0)

+ +(defun check-timeouts(id job) +; (format-log t "Checking timeout on ~A~%" job) + (when (> (get-universal-high-res) (+ (start-time job) (timeout job))) + (format-log t "Timeout detected ~A ~A~%" id job) + t)) + +(defun finish-job (request-id sm result) + "Remove the job from the request map and call the callback with the result" + (let ((remote-job (gethash request-id (request-map sm)))) + (when remote-job + (remhash request-id (request-map sm)) + (funcall (callback remote-job) result)))) + +(defmethod process-timeout((sm yarpc-client-state-machine)) + (let ((requests (request-map sm))) +#+nio-debug (format-log t "yarpc-client-state-machine:process-outgoing-packet called, searching for timeouts in ~A ~%" requests) + (maphash #'(lambda (id job) + (when (check-timeouts id job) (finish-job id sm nil))) + requests))) + + (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-utils:take (job-queue sm) :blocking-call nil))) @@ -88,8 +115,7 @@ (request-id (request-id response))) #+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)))) + (finish-job request-id sm result)))

(defparameter *simulate-calls* nil)

@@ -107,3 +133,14 @@ (setf (nio:active-conn node) (nio::create-state-machine 'yarpc-client-state-machine 1 1 6)) (push node nio::*nodes-list*)))

+ + +(defun test-timeout() + (let* ((done nil) + (job (remote-job #'(lambda(x) (format-log t "~A finished~%" x) (setf done t)) :timeout 30))) + (format-log t "Job: ~A~%" job) + (loop while (not done) do + (check-timeouts 99 job) + (format-log t ".~%") + (sleep 1)) + (format-log t "done test~%"))) \ No newline at end of file

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 Tue Apr 17 00:26:29 2007 @@ -52,12 +52,6 @@ (defconstant STATE-INITIALISED 0) (defconstant STATE-SEND-RESPONSE 1)

- -(defparameter +process-jobs-inline+ t - "Set this to make the NIO thread process the RPC calls - warning the procedure should not block!") - - - (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-utils:take nio-yarpc:job-queue :blocking-call blocking))) @@ -66,6 +60,7 @@ #+nio-debug (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job) (nio-utils:add result-queue (list request-id (nio-yarpc:execute-call job)))))))

+(defmethod process-timeout((sm yarpc-state-machine)))

(defmethod process-outgoing-packet((sm yarpc-state-machine)) #+nio-debug2 (format-log t "yarpc-state-machine:process-outgoing-packet - called, polling the results-queue ~%" ) @@ -78,8 +73,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-utils:add job-queue (list (call-string call) (request-id call) (result-queue sm))) - (when +process-jobs-inline+ (run-job :blocking nil))) + (nio-utils:add job-queue (list (call-string call) (request-id call) (result-queue sm))))