;;======================================================================
;; Copyright 2017, Matthew Welland.
;;
;; This file is part of Megatest.
;;
;; Megatest is free software: you can redistribute it and/or modify
;; it under the terms of the GNU General Public License as published by
;; the Free Software Foundation, either version 3 of the License, or
;; (at your option) any later version.
;;
;; Megatest is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
;; GNU General Public License for more details.
;;
;; You should have received a copy of the GNU General Public License
;; along with Megatest. If not, see <http://www.gnu.org/licenses/>.
;;======================================================================
(declare (unit rmtmod))
(declare (uses commonmod))
(declare (uses configfmod))
(declare (uses apimod))
(declare (uses itemsmod))
(declare (uses debugprint))
(declare (uses mtver))
(declare (uses tasksmod))
(declare (uses pgdb))
(declare (uses mtargs))
(declare (uses dbmod))
(declare (uses http-transportmod))
(declare (uses servermod))
(declare (uses clientmod))
(module rmtmod
*
(import scheme
chicken.random
chicken.string
chicken.base
chicken.condition
chicken.sort
chicken.time
chicken.base
chicken.file
chicken.format
chicken.process
chicken.file.posix
chicken.process-context.posix
chicken.process-context
(prefix sqlite3 sqlite3:)
typed-records
srfi-1
srfi-13
srfi-18
srfi-69
commonmod
apimod
itemsmod
debugprint
mtver
regex
tasksmod
pgdb
(prefix mtargs args:)
dbmod
http-transportmod
servermod
clientmod
configfmod
)
(defstruct alldat
(areapath #f)
(ulexdat #f)
)
;; (include "db_records.scm")
;;======================================================================
;; return the handle struct for sending queries to a specific database
;; - initializes the connection object if this is the first access
;; - finds the "captain" and asks who to talk to for the given dbfname
;; - establishes the connection to the current dbowner
;;
#;(define (rmt:connect alldat dbfname dbtype)
(let* ((ulexdat (or (alldat-ulexdat alldat)
(rmt:setup-ulex alldat))))
(ulex:connect ulexdat dbfname dbtype)))
;; setup the remote calls
#;(define (rmt:setup-ulex alldat)
(let* ((udata (ulex:setup))) ;; establish connection to ulex
(alldat-ulexdat-set! alldat udata)
;; register all needed procs
(ulex:register-handler udata 'ping cmod:get-full-version) ;; override ping with get-full-version
(ulex:register-handler udata 'login cmod:get-full-version) ;; force setup of the connection
(ulex:register-handler udata 'execute api:execute-requests)
udata))
;; set up a connection to the current owner of the dbfile associated with rid
;; then send the query to that dbfile owner and wait for a response.
;;
#;(define (rmt:send-receive alldat cmd rid params #!key (attemptnum 1)(area-dat #f)) ;; start attemptnum at 1 so the modulo below works as expected
(let* (;; (alldat *alldat*)
(areapath (alldat-areapath alldat))
(dbtype (if (or (not rid)(< rid 1)) ;; this is the criteria for "main.db"
"main" "runs"))
(dbfname (if (equal? dbtype "main")
"main.db"
(conc rid ".db")))
(dbfile (conc areapath "/.db/" dbfname))
(ulexconn (rmt:connect alldat dbfname dbtype)) ;; ulexconn is our new *runremote*, it is a dbowner struct < pdat lastrefresh >
(udata (alldat-ulexdat alldat)))
(ulex:remote-request udata ulexconn 'immediate dbfile 'execute rid params)))
;; need to call this on the other side
;; (api:execute-requests dbstruct-local (vector (symbol->string cmd) params))))
#;(with-input-from-string
(ulex:remote-request udata ulexconn 'immediate dbfile 'execute rid (with-output-to-string (lambda ()(serialize params))))
(lambda ()(deserialize)))
;;
;; THESE ARE ALL CALLED ON THE CLIENT SIDE!!!
;;
;; generate entries for ~/.megatestrc with the following
;;
;; grep define ../rmt.scm | grep rmt: |perl -pi -e 's/\(define\s+\((\S+)\W.*$/\1/'|sort -u
;;======================================================================
;; S U P P O R T F U N C T I O N S
;;======================================================================
;; if a server is either running or in the process of starting call client:setup
;; else return #f to let the calling proc know that there is no server available
;;
(define (rmt:get-connection-info areapath #!key (area-dat #f)) ;; TODO: push areapath down.
(let* ((runremote (or area-dat *runremote*))
(cinfo (if (remote? runremote)
(remote-conndat runremote)
#f)))
(if cinfo
cinfo
(if (server:check-if-running areapath)
(client:setup areapath)
#f))))
(defstruct rmt:remote
(conns (make-hash-table)) ;; apath/dbname => rmt:conn
)
(defstruct rmt:conn
(apath #f)
(dbname #f)
(fullname #f)
(hostport #f)
(lastmsg 0)
(expires 0))
(define *rmt:remote* (make-rmt:remote))
;; do we have a connection to apath dbname and
;; is it not expired? then return it
;;
(define (rmt:get-existing-live-conn remote apath dbname)
(let* ((fullname (db:dbname->path apath dbname))
(conn (hash-table-ref/default (rmt:remote-conns remote) fullname #f)))
(if (and conn
(> (current-seconds) (rmt:conn-expires conn)))
conn
#f)))
;; looks for a connection to main
;; connections for other servers happens by requesting from main
;;
(define (rmt:open-main-connection remote apath)
(let* ((pktsdir (get-pkts-dir apath))
(all-srvpkts (get-all-server-pkts pktsdir *srvpktspec*))
(viable-srvs (get-viable-servers all-srvpkts apath))
(the-srv (get-the-server viable-srvs apath))
(dbname (db:run-id->dbname #f))
(start-main-srv (lambda ()
;; srv not ready, delay a little and try again
(system (conc "nbfake megatest -server - -area "apath" -db "dbname))
(thread-sleep! 1.5)
(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))
(srvready (server-ready? srv-addr))
(fullpath (db:dbname->path apath dbname)))
(if srvready
(hash-table-set! (rmt:remote-conns remote)
fullpath
(make-rmt:conn
apath: apath
dbname: dbname
fullname: fullpath
hostport: srv-addr
lastmsg: (current-seconds)
expires: (+ (current-seconds) 60) ;; this needs to be gathered during the ping
))
(start-main-srv)))
(start-main-srv))))
(define (rmt:general-open-connection remote apath dbname)
(let ((mainconn (rmt:get-existing-live-conn remote apath (db:run-id->dbname #f))))
(if (not mainconn)(rmt:open-main-connection remote apath))
;; TODO - call main for connection info
))
;;======================================================================
;; RA => e.g. usage (rmt:send-receive 'get-var #f (list varname))
;;
(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 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 params)
;; do we have a connection to the needed db?
;; has the connection expired?
(let connloop ((conn (rmt:get-existing-live-conn remote apath dbname)))
(if (not conn)
(connloop (rmt:general-open-connection remote apath dbname))
(begin
#t ;; here we do the actual connection work
))))
;; ;; ;; start attemptnum at 1 so the modulo below works as expected
;; ;; #;(common:telemetry-log (conc "rmt:"(->string cmd))
;; ;; payload: `((rid . ,rid)
;; ;; (params . ,params)))
;; ;;
;; ;; (if (> attemptnum 2)
;; ;; (debug:print 0 *default-log-port* "INFO: attemptnum in rmt:send-receive is " attemptnum))
;; ;;
;; ;; (cond
;; ;; ((> attemptnum 2) (thread-sleep! 0.053))
;; ;; ((> attemptnum 10) (thread-sleep! 0.5))
;; ;; ((> attemptnum 20) (thread-sleep! 1)))
;; ;; (if (and (> attemptnum 5) (= 0 (modulo attemptnum 15)))
;; ;; (begin (server:run *toppath*) (thread-sleep! 3)))
;; ;;
;; ;;
;; ;; ;;DOT digraph megatest_state_status {
;; ;; ;;DOT ranksep=0;
;; ;; ;;DOT // rankdir=LR;
;; ;; ;;DOT node [shape="box"];
;; ;; ;;DOT "rmt:send-receive" -> MUTEXLOCK;
;; ;; ;;DOT { edge [style=invis];"case 1" -> "case 2" -> "case 3" -> "case 4" -> "case 5" -> "case 6" -> "case 7" -> "case 8" -> "case 9" -> "case 10" -> "case 11"; }
;; ;; ;; do all the prep locked under the rmt-mutex
;; ;; (mutex-lock! *rmt-mutex*)
;; ;;
;; ;; ;; 1. check if server is started IFF cmd is a write OR if we are not on the homehost, store in runremote
;; ;; ;; 2. check the age of the connections. refresh the connection if it is older than timeout-20 seconds.
;; ;; ;; 3. do the query, if on homehost use local access
;; ;; ;;
;; ;; (let* ((start-time (current-seconds)) ;; snapshot time so all use cases get same value
;; ;; (areapath *toppath*);; TODO - resolve from dbstruct to be compatible with multiple areas
;; ;; (runremote (or area-dat
;; ;; *runremote*))
;; ;; (attemptnum (+ 1 attemptnum))
;; ;; (readonly-mode (rmtmod:calc-ro-mode runremote *toppath*)))
;; ;;
;; ;; ;; DOT INIT_RUNREMOTE; // leaving off - doesn't really add to the clarity
;; ;; ;; DOT MUTEXLOCK -> INIT_RUNREMOTE [label="no remote?"];
;; ;; ;; DOT INIT_RUNREMOTE -> MUTEXLOCK;
;; ;; ;; ensure we have a record for our connection for given area
;; ;; (if (not runremote) ;; can remove this one. should never get here.
;; ;; (begin
;; ;; (set! *runremote* (make-and-init-remote))
;; ;; (let* ((server-info (remote-server-info *runremote*)))
;; ;; (if server-info
;; ;; (begin
;; ;; (remote-server-url-set! *runremote* (server:record->url server-info))
;; ;; (remote-server-id-set! *runremote* (server:record->id server-info)))))
;; ;; (set! runremote *runremote*))) ;; new runremote will come from this on next iteration
;; ;;
;; ;; ;; DOT SET_HOMEHOST; // leaving off - doesn't really add to the clarity
;; ;; ;; DOT MUTEXLOCK -> SET_HOMEHOST [label="no homehost?"];
;; ;; ;; DOT SET_HOMEHOST -> MUTEXLOCK;
;; ;; ;; ensure we have a homehost record
;; ;; (if (not (pair? (remote-hh-dat runremote))) ;; not on homehost
;; ;; (thread-sleep! 0.1) ;; since we shouldn't get here, delay a little
;; ;; (remote-hh-dat-set! runremote (common:get-homehost)))
;; ;;
;; ;; ;;(print "BB> readonly-mode is "readonly-mode" dbfile is "dbfile)
;; ;; (cond
;; ;; ;;DOT EXIT;
;; ;; ;;DOT MUTEXLOCK -> EXIT [label="> 15 attempts"]; {rank=same "case 1" "EXIT" }
;; ;; ;; give up if more than 150 attempts
;; ;; ((> attemptnum 150)
;; ;; (debug:print 0 *default-log-port* "ERROR: 150 tries to start/connect to server. Giving up.")
;; ;; (exit 1))
;; ;;
;; ;; ;;DOT CASE2 [label="local\nreadonly\nquery"];
;; ;; ;;DOT MUTEXLOCK -> CASE2; {rank=same "case 2" CASE2}
;; ;; ;;DOT CASE2 -> "rmt:open-qry-close-locally";
;; ;; ;; readonly mode, read request- handle it - case 2
;; ;; ((and readonly-mode
;; ;; (member cmd api:read-only-queries))
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 2")
;; ;; (rmt:open-qry-close-locally cmd 0 params)
;; ;; )
;; ;;
;; ;; ;;DOT CASE3 [label="write in\nread-only mode"];
;; ;; ;;DOT MUTEXLOCK -> CASE3 [label="readonly\nmode?"]; {rank=same "case 3" CASE3}
;; ;; ;;DOT CASE3 -> "#f";
;; ;; ;; readonly mode, write request. Do nothing, return #f
;; ;; (readonly-mode (extras-readonly-mode *rmt-mutex* *default-log-port* cmd params))
;; ;;
;; ;; ;; This block was for pre-emptively resetting the connection if there had been no communication for some time.
;; ;; ;; I don't think it adds any value. If the server is not there, just fail and start a new connection.
;; ;; ;; also, the expire-time calculation might not be correct. We want, time-since-last-server-access > (server:get-timeout)
;; ;; ;;
;; ;; ;;DOT CASE4 [label="reset\nconnection"];
;; ;; ;;DOT MUTEXLOCK -> CASE4 [label="have connection,\nlast_access > expire_time"]; {rank=same "case 4" CASE4}
;; ;; ;;DOT CASE4 -> "rmt:send-receive";
;; ;; ;; reset the connection if it has been unused too long
;; ;; ((and runremote
;; ;; (remote-conndat runremote)
;; ;; (> (current-seconds) ;; if it has been more than server-timeout seconds since last contact, close this connection and start a new on
;; ;; (+ (http-transport:server-dat-get-last-access (remote-conndat runremote))
;; ;; (remote-server-timeout runremote))))
;; ;; (debug:print-info 0 *default-log-port* "Connection to " (remote-server-url runremote) " expired due to no accesses, forcing new connection.")
;; ;; (http-transport:close-connections area-dat: runremote)
;; ;; (remote-conndat-set! runremote #f) ;; invalidate the connection, thus forcing a new connection.
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (rmt:send-receive cmd rid params attemptnum: attemptnum))
;; ;;
;; ;; ;;DOT CASE5 [label="local\nread"];
;; ;; ;;DOT MUTEXLOCK -> CASE5 [label="server not required,\non homehost,\nread-only query"]; {rank=same "case 5" CASE5};
;; ;; ;;DOT CASE5 -> "rmt:open-qry-close-locally";
;; ;;
;; ;; ;; on homehost and this is a read
;; ;; ((and (not (remote-force-server runremote)) ;; honor forced use of server, i.e. server NOT required
;; ;; (cdr (remote-hh-dat runremote)) ;; on homehost
;; ;; (member cmd api:read-only-queries)) ;; this is a read
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 5")
;; ;; (rmt:open-qry-close-locally cmd 0 params))
;; ;;
;; ;; ;;DOT CASE6 [label="init\nremote"];
;; ;; ;;DOT MUTEXLOCK -> CASE6 [label="on homehost,\nwrite query,\nhave server,\ncan't reach it"]; {rank=same "case 6" CASE6};
;; ;; ;;DOT CASE6 -> "rmt:send-receive";
;; ;; ;; on homehost and this is a write, we already have a server, but server has died
;; ;; ((and (cdr (remote-hh-dat runremote)) ;; on homehost
;; ;; (not (member cmd api:read-only-queries)) ;; this is a write
;; ;; (remote-server-url runremote) ;; have a server
;; ;; (not (server:ping (remote-server-url runremote) (remote-server-id runremote)))) ;; server has died. NOTE: this is not a cheap call! Need better approach.
;; ;; (set! *runremote* (make-and-init-remote))
;; ;; (let* ((server-info (remote-server-info *runremote*)))
;; ;; (if server-info
;; ;; (begin
;; ;; (remote-server-url-set! *runremote* (server:record->url server-info))
;; ;; (remote-server-id-set! *runremote* (server:record->id server-info)))))
;; ;; (remote-force-server-set! runremote (common:force-server?))
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 6")
;; ;; (rmt:send-receive cmd rid params attemptnum: attemptnum))
;; ;;
;; ;; ;;DOT CASE7 [label="homehost\nwrite"];
;; ;; ;;DOT MUTEXLOCK -> CASE7 [label="server not required,\non homehost,\na write,\nhave a server"]; {rank=same "case 7" CASE7};
;; ;; ;;DOT CASE7 -> "rmt:open-qry-close-locally";
;; ;; ;; on homehost and this is a write, we already have a server
;; ;; ((and (not (remote-force-server runremote)) ;; honor forced use of server, i.e. server NOT required
;; ;; (cdr (remote-hh-dat runremote)) ;; on homehost
;; ;; (not (member cmd api:read-only-queries)) ;; this is a write
;; ;; (remote-server-url runremote)) ;; have a server
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 4.1")
;; ;; (rmt:open-qry-close-locally cmd 0 params))
;; ;;
;; ;; ;;DOT CASE8 [label="force\nserver"];
;; ;; ;;DOT MUTEXLOCK -> CASE8 [label="server not required,\nhave homehost info,\nno connection yet,\nnot a read-only query"]; {rank=same "case 8" CASE8};
;; ;; ;;DOT CASE8 -> "rmt:open-qry-close-locally";
;; ;; ;; on homehost, no server contact made and this is a write, passively start a server
;; ;; ((and (not (remote-force-server runremote)) ;; honor forced use of server, i.e. server NOT required
;; ;; (cdr (remote-hh-dat runremote)) ;; have homehost
;; ;; (not (remote-server-url runremote)) ;; no connection yet
;; ;; (not (member cmd api:read-only-queries))) ;; not a read-only query
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 8")
;; ;; (let ((server-info (server:check-if-running *toppath*))) ;; (server:read-dotserver->url *toppath*))) ;; (server:check-if-running *toppath*))) ;; Do NOT want to run server:check-if-running - very expensive to do for every write call
;; ;; (if server-info
;; ;; (begin
;; ;; (remote-server-url-set! runremote (server:record->url server-info)) ;; the string can be consumed by the client setup if needed
;; ;; (remote-server-id-set! runremote (server:record->id server-info)))
;; ;; (if (common:force-server?)
;; ;; (server:start-and-wait *toppath*)
;; ;; (server:kind-run *toppath*)))
;; ;; (remote-force-server-set! runremote (common:force-server?))
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 8.1")
;; ;; (rmt:open-qry-close-locally cmd 0 params)))
;; ;;
;; ;; ;;DOT CASE9 [label="force server\nnot on homehost"];
;; ;; ;;DOT MUTEXLOCK -> CASE9 [label="no connection\nand either require server\nor not on homehost"]; {rank=same "case 9" CASE9};
;; ;; ;;DOT CASE9 -> "start\nserver" -> "rmt:send-receive";
;; ;; ((or (and (remote-force-server runremote) ;; we are forcing a server and don't yet have a connection to one
;; ;; (not (remote-conndat runremote)))
;; ;; (and (not (cdr (remote-hh-dat runremote))) ;; not on a homehost
;; ;; (not (remote-conndat runremote)))) ;; and no connection
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 9, hh-dat: " (remote-hh-dat runremote) " conndat: " (remote-conndat runremote))
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (if (not (server:check-if-running *toppath*)) ;; who knows, maybe one has started up?
;; ;; (server:start-and-wait *toppath*))
;; ;; (remote-conndat-set! runremote (rmt:get-connection-info *toppath*)) ;; calls client:setup which calls client:setup-http
;; ;; (rmt:send-receive cmd rid params attemptnum: attemptnum)) ;; TODO: add back-off timeout as
;; ;;
;; ;; ;;DOT CASE10 [label="on homehost"];
;; ;; ;;DOT MUTEXLOCK -> CASE10 [label="server not required,\non homehost"]; {rank=same "case 10" CASE10};
;; ;; ;;DOT CASE10 -> "rmt:open-qry-close-locally";
;; ;; ;; all set up if get this far, dispatch the query
;; ;; ((and (not (remote-force-server runremote))
;; ;; (cdr (remote-hh-dat runremote))) ;; we are on homehost
;; ;; (mutex-unlock! *rmt-mutex*)
;; ;; (debug:print-info 12 *default-log-port* "rmt:send-receive, case 10")
;; ;; (rmt:open-qry-close-locally cmd (if rid rid 0) params))
;; ;;
;; ;; ;;DOT CASE11 [label="send_receive"];
;; ;; ;;DOT MUTEXLOCK -> CASE11 [label="else"]; {rank=same "case 11" CASE11};
;; ;; ;;DOT CASE11 -> "rmt:send-receive" [label="call failed"];
;; ;; ;;DOT CASE11 -> "RESULT" [label="call succeeded"];
;; ;; ;; not on homehost, do server query
;; ;; (else (extras-case-11 *default-log-port* runremote cmd params attemptnum rid)))))
;; ;; ;;DOT }
;; bunch of small functions factored out of send-receive to make debug easier
;;
;; No Title
;; Error: (vector-ref) out of range
;; #(#<condition: (exn type)> (#("db.scm:3740: regex#regexp" #f #f) #("db.scm:3739: regex#string-substitute" #f #f) #("db.scm:3738: base64#base64-decode" #f #f) #("db.scm:3737: z3#z3:decode-buffer" #f #f) #("db.scm:3736: with-input-from-string" #f #f) #("db.scm:3741: s11n#deserialize" #f #f) #("api.scm:374: api:execute-requests" #f #f) #("api.scm:139: call-with-current-continuation" #f #f) #("api.scm:139: with-exception-handler" #f #f) #("api.scm:139: ##sys#call-with-values" #f #f) #("api.scm:158: string->symbol" #f #f) #("api.scm:160: current-milliseconds" #f #f) #("api.scm:161: dbr:dbstruct-read-only" #f #f) #("api.scm:139: k15" #f #f) #("api.scm:139: g19" #f #f) #("api.scm:142: get-call-chain" #f #f)) #("get-test-info-by-id" (1102 507299)))
;; 6
;;
;; Call history:
;;
;; http-transport.scm:306: thread-terminate!
;; http-transport.scm:307: debug:print-info
;; common_records.scm:235: debug:debug-mode
;; rmt.scm:259: k587
;; rmt.scm:259: g591
;; rmt.scm:276: http-transport:server-dat-update-last-access
;; http-transport.scm:364: current-seconds
;; rmt.scm:282: debug:print-info
;; common_records.scm:235: debug:debug-mode
;; rmt.scm:283: mutex-unlock!
;; rmt.scm:287: extras-transport-succeded <--
;; +-----------------------------------------------------------------------------+
;; | Exit Status : 70
;;
(define (extras-case-11 *default-log-port* runremote cmd params attemptnum rid)
;; (mutex-unlock! *rmt-mutex*)
(debug:print-info 12 *default-log-port* "rmt:send-receive, case 9")
;; (mutex-lock! *rmt-mutex*)
(let* ((conninfo (remote-conndat runremote))
(dat-in (condition-case ;; handling here has
;; caused a lot of
;; problems. However it
;; is needed to deal with
;; attemtped
;; communication to
;; servers that have gone
;; away
(http-transport:client-api-send-receive 0 conninfo cmd params)
((servermismatch) (vector #f "Server id mismatch" ))
((commfail)(vector #f "communications fail"))
((exn)(vector #f "other fail" (print-call-chain)))))
(dat (if (and (vector? dat-in) ;; ... check it is a correct size
(> (vector-length dat-in) 1))
dat-in
(vector #f (conc "communications fail (type 2), dat-in=" dat-in))))
(success (if (vector? dat) (vector-ref dat 0) #f))
(res (if (vector? dat) (vector-ref dat 1) #f)))
(if (and (vector? conninfo) (< 5 (vector-length conninfo)))
(http-transport:server-dat-update-last-access conninfo) ;; refresh access time
(begin
(debug:print 0 *default-log-port* "INFO: Should not get here! conninfo=" conninfo)
(set! conninfo #f)
(remote-conndat-set! *runremote* #f) ;; NOTE: *runremote* is global copy of runremote. Purpose: factor out global.
(http-transport:close-connections area-dat: runremote)))
(debug:print-info 13 *default-log-port* "rmt:send-receive, case 9. conninfo=" conninfo " dat=" dat " runremote = " runremote)
(mutex-unlock! *rmt-mutex*)
(if success ;; success only tells us that the transport was
;; successful, have to examine the data to see if
;; there was a detected issue at the other end
(extras-transport-succeded *default-log-port* *rmt-mutex* attemptnum runremote res params rid cmd)
(begin
(debug:print-error 0 *default-log-port* " dat=" dat)
(extras-transport-failed *default-log-port* *rmt-mutex* attemptnum runremote cmd rid params))
)))
(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"))
(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)
(vector-ref (hash-table-ref *db-stats* b) 0)))))))
(define (rmt:get-max-query-average run-id)
(mutex-lock! *db-stats-mutex*)
(let* ((runkey (conc "run-id=" run-id " "))
(cmds (filter (lambda (x)
(substring-index runkey x))
(hash-table-keys *db-stats*)))
(res (if (null? cmds)
(cons 'none 0)
(let loop ((cmd (car cmds))
(tal (cdr cmds))
(max-cmd (car cmds))
(res 0))
(let* ((cmd-dat (hash-table-ref *db-stats* cmd))
(tot (vector-ref cmd-dat 0))
(curravg (/ (vector-ref cmd-dat 1) (vector-ref cmd-dat 0))) ;; count is never zero by construction
(currmax (max res curravg))
(newmax-cmd (if (> curravg res) cmd max-cmd)))
(if (null? tal)
(if (> tot 10)
(cons newmax-cmd currmax)
(cons 'none 0))
(loop (car tal)(cdr tal) newmax-cmd currmax)))))))
(mutex-unlock! *db-stats-mutex*)
res))
(define (rmt:open-qry-close-locally cmd run-id params #!key (remretries 5))
(let* ((qry-is-write (not (member cmd api:read-only-queries)))
(db-file-path (common:get-db-tmp-area)) ;; 0))
(dbstruct-local (db:setup #t)) ;; make-dbr:dbstruct path: dbdir local: #t)))
(read-only (not (file-writable? db-file-path)))
(start (current-milliseconds))
(resdat (if (not (and read-only qry-is-write))
(let ((v (api:execute-requests dbstruct-local (vector (symbol->string cmd) params))))
(handle-exceptions ;; there has been a long history of receiving strange errors from values returned by the client when things go wrong..
exn ;; This is an attempt to detect that situation and recover gracefully
(begin
(debug:print 0 *default-log-port* "ERROR: bad data from server " v " message: " ((condition-property-accessor 'exn 'message) exn) ", exn=" exn)
(vector #t '())) ;; should always get a vector but if something goes wrong return a dummy
(if (and (vector? v)
(> (vector-length v) 1))
(let ((newvec (vector (vector-ref v 0)(vector-ref v 1))))
newvec) ;; by copying the vector while inside the error handler we should force the detection of a corrupted record
(vector #t '())))) ;; we could also check that the returned types are valid
(vector #t '())))
(success (vector-ref resdat 0))
(res (vector-ref resdat 1))
(duration (- (current-milliseconds) start)))
(if (and read-only qry-is-write)
(debug:print 0 *default-log-port* "ERROR: attempt to write to read-only database ignored. cmd=" cmd))
(if (not success)
(if (> remretries 0)
(begin
(debug:print-error 0 *default-log-port* "local query failed. Trying again.")
(thread-sleep! (/ (pseudo-random-integer 5000) 1000)) ;; some random delay
(rmt:open-qry-close-locally cmd run-id params remretries: (- remretries 1)))
(begin
(debug:print-error 0 *default-log-port* "too many retries in rmt:open-qry-close-locally, giving up")
#f))
(begin
;; (rmt:update-db-stats run-id cmd params duration)
;; mark this run as dirty if this was a write, the watchdog is responsible for syncing it
(if qry-is-write
(let ((start-time (current-seconds)))
(mutex-lock! *db-multi-sync-mutex*)
(set! *db-last-access* start-time) ;; THIS IS PROBABLY USELESS? (we are on a client)
(mutex-unlock! *db-multi-sync-mutex*)))))
res))
;;======================================================================
;;
;; A C T U A L A P I C A L L S
;;
;;======================================================================
;;======================================================================
;; S E R V E R
;;======================================================================
(define (rmt:kill-server run-id)
(rmt:send-receive 'kill-server run-id (list run-id)))
(define (rmt:start-server run-id)
(rmt:send-receive 'start-server 0 (list run-id)))
;;======================================================================
;; M I S C
;;======================================================================
(define (rmt:login run-id)
(rmt:send-receive 'login run-id (list *toppath* megatest-version *my-client-signature*)))
;; rmt:login-no-auto-client-setup
;; rmt:send-receive-no-auto-client-setup
;; hand off a call to one of the db:queries statements
;; added run-id to make looking up the correct db possible
;;
(define (rmt:general-call stmtname run-id . params)
(rmt:send-receive 'general-call run-id (append (list stmtname run-id) params)))
;; given a hostname, return a pair of cpu load and update time representing latest intelligence from tests running on that host
(define (rmt:get-latest-host-load hostname)
(rmt:send-receive 'get-latest-host-load 0 (list hostname)))
(define (rmt:sdb-qry qry val run-id)
;; add caching if qry is 'getid or 'getstr
(rmt:send-receive 'sdb-qry run-id (list qry val)))
;; NOT COMPLETED
(define (rmt:runtests user run-id testpatt params)
(rmt:send-receive 'runtests run-id testpatt))
(define (rmt:get-run-record-ids target run keynames test-patt)
(rmt:send-receive 'get-run-record-ids #f (list target run keynames test-patt)))
(define (rmt:get-changed-record-ids since-time)
(rmt:send-receive 'get-changed-record-ids #f (list since-time)) )
(define (rmt:drop-all-triggers)
(rmt:send-receive 'drop-all-triggers #f '()))
(define (rmt:create-all-triggers)
(rmt:send-receive 'create-all-triggers #f '()))
;;======================================================================
;; T E S T M E T A
;;======================================================================
(define (rmt:get-tests-tags)
(rmt:send-receive 'get-tests-tags #f '()))
;;======================================================================
;; K E Y S
;;======================================================================
;; These require run-id because the values come from the run!
;;
(define (rmt:get-key-val-pairs run-id)
(rmt:send-receive 'get-key-val-pairs run-id (list run-id)))
(define (rmt:get-keys)
(if *db-keys* *db-keys*
(let ((res (rmt:send-receive 'get-keys #f '())))
(set! *db-keys* res)
res)))
(define (rmt:get-keys-write) ;; dummy query to force server start
(let ((res (rmt:send-receive 'get-keys-write #f '())))
(set! *db-keys* res)
res))
;; we don't reuse run-id's (except possibly *after* a db cleanup) so it is safe
;; to cache the resuls in a hash
;;
(define (rmt:get-key-vals run-id)
(or (hash-table-ref/default *keyvals* run-id #f)
(let ((res (rmt:send-receive 'get-key-vals #f (list run-id))))
(hash-table-set! *keyvals* run-id res)
res)))
(define (rmt:get-targets)
(rmt:send-receive 'get-targets #f '()))
(define (rmt:get-target run-id)
(rmt:send-receive 'get-target run-id (list run-id)))
(define (rmt:get-run-times runpatt targetpatt)
(rmt:send-receive 'get-run-times #f (list runpatt targetpatt )))
;;======================================================================
;; T E S T S
;;======================================================================
;; Just some syntatic sugar
(define (rmt:register-test run-id test-name item-path)
(rmt:general-call 'register-test run-id run-id test-name item-path))
(define (rmt:get-test-id run-id testname item-path)
(rmt:send-receive 'get-test-id run-id (list run-id testname item-path)))
;; run-id is NOT used
;;
(define (rmt:get-test-info-by-id run-id test-id)
(if (number? test-id)
(rmt:send-receive 'get-test-info-by-id run-id (list run-id test-id))
(begin
(debug:print 0 *default-log-port* "WARNING: Bad data handed to rmt:get-test-info-by-id run-id=" run-id ", test-id=" test-id)
(print-call-chain (current-error-port))
#f)))
(define (rmt:test-get-rundir-from-test-id run-id test-id)
(rmt:send-receive 'test-get-rundir-from-test-id run-id (list run-id test-id)))
;; (define (rmt:open-test-db-by-test-id run-id test-id #!key (work-area #f))
;; (let* ((test-path (if (string? work-area)
;; work-area
;; (rmt:test-get-rundir-from-test-id run-id test-id))))
;; (debug:print 3 *default-log-port* "TEST PATH: " test-path)
;; (open-test-db test-path)))
;; WARNING: This currently bypasses the transaction wrapped writes system
(define (rmt:test-set-state-status-by-id run-id test-id newstate newstatus newcomment)
(rmt:send-receive 'test-set-state-status-by-id run-id (list run-id test-id newstate newstatus newcomment)))
(define (rmt:set-tests-state-status run-id testnames currstate currstatus newstate newstatus)
(rmt:send-receive 'set-tests-state-status run-id (list run-id testnames currstate currstatus newstate newstatus)))
(define (rmt:get-tests-for-run run-id testpatt states statuses offset limit not-in sort-by sort-order qryvals last-update mode)
;; (if (number? run-id)
(rmt:send-receive 'get-tests-for-run run-id (list run-id testpatt states statuses offset limit not-in sort-by sort-order qryvals last-update mode)))
;; (begin
;; (debug:print-error 0 *default-log-port* "rmt:get-tests-for-run called with bad run-id=" run-id)
;; (print-call-chain (current-error-port))
;; '())))
(define (rmt:get-tests-for-run-state-status run-id testpatt last-update)
(rmt:send-receive 'get-tests-for-run-state-status run-id (list run-id testpatt last-update)))
;; get stuff via synchash
(define (rmt:synchash-get run-id proc synckey keynum params)
(rmt:send-receive 'synchash-get run-id (list run-id proc synckey keynum params)))
(define (rmt:get-tests-for-run-mindata run-id testpatt states status not-in)
(rmt:send-receive 'get-tests-for-run-mindata run-id (list run-id testpatt states status not-in)))
;; IDEA: Threadify these - they spend a lot of time waiting ...
;;
(define (rmt:get-tests-for-runs-mindata run-ids testpatt states status not-in)
(let ((multi-run-mutex (make-mutex))
(run-id-list (if run-ids
run-ids
(rmt:get-all-run-ids)))
(result '()))
(if (null? run-id-list)
'()
(let loop ((hed (car run-id-list))
(tal (cdr run-id-list))
(threads '()))
(if (> (length threads) 5)
(loop hed tal (filter (lambda (th)(not (member (thread-state th) '(terminated dead)))) threads))
(let* ((newthread (make-thread
(lambda ()
(let ((res (rmt:send-receive 'get-tests-for-run-mindata hed (list hed testpatt states status not-in))))
(if (list? res)
(begin
(mutex-lock! multi-run-mutex)
(set! result (append result res))
(mutex-unlock! multi-run-mutex))
(debug:print-error 0 *default-log-port* "get-tests-for-run-mindata failed for run-id " hed ", testpatt " testpatt ", states " states ", status " status ", not-in " not-in))))
(conc "multi-run-thread for run-id " hed)))
(newthreads (cons newthread threads)))
(thread-start! newthread)
(thread-sleep! 0.054) ;; give that thread some time to start
(if (null? tal)
newthreads
(loop (car tal)(cdr tal) newthreads))))))
result))
;; ;; IDEA: Threadify these - they spend a lot of time waiting ...
;; ;;
;; (define (rmt:get-tests-for-runs-mindata run-ids testpatt states status not-in)
;; (let ((run-id-list (if run-ids
;; run-ids
;; (rmt:get-all-run-ids))))
;; (apply append (map (lambda (run-id)
;; (rmt:send-receive 'get-tests-for-run-mindata run-id (list run-ids testpatt states status not-in)))
;; run-id-list))))
(define (rmt:delete-test-records run-id test-id)
(rmt:send-receive 'delete-test-records run-id (list run-id test-id)))
(define (rmt:test-set-state-status run-id test-id state status msg)
(rmt:send-receive 'test-set-state-status run-id (list run-id test-id state status msg)))
(define (rmt:test-toplevel-num-items run-id test-name)
(rmt:send-receive 'test-toplevel-num-items run-id (list run-id test-name)))
;; (define (rmt:get-previous-test-run-record run-id test-name item-path)
;; (rmt:send-receive 'get-previous-test-run-record run-id (list run-id test-name item-path)))
(define (rmt:get-matching-previous-test-run-records run-id test-name item-path)
(rmt:send-receive 'get-matching-previous-test-run-records run-id (list run-id test-name item-path)))
(define (rmt:test-get-logfile-info run-id test-name)
(rmt:send-receive 'test-get-logfile-info run-id (list run-id test-name)))
(define (rmt:test-get-records-for-index-file run-id test-name)
(rmt:send-receive 'test-get-records-for-index-file run-id (list run-id test-name)))
(define (rmt:get-testinfo-state-status run-id test-id)
(rmt:send-receive 'get-testinfo-state-status run-id (list run-id test-id)))
(define (rmt:test-set-log! run-id test-id logf)
(if (string? logf)(rmt:general-call 'test-set-log run-id logf test-id)))
(define (rmt:test-set-top-process-pid run-id test-id pid)
(rmt:send-receive 'test-set-top-process-pid run-id (list run-id test-id pid)))
(define (rmt:test-get-top-process-pid run-id test-id)
(rmt:send-receive 'test-get-top-process-pid run-id (list run-id test-id)))
(define (rmt:get-run-ids-matching-target keynames target res runname testpatt statepatt statuspatt)
(rmt:send-receive 'get-run-ids-matching-target #f (list keynames target res runname testpatt statepatt statuspatt)))
;; NOTE: This will open and access ALL run databases.
;;
(define (rmt:test-get-paths-matching-keynames-target-new keynames target res testpatt statepatt statuspatt runname)
(let ((run-ids (rmt:get-run-ids-matching-target keynames target res runname testpatt statepatt statuspatt)))
(apply append
(map (lambda (run-id)
(rmt:send-receive 'test-get-paths-matching-keynames-target-new run-id (list run-id keynames target res testpatt statepatt statuspatt runname)))
run-ids))))
(define (rmt:get-prereqs-not-met run-id waitons ref-test-name ref-item-path #!key (mode '(normal))(itemmaps #f))
(rmt:send-receive 'get-prereqs-not-met run-id (list run-id waitons ref-test-name ref-item-path mode itemmaps)))
(define (rmt:get-count-tests-running-for-run-id run-id)
(rmt:send-receive 'get-count-tests-running-for-run-id run-id (list run-id)))
(define (rmt:get-not-completed-cnt run-id)
(rmt:send-receive 'get-not-completed-cnt run-id (list run-id)))
;; Statistical queries
(define (rmt:get-count-tests-running run-id)
(rmt:send-receive 'get-count-tests-running run-id (list run-id)))
(define (rmt:get-count-tests-running-for-testname run-id testname)
(rmt:send-receive 'get-count-tests-running-for-testname run-id (list run-id testname)))
(define (rmt:get-count-tests-running-in-jobgroup run-id jobgroup)
(rmt:send-receive 'get-count-tests-running-in-jobgroup run-id (list run-id jobgroup)))
;; state and status are extra hints not usually used in the calculation
;;
(define (rmt:set-state-status-and-roll-up-items run-id test-name item-path state status comment)
(rmt:send-receive 'set-state-status-and-roll-up-items run-id (list run-id test-name item-path state status comment)))
(define (rmt:set-state-status-and-roll-up-run run-id state status)
(rmt:send-receive 'set-state-status-and-roll-up-run run-id (list run-id state status)))
(define (rmt:update-pass-fail-counts run-id test-name)
(rmt:general-call 'update-pass-fail-counts run-id test-name test-name test-name))
(define (rmt:top-test-set-per-pf-counts run-id test-name)
(rmt:send-receive 'top-test-set-per-pf-counts run-id (list run-id test-name)))
(define (rmt:get-raw-run-stats run-id)
(rmt:send-receive 'get-raw-run-stats run-id (list run-id)))
(define (rmt:get-test-times runname target)
(rmt:send-receive 'get-test-times #f (list runname target )))
;;======================================================================
;; R U N S
;;======================================================================
(define (rmt:get-run-info run-id)
(rmt:send-receive 'get-run-info run-id (list run-id)))
(define (rmt:get-num-runs runpatt)
(rmt:send-receive 'get-num-runs #f (list runpatt)))
(define (rmt:get-runs-cnt-by-patt runpatt targetpatt keys)
(rmt:send-receive 'get-runs-cnt-by-patt #f (list runpatt targetpatt keys)))
;; Use the special run-id == #f scenario here since there is no run yet
(define (rmt:register-run keyvals runname state status user contour)
(rmt:send-receive 'register-run #f (list keyvals runname state status user contour)))
(define (rmt:get-run-name-from-id run-id)
(rmt:send-receive 'get-run-name-from-id run-id (list run-id)))
(define (rmt:delete-run run-id)
(rmt:send-receive 'delete-run run-id (list run-id)))
(define (rmt:update-run-stats run-id stats)
(rmt:send-receive 'update-run-stats #f (list run-id stats)))
(define (rmt:delete-old-deleted-test-records)
(rmt:send-receive 'delete-old-deleted-test-records #f '()))
(define (rmt:get-runs runpatt count offset keypatts)
(rmt:send-receive 'get-runs #f (list runpatt count offset keypatts)))
(define (rmt:simple-get-runs runpatt count offset target last-update)
(rmt:send-receive 'simple-get-runs #f (list runpatt count offset target last-update)))
(define (rmt:get-all-run-ids)
(rmt:send-receive 'get-all-run-ids #f '()))
(define (rmt:get-prev-run-ids run-id)
(rmt:send-receive 'get-prev-run-ids #f (list run-id)))
(define (rmt:lock/unlock-run run-id lock unlock user)
(rmt:send-receive 'lock/unlock-run #f (list run-id lock unlock user)))
;; set/get status
(define (rmt:get-run-status run-id)
(rmt:send-receive 'get-run-status #f (list run-id)))
(define (rmt:get-run-state run-id)
(rmt:send-receive 'get-run-state #f (list run-id)))
(define (rmt:set-run-status run-id run-status #!key (msg #f))
(rmt:send-receive 'set-run-status #f (list run-id run-status msg)))
(define (rmt:set-run-state-status run-id state status )
(rmt:send-receive 'set-run-state-status #f (list run-id state status)))
(define (rmt:update-tesdata-on-repilcate-db old-lt new-lt)
(rmt:send-receive 'update-tesdata-on-repilcate-db #f (list old-lt new-lt)))
(define (rmt:update-run-event_time run-id)
(rmt:send-receive 'update-run-event_time #f (list run-id)))
(define (rmt:get-runs-by-patt keys runnamepatt targpatt offset limit fields last-runs-update #!key (sort-order "asc")) ;; fields of #f uses default
(rmt:send-receive 'get-runs-by-patt #f (list keys runnamepatt targpatt offset limit fields last-runs-update sort-order)))
(define (rmt:find-and-mark-incomplete run-id ovr-deadtime)
;; (if (rmt:send-receive 'have-incompletes? run-id (list run-id ovr-deadtime))
(rmt:send-receive 'mark-incomplete run-id (list run-id ovr-deadtime))
) ;; )
(define (rmt:get-main-run-stats run-id)
(rmt:send-receive 'get-main-run-stats #f (list run-id)))
(define (rmt:get-var varname)
(rmt:send-receive 'get-var #f (list varname)))
(define (rmt:del-var varname)
(rmt:send-receive 'del-var #f (list varname)))
(define (rmt:set-var varname value)
(rmt:send-receive 'set-var #f (list varname value)))
(define (rmt:inc-var varname)
(rmt:send-receive 'inc-var #f (list varname)))
(define (rmt:dec-var varname)
(rmt:send-receive 'dec-var #f (list varname)))
(define (rmt:add-var varname value)
(rmt:send-receive 'add-var #f (list varname value)))
;;======================================================================
;; M U L T I R U N Q U E R I E S
;;======================================================================
;; Need to move this to multi-run section and make associated changes
(define (rmt:find-and-mark-incomplete-all-runs #!key (ovr-deadtime #f))
(let ((run-ids (rmt:get-all-run-ids)))
(for-each (lambda (run-id)
(rmt:find-and-mark-incomplete run-id ovr-deadtime))
run-ids)))
;; get the previous record for when this test was run where all keys match but runname
;; returns #f if no such test found, returns a single test record if found
;;
;; Run this at the client end since we have to connect to multiple run-id dbs
;;
(define (rmt:get-previous-test-run-record run-id test-name item-path)
(let* ((keyvals (rmt:get-key-val-pairs run-id))
(keys (rmt:get-keys))
(selstr (string-intersperse keys ","))
(qrystr (string-intersperse (map (lambda (x)(conc x "=?")) keys) " AND ")))
(if (not keyvals)
#f
(let ((prev-run-ids (rmt:get-prev-run-ids run-id)))
;; for each run starting with the most recent look to see if there is a matching test
;; if found then return that matching test record
(debug:print 4 *default-log-port* "selstr: " selstr ", qrystr: " qrystr ", keyvals: " keyvals ", previous run ids found: " prev-run-ids)
(if (null? prev-run-ids) #f
(let loop ((hed (car prev-run-ids))
(tal (cdr prev-run-ids)))
(let ((results (rmt:get-tests-for-run hed (conc test-name "/" item-path) '() '() ;; run-id testpatt states statuses
#f #f #f ;; offset limit not-in hide/not-hide
#f #f #f #f 'normal))) ;; sort-by sort-order qryvals last-update mode
(debug:print 4 *default-log-port* "Got tests for run-id " run-id ", test-name " test-name ", item-path " item-path ": " results)
(if (and (null? results)
(not (null? tal)))
(loop (car tal)(cdr tal))
(if (null? results) #f
(car results))))))))))
(define (rmt:get-run-stats)
(rmt:send-receive 'get-run-stats #f '()))
;;======================================================================
;; S T E P S
;;======================================================================
;; Getting steps is more complicated.
;;
;; If given work area
;; 1. Find the testdat.db file
;; 2. Open the testdat.db file and do the query
;; If not given the work area
;; 1. Do a remote call to get the test path
;; 2. Continue as above
;;
;;(define (rmt:get-steps-for-test run-id test-id)
;; (rmt:send-receive 'get-steps-data run-id (list test-id)))
(define (rmt:teststep-set-status! run-id test-id teststep-name state-in status-in comment logfile)
(let* ((state (items:check-valid-items "state" state-in))
(status (items:check-valid-items "status" status-in)))
(if (or (not state)(not status))
(debug:print 3 *default-log-port* "WARNING: Invalid " (if status "status" "state")
" value \"" (if status state-in status-in) "\", update your validvalues section in megatest.config"))
(rmt:send-receive 'teststep-set-status! run-id (list run-id test-id teststep-name state-in status-in comment logfile))))
(define (rmt:delete-steps-for-test! run-id test-id)
(rmt:send-receive 'delete-steps-for-test! run-id (list run-id test-id)))
(define (rmt:get-steps-for-test run-id test-id)
(rmt:send-receive 'get-steps-for-test run-id (list run-id test-id)))
(define (rmt:get-steps-info-by-id test-step-id)
(rmt:send-receive 'get-steps-info-by-id #f (list test-step-id)))
;;======================================================================
;; T E S T D A T A
;;======================================================================
(define (rmt:read-test-data run-id test-id categorypatt #!key (work-area #f))
(rmt:send-receive 'read-test-data run-id (list run-id test-id categorypatt)))
(define (rmt:read-test-data-varpatt run-id test-id categorypatt varpatt #!key (work-area #f))
(rmt:send-receive 'read-test-data-varpatt run-id (list run-id test-id categorypatt varpatt)))
(define (rmt:get-data-info-by-id test-data-id)
(rmt:send-receive 'get-data-info-by-id #f (list test-data-id)))
(define (rmt:testmeta-add-record testname)
(rmt:send-receive 'testmeta-add-record #f (list testname)))
(define (rmt:testmeta-get-record testname)
(rmt:send-receive 'testmeta-get-record #f (list testname)))
(define (rmt:testmeta-update-field test-name fld val)
(rmt:send-receive 'testmeta-update-field #f (list test-name fld val)))
(define (rmt:test-data-rollup run-id test-id status)
(rmt:send-receive 'test-data-rollup run-id (list run-id test-id status)))
(define (rmt:csv->test-data run-id test-id csvdata)
(rmt:send-receive 'csv->test-data run-id (list run-id test-id csvdata)))
;;======================================================================
;; T A S K S
;;======================================================================
(define (rmt:tasks-find-task-queue-records target run-name test-patt state-patt action-patt)
(rmt:send-receive 'find-task-queue-records #f (list target run-name test-patt state-patt action-patt)))
(define (rmt:tasks-add action owner target runname testpatt params)
(rmt:send-receive 'tasks-add #f (list action owner target runname testpatt params)))
(define (rmt:tasks-set-state-given-param-key param-key new-state)
(rmt:send-receive 'tasks-set-state-given-param-key #f (list param-key new-state)))
(define (rmt:tasks-get-last target runname)
(rmt:send-receive 'tasks-get-last #f (list target runname)))
;;======================================================================
;; N O S Y N C D B
;;======================================================================
(define (rmt:no-sync-set var val)
(rmt:send-receive 'no-sync-set #f `(,var ,val)))
(define (rmt:no-sync-get/default var default)
(rmt:send-receive 'no-sync-get/default #f `(,var ,default)))
(define (rmt:no-sync-del! var)
(rmt:send-receive 'no-sync-del! #f `(,var)))
(define (rmt:no-sync-get-lock keyname)
(rmt:send-receive 'no-sync-get-lock #f `(,keyname)))
;;======================================================================
;; A R C H I V E S
;;======================================================================
(define (rmt:archive-get-allocations testname itempath dneeded)
(rmt:send-receive 'archive-get-allocations #f (list testname itempath dneeded)))
(define (rmt:archive-register-block-name bdisk-id archive-path)
(rmt:send-receive 'archive-register-block-name #f (list bdisk-id archive-path)))
(define (rmt:archive-allocate-testsuite/area-to-block block-id testsuite-name areakey)
(rmt:send-receive 'archive-allocate-test-to-block #f (list block-id testsuite-name areakey)))
(define (rmt:archive-register-disk bdisk-name bdisk-path df)
(rmt:send-receive 'archive-register-disk #f (list bdisk-name bdisk-path df)))
(define (rmt:test-set-archive-block-id run-id test-id archive-block-id)
(rmt:send-receive 'test-set-archive-block-id run-id (list run-id test-id archive-block-id)))
(define (rmt:test-get-archive-block-info archive-block-id)
(rmt:send-receive 'test-get-archive-block-info #f (list archive-block-id)))
(define (rmtmod:calc-ro-mode runremote *toppath*)
(if (and runremote
(remote-ro-mode-checked runremote))
(remote-ro-mode runremote)
(let* ((dbfile (conc *toppath* "/megatest.db"))
(ro-mode (not (file-writable? dbfile)))) ;; TODO: use dbstruct or runremote to figure this out in future
(if runremote
(begin
(remote-ro-mode-set! runremote ro-mode)
(remote-ro-mode-checked-set! runremote #t)
ro-mode)
ro-mode))))
(define (extras-readonly-mode rmt-mutex log-port cmd params)
(mutex-unlock! rmt-mutex)
(debug:print-info 12 log-port "rmt:send-receive, case 3")
(debug:print 0 log-port "WARNING: write transaction requested on a readonly area. cmd="cmd" params="params)
#f)
(define (extras-transport-failed *default-log-port* *rmt-mutex* attemptnum runremote cmd rid params)
(debug:print 0 *default-log-port* "WARNING: communication failed. Trying again, try num: " attemptnum)
(mutex-lock! *rmt-mutex*)
(remote-conndat-set! runremote #f)
(http-transport:close-connections area-dat: runremote)
(remote-server-url-set! runremote #f)
(mutex-unlock! *rmt-mutex*)
(debug:print-info 12 *default-log-port* "rmt:send-receive, case 9.1")
(rmt:send-receive cmd rid params attemptnum: (+ attemptnum 1)))
(define (extras-transport-succeded *default-log-port* *rmt-mutex* attemptnum runremote res params rid cmd)
(if (and (vector? res)
(eq? (vector-length res) 2)
(eq? (vector-ref res 1) 'overloaded)) ;; since we are
;; looking at the
;; data to carry the
;; error we'll use a
;; fairly obtuse
;; combo to minimise
;; the chances of
;; some sort of
;; collision. this
;; is the case where
;; the returned data
;; is bad or the
;; server is
;; overloaded and we
;; want to ease off
;; the queries
(let ((wait-delay (+ attemptnum (* attemptnum 10))))
(debug:print 0 *default-log-port* "WARNING: server is overloaded. Delaying " wait-delay " seconds and trying call again.")
(mutex-lock! *rmt-mutex*)
(http-transport:close-connections area-dat: runremote)
(set! *runremote* #f) ;; force starting over
(mutex-unlock! *rmt-mutex*)
(thread-sleep! wait-delay)
(rmt:send-receive cmd rid params attemptnum: (+ attemptnum 1)))
res)) ;; All good, return res
#;(set-functions rmt:send-receive remote-server-url-set!
http-transport:close-connections remote-conndat-set!
debug:print debug:print-info
remote-ro-mode remote-ro-mode-set!
remote-ro-mode-checked-set! remote-ro-mode-checked)
;; gets mtpg-run-id and syncs the record if different
;;
(define (tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time)
(let* ((runs-ht (hash-table-ref cached-info 'runs))
(runinf (hash-table-ref/default runs-ht run-id #f))
(area-id (vector-ref area-info 0)))
(if runinf
runinf ;; already cached
(let* ((run-dat (rmt:get-run-info run-id)) ;; NOTE: get-run-info returns a vector < row header >
(run-name (rmt:get-run-name-from-id run-id))
(row (db:get-rows run-dat)) ;; yes, this returns a single row
(header (db:get-header run-dat))
(state (db:get-value-by-header row header "state"))
(status (db:get-value-by-header row header "status"))
(owner (db:get-value-by-header row header "owner"))
(event-time (db:get-value-by-header row header "event_time"))
(comment (db:get-value-by-header row header "comment"))
(fail-count (db:get-value-by-header row header "fail_count"))
(pass-count (db:get-value-by-header row header "pass_count"))
(db-contour (db:get-value-by-header row header "contour"))
(contour (if (args:get-arg "-prepend-contour")
(if (and db-contour (not (equal? db-contour "")) (string? db-contour ))
(begin
(debug:print-info 10 *default-log-port* "db-contour" db-contour)
db-contour)
(args:get-arg "-contour"))))
(run-tag (if (args:get-arg "-run-tag")
(args:get-arg "-run-tag")
""))
(last-update (db:get-value-by-header row header "last_update"))
(keytarg (if (or (args:get-arg "-prepend-contour") (args:get-arg "-prefix-target"))
(conc "MT_CONTOUR/MT_AREA/" (string-intersperse (rmt:get-keys) "/")) (string-intersperse (rmt:get-keys) "/"))) ;; e.g. version/iteration/platform
(base-target (rmt:get-target run-id))
(target (if (or (args:get-arg "-prepend-contour") (args:get-arg "-prefix-target"))
(conc (or (args:get-arg "-prefix-target") (conc contour "/" (common:get-area-name) "/")) base-target) base-target)) ;; e.g. v1.63/a3e1/ubuntu
(spec-id (pgdb:get-ttype dbh keytarg))
(publish-time (if (args:get-arg "-cp-eventtime-to-publishtime")
event-time
(current-seconds)))
(new-run-id (if (and run-name base-target) (pgdb:get-run-id dbh spec-id target run-name area-id) #f)))
(if new-run-id
(begin ;; let ((run-record (pgdb:get-run-info dbh new-run-id))
(hash-table-set! runs-ht run-id new-run-id)
;; ensure key fields are up to date
;; if last_update == pgdb_last_update do not update smallest-last-update-time
(let* ((pgdb-last-update (pgdb:get-run-last-update dbh new-run-id))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time)))
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:refresh-run-info
dbh
new-run-id
state status owner event-time comment fail-count pass-count area-id last-update publish-time)
(debug:print-info 4 *default-log-port* "Working on run-id " run-id " pgdb-id " new-run-id )
(if (not (equal? run-tag ""))
(task:add-run-tag dbh new-run-id run-tag))
new-run-id)
(if (or (not state) (equal? state "deleted"))
(begin
(debug:print-info 1 *default-log-port* "Warning: Run with id " run-id " was created after previous sync and deleted before the sync") #f)
(if (handle-exceptions
exn
(begin (print-call-chain)
(print ((condition-property-accessor 'exn 'message) exn))
#f)
(pgdb:insert-run
dbh
spec-id target run-name state status owner event-time comment fail-count pass-count area-id last-update publish-time))
(let* ((smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f)))
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time))
#f)))))))
(define (tasks:sync-test-gen-data dbh cached-info test-data-ids smallest-last-update-time)
(let ((test-ht (hash-table-ref cached-info 'tests))
(data-ht (hash-table-ref cached-info 'data)))
(for-each
(lambda (test-data-id)
(let* ((test-data-info (rmt:get-data-info-by-id test-data-id))
(data-id (db:test-data-get-id test-data-info))
(test-id (db:test-data-get-test_id test-data-info))
(category (db:test-data-get-category test-data-info))
(variable (db:test-data-get-variable test-data-info))
(value (db:test-data-get-value test-data-info))
(expected (db:test-data-get-expected test-data-info))
(tol (db:test-data-get-tol test-data-info))
(units (db:test-data-get-units test-data-info))
(comment (db:test-data-get-comment test-data-info))
(status (db:test-data-get-status test-data-info))
(type (db:test-data-get-type test-data-info))
(last-update (db:test-data-get-last_update test-data-info))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f))
(pgdb-test-id (hash-table-ref/default test-ht test-id #f))
(pgdb-data-id (if pgdb-test-id
(pgdb:get-test-data-id dbh pgdb-test-id category variable)
#f)))
(if data-id
(begin
(if pgdb-test-id
(begin
(if pgdb-data-id
(begin
(debug:print-info 4 *default-log-port* "Updating existing test-data with test-id: " test-id " and data-id " data-id " pgdb test id: " pgdb-test-id " pgdb data id " pgdb-data-id)
(let* ((pgdb-last-update (pgdb:get-test-data-last-update dbh pgdb-data-id)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time)))
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:update-test-data dbh pgdb-data-id pgdb-test-id category variable value expected tol units comment status type last-update))
(begin
(debug:print-info 4 *default-log-port* "Inserting test-data with test-id: " test-id " and data-id " data-id " pgdb test id: " pgdb-test-id)
(if (handle-exceptions
exn
(begin (print-call-chain)
(print ((condition-property-accessor 'exn 'message) exn))
#f)
(pgdb:insert-test-data dbh pgdb-test-id category variable value expected tol units comment status type last-update))
;(tasks:run-id->mtpg-run-id dbh cached-info run-id area-info)
(begin
;(pgdb:insert-test-data dbh pgdb-test-id category variable value expected tol units comment status type )
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(set! pgdb-data-id (pgdb:get-test-data-id dbh pgdb-test-id category variable)))
#f)))
(hash-table-set! data-ht data-id pgdb-data-id ))
(begin
(debug:print-info 1 *default-log-port* "Error: Test not in pgdb"))))
(debug:print-info 1 *default-log-port* "Error: Could not get test data info for data id " test-data-id )))) ;; this is a wierd senario need to debug
test-data-ids)))
(define (task:get-test-times)
(let* ((runname (if (args:get-arg "-runname")
(args:get-arg "-runname")
#f))
(target (if (args:get-arg "-target")
(args:get-arg "-target")
#f))
(test-times (rmt:get-test-times runname target )))
(if (not runname)
(begin
(print "Error: Missing argument -runname")
(exit)))
(if (string-contains runname "%")
(begin
(print "Error: Invalid runname, '%' not allowed (" runname ") ")
(exit)))
(if (not target)
(begin
(print "Error: Missing argument -target")
(exit)))
(if (string-contains target "%")
(begin
(print "Error: Invalid target, '%' not allowed (" target ") ")
(exit)))
(if (eq? (length test-times) 0)
(begin
(print "Data not found!!")
(exit)))
(if (equal? (args:get-arg "-dumpmode") "json")
(task:print-testtime-as-json test-times)
(if (equal? (args:get-arg "-dumpmode") "csv")
(task:print-testtime test-times ",")
(task:print-testtime test-times " ")))))
(define (tasks:sync-test-steps dbh cached-info test-step-ids smallest-last-update-time)
; (print "Sync Steps " test-step-ids )
(let ((test-ht (hash-table-ref cached-info 'tests))
(step-ht (hash-table-ref cached-info 'steps)))
(for-each
(lambda (test-step-id)
(let* ((test-step-info (rmt:get-steps-info-by-id test-step-id))
(step-id (tdb:step-get-id test-step-info))
(test-id (tdb:step-get-test_id test-step-info))
(stepname (tdb:step-get-stepname test-step-info))
(state (tdb:step-get-state test-step-info))
(status (tdb:step-get-status test-step-info))
(event_time (tdb:step-get-event_time test-step-info))
(comment (tdb:step-get-comment test-step-info))
(logfile (tdb:step-get-logfile test-step-info))
(last-update (tdb:step-get-last_update test-step-info))
(pgdb-test-id (hash-table-ref/default test-ht test-id #f))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f))
(pgdb-step-id (if pgdb-test-id
(pgdb:get-test-step-id dbh pgdb-test-id stepname state)
#f)))
(if step-id
(begin
(if pgdb-test-id
(begin
(if pgdb-step-id
(begin
(debug:print-info 4 *default-log-port* "Updating existing test-step with test-id: " test-id " and step-id " step-id " pgdb test id: " pgdb-test-id " pgdb step id " pgdb-step-id )
(let* ((pgdb-last-update (pgdb:get-test-step-last-update dbh pgdb-step-id)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time)))
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:update-test-step dbh pgdb-step-id pgdb-test-id stepname state status event_time comment logfile last-update))
(begin
(debug:print-info 4 *default-log-port* "Inserting test-step with test-id: " test-id " and step-id " step-id " pgdb test id: " pgdb-test-id)
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(pgdb:insert-test-step dbh pgdb-test-id stepname state status event_time comment logfile last-update )
(set! pgdb-step-id (pgdb:get-test-step-id dbh pgdb-test-id stepname state))))
(hash-table-set! step-ht step-id pgdb-step-id ))
(debug:print-info 1 *default-log-port* "Error: Test not cashed")))
(debug:print-info 1 *default-log-port* "Error: Could not get test step info for step id " test-step-id )))) ;; this is a wierd senario need to debug
test-step-ids)))
(define (tasks:sync-tests-data dbh cached-info test-ids area-info smallest-last-update-time)
(let ((test-ht (hash-table-ref cached-info 'tests)))
(for-each
(lambda (test-id)
; (print test-id)
(let* ((test-info (rmt:get-test-info-by-id #f test-id))
(run-id (db:test-get-run_id test-info)) ;; look these up in db_records.scm
(test-id (db:test-get-id test-info))
(test-name (db:test-get-testname test-info))
(item-path (db:test-get-item-path test-info))
(state (db:test-get-state test-info))
(status (db:test-get-status test-info))
(host (db:test-get-host test-info))
(pid (db:test-get-process_id test-info))
(cpuload (db:test-get-cpuload test-info))
(diskfree (db:test-get-diskfree test-info))
(uname (db:test-get-uname test-info))
(run-dir (db:test-get-rundir test-info))
(log-file (db:test-get-final_logf test-info))
(run-duration (db:test-get-run_duration test-info))
(comment (db:test-get-comment test-info))
(event-time (db:test-get-event_time test-info))
(archived (db:test-get-archived test-info))
(last-update (db:test-get-last_update test-info))
(pgdb-run-id (tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f))
(pgdb-test-id (if pgdb-run-id
(begin
;(print pgdb-run-id)
(pgdb:get-test-id dbh pgdb-run-id test-name item-path))
#f)))
;; "id" "run_id" "testname" "state" "status" "event_time"
;; "host" "cpuload" "diskfree" "uname" "rundir" "item_path"
;; "run_duration" "final_logf" "comment" "shortdir" "attemptnum" "archived"
(if (or (not item-path) (string-null? item-path))
(debug:print-info 0 *default-log-port* "Working on Run id : " run-id "and test name : " test-name))
(if pgdb-run-id
(begin
(if pgdb-test-id ;; have a record
(begin ;; let ((key-name (conc run-id "/" test-name "/" item-path)))
(debug:print-info 4 *default-log-port* "Updating existing test with run-id: " run-id " and test-id: " test-id " pgdb run id: " pgdb-run-id " pgdb-test-id " pgdb-test-id)
(let* ((pgdb-last-update (pgdb:get-test-last-update dbh pgdb-test-id)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time))) ;;if last-update is same as pgdb-last-update then it is safe to assume the records are identical and we can use a larger last update time.
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:update-test dbh pgdb-test-id pgdb-run-id test-name item-path state status host cpuload diskfree uname run-dir log-file run-duration comment event-time archived last-update pid))
(begin
(debug:print-info 4 *default-log-port* "Inserting test with run-id: " run-id " and test-id: " test-id " pgdb run id: " pgdb-run-id)
(pgdb:insert-test dbh pgdb-run-id test-name item-path state status host cpuload diskfree uname run-dir log-file run-duration comment event-time archived last-update pid)
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(set! pgdb-test-id (pgdb:get-test-id dbh pgdb-run-id test-name item-path))))
(hash-table-set! test-ht test-id pgdb-test-id))
(debug:print-info 1 *default-log-port* "WARNING: Skipping run with run-id:" run-id ". This run was created after privious sync and removed before this sync."))))
test-ids)))
;; get runs changed since last sync
;; (define (tasks:sync-test-data dbh cached-info area-info)
;; (let* ((
(define (tasks:sync-to-postgres configdat dest)
(print "In sync")
(let* ((dbh (pgdb:open configdat dbname: dest))
(area-info (pgdb:get-area-by-path dbh *toppath*))
(cached-info (make-hash-table))
(start (current-seconds))
(test-patt (if (args:get-arg "-testpatt")
(args:get-arg "-testpatt")
"%"))
(target (if (args:get-arg "-target")
(args:get-arg "-target")
#f))
(run-name (if (args:get-arg "-runname")
(args:get-arg "-runname")
#f)))
(if (and target (not run-name))
(begin
(print "Error: Provide runname")
(exit 1)))
(if (and (not target) run-name)
(begin
(print "Error: Provide target")
(exit 1)))
;(print "123")
;(exit 1)
(for-each (lambda (dtype)
(hash-table-set! cached-info dtype (make-hash-table)))
'(runs targets tests steps data))
(hash-table-set! cached-info 'start start) ;; when done we'll set sync times to this
(if area-info
(let* ((last-sync-time (vector-ref area-info 3))
(smallest-last-update-time (make-hash-table))
(changed (if (and target run-name)
(rmt:get-run-record-ids target run-name (rmt:get-keys) test-patt)
(rmt:get-changed-record-ids last-sync-time)))
(run-ids (alist-ref 'runs changed))
(test-ids (alist-ref 'tests changed))
(test-step-ids (alist-ref 'test_steps changed))
(test-data-ids (alist-ref 'test_data changed))
(run-stat-ids (alist-ref 'run_stats changed))
(area-tag (if (args:get-arg "-area-tag")
(args:get-arg "-area-tag")
(if (args:get-arg "-area")
(args:get-arg "-area")
""))))
(if (and (equal? area-tag "") (not (pgdb:is-area-taged dbh (vector-ref area-info 0))))
(set! area-tag *default-area-tag*))
(if (not (equal? area-tag ""))
(task:add-area-tag dbh area-info area-tag))
(if (or (not (null? test-ids)) (not (null? run-ids)))
(begin
(debug:print-info 0 *default-log-port* "syncing runs")
(tasks:sync-run-data dbh cached-info run-ids area-info smallest-last-update-time)
(debug:print-info 0 *default-log-port* "syncing tests")
(tasks:sync-tests-data dbh cached-info test-ids area-info smallest-last-update-time)
(debug:print-info 0 *default-log-port* "syncing test steps")
(tasks:sync-test-steps dbh cached-info test-step-ids smallest-last-update-time)
(debug:print-info 0 *default-log-port* "syncing test data")
(tasks:sync-test-gen-data dbh cached-info test-data-ids smallest-last-update-time)
(print "----------done---------------")))
(let* ((smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" (current-seconds))))
(debug:print-info 0 "smallest-time :" smallest-time " last-sync-time " last-sync-time)
(if (not (and target run-name))
(if (or (and smallest-time (> smallest-time last-sync-time)) (and smallest-time (eq? last-sync-time 0)))
(pgdb:write-sync-time dbh area-info smallest-time))))) ;;this needs to be changed
(if (tasks:set-area dbh configdat)
(tasks:sync-to-postgres configdat dest)
(begin
(debug:print 0 *default-log-port* "ERROR: unable to create an area record")
#f)))))
(define (tasks:sync-run-data dbh cached-info run-ids area-info smallest-last-update-time)
(for-each
(lambda (run-id)
(debug:print-info 4 *default-log-port* "Check if run with " run-id " needs to be synced" )
(tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time))
run-ids))
;;======================================================================
;; simple lock. improve and converge on this one.
;;
(define (common:simple-lock keyname)
(rmt:no-sync-get-lock keyname))
(define (common:simple-unlock keyname #!key (force #f))
(rmt:no-sync-del! keyname))
;;======================================================================
;; S T A T E A N D S T A T U S F O R T E S T S
;;======================================================================
;; speed up for common cases with a little logic
(define (mt:test-set-state-status-by-id run-id test-id newstate newstatus newcomment)
(if (not (and run-id test-id))
(begin
(debug:print-error 0 *default-log-port* "bad data handed to mt:test-set-state-status-by-id, run-id=" run-id ", test-id=" test-id ", newstate=" newstate)
(print-call-chain (current-error-port))
#f)
(begin
;; cond
;; ((and newstate newstatus newcomment)
;; (rmt:general-call 'state-status-msg run-id newstate newstatus newcomment test-id))
;; ((and newstate newstatus)
;; (rmt:general-call 'state-status run-id newstate newstatus test-id))
;; (else
;; (if newstate (rmt:general-call 'set-test-state run-id newstate test-id))
;; (if newstatus (rmt:general-call 'set-test-status run-id newstatus test-id))
;; (if newcomment (rmt:general-call 'set-test-comment run-id newcomment test-id))))
(rmt:set-state-status-and-roll-up-items run-id test-id #f newstate newstatus newcomment)
;; (mt:process-triggers run-id test-id newstate newstatus)
#t)))
(define (mt:test-set-state-status-by-id-unless-completed run-id test-id newstate newstatus newcomment)
(let* ((test-vec (rmt:get-testinfo-state-status run-id test-id))
(state (vector-ref test-vec 3)))
(if (equal? state "COMPLETED")
#t
(rmt:set-state-status-and-roll-up-items run-id test-id #f newstate newstatus newcomment))))
(define (mt:test-set-state-status-by-testname run-id test-name item-path new-state new-status new-comment)
;(let ((test-id (rmt:get-test-id run-id test-name item-path)))
(rmt:set-state-status-and-roll-up-items run-id test-name item-path new-state new-status new-comment)
;; (mt:process-triggers run-id test-id new-state new-status)
#t);)
;;(mt:test-set-state-status-by-id run-id test-id new-state new-status new-comment)))
(define (mt:test-set-state-status-by-testname-unless-completed run-id test-name item-path new-state new-status new-comment)
(let ((test-id (rmt:get-test-id run-id test-name item-path)))
(mt:test-set-state-status-by-id-unless-completed run-id test-id new-state new-status new-comment)))
;;======================================================================
;; R U N S
;;======================================================================
;; runs:get-runs-by-patt
;; get runs by list of criteria
;; register a test run with the db
;;
;; Use: (db-get-value-by-header (db:get-header runinfo)(db:get-rows runinfo))
;; to extract info from the structure returned
;;
(define (mt:get-runs-by-patt keys runnamepatt targpatt)
(let loop ((runsdat (rmt:get-runs-by-patt keys runnamepatt targpatt 0 500 #f 0))
(res '())
(offset 0)
(limit 500))
;; (print "runsdat: " runsdat)
(let* ((header (vector-ref runsdat 0))
(runslst (vector-ref runsdat 1))
(full-list (append res runslst))
(have-more (eq? (length runslst) limit)))
;; (debug:print 0 *default-log-port* "header: " header " runslst: " runslst " have-more: " have-more)
(if have-more
(let ((new-offset (+ offset limit))
(next-batch (rmt:get-runs-by-patt keys runnamepatt targpatt offset limit #f 0)))
(debug:print-info 4 *default-log-port* "More than " limit " runs, have " (length full-list) " runs so far.")
(debug:print-info 0 *default-log-port* "next-batch: " next-batch)
(loop next-batch
full-list
new-offset
limit))
(vector header full-list)))))
;;======================================================================
;; T E S T S
;;======================================================================
(define (mt:get-tests-for-run run-id testpatt states status #!key (not-in #t) (sort-by 'event_time) (sort-order "ASC") (qryvals #f)(last-update #f))
(let loop ((testsdat (rmt:get-tests-for-run run-id testpatt states status 0 500 not-in sort-by sort-order qryvals last-update 'normal))
(res '())
(offset 0)
(limit 500))
(let* ((full-list (append res testsdat))
(have-more (eq? (length testsdat) limit)))
(if have-more
(let ((new-offset (+ offset limit)))
(debug:print-info 4 *default-log-port* "More than " limit " tests, have " (length full-list) " tests so far.")
(loop (rmt:get-tests-for-run run-id testpatt states status new-offset limit not-in sort-by sort-order qryvals last-update 'normal)
full-list
new-offset
limit))
full-list))))
(define (mt:lazy-get-prereqs-not-met run-id waitons ref-item-path #!key (mode '(normal))(itemmaps #f) )
(let* ((key (list run-id waitons ref-item-path mode))
(res (hash-table-ref/default *pre-reqs-met-cache* key #f))
(useres (let ((last-time (if (vector? res) (vector-ref res 0) #f)))
(if last-time
(< (current-seconds)(+ last-time 5))
#f))))
(if useres
(let ((result (vector-ref res 1)))
(debug:print 4 *default-log-port* "Using lazy value res: " result)
result)
(let ((newres (rmt:get-prereqs-not-met run-id waitons ref-item-path mode: mode itemmaps: itemmaps)))
(hash-table-set! *pre-reqs-met-cache* key (vector (current-seconds) newres))
newres))))
;;======================================================================
;; from metadat lookup MEGATEST_VERSION
;;
(define (common:get-last-run-version) ;; RADT => How does this work in send-receive function??; assume it is the value saved in some DB
(rmt:get-var "MEGATEST_VERSION"))
(define (common:get-last-run-version-number)
(string->number
(substring (common:get-last-run-version) 0 6)))
(define (common:set-last-run-version)
(rmt:set-var "MEGATEST_VERSION" (common:version-signature)))
;;======================================================================
;; faux-lock is deprecated. Please use simple-lock below
;;
(define (common:faux-lock keyname #!key (wait-time 8)(allow-lock-steal #t))
(if (rmt:no-sync-get/default keyname #f) ;; do not be tempted to compare to pid. locking is a one-shot action, if already locked for this pid it doesn't actually count
(if (> wait-time 0)
(begin
(thread-sleep! 1)
(if (eq? wait-time 1) ;; only one second left, steal the lock
(begin
(debug:print-info 0 *default-log-port* "stealing lock for " keyname)
(common:faux-unlock keyname force: #t)))
(common:faux-lock keyname wait-time: (- wait-time 1)))
#f)
(begin
(rmt:no-sync-set keyname (conc (current-process-id)))
(equal? (conc (current-process-id)) (conc (rmt:no-sync-get/default keyname #f))))))
(define (common:faux-unlock keyname #!key (force #f))
(if (or force (equal? (conc (current-process-id)) (conc (rmt:no-sync-get/default keyname #f))))
(begin
(if (rmt:no-sync-get/default keyname #f) (rmt:no-sync-del! keyname))
#t)
#f))
;;======================================================================
;; postive number if megatest version > db version
;; negative number if megatest version < db version
(define (common:version-db-delta)
(- megatest-version (common:get-last-run-version-number)))
(define (common:version-changed?)
(not (equal? (common:get-last-run-version)
(common:version-signature))))
(define (common:api-changed?)
(not (equal? (substring (->string megatest-version) 0 4)
(substring (conc (common:get-last-run-version)) 0 4))))
;;======================================================================
;; Move me elsewhere ...
;; RADT => Why do we meed the version check here, this is called only if version misma
;;
#;(define (common:cleanup-db dbstruct #!key (full #f))
(apply db:multi-db-sync
dbstruct
'schema
;; 'new2old
'killservers
'adj-target
;; 'old2new
'new2old
;; (if full
'(dejunk)
;; '())
)
(if (common:api-changed?)
(common:set-last-run-version)))
;; This login does no retries under the hood - it acts a bit like a ping.
;; Deprecated for nmsg-transport.
;;
(define (rmt:login-no-auto-client-setup connection-info)
(rmt:send-receive-no-auto-client-setup connection-info 'login 0 (list *toppath* megatest-version *my-client-signature*)))
(define (rmt:send-receive-no-auto-client-setup connection-info cmd run-id params)
(let* ((run-id (if run-id run-id 0))
(res (handle-exceptions
exn
#f
(http-transport:client-api-send-receive run-id connection-info cmd params))))
(if (and res (vector-ref res 0))
(vector-ref res 1) ;;; YES!! THIS IS CORRECT!! CHANGE IT HERE, THEN CHANGE rmt:send-receive ALSO!!!
#f)))
(define (std-exit-procedure)
;;(common:telemetry-log-close)
(on-exit (lambda () 0))
;;(debug:print-info 13 *default-log-port* "std-exit-procedure called; *time-to-exit*="*time-to-exit*)
(let ((no-hurry (if (bdat-time-to-exit *bdat*) ;; hurry up
#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
(if *dbstruct-db* (db:close-all *dbstruct-db*)) ;; one second allocated
(if *server-info*
(let ((pkt-file (conc (get-pkts-dir *toppath*)
"/" (servdat-uuid *server-info*)
".pkt"))
(dbfile (servdat-dbfile *server-info*)))
(debug:print-info 0 *default-log-port* "removing pkt "pkt-file)
(delete-file* pkt-file)
(if (and dbfile
(string-match ".*/main.db$" dbfile))
(begin
(debug:print-info 0 *default-log-port* "Releasing lock for "dbfile)
(with-lock-db (servdat-dbfile *server-info*)
(lambda (dbh dbfile)
(db:release-lock dbh)))))))
(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
(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. 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)
)
)
0)
;;======================================================================
;; Force a megatest cleanup-db if version is changed and skip-version-check not specified
;; Do NOT check if not on homehost!
;;
#;(define (common:exit-on-version-changed)
(if (common:on-homehost?)
(if (common:api-changed?)
(let* ((mtconf (conc (get-environment-variable "MT_RUN_AREA_HOME") "/megatest.config"))
(dbfile (conc (get-environment-variable "MT_RUN_AREA_HOME") "/megatest.db"))
(read-only (not (file-writable? dbfile)))
(dbstruct (db:setup #t)))
(debug:print 0 *default-log-port*
"WARNING: Version mismatch!\n"
" expected: " (common:version-signature) "\n"
" got: " (common:get-last-run-version))
(cond
((get-environment-variable "MT_SKIP_DB_MIGRATE") #t)
((and (common:file-exists? mtconf) (common:file-exists? dbfile) (not read-only)
(eq? (current-user-id)(file-owner mtconf))) ;; safe to run -cleanup-db
(debug:print 0 *default-log-port* " I see you are the owner of megatest.config, attempting to cleanup and reset to new version")
(handle-exceptions
exn
(begin
(debug:print 0 *default-log-port* "Failed to switch versions. exn=" exn)
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(print-call-chain (current-error-port))
(exit 1))
(common:cleanup-db dbstruct)))
((not (common:file-exists? mtconf))
(debug:print 0 *default-log-port* " megatest.config does not exist in this area. Cannot proceed with megatest version migration.")
(exit 1))
((not (common:file-exists? dbfile))
(debug:print 0 *default-log-port* " megatest.db does not exist in this area. Cannot proceed with megatest version migration.")
(exit 1))
((not (eq? (current-user-id)(file-owner mtconf)))
(debug:print 0 *default-log-port* " You do not own megatest.db in this area. Cannot proceed with megatest version migration.")
(exit 1))
(read-only
(debug:print 0 *default-log-port* " You have read-only access to this area. Cannot proceed with megatest version migration.")
(exit 1))
(else
(debug:print 0 *default-log-port* " to switch versions you can run: \"megatest -cleanup-db\"")
(exit 1)))))))
;;======================================================================
;; (begin
;; (debug:print 0 *default-log-port* "ERROR: cannot migrate version unless on homehost. Exiting.")
;; (exit 1))))
(define (common:run-sync?)
;; (and (common:on-homehost?)
(args:get-arg "-server"))
;; this one seems to be the general entry point
;;
(define (server:start-and-wait areapath #!key (timeout 60))
(let ((give-up-time (+ (current-seconds) timeout)))
(let loop ((server-info (server:check-if-running areapath))
(try-num 0))
(if (or server-info
(> (current-seconds) give-up-time)) ;; server-url will be #f if no server available.
(server:record->url server-info)
(let ((num-ok (length (server:get-best (server:get-list areapath)))))
(if (and (> try-num 0) ;; first time through simply wait a little while then try again
(< num-ok 1)) ;; if there are no decent candidates for servers then try starting a new one
(server:kind-run areapath))
(thread-sleep! 5)
(loop (server:check-if-running areapath)
(+ try-num 1)))))))
(define (make-and-init-remote)
(make-remote ;; hh-dat: (common:get-homehost)
server-info: (if *toppath* (server:check-if-running *toppath*) #f)
server-timeout: (server:expiration-timeout)))
;; called in megatest.scm, host-port is string hostname:port
;;
;; NOTE: This is NOT called directly from clients as not all transports support a client running
;; in the same process as the server.
;;
(define (server:ping host-port-in server-id #!key (do-exit #f))
(let ((host:port (if (not host-port-in) ;; use read-dotserver to find
#f ;; (server:check-if-running *toppath*)
;; (if (number? host-port-in) ;; we were handed a server-id
;; (let ((srec (tasks:get-server-by-id (db:delay-if-busy (tasks:open-db)) host-port-in)))
;; ;; (print "srec: " srec " host-port-in: " host-port-in)
;; (if srec
;; (conc (vector-ref srec 3) ":" (vector-ref srec 4))
;; (conc "no such server-id " host-port-in)))
host-port-in))) ;; )
(let* ((host-port (if host:port
(let ((slst (string-split host:port ":")))
(if (eq? (length slst) 2)
(list (car slst)(string->number (cadr slst)))
#f))
#f)))
;; (toppath (launch:setup)))
;; (print "host-port=" host-port)
(if (not host-port)
(begin
(if host-port-in
(debug:print 0 *default-log-port* "ERROR: bad host:port"))
(if do-exit (exit 1))
#f)
(let* ((iface (car host-port))
(port (cadr host-port))
(server-dat (http-transport:client-connect iface port server-id))
(login-res (rmt:login-no-auto-client-setup server-dat)))
(if (and (list? login-res)
(car login-res))
(begin
;; (print "LOGIN_OK")
(if do-exit (exit 0))
#t)
(begin
;; (print "LOGIN_FAILED")
(if do-exit (exit 1))
#f)))))))
;; ping the given server
;;
(define (server:check-server server-record)
(let* ((server-url (server:record->url server-record))
(server-id (server:record->id server-record))
(res (case *transport-type*
((http)(server:ping server-url server-id))
;; ((nmsg)(nmsg-transport:ping (tasks:hostinfo-get-interface server)
)))
(if res
server-url
#f)))
;; 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)
(and (list? servers)
(< (length servers) (pseudo-random-integer ns)))) ;; somewhere between 0 and numservers
#f
(let loop ((hed (car servers))
(tal (cdr servers)))
(let ((res (server:check-server hed)))
(if res
hed
(if (null? tal)
#f
(loop (car tal)(cdr tal)))))))))
;; kind start up of servers, wait 40 seconds before allowing another server for a given
;; run-id to be launched
;;
(define (server:kind-run areapath)
;; look for $MT_RUN_AREA_HOME/logs/server-start-last
;; and wait for it to be at least 3 seconds old
(server:wait-for-server-start-last-flag areapath)
(if (not (server:check-if-running areapath)) ;; why try if there is already a server running?
(let* ((last-run-dat (hash-table-ref/default *server-kind-run* areapath '(0 0))) ;; callnum, whenrun
(call-num (car last-run-dat))
(when-run (cadr last-run-dat))
(run-delay (+ (case call-num
((0) 0)
((1) 20)
((2) 300)
(else 600))
(pseudo-random-integer 5))) ;; add a small random number just in case a lot of jobs hit the work hosts simultaneously
(lock-file (conc areapath "/logs/server-start.lock")))
(if (> (- (current-seconds) when-run) run-delay)
(let* ((start-flag (conc areapath "/logs/server-start-last")))
(common:simple-file-lock-and-wait lock-file expire-time: 15)
(debug:print-info 0 *default-log-port* "server:kind-run: touching " start-flag)
(system (conc "touch " start-flag)) ;; lazy but safe
(server:run areapath)
(thread-sleep! 2) ;; don't release the lock for at least a few seconds
(common:simple-file-release-lock lock-file)))
(hash-table-set! *server-kind-run* areapath (list (+ call-num 1)(current-seconds))))))
#;(define (client:connect iface port)
(http-transport:client-connect iface port)
#;(case (server:get-transport)
((rpc) (rpc:client-connect iface port))
((http) (http:client-connect iface port))
((zmq) (zmq:client-connect iface port))
(else (rpc:client-connect iface port))))
(define (client:setup areapath #!key (remaining-tries 100) (failed-connects 0))
(client:setup-http areapath remaining-tries: remaining-tries failed-connects: failed-connects)
#;(case (server:get-transport)
((rpc) (rpc-transport:client-setup remaining-tries: remaining-tries failed-connects: failed-connects)) ;;(client:setup-rpc run-id))
((http)(client:setup-http areapath remaining-tries: remaining-tries failed-connects: failed-connects))
(else (rpc-transport:client-setup remaining-tries: remaining-tries failed-connects: failed-connects)))) ;; (client:setup-rpc run-id))))
;; Do all the connection work, look up the transport type and set up the
;; connection if required.
;;
;; There are two scenarios.
;; 1. We are a test manager and we received *transport-type* and *runremote* via cmdline
;; 2. We are a run tests, list runs or other interactive process and we must figure out
;; *transport-type* and *runremote* from the monitor.db
;;
;; client:setup
;;
;; lookup_server, need to remove *runremote* stuff
;;
(define (client:setup-http areapath #!key (remaining-tries 100) (failed-connects 0)(area-dat #f))
(debug:print-info 2 *default-log-port* "client:setup remaining-tries=" remaining-tries)
(server:start-and-wait areapath)
(if (<= remaining-tries 0)
(begin
(debug:print-error 0 *default-log-port* "failed to start or connect to server")
(exit 1))
;;
;; Alternatively here, we can get the list of candidate servers and work our way
;; through them searching for a good one.
;;
(let* ((server-dat (server:get-rand-best areapath)) ;; (server:get-first-best areapath))
(runremote (or area-dat *runremote*)))
(if (not server-dat) ;; no server found
(client:setup-http areapath remaining-tries: (- remaining-tries 1))
(let ((host (cadr server-dat))
(port (caddr server-dat))
(server-id (caddr (cddr server-dat))))
(debug:print-info 4 *default-log-port* "client:setup server-dat=" server-dat ", remaining-tries=" remaining-tries)
(if (and (not area-dat)
(not *runremote*))
(begin
(set! *runremote* (make-and-init-remote))
(let* ((server-info (remote-server-info *runremote*)))
(if server-info
(begin
(remote-server-url-set! *runremote* (server:record->url server-info))
(remote-server-id-set! *runremote* (server:record->id server-info)))))))
(if (and host port server-id)
(let* ((start-res (case *transport-type*
((http)(http-transport:client-connect host port server-id))))
(ping-res (case *transport-type*
((http)(rmt:login-no-auto-client-setup start-res)))))
(if (and start-res
ping-res)
(let ((runremote (or area-dat *runremote*))) ;; it might have been generated only a few statements ago
(remote-conndat-set! runremote start-res) ;; (hash-table-set! runremote run-id start-res)
(debug:print-info 2 *default-log-port* "connected to " (http-transport:server-dat-make-url start-res))
start-res)
(begin ;; login failed but have a server record, clean out the record and try again
(debug:print-info 0 *default-log-port* "client:setup, login unsuccessful, will attempt to start server ... start-res=" start-res ", server-dat=" server-dat) ;; had runid. Fixes part of Randy;s ticket 1405717332
(case *transport-type*
((http)(http-transport:close-connections)))
(remote-conndat-set! runremote #f) ;; (hash-table-delete! runremote run-id)
(thread-sleep! 1)
(client:setup-http areapath remaining-tries: (- remaining-tries 1))
)))
(begin ;; no server registered
;; (server:kind-run areapath)
(server:start-and-wait areapath)
(debug:print-info 0 *default-log-port* "client:setup, no server registered, remaining-tries=" remaining-tries)
(thread-sleep! 1) ;; (+ 5 (pseudo-random-integer (- 20 remaining-tries)))) ;; give server a little time to start up, randomize a little to avoid start storms.
(client:setup-http areapath remaining-tries: (- remaining-tries 1)))))))))
)