Megatest

;; timeout is hms string: 1h 5m 3s, default is 10 minutes
;;
(define (server:expiration-timeout)
  (let ((tmo (configf:lookup *configdat* "server" "timeout")))
    (if (and (string? tmo)
	     (common:hms-string->seconds tmo)) ;; BUG: hms-string->seconds is broken, if given "10" returns 0. Also, it doesn't belong in this logic unless the string->number is changed below
        (* 3600 (string->number tmo))
	60))) ;; default is one minute

(define (runs:get-mt-env-alist run-id runname target testname itempath)
  ;;(bb-check-path msg: "runs:set-megatest-env-vars entry")
  `(("MT_TEST_NAME"     . ,testname)
    
    ("MT_ITEMPATH"      . ,itempath)

(define (db:sync-inmem->disk dbstruct apath dbfile #!key (force-sync #f))
  (let* ((dbdat       (db:get-dbdat dbstruct apath dbfile))
	 (db          (dbr:dbdat-db dbdat))
	 (inmem       (dbr:dbdat-inmem dbdat))
	 (start-t     (current-seconds))
	 (last-update (dbr:dbdat-last-write dbdat))
	 (last-sync   (dbr:dbdat-last-sync dbdat)))
    (debug:print-info 0 *default-log-port* "Syncing for dbfile: " dbfile)
    (mutex-lock! *db-multi-sync-mutex*)
    (let* ((update_info (cons (if force-sync 0 last-update) "last_update"))
    	   (need-sync   (or force-sync (>= last-update last-sync))))
       (if need-sync
	  (db:sync-tables (db:sync-all-tables-list) update_info inmem db)
	  (debug:print 0 *default-log-port* "Skipping sync as nothing touched.")))
    (dbr:dbdat-last-sync-set! dbdat start-t)

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

    subgraph cluster_1 {
        node [style=filled,shape=box];

	rmt:send-receive -> "init-*remote* if needed" -> rmt:general-open-connection ->
	  rmt:send-receive-real;
	

//	check_available_queue       -> remove_entries_over_10s_old;
//	remove_entries_over_10s_old -> set_available [label="num_avail < 3"];
//	remove_entries_over_10s_old -> exit [label="num_avail > 2"];
//
//	set_available               -> delay_2s;
//	delay_2s          -> check_place_in_queue;
//
//	check_place_in_queue        -> "http:transport-launch" [label="at head"];
//	check_place_in_queue        -> exit [label="not at head"];
//
//	"client:login"              -> "server:shutdown" [label="login failed"];
//	"server:shutdown"           -> exit;	
//
//	subgraph cluster_2 {
//		"http:transport-launch"       -> "http:transport-run";
//		"http:transport-launch"       -> "http:transport-keep-running";
//
//		"http:transport-keep-running" -> "tests running?";
//		"tests running?"              -> "client:login" [label=yes];
//		"tests running?"              -> "server:shutdown" [label=no];
//		"client:login"                -> delay_5s [label="login ok"];
//		delay_5s                      -> "http:transport-keep-running";

//	}











//
//	// start_server -> "server_running?";
//	// "server_running?" -> set_available [label="no"];
//	// "server_running?" -> delay_2s [label="yes"];
//	// delay_2s -> "still_running?";
//	// "still_running?" -> ping_server [label=yes];
//	// "still_running?" -> set_available [label=no];
//	// ping_server -> exit [label=alive];
//	// ping_server -> remove_server_record [label=dead];
//	// remove_server_record -> set_available;
//	// set_available -> avail_delay [label="delay 3s"];
//	// avail_delay -> "first_in_queue?";
//	// 
//	// "first_in_queue?" -> set_running [label=yes];
//	// set_running -> get_next_port -> handle_requests;
//	// "first_in_queue?" -> "dead_entry_in_queue?" [label=no];
//	// "dead_entry_in_queue?" -> "server_running?" [label=no];
//	// "dead_entry_in_queue?" -> "remove_dead_entries" [label=yes];
//	// remove_dead_entries -> "server_running?";
//	// 
//	// handle_requests -> start_shutdown [label="no traffic\nno running tests"];
//	// handle_requests -> shutdown_request;
//	// start_shutdown -> shutdown_delay;
//	// shutdown_request -> shutdown_delay;
//	// shutdown_delay -> exit;
//	
//        label = "server:launch";
//        color=brown;
    }

//     client_start_server -> start_server;
//     handle_requests -> read_write;
//     read_write -> handle_requests;
}

  (let* ((start-time (current-seconds))
	 (am-server  (args:get-arg "-server"))
	 (dbfile     (args:get-arg "-db"))
	 (apath      *toppath*))
    (let loop ()
      (thread-sleep! 5) ;; add control / setting for this
      (if am-server
	  (if (not  *dbstruct-db*) ;; skip syncing until db is setup
	      (loop)
	      (begin
		(debug:print-info 0 *default-log-port* "syncing "apath" "dbfile" at "(current-seconds))
		(db:sync-inmem->disk *dbstruct-db* apath dbfile)
		(loop)))))))
  
;; 
;; (let ((dbstruct 
;; 	    (debug:print-info 13 *default-log-port* "after db:setup with dbstruct=" dbstruct)
;; 	    (cond
;; 	     ((dbr:dbstruct-read-only dbstruct)
;; 	      (debug:print-info 13 *default-log-port* "loading read-only watchdog")

	
	directory-utils
	;; http-client
	;; intarweb
	matchable
	md5
	message-digest
	nanomsg
	(prefix base64 base64:)
	(prefix sqlite3 sqlite3:)
	regex
	s11n
	;; spiffy
	;; spiffy-directory-listing
	;; spiffy-request-vars
	srfi-1
	srfi-13
	srfi-18
	srfi-69
	stack
	system-information
	;; tcp6
	typed-records
	uri-common
	z3
       
	apimod
	commonmod
	configfmod

;; info about me as a server
;;
(defstruct servdat
  (host #f)
  (port #f)
  (uuid #f)
  (rep  #f)
  (dbfile #f)
  (api-url #f)
  (api-uri #f)
  (api-req #f)
  (status 'starting)
  (trynum 0) ;; count the number of ports we've tried
  ) 

		#t)
	      (start-main-srv)))
	(start-main-srv))))

;; NB// remote is a rmt:remote struct
;;
(define (rmt:general-open-connection remote apath dbname #!key (num-tries 5))
  (let* ((mdbname (db:run-id->dbname #f))
	 (mconn   (rmt:get-conn remote apath mdbname)))
    (cond
     ((or (not mconn) ;; no channel open to main?
	  (< (rmt:conn-expires mconn)(+ (current-seconds) 2))) ;; restablish connection if less than 2 seconds on the lease
      (rmt:open-main-connection remote apath)

      (rmt:general-open-connection remote apath mdbname))
     ((not (rmt:get-conn remote apath dbname))                 ;; no channel open to dbname?     
      (let* ((res (rmt:send-receive-real remote apath mdbname 'get-server `(,apath ,dbname))))
	(case res
	  ((server-started)
	   (if (> num-tries 0)
	       (begin
		 (thread-sleep! 2)
		 (rmt:general-open-connection remote apath dbname num-tries: (- num-tries 1)))
	       (begin
		 (debug:print-error 0 *default-log-port* "Failed to start servers needed or open channel to "apath", "dbname)
		 (exit 1))))
	  (else
	   (if (list? res) ;; server has been registered and the info was returned. pass it on.
	       (begin ;;  ("192.168.0.9" 53817
		      ;;  "5e34239f48e8973b3813221e54701a01" "24310"
		      ;;  "192.168.0.9"
		      ;;  "/home/matt/data/megatest/tests/simplerun"
		 ;;  ".db/1.db")
		 (match
		  res
		  ((host port servkey pid ipaddr apath dbname)
		   (debug:print-info 0 *default-log-port* "got "res)
		   (hash-table-set! (rmt:remote-conns remote)
				    dbname
				    (make-rmt:conn
				     apath: apath
				     dbname: dbname
				     hostport: (conc host":"port)
				     ipaddr: ipaddr
				     port: port
				     srvkey: servkey
				     lastmsg: (current-seconds)
				     expires: (+ (current-seconds) 60))))
		  (else
		   (debug:print-info 0 *default-log-port* "return data from starting server did not match host port servkey pid ipaddr apath dbname " res)))
		 res)
	       (begin
		 (debug:print-info 0 *default-log-port* "Unexpected result: " res)
		 res))))))
     

     )))

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


;; Defaults to current area
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f))

  
;; 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 cmd params)
  (let* ((conn (rmt:get-conn remote apath dbname)))
    (assert conn "FATAL: rmt:send-receive-real called without the needed channels opened")
    (let* ((key     #f)

	   (host    (rmt:conn-ipaddr conn))
	   (port    (rmt:conn-port   conn))

	   (payload `((cmd    . ,cmd)
		      (key    . ,(rmt:conn-srvkey conn))
		      (params . ,params)))
	   (res      (open-send-receive-nn (conc host":"port)



					   (sexpr->string payload))))




      (string->sexpr res))))




;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;
;; Purpose - call the main.db server and request a server be started
;; for the given area path and dbname
;;
;; (define (rmt:send-receive-server-start remote apath dbname)
;;   (let* ((conn (rmt:get-conn remote apath dbname)))
;;     (assert conn "FATAL: Unable to connect to db "apath"/"dbname)
;;     #;(let* ((res      (with-input-from-request
;; 		      (rmt:conn->uri conn "api") 
;; 		      `((params . (,apath ,dbname)))
;; 		      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, "(seconds->year-week/day-time (current-seconds))"\n=====================")
    (debug:print 18 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg"))
    (for-each (lambda (cmd)
		(let ((cmd-dat (hash-table-ref *db-stats* cmd)))
		  (debug:print 18 *default-log-port* (format #f fmtstr cmd (vector-ref cmd-dat 0) (vector-ref cmd-dat 1) (/ (vector-ref cmd-dat 1)(vector-ref cmd-dat 0))))))
	      (sort (hash-table-keys *db-stats*)
		    (lambda (a b)
		      (> (vector-ref (hash-table-ref *db-stats* a) 0)

		       #f
		       (begin
			 (bdat-time-to-exit-set! *bdat* #t)
			 #t))))
    (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.")
    (if (and no-hurry (debug:debug-mode 18))
	(rmt:print-db-stats))
    (let ((th1 (make-thread
		(lambda () ;; thread for cleaning up, give it five seconds
		  (let* ((start-time (current-seconds)))
		    (if *server-info*



			(let ((dbfile   (servdat-dbfile *server-info*)))
			  (if dbfile
			      (let* ((am-server  (args:get-arg "-server"))
				     (dbfile     (args:get-arg "-db"))
				     (apath      *toppath*))
				;; do a final sync here
				(debug:print-info 0 *default-log-port* "Doing final sync for "apath" "dbfile" at "(current-seconds))
				(db:sync-inmem->disk *dbstruct-db* apath dbfile)
				(if am-server
				    (if (string-match ".*/main.db$" dbfile)
					(let ((pkt-file (conc (get-pkts-dir *toppath*)
							      "/" (servdat-uuid *server-info*)
							      ".pkt")))
					  (debug:print-info 0 *default-log-port* "removing pkt "pkt-file)
					  (delete-file* pkt-file)
					  (debug:print-info 0 *default-log-port* "Releasing lock for "dbfile)
					  (db:with-lock-db (servdat-dbfile *server-info*)
							   (lambda (dbh dbfile)
							     (db:release-lock dbh dbfile))))
					(let* ((sdat *server-info*) ;; we have a run-id server





					       (host (servdat-host sdat))
					       (port (servdat-port sdat))
					       (uuid (servdat-uuid sdat)))
					  (debug:print-info 0 *default-log-port* "deregistering server "host":"port" with uuid "uuid)
					  )))))))
		    (debug:print-info 0 *default-log-port* "Shutdown activities completed in "(- (current-seconds) start-time)" seconds"))
		  ;; (if *dbstruct-db* (db:close-all *dbstruct-db*)) ;; one second allocated
		  #;(if (bdat-task-db *bdat*)    ;; TODO: Check that this is correct for task db
		  (let ((db (cdr (bdat-task-db *bdat*))))
		  (if (sqlite3:database? db)
		  (begin
		  (debug:print-info 0 *default-log-port* "Closing down task db "db)
		  (sqlite3:interrupt! db)
		  (sqlite3:finalize! db #t)
		  (bdat-task-db-set! *bdat* #f)))))
		  #;(http-client#close-idle-connections!)
		  (if (not (eq? *default-log-port* (current-error-port)))
		      (close-output-port *default-log-port*))
		  (set! *default-log-port* (current-error-port))) "Cleanup db exit thread"))
	  (th2 (make-thread (lambda ()
			      (debug:print 4 *default-log-port* "Attempting clean exit. Mode="(if no-hurry "no-hurry" "normal")
					   " Please be patient and wait a few seconds...")
			      (if no-hurry
                                  (begin
                                    (thread-sleep! 5)) ;; give the clean up few seconds to do it's stuff
                                  (begin
				    (thread-sleep! 2)))
      			      (debug:print 4 *default-log-port* " ... done")
      			      )
			    "clean exit")))
      (thread-start! th1)
      (thread-start! th2)
      (thread-join! th1)
      )

      #f
      (conc "http://" (car hostport) ":" (cadr hostport))))

;;======================================================================
;; 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 hostn)
  ;;  ;; Configurations for server
  ;;  (tcp-buffer-size 2048)
  ;;  (max-connections 2048) 
  (debug:print 2 *default-log-port* "Attempting to start the server ...")
  (let* ((db              #f) ;;        (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily
	 (hostname        (get-host-name))

    (if *server-info*
	(begin
	  (servdat-host-set! *server-info* ipaddrstr)
	  (servdat-port-set! *server-info* port)
	  (servdat-status-set! *server-info* 'trying-port)
	  (servdat-trynum-set! *server-info* (+ (servdat-trynum *server-info*) 1)))
	(set! *server-info* (make-servdat host: ipaddrstr port: port)))
    (let* ((rep (rmt:try-start-server ipaddrstr port)))



      (let loop ((instr (nn-recv rep)))
	(let* ((data   (string->sexpr instr))




	       (res    (api:process-request *dbstruct-db* data))
	       (resdat (sexpr->string res)))
	  (set! *db-last-access* (current-seconds))
	  (nn-send rep resdat)
	  (loop (nn-recv rep)))))
    (let* ((portnum (servdat-port *server-info*)))
      (portlogger:open-run-close portlogger:set-port portnum "released")
      (debug:print 1 *default-log-port* "INFO: server has been stopped"))))

(define (rmt:try-start-server ipaddrstr portnum)
  (if *server-info* ;; update the server info as we might be trying next port
      (begin
	(servdat-host-set! *server-info* ipaddrstr)
	(servdat-port-set! *server-info* portnum)
	(servdat-status-set! *server-info* 'trying-port)
	(servdat-trynum-set! *server-info*
			     (+ (servdat-trynum *server-info*) 1)))
      (set! *server-info* (make-servdat host: ipaddrstr port: portnum)))
  (debug:print-info 0 *default-log-port* "rmt:try-start-server time="
		    (seconds->time-string (current-seconds))
		    " ipaddrsstr=" ipaddrstr
		    " portnum=" portnum)
  (if (is-port-in-use portnum)
      (begin
	(portlogger:open-run-close portlogger:set-failed portnum)
	(debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
	;; (thread-sleep! 0.1)
	(rmt:try-start-server ipaddrstr
			      (portlogger:open-run-close
			       portlogger:find-port)))
      (begin
	(if (not *server-info*)
	    (set! *server-info* (make-servdat
				 host: ipaddrstr
				 port: portnum)))
	(servdat-status-set! *server-info* 'starting)
	(servdat-port-set!   *server-info* portnum)
	(if (not (servdat-rep *server-info*))
	    (servdat-rep-set!    *server-info*  (nn-socket 'rep)))
	(let* ((rep (servdat-rep *server-info*)))
	  (debug:print 0 *default-log-port* "INFO: Trying to start server on " ipaddrstr ":" portnum)
	  (handle-exceptions
	   exn
	   (begin
	     (print-error-message exn)
	     (if (< portnum 64000)
		 (begin 
		   (debug:print 0 *default-log-port* "WARNING: attempt to start server failed. Trying again ...")
		   (debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
		   (debug:print 5 *default-log-port* "exn=" (condition->list exn))
		   (portlogger:open-run-close portlogger:set-failed portnum)
		   (debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
		   ;; (thread-sleep! 0.1)
		   (rmt:try-start-server ipaddrstr
					 (portlogger:open-run-close portlogger:find-port)))
		 (begin
		   (print "ERROR: Tried and tried but could not start the server, stopping at port "portnum))))




	   (nn-bind rep (conc "tcp://*:" portnum))

	   rep)))))

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

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

	 all-pkt-files)))

(define (server-address srv-pkt)
  (conc (alist-ref 'host srv-pkt) ":"
	(alist-ref 'port srv-pkt)))
	
(define (server-ready? host port key) ;; server-address is host:port
;;  (let-values (((i o)(handle-exceptions
;;		      exn
;;		      (values #f #f)
;;		      (tcp-connect host port))))
;;    (if (and i o)

  (let* ((data (sexpr->string  `((cmd . ping)
				 (key . ,key)
				 (params . ()))))
	 (res  (open-send-receive-nn (conc host ":" port) data)))
    (string->sexpr res)))

;; (let ((res (with-input-from-port i
;; 		       read)))
;; 	    (close-output-port o)
;; 	    (close-input-port i)
;; 	    res))
;; 	    (if (string? res)
;; 		(string->sexpr res)
;; 		res)))
;;     (begin ;; connection failed
;; 	  (debug:print-info 0 *default-log-port* "Server at "host":"port" is not responding.")
;; 	  #f))))

;; (define (loop-test host port data) ;; server-address is host:port
;;   ;; ping the server and ask it
;;   ;; if it ready
;;   ;; (let* ((sdat (servdat-init #f host port #f)))
;;   ;;   (http-transport:send-receive sdat "abc" 'ping '())))
;;   (let* ((payload (sexpr->string data))
;; 	 (res     (with-input-from-request

		    (exit))
		  (loop start-time
			(equal? sdat last-sdat)
			sdat))))))))

(define (rmt:register-server remote apath iface port server-key dbname)
  (rmt:open-main-connection remote apath) ;; we need a channel to main.db
  (rmt:send-receive-real remote apath      ;; params: host port servkey pid ipaddr dbpath
			 (db:run-id->dbname #f)
			 'register-server `(,iface
					    ,port
					    ,server-key
					    ,(current-process-id)
					    ,iface
					    ,apath
					    ,dbname)))

(define (http-transport:wait-for-stable-interface #!optional (num-tries-allowed 100))
  ;; wait until *server-info* stops changing
  (let* ((stime (current-seconds)))
    (let loop ((last-host  #f)
	       (last-port  #f)
	       (tries 0))

    ;; this is our forever loop
    (let* ((iface             (servdat-host *server-info*))
	   (port              (servdat-port *server-info*)))
      (let loop ((count         0)
		 (bad-sync-count 0)
		 (start-time     (current-milliseconds)))
	
	(if (and (not is-main)
		 (common:low-noise-print 60 "servdat-status"))
	    (debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *server-info*)))

	;; set up the database handle
	(mutex-lock! *heartbeat-mutex*)
	(if (not *dbstruct-db*) ;; no db opened yet, open the db and register with main if appropriate
	    (let ((watchdog (bdat-watchdog *bdat*)))		 
	      (debug:print 0 *default-log-port* "SERVER: dbprep")

		 (> (+ last-access server-timeout)
		    (current-seconds)))
	    (if (common:low-noise-print 120 "server continuing")
		(debug:print-info 0 *default-log-port* "Server continuing, seconds since last db access: " (- (current-seconds) last-access)))
	    (loop 0 bad-sync-count (current-milliseconds)))
	   (else
	    (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
	    (if (not (string-match ".db/main.db" (args:get-arg "-db")))
		(let* ((res (rmt:send-receive 'deregister-server #f
					      `(,(servdat-uuid sdat)
						,(current-process-id)
						,(servdat-host sdat)   ;; iface
						,(servdat-port sdat)))))
		(debug:print-info 0 *default-log-port* "deregistered-server, res="res)))
	    (http-transport:server-shutdown port))))))))

(define (http-transport:server-shutdown port)
  (begin
    ;;(BB> "http-transport:server-shutdown called")
    (debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id))
    ;;
    ;; start_shutdown
    ;;

    ;; deregister the server

    
    (bdat-time-to-exit-set! *bdat* #t) ;; tell on-exit to be fast as we've already cleaned up
    ;; (portlogger:open-run-close portlogger:set-port port "released") ;; done in rmt:run
    (thread-sleep! 1)

    ;; (debug:print-info 0 *default-log-port* "Max cached queries was    " *max-cache-size*)
    ;; (debug:print-info 0 *default-log-port* "Number of cached writes   " *number-of-writes*)
    ;; (debug:print-info 0 *default-log-port* "Average cached write time "
    ;; 		      (if (eq? *number-of-writes* 0)
    ;; 			  "n/a (no writes)"
    ;; 			  (/ *writes-total-delay*
    ;; 			     *number-of-writes*))
    ;; 		      " ms")
    ;; (debug:print-info 0 *default-log-port* "Number non-cached queries "  *number-non-write-queries*)
    ;; (debug:print-info 0 *default-log-port* "Average non-cached time   "
    ;; 		      (if (eq? *number-non-write-queries* 0)
    ;; 			  "n/a (no queries)"
    ;; 			  (/ *total-non-write-delay* 
    ;; 			     *number-non-write-queries*))
    ;; 		      " ms")
    
    (db:print-current-query-stats)
    #;(common:save-pkt `((action . exit)
                       (T      . server)
                       (pid    . ,(current-process-id)))
                     *configdat* #t)
    (debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting")
    (exit)))

;; Call this to start the actual server

					  (argv)))))))

(define (rmt:get-signature) 
  (if *my-signature* *my-signature*
      (let ((sig (rmt:mk-signature)))
        (set! *my-signature* sig)
        *my-signature*)))

;;======================================================================
;; Nanomsg transport
;;======================================================================

(define (is-port-in-use port-num)
  (let* ((ret #f))
    (let-values (((inp oup pid)
		  (process "netstat" (list  "-tulpn" ))))
      (let loop ((inl (read-line inp)))
        (if (not (eof-object? inl))
            (begin 
	      (if (string-search (regexp (conc ":" port-num)) inl)
		  (begin
					;(print "Output: "  inl)
		    (set! ret  #t))
		  (loop (read-line inp)))))))
    ret))

;;start a server, returns the connection
;;
(define (start-nn-server portnum )
  (let ((rep (nn-socket 'rep)))
    (handle-exceptions
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       (print "ERROR: Failed to start server \"" emsg "\"")
       (exit 1))
      
     (nn-bind rep (conc "tcp://*:" portnum)))
    rep))

;; open connection to server, send message, close connection
;;
(define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
  (let ((req  (nn-socket 'req))
        (uri  (conc "tcp://" host-port))
        (res  #f)
        ;; (contacts (alist-ref 'contact attrib))
        ;; (mode (alist-ref 'mode attrib))
	) 
    (handle-exceptions
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       ;; Send notification       
       (debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\"" )
       #f)
     (nn-connect req uri)
     ;; (print "Connected to the server " )
     (nn-send req msg)
     ;; (print "Request Sent")  
     (let* ((th1  (make-thread (lambda ()
                                 (let ((resp (nn-recv req)))
                                   (nn-close req)
                                   (set! res (if (equal? resp "ok")
                                                 #t
                                                 #f))))
                               "recv thread"))
            (th2 (make-thread (lambda ()
                                (thread-sleep! timeout)
                                (thread-terminate! th1))
			      "timer thread")))
       (thread-start! th1)
       (thread-start! th2)
       (thread-join! th1)
       res))))

(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
  (let ((req  (nn-socket 'req))
        (uri  (conc "tcp://" host-port))
        (res  #f)
	;;        (contacts (alist-ref 'contact attrib))
        ;; (mode (alist-ref 'mode attrib))
	) 
    (handle-exceptions
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       ;; Send notification      
       (debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\"" )
       #f)
     (nn-connect req uri)
     ;; (print "Connected to the server " )
     (nn-send req msg)
     ;; (print "Request Sent")  
     ;; receive code here
     ;;(print (nn-recv req))
     (let* ((th1  (make-thread (lambda ()
                                 (let ((resp (nn-recv req)))
                                   (nn-close req)
                                   (print resp)
                                   (set! res resp)))
                               "recv thread"))
            (th2 (make-thread (lambda ()
                                (thread-sleep! timeout)
                                (thread-terminate! th1))
                             "timer thread")))
       (thread-start! th1)
       (thread-start! th2)
       (thread-join! th1)
       res))))

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

;; run ping in separate process, safest way in some cases
;;

 ;; db:get-dbdat
 ;; rmt:find-main-server
;;  rmt:send-receive-real
;;  rmt:send-receive
 ;; sexpr->string
 ;; server-ready?
 ;; rmt:register-server
 rmt:open-main-connection
 rmt:general-open-connection
 rmt:get-conn
 ;; common:watchdog
 ;; rmt:find-main-server
 ;; get-all-server-pkts
 ;; get-viable-servers
 ;; get-best-candidate
 ;; api:run-server-process
 ;; rmt:run
 ;; rmt:try-start-server
 )

(define *db* (db:setup #f))

;; these let me cut and paste from source easily
(define apath *toppath*)
(define dbname ".db/2.db")
(define remote *rmt:remote*)
(define keyvals  '(("SYSTEM" "a")("RELEASE" "b")))

(test #f #t (rmt:open-main-connection remote apath))
(test #f #t (rmt:conn? (rmt:get-conn *rmt:remote* *toppath* ".db/main.db")))
(test #f #t (rmt:conn? (rmt:get-conn *rmt:remote* *toppath* ".db/main.db")))
(test #f 'server-started (rmt:send-receive-real *rmt:remote* *toppath* ".db/main.db" 'get-server `(,apath ,dbname)))

(thread-sleep! 2)
(test #f #t (list? (rmt:general-open-connection *rmt:remote* *toppath* ".db/2.db")))

(test #f '("SYSTEM" "RELEASE") (rmt:get-keys))
(test #f 1 (rmt:send-receive 'register-run #f (list keyvals "run1" "new" "n/a" "justme" #f)))
(print "Got here.")

(test #f 1 (rmt:send-receive 'register-run 1 (list keyvals "run1" "new" "n/a" "justme" #f)))

;; (test #f 1 (rmt:register-run keyvals "run2" "new" "n/a" "justme" #f))

(thread-sleep! 5)
(exit)


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

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

;; 
;; (test #f '(#t "exit process started") (rmt:kill-server)) ;; *toppath* *my-client-signature* #f)))
;; 

;; (exit)


;; ;; ;; ;; all old stuff below
;; ;; ;; 
;; ;; ;; 
;; ;; ;; 
;; ;; ;; 
;; ;; ;; (delete-file* "logs/1.log")
;; ;; ;; (define run-id 1)
;; ;; ;; 
;; ;; ;; (test "setup for run" #t (begin (launch:setup-for-run)
;; ;; ;;  				(string? (getenv "MT_RUN_AREA_HOME"))))
;; ;; ;; 
;; ;; ;; ;; Insert data into db
;; ;; ;; ;;
;; ;; ;; (define user    (current-user-name))
;; ;; ;; (define runname "mytestrun")
;; ;; ;; (define keys    (rmt:get-keys))
;; ;; ;; (define runinfo #f)
;; ;; ;; (define keyvals '(("SYSTEM" "abc")("RELEASE" "def")))
;; ;; ;; (define header  (list "SYSTEM" "RELEASE" "id" "runname" "state" "status" "owner" "event_time"))
;; ;; ;; 
;; ;; ;; (test #f 1 (rmt:register-run keyvals runname "new" "n/a" user))
;; ;; ;; ;; (test #f #f (rmt:get-runs-by-patt keys runname))
;; ;; ;; (test #f #t (rmt:general-call 'register-test run-id run-id "test-one" ""))
;; ;; ;; (define test-one-id #f)
;; ;; ;; (test #f 30001  (let ((test-id (rmt:get-test-id run-id "test-one" "")))
;; ;; ;; 	      (set! test-one-id test-id)
;; ;; ;; 	      test-id))
;; ;; ;; (define test-one-rec #f)
;; ;; ;; (test #f "test-one" (let ((test-rec (rmt:get-test-info-by-id run-id test-one-id)))
;; ;; ;; 		      (set! test-one-rec test-rec)
;; ;; ;; 		      (vector-ref test-rec 2)))
;; ;; ;; 
;; ;; ;; (use trace)
;; ;; ;; (import trace)
;; ;; ;; ;; (trace
;; ;; ;; ;;  rmt:send-receive
;; ;; ;; ;;  rmt:open-qry-close-locally
;; ;; ;; ;; )
;; ;; ;; 
;; ;; ;; ;; Tests to assess reading/writing while servers are starting/stopping
;; ;; ;; (define start-time (current-seconds))
;; ;; ;; (let loop ((test-state 'start))
;; ;; ;;   (let* ((server-dats (tasks:get-server-records (db:delay-if-busy (tasks:open-db)) run-id))
;; ;; ;; 	 (first-dat   (if (not (null? server-dats))
;; ;; ;; 			  (car server-dats)
;; ;; ;; 			  #f))
;; ;; ;; 	 (server-state (or (and first-dat (string->symbol (vector-ref first-dat 8))) 'no-dat)))
;; ;; ;;     (if first-dat
;; ;; ;; 	(map (lambda (dat)
;; ;; ;; 	       (apply print (intersperse (vector->list dat) ", ")))
;; ;; ;; 	     server-dats)
;; ;; ;; 	(print "No server"))
;; ;; ;;     (test #f test-one-rec (rmt:get-test-info-by-id run-id test-one-id))
;; ;; ;;     (thread-sleep! 1)
;; ;; ;;     (case test-state
;; ;; ;;       ((start)
;; ;; ;;        (print "Trying to start server")
;; ;; ;;        (server:kind-run run-id)
;; ;; ;;        (loop 'server-started))
;; ;; ;;       ((server-started)
;; ;; ;;        (case server-state 
;; ;; ;; 	 ((running)
;; ;; ;; 	  (print "Server appears to be running. Now ask it to shutdown")
;; ;; ;; 	  (rmt:kill-server run-id)
;; ;; ;; 	  ;; (trace rmt:open-qry-close-locally rmt:send-receive)
;; ;; ;; 	  (loop 'shutdown-started))
;; ;; ;; 	 ((available)
;; ;; ;; 	  (loop test-state))
;; ;; ;; 	 ((shutting-down)
;; ;; ;; 	  (loop test-state))
;; ;; ;; 	 ((no-dat)
;; ;; ;; 	  (loop test-state))
;; ;; ;; 	 (else (print "Don't know what to do if get here"))))
;; ;; ;;       ((shutdown-started)
;; ;; ;;        (case server-state 
;; ;; ;; 	 ((no-dat)
;; ;; ;; 	  (print "Server appears to have shutdown, ending this test"))
;; ;; ;; 	 (else
;; ;; ;; 	  (loop test-state)))))))
;; ;; ;; 
;; ;; ;; (exit)	 
;; ;; ;;        
;; ;; ;; ;; (set! *transport-type* 'http)
;; ;; ;; ;; 
;; ;; ;; ;; (test "setup for run" #t (begin (setup-for-run)
;; ;; ;; ;; 				(string? (getenv "MT_RUN_AREA_HOME"))))
;; ;; ;; ;; 
;; ;; ;; ;; (test "server-register, get-best-server" #t (let ((res #f))
;; ;; ;; ;; 					      (open-run-close tasks:server-register tasks:open-db 1 "bob" 1234 100 'live 'http)
;; ;; ;; ;; 					      (set! res (open-run-close tasks:get-best-server tasks:open-db))
;; ;; ;; ;; 					      (number? (vector-ref res 3))))
;; ;; ;; ;; 
;; ;; ;; ;; (test "de-register server" #f (let ((res #f))
;; ;; ;; ;; 				(open-run-close tasks:server-deregister tasks:open-db "bob" port: 1234)
;; ;; ;; ;; 				(vector? (open-run-close tasks:get-best-server tasks:open-db))))
;; ;; ;; ;; 
;; ;; ;; ;; (define server-pid #f)
;; ;; ;; ;; 
;; ;; ;; ;; ;; Not sure how the following should work, replacing it with system of megatest -server
;; ;; ;; ;; ;; (test "launch server" #t (let ((pid (process-fork (lambda ()
;; ;; ;; ;; ;; 						    ;; (daemon:ize)
;; ;; ;; ;; ;; 						    (server:launch 'http)))))
;; ;; ;; ;; ;; 			   (set! server-pid pid)
;; ;; ;; ;; ;; 			   (number? pid)))
;; ;; ;; ;; (system "../../bin/megatest -server - -debug 22 > server.log 2> server.log &")
;; ;; ;; ;; 
;; ;; ;; ;; (let loop ((n 10))
;; ;; ;; ;;   (thread-sleep! 1) ;; need to wait for server to start.
;; ;; ;; ;;   (let ((res (open-run-close tasks:get-best-server tasks:open-db)))
;; ;; ;; ;;     (print "tasks:get-best-server returned " res)
;; ;; ;; ;;     (if (and (not res)
;; ;; ;; ;; 	     (> n 0))
;; ;; ;; ;; 	(loop (- n 1)))))
;; ;; ;; ;; 
;; ;; ;; ;; (test "get-best-server" #t (begin 
;; ;; ;; ;; 			     (client:launch)
;; ;; ;; ;; 			     (let ((dat (open-run-close tasks:get-best-server tasks:open-db)))
;; ;; ;; ;; 			       (vector? dat))))
;; ;; ;; ;; 
;; ;; ;; ;; (define *keys*               (keys:config-get-fields *configdat*))
;; ;; ;; ;; (define *keyvals*            (keys:target->keyval *keys* "a/b/c"))
;; ;; ;; ;; 
;; ;; ;; ;; (test #f #t                       (string? (car *runremote*)))
;; ;; ;; ;; (test #f '(#t "successful login") (rmt:login)) ;;  *runremote* *toppath* *my-client-signature*)))
;; ;; ;; ;; 
;; ;; ;; ;; (test #f #f                       (rmt:get-test-info-by-id 1 99)) ;; get non-existant test
;; ;; ;; ;; 
;; ;; ;; ;; ;; RUNS
;; ;; ;; ;; (test #f 1                        (rmt:register-run  *keyvals* "firstrun" "new" "n/a" (current-user-name)))
;; ;; ;; ;; (test "get run info"  "firstrun"  (let ((rinfo (rmt:get-run-info 1)))
;; ;; ;; ;; 				    (vector-ref (vector-ref rinfo 1) 3)))
;; ;; ;; ;; (test "get runname from id" "firstrun" (rmt:get-run-name-from-id 1))
;; ;; ;; ;; 
;; ;; ;; ;; ;; TESTS
;; ;; ;; ;; (test "get tests (no data)" '()   (rmt:get-tests-for-run 1 "%" '() '() #f #f #f #f #f #f))
;; ;; ;; ;; (test "register test"       #t    (rmt:general-call 'register-test 1 1 "test1" ""))
;; ;; ;; ;; (test "get tests (some data)"  1  (length (rmt:get-tests-for-run 1 "%" '() '() #f #f #f #f #f #f)))
;; ;; ;; ;; (test "get test id"            1  (rmt:get-test-id 1 "test1" ""))
;; ;; ;; ;; 
;; ;; ;; ;; (test "sync back"              #t (> (rmt:sync-inmem->db) 0))
;; ;; ;; ;; (test "get test id from main"  1  (db:get-test-id *db* 1 "test1" ""))
;; ;; ;; ;; 
;; ;; ;; ;; (test "get keys"               #t (list? (rmt:get-keys)))
;; ;; ;; ;; (test "set comment"            #t (begin (rmt:general-call 'set-test-comment 1 "this is a comment" 1) #t))
;; ;; ;; ;; (test "get comment" "this is a comment" (let ((trec (rmt:get-test-info-by-id 1 1)))
;; ;; ;; ;; 					  (db:test-get-comment trec)))
;; ;; ;; ;; 
;; ;; ;; ;; ;; MORE RUNS
;; ;; ;; ;; (test "get runs"  #t (let* ((runs   (rmt:get-runs "%" #f #f '()))
;; ;; ;; ;; 			    (header (vector-ref runs 0))
;; ;; ;; ;; 			    (data   (vector-ref runs 1)))
;; ;; ;; ;; 		       (and (list?   header)
;; ;; ;; ;; 			    (list?   data)
;; ;; ;; ;; 			    (vector? (car data)))))
;; ;; ;; ;; 
;; ;; ;; ;; (test "get local testinfo" "test1" (vector-ref (db:get-testinfo-state-status *db* 1 1) 2))
;; ;; ;; ;; (test "get testinfo"       "test1" (vector-ref (rmt:get-testinfo-state-status 1 1) 2))
;; ;; ;; ;; 
;; ;; ;; ;; ;;======================================================================
;; ;; ;; ;; ;; D B
;; ;; ;; ;; ;;======================================================================
;; ;; ;; ;; 
;; ;; ;; ;; (test "pass fail counts" #t (rmt:general-call 'pass-fail-counts 10 9 1))
;; ;; ;; ;; (test "get pass fail counts" 19 (let ((dat (rmt:get-test-info-by-id 1)))
;; ;; ;; ;; 				  (+ (db:test-get-pass_count dat)
;; ;; ;; ;; 				     (db:test-get-fail_count dat))))
;; ;; ;; ;; 
;; ;; ;; ;; (define testregistry (make-hash-table))
;; ;; ;; ;; (for-each
;; ;; ;; ;;  (lambda (tname)
;; ;; ;; ;;    (for-each
;; ;; ;; ;;     (lambda (itempath)
;; ;; ;; ;;       (let ((tkey  (conc tname "/" itempath))
;; ;; ;; ;; 	    (rpass (random 10))
;; ;; ;; ;; 	    (rfail (random 10)))
;; ;; ;; ;; 	(hash-table-set! testregistry tkey (list tname itempath))
;; ;; ;; ;; 	(rmt:general-call 'register-test 1 tname itempath)
;; ;; ;; ;; 	(let* ((tid  (rmt:get-test-id 1 tname itempath))
;; ;; ;; ;; 	       (tdat (rmt:get-test-info-by-id tid)))
;; ;; ;; ;; 	  (rmt:general-call 'pass-fail-counts rpass rfail (db:test-get-id tdat))
;; ;; ;; ;; 	  (let* ((resdat (rmt:get-test-info-by-id tid)))
;; ;; ;; ;; 	    (test "set/get pass fail counts" (list rpass rfail)
;; ;; ;; ;; 		  (list (db:test-get-pass_count resdat)
;; ;; ;; ;; 			(db:test-get-fail_count resdat)))))))
;; ;; ;; ;;     (list "" "a" "b" "c" "d" "e" "f" "g" "h" "i" "j")))
;; ;; ;; ;;  (list "test1" "test2" "test3" "test4" "test5"))
;; ;; ;; ;; 
;; ;; ;; ;; 
;; ;; ;; ;; (test #f '(#t "exit process started") (rmt:kill-server)) ;; *toppath* *my-client-signature* #f)))
;; ;; ;; ;; 
;; ;; ;;