X-Git-Url: http://git.kpe.io/?a=blobdiff_plain;f=listener.lisp;h=dfd70f490985ffed9c1cb3d33a60a4d31fba0a86;hb=0dc565c13310ce9f59b42b4e4bdd9167e24ca756;hp=d37d72d924c54f51a3c82c948ba149e5d71667d0;hpb=0ddc73d14e764eeaede64f92b620a0c6de46cfd3;p=kmrcl.git

diff --git a/listener.lisp b/listener.lisp
index d37d72d..dfd70f4 100644
--- a/listener.lisp
+++ b/listener.lisp
@@ -5,9 +5,9 @@
 ;;;; Name:          listener.lisp
 ;;;; Purpose:       Listener and worker processes
 ;;;; Programmer:    Kevin M. Rosenberg
-;;;; Date Started:  Dec 2002
+;;;; Date Started:  Jun 2003
 ;;;;
-;;;; $Id: listener.lisp,v 1.2 2003/07/09 22:12:52 kevin Exp $
+;;;; $Id: listener.lisp,v 1.4 2003/07/11 02:37:33 kevin Exp $
 ;;;; *************************************************************************
 
 (in-package #:kmrcl)
@@ -29,28 +29,34 @@
 	     :initform nil)
    (function-args :initarg :function-args :accessor function-args
 		  :initform nil)
-   (process :initarg :process :accessor process)
-   (socket :initarg :socket :accessor socket)
+   (process :initarg :process :accessor process :initform nil)
+   (socket :initarg :socket :accessor socket :initform nil)
    (workers :initform nil :accessor workers
 	    :documentation "list of worker threads")
    (name :initform "" :accessor name :initarg :name)
    (base-name :initform "listener" :accessor base-name :initarg :base-name)
    (wait :initform nil :accessor wait :initarg :wait)
+   (timeout :initform nil :accessor timeout :initarg :timeout)
+   (number-fixed-workers :initform nil :accessor number-fixed-workers
+			 :initarg :number-fixed-workers)
    (catch-errors :initform nil :accessor catch-errors :initarg :catch-errors)
    (format :initform :text :accessor listener-format :initarg :format)))
 
-(defclass worker ()
+(defclass fixed-worker ()
   ((listener :initarg :listener :accessor listener :initform nil)
-   (connection :initarg :connection :accessor connection :initform nil)
    (name :initarg :name :accessor name :initform nil)
-   (thread-fun :initarg :thread-fun :accessor thread-fun :initform nil)
    (process :initarg :process :accessor process :initform nil)))
 
+(defclass worker (fixed-worker)
+  ((connection :initarg :connection :accessor connection :initform nil)
+   (thread-fun :initarg :thread-fun :accessor thread-fun :initform nil)))
+
+
 (defmethod print-object ((obj listener) s)
   (print-unreadable-object (obj s :type t :identity nil)
     (format s "port ~A" (port obj))))
 
-(defmethod print-object ((obj worker) s)
+(defmethod print-object ((obj fixed-worker) s)
   (print-unreadable-object (obj s :type t :identity nil)
     (format s "port ~A" (port (listener obj)))))
   
@@ -80,16 +86,22 @@
        (warn "~&listener is not in active list")
        (return-from init/listener listener))
      (dolist (worker (workers listener))
-       (close-active-socket (connection worker))
-       (destroy-process (process worker))
-       (setf (connection worker) nil)
-       (setf (process worker) nil))
+       (when (and (typep worker 'worker)
+		  (connection worker))
+	 (errorset (close-active-socket
+		    (connection worker)) nil)
+	 (setf (connection worker) nil))
+       (when (process worker)
+	 (errorset (destroy-process (process worker)) nil)
+	 (setf (process worker) nil)))
      (setf (workers listener) nil)
      (with-slots (process socket) listener
-       (errorset (close-passive-socket socket) t)
-       (errorset (destroy-process process) t)
-       (setf process nil)
-       (setf socket nil))
+       (when socket
+	 (errorset (close-passive-socket socket) nil)
+	 (setf socket nil))
+       (when process
+	 (errorset (destroy-process process) nil)
+	 (setf process nil)))
      (setq *active-listeners* (remove listener *active-listeners*)))
     (:restart
      (init/listener listener :stop)
@@ -109,40 +121,60 @@
   (format nil "~A-worker-~D" base-name (incf *worker-count*)))
 
 (defun make-socket-server (listener)
-  (setf (socket listener) (create-inet-listener
-			   (port listener)
-			   :format (listener-format listener)))
-  (setf (process listener) (make-process
-			    (name listener)
-			    #'(lambda () (start-socket-server listener))))
+  #+lispworks
+  (progn
+    (setf (process listener)
+      (comm:start-up-server :process-name (name listener)
+			    :service (port listener) 
+			    :function
+			    #'(lambda (handle) 
+				(lw-worker handle listener)))))
+  #-lispworks
+  (progn
+    (setf (socket listener) (create-inet-listener
+			     (port listener)
+			     :format (listener-format listener)))
+    (if (number-fixed-workers listener)
+	(start-fixed-number-of-workers listener)
+	(setf (process listener) (make-process
+				  (name listener)
+				  #'(lambda ()
+				      (start-socket-server listener))))))
   listener)
 
 
 (defmethod initialize-instance :after
     ((self worker) &key listener connection name &allow-other-keys)
-  (unless connection
-    (error "connection not provided to modlisp-worker"))
-  (setf (slot-value self 'listener) listener)
-  (setf (slot-value self 'name) name)
-  (setf (slot-value self 'connection) connection)
-  (setf (slot-value self 'thread-fun)
-	#'(lambda ()
-	    (unwind-protect
-		(if (catch-errors listener)
-		    (handler-case
-			(apply (listener-function listener)
-			       connection
-			       (function-args listener))
-		      (error (e)
-			(cmsg "Error ~A [~A]" e name)))
-		  (apply (listener-function listener)
-			 connection
-			 (function-args listener)))
-	  (progn
-	    (errorset (close-active-socket connection) nil)
-	    (cmsg-c :threads "~A ended" name)
-	    (setf (workers listener)
-		  (remove self (workers listener))))))))
+  (flet ((do-work ()
+	   (apply (listener-function listener)
+		  connection
+		  (function-args listener))))
+    (unless connection
+      (error "connection not provided to modlisp-worker"))
+    (setf (slot-value self 'listener) listener)
+    (setf (slot-value self 'name) name)
+    (setf (slot-value self 'connection) connection)
+    (setf (slot-value self 'thread-fun)
+	  #'(lambda ()
+	      (unwind-protect
+		   (if (catch-errors listener)
+		       (handler-case
+			   (if (timeout listener)
+			       (with-timeout ((timeout listener))
+				 (do-work))
+			       (do-work))
+			 (error (e)
+			   (cmsg "Error ~A [~A]" e name)))
+		       (if (timeout listener)
+			   (with-timeout ((timeout listener))
+			     (do-work))
+			   (do-work)))
+		(progn
+		  (errorset (finish-output connection) nil)
+		  (errorset (close-active-socket connection) nil)
+		  (cmsg-c :threads "~A ended" name)
+		  (setf (workers listener)
+			(remove self (workers listener)))))))))
 
 (defun start-socket-server (listener)
   (unwind-protect
@@ -153,12 +185,75 @@
 		  (apply (listener-function listener)
 			 connection
 			 (function-args listener))
-	       (errorset (close connection) nil))
+	       (progn
+		 (errorset (finish-output connection) nil)
+		 (errorset (close-active-socket connection) nil)))
 	     (let ((worker (make-instance 'worker :listener listener
 					  :connection connection
 					  :name (next-worker-name
 						 (base-name listener)))))
-	      (setf (process worker)
-		(make-process (name worker) (thread-fun worker)))
-	      (push worker (workers listener))))))
+	       (setf (process worker)
+		     (make-process (name worker) (thread-fun worker)))
+	       (push worker (workers listener))))))
     (errorset (close-passive-socket (socket listener)) nil)))
+
+#+lispworks
+(defun lw-worker (handle listener)
+  (let ((connection (make-instance 'comm:socket-stream
+		      :socket handle
+		      :direction :io
+		      :element-type 'base-char)))
+    (if (wait listener)
+	(progn
+	  (apply (listener-function listener)
+		 connection
+		 (function-args listener))
+	  (finish-output connection))
+	(let ((worker (make-instance 'worker :listener listener
+				     :connection connection
+				     :name (next-worker-name
+					    (base-name listener)))))
+	  (setf (process worker)
+		(make-process (name worker) (thread-fun worker)))
+	  (push worker (workers listener))))))
+
+;; Fixed pool of workers
+
+(defun start-fixed-number-of-workers (listener)
+  (dotimes (i (number-fixed-workers listener))
+    (let ((name (next-worker-name (base-name listener))))
+      (push
+       (make-instance 'fixed-worker
+		      :name name
+		      :listener listener
+		      :process
+		      (make-process
+		       name #'(lambda () (fixed-worker name listener))))
+       (workers listener)))))
+
+
+(defun fixed-worker (name listener)
+  (loop 
+   (let ((connection (accept-tcp-connection (socket listener))))
+     (flet ((do-work ()
+	      (apply (listener-function listener)
+		     connection
+		     (function-args listener))))
+       (unwind-protect
+	    (handler-case
+		(if (catch-errors listener)
+		    (handler-case
+			(if (timeout listener)
+			    (with-timeout ((timeout listener))
+			      (do-work))
+			    (do-work))
+		      (error (e)
+			(cmsg "Error ~A [~A]" e name)))
+		    (if (timeout listener)
+			(with-timeout ((timeout listener))
+			  (do-work))
+			(do-work)))
+	      (error (e)
+		(format t "Error: ~A" e)))
+	 (errorset (finish-output connection) nil)
+	 (errorset (close connection) nil))))))