Changeset 40 for branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp

Timestamp:

01/17/07 05:06:13 (18 years ago)

Author:

psmith

Message:

yarpc work, saving...

File:

• branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp (modified) (4 diffs)

branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp ¶

@@ -33 +33 @@
 ;; A server that processes remote procedure calls and returns results
 ;;
-;; Test with: 
-;; > telnet 127.0.0.1 16323
-;; Trying 127.0.0.1...
-;; Connected to 127.0.0.1.
-;; Escape character is '^]'.
-;; (test-rpc "who" 2 's)
-;; response - who 2 'S
-;;
 (defclass yarpc-state-machine (state-machine)
-  ((outgoing-packet :initarg :outgoing-packet
-                    :accessor outgoing-packet
-                    :initform nil)))
+  ((job-queue :initarg :job-queue
+              :initform (error "Must supply a job queue to write work to.")
+              :accessor job-queue
+              :documentation "The queue used to hand off work from the NIO thread to an external thread for execution")
+   (result-queue :initform (nio-compat:concurrent-queue)
+                 :accessor result-queue
+                 :documentation "The queue used to return results from an external thread to the nio thread")))
 
-(defun yarpc-state-machine ()
-    (make-instance 'yarpc-state-machine))
+(defun yarpc-state-machine (read-fd write-fd socket job-queue)
+  (let ((sm (make-instance 'yarpc-state-machine :read-fd read-fd :write-fd write-fd :socket socket :job-queue job-queue)))
+    (nio-buffer:clear (foreign-read-buffer sm))
+    (nio-buffer:clear (foreign-write-buffer sm))
+    (format t "yarpc-state-machine - Created ~S~%" sm)
+    sm))
 
 (defparameter yarpc-pf (yarpc-packet-factory))
@@ -53 +53 @@
 (defmethod get-packet-factory((sm yarpc-state-machine))
   yarpc-pf)
+
+(defmethod print-object ((sm yarpc-state-machine) stream)
+  (format stream "#<YARPC-STATE-MACHINE ~A >" (call-next-method sm nil)))
+
+(defconstant STATE-INITIALISED 0)
+(defconstant STATE-SEND-RESPONSE 1)
+
+(defparameter state STATE-INITIALISED)
+
+
+(defmethod process-outgoing-packet((sm yarpc-state-machine))
+  (format t "yarpc-state-machine: process-outgoing-packet called, polling the results-queue ~%")
+  (let ((packet (nio-compat:take (result-queue sm) :blocking-call nil)))
+    (format t "yarpc-state-machine: process-outgoing-packet got result ~A ~%" packet)
+    packet))
+
+
+;Process a call method packet, returns 
+(defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet))
+  (assert (eql state STATE-INITIALISED))
+  (format t "yarpc-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm call)
+  (nio-compat:add (job-queue sm) (cons (call-string call) (result-queue sm))))  
+
+
+;Called from an external thread i.e. *not* the nio thread
+;Blocks waiting for a job (call-string,result-queue) to process and return the result into the result queue
+;(defmethod get-job ((sm yarpc-state-machine))
+;  (values (nio-compat:take (job-queue sm)) (result-queue sm)))
+
+
 
 ;;TODO move somewhere suitable
@@ -72 +102 @@
   (format nil "response - ~A ~A ~A ~A~%" a b c (code-char #x2211)))
 
-;;end move TODO
-
-
-;;;Utils
-
-(defun print-hashtable (table &optional (stream t))
-  (maphash #'(lambda (k v) (format stream "~a -> ~a~%" k v)) table))
-;;;
-
-
-(defmethod print-object ((sm yarpc-state-machine) stream)
-  (format stream "#<YARPC-STATE-MACHINE ~A >" (call-next-method sm nil)))
-
-(defconstant STATE-INITIALISED 0)
-(defconstant STATE-SEND-RESPONSE 1)
-
-(defparameter state STATE-INITIALISED)
-
 (define-condition authorization-error (error) ())
-
-
-(defmethod process-outgoing-packet((sm yarpc-state-machine))
-  (format t "process-outgoing-packet called~%")
-  (let ((packet (outgoing-packet sm)))
-    (setf (outgoing-packet sm) nil)
-    packet))
-
-;TODO queue and thread stuf
-(defmethod queue-outgoing-packet((sm yarpc-state-machine) packet)
-  (setf (outgoing-packet sm) packet))
-
-;Process a call method packet, returns 
-(defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet))
-  (assert (eql state STATE-INITIALISED))
-  (format t "yarpc-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm call)
-  (handler-case
-    (let ((result (execute-call (call-string call))))
-      (when result 
-        (let ((response-packet (progn 
-                                  (setf state STATE-SEND-RESPONSE)
-                                  (queue-outgoing-packet sm (method-response-packet result)))))
-          t)))
-    (reader-error (re) (format t "No such function ~A~%" (call-string call)))
-    (authorization-error (ae) (format t "Function not declared with defremote ~A~%" (call-string call)))))
-
-(defmethod process-incoming-packet ((sm yarpc-state-machine) (response method-response-packet))
-  (assert (eql state STATE-INITIALISED))
-  (format t "yarpc-state-machine:process-incoming-packet called :sm ~A :packet ~A~%" sm response))
-  
 
 (defun execute-call (call-string)
@@ -130 +112 @@
             (error 'authorization-error))))
 
+;;end move TODO
 
-(defmethod remote-execute ((sm yarpc-state-machine) call-string)
-  (queue-outgoing-packet sm (make-instance 'call-method-packet :call-string call-string)))
-    
+
+
+
+
+;  (handler-case
+;    (let ((result (execute-call (call-string call))))
+;      (when result 
+;       (let ((response-packet (progn 
+;                                  (setf state STATE-SEND-RESPONSE)
+;                                  (queue-outgoing-packet sm (method-response-packet result)))))
+;          t)))
+;    (reader-error (re) (format t "No such function ~A~%" (call-string call)))
+;    (authorization-error (ae) (format t "Function not declared with defremote ~A~%" (call-string call))))