Megatest

mofiles/keysmod.o : mofiles/debugprint.o
mofiles/launchmod.o : mofiles/bigmod.o
mofiles/launchmod.o : mofiles/ezstepsmod.o
mofiles/launchmod.o : mofiles/rmtmod.o mofiles/servermod.o
mofiles/mtmod.o : mofiles/debugprint.o
mofiles/portloggermod.o : mofiles/tasksmod.o
mofiles/rmtmod.o : mofiles/apimod.o
mofiles/rmtmod.o : mofiles/commonmod.o
mofiles/rmtmod.o : mofiles/commonmod.o mofiles/http-transportmod.o
mofiles/rmtmod.o : mofiles/itemsmod.o mofiles/clientmod.o
mofiles/runsmod.o : mofiles/rmtmod.o mofiles/archivemod.o
mofiles/servermod.o : mofiles/commonmod.o
mofiles/servermod.o : mofiles/http-transportmod.o
mofiles/stml2.o : mofiles/cookie.o mofiles/dbi.o
mofiles/tasksmod.o : mofiles/pgdb.o mofiles/dbmod.o
mofiles/testsmod.o : mofiles/commonmod.o

	chicken.base
	chicken.process-context.posix
	chicken.string
	chicken.time
	chicken.condition
	chicken.process
	chicken.random
	chicken.file
	
	;; (prefix sqlite3 sqlite3:)
	typed-records
	srfi-18
	srfi-69

	commonmod
	dbmod
	debugprint
	tasksmod
	servermod
	matchable
	
	)
;; allow these queries through without starting a server
;;
(define api:read-only-queries
  '(get-key-val-pairs
    get-var

    testmeta-update-field

    ;; TASKS
    tasks-add
    tasks-set-state-given-param-key
    ))

(define (api:run-server-process apath dbname)
  (let* ((cmd  (conc "nbfake megatest -server - -area "apath
		     " -db "dbname))
	 (cleandbname (string-translate dbname "./" "_-"))
	 (logd (conc apath "/logs")) 
	 (logf (conc logd "/server-"(current-seconds)cleandbname".log")))
    (if (not (directory-exists? logd))
	(create-directory logd #t))
    (system (conc "NBFAKE_LOG="logf" "cmd))))

;; special function to get server
;; look up in db
;; if found -> return it
;; if not found -> start server, return starting
;;
(define (api:start-server dbstruct params)
  (let* ((res (apply db:get-server-info dbstruct params)))
    (if res
	res
	(match params
	  ((apath dbname)
	   (api:run-server-process apath dbname)
	   'server-started)
	  (else
	   (debug:print-info 0 *default-log-port* "api:start-server called with wrong params: "params)
	   'bad-params)))))

	
(define (api:dispatch-cmd dbstruct cmd params)
  (case cmd
    ;;===============================================
    ;; READ/WRITE QUERIES
    ;;===============================================

    ((get-keys-write)                        (db:get-keys dbstruct)) ;; force a dummy "write" query to force server; for debug in -repl
    
    ;; SERVERS
    ;; ((start-server)                    (apply server:kind-run params))
    ((kill-server)                       (set! *server-run* #f))
    ((get-server)                        (apply db:get-server-info dbstruct params))
    ((get-server)                        (api:start-server dbstruct params))

    ;; TESTS

    ;;((test-set-state-status-by-id)     (apply mt:test-set-state-status-by-id dbstruct params))
    ;;BB - commented out above because it was calling below, eventually, incorrectly (dbstruct passed to mt:test-set-state-status-by-id, which previosly did more, but now only passes thru to db:set-state-status-and-roll-up-items.
    ((test-set-state-status-by-id)

    ;; TASKS 
    ((find-task-queue-records)   (apply tasks:find-task-queue-records dbstruct params))
    (else
     (debug:print 0 *default-log-port* "ERROR: bad api call " cmd)
     (conc "ERROR: BAD api call " cmd))))


;; These are called by the server on recipt of /api calls
;;    - keep it simple, only return the actual result of the call, i.e. no meta info here
;;
;;    - returns #( flag result )
;;
(define (api:execute-requests dbstruct cmd params)
;;   (handle-exceptions
;;    exn
;;    (let ((call-chain (get-call-chain)))
;;      (debug:print 0 *default-log-port* "WARNING: api:execute-requests received an exception from peer, dat=" dat ", exn=" exn)
;;      (print-call-chain (current-error-port))
;;      (debug:print 0 *default-log-port* " message: "  ((condition-property-accessor 'exn 'message) exn))       
;;      (vector #f (vector exn call-chain dat))) ;; return some stuff for debug if an exception happens
;;    (cond
;;     ((not (vector? dat))                    ;; it is an error to not receive a vector
;;      (vector #f (vector #f "remote must be called with a vector")))
;;     ((> *api-process-request-count* 20) ;; 20)
;;      (debug:print 0 *default-log-port* "WARNING: api:execute-requests received an overloaded message.")
;;      (set! *server-overloaded* #t)
;;      (vector #f (vector #f 'overloaded))) ;; the inner vector is what gets returned. nope, don't know why. please refactor!
;;     (else  
  (let* (;; (cmd-in            (vector-ref dat 0))
            ;; (cmd               (if (symbol? cmd-in)
	    ;; 			   cmd-in
	    ;; 			   (string->symbol cmd-in)))
            ;; (params            (vector-ref dat 1))
            (start-t           (current-milliseconds))
            ;; (readonly-mode     (dbr:dbstruct-read-only dbstruct))
            ;; (readonly-command  (member cmd api:read-only-queries))
  (let* ((start-t           (current-milliseconds))
	 ;; (readonly-mode     (dbr:dbstruct-read-only dbstruct))
	 ;; (readonly-command  (member cmd api:read-only-queries))
            ;; (writecmd-in-readonly-mode (and readonly-mode (not readonly-command)))
	    (res        (api:dispatch-cmd dbstruct cmd params)))

	 (res        (api:dispatch-cmd dbstruct cmd params)))
    
    ;; (if writecmd-in-readonly-mode
    ;; (conc "attempt to run write command "cmd" on a read-only database")

    ;; save all stats
    (let ((delta-t (- (current-milliseconds)
		      start-t)))
      (hash-table-set! *db-api-call-time* cmd

;; http-server  send-response
;;                 api:process-request
;;                    db:*
;;
;; NB// Runs on the server as part of the server loop
;;
(define (api:process-request dbstruct $) ;; the $ is the request vars proc
  (debug:print 4 *default-log-port* "server-id:"  *server-id*)
  (debug:print 0 *default-log-port* "server-id:"  *server-id*)
  (let* ((cmd     ($ 'cmd))
	 (params  (string->sexpr ($ 'params)))
         (key     ($ 'key))    ;; TODO - add this back
	 )
    (debug:print 0 *default-log-port* "cmd:" cmd " with params " params "key " key)
    (if (equal? key "nokey") ;; *server-id*) ;; TODO - get real key involved
	(begin

	(let* (;; (dbfile   (db:run-id->path apath run-id))
	       (newdbdat (db:open-dbdat apath dbfile db:initialize-db)))
	  (dbr:dbstruct-dbdat-put! dbstruct dbfile newdbdat)
	  newdbdat))))

;; get the inmem db for actual db operations
;;
(define (db:get-inmem dbstruct dbfile)
  (dbr:dbdat-inmem (db:get-dbdat dbstruct dbfile)))
(define (db:get-inmem dbstruct apath dbfile)
  (dbr:dbdat-inmem (db:get-dbdat dbstruct apath dbfile)))

;; get the handle for the on-disk db
;;
(define (db:get-ddb dbstruct apath dbfile)
  (dbr:dbdat-db (db:get-dbdat dbstruct apath dbfile)))

;; open or create the disk db file

		     ))

;; (db:with-db dbstruct run-id sqlite3:exec "select blah fgrom blaz;")
;; r/w is a flag to indicate if the db is modified by this query #t = yes, #f = no
;;
(define (db:with-db dbstruct run-id r/w proc . params)
  (assert (dbr:dbstruct? dbstruct) "FATAL: db:with-db called with bad dbstruct")
  (let* ((dbpath    (db:run-id->dbname run-id))
  (let* ((dbdat     (db:get-dbdat dbstruct run-id))
	 (dbdat     (db:get-dbdat dbstruct *toppath* dbpath))
	 (db        (dbr:dbdat-inmem dbdat))
	 (fname     (dbr:dbdat-fname dbdat))
	 (use-mutex (> *api-process-request-count* 25))) ;; was 25
    (if (and use-mutex
	     (common:low-noise-print 120 "over-50-parallel-api-requests"))
	(debug:print-info 0 *default-log-port* *api-process-request-count* " parallel api requests being processed in process " (current-process-id) ", throttling access"))
    (if (common:low-noise-print 600 (conc "parallel-api-requests" *max-api-process-requests*))

  (db:with-db
   dbstruct
   #f #f
   (lambda (db)
     (sqlite3:execute db "INSERT INTO servers (host,port,servkey,pid,ipaddr,dbpath) VALUES (?,?,?,?,?,?);"
		      host port servkey pid ipaddr dbpath))))

(define (db:get-server-info dbstruct dbpath)
(define (db:get-server-info dbstruct apath dbname)
  (db:with-db
   dbstruct
   #f #f
   (lambda (db)
     (sqlite3:fold-row
      (lambda (res host port servkey pid ipaddr dbpath)
	(list host port servkey pid ipaddr dbpath))
      '()
      #f
      db
      "SELECT host,port,servkey,pid,ipaddr,dbpath FROM servers WHERE dbpath=?;"
      dbpath))))
      (conc apath "/" dbname)))))

)

					   ;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
					   (server:get-best-guess-address hostname)
					   #f)))
			    (if ipstr ipstr hostn))) ;; hostname))) 
	 (start-port      (portlogger:open-run-close portlogger:find-port))
	 (link-tree-path  (common:get-linktree))
	 (tmp-area        (common:get-db-tmp-area))
	 (start-file      (conc tmp-area "/.server-start")))
	 #;(start-file      (conc tmp-area "/.server-start")))
    (debug:print-info 0 *default-log-port* "portlogger recommended port: " start-port)
    ;; set some parameters for the server
    (root-path     (if link-tree-path 
		       link-tree-path
		       (current-directory))) ;; WARNING: SECURITY HOLE. FIX ASAP!
    (handle-directory spiffy-directory-listing)
    (handle-exception (lambda (exn chain)
    #;(handle-exception (lambda (exn chain)
			(signal (make-composite-condition
				 (make-property-condition 
				  'server
				  'message "server error")))))

    ;; Setup the web server and a /ctrl interface
    ;;
    (vhost-map `(((* any) . ,(lambda (continue)
			       ;; open the db on the first call 
				 ;; This is were we set up the database connections
			       (let* (($   (request-vars source: 'both))
				      (dat ($ 'dat))
				      (res #f))
				 (cond
				  ((equal? (uri-path (request-uri (current-request)))
					   '(/ "api"))
				   (debug:print 0 *default-log-port* "In api request $=" $)
				   (send-response ;; the $ is the request vars proc
				    body: ((api-proc) *dbstruct-db* $) ;; ($) => alist
				    headers: '((content-type text/plain)))
				   (set! *db-last-access* (current-seconds)))
				  ((equal? (uri-path (request-uri (current-request))) 
					   '(/ "ping"))
				   (send-response body: (conc *toppath*"/"(args:get-arg "-db"))

  (mutex-lock! *http-mutex*)
  (set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*)))

;; serverdat contains uuid to be used for connection validation
;;
;; NOTE: serverdat must be initialized or created by servdat-init
;;
;; DO NOT USE. Moved to rmt:set-receive-real
;;
#;(define (http-transport:send-receive sdat qry-key cmd params #!key (numretries 3))
  (let* ((res        #f)
	 (success    #t)
	 (sparams    (with-output-to-string
		       (lambda ()(write params)))))
    ;; send the data and get the response extract the needed info from
    ;; the http data and process and return it.
    (let* ((send-recieve (lambda ()
			   (set! res
;; (define (http-transport:send-receive conn qry-key cmd params #!key (numretries 3))
;;   (let* ((res        #f)
;; 	 (success    #t)
;; 	 (sparams    (with-output-to-string
;; 		       (lambda ()(write params)))))
;;     ;; send the data and get the response extract the needed info from
;;     ;; the http data and process and return it.
;;     (let* ((send-recieve (lambda ()
;; 			   (set! res
				 (vector
				  #t ;; success
				  (with-input-from-request
				   (servdat-api-uri sdat)
				   (list (cons 'key qry-key)
					 ;; (cons 'srvid (servdat-uuid sdat))
					 (cons 'cmd cmd)
					 (cons 'params sparams))
				   read-string))))) ;; or read-string?
	      (time-out     (lambda ()
			      (thread-sleep! 45)
			      (debug:print 0 *default-log-port* "WARNING: send-receive took more than 45 seconds!!")
			      #f))
	      (th1 (make-thread send-recieve "with-input-from-request"))
	      (th2 (make-thread time-out     "time out")))
	 (thread-start! th1)
	 (thread-start! th2)
	 (thread-join! th1)
	 (close-idle-connections!)
	 (thread-terminate! th2)
	 (if (string? res)
	     (with-input-from-string res
	       (lambda () read))
	     res))))
;; 				 (with-input-from-request
;; 				  (rmt:conn->uri conn "api")
;; 				  (list (cons 'key qry-key)
;; 					;; (cons 'srvid (servdat-uuid sdat))
;; 					(cons 'cmd cmd)
;; 					(cons 'params sparams))
;; 				   read-string))))
;; 	   (time-out     (lambda ()
;; 			   (thread-sleep! 45)
;; 			   (debug:print 0 *default-log-port* "WARNING: send-receive took more than 45 seconds!!")
;; 			   #f))
;; 	   (th1 (make-thread send-recieve "with-input-from-request"))
;; 	   (th2 (make-thread time-out     "time out")))
;;       (thread-start! th1)
;;       (thread-start! th2)
;;       (thread-join! th1)
;;       (close-idle-connections!)
;;       (thread-terminate! th2)
;;       (if (string? res)
;; 	  (with-input-from-string res
;; 	    (lambda () read))
;; 	  res))))

;; careful closing of connections stored in *runremote*
;;
(define (http-transport:close-connections #!key (area-dat #f))
  (debug:print-info 0 *default-log-port* "http-transport:close-connections doesn't do anything now!"))
;;   (let* ((runremote  (or area-dat *runremote*))
;; 	 (server-dat (if runremote

  (srvpkt   #f)
  (lastmsg  0)
  (expires  0))

;; replaces *runremote*
(define *rmt:remote* (make-rmt:remote))

;; -> http://abc.com:900/<entrypoint>
;;
(define (rmt:conn->uri conn entrypoint)
  (conc "http://"(rmt:conn-ipaddr conn)":"(rmt:conn-port conn)"/"entrypoint))

;; set up the api proc, seems like there should be a better place for this?
(api-proc api:process-request)

;; do we have a connection to apath dbname and
;; is it not expired? then return it
;;
;; else setup a connection

;; connections for other servers happens by requesting from main
;;
(define (rmt:open-main-connection remote apath)
  (let* ((dbname         (db:run-id->dbname #f))
	 (the-srv        (rmt:find-main-server apath dbname))
	 (start-main-srv (lambda ()
			   ;; srv not ready, delay a little and try again
			   (system (conc "nbfake megatest -server - -area "apath
					 " -db "dbname))
			   (api:run-server-process apath dbname)
			   (thread-sleep! 2)
			   (rmt:open-main-connection remote apath) ;; TODO: Add limit to number of tries
			   )))
    (if the-srv ;; yes, we have a server, now try connecting to it
	(let* ((srv-addr (server-address the-srv))
	       (ipaddr   (alist-ref 'ipaddr the-srv))
	       (port     (alist-ref 'port   the-srv))

	      (start-main-srv)))
	(start-main-srv))))

;; NB// remote is a rmt:remote struct
;;
(define (rmt:general-open-connection remote apath dbname)
  (let  ((mainconn (rmt:get-connection remote apath (db:run-id->dbname #f))))
    (debug:print 0 *default-log-port* "remote: " remote)
    (if (not mainconn)
	(begin
	  (rmt:open-main-connection remote apath)
	  (thread-sleep! 1)
	  (rmt:general-open-connection remote apath dbname))
	;; we have a connection to main, ask for contact info for dbname
	(let* ((res (rmt:send-receive-real remote apath ".db/main.db" #f 'get-server `(,apath ,dbname))))
	  (print "rmt:general-open-connection got res="res)))))
	(let* ((res (rmt:send-receive mainconn "querykeyhere" 'get-server `(,apath ,dbname))))
	  (print "rmt:general-open-connection got res="res)
	  res))))
	  

;;======================================================================

;; Defaults to 
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f))
  (if (not *rmt:remote*)(set! *rmt:remote* (make-rmt:remote)))
  (let* ((apath *toppath*)
	 (conns *rmt:remote*)
	 (dbname (db:run-id->dbname rid)))
    (rmt:send-receive-real conns apath dbname rid cmd params)))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future
;;
(define (rmt:send-receive-real remote apath dbname rid cmd params)
  (let* ((conn (rmt:get-connection remote apath dbname)))
    (assert conn "FATAL: Unable to connect to db "apath"/"dbname)
    (let* ((host    (rmt:conn-ipaddr conn))
	   (port    (rmt:conn-port   conn))
    (let* (;; (host    (rmt:conn-ipaddr conn))
	   ;; (port    (rmt:conn-port   conn))
	   (payload (sexpr->string params))
	   (res      (with-input-from-request
		      (conc "http://"host":"port"/api")
		      (rmt:conn->uri conn "api") ;; (conc "http://"host":"port"/api")
		      `((params . ,payload)
			(cmd    . ,cmd)
			(key    . "nokey"))
		      read-string)))
      (string->sexpr res))))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future
;;
(define (rmt:send-receive-server-start remote apath dbname)
  (let* ((conn (rmt:get-connection remote apath dbname)))
    (assert conn "FATAL: Unable to connect to db "apath"/"dbname)
    (let* (;; (host    (rmt:conn-ipaddr conn))
	   ;; (port    (rmt:conn-port   conn))
	   ;; (payload (sexpr->string params))
	   (res      (with-input-from-request
		      (rmt:conn->uri conn "api") ;; (conc "http://"host":"port"/api")
		      `((params . (,apath ,dbname))
			;; (cmd    . ,cmd)
			#;(key    . "nokey"))
		      read-string)))
      (string->sexpr res))))

(define (rmt:print-db-stats)
  (let ((fmtstr "~40a~7-d~9-d~20,2-f")) ;; "~20,2-f"
    (debug:print 18 *default-log-port* "DB Stats\n========")
    (debug:print 18 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg"))

  (let* ((server-url (server:record->url server-record))
         (server-id (server:record->id server-record)) 
         (res       (server:ping server-url server-id)))
    (if res
        server-url
	#f)))

;; no longer care if multiple servers are started by accident. older servers will drop off in time.
;; no longer care if multiple servers are started by accident. older
;; servers will drop off in time.
;;
(define (server:check-if-running areapath) ;;  #!key (numservers "2"))
  (let* ((ns            (server:get-num-servers))
	 (servers       (server:get-best (server:get-list areapath))))
    (if (or (and servers
		 (null? servers))
	    (not servers)

;;     You should have received a copy of the GNU General Public License
;;     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.

;; Run like this:
;;
;;  ./rununittest.sh server 1;(cd simplerun;megatest -stop-server 0)

(import rmtmod trace http-transportmod http-client apimod)
(import rmtmod trace http-transportmod http-client apimod dbmod)
(trace-call-sites #t)
(trace
 ;; db:get-dbdat
 ;; rmt:find-main-server
 )

(test #f #t (rmt:remote? (let ((r (make-rmt:remote)))
			   (set! *rmt:remote* r)
			   r)))
(test #f #f (rmt:get-connection *rmt:remote* *toppath* ".db/main.db"))
(test #f #f (rmt:find-main-server *toppath* ".db/main.db"))
(test #f #t (rmt:open-main-connection *rmt:remote* *toppath*))
(pp (hash-table->alist (rmt:remote-conns *rmt:remote*)))
(test #f #t (rmt:conn? (rmt:get-connection *rmt:remote* *toppath* ".db/main.db")))

(define *main*  (rmt:get-connection *rmt:remote* *toppath* ".db/main.db"))

(test #f 'a (loop-test (rmt:conn-ipaddr *main*)(rmt:conn-port *main*) 'a))
(trace
 rmt:send-receive
 with-input-from-request
 rmt:get-connection
 with-input-from-request
 )

(define *db* (db:setup #f))
(test #f #f (api:execute-requests *db* 'get-server (list (conc *toppath*"/.db/1.db"))))
(test #f #f (rmt:general-open-connection *rmt:remote* (list (conc *toppath*"/.db/1.db"))))
(test #f 'server-started (api:execute-requests *db* 'get-server (list *toppath* ".db/1.db")))
(test #f #f (rmt:general-open-connection *rmt:remote* *toppath* ".db/1.db"))

;; (delete-file* "logs/1.log")
;; (define run-id 1)

;; (test "setup for run" #t (begin (launch:setup)
;;  				(string? (getenv "MT_RUN_AREA_HOME"))))
;;