1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
+
+
+
|
;; Copyright 2006-2012, Matthew Welland.
;;
;; This program is made available under the GNU GPL version 2.0 or
;; greater. See the accompanying file COPYING for details.
;;
;; This program is distributed WITHOUT ANY WARRANTY; without even the
;; implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
;; PURPOSE.
;; strftime('%m/%d/%Y %H:%M:%S','now','localtime')
(use sqlite3 srfi-1 posix regex regex-case srfi-69 dot-locking format)
(import (prefix sqlite3 sqlite3:))
(include "/nfs/site/disks/icf_fdk_cw_gwa002/srehman/fossil/dbi/dbi.scm")
(import (prefix dbi dbi:))
(declare (unit tasks))
(declare (uses db))
(declare (uses rmt))
(declare (uses common))
(include "task_records.scm")
|
| ︙ | | |
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
152
153
154
155
156
|
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
152
153
154
155
156
157
158
159
160
161
162
|
+
+
-
-
-
-
+
+
+
+
+
-
-
+
+
-
+
-
+
-
+
-
+
|
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(debug:print 0 *default-log-port* " exn=" (condition->list exn))))
(let* ((dbpath (tasks:get-task-db-path))
(dbfile (conc dbpath "/monitor.db"))
(avail (tasks:wait-on-journal dbpath 10)) ;; wait up to about 10 seconds for the journal to go away
(exists (file-exists? dbpath))
(write-access (file-write-access? dbpath))
(dbdat '())
(mdb (cond ;; what the hek is *toppath* doing here?
((and (string? *toppath*)(file-write-access? *toppath*))
(set! dbdat (cons (cons 'dbname dbfile) dbdat))
(sqlite3:open-database dbfile))
((file-read-access? dbpath) (sqlite3:open-database dbfile))
(else (sqlite3:open-database ":memory:")))) ;; (never-give-up-open-db dbpath))
(handler (make-busy-timeout 36000)))
(dbi:open 'sqlite3 dbdat))
((file-read-access? dbpath) (dbi:open 'sqlite3 dbdat))
(else (dbi:open 'sqlite3 '((dbname . ":memory:"))))))) ;; (never-give-up-open-db dbpath))
;;(handler (make-busy-timeout 36000)))
(if (and exists
(not write-access))
(set! *db-write-access* write-access)) ;; only unset so other db's also can use this control
(sqlite3:set-busy-handler! mdb handler)
(db:set-sync mdb) ;; (sqlite3:execute mdb (conc "PRAGMA synchronous = 0;"))
;;(sqlite3:set-busy-handler! mdb handler)
(db:set-sync mdb) ;; (dbi:exec mdb (conc "PRAGMA synchronous = 0;"))
;; (if (or (and (not exists)
;; (file-write-access? *toppath*))
;; (not (file-read-access? dbpath)))
;; (begin
;;
;; TASKS QUEUE MOVED TO main.db
;;
;; (sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS tasks_queue (id INTEGER PRIMARY KEY,
;; (dbi:exec mdb "CREATE TABLE IF NOT EXISTS tasks_queue (id INTEGER PRIMARY KEY,
;; action TEXT DEFAULT '',
;; owner TEXT,
;; state TEXT DEFAULT 'new',
;; target TEXT DEFAULT '',
;; name TEXT DEFAULT '',
;; testpatt TEXT DEFAULT '',
;; keylock TEXT,
;; params TEXT,
;; creation_time TIMESTAMP,
;; execution_time TIMESTAMP);")
(sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS monitors (id INTEGER PRIMARY KEY,
(dbi:exec mdb "CREATE TABLE IF NOT EXISTS monitors (id INTEGER PRIMARY KEY,
pid INTEGER,
start_time TIMESTAMP,
last_update TIMESTAMP,
hostname TEXT,
username TEXT,
CONSTRAINT monitors_constraint UNIQUE (pid,hostname));")
(sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS servers (id INTEGER PRIMARY KEY,
(dbi:exec mdb "CREATE TABLE IF NOT EXISTS servers (id INTEGER PRIMARY KEY,
pid INTEGER,
interface TEXT,
hostname TEXT,
port INTEGER,
pubport INTEGER,
start_time TIMESTAMP,
priority INTEGER,
state TEXT,
mt_version TEXT,
heartbeat TIMESTAMP,
transport TEXT,
run_id INTEGER);")
;; CONSTRAINT servers_constraint UNIQUE (pid,hostname,port));")
(sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS clients (id INTEGER PRIMARY KEY,
(dbi:exec mdb "CREATE TABLE IF NOT EXISTS clients (id INTEGER PRIMARY KEY,
server_id INTEGER,
pid INTEGER,
hostname TEXT,
cmdline TEXT,
login_time TIMESTAMP,
logout_time TIMESTAMP DEFAULT -1,
CONSTRAINT clients_constraint UNIQUE (pid,hostname));")
|
| ︙ | | |
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
|
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
|
-
+
-
+
+
-
+
-
+
+
-
+
-
+
-
+
-
+
-
-
+
+
-
+
-
+
-
-
+
+
-
+
-
+
-
+
+
-
+
|
(tasks:server-set-available mdb run-id)
(thread-sleep! (/ (random 1500) 1000)) ;; (thread-sleep! 2) ;; Try removing this. It may not be needed.
(tasks:server-am-i-the-server? mdb run-id))
#f))
;; register that this server may come online (first to register goes though with the process)
(define (tasks:server-set-available mdb run-id)
(sqlite3:execute
(dbi:exec
mdb
"INSERT INTO servers (pid,hostname,port,pubport,start_time, priority,state,mt_version,heartbeat, interface,transport,run_id)
VALUES(?, ?, ?, ?, strftime('%s','now'), ?, ?, ?,-1,?, ?, ?);"
(current-process-id) ;; pid
(get-host-name) ;; hostname
-1 ;; port
-1 ;; pubport
(random 1000) ;; priority (used a tiebreaker on get-available)
"available" ;; state
(common:version-signature) ;; mt_version
-1 ;; interface
;; (conc (server:get-transport)) ;; transport
(conc *transport-type*) ;; transport
run-id
))
(define (tasks:num-in-available-state mdb run-id)
(let ((res 0))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (num-in-queue)
(set! res num-in-queue))
(set! res num-in-queue)))
mdb
"SELECT count(id) FROM servers WHERE run_id=? AND state = 'available' AND (strftime('%s','now') - start_time) < 30 ;"
run-id)
res))
(define (tasks:num-servers-non-zero-running mdb)
(let ((res 0))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (num-running)
(set! res num-running))
(set! res num-running)))
mdb
"SELECT count(id) FROM servers WHERE run_id != 0 AND state = 'running';")
res))
(define (tasks:server-clean-out-old-records-for-run-id mdb run-id tag)
(sqlite3:execute mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE state in ('available','dbprep','shutting-down') AND (strftime('%s','now') - start_time) > 50 AND run_id=?;"
(dbi:exec mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE state in ('available','dbprep','shutting-down') AND (strftime('%s','now') - start_time) > 50 AND run_id=?;"
(conc "defunct" tag) run-id))
(define (tasks:server-force-clean-running-records-for-run-id mdb run-id tag)
(sqlite3:execute mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE state = 'running' AND run_id=?;"
(dbi:exec mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE state = 'running' AND run_id=?;"
(conc "defunct" tag) run-id))
(define (tasks:server-force-clean-run-record mdb run-id iface port tag)
(sqlite3:execute mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE state = 'running' AND run_id=? AND interface=? AND port=?;"
(dbi:exec mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE state = 'running' AND run_id=? AND interface=? AND port=?;"
(conc "defunct" tag) run-id iface port))
;; BB> adding missing func for --list-servers
(define (tasks:server-deregister mdb hostname #!key (pullport #f) (pid #f) (action #f)) ;;pullport pid: pid action: 'delete))
(if (eq? action 'delete)
(sqlite3:execute mdb "DELETE FROM servers WHERE pid=? AND port=? AND hostname=?;" pid pullport hostname)
(sqlite3:execute mdb "UPDATE servers SET state='defunct', heartbeat=strftime('%s','now') WHERE hostname=? AND pid=?;"
(dbi:exec mdb "DELETE FROM servers WHERE pid=? AND port=? AND hostname=?;" pid pullport hostname)
(dbi:exec mdb "UPDATE servers SET state='defunct', heartbeat=strftime('%s','now') WHERE hostname=? AND pid=?;"
hostname pid)))
(define (tasks:server-delete-records-for-this-pid mdb tag)
(sqlite3:execute mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE hostname=? AND pid=?;"
(dbi:exec mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE hostname=? AND pid=?;"
(conc "defunct" tag) (get-host-name) (current-process-id)))
(define (tasks:server-delete-record mdb server-id tag)
(sqlite3:execute mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE id=?;"
(dbi:exec mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE id=?;"
(conc "defunct" tag) server-id)
;; use this opportuntity to clean out records over one month old or over 10 minutes old with port = -1 (i.e. a never used placeholder)
(sqlite3:execute mdb "DELETE FROM servers WHERE state not in ('running','shutting-down','dbprep') AND (strftime('%s','now') - start_time) > 2628000;")
(sqlite3:execute mdb "DELETE FROM servers WHERE state like 'defunct%' AND port=-1 AND (strftime('%s','now') - start_time) > 600;")
(dbi:exec mdb "DELETE FROM servers WHERE state not in ('running','shutting-down','dbprep') AND (strftime('%s','now') - start_time) > 2628000;")
(dbi:exec mdb "DELETE FROM servers WHERE state like 'defunct%' AND port=-1 AND (strftime('%s','now') - start_time) > 600;")
)
(define (tasks:server-set-state! mdb server-id state)
(sqlite3:execute mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE id=?;" state server-id))
(dbi:exec mdb "UPDATE servers SET state=?,heartbeat=strftime('%s','now') WHERE id=?;" state server-id))
(define (tasks:server-set-interface-port mdb server-id interface port)
(sqlite3:execute mdb "UPDATE servers SET interface=?,port=?,heartbeat=strftime('%s','now') WHERE id=?;" interface port server-id))
(dbi:exec mdb "UPDATE servers SET interface=?,port=?,heartbeat=strftime('%s','now') WHERE id=?;" interface port server-id))
;; Get random port not used in long time
;;
(define (tasks:server-get-next-port mdb)
(let* ((lownum 30000)
(highnum 64000)
(used-ports '())
(get-rand-port (lambda ()
(+ lownum (random (- highnum lownum)))))
(port-param (if (and (args:get-arg "-port")
(string->number (args:get-arg "-port")))
(string->number (args:get-arg "-port"))
#f))
;; (config-port (if (and (config-lookup *configdat* "server" "port")
;; (string->number (config-lookup *configdat* "server" "port")))
;; (string->number (config-lookup *configdat* "server" "port"))
;; #f))
)
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (port)
(set! used-ports (cons port used-ports)))
(set! used-ports (cons port used-ports))))
mdb
"SELECT port FROM servers;")
(cond
((and port-param res) (if (> res port-param) res port-param))
(port-param port-param)
;; ((and config-port res) (if (> res config-port) res config-port))
;; (config-port config-port)
|
| ︙ | | |
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
|
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
|
-
+
+
-
+
|
;; Use: (db:get-value-by-header (car (db:get-rows dat)) (db:get-header dat) "fieldname")
;; to extract info from the structure returned
;;
(define (tasks:server-get-servers-vying-for-run-id mdb run-id)
(let* ((header (list "id" "hostname" "pid" "interface" "port" "pubport" "state" "run_id" "priority" "start_time"))
(selstr (string-intersperse header ","))
(res '()))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (a . b)
(set! res (cons (apply vector a b) res)))
(set! res (cons (apply vector a b) res))))
mdb
(conc "SELECT " selstr " FROM servers WHERE run_id=? AND state in ('available','running','dbprep') ORDER BY start_time DESC;")
run-id)
(vector header res)))
(define (tasks:get-server mdb run-id #!key (retries 10))
(let ((res #f)
|
| ︙ | | |
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
|
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
|
-
+
+
-
+
-
+
+
-
+
-
+
+
-
+
|
(print-call-chain (current-error-port))
(if (> retries 0)
(begin
(debug:print 0 *default-log-port* " trying call to tasks:get-server again in 10 seconds")
(thread-sleep! 10)
(tasks:get-server mdb run-id retries: (- retries 0)))
(debug:print 0 *default-log-port* "10 tries of tasks:get-server all crashed and burned. Giving up and returning \"no server found\"")))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id interface port pubport transport pid hostname)
(set! res (vector id interface port pubport transport pid hostname)))
(set! res (vector id interface port pubport transport pid hostname))))
mdb
;; removed:
;; strftime('%s','now')-heartbeat < 10 AND mt_version = ?
"SELECT id,interface,port,pubport,transport,pid,hostname FROM servers
WHERE run_id=? AND state='running'
ORDER BY start_time DESC LIMIT 1;" run-id) ;; (common:version-signature) run-id)
res)))
(define (tasks:server-running-or-starting? mdb run-id)
(let ((res #f))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id)
(set! res id))
(set! res id)))
mdb ;; NEEDS dbprep ADDED
"SELECT id FROM servers WHERE run_id=? AND (state = 'running' OR (state = 'dbprep' AND (strftime('%s','now') - start_time) < 60));" run-id)
res))
(define (tasks:server-running? mdb run-id)
(let ((res #f))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id)
(set! res id))
(set! res id)))
mdb ;; NEEDS dbprep ADDED
"SELECT id FROM servers WHERE run_id=? AND state = 'running';" run-id)
res))
(define (tasks:need-server run-id)
(equal? (configf:lookup *configdat* "server" "required") "yes"))
|
| ︙ | | |
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
|
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
|
-
+
+
-
+
-
+
+
+
-
+
-
+
+
+
-
+
|
(server:read-dotserver *toppath*))) ;; no point in trying
(loop (tasks:get-server (db:delay-if-busy tdbdat) run-id)(+ delay-time 1))
#f))
#f)))
(define (tasks:get-all-servers mdb)
(let ((res '()))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id)
;; 0 1 2 3 4 5 6 7 8 9 10 11 12
(set! res (cons (vector id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id) res)))
(set! res (cons (vector id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id) res))))
mdb
"SELECT id,pid,hostname,interface,port,pubport,start_time,priority,state,mt_version,strftime('%s','now')-heartbeat AS last_update,transport,run_id
FROM servers WHERE state NOT LIKE 'defunct%' ORDER BY start_time DESC;")
res))
(define (tasks:get-server-by-id mdb id)
(let ((res #f))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id)
;; 0 1 2 3 4 5 6 7 8 9 10 11 12
(set! res (vector id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id)))
(set! res (vector id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id))))
mdb
"SELECT id,pid,hostname,interface,port,pubport,start_time,priority,state,mt_version,strftime('%s','now')-heartbeat AS last_update,transport,run_id
FROM servers WHERE id=?;"
id)
res))
(define (tasks:get-server-records mdb run-id)
(let ((res '()))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id)
;; 0 1 2 3 4 5 6 7 8 9 10 11 12
(set! res (cons (vector id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id) res)))
(set! res (cons (vector id pid hostname interface port pubport start-time priority state mt-version last-update transport run-id) res))))
mdb
"SELECT id,pid,hostname,interface,port,pubport,start_time,priority,state,mt_version,strftime('%s','now')-heartbeat AS last_update,transport,run_id
FROM servers WHERE run_id=? AND state NOT LIKE 'defunct%' ORDER BY start_time DESC;"
run-id)
(reverse res)))
;; no elegance here ...
|
| ︙ | | |
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
|
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
|
-
+
-
+
+
-
+
|
))
;;======================================================================
;; M O N I T O R S
;;======================================================================
(define (tasks:remove-monitor-record mdb)
(sqlite3:execute mdb "DELETE FROM monitors WHERE pid=? AND hostname=?;"
(dbi:exec mdb "DELETE FROM monitors WHERE pid=? AND hostname=?;"
(current-process-id)
(get-host-name)))
(define (tasks:get-monitors mdb)
(let ((res '()))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (a . rem)
(set! res (cons (apply vector a rem) res)))
(set! res (cons (apply vector a rem) res))))
mdb
"SELECT id,pid,strftime('%m/%d/%Y %H:%M',datetime(start_time,'unixepoch'),'localtime'),strftime('%m/%d/%Y %H:%M:%S',datetime(last_update,'unixepoch'),'localtime'),hostname,username FROM monitors ORDER BY last_update ASC;")
(reverse res)
))
(define (tasks:monitors->text-table monitors)
(let ((fmtstr "~4a~8a~20a~20a~10a~10a"))
|
| ︙ | | |
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
|
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
|
-
+
-
+
+
-
+
-
+
-
+
-
+
+
-
+
|
(tasks:monitor-get-username monitor)))
monitors)
"\n"))))
;; update the last_update field with the current time and
;; if any monitors appear dead, remove them
(define (tasks:monitors-update mdb)
(sqlite3:execute mdb "UPDATE monitors SET last_update=strftime('%s','now') WHERE pid=? AND hostname=?;"
(dbi:exec mdb "UPDATE monitors SET last_update=strftime('%s','now') WHERE pid=? AND hostname=?;"
(current-process-id)
(get-host-name))
(let ((deadlist '()))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id pid host last-update delta)
(print "Going to delete stale record for monitor with pid " pid " on host " host " last updated " delta " seconds ago")
(set! deadlist (cons id deadlist)))
(set! deadlist (cons id deadlist))))
mdb
"SELECT id,pid,hostname,last_update,strftime('%s','now')-last_update AS delta FROM monitors WHERE delta > 700;")
(sqlite3:execute mdb (conc "DELETE FROM monitors WHERE id IN ('" (string-intersperse (map conc deadlist) "','") "');")))
(dbi:exec mdb (conc "DELETE FROM monitors WHERE id IN ('" (string-intersperse (map conc deadlist) "','") "');")))
)
(define (tasks:register-monitor db port)
(let* ((pid (current-process-id))
(hostname (get-host-name))
(userinfo (user-information (current-user-id)))
(username (car userinfo)))
(print "Register monitor, pid: " pid ", hostname: " hostname ", port: " port ", username: " username)
(sqlite3:execute db "INSERT INTO monitors (pid,start_time,last_update,hostname,username) VALUES (?,strftime('%s','now'),strftime('%s','now'),?,?);"
(dbi:exec db "INSERT INTO monitors (pid,start_time,last_update,hostname,username) VALUES (?,strftime('%s','now'),strftime('%s','now'),?,?);"
pid hostname username)))
(define (tasks:get-num-alive-monitors mdb)
(let ((res 0))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (count)
(set! res count))
(set! res count)))
mdb
"SELECT count(id) FROM monitors WHERE last_update < (strftime('%s','now') - 300) AND username=?;"
(car (user-information (current-user-id))))
res))
;;
(define (tasks:start-monitor db mdb)
|
| ︙ | | |
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
|
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
|
-
+
|
;; register a task
(define (tasks:add dbstruct action owner target runname testpatt params)
(db:with-db
dbstruct #f #t
(lambda (db)
(sqlite3:execute db "INSERT INTO tasks_queue (action,owner,state,target,name,testpatt,params,creation_time,execution_time)
(dbi:exec db "INSERT INTO tasks_queue (action,owner,state,target,name,testpatt,params,creation_time,execution_time)
VALUES (?,?,'new',?,?,?,?,strftime('%s','now'),0);"
action
owner
target
runname
testpatt
(if params params "")))))
|
| ︙ | | |
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
|
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
|
-
+
-
+
+
-
+
-
+
-
+
+
-
+
-
+
-
+
+
-
+
-
-
-
+
+
+
+
-
+
|
(define (tasks:snag-a-task dbstruct)
(let ((res #f)
(keytxt (conc (current-process-id) "-" (get-host-name) "-" (car (user-information (current-user-id))))))
(db:with-db
dbstruct #f #t
(lambda (db)
;; first randomly set a new to pid-hostname-hostname
(sqlite3:execute
(dbi:exec
db
"UPDATE tasks_queue SET keylock=? WHERE id IN
(SELECT id FROM tasks_queue
WHERE state='new' OR
(state='waiting' AND (strftime('%s','now')-execution_time) > 10) OR
state='reset'
ORDER BY RANDOM() LIMIT 1);" keytxt)
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id . rem)
(set! res (apply vector id rem)))
(set! res (apply vector id rem))))
db
"SELECT id,action,owner,state,target,name,test,item,params,creation_time,execution_time FROM tasks_queue WHERE keylock=? ORDER BY execution_time ASC LIMIT 1;" keytxt)
(if res ;; yep, have work to be done
(begin
(sqlite3:execute db "UPDATE tasks_queue SET state='inprogress',execution_time=strftime('%s','now') WHERE id=?;"
(dbi:exec db "UPDATE tasks_queue SET state='inprogress',execution_time=strftime('%s','now') WHERE id=?;"
(tasks:task-get-id res))
res)
#f)))))
(define (tasks:reset-stuck-tasks dbstruct)
(let ((res '()))
(db:with-db
dbstruct #f #t
(lambda (db)
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id delta)
(set! res (cons id res)))
(set! res (cons id res))))
db
"SELECT id,strftime('%s','now')-execution_time AS delta FROM tasks_queue WHERE state='inprogress' AND delta>700 ORDER BY delta DESC LIMIT 2;")
(sqlite3:execute
(dbi:exec
db
(conc "UPDATE tasks_queue SET state='reset' WHERE id IN ('" (string-intersperse (map conc res) "','") "');")
)))))
;; return all tasks in the tasks_queue table
;;
(define (tasks:get-tasks dbstruct types states)
(let ((res '()))
(db:with-db
dbstruct #f #f
(lambda (db)
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (id . rem)
(set! res (cons (apply vector id rem) res)))
(set! res (cons (apply vector id rem) res))))
db
(conc "SELECT id,action,owner,state,target,name,test,item,params,creation_time,execution_time
FROM tasks_queue "
;; WHERE
;; state IN " statesstr " AND
;; action IN " actionsstr
" ORDER BY creation_time DESC;"))
res))))
(define (tasks:get-last dbstruct target runname)
(let ((res #f))
(db:with-db
dbstruct #f #f
(lambda (db)
(sqlite3:for-each-row
(lambda (id . rem)
(set! res (apply vector id rem)))
(dbi:for-each-row
(lambda (output)
(lambda (id . rem)
(set! res (apply vector id rem))))
db
(conc "SELECT id,action,owner,state,target,name,testpatt,keylock,params,creation_time,execution_time
FROM tasks_queue
WHERE
target = ? AND name =?
ORDER BY creation_time DESC LIMIT 1;")
target runname)
res))))
;; remove tasks given by a string of numbers comma separated
(define (tasks:remove-queue-entries dbstruct task-ids)
(db:with-db
dbstruct #f #t
(lambda (db)
(sqlite3:execute db (conc "DELETE FROM tasks_queue WHERE id IN (" task-ids ");")))))
(dbi:exec db (conc "DELETE FROM tasks_queue WHERE id IN (" task-ids ");")))))
(define (tasks:process-queue dbstruct)
(let* ((task (tasks:snag-a-task dbstruct))
(action (if task (tasks:task-get-action task) #f)))
(if action (print "tasks:process-queue task: " task))
(if action
(case (string->symbol action)
|
| ︙ | | |
738
739
740
741
742
743
744
745
746
747
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
|
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
|
-
+
-
+
-
+
-
+
-
+
+
-
+
|
(tasks:task-get-params task)))
tasks) "\n"))))
(define (tasks:set-state dbstruct task-id state)
(db:with-db
dbstruct #f #t
(lambda (db)
(sqlite3:execute db "UPDATE tasks_queue SET state=? WHERE id=?;"
(dbi:exec db "UPDATE tasks_queue SET state=? WHERE id=?;"
state
task-id))))
;;======================================================================
;; Access using task key (stored in params; (hash-table->alist flags) hostname pid
;;======================================================================
(define (tasks:param-key->id dbstruct task-params)
(db:with-db
dbstruct #f #f
(lambda (db)
(handle-exceptions
exn
#f
(sqlite3:first-result db "SELECT id FROM tasks_queue WHERE params LIKE ?;"
(dbi:get-one db "SELECT id FROM tasks_queue WHERE params LIKE ?;"
task-params)))))
(define (tasks:set-state-given-param-key dbstruct param-key new-state)
(db:with-db
dbstruct #f #t
(lambda (db)
(sqlite3:execute db "UPDATE tasks_queue SET state=? WHERE params LIKE ?;" new-state param-key))))
(dbi:exec db "UPDATE tasks_queue SET state=? WHERE params LIKE ?;" new-state param-key))))
(define (tasks:get-records-given-param-key dbstruct param-key state-patt action-patt test-patt)
(db:with-db
dbstruct #f #f
(lambda (db)
(handle-exceptions
exn
'()
(sqlite3:first-row db "SELECT id,action,owner,state,target,name,testpatt,keylock,params WHERE
(dbi:get-one-row db "SELECT id,action,owner,state,target,name,testpatt,keylock,params WHERE
params LIKE ? AND state LIKE ? AND action LIKE ? AND testpatt LIKE ?;"
param-key state-patt action-patt test-patt)))))
(define (tasks:find-task-queue-records dbstruct target run-name test-patt state-patt action-patt)
;; (handle-exceptions
;; exn
;; '()
;; (sqlite3:first-row
(let ((db (db:delay-if-busy (db:get-db dbstruct #f)))
(res '()))
(sqlite3:for-each-row
(dbi:for-each-row
(lambda (output)
(lambda (a . b)
(set! res (cons (cons a b) res)))
(set! res (cons (cons a b) res))))
db "SELECT id,action,owner,state,target,name,testpatt,keylock,params FROM tasks_queue
WHERE
target = ? AND name = ? AND state LIKE ? AND action LIKE ? AND testpatt LIKE ?;"
target run-name state-patt action-patt test-patt)
res)) ;; )
;; kill any runner processes (i.e. processes handling -runtests) that match target/runname
|
| ︙ | | |