︙ | | |
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
|
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
|
-
+
-
-
+
+
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
|
;; open or create the disk db file
;; create and fill the inmemory db
;; assemble into dbr:dbdat struct and return
;;
(define (db:open-dbdat apath dbfile dbinit-proc)
(let* (;; (dbfile (db:run-id->path apath run-id))
(db (db:open-run-db dbfile dbinit-proc))
(inmem (db:open-inmem-db dbinit-proc))
;; (inmem (db:open-inmem-db dbinit-proc))
(dbdat (make-dbr:dbdat
db: db
inmem: inmem
db: #f ;; db
inmem: db ;; inmem
;; run-id: run-id ;; no can do, there are many run-id values that point to single db
fname: dbfile)))
;; now sync the disk file data into the inmemory db
(db:sync-tables (db:sync-all-tables-list) #f db inmem)
;; (db:sync-tables (db:sync-all-tables-list) '("last_update" . 0) db inmem)
;; (sqlite3:finalize! db) ;; open and close every sync
dbdat))
;; (define (db:open-dbdat apath dbfile dbinit-proc)
;; (let* (;; (dbfile (db:run-id->path apath run-id))
;; (db (db:open-run-db dbfile dbinit-proc))
;; (inmem (db:open-inmem-db dbinit-proc))
;; (dbdat (make-dbr:dbdat
;; db: #f ;; db
;; inmem: inmem
;; ;; run-id: run-id ;; no can do, there are many run-id values that point to single db
;; fname: dbfile)))
;; ;; now sync the disk file data into the inmemory db
;; (db:sync-tables (db:sync-all-tables-list) '("last_update" . 0) db inmem)
;; (sqlite3:finalize! db) ;; open and close every sync
;; dbdat))
;; open the disk database file
;; NOTE: May need to add locking to file create process here
;; returns an sqlite3 database handle
;;
(define (db:open-run-db dbfile dbinit-proc)
(let* ((parent-dir (pathname-directory dbfile)))
(if (not (directory-exists? parent-dir))
(create-directory parent-dir #t))
(let* ((exists (file-exists? dbfile))
(db (sqlite3:open-database dbfile))
(handler (sqlite3:make-busy-timeout 3600)))
(sqlite3:set-busy-handler! db handler)
(db:set-sync db)
;; (db:set-sync db) ;; we don't mind that this is slow?
(if (not exists)
(dbinit-proc db))
db)))
;; open and initialize the inmem db
;; NOTE: Does NOT sync in the data from the disk db
;;
|
︙ | | |
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
|
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
-
+
-
+
|
;; - uses db:initialize-db to create the schema
;;
;; Make the dbstruct, call for main db at least once
;; sync disk db to inmem
;;
;; called in http-transport and replicated in rmt.scm for *local* access.
;;
(define (db:setup run-id)
(define (db:setup db-file) ;; run-id)
(assert *toppath* "FATAL: db:setup called before toppath is available.")
(let* ((dbstruct (or *dbstruct-db* (make-dbr:dbstruct)))
(db-file (db:run-id->path *toppath* run-id)))
#;(db-file (db:run-id->path *toppath* run-id)))
(db:get-dbdat dbstruct *toppath* db-file)
(if (not *dbstruct-db*)(set! *dbstruct-db* dbstruct))
dbstruct))
;;======================================================================
;; setting/getting a lock on the db for only one server per db
;;
|
︙ | | |
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
|
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
|
-
+
|
(db:with-lock-db dbfile
(lambda (dbh dbfile)
(db:get-iam-server-lock dbh dbfile))))
(define (db:with-lock-db dbfile proc)
(let* ((dbh (db:open-run-db dbfile db:initialize-db))
(res (proc dbh dbfile)))
(sqlite3:finalize! dbh)
;; (sqlite3:finalize! dbh)
res))
;; called before db is open?
;;
(define (db:get-iam-server-lock dbh dbfname)
(sqlite3:with-transaction
dbh
|
︙ | | |
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
451
452
453
454
455
456
457
458
459
460
461
462
463
|
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
469
470
471
472
473
474
475
476
477
478
479
480
481
|
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
-
+
|
;; ;; (mutex-unlock! *db-multi-sync-mutex*)
;; ;; (stack-push! (dbr:dbstruct-dbstack dbstruct) tmpdb)))
;; NOTE: touched logic is disabled/not done
;; sync run to disk if touched
;;
(define (db:sync-inmem->disk dbstruct apath dbfile #!key (force-sync #f))
#f) ;; disabled
(let* ((dbdat (db:get-dbdat dbstruct apath dbfile))
(db (dbr:dbdat-db dbdat))
(inmem (dbr:dbdat-inmem dbdat))
(start-t (current-seconds))
(last-update (dbr:dbdat-last-write dbdat))
(last-sync (dbr:dbdat-last-sync dbdat)))
(debug:print-info 4 *default-log-port* "Syncing for dbfile: " dbfile)
(mutex-lock! *db-multi-sync-mutex*)
(let* ((update_info (cons (if force-sync 0 last-update) "last_update"))
(need-sync (or force-sync (>= last-update last-sync))))
(if need-sync
(db:sync-tables (db:sync-all-tables-list) update_info inmem db)
(debug:print 0 *default-log-port* "Skipping sync as nothing touched.")))
(dbr:dbdat-last-sync-set! dbdat start-t)
(mutex-unlock! *db-multi-sync-mutex*)))
;; (let* ((dbdat (db:get-dbdat dbstruct apath dbfile))
;; (dbfullname (conc apath "/" dbfile))
;; (db (db:open-run-db dbfullname db:initialize-db)) ;; (dbr:dbdat-db dbdat))
;; (inmem (dbr:dbdat-inmem dbdat))
;; (start-t (current-seconds))
;; (last-update (dbr:dbdat-last-write dbdat))
;; (last-sync (dbr:dbdat-last-sync dbdat)))
;; (debug:print-info 0 *default-log-port* "Syncing for dbfile: "dbfile", last-update: "last-update", last-sync: "last-sync)
;; (mutex-lock! *db-multi-sync-mutex*)
;; (let* ((update_info (cons "last_update" (if force-sync 0 last-update))) ;; "last_update"))
;; (need-sync (or force-sync (>= last-update last-sync))))
;; (if need-sync
;; (begin
;; (db:sync-tables (db:sync-all-tables-list) update_info inmem db)
;; (dbr:dbdat-last-sync-set! dbdat start-t))
;; (debug:print 0 *default-log-port* "Skipping sync as nothing touched.")))
;; (sqlite3:finalize! db)
;; (mutex-unlock! *db-multi-sync-mutex*)))
;; TODO: Add final sync to this
;;
(define (db:safely-close-sqlite3-db db stmt-cache #!key (try-num 3))
#;(define (db:safely-close-sqlite3-db db stmt-cache #!key (try-num 3))
(if (<= try-num 0)
#f
(handle-exceptions
exn
(begin
(print "Attempt to safely close sqlite3 db failed. Trying again. exn=" exn)
(thread-sleep! 3)
(sqlite3:interrupt! db)
(db:safely-close-sqlite3-db db stmt-cache try-num: (- try-num 1)))
(if (sqlite3:database? db)
(let* ((stmts (and stmt-cache (hash-table-ref/default stmt-cache db #f))))
(if stmts (map sqlite3:finalize! (hash-table-values stmts)))
(sqlite3:finalize! db)
#t)
#f))))
;; close all opened run-id dbs
(define (db:close-all dbstruct)
#;(define (db:close-all dbstruct)
(assert (dbr:dbstruct? dbstruct) "FATAL: db:close-all called with dbstruct not set up.")
(handle-exceptions
exn
(begin
(debug:print 0 *default-log-port* "WARNING: Finalizing failed, " ((condition-property-accessor 'exn 'message) exn) ", note - exn=" exn)
(print-call-chain *default-log-port*))
;; (db:sync-touched dbstruct 0 force-sync: #t) ;; NO. Do not do this here. Instead we rely on a server to be started when there are writes, even if the server itself is not going to be used as a server.
|
︙ | | |
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
|
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
|
+
+
-
-
-
+
+
+
-
-
-
-
-
-
-
-
-
-
-
+
-
-
-
-
+
+
-
-
+
-
-
-
-
|
(sqlite3:execute "DELETE FROM servers WHERE state LIKE 'defunct%';"))
(else
(sqlite3:execute db "vacuum;")))
(sqlite3:finalize! db)
#t))))))
;; last-update is *always* a pair ( fieldname|#f . last-update-seconds|#f)
(define (db:sync-one-table fromdb todb tabledat last-update numrecs)
(assert (pair? last-update) "FATAL: last-update must always be a pair.")
(let* ((tablename (car tabledat))
(fields (cdr tabledat))
(has-last-update (member "last_update" fields))
(use-last-update (cond
((and has-last-update
(member "last_update" fields))
(last-update-field (or (car last-update)
(if has-last-update
"last_update"
#t) ;; if given a number, just use it for all fields
((number? last-update) #f) ;; if not matched first entry then ignore last-update for this table
((and (pair? last-update)
(member (car last-update) ;; last-update field name
(map car fields)))
#t)
(last-update
(debug:print 0 *default-log-port* "ERROR: parameter last-update for db:sync-tables must be a pair or a number, received: " last-update);; found in fields
#f)
(else
#f)))
#f)))
(last-update-value (if use-last-update ;; no need to check for has-last-update - it is already accounted for
(if (number? last-update)
last-update
(cdr last-update))
(has-field (member last-update-field fields))
(last-update-value (cdr last-update))
#f))
(last-update-field (if use-last-update
(use-last-update (and has-field last-update-field last-update-value))
(if (number? last-update)
"last_update"
(car last-update))
#f))
(num-fields (length fields))
(field->num (make-hash-table))
(num->field (apply vector (map car fields))) ;; BBHERE
(full-sel (conc "SELECT " (string-intersperse (map car fields) ",")
" FROM " tablename (if use-last-update ;; apply last-update criteria
(conc " WHERE " last-update-field " >= " last-update-value)
"")
|
︙ | | |
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
|
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
|
+
-
+
|
(not (equal? (vector-ref fromrow i)(vector-ref curr i))))
(set! same #f))
(if (and same
(< i (- num-fields 1)))
(loop (+ i 1))))
(if (not same)
(begin
(debug:print 0 *default-log-port* "applying data "fromrow"to table "tablename", numrecs="numrecs)
(apply sqlite3:execute stmth (vector->list fromrow))
(hash-table-set! numrecs tablename (+ 1 (hash-table-ref/default numrecs tablename 0)))))))
fromdat-lst))))
fromdats)
(sqlite3:finalize! stmth)
(if (member "last_update" field-names)
(db:create-trigger db tablename)))))
;; tbls is ( ("tablename" ( "field1" [#f|proc1] ) ( "field2" [#f|proc2] ) .... ) )
;; db's are sqlite3 handles
;;
;; if last-update specified ("field-name" . time-in-seconds)
;; then sync only records where field-name >= time-in-seconds
;; IFF field-name exists
;;
(define (db:sync-tables tbls last-update fromdb todb)
(assert (pair? last-update) "FATAL: last-update must always be a pair")
;; NOTE: I'm moving all the checking OUT of this routine. Check for read/write access, existance, etc
;; BEFORE calling this sync
(let ((stmts (make-hash-table)) ;; table-field => stmt
(all-stmts '()) ;; ( ( stmt1 value1 ) ( stml2 value2 ))
(numrecs (make-hash-table))
(start-time (current-milliseconds))
|
︙ | | |
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
|
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
|
-
+
|
(if should-print (debug:print 3 *default-log-port* "INFO: db sync, total run time " runtime " ms"))
(for-each
(lambda (dat)
(let ((tblname (car dat))
(count (cdr dat)))
(set! tot-count (+ tot-count count))
(if (> count 0)
(if should-print (debug:print 0 *default-log-port* (format #f " ~10a ~5a" tblname count))))))
(if should-print (debug:print 0 *default-log-port* " "tblname" "count))))) ;; (format #f " ~10a ~5a" tblname count))))))
(sort (hash-table->alist numrecs)(lambda (a b)(> (cdr a)(cdr b))))))
tot-count))
(define (db:patch-schema-rundb frundb)
;;
;; remove this some time after September 2016 (added in version v1.6031
;;
|
︙ | | |
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
|
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
|
-
+
|
(string->number res)
#f)))
(if newres
newres
res))
res)))
(define (db:no-sync-close-db db stmt-cache)
#;(define (db:no-sync-close-db db stmt-cache)
(db:safely-close-sqlite3-db db stmt-cache))
;; transaction protected lock aquisition
;; either:
;; fails returns (#f . lock-creation-time)
;; succeeds (returns (#t . lock-creation-time)
;; use (db:no-sync-del! db keyname) to release the lock
|
︙ | | |
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
|
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
|
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
(debug:print 2 *default-log-port* "NOTE: using target " (string-intersperse (map cadr keyvals) "/") " for this run")
(if (and runname (null? (filter (lambda (x)(not x)) keyvals))) ;; there must be a better way to "apply and"
(db:with-db
dbstruct #f #f
(lambda (db)
(let ((res #f))
(apply sqlite3:execute db (conc "INSERT OR IGNORE INTO runs (runname,state,status,owner,event_time,contour" comma keystr ") VALUES (?,?,?,?,strftime('%s','now'),?" comma valslots ");")
allvals)
(apply sqlite3:for-each-row
(lambda (id)
(set! res id))
db
(let ((qry (conc "SELECT id FROM runs WHERE (runname=? " andstr key=?str ");")))
qry)
qryvals)
(sqlite3:execute db "UPDATE runs SET state=?,status=?,event_time=strftime('%s','now') WHERE id=? AND state='deleted';" state status res)
res)))
(begin
(debug:print-error 0 *default-log-port* "Called without all necessary keys")
#f))))
;; register a test run with the db, this accesses the main.db and does NOT
;; use server api
;;
(define (db:insert-run dbstruct run-id keyvals runname state status user contour-in)
(let* ((keys (map car keyvals))
(keystr (keys->keystr keys))
(contour (or contour-in "")) ;; empty string to force no hierarcy and be backwards compatible.
(comma (if (> (length keys) 0) "," ""))
(andstr (if (> (length keys) 0) " AND " ""))
(valslots (keys->valslots keys)) ;; ?,?,? ...
(allvals (append (list runname state status user contour) (map cadr keyvals)))
(qryvals (append (list runname) (map cadr keyvals)))
(key=?str (string-intersperse (map (lambda (k)(conc k "=?")) keys) " AND ")))
(debug:print 3 *default-log-port* "keys: " keys " allvals: " allvals " keyvals: " keyvals " key=?str is " key=?str)
(debug:print 2 *default-log-port* "NOTE: using target " (string-intersperse (map cadr keyvals) "/") " for this run")
(if (and runname (null? (filter (lambda (x)(not x)) keyvals))) ;; there must be a better way to "apply and"
(db:with-db
dbstruct #f #f
(lambda (db)
(let ((res #f))
(apply sqlite3:execute db (conc "INSERT OR IGNORE INTO runs (id,runname,state,status,owner,event_time,contour" comma keystr ") VALUES (?,?,?,?,?,strftime('%s','now'),?" comma valslots ");")
run-id
allvals)
(apply sqlite3:for-each-row
(lambda (id)
(set! res id))
db
(let ((qry (conc "SELECT id FROM runs WHERE (runname=? " andstr key=?str ");")))
qry)
|
︙ | | |
5535
5536
5537
5538
5539
5540
5541
5542
5543
5544
5545
5546
5547
5548
5549
5550
5551
5552
5553
5554
5555
|
5575
5576
5577
5578
5579
5580
5581
5582
5583
5584
5585
5586
5587
5588
5589
5590
5591
5592
5593
5594
5595
5596
5597
5598
5599
5600
5601
5602
5603
5604
5605
5606
5607
5608
5609
5610
5611
5612
5613
5614
5615
5616
5617
5618
5619
5620
5621
5622
5623
5624
5625
5626
5627
5628
|
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
+
|
(debug:print-info 0 *default-log-port* "Server already running at "sinfo ", while trying to register server " host":"port)
#f) ;; server already registered
(begin
(sqlite3:execute db "INSERT INTO servers (host,port,servkey,pid,ipaddr,apath,dbname) VALUES (?,?,?,?,?,?,?);"
host port servkey pid ipaddr apath dbname)
(db:get-server-info dbstruct apath dbname)))))))))
;; run this one in a transaction where first check if host:port is taken
(define (db:deregister-server dbstruct host port servkey pid ipaddr apath dbname)
(db:with-db
dbstruct
#f #f
(lambda (db)
(sqlite3:with-transaction
db
(lambda ()
(let* ((sinfo (db:get-server-info dbstruct apath dbname)))
(if (not sinfo)
(begin
(debug:print-info 0 *default-log-port* "Server already removed for "apath", "dbname) ;; at "sinfo ", while trying to register server " host":"port)
#f) ;; server already deregistered
(begin
(sqlite3:execute db "DELETE FROM servers WHERE apath=? AND dbname=?;" ;; (host,port,servkey,pid,ipaddr,apath,dbname) VALUES (?,?,?,?,?,?,?);"
;; host port servkey pid ipaddr
apath dbname)
#;(db:get-server-info dbstruct apath dbname)))))))))
(define (db:get-server-info dbstruct apath dbname)
(db:with-db
dbstruct
#f #f
(lambda (db)
(sqlite3:fold-row
(lambda (res host port servkey pid ipaddr apath dbpath)
(list host port servkey pid ipaddr apath dbpath))
#f
db
"SELECT host,port,servkey,pid,ipaddr,apath,dbname FROM servers WHERE apath=? AND dbname=?;"
apath dbname))))
(define (db:get-count-servers dbstruct apath)
(db:with-db
dbstruct
#f #f
(lambda (db)
(sqlite3:fold-row
(lambda (res count)
(max res count))
0
db
"SELECT count(*) FROM servers WHERE apath=?;"
apath))))
)
)
|