Megatest

	srfi-1
	regex regex-case srfi-69
	typed-records
	sparse-vectors
	format
	srfi-4
	srfi-14
	srfi-18
	)

;; (include "common_records.scm")
;; (include "db_records.scm")
;; (include "run_records.scm")
;; (include "task_records.scm")
;; (include "megatest-version.scm")

(import dashboard)

;; ease debugging by loading ~/.dashboardrc
(let ((debugcontrolf (get-debugcontrolf)))
  (if debugcontrolf
      (load debugcontrolf)))

(import srfi-18)

(thread-join!
 (thread-start!
  (make-thread main "main")))

;;======================================================================
;; S E R V E R
;; ======================================================================

(define (http-get-function fnkey)
  (hash-table-ref/default *http-functions* fnkey (lambda () "nothing here yet")))

(define *rmt:run-mutex* (make-mutex))
(define *rmt:run-flag* #f)

;; Main entry point to start a server. was start-server
(define (rmt:run hostn)
  (mutex-lock! *rmt:run-mutex*)
  (if *rmt:run-flag*
      (begin
	(debug:print-warn 0 *default-log-port* "rmt:run already running.")
	(mutex-unlock! *rmt:run-mutex*))
      (begin
	(set! *rmt:run-flag* #t)
	(mutex-unlock! *rmt:run-mutex*)
	;;  ;; Configurations for server
	;;  (tcp-buffer-size 2048)
	;;  (max-connections 2048) 
	(debug:print 2 *default-log-port* "PID: "(current-process-id)". Attempting to start the server ...")
	(if (and *db-serv-info*
		 (servdat-uconn *db-serv-info*))
	    (let* ((uconn (servdat-uconn *db-serv-info*)))
	      (wait-and-close uconn))
	    (let* ((port            (portlogger:open-run-close portlogger:find-port))
		   (handler-proc    (lambda (rem-host-port qrykey cmd params) ;;
				      (set! *db-last-access* (current-seconds))
				      (assert (list? params) "FATAL: handler called with non-list params")
				      (api:execute-requests *dbstruct-db* cmd params))))
	      ;; (api:process-request *dbstuct-db* 
	      (if (not *db-serv-info*)
		  (set! *db-serv-info* (make-servdat host: hostn port: port)))
	      (let* ((uconn (run-listener handler-proc port))
		     (rport (udat-port uconn))) ;; the real port
		(servdat-host-set! *db-serv-info* hostn)
		(servdat-port-set! *db-serv-info* rport)
		(servdat-uconn-set! *db-serv-info* uconn)
		(wait-and-close uconn)
		(db:print-current-query-stats)
		)))
	(let* ((host (servdat-host *db-serv-info*))
	       (port (servdat-port *db-serv-info*))
	       (mode (or (servdat-mode *db-serv-info*)
			 "non-db")))
	  ;; server exit stuff here
	  ;; (rmt:server-shutdown host port) - always do in on-exit
	  ;; (portlogger:open-run-close portlogger:set-port port "released") ;; moved to on-exit 
	  (debug:print-info 0 *default-log-port* "Server "host":"port" mode "mode"shutdown complete. Exiting")
	  ))))

;;======================================================================
;; S E R V E R   U T I L I T I E S 
;;======================================================================

;;======================================================================
;; C L I E N T S

(define th1 
  (make-thread
    (lambda ()
(let loop ((r 1)
	   (i 1))
   (print "register-test "r" test"i)
   (rmt:register-test r "test1" (conc "item_" i))
   (if (< i 100000)
       (loop r (+ i 1))
       (if (< r 100)
	   (begin
	      (print "get-tests-for-run "r)
              (rmt:get-tests-for-run r "%" '() '() 0 #f #f #f #f #f 0 #f)
 	      (loop (+ r 1) 0)))))
)))

	chicken.string
	chicken.sort
	chicken.pretty-print
	
	address-info
	mailbox
	matchable
	;; queues
	regex
	regex-case
	s11n
	srfi-1
	srfi-18
	srfi-4
	srfi-69

  ;; the listener side
  (port #f)
  (host-port #f)
  (socket #f)
  ;; the peers
  (peers  (make-hash-table)) ;; host:port->peer
  ;; work handling
  (work-queue (make-mailbox))
  (work-proc  #f)                ;; set by user
  (cnum       0)                 ;; cookie number
  (mboxes     (make-hash-table)) ;; for the replies
  (avail-cmboxes '())            ;; list of (<cookie> . <mbox>) for re-use
  ;; threads
  (numthreads 10)
  (cmd-thread #f)
  (work-queue-thread #f)
  ) 

;; ;; struct for keeping track of others we are talking to
;; ;;
;; (defstruct pdat

;;======================================================================
;; work queues - this is all happening on the listener side
;;======================================================================

;; rdat is (rem-host-port qrykey cmd params)
					     
(define (add-to-work-queue uconn rdat)
  #;(queue-add! (udat-work-queue uconn) rdat)
  (mailbox-send! (udat-work-queue uconn) rdat))

(define (do-work uconn rdat)
  (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change
    ;; put this following into a do-work procedure
    (match rdat
      ((rem-host-port qrykey cmd params)
       (let* ((result (proc rem-host-port qrykey cmd params)))
	 ;; send 'response as cmd and result as params
	 (send uconn rem-host-port qrykey 'response result))) ;; could check for ack
      (else
       (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params")))))
     

(define (process-work-queue uconn) 
  (let ((wqueue (udat-work-queue uconn))
	(proc   (udat-work-proc  uconn))
	(numthr (udat-numthreads uconn)))
    (let loop ((thnum    1)

	       (threads '()))
      (let ((thlst (cons (make-thread (lambda ()
					(let ((rdat (mailbox-receive! wqueue #f 'MBOX_TIMEOUT)))
					  (do-work uconn rdat)))
				      (conc "work thread " thnum))
			 threads)))
	(if (< thnum numthr)
	    (loop (+ thnum 1)
		  thlst)
	    (begin
	      (print "ULEX: Starting "(length thlst)" worker threads.")
	      (map thread-start! thlst)
	      (print "ULEX: Threads started. Joining all.")
	      (map thread-join! thlst)))))))

;; below was to enable re-use of connections. This seems non-trivial so for
;; now lets open on each call
;;
;; ;; given host-port get or create peer struct
;; ;;
;; (define (udat-get-peer uconn host-port)