︙ | | |
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
-
+
+
|
(import debugprint)
(import tcp-transportmod)
(use srfi-69
srfi-18
posix
matchable
s11n)
s11n
typed-records)
;; allow these queries through without starting a server
;;
(define api:read-only-queries
'(get-key-val-pairs
get-var
get-keys
|
︙ | | |
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
|
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
|
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
|
;; TASKS
tasks-add
tasks-set-state-given-param-key
))
(define *db-write-mutexes* (make-hash-table))
(define *server-signature* #f)
;; ;; These are called by the server on recipt of /api calls
;; ;; - keep it simple, only return the actual result of the call, i.e. no meta info here
;; ;;
;; ;; - returns #( flag result )
;; ;;
;; (define (api:execute-requests dbstruct dat)
;; (if (> *api-process-request-count* 50)
;; (begin
;; (if (common:low-noise-print 30 "too many threads")
;; (debug:print 0 *default-log-port* "WARNING: "*api-process-request-count*" threads, potential overload, adding 0.5 sec delay."))
;; ;; (thread-sleep! 0.5) ;; take a nap - no, the napping is moved to the clients via tt:backoff-incr
;; ))
;; (cond
;; ((not (vector? dat)) ;; it is an error to not receive a vector
;; (vector #f (vector #f "remote must be called with a vector")))
;; (else
;; (let* ((cmd-in (vector-ref dat 0))
;; (cmd (if (symbol? cmd-in)
;; cmd-in
;; (string->symbol cmd-in)))
;; (params (vector-ref dat 1))
;; (run-id (if (null? params)
;; 0
;; (car params)))
;; (write-mutex (if (hash-table-exists? *db-write-mutexes* run-id)
;; (hash-table-ref *db-write-mutexes* run-id)
;; (let* ((newmutex (make-mutex)))
;; (hash-table-set! *db-write-mutexes* run-id newmutex)
;; newmutex)))
;; (start-t (current-milliseconds))
;; (readonly-mode (dbr:dbstruct-read-only dbstruct))
;; (readonly-command (member cmd api:read-only-queries))
;; (writecmd-in-readonly-mode (and readonly-mode (not readonly-command))))
;; (if (not readonly-command)
;; (mutex-lock! write-mutex))
;; (let* ((tmppath (dbr:dbstruct-tmppath dbstruct))
;; (clean-run-id (cond
;; ((number? run-id) run-id)
;; ((equal? run-id #f) "main")
;; (else "other")))
;; (crumbfile (dbfile:wait-for-qif tmppath clean-run-id (cons cmd params)))
;; (res
;; (if writecmd-in-readonly-mode
;; (conc "attempt to run write command "cmd" on a read-only database")
;; (api:dispatch-request dbstruct cmd run-id params))))
;; (delete-file* crumbfile)
;; (if (not readonly-command)
;; (mutex-unlock! write-mutex))
;;
;; ;; save all stats
;; (let ((delta-t (- (current-milliseconds)
;; start-t))
;; (modified-cmd (if (eq? cmd 'general-call)
;; (string->symbol (conc "general-call-" (car params)))
;; cmd)))
;; (hash-table-set! *db-api-call-time* modified-cmd
;; (cons delta-t (hash-table-ref/default *db-api-call-time* modified-cmd '()))))
;; (if writecmd-in-readonly-mode
;; (begin
;; #;(common:telemetry-log (conc "api-out:"(->string cmd))
;; payload: `((params . ,params)
;; (ok-res . #t)))
;; (vector #f res))
;; (begin
;; #;(common:telemetry-log (conc "api-out:"(->string cmd))
;; payload: `((params . ,params)
;; (ok-res . #f)))
;; (vector #t res))))))))
(define *api-threads* '())
(define (api:register-thread th-in)
(set! *api-threads* (cons (cons th-in (current-seconds)) *api-threads*)))
(define (api:unregister-thread th-in)
(set! *api-threads* (filter (lambda (thdat)
(not (eq? th-in (car thdat))))
*api-threads*)))
(define (api:remove-dead-or-terminated)
(set! *api-threads* (filter (lambda (thdat)
(not (member (thread-state (car thdat)) '(terminated dead))))
*api-threads*)))
(define (api:get-count-threads-alive)
(length *api-threads*))
(define *api:last-stats-print* 0)
(define *api-print-db-stats-mutex* (make-mutex))
(define (api:print-db-stats)
(debug:print-info 0 *default-log-port* "Started periodic db stats printer")
(let loop ()
(mutex-lock! *api-print-db-stats-mutex*)
(if (> (- (current-seconds) *api:last-stats-print*) 15)
(begin
(rmt:print-db-stats)
(set! *api:last-stats-print* (current-seconds))))
(mutex-unlock! *api-print-db-stats-mutex*)
(thread-sleep! 5)
(loop)))
;; indat is (cmd run-id params meta)
;;
;; WARNING: Do not print anything in the lambda of this function as it
;; reads/writes to current in/out port
;;
(define (api:tcp-dispatch-request-make-handler dbstruct) ;; cmd run-id params)
(assert *toppath* "FATAL: api:tcp-dispatch-request-make-handler called but *toppath* not set.")
(if (not *server-signature*)
(set! *server-signature* (tt:mk-signature *toppath*)))
(lambda (indat)
(api:register-thread (current-thread))
(let* (;; (indat (deserialize))
(let* ((result
(newcount (+ *api-process-request-count* 1))
(numthreads (api:get-count-threads-alive))
(delay-wait (if (> newcount 10)
(- newcount 10)
0))
(normal-proc (lambda (cmd run-id params)
(case cmd
((ping) *server-signature*)
(else
(api:dispatch-request dbstruct cmd run-id params))))))
(set! *api-process-request-count* newcount)
(set! *db-last-access* (current-seconds))
(if (not (eq? newcount numthreads))
(begin
(api:remove-dead-or-terminated)
(let ((threads-now (api:get-count-threads-alive)))
(debug:print 0 *default-log-port* "WARNING: newcount="newcount", numthreads="numthreads", remaining="threads-now)
(set! newcount threads-now))))
(match indat
((cmd run-id params meta)
(let* ((db-ok (let* ((dbfname (dbmod:run-id->dbfname run-id))
(ok (equal? dbfname (dbr:dbstruct-dbfname dbstruct))))
(case cmd
((ping) #t) ;; we are fine
(else
(let* ((numthreads (api:get-count-threads-alive))
(delay-wait (if (> numthreads 10)
(- numthreads 10)
0))
(normal-proc (lambda (cmd run-id params)
(case cmd
((ping) *server-signature*)
(else
(api:dispatch-request dbstruct cmd run-id params))))))
(set! *api-process-request-count* numthreads)
(set! *db-last-access* (current-seconds))
;; (if (not (eq? numthreads numthreads))
;; (begin
;; (api:remove-dead-or-terminated)
;; (let ((threads-now (api:get-count-threads-alive)))
;; (debug:print 0 *default-log-port* "WARNING: numthreads="numthreads", numthreads="numthreads", remaining="threads-now)
;; (set! numthreads threads-now))))
(match indat
((cmd run-id params meta)
(let* ((start-t (current-milliseconds))
(db-ok (let* ((dbfname (dbmod:run-id->dbfname run-id))
(ok (equal? dbfname (dbr:dbstruct-dbfname dbstruct))))
(case cmd
((ping) #t) ;; we are fine
(else
(if (not ok)(debug:print 0 *default-log-port* "ERROR: "cmd", run-id "run-id", not correct for dbfname "(dbr:dbstruct-dbfname dbstruct)))
(assert ok "FATAL: database file and run-id not aligned.")))))
(ttdat *server-info*)
(server-state (tt-state ttdat))
(status (cond
((> newcount 3) 'busy)
;; ((> newcount 5) 'loaded) ;; this gets transmitted to the client which calls tt:backoff-incr to slow stuff down.
(else 'ok)))
(errmsg (case status
((busy) (conc "Server overloaded, "newcount" threads in flight"))
((loaded) (conc "Server loaded, "newcount" threads in flight"))
(else #f)))
(result (case status
((busy)
(if (eq? cmd 'ping)
(normal-proc cmd run-id params)
;; newcount must be greater than 5 for busy
(* 1 (- newcount 3)) ;; was 15
)) ;; (- newcount 29)) ;; call back in as many seconds
((loaded)
;; (if (eq? (rmt:transport-mode) 'tcp)
;; (thread-sleep! 0.5))
(normal-proc cmd run-id params))
(else
(normal-proc cmd run-id params))))
(meta (case cmd
((ping) `((sstate . ,server-state)))
(else `((wait . ,delay-wait)))))
(payload (list status errmsg result meta)))
(set! *api-process-request-count* (- *api-process-request-count* 1))
;; (serialize payload)
(api:unregister-thread (current-thread))
payload))
(else
(assert #f "FATAL: failed to deserialize indat "indat))))))
(assert ok "FATAL: database file and run-id not aligned.")))))
(ttdat *server-info*)
(server-state (tt-state ttdat))
(maxthreads 20) ;; make this a parameter?
(status (cond
((and (> numthreads maxthreads)
(> (random 100) 70)) ;; allow a 30% probability to go through so we can figure out what is going wrong in main.db server.
'busy)
;; ((> numthreads 5) 'loaded) ;; this gets transmitted to the client which calls tt:backoff-incr to slow stuff down.
(else 'ok)))
(errmsg (case status
((busy) (conc "Server overloaded, "numthreads" threads in flight"))
((loaded) (conc "Server loaded, "numthreads" threads in flight"))
(else #f)))
(result (case status
((busy)
(if (eq? cmd 'ping)
(normal-proc cmd run-id params)
;; numthreads must be greater than 5 for busy
(* 0.1 (- numthreads maxthreads)) ;; was 15
)) ;; (- numthreads 29)) ;; call back in as many seconds
((loaded)
;; (if (eq? (rmt:transport-mode) 'tcp)
;; (thread-sleep! 0.5))
(normal-proc cmd run-id params))
(else
(normal-proc cmd run-id params))))
(meta (case cmd
((ping) `((sstate . ,server-state)))
(else `((wait . ,delay-wait)))))
(payload (list status errmsg result meta)))
;; (cmd run-id params meta)
(db:add-stats cmd run-id params (- (current-milliseconds) start-t))
payload))
(else
(assert #f "FATAL: failed to deserialize indat "indat))))))
;; (set! *api-process-request-count* (- *api-process-request-count* 1))
;; (serialize payload)
(api:unregister-thread (current-thread))
result)))
(define *api-halt-writes* #f)
(define (api:dispatch-request dbstruct cmd run-id params)
(if (not *no-sync-db*)
(db:open-no-sync-db))
(let* ((start-time (current-milliseconds)))
(if (member cmd api:write-queries)
(let loop ()
(if *api-halt-writes*
(begin
(thread-sleep! 0.2)
(if (< (- (current-milliseconds) start-time)
5000) ;; hope it don't take more than five seconds to sync
(loop-time)
#;(debug:print 0 *default-log-port* "ERROR: writes halted for more than 5 seconds, sync might be taking too long"))))))
(db:add-stats 'api-write-blocking-for-sync run-id params (- (current-milliseconds) start-time)))
(case cmd
;;===============================================
;; READ/WRITE QUERIES
;;===============================================
((get-keys-write) (db:get-keys dbstruct)) ;; force a dummy "write" query to force server; for debug in -repl
;; SERVERS
((start-server) (apply server:kind-run params))
((start-server) (apply tt:server-process-run params))
((kill-server) (set! *server-run* #f))
;; TESTS
;;((test-set-state-status-by-id) (apply mt:test-set-state-status-by-id dbstruct params))
;;BB - commented out above because it was calling below, eventually, incorrectly (dbstruct passed to mt:test-set-state-status-by-id, which previosly did more, but now only passes thru to db:set-state-status-and-roll-up-items.
((test-set-state-status-by-id)
|
︙ | | |
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
|
478
479
480
481
482
483
484
|
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
|
;; TASKS
((find-task-queue-records) (apply tasks:find-task-queue-records dbstruct params))
(else
(debug:print 0 *default-log-port* "ERROR: bad api call " cmd)
(conc "ERROR: BAD api call " cmd))))
;; http-server send-response
;; api:process-request
;; db:*
;;
;; NB// Runs on the server as part of the server loop
;;
(define (api:process-request dbstruct $) ;; the $ is the request vars proc
(debug:print 4 *default-log-port* "server-id:" *server-id*)
(let* ((cmd ($ 'cmd))
(paramsj ($ 'params))
(key ($ 'key))
(params (db:string->obj paramsj transport: 'http))) ;; incoming data from the POST (or is it a GET?)
(debug:print 4 *default-log-port* "cmd:" cmd " with params " params "key " key)
(if (equal? key *server-id*)
(begin
(set! *api-process-request-count* (+ *api-process-request-count* 1))
(let* ((resdat (api:execute-requests dbstruct (vector cmd params))) ;; process the request, resdat = #( flag result )
(success (vector-ref resdat 0))
(res (vector-ref resdat 1))) ;; (vector flag payload), get the payload, ignore the flag (why?)
(debug:print 4 *default-log-port* "res:" res)
(if (not success)
(debug:print 0 *default-log-port* "ERROR: success flag is #f for " cmd " with params " params))
(if (> *api-process-request-count* *max-api-process-requests*)
(set! *max-api-process-requests* *api-process-request-count*))
(set! *api-process-request-count* (- *api-process-request-count* 1))
;; This can be here but needs controls to ensure it doesn't run more than every 4 seconds
;; (rmt:dat->json-str
;; (if (or (string? res)
;; (list? res)
;; (number? res)
;; (boolean? res))
;; res
;; (list "ERROR, not string, list, number or boolean" 1 cmd params res)))))
(db:obj->string res transport: 'http)))
(begin
(debug:print 0 *default-log-port* "Server refused to process request. Server id mismatch. recived " key " expected: " *server-id* ".\nOther arguments recived: cmd=" cmd " params = " params)
(db:obj->string (conc "Server refused to process request server-id mismatch: " key ", " *server-id*) transport: 'http)))))
|
︙ | | |
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
+
|
;;======================================================================
(declare (unit common))
(declare (uses commonmod))
(declare (uses rmtmod))
(declare (uses debugprint))
(declare (uses mtargs))
(use srfi-1 data-structures posix regex-case (prefix base64 base64:)
format dot-locking csv-xml z3 udp ;; sql-de-lite
hostinfo md5 message-digest typed-records directory-utils stack
matchable regex posix (srfi 18) extras ;; tcp
(prefix nanomsg nmsg:)
(prefix sqlite3 sqlite3:)
|
︙ | | |
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
-
-
-
-
|
;; (define *common:denoise* (make-hash-table)) ;; for low noise printing
(define *default-log-port* (current-error-port))
(define *time-zero* (current-seconds)) ;; for the watchdog
(define *on-exit-procs* '()) ;; add procs to this list to be executed on exit
(define *default-area-tag* "local")
;; DATABASE
;; (define *dbstruct-dbs* #f) ;; used to cache the dbstruct in db:setup. Goal is to remove this.
;; db stats
(define *db-stats* (make-hash-table)) ;; hash of vectors < count duration-total >
(define *db-stats-mutex* (make-mutex))
;; db access
(define *db-last-access* (current-seconds)) ;; last db access, used in server
;; (define *db-write-access* #t)
;; db sync
;; (define *db-last-sync* 0) ;; last time the sync to megatest.db happened
(define *db-sync-in-progress* #f) ;; if there is a sync in progress do not try to start another
;; (define *db-multi-sync-mutex* (make-mutex)) ;; protect access to *db-sync-in-progress*, *db-last-sync*
|
︙ | | |
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
|
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
-
|
(define *transport-type* 'http) ;; override with [server] transport http|rpc|nmsg
(define *runremote* #f) ;; if set up for server communication this will hold <host port>
;; (define *max-cache-size* 0)
(define *logged-in-clients* (make-hash-table))
(define *server-id* #f)
(define *server-info* #f) ;; good candidate for easily convert to non-global
(define *time-to-exit* #f)
(define *server-run* #t)
(define *run-id* #f)
(define *server-kind-run* (make-hash-table))
(define *home-host* #f)
;; (define *total-non-write-delay* 0)
(define *heartbeat-mutex* (make-mutex))
;; (define *api-process-request-count* 0)
;; (define *max-api-process-requests* 0)
|
︙ | | |
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
|
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
-
+
|
;;======================================================================
(define *common:this-exe-fullpath* (common:get-this-exe-fullpath))
(define *common:this-exe-dir* (pathname-directory *common:this-exe-fullpath*))
(define *common:this-exe-name* (pathname-strip-directory *common:this-exe-fullpath*))
(define (common:get-sync-lock-filepath)
(let* ((tmp-area (common:get-db-tmp-area))
(let* ((tmp-area (common:make-tmpdir-name *toppath* ""))
(lockfile (conc tmp-area "/megatest.db.lock")))
lockfile))
(define *common:logpro-exit-code->status-sym-alist*
'( ( 0 . pass )
( 1 . fail )
( 2 . warn )
|
︙ | | |
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
|
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
|
-
+
|
;;======================================================================
;; lazy-safe get file mod time. on any error (file not existing etc.) return 0
;;
(define (common:lazy-modification-time fpath)
(handle-exceptions
exn
(begin
(debug:print 0 *default-log-port* "Failed to get modifcation time for " fpath ", treating it as zero. exn=" exn)
(debug:print 2 *default-log-port* "Failed to get modification time for " fpath ", treating it as zero. exn=" exn)
0)
(if (file-exists? fpath)
(file-modification-time fpath)
0)))
;;======================================================================
;; find timestamp of newest file associated with a sqlite db file
|
︙ | | |
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
|
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
|
-
+
|
;; returns: ok/not dbspace required-space
;;
(define (common:check-db-dir-space)
(let* ((required (string->number
;; default is 1GB (or actually a billion bytes) This is the number of 1 kB blocks.
(or (configf:lookup *configdat* "setup" "dbdir-space-required")
"1000000")))
(dbdir (common:get-db-tmp-area)) ;; (db:get-dbdir))
(dbdir (common:make-tmpdir-name *toppath* "")) ;; (db:get-dbdir))
(tdbspace (common:check-space-in-dir dbdir required))
(mdbspace (common:check-space-in-dir *toppath* required)))
(sort (list tdbspace mdbspace) (lambda (a b)
(< (cadr a)(cadr b))))))
;;======================================================================
;; check available space in dbdir, exit if insufficient
|
︙ | | |
︙ | | |
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
-
+
|
(debug:print-error 0 *default-log-port* " query " stmt " failed, params: " params ", error: " ((condition-property-accessor 'exn 'message) exn) ", exn=" exn)
(print-call-chain (current-error-port))
default)))
(apply sqlite3:first-result db stmt params)))
(define (db:setup do-sync)
(assert *toppath* "FATAL: db:setup called before launch:setup has been run.")
(let* ((tmpdir (common:get-db-tmp-area)))
(let* ((tmpdir (common:make-tmpdir-name *toppath* "")))
(if (not *dbstruct-dbs*)
(dbfile:setup do-sync *toppath* tmpdir)
*dbstruct-dbs*)))
;; moved from dbfile
;;
;; ADD run-id SUPPORT
|
︙ | | |
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
|
265
266
267
268
269
270
271
272
273
274
275
276
277
278
|
-
-
-
-
-
-
-
|
(debug:print-error 0 *default-log-port* " params: " params
", error: " ((condition-property-accessor 'exn 'message) exn)
", arguments: " ((condition-property-accessor 'exn 'arguments) exn)
", location: " ((condition-property-accessor 'exn 'location) exn)
))
;; NB// #f => return dbdir only
;; (was planned to be; zeroth db with name=main.db)
;;
;; If run-id is #f return to create and retrieve the path where the db will live.
;;
(define db:dbfile-path common:get-db-tmp-area)
(define (db:set-sync db)
(let ((syncprag (configf:lookup *configdat* "setup" "sychronous")))
(sqlite3:execute db (conc "PRAGMA synchronous = " (or syncprag 0) ";"))))
(define (db:get-last-update-time db)
(let ((last-update-time #f))
|
︙ | | |
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
|
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
|
-
+
|
(max (get-mtime fname)
(get-mtime wal-file)
(get-mtime shm-file))))
;; (define (db:all-db-sync dbstruct)
;; (let* ((dbdat (db:open-db dbstruct #f db:initialize-main-db))
;; (data-synced 0) ;; count of changed records
;; (tmp-area (common:get-db-tmp-area))
;; (tmp-area (common:make-tmpdir-name *toppath*))
;; (dbfiles (glob (conc tmp-area"/.mtdb/*.db")))
;; (sync-durations (make-hash-table))
;; (no-sync-db (db:open-no-sync-db)))
;; (for-each
;; (lambda (file) ;; tmp db file
;; (debug:print-info 3 *default-log-port* "file: " file)
;; (let* ((fname (conc (pathname-file file) ".db")) ;; fname is tmp db file
|
︙ | | |
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
|
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
|
-
+
|
;; 'closeall - close all opened dbs
;; 'schema - attempt to apply schema changes
;; run-ids: '(1 2 3 ...) or #f (for all)
;;
(define (db:multi-db-sync dbstruct . options)
(let* (;; (dbdat (db:open-db dbstruct #f dbfile:db-init-proc))
(data-synced 0) ;; count of changed records
(tmp-area (common:get-db-tmp-area))
(tmp-area (common:make-tmpdir-name *toppath* ""))
(old2new (member 'old2new options))
(dejunk (member 'dejunk options))
(killservers (member 'killservers options))
(src-area (if old2new *toppath* tmp-area))
(dest-area (if old2new tmp-area *toppath*))
(dbfiles (if old2new (glob (conc *toppath* "/.mtdb/*.db"))
(glob (conc tmp-area "/.mtdb/*.db"))))
|
︙ | | |
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
|
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
|
-
+
|
;;======================================================================
;; no-sync.db - small bits of data to be shared between servers
;;======================================================================
(define (db:get-dbsync-path)
(case (rmt:transport-mode)
((http)(common:get-db-tmp-area))
((http)(common:make-tmpdir-name *toppath* ""))
((tcp) (conc *toppath*"/.mtdb"))
((nfs) (conc *toppath*"/.mtdb"))
(else "/tmp/dunno-this-gonna-exist")))
;; This is needed for api.scm
(define (db:open-no-sync-db)
(dbfile:open-no-sync-db (db:get-dbsync-path)))
|
︙ | | |
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
|
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
|
-
+
|
res))
;; TODO: Switch this to use max(update_time) from each run db? Then if using a server there is no disk traffic (using cachedb db) ???
;;
;; NOTE: This DOESN'T (necessarily) get the real run ids, but the number of the <number>.db!!
(define (db:get-changed-run-ids since-time)
(let* ((dbdir (db:dbfile-path)) ;; (configf:lookup *configdat* "setup" "dbdir"))
(let* ((dbdir (common:make-tmpdir-name *toppath* "")) ;; (configf:lookup *configdat* "setup" "dbdir"))
(alldbs (glob (conc *toppath* "/.mtdb/[0-9]*.db*")))
(changed (filter (lambda (dbfile)
(> (file-modification-time dbfile) since-time))
alldbs)))
(delete-duplicates
(map (lambda (dbfile)
(let* ((res (string-match ".*\\/(\\d+)\\.db*" dbfile)))
|
︙ | | |
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
|
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
|
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
'()
db
qry
run-id
(or last-update 0))))))
(define (db:get-testinfo-state-status dbstruct run-id test-id)
(let ((res #f))
(db:with-db dbstruct run-id #f
(lambda (dbdat db)
(sqlite3:for-each-row
(lambda (run-id testname item-path state status)
;; id,run_id,testname,state,status,event_time,host,cpuload,diskfree,uname,rundir,item_path,run_duration,final_logf,comment
(set! res (vector test-id run-id testname state status -1 "" -1 -1 "" "-" item-path -1 "-" "-")))
db
"SELECT run_id,testname,item_path,state,status FROM tests WHERE id=? and run_id=?;"
test-id run-id)))
res))
(db:with-db
dbstruct run-id #f
(lambda (dbdat db)
(let* ((res #f)
(stmth (db:get-cache-stmth dbdat db "SELECT run_id,testname,item_path,state,status FROM tests WHERE id=? and run_id=?;")))
(sqlite3:for-each-row
(lambda (run-id testname item-path state status)
;; id,run_id,testname,state,status,event_time,host,cpuload,diskfree,uname,rundir,item_path,run_duration,final_logf,comment
(set! res (vector test-id run-id testname state status -1 "" -1 -1 "" "-" item-path -1 "-" "-")))
;; db
;; "SELECT run_id,testname,item_path,state,status FROM tests WHERE id=? and run_id=?;"
stmth
test-id run-id)
res))))
;; get a useful subset of the tests data (used in dashboard
;; use db:mintest-get-{id ,run_id,testname ...}
;;
(define (db:get-tests-for-run-mindata dbstruct run-id testpatt states statuses not-in)
(db:get-tests-for-run dbstruct run-id testpatt states statuses #f #f not-in #f #f "id,run_id,testname,state,status,event_time,item_path" 0 #f))
|
︙ | | |
2635
2636
2637
2638
2639
2640
2641
2642
2643
2644
2645
2646
2647
2648
2649
2650
2651
2652
2653
2654
2655
2656
|
2631
2632
2633
2634
2635
2636
2637
2638
2639
2640
2641
2642
2643
2644
2645
2646
2647
2648
2649
2650
2651
2652
|
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
|
;;
(define (db:get-test-state-status-by-id dbstruct run-id test-id)
(db:with-db
dbstruct
run-id
#f
(lambda (dbdat db)
(let ((res (cons #f #f)))
;; (stmth (db:get-cache-stmth dbdat db "SELECT state,status FROM tests WHERE id=?;")))
(sqlite3:for-each-row ;; attemptnum added to hold pid of top process (not Megatest) controlling a test
(lambda (state status)
(cons state status))
db
"SELECT state,status FROM tests WHERE id=? AND run_id=?;" ;; stmth try not compiling this one - yes, this fixed the bind issue
test-id run-id)
(let ((res (cons #f #f))
(stmth (db:get-cache-stmth dbdat db "SELECT state,status FROM tests WHERE id=? AND run_id=?;")))
(sqlite3:for-each-row ;; attemptnum added to hold pid of top process (not Megatest) controlling a test
(lambda (state status)
(cons state status))
;; db
stmth ;;"SELECT state,status FROM tests WHERE id=? AND run_id=?;" ;; stmth try not compiling this one - yes, this fixed the bind issue
test-id run-id)
res))))
;; Use db:test-get* to access
;; Get test data using test_ids. NB// Only works within a single run!!
;;
(define (db:get-test-info-by-ids dbstruct run-id test-ids)
(db:with-db
|
︙ | | |
3726
3727
3728
3729
3730
3731
3732
3733
3734
3735
3736
3737
3738
3739
3740
3741
3742
3743
3744
3745
3746
3747
3748
3749
3750
3751
3752
3753
3754
|
3722
3723
3724
3725
3726
3727
3728
3729
3730
3731
3732
3733
3734
3735
3736
3737
3738
3739
3740
3741
3742
3743
3744
3745
3746
3747
3748
3749
3750
3751
3752
3753
3754
3755
3756
3757
3758
3759
3760
|
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
(delete-duplicates
(cons testname (hash-table-ref/default res tag '())))))
tags)))
db
"SELECT testname,tags FROM test_meta")
(hash-table->alist res)))))
;; testmeta doesn't change, we can cache it for up too an hour
(define *db:testmeta-cache* (make-hash-table))
(define *db:testmeta-last-update* 0)
;; read the record given a testname
(define (db:testmeta-get-record dbstruct testname)
(if (and (< (- (current-seconds) *db:testmeta-last-update*) 600)
(hash-table-exists? *db:testmeta-cache* testname))
(hash-table-ref *db:testmeta-cache* testname)
(let ((res #f))
(db:with-db
dbstruct
#f
#f
(lambda (dbdat db)
(sqlite3:for-each-row
(lambda (id testname author owner description reviewed iterated avg_runtime avg_disk tags jobgroup)
(set! res (vector id testname author owner description reviewed iterated avg_runtime avg_disk tags jobgroup)))
db
"SELECT id,testname,author,owner,description,reviewed,iterated,avg_runtime,avg_disk,tags,jobgroup FROM test_meta WHERE testname=?;"
testname)
res))))
(let ((res #f))
(db:with-db
dbstruct
#f
#f
(lambda (dbdat db)
(sqlite3:for-each-row
(lambda (id testname author owner description reviewed iterated avg_runtime avg_disk tags jobgroup)
(set! res (vector id testname author owner description reviewed iterated avg_runtime avg_disk tags jobgroup)))
db
"SELECT id,testname,author,owner,description,reviewed,iterated,avg_runtime,avg_disk,tags,jobgroup FROM test_meta WHERE testname=?;"
testname)))
(hash-table-set! *db:testmeta-cache* testname res)
(set! *db:testmeta-last-update* (current-seconds))
res)))
;; create a new record for a given testname
(define (db:testmeta-add-record dbstruct testname)
(db:with-db dbstruct #f #t
(lambda (dbdat db)
(sqlite3:execute
db
|
︙ | | |
4312
4313
4314
4315
4316
4317
4318
4319
4320
4321
4322
4323
4324
4325
4326
|
4318
4319
4320
4321
4322
4323
4324
4325
4326
4327
4328
4329
4330
4331
4332
|
-
+
|
(debug:print 0 *default-log-port* "could not get lock for " from-db " from no-sync-db")
#f
))))
;; sync for filesystem local db writes
;;
(define (db:run-lock-and-sync no-sync-db)
(let* ((tmp-area (common:get-db-tmp-area))
(let* ((tmp-area (common:make-tmpdir-name *toppath* ""))
(dbfiles (glob (conc tmp-area"/.mtdb/*.db")))
(sync-durations (make-hash-table)))
;; (debug:print-info 0 *default-log-port* "lock-and-sync, dbfiles: "dbfiles)
(for-each
(lambda (file)
(let* ((fname (conc (pathname-file file) ".db"))
(fulln (conc *toppath*"/.mtdb/"fname))
|
︙ | | |
4368
4369
4370
4371
4372
4373
4374
4375
4376
4377
4378
4379
4380
4381
4382
|
4374
4375
4376
4377
4378
4379
4380
4381
4382
4383
4384
4385
4386
4387
4388
|
-
+
|
(thread-sleep! 0.05) ;; delay for startup
(let ((legacy-sync (common:run-sync?))
(sync-stale-seconds (configf:lookup-number *configdat* "server" "sync-stale-seconds" default: 300))
(debug-mode (debug:debug-mode 1))
(last-time (current-seconds)) ;; last time through the sync loop
(no-sync-db (db:open-no-sync-db))
(sync-duration 0) ;; run time of the sync in milliseconds
(tmp-area (common:get-db-tmp-area)))
(tmp-area (common:make-tmpdir-name *toppath* "")))
;; Sync moved to http-transport keep-running loop
(debug:print-info 2 *default-log-port* "Periodic copy-based sync thread started. syncer is copy-sync, tmp-area is " tmp-area)
(debug:print-info 3 *default-log-port* "watchdog starting. syncer is copy-sync pid="(current-process-id));; " this-wd-num="this-wd-num)
(if (and legacy-sync (not *time-to-exit*))
(begin
(debug:print-info 0 *default-log-port* "Server running, periodic copy-based sync started.")
|
︙ | | |
4476
4477
4478
4479
4480
4481
4482
4483
4484
4485
4486
4487
4488
4489
4490
|
4482
4483
4484
4485
4486
4487
4488
4489
4490
4491
4492
4493
4494
4495
4496
|
-
+
|
;;
(for-each
(lambda (subdb)
(let* (;;(dbstruct (db:setup))
(mtdb (dbr:subdb-mtdb subdb))
(mtpath (db:dbdat-get-path mtdb))
(tmp-area (common:get-db-tmp-area))
(tmp-area (common:make-tmpdir-name *toppath* ""))
(res (db:sync-to-megatest.db dbstruct no-sync-db: no-sync-db))) ;; did we sync any data? If so need to set the db touched flag to keep the server alive
(set! sync-duration (- (current-milliseconds) sync-start))
(if (> res 0) ;; some records were transferred, keep the db alive
(begin
(mutex-lock! *heartbeat-mutex*)
(set! *db-last-access* (current-seconds))
(mutex-unlock! *heartbeat-mutex*)
|
︙ | | |
4523
4524
4525
4526
4527
4528
4529
4530
4531
4532
4533
4534
4535
4536
4537
|
4529
4530
4531
4532
4533
4534
4535
4536
4537
4538
4539
4540
4541
4542
|
-
|
(if (not *time-to-exit*) (loop))))
;; ;; time to exit, close the no-sync db here
;; (db:no-sync-close-db no-sync-db stmt-cache)
(if (common:low-noise-print 30)
(debug:print-info 0 *default-log-port* "Exiting watchdog timer, *time-to-exit* = " *time-to-exit*" pid="(current-process-id) ))))
))
(define (std-exit-procedure)
;;(common:telemetry-log-close)
(on-exit (lambda () 0)) ;; why is this here?
;;(debug:print-info 13 *default-log-port* "std-exit-procedure called; *time-to-exit*="*time-to-exit*)
(let ((no-hurry (if *time-to-exit* ;; hurry up
#f
|
︙ | | |
︙ | | |
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
|
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
+
-
+
|
#t
)
#f
)
)
(define (dbfile:make-tmpdir-name areapath tmpadj)
(let* ((area (pathname-file areapath))
(let* ((dname (conc "/tmp/"(current-user-name)"/" (string-translate areapath "/" ".") tmpadj)))
(dname (conc "/tmp/"(current-user-name)"/megatest_localdb/" area "/" (string-translate areapath "/" ".") tmpadj "/.mtdb")))
(unless (directory-exists? dname)
(create-directory dname #t))
dname))
(define (dbfile:run-id->path apath run-id)
(conc apath"/"(dbfile:run-id->dbname run-id)))
|
︙ | | |
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
|
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
|
-
+
|
;; opens and returns handle and nothing else
;;
;; NOTE: this is already protected by mutex *no-sync-db-mutex*
;;
(define (dbfile:raw-open-no-sync-db dbpath)
(if (not (file-exists? dbpath))
(create-directory dbpath #t))
(debug:print-info 0 *default-log-port* "Opening "dbpath"/no-sync.db")
(debug:print-info 2 *default-log-port* "(dbfile:raw-open-no-sync-db: Opening "dbpath"/no-sync.db")
(let* ((dbname (conc dbpath "/no-sync.db"))
(db-exists (file-exists? dbname))
(init-proc (lambda (db)
(sqlite3:with-transaction
db
(lambda ()
;; I have been having trouble with init of no-sync.db so
|
︙ | | |
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
|
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
|
-
-
+
+
+
-
-
-
-
+
+
+
+
|
dbname TEXT,
mtversion TEXT,
reason TEXT DEFAULT 'none',
CONSTRAINT no_sync_processes UNIQUE (host,pid));"
))))))
(on-tmp (equal? (car (string-split dbpath "/")) "tmp"))
(db (if on-tmp
(dbfile:cautious-open-database dbname init-proc 0 "WAL" force-init: #t)
(dbfile:cautious-open-database dbname init-proc 0 #f force-init: #t)
(dbfile:cautious-open-database dbname init-proc 1 "WAL" force-init: #t) ;; WAL MODE should use syncronous=1
;; (dbfile:cautious-open-database dbname init-proc 0 #f force-init: #t)
(dbfile:cautious-open-database dbname init-proc 0 "MEMORY" force-init: #t) ;; Journal mode = memory is fastest?
;; (sqlite3:open-database dbname)
)))
(if on-tmp ;; done in cautious-open-database
(begin
(sqlite3:execute db "PRAGMA synchronous = 0;")
(sqlite3:set-busy-handler! db (sqlite3:make-busy-timeout 136000))))
;; (if on-tmp ;; done in cautious-open-database
;; (begin
;; (sqlite3:execute db "PRAGMA synchronous = 0;") ;; why was this here when is is handled by cautious-open-database?
(sqlite3:set-busy-handler! db (sqlite3:make-busy-timeout 136000)) ;; ))
db))
;; mtest processes registry calls
(define (dbfile:insert-or-update-process nsdb dat)
(let* ((host (procinf-host dat))
(pid (procinf-pid dat))
|
︙ | | |
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
|
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
|
+
+
-
+
+
+
+
+
+
+
+
+
-
+
-
+
|
(define (dbfile:register-process nsdb host port pid starttime endtime status purpose dbname mtversion)
(sqlite3:execute nsdb "INSERT INTO processes (host,port,pid,starttime,endtime,status,purpose,dbname,mtversion) VALUES (?,?,?,?,?,?,?,?,?);"
host port pid starttime endtime status purpose dbname mtversion))
(define (dbfile:set-process-status nsdb host pid newstatus)
(sqlite3:execute nsdb "UPDATE processes SET status=? WHERE host=? AND pid=?;" newstatus host pid))
;; as sorted should be stable. can use to choose "winner"
;;
(define (dbfile:get-process-options nsdb purpose dbname)
(sqlite3:fold-row
;; host port pid starttime status mtversion
(lambda (res . row)
(cons row res))
'()
nsdb
"SELECT host,port,pid,starttime,endtime,status,mtversion FROM processes WHERE purpose=? AND dbname LIKE ? AND status='alive';"
"SELECT host,port,pid,starttime,endtime,status,mtversion FROM processes WHERE purpose=? AND dbname LIKE ? AND status IN ('running','alive') ORDER BY starttime ASC,host,port;"
purpose dbname))
(define (dbfile:get-process-info nsdb host pid)
(let ((res (sqlite3:fold-row
;; host port pid starttime status mtversion
(lambda (res . row)
(cons row res))
'()
nsdb
"SELECT host,port,pid,starttime,endtime,status,purpose,dbname,mtversion FROM processes WHERE host=? AND pid=?;"
host pid)))
(if (null? res)
#f
(car res))))
(define (dbfile:row->procinf row)
(match row
((host port pid starttime endtime status mtversion)
(make-procinf host: host port: port pid: pid starttime: starttime endtime: endtime status: status mtversion: mtversion))
(else
(debug:print 0 *default-log-port* "ERROR: row "row" did not match host,port,pid,starttime,endtime,status,mtversion")
#f)))
(define (dbfile:set-process-done nsdb host pid reason)
(sqlite3:execute nsdb "UPDATE processes SET status='ended',endtime=?,reason=? WHERE host=? AND pid=?;" (current-seconds) reason host pid)
(sqlite3:execute nsdb "UPDATE processes SET status='done',endtime=?,reason=? WHERE host=? AND pid=?;" (current-seconds) reason host pid)
(dbfile:cleanup-old-entries nsdb))
(define (dbfile:cleanup-old-entries nsdb)
(sqlite3:execute nsdb "DELETE FROM process WHERE status='ended' AND endtime<?;" (- (current-seconds) (* 3600 48))))
(sqlite3:execute nsdb "DELETE FROM process WHERE status='done' AND endtime<?;" (- (current-seconds) (* 3600 48))))
;; other no-sync functions
(define (dbfile:with-no-sync-db dbpath proc)
(mutex-lock! *no-sync-db-mutex*)
(let* ((already-open *no-sync-db*)
(db (or already-open (dbfile:raw-open-no-sync-db dbpath)))
|
︙ | | |
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
|
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
|
+
+
+
+
-
+
+
+
|
;; transaction protected lock aquisition
;; either:
;; fails returns (#f lock-creation-time identifier)
;; succeeds (returns (#t lock-creation-time identifier)
;; use (db:no-sync-del! db keyname) to release the lock
;;
(define (db:no-sync-get-lock-with-id db keyname identifier)
(debug:print 2 *default-log-port* "db:no-sync-get-lock-with-id: db: " db " keyname: " keyname " identifier: " identifier)
(sqlite3:with-transaction
db
(lambda ()
(condition-case
(let* ((curr-val (db:no-sync-get/default db keyname #f)))
(debug:print 2 *default-log-port* "db:no-sync-get-lock-with-id: curr-val: " curr-val)
(if curr-val
(match (db:extract-time-identifier curr-val) ;; result->timestamp, identifier
((timestamp . ident)
(cons (equal? ident identifier) timestamp))
(else
(debug:print 2 *default-log-port* "db:no-sync-get-lock-with-id: malformed lock")
(else (cons #f 'malformed-lock))) ;; lock malformed
(cons #f 'malformed-lock)
)
) ;; lock malformed
(let ((curr-sec (current-seconds))
(lock-value (if identifier
(conc (current-seconds)"+"identifier)
(current-seconds))))
(sqlite3:execute db "INSERT OR REPLACE INTO no_sync_metadat (var,val) VALUES(?,?);" keyname lock-value)
(cons #t curr-sec))))
(exn (io-error) (dbfile:print-err "ERROR: i/o error with no-sync db. Check permissions, disk space etc. and try again."))
|
︙ | | |
1570
1571
1572
1573
1574
1575
1576
1577
1578
|
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
|
+
+
+
+
+
+
+
-
+
+
|
(result (or stmth
(let* ((newstmth (sqlite3:prepare db stmt)))
;; (db:hoh-set! stmt-cache db stmt newstmth)
(hash-table-set! stmt-cache stmt newstmth)
newstmth))))
(mutex-unlock! *get-cache-stmth-mutex*)
result))
;; (define *mutex-stmth-call* (make-mutex))
;;
;; (define (db:with-mutex-for-stmth proc)
;; (mutex-lock! *mutex-stmth-call*)
;; (let* ((res (proc)))
;; (mutex-unlock! *mutex-stmth-call*)
;; res))
)
)
|
︙ | | |
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
+
|
(import scheme
chicken
data-structures
extras
files
format
(prefix sqlite3 sqlite3:)
matchable
posix
typed-records
srfi-1
srfi-18
srfi-69
|
︙ | | |
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
-
-
-
+
+
+
-
-
+
+
|
;;======================================================================
;; The cachedb one-db file per server method goes in here
;;======================================================================
;; NOTE: the r/w is now w/r, #t=db modified by query, #f=db NOT modified by query
(define (dbmod:with-db dbstruct run-id w/r proc params)
(let* ((use-mutex (or (and w/r ;; use the mutex on queries that modify the db and for sync to disk
(> *api-process-request-count* 5)) ;; when writes are happening throttle more
(> *api-process-request-count* 50)))
(let* ((use-mutex w/r) ;; (or (and w/r ;; use the mutex on queries that modify the db and for sync to disk
;; (> *api-process-request-count* 5)) ;; when writes are happening throttle more
;; (> *api-process-request-count* 50)))
(dbdat (dbmod:open-db dbstruct run-id (dbfile:db-init-proc)))
(dbh (dbr:dbdat-dbh dbdat)) ;; this will be the cachedb handle
(dbfile (dbr:dbdat-dbfile dbdat)))
;; if nfs mode do a sync if delta > 2
(let* ((last-update (dbr:dbstruct-last-update dbstruct))
(sync-proc (dbr:dbstruct-sync-proc dbstruct))
#;(let* ((last-update (dbr:dbstruct-last-update dbstruct))
;; (sync-proc (dbr:dbstruct-sync-proc dbstruct))
(curr-secs (current-seconds)))
(if (> (- curr-secs last-update) 5)
(begin
(sync-proc last-update)
;; MOVE THIS CALL TO INSIDE THE sync-proc CALL
(dbr:dbstruct-last-update-set! dbstruct curr-secs)
|
︙ | | |
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
|
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
-
+
|
(tmpadj "") ;; add to tmp path
(syncdir 'todisk)) ;; todisk is used when caching in /tmp and writing data back to MTRAH
(let* ((dbstruct (or dbstruct-in (make-dbr:dbstruct areapath: areapath)))
(dbfname (or dbfname-in (dbmod:run-id->dbfname run-id)))
(dbpath (dbmod:get-dbdir dbstruct)) ;; directory where all the .db files are kept
(dbfullname (conc dbpath"/"dbfname)) ;; (dbmod:run-id->full-dbfname dbstruct run-id))
(dbexists (file-exists? dbfullname))
(tmpdir (dbfile:make-tmpdir-name areapath tmpadj))
(tmpdir (common:make-tmpdir-name areapath tmpadj))
(tmpdb (let* ((fname (conc tmpdir"/"dbfname)))
fname))
(cachedb (dbmod:open-cachedb-db init-proc
;; (if (eq? (dbfile:cache-method) 'cachedb)
;; #f
tmpdb
;; )
|
︙ | | |
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
|
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-
-
-
-
-
+
+
+
+
+
+
+
-
-
+
+
+
-
-
-
-
|
(dbr:dbstruct-dbfile-set! dbstruct dbfullname)
(dbr:dbstruct-dbtmpname-set! dbstruct tmpdb)
(dbr:dbstruct-dbfname-set! dbstruct dbfname)
(dbr:dbstruct-sync-proc-set! dbstruct
(lambda (last-update)
(if *sync-in-progress*
(debug:print 3 *default-log-port* "WARNING: overlapping calls to sync to disk")
(let* ((syncer-logfile (conc areapath"/logs/"dbfname"-syncer.log"))
(sync-cmd (if (eq? syncdir 'todisk)
(conc "(NBFAKE_LOG="syncer-logfile" nbfake megatest -db2db -from "tmpdb" -to "dbfullname" -period 5 -timeout 10 > /dev/null 2&>1)&")
(conc "(NBFAKE_LOG="syncer-logfile" nbfake megatest -db2db -from "dbfullname" -to "tmpdb" -period 5 -timeout 10 > /dev/null 2&>1)&")))
(synclock-file (conc dbfullname".lock"))
(syncer-running-file (conc dbfullname"-sync-running"))
(synclock-mod-time (if (file-exists? synclock-file)
(handle-exceptions
exn
#f
(file-modification-time synclock-file))
#f))
(thethread (lambda ()
(thread-start!
(make-thread
(lambda ()
(set! *sync-in-progress* #t)
(debug:print-info "Running "sync-cmd)
(if (file-exists? syncer-running-file)
(debug:print-info 0 *default-log-port* "Syncer still running, skipping syncer start.")
(system sync-cmd))
(set! *sync-in-progress* #f)))))))
(if ((if (eq? syncdir 'todisk) < >) ;; use less than for todisk, greater than for from disk
(file-modification-time tmpdb)
(file-modification-time dbfullname))
(debug:print 4 *default-log-port* "Skipping sync, "tmpdb" older than "dbfullname)
(if synclock-mod-time
(if (> (- (current-seconds) synclock-mod-time) 20) ;; something wrong with sync, remove file
(begin
(begin
(handle-exceptions
exn
#f
(begin
(debug:print 0 *default-log-port* "Sync lock file " synclock-file "is older than 20 seconds (" synclock-mod-time " seconds). Removing it")
;; turn off writes - send busy or block?
;; call db2db internally
;; turn writes back on
;;
(set! *api-halt-writes* #t) ;; do we need a mutex?
;; (dbmod:db-to-db-sync src-db dest-db last-update (dbfile:db-init-proc) keys)
(debug:print-info 2 *default-log-port* "Internal sync running from "tmpdb" to "dbfullname)
(delete-file synclock-file)
)
(dbmod:db-to-db-sync tmpdb dbfullname last-update (dbfile:db-init-proc) keys)
(set! *api-halt-writes* #f)
))))
)
(thethread))
(debug:print 0 *default-log-port* "Skipping sync, lockfile "synclock-file" found."))
(thethread)))))))
;; (dbmod:sync-tables tables #f db cachedb)
;;
(thread-sleep! 1) ;; let things settle before syncing in needed data
(dbmod:sync-gasket tables #f cachedb db dbfullname 'fromdest keys) ;; ) ;; load into cachedb
(dbr:dbstruct-last-update-set! dbstruct (+ (current-seconds) -10)) ;; should this be offset back in time by one second?
dbstruct))
|
︙ | | |
854
855
856
857
858
859
860
861
|
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
|
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
(sdb (dbmod:safely-open-db src-db init-proc #t))
(ddb (dbmod:safely-open-db dest-db init-proc d-wr))
(res (dbmod:sync-gasket tables last-update sdb ddb dest-db 'todisk keys)))
(sqlite3:finalize! sdb)
(sqlite3:finalize! ddb)
res)))
#f))
)
;; ======================================================================
;; dbstats
;;======================================================================
;; (define *dbstruct-dbs* #f) ;; used to cache the dbstruct in db:setup. Goal is to remove this.
;; db stats
(define *db-stats* (make-hash-table)) ;; hash of vectors < count duration-total >
(define *db-stats-mutex* (make-mutex))
(define (rmt:print-db-stats)
(let ((fmtstr "~40a~7-d~9-d~20,2-f")) ;; "~20,2-f"
(debug:print 0 *default-log-port* "DB Stats\n========")
(debug:print 0 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg"))
(for-each (lambda (cmd)
(let* ((dat (hash-table-ref *db-stats* cmd))
(count (dbstat-cnt dat))
(tottime (dbstat-tottime dat)))
(debug:print 0 *default-log-port*
(format #f fmtstr cmd count tottime
(/ tottime count)))))
(sort (hash-table-keys *db-stats*)
(lambda (a b)
(> (dbstat-tottime (hash-table-ref *db-stats* a))
(dbstat-tottime (hash-table-ref *db-stats* b))))))))
(defstruct dbstat
(cnt 0)
(tottime 0))
(define (db:add-stats cmd run-id params delta)
(let* ((modified-cmd (if (eq? cmd 'general-call)
(string->symbol (conc "general-call-" (car params)))
cmd))
(rec (hash-table-ref/default *db-stats* modified-cmd #f)))
(if (not rec)
(let ((new-rec (make-dbstat)))
(hash-table-set! *db-stats* modified-cmd new-rec)
(set! rec new-rec)))
(dbstat-cnt-set! rec (+ (dbstat-cnt rec) 1))
(dbstat-tottime-set! rec (+ (dbstat-tottime rec) delta))))
)
;; ATTIC
#;(let* ((syncer-logfile (conc areapath"/logs/"dbfname"-syncer.log"))
(sync-cmd (if (eq? syncdir 'todisk)
(conc "(NBFAKE_LOG="syncer-logfile" nbfake megatest -db2db -from "tmpdb" -to "dbfullname" -period 5 -timeout 10 > /dev/null 2&>1)&")
(conc "(NBFAKE_LOG="syncer-logfile" nbfake megatest -db2db -from "dbfullname" -to "tmpdb" -period 5 -timeout 10 > /dev/null 2&>1)&")))
(synclock-file (conc dbfullname".lock"))
(syncer-running-file (conc dbfullname"-sync-running"))
(synclock-mod-time (if (file-exists? synclock-file)
(handle-exceptions
exn
#f
(file-modification-time synclock-file))
#f))
(thethread (lambda ()
(thread-start!
(make-thread
(lambda ()
(set! *sync-in-progress* #t)
(debug:print-info "Running "sync-cmd)
(if (file-exists? syncer-running-file)
(debug:print-info 0 *default-log-port* "Syncer still running, skipping syncer start.")
(system sync-cmd))
(set! *sync-in-progress* #f)))))))
(if ((if (eq? syncdir 'todisk) < >) ;; use less than for todisk, greater than for from disk
(file-modification-time tmpdb)
(file-modification-time dbfullname))
(debug:print 4 *default-log-port* "Skipping sync, "tmpdb" older than "dbfullname)
(if synclock-mod-time
(if (> (- (current-seconds) synclock-mod-time) 20) ;; something wrong with sync, remove file
(begin
(handle-exceptions
exn
#f
(begin
(debug:print 0 *default-log-port* "Sync lock file " synclock-file "is older than 20 seconds (" synclock-mod-time " seconds). Removing it")
(delete-file synclock-file)
)
)
(thethread))
(debug:print 0 *default-log-port* "Skipping sync, lockfile "synclock-file" found."))
(thethread))))
|
︙ | | |
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
-
-
+
-
-
-
+
-
-
-
+
-
-
+
-
-
-
-
-
+
-
-
-
-
-
-
-
-
-
-
+
-
-
-
+
+
-
-
-
-
+
+
+
+
-
-
-
+
-
-
+
+
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
|
((tcp) (tt:make-remote areapath))
(else #f)))
;;======================================================================
(define *send-receive-mutex* (make-mutex)) ;; should have separate mutex per run-id
;; RA => e.g. usage (rmt:send-receive 'get-var #f (list varname))
;;
;; how to make area-dat
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f)) ;; start attemptnum at 1 so the modulo below works as expected
(assert *toppath* "FATAL: rmt:send-receive called with *toppath* not set.")
(define (rmt:set-ttdat areapath ttdat)
(if (not (eq? (rmt:transport-mode) 'nfs))
(begin
(if (> attemptnum 2)
(if ttdat
(debug:print 0 *default-log-port* "INFO: attemptnum in rmt:send-receive is " attemptnum))
ttdat
(cond
((> attemptnum 2) (thread-sleep! 0.05))
((> attemptnum 10) (thread-sleep! 0.5))
((> attemptnum 20) (thread-sleep! 1)))
(let* ((newremote (make-and-init-remote areapath)))
;; I'm turning this off, it may make sense to move it
;; into http-transport-handler
(if (and (> attemptnum 5) (= 0 (modulo attemptnum 15)))
(begin
(debug:print 0 *default-log-port* "ERROR: can't connect to server, trying to start a server.")
(case (rmt:transport-mode)
((http)
(server:run *toppath*)
(thread-sleep! 3))
(else
(set! *ttdat* newremote)
(thread-sleep! 1) ;; for tcp the server is started by routines in tcp-transportmod. For nfs there is no server
))))))
newremote)))
;; 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
;;
;; NB// area-dat replaced by ttdat
;;
(define (rmt:send-receive cmd run-id params #!key (attemptnum 1)(ttdat #f))
(assert *toppath* "FATAL: rmt:send-receive called with *toppath* not set.")
(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*))
(readonly-mode (rmtmod:calc-ro-mode ttdat *toppath*))
(testsuite (common:get-testsuite-name))
(mtexe (common:find-local-megatest)))
(mtexe (common:find-local-megatest))
(dbfname (conc (dbfile:run-id->dbnum run-id)".db"))
(case (rmt:transport-mode)
((http)(http-transport-handler runremote cmd rid params attemptnum area-dat areapath readonly-mode))
((tcp) (tcp-transport-handler runremote cmd rid params attemptnum area-dat areapath readonly-mode testsuite mtexe))
((nfs) (nfs-transport-handler runremote cmd rid params attemptnum area-dat areapath readonly-mode testsuite mtexe))
)))
(ttdat (rmt:set-ttdat areapath ttdat))
(conn (tt:get-conn ttdat dbfname))
(is-main (equal? dbfname "main.db")) ;; why not (not run-id) ?
(server-start-proc (if is-main
#f
(lambda ()
;; (debug:print-info 0 *default-log-port* "starting server for dbfname: "dbfname)
(rmt:start-server ;; tt:server-process-run
areapath
testsuite ;; (dbfile:testsuite-name)
mtexe
run-id)))))
;; here we look at ttdat, if dbfname is NOT main.db we check that a conn exists for it
;; and if there is no conn we first send a request to the main.db server to start a
;; server for the dbfname.
#;(if (and (not is-main)(not conn)) ;; no existing connection to non-main server, call in a start up request
(begin
(server-start-proc)
(thread-sleep! 1)))
(tt:handler ttdat cmd run-id params attemptnum readonly-mode dbfname testsuite mtexe server-start-proc)))
;; KEEP THIS HERE FOR A WHILE, WE MAY WANT TO RESURECT IT
(define (nfs-transport-handler runremote cmd run-id params attemptnum area-dat areapath readonly-mode testsuite mtexe)
(let* ((keys (common:get-fields *configdat*))
(dbstruct (dbmod:nfs-get-dbstruct run-id keys (dbfile:db-init-proc) areapath tmpadj: "/dashboard")))
(api:dispatch-request dbstruct cmd run-id params)))
;; (define (nfs-transport-handler runremote cmd run-id params attemptnum area-dat areapath readonly-mode testsuite mtexe)
;; (let* ((keys (common:get-fields *configdat*))
;; (dbstruct (dbmod:nfs-get-dbstruct run-id keys (dbfile:db-init-proc) areapath tmpadj: "/dashboard")))
;; (api:dispatch-request dbstruct cmd run-id params)))
(define (tcp-transport-handler runremote cmd run-id params attemptnum area-dat areapath readonly-mode testsuite mtexe)
(if (not runremote)
(let* ((newremote (make-and-init-remote areapath)))
(set! *runremote* newremote)
(set! runremote newremote)))
(let* ((dbfname (conc (dbfile:run-id->dbnum run-id)".db"))) ;;(dbfile:run-id->path areapath run-id)))
(tt:handler runremote cmd run-id params attemptnum area-dat areapath readonly-mode dbfname testsuite mtexe)))
(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)
|
︙ | | |
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
-
+
|
(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 (db:dbfile-path)) ;; 0))
(db-file-path (common:make-tmpdir-name *toppath* "")) ;; 0))
(dbstructs-local (db:setup #t))
(read-only (not (file-write-access? db-file-path)))
(start (current-milliseconds))
(resdat (if (not (and read-only qry-is-write))
(let ((v (api:execute-requests dbstructs-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
|
︙ | | |
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
|
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
|
-
+
-
-
+
+
-
+
-
+
|
#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)
(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)))
(define (rmt:start-server areapath testsuite mtexe run-id) ;; run on main.db server
(rmt:send-receive 'start-server #f (list areapath testsuite mtexe run-id)))
;;======================================================================
;; M I S C
;;======================================================================
(define (rmt:login run-id)
(rmt:send-receive 'login run-id (list *toppath* megatest-version (client:get-signature))))
;; 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 runremote)
;; (rmt:send-receive-no-auto-client-setup runremote 'login 0 (list *toppath* megatest-version (client:get-signature))))
;; (rmt:send-receive-no-auto-client-setup runremote 'login #f (list *toppath* megatest-version (client:get-signature))))
;; 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)))
(rmt:send-receive 'get-latest-host-load #f (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)
|
︙ | | |
︙ | | |
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
|
+
-
+
+
+
+
-
+
+
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
+
-
+
-
+
+
+
-
+
-
+
+
-
+
-
+
-
+
-
+
+
+
+
-
+
|
;; parameters
;;
(define tt-server-timeout-param (make-parameter 600))
;; make ttdat visible
(define *server-info* #f)
(define *server-run* #t)
(define (tt:make-remote areapath)
(make-tt areapath: areapath))
;; 1 ... or #f
;; and check that dbfname matches. FIXME: the propagation of dbfname and run-id
;; might not make the best sense
;;
(define (tt:valid-run-id run-id dbfname)
(and (or (number? run-id)
(not run-id))
(equal? (dbfile:run-id->dbfname run-id) dbfname)))
(tcp-buffer-size 2048)
;; (max-connections 4096)
;; (max-connections 4096)
(define (tt:get-conn ttdat dbfname)
(hash-table-ref/default (tt-conns ttdat) dbfname #f))
;; do all the busy work of finding and setting up conn for
;; connecting to a server
;;
(define (tt:client-connect-to-server ttdat dbfname run-id testsuite)
(define (tt:client-connect-to-server ttdat dbfname run-id testsuite server-start-proc)
(assert (tt:valid-run-id run-id dbfname) "FATAL: invalid run-id "run-id)
(debug:print-info 2 *default-log-port* "tt:client-connect-to-server " dbfname " " run-id)
(let* ((conn (hash-table-ref/default (tt-conns ttdat) dbfname #f))
(server-start-proc (lambda ()
(tt:server-process-run
(tt-areapath ttdat)
testsuite ;; (dbfile:testsuite-name)
(common:find-local-megatest)
run-id))))
(let* ((conn (tt:get-conn ttdat dbfname))
(server-start-proc (or server-start-proc
(lambda ()
(assert (equal? dbfname "main.db") ;; only main.db is started here
"FATAL: called server-start-proc for db other than main.db")
(tt:server-process-run
(tt-areapath ttdat)
testsuite ;; (dbfile:testsuite-name)
(common:find-local-megatest)
run-id)))))
(if conn
(begin
; (debug:print-info 0 *default-log-port* "already connected to the server")
(debug:print-info 2 *default-log-port* "already connected to a server")
conn) ;; we are already connected to the server
(let* ((sdat (tt:get-current-server-info ttdat dbfname)))
(match sdat
((host port start-time server-id pid dbfname2 servinffile)
(assert (equal? dbfname dbfname2) "FATAL: read server info from wrong file.")
;(debug:print-info 0 *default-log-port* "in match servinffile:" servinffile)
(debug:print-info 2 *default-log-port* "no conn - in match servinffile:" servinffile)
(let* ((host-port (conc host":"port))
(conn (make-tt-conn
host: host
port: port
host-port: host-port
dbfname: dbfname
servinf-file: servinffile
server-id: server-id
server-start: start-time
pid: pid)))
;; verify we can talk to this server
(let* ((result (tt:timed-ping host port server-id))
(ping-res (car result))
(ping (cdr result)))
(debug:print-info 0 *default-log-port* "ping time: " ping)
(debug:print-info 2 *default-log-port* "host " host " port " port " ping time: " ping " result " ping-res)
(case ping-res
((running)
(debug:print-info 2 *default-log-port* "Setting conn = " conn " in hash table")
(hash-table-set! (tt-conns ttdat) dbfname conn) ;;; is this ok to save before validating that the connection is good?
conn)
((starting)
(thread-sleep! 0.5)
(debug:print-info 0 *default-log-port* "server for " dbfname " is in starting state, retrying connect")
(tt:client-connect-to-server ttdat dbfname run-id testsuite))
(tt:client-connect-to-server ttdat dbfname run-id testsuite server-start-proc))
(else
(let* ((curr-secs (current-seconds)))
;; rm the (last server) would go here
(if (> (- curr-secs (tt-last-serv-start ttdat)) 10)
(begin
(tt-last-serv-start-set! ttdat curr-secs)
(server-start-proc))) ;; start server if 30 sec since last attempt
(server-start-proc))) ;; start server if 10 sec since last attempt
(thread-sleep! 1)
(debug:print-info 2 *default-log-port* "server ping result was neither running nor starting. Retrying connect")
(tt:client-connect-to-server ttdat dbfname run-id testsuite)))))))
(tt:client-connect-to-server ttdat dbfname run-id testsuite server-start-proc)))))))
(else ;; no good server found, if haven't started server in > 5 secs, start another
(if (> (- (current-seconds) (tt-last-serv-start ttdat)) 5) ;; BUG - grow this number really do not want to swamp the machine with servers
(if (> (- (current-seconds) (tt-last-serv-start ttdat)) 3) ;; BUG - grow this number really do not want to swamp the machine with servers
(begin
(debug:print-info 0 *default-log-port* "No server found. Starting one for run-id "run-id" in dbfile "dbfname)
(debug:print-info 0 *default-log-port* "Starting server for "dbfname)
(server-start-proc)
(tt-last-serv-start-set! ttdat (current-seconds))))
(tt-last-serv-start-set! ttdat (current-seconds))
(thread-sleep! 3)
))
(thread-sleep! 1)
(debug:print-info 0 *default-log-port* "Connect to server for " dbfname)
(tt:client-connect-to-server ttdat dbfname run-id testsuite)))))))
(tt:client-connect-to-server ttdat dbfname run-id testsuite server-start-proc)))))))
(define (tt:timed-ping host port server-id)
(let* ((start-time (current-milliseconds))
(result (tt:ping host port server-id)))
(cons result (- (current-milliseconds) start-time))))
|
︙ | | |
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
|
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
|
-
-
-
+
+
+
+
-
-
-
+
+
+
-
+
-
+
-
+
-
+
-
+
-
+
-
+
-
+
-
+
-
+
-
+
|
;; (debug:print 0 *default-log-port* "res not in form (status errmsg result meta), got: "res)
(try-again)))))
;; client side handler
;;
;;(tt:handler #<tt> get-keys #f () 2 #f "/home/matt/data/megatest/ext-tests" #f "main.db" "ext-tests" "/home/matt/data/megatest/bin/.22.04/../megatest")
;;
(define (tt:handler ttdat cmd run-id params attemptnum area-dat areapath readonly-mode dbfname testsuite mtexe)
;; NOTE: areapath is passed in and in tt struct. We'll use passed in value for now.
(let* ((conn (tt:client-connect-to-server ttdat dbfname run-id testsuite))) ;; (hash-table-ref/default (tt-conns ttdat) dbfname #f)))
(define (tt:handler ttdat cmd run-id params attemptnum readonly-mode dbfname testsuite mtexe server-start-proc)
;; connect-to-server will start a server if needed.
(let* ((areapath (tt-areapath ttdat))
(conn (tt:client-connect-to-server ttdat dbfname run-id testsuite server-start-proc))) ;; looks up conn keyed by dbfname
(if conn
;; have connection, call the server
(let* ((res (tt:send-receive ttdat conn cmd run-id params)))
;; res is (status errmsg result meta)
; (debug:print 0 *default-log-port* "conn:" conn " res: " res)
(match res
((status errmsg result meta)
(if (list? meta)
(let* ((delay-wait (alist-ref 'delay-wait meta)))
(if (and (number? delay-wait)
(> delay-wait 0))
(begin
(debug:print 0 *default-log-port* "Server is loaded, delaying "delay-wait" seconds")
(thread-sleep! delay-wait)))))
(case status
((busy) ;; result will be how long the server wants you to delay
(let* ((dly (if (number? result) result 0.1)))
(debug:print 0 *default-log-port* "WARNING: server for "dbfname" is busy, will try again in "dly" seconds.")
(let* ((raw-dly (if (number? result) result 0.1))
(dly (+ raw-dly (/ attemptnum 10)))) ;; (* raw-dly (/ attemptnum 2))))
(debug:print 0 *default-log-port* "WARNING: server for "dbfname" is busy, cmd is "cmd", will try again in "dly" seconds. This is attempt "(- attemptnum 1))
(thread-sleep! dly)
(tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe)))
(tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe server-start-proc)))
((loaded)
(debug:print 0 *default-log-port* "WARNING: server for "dbfname" is loaded, slowing queries.")
(tt:backoff-incr (tt-conn-host conn)(tt-conn-port conn))
result) ;; (tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe))
result) ;; (tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe))
(else
result)))
(else ;; did not receive properly formated result
(if (not res) ;; tt:handler is telling us that communication failed
(if (not res) ;; tt:send-receive telling us that communication failed
(let* ((host (tt-conn-host conn))
(port (tt-conn-port conn))
;; (dbfname (tt-conn-port conn)) ;; 192.168.0.127:4242-726924:4.db
(pid (tt-conn-pid conn))
;;(servinf (tt-conn-servinf-file conn)))
(servinf (tt-servinf-file ttdat))) ;; (conc areapath"/.servinfo/"host":"port"-"pid":"dbfname))) ;; TODO, use (server:get-servinfo-dir areapath)
(hash-table-set! (tt-conns ttdat) dbfname #f)
(hash-table-set! (tt-conns ttdat) dbfname #f) ;; clear out the conn for this dbfname to force finding new server
(if (and servinf (file-exists? servinf))
(begin
(if (< attemptnum 10)
(begin
(thread-sleep! 0.5)
(tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe))
(tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe server-start-proc))
(begin
(debug:print 0 *default-log-port* "INFO: no response from server "host":"port" for "dbfname)
(if (and (file-exists? servinf)
(> (- (current-seconds)(file-modification-time servinf)) 60))
(begin
(debug:print 0 *default-log-port* "INFO: "servinf" file seems old and no ping response, removing it.")
(handle-exceptions
exn
#f
(delete-file* servinf))
(tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe))
(tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe server-start-proc))
(begin
;; start server - addressed in client-connect-to-server
;; delay - addressed in client-connect-to-server
;; try again
(thread-sleep! 0.25) ;; dunno, I think this needs to be here
(tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe))
(tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe server-start-proc))
))))
(begin ;; no server file, delay and try again
(debug:print 0 *default-log-port* "INFO: connection to server "host":"port" broken for "dbfname", but do not see servinf file "servinf)
(debug:print 2 *default-log-port* "INFO: connection to server "host":"port" broken for "dbfname", no servinf file. Server exited? ")
(thread-sleep! 0.5)
(tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe))))
(tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe server-start-proc))))
(begin ;; this case is where res is malformed. Probably should abort
(assert #f "FATAL: tt:handler received bad data "res)
;; (debug:print 0 *default-log-port* "INFO: got corrupt data from server "host":"port", "res", for "dbfname", will try again.")
;; (tt:handler ttdat cmd run-id params (+ attemptnum 1) area-dat areapath readonly-mode dbfname testsuite mtexe)
;; (tt:handler ttdat cmd run-id params (+ attemptnum 1) readonly-mode dbfname testsuite mtexe)
)))))
(begin
(thread-sleep! 1) ;; no conn yet set up, give it a rest and try again
(tt:handler ttdat cmd run-id params attemptnum area-dat areapath readonly-mode dbfname testsuite mtexe)))))
(tt:handler ttdat cmd run-id params attemptnum readonly-mode dbfname testsuite mtexe server-start-proc)))))
(define (tt:bid-for-servership run-id)
#f)
;; gets server info and appends path to server file
;; sorts by age, oldest first
;;
|
︙ | | |
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
|
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
|
+
+
-
+
|
;; This is the routine called in megatest.scm to start a server
;;
;; Server viability is checked in keep-running. Blindly start and run here.
;;
(define (tt:start-server areapath run-id dbfname-in handler keys)
(assert areapath "FATAL: areapath not provided for tt:start-server")
;; is there already a server for this dbfile? Then exit.
(debug:print 2 *default-log-port* "tt:start-server: " dbfname-in)
(let* ((ttdat (make-tt areapath: areapath))
(dbfname (or dbfname-in (dbmod:run-id->dbfname run-id)))
(servers (tt:find-server areapath dbfname))) ;; should use tt:get-current-server-info instead
(debug:print 0 *default-log-port* "Found " (length servers) " already running for " dbfname)
(if (> (length servers) 4)
(if (> (length servers) 0)
(begin
(debug:print 0 *default-log-port* "INFO: found server(s) already running for db "dbfname", "(string-intersperse servers ",")" Exiting.")
(exit))
(let* ((dbstruct (dbmod:open-dbmoddb areapath run-id dbfname (dbfile:db-init-proc) keys)))
(tt-handler-set! ttdat (handler dbstruct))
(let* ((tcp-thread (make-thread
(lambda ()
|
︙ | | |
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
|
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
|
-
+
-
+
+
-
+
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
|
(begin
(procinf-port-set! *procinf* (tt-port ttdat))
(procinf-dbname-set! *procinf* dbfname)
(dbfile:with-no-sync-db
nosyncdbpath
(lambda (nsdb)
(dbfile:insert-or-update-process nsdb *procinf*))))
(if (< count 5)
(if (< count 10)
(begin
(thread-sleep! 0.5)
(thread-sleep! 0.25)
(loop (+ count 1)))
(debug:print 0 *default-log-port* "ERROR: (tt-port ttdat) no port set!"))))
(thread-join! run-thread) ;; run thread will exit on timeout or other conditions
;; replace with call to (dbfile:set-process-done nsdb host pid reason)
(procinf-status-set! *procinf* "done")
(procinf-end-set! *procinf* (current-seconds))
;; either convert this to use set-process-done or get rid of set-process-done
(dbfile:with-no-sync-db
nosyncdbpath
(lambda (nsdb)
(dbfile:insert-or-update-process nsdb *procinf*)))
(debug:print 0 *default-log-port* "Exiting now.")
(exit)))))))
(define (tt:keep-running ttdat dbfname dbstruct)
;; verfiy conn for ready
;; listener socket has been started by this stage
;; wait for a port before creating the registration file
;;
(let* ((db-locked-in #f)
(areapath (tt-areapath ttdat))
(nosyncdbpath (conc areapath"/.mtdb"))
(nosyncdbpath (conc areapath"/.mtdb")))
(cleanup (lambda ()
(if (tt-cleanup-proc ttdat)
((tt-cleanup-proc ttdat)))
(dbfile:with-no-sync-db nosyncdbpath
(lambda (db)
(let* ((dbtmpname (dbr:dbstruct-dbtmpname dbstruct)))
(debug:print-info 0 *default-log-port* "Running clean up, including removing db file "dbtmpname)
(db:no-sync-del! db dbfname)
#;(if dbtmpname
(delete-file dbtmpname))))))))
(set! *server-info* ttdat)
(let loop ((count 0))
(if (> count 240)
(begin
(debug:print 0 *default-log-port* "FATAL: Could not start a tcp server, giving up.")
(exit 1))
(if (not (tt-port ttdat)) ;; no connection yet
(begin
(thread-sleep! 0.25)
(loop (+ count 1))))))
(tt:create-server-registration-file ttdat dbfname)
;; now start watching the last-access, if it hasn't been touched
;; in over ten seconds we exit
(thread-sleep! 0.05) ;; any real need for delay here?
(let loop ()
(let* ((servers (tt:get-server-info-sorted ttdat dbfname))
(ok (cond
((not *server-run*)
(debug:print 0 *default-log-port* "WARNING: received a stop server from client by remote request.")
#f)
((null? servers) #f) ;; not ok
((equal? (list-ref (car servers) 6) ;; compare the servinfofile
(tt-servinf-file ttdat))
(let* ((res (if db-locked-in
#t
;;
;; let's replace the below "winning" lock method with:
;; 1. create a lock file with pid etc.
;; 2. if there are no other lock files make an entry in the no-sync db
;; 3. gather the lock entries, apply the "winner" heuristic
;; 4. if I'm the winner, set tt-state to 'running else set to 'notthewinner
;;
;; New idea:
;; 1. check all processes entries that match the db
;; 2. sort by fixed heuristic
;; 3. if I'm number one, set state to 'running and db-locked-in to #t
(let* ((candidates (dbfile:with-no-sync-db
nosyncdbpath
(lambda (nsdb)
(dbfile:get-process-options nsdb "server" dbfname))))
(primecand (begin
(assert (not (null? candidates))
"HOW CAN WE NOT BE IN THE PROCESSES DB AS A SERVER?")
(dbfile:row->procinf (car candidates)))))
;; compare primecand with myself
;; if not me check that it is reachable
;; if reachable exit
#f)
(let* ((lock-result ;; this is the primary lock - need to double verify that got it
#;(let* ((lock-result ;; this is the primary lock - need to double verify that got it
(dbfile:with-no-sync-db
nosyncdbpath
(lambda (db)
(db:no-sync-lock-and-check db dbfname
(tt-servinf-file ttdat)
;; (dbr:dbstruct-dbtmpname dbstruct)
))))
|
︙ | | |
574
575
576
577
578
579
580
581
582
583
584
585
586
587
|
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
|
+
|
(debug:print 0 *default-log-port* "Failed to get server lock for "dbfname)
#f))))))
(if (and res (common:low-noise-print 120 "top server message"))
(debug:print-info 0 *default-log-port* "Keep running, I'm the top server for "
dbfname" on "(tt-host ttdat)":"(tt-port ttdat)))
res))
(else
;; wrong servinfo file
(debug:print-info 0 *default-log-port* "I'm not the lead server: "servers)
(let* ((leadsrv (car servers)))
(match leadsrv
((host port startseconds server-id pid dbfname servinfofile)
(let* ((result (tt:timed-ping host port server-id))
(res (car result))
(ping (cdr result)))
|
︙ | | |
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
|
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
|
-
+
-
-
+
+
+
-
+
-
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-
+
+
-
-
-
-
-
+
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
+
+
+
|
(else ;; should never get here
(debug:print 0 *default-log-port* "BAD SERVER RECORD: "leadsrv)
(assert #f "Bad server record "leadsrv))))))))
(if ok
(tt-last-access-set! ttdat *db-last-access*) ;; bit silly, just use db-last-access
(begin
(debug:print 0 *default-log-port* "Exiting immediately")
(cleanup)
(tt:shutdown-server ttdat)
(exit)))
(let* ((last-update (dbr:dbstruct-last-update dbstruct))
(curr-secs (current-seconds)))
(if (and (eq? (tt-state ttdat) 'running)
(> (- curr-secs last-update) 3)) ;; every 3-4 seconds update the db?
(begin
(set! (file-modification-time (tt-servinf-file ttdat)) (current-seconds))
(let* ((sinfo-file (tt-servinf-file ttdat)))
;; (debug:print 0 *default-log-port* "INFO: touching "sinfo-file)
(set! (file-modification-time sinfo-file) (current-seconds))
((dbr:dbstruct-sync-proc dbstruct) last-update)
(dbr:dbstruct-last-update-set! dbstruct curr-secs))))
(if (< (- (current-seconds) (tt-last-access ttdat)) (tt-server-timeout-param))
(begin
(thread-sleep! 5)
(loop)))))
(cleanup)
;; (cleanup) ;; all done by tt:shutdown-server
(debug:print 0 *default-log-port* "INFO: Server timed out, exiting from tt:keep-running.")))
;; ;; given an already set up uconn start the cmd-loop
;; ;;
;; (define (tt:cmd-loop ttdat)
;; (let* ((serv-listener (-socket uconn))
;; (listener (lambda ()
;; (let loop ((state 'start))
;; (let-values (((inp oup)(tcp-accept serv-listener)))
;; ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; (let* ((rdat (deserialize inp)) ;; '(my-host-port qrykey cmd params)
;; (resp (ulex-handler uconn rdat)))
;; (serialize resp oup)
;; (close-input-port inp)
;; (close-output-port oup)
;; ;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; )
;; (loop state))))))
;; ;; start N of them
;; (let loop ((thnum 0)
;; (threads '()))
;; (if (< thnum 100)
;; (let* ((th (make-thread listener (conc "listener" thnum))))
;; (thread-start! th)
;; (loop (+ thnum 1)
;; (cons th threads)))
;; (map thread-join! threads)))))
;;
;;
;;
;; (define (wait-and-close uconn)
;; (thread-join! (udat-cmd-thread uconn))
;; (tcp-close (udat-socket uconn)))
;;
;;
(define (tt:shutdown-server ttdat)
(let* ((cleanproc (tt-cleanup-proc ttdat))
(port (tt-port ttdat)))
(let* ((host (tt-host ttdat))
(port (tt-port ttdat))
(sinf (tt-servinf-file ttdat)))
(tt-state-set! ttdat 'shutdown)
(portlogger:open-run-close portlogger:set-port port "released")
(if cleanproc (cleanproc))
(if (file-exists? sinf)
(delete-file* sinf))
(tcp-close (tt-socket ttdat)) ;; close up ports here
))
;; (define (wait-and-close uconn)
;; (thread-join! (tt-cmd-thread uconn))
;; (tcp-close (tt-socket uconn)))
;; return servid
;; side-effects:
;; ttdat-cleanup-proc is populated with function to remove the serverinfo file
(define (tt:create-server-registration-file ttdat dbfname)
(let* ((areapath (tt-areapath ttdat))
(servdir (tt:get-servinfo-dir areapath))
(host (tt-host ttdat))
(port (tt-port ttdat))
(servinf (conc servdir"/"host":"port"-"(current-process-id)":"dbfname))
(serv-id (tt:mk-signature areapath))
(serv-id (tt:mk-signature areapath)))
(clean-proc (lambda ()
(delete-file* servinf)
)))
(assert (and host port) "FATAL: tt:create-server-registration-file called with no conn, dbfname="dbfname)
(tt-cleanup-proc-set! ttdat clean-proc)
(tt-servinf-file-set! ttdat servinf)
(with-output-to-file servinf
(lambda ()
(print "SERVER STARTED: "host":"port" AT "(current-seconds)" server-id: "serv-id" pid: "(current-process-id)" dbfname: "dbfname)))
serv-id))
;; find valid server
;; get servers listed, last part of name must match :<dbfname>
;; if more than one, wait one second and look again
;; future: ping oldest, if alive remove other :<dbfname> files
;;
(define (tt:find-server areapath dbfname)
(let* ((servdir (tt:get-servinfo-dir areapath))
(sfiles (glob (conc servdir"/*:"dbfname))))
sfiles))
(sfiles (glob (conc servdir"/*:"dbfname)))
(goodfiles '()))
;; filter the files here by looking in processes table (if we are not main.db)
;; and or look at the time stamp on the servinfo file, a running server will
;; touch the file every minute (again, this will only apply for main.db)
(for-each (lambda (fname)
(let* ((age (- (current-seconds)(file-modification-time fname))))
(if (> age 200) ;; can't trust it if over 200 seconds old
(begin
(debug:print 0 *default-log-port* "WARNING: removing stale servinfo file "fname", it is "age" seconds old")
(handle-exceptions
exn
(debug:print 0 *default-log-port* "WARNING: error attempting to remove stale servinfo file "fname)
(delete-file fname))) ;;
(set! goodfiles (cons fname goodfiles)))))
sfiles)
goodfiles))
;; given a path to a server info file return: host port startseconds server-id pid dbfname logf
;; example of what it's looking for in the log file:
;; SERVER STARTED: 10.38.175.67:50216 AT 1616502350.0 server-id: 4907e90fc55c7a09694e3f658c639cf4
;;
(define (tt:server-get-info logf)
(let ((server-rx (regexp "^SERVER STARTED: (\\S+):(\\d+) AT ([\\d\\.]+) server-id: (\\S+) pid: (\\d+) dbfname: (\\S+)")) ;; SERVER STARTED: host:port AT timesecs server id
(dbprep-rx (regexp "^SERVER: dbprep"))
(dbprep-found 0)
(bad-dat (list #f #f #f #f #f #f logf)))
(let ((fdat (handle-exceptions
exn
(begin
;; WARNING: this is potentially dangerous to blanket ignore the errors
(debug:print-info 0 *default-log-port* "Unable to get server info from "logf", exn="(condition->list exn))
;; BUG, TODO: add err checking, for now blanket ignore the errors?
(debug:print-info 0 *default-log-port* "Unable to get server info from "logf
", exn="(condition->list exn))
'()) ;; no idea what went wrong, call it a bad server, return empty list
(with-input-from-file logf read-lines))))
(if (null? fdat) ;; bad data, return bad-dat
bad-dat
(let loop ((inl (car fdat))
(tail (cdr fdat))
(lnum 0))
|
︙ | | |
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
|
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
|
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
|
(string->number pid)
dbfname
logf))
(else
(debug:print 0 *default-log-port* "ERROR: did not recognise SERVER line info "mlst)
bad-dat)))))))))
(define *last-server-start* (make-hash-table))
(define (tt:too-recent-server-start dbfname)
(let* ((last-run-time (hash-table-ref/default *last-server-start* dbfname #f)))
(and last-run-time
(< (- (current-seconds) last-run-time) 5))))
;; Given an area path, start a server process ### NOTE ### > file 2>&1
;; if the target-host is set
;; try running on that host
;; incidental: rotate logs in logs/ dir.
;;
(define (tt:server-process-run areapath testsuite mtexe run-id #!key (profile-mode "")) ;; areapath is *toppath* for a given testsuite area
(assert areapath "FATAL: tt:server-process-run called without areapath defined.")
(assert testsuite "FATAL: tt:server-process-run called without testsuite defined.")
(assert mtexe "FATAL: tt:server-process-run called without mtexe defined.")
;; mtest -server - -m testsuite:ext-tests -db 6.db
(let* ((dbfname (dbmod:run-id->dbfname run-id))
(load (get-normalized-cpu-load))
(trying (length (tt:find-server areapath dbfname)))
(nrun (number-of-processes-running (conc "mtest.*server.*"testsuite".*"dbfname))))
(cond
((> load 2.0)
(debug:print 0 *default-log-port* "Normalized load "load" on " (get-host-name) " is over the limit of 2.0. Not starting a server.")
(thread-sleep! 1))
((> nrun 100)
(debug:print 0 *default-log-port* nrun" servers running on " (get-host-name) ", not starting another.")
(thread-sleep! 1))
((> trying 4)
(debug:print 0 *default-log-port* trying" servers registered in .servinfo dir. not starting another.")
(thread-sleep! 1))
(else
(if (not (file-exists? (conc areapath"/logs")))
(create-directory (conc areapath"/logs") #t))
(let* ((logfile (conc areapath "/logs/server-"dbfname"-"(current-process-id)".log")) ;; -" curr-pid "-" target-host ".log"))
(cmdln (conc
mtexe
" -startdir "areapath
" -server - ";; (or target-host "-")
" -m testsuite:"testsuite
" -db "dbfname ;; (dbmod:run-id->dbfname run-id)
" " profile-mode
(conc " >> " logfile " 2>&1 &"))))
;; we want the remote server to start in *toppath* so push there
;; (push-directory areapath) ;; use cd in the command line instead
(debug:print 0 *default-log-port* "INFO: Trying to start server in tcp mode (" cmdln ") at "(common:human-time)" for "areapath)
;; (debug:print 0 *default-log-port* "INFO: starting server at " (common:human-time))
(let* ((dbfname (dbmod:run-id->dbfname run-id)))
(if (tt:too-recent-server-start dbfname)
#f
(let* ((load (get-normalized-cpu-load))
(srvrs (tt:find-server areapath dbfname))
(trying (length srvrs))
(nrun (number-of-processes-running (conc "mtest.*server.*"testsuite".*"dbfname))))
(cond
((> load 2.0)
(debug:print 0 *default-log-port* "Normalized load "load" on " (get-host-name) " is over the limit of 2.0. Not starting a server. Please reduce the load on "(get-host-name)" by killing some processes")
(thread-sleep! 1)
#f)
((> nrun 100)
(debug:print 0 *default-log-port* nrun" servers running on " (get-host-name) ", not starting another.")
(thread-sleep! 1)
#f)
((> trying 2)
(debug:print 0 *default-log-port* trying" servers registered in .servinfo dir. not starting another.")
(thread-sleep! 1)
#f)
(else
(if (not (file-exists? (conc areapath"/logs")))
(create-directory (conc areapath"/logs") #t))
(let* ((logfile (conc areapath "/logs/server-"dbfname"-"(current-process-id)".log")) ;; -" curr-pid "-" target-host ".log"))
(cmdln (conc
mtexe
" -startdir "areapath
" -server - ";; (or target-host "-")
" -m testsuite:"testsuite
" -db "dbfname ;; (dbmod:run-id->dbfname run-id)
" " profile-mode
(conc " >> " logfile " 2>&1 &"))))
;; we want the remote server to start in *toppath* so push there
;; (push-directory areapath) ;; use cd in the command line instead
(debug:print 2 *default-log-port* "INFO: Trying to start server in tcp mode (" cmdln ") at "(common:human-time)" for "areapath)
;; (debug:print 0 *default-log-port* "INFO: starting server at " (common:human-time))
(system cmdln)
;; ;; use below to go back to nbfake - nbfake does cause trouble ...
;; (setenv "NBFAKE_QUIET" "yes") ;; BUG: change to with-environment-variable ...
;; (setenv "NBFAKE_LOG" logfile)
;; (system (conc "cd "areapath" ; nbfake " cmdln))
;; (unsetenv "NBFAKE_QUIET")
;; (unsetenv "NBFAKE_LOG")
;;(pop-directory)
)))))
(system cmdln)
(hash-table-set! *last-server-start* dbfname (current-seconds))
;; ;; use below to go back to nbfake - nbfake does cause trouble ...
;; (setenv "NBFAKE_QUIET" "yes") ;; BUG: change to with-environment-variable ...
;; (setenv "NBFAKE_LOG" logfile)
;; (system (conc "cd "areapath" ; nbfake " cmdln))
;; (unsetenv "NBFAKE_QUIET")
;; (unsetenv "NBFAKE_LOG")
;;(pop-directory)
#t)))))))
;;======================================================================
;; tcp connection stuff
;;======================================================================
;; find a port and start tcp-server. This only starts the tcp portion of
;; the server, look at (tt:start-server ...) above for the entry point
|
︙ | | |