16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
;; 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 apimod))
(declare (uses dbmod))
;; (declare (uses transport))
;; (declare (uses servermod))
(import scheme chicken data-structures extras)
(import (prefix sqlite3 sqlite3:) posix typed-records srfi-18)
(import commonmod)
(import dbmod)
(import apimod)
;; (import transport)
;; (import servermod)
(defstruct alldat
(areapath #f)
(ulexdat #f)
)
(define (rmtmod:calc-ro-mode runremote *toppath*)
|
>
|
|
>
|
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
|
|
|
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
|
;; 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 debugprint))
(declare (uses apimod))
(declare (uses dbmod))
(declare (uses configfmod))
(declare (uses margsmod))
(declare (uses portlogger))
(module rmtmod
*
(import scheme chicken data-structures extras ports)
(import (prefix sqlite3 sqlite3:)
directory-utils
intarweb
matchable
md5
message-digest
uri-common
spiffy
spiffy-directory-listing
spiffy-request-vars
http-client
posix
posix-extras
regex
typed-records
srfi-1
srfi-13
srfi-18
srfi-69
tcp)
(import apimod)
(import commonmod)
(import debugprint)
(import dbmod)
(import configfmod)
(import margsmod)
(import portlogger)
(defstruct alldat
(areapath #f)
(ulexdat #f)
)
(define (rmtmod:calc-ro-mode runremote *toppath*)
|
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
|
(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)))
(define (server:get-best-guess-address hostname)
(let ((res #f))
(for-each
(lambda (adr)
(if (not (eq? (u8vector-ref adr 0) 127))
(set! res adr)))
;; NOTE: This can fail when there is no mention of the host in /etc/hosts. FIXME
(vector->list (hostinfo-addresses (hostname->hostinfo hostname))))
(string-intersperse
(map number->string
(u8vector->list
(if res res (hostname->ip hostname)))) ".")))
;; moving this here as it needs access to db and cannot be in common.
;;
(define (server:get-bruteforce-syncer dbstruct #!key (fork-to-background #f) (persist-until-sync #f))
(let* ((sqlite-exe (or (get-environment-variable "MT_SQLITE3_EXE"))) ;; defined in cfg.sh
(sync-log (or ;; (args:get-arg "-sync-log")
(conc *toppath* "/logs/sync-" (current-process-id) "-" (get-host-name) ".log")))
(tmp-area (common:get-db-tmp-area))
(tmp-db (conc tmp-area "/megatest.db"))
(staging-file (conc *toppath* "/.megatest.db"))
(mtdbfile (conc *toppath* "/megatest.db"))
(lockfile (common:get-sync-lock-filepath))
(sync-cmd-core (conc sqlite-exe" " tmp-db " .dump | "sqlite-exe" " staging-file "&>"sync-log))
(sync-cmd (if fork-to-background
(conc "/usr/bin/env NBFAKE_LOG="*toppath*"/logs/last-server-sync-"(current-process-id)".log nbfake \""sync-cmd-core" && /bin/mv -f " staging-file " " mtdbfile" \"")
sync-cmd-core))
(default-min-intersync-delay 2)
(min-intersync-delay (configf:lookup-number *configdat* "server" "minimum-intersync-delay" default: default-min-intersync-delay))
(default-duty-cycle 0.1)
(duty-cycle (configf:lookup-number *configdat* "server" "sync-duty-cycle" default: default-duty-cycle))
(last-sync-seconds 10) ;; we will adjust this to a measurement and delay last-sync-seconds * (1 - duty-cycle)
(calculate-off-time (lambda (work-duration duty-cycle)
(* (/ (- 1 duty-cycle) duty-cycle) last-sync-seconds)))
(off-time min-intersync-delay) ;; adjusted in closure below.
(do-a-sync
(lambda ()
;; (BB> "Start do-a-sync with fork-to-background="fork-to-background" persist-until-sync="persist-until-sync)
(let* ((finalres
(let retry-loop ((num-tries 0))
(if (common:simple-file-lock lockfile)
(begin
(cond
((not (or fork-to-background persist-until-sync))
(debug:print 0 *default-log-port* "INFO: syncer thread sleeping for max of (server.minimum-intersync-delay="min-intersync-delay
" , off-time="off-time" seconds ]")
(thread-sleep! (max off-time min-intersync-delay)))
(else
(debug:print 0 *default-log-port* "INFO: syncer thread NOT sleeping ; maybe time-to-exit...")))
(if (not (configf:lookup *configdat* "server" "disable-db-snapshot"))
(common:snapshot-file mtdbfile subdir: ".db-snapshot"))
(delete-file* staging-file)
(let* ((start-time (current-milliseconds))
(res (system sync-cmd))
(dbbackupfile (conc mtdbfile ".backup"))
(res2
(cond
((eq? 0 res )
(handle-exceptions
exn
#f
(if (file-exists? dbbackupfile)
(delete-file* dbbackupfile)
)
(if (eq? 0 (file-size sync-log))
(delete-file* sync-log))
(system (conc "/bin/mv " staging-file " " mtdbfile))
(set! last-sync-seconds (/ (- (current-milliseconds) start-time) 1000))
(set! off-time (calculate-off-time
last-sync-seconds
(cond
((and (number? duty-cycle) (> duty-cycle 0) (< duty-cycle 1))
duty-cycle)
(else
(debug:print 0 *default-log-port* "WARNING: ["(common:human-time)"] server.sync-duty-cycle is invalid. Should be a number between 0 and 1, but "duty-cycle" was specified. Using default value: "default-duty-cycle)
default-duty-cycle))))
(debug:print 1 *default-log-port* "INFO: ["(common:human-time)"] pid="(current-process-id)" SYNC took "last-sync-seconds" sec")
(debug:print 1 *default-log-port* "INFO: ["(common:human-time)"] pid="(current-process-id)" SYNC took "last-sync-seconds" sec ; with duty-cycle of "duty-cycle" off time is now "off-time)
'sync-completed))
(else
(system (conc "/bin/cp "sync-log" "sync-log".fail"))
(debug:print 0 *default-log-port* "ERROR: ["(common:human-time)"] Sync failed. See log at "sync-log".fail")
(if (file-exists? (conc mtdbfile ".backup"))
(system (conc "/bin/cp "mtdbfile ".backup " mtdbfile)))
#f))))
(common:simple-file-release-lock lockfile)
;; (BB> "released lockfile: " lockfile)
#;(when (common:file-exists? lockfile)
(BB> "DID NOT ACTUALLY RELEASE LOCKFILE"))
res2) ;; end let
);; end begin
;; else
(cond
(persist-until-sync
(thread-sleep! 1)
(debug:print 1 *default-log-port* "INFO: ["(common:human-time)"] pid="(current-process-id)" other SYNC in progress; we're in a fork-to-background so we need to succeed. Let's wait a jiffy and and try again. num-tries="num-tries" (waiting for lockfile="lockfile" to disappear)")
(retry-loop (add1 num-tries)))
(else
(thread-sleep! (max off-time (+ last-sync-seconds min-intersync-delay)))
(debug:print 1 *default-log-port* "INFO: ["(common:human-time)"] pid="(current-process-id)" other SYNC in progress; not syncing.")
'parallel-sync-in-progress))
) ;; end if got lockfile
)
))
;; (BB> "End do-a-sync with fork-to-background="fork-to-background" persist-until-sync="persist-until-sync" and result="finalres)
finalres)
) ;; end lambda
))
do-a-sync))
|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
|
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
|
(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)))
;; (define server:sync-lock-token "SERVER_SYNC_LOCK")
;; (define (server:release-sync-lock)
;; (db:no-sync-del! *no-sync-db* server:sync-lock-token))
;; (define (server:have-sync-lock?)
;; (let* ((have-lock-pair (db:no-sync-get-lock *no-sync-db* server:sync-lock-token))
;; (have-lock? (car have-lock-pair))
;; (lock-time (cdr have-lock-pair))
;; (lock-age (- (current-seconds) lock-time)))
;; (cond
;; (have-lock? #t)
;; ((>lock-age
;; (* 3 (configf:lookup-number *configdat* "server" "minimum-intersync-delay" default: 180)))
;; (server:release-sync-lock)
;; (server:have-sync-lock?))
;; (else #f))))
;; (define (debug:print . params) #f)
;; (define (debug:print-info . params) #f)
;;
;; (define (set-functions dbgp dbgpinfo)
;; (set! debug:print dbgp)
;; (set! debug:print-info dbgpinfo))
|
2385
2386
2387
2388
2389
2390
2391
2392
2393
2394
2395
2396
2397
2398
|
(define (common:set-last-run-version)
(rmt:set-var "MEGATEST_VERSION" (common:version-signature)))
(define (common:get-last-run-version-number)
(string->number
(substring (common:get-last-run-version) 0 6)))
;;======================================================================
;; see defstruct host at top of file.
;; host: reachable last-update last-used last-cpuload
;;
(define (common:update-host-loads-table hosts-raw)
(let* ((hosts (filter (lambda (x)
(string-match (regexp "^\\S+$") x))
|
>
>
>
>
>
>
>
>
>
>
>
|
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315
2316
2317
2318
|
(define (common:set-last-run-version)
(rmt:set-var "MEGATEST_VERSION" (common:version-signature)))
(define (common:get-last-run-version-number)
(string->number
(substring (common:get-last-run-version) 0 6)))
(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? dbstruct)
(not (equal? (substring (->string megatest-version) 0 4)
(substring (conc (common:get-last-run-version)) 0 4))))
;;======================================================================
;; see defstruct host at top of file.
;; host: reachable last-update last-used last-cpuload
;;
(define (common:update-host-loads-table hosts-raw)
(let* ((hosts (filter (lambda (x)
(string-match (regexp "^\\S+$") x))
|
2550
2551
2552
2553
2554
2555
2556
2557
2558
2559
|
(loop (car tal)(cdr tal)))))))
;; no match, try again
(if (null? tal)
fallback-launcher
(loop (car tal)(cdr tal))))))))
fallback-launcher)))
)
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
2470
2471
2472
2473
2474
2475
2476
2477
2478
2479
2480
2481
2482
2483
2484
2485
2486
2487
2488
2489
2490
2491
2492
2493
2494
2495
2496
2497
2498
2499
2500
2501
2502
2503
2504
2505
2506
2507
2508
2509
2510
2511
2512
2513
2514
2515
2516
2517
2518
2519
2520
2521
2522
2523
2524
2525
2526
2527
2528
2529
2530
2531
2532
2533
2534
2535
2536
2537
2538
2539
2540
2541
2542
2543
2544
2545
2546
2547
2548
2549
2550
2551
2552
2553
2554
2555
2556
2557
2558
2559
2560
2561
2562
2563
2564
2565
2566
2567
2568
2569
2570
2571
2572
2573
2574
2575
|
(loop (car tal)(cdr tal)))))))
;; no match, try again
(if (null? tal)
fallback-launcher
(loop (car tal)(cdr tal))))))))
fallback-launcher)))
;;======================================================================
;; everything from client moved here
;;======================================================================
;; client:get-signature
(define (client:get-signature)
(if *my-client-signature* *my-client-signature*
(let ((sig (conc (get-host-name) " " (current-process-id))))
(set! *my-client-signature* sig)
*my-client-signature*)))
;; Not currently used! But, I think it *should* be used!!!
#;(define (client:logout serverdat)
(let ((ok (and (socket? serverdat)
(cdb:logout serverdat *toppath* (client:get-signature)))))
ok))
#;(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
;; POSSIBLE BUG. I removed the full initialization call. mrw
(set! *runremote* (create-remote-record))
(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 (http-transport:client-connect host port server-id))
(ping-res (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 (random (- 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)))))))))
)
|