Index: common.scm ================================================================== --- common.scm +++ common.scm @@ -149,17 +149,17 @@ (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* ;; task db (define *task-db* #f) ;; (vector db path-to-db) (define *db-access-allowed* #t) ;; flag to allow access -(define *db-access-mutex* (make-mutex)) +;; (define *db-access-mutex* (make-mutex)) ;; moved to dbfile (define *db-transaction-mutex* (make-mutex)) (define *db-cache-path* #f) (define *db-with-db-mutex* (make-mutex)) (define *db-api-call-time* (make-hash-table)) ;; hash of command => (list of times) ;; no sync db -(define *no-sync-db* #f) +;; (define *no-sync-db* #f) ;; moved to dbfile ;; SERVER (define *my-client-signature* #f) (define *transport-type* 'http) ;; override with [server] transport http|rpc|nmsg (define *runremote* #f) ;; if set up for server communication this will hold Index: db.scm ================================================================== --- db.scm +++ db.scm @@ -2109,69 +2109,12 @@ ;;====================================================================== (define (db:open-no-sync-db) (dbfile:open-no-sync-db (db:dbfile-path))) -;; if we are not a server create a db handle. this is not finalized -;; so watch for problems. I'm still not clear if it is needed to manually -;; finalize sqlite3 dbs with the sqlite3 egg. -;; -(define (db:no-sync-db db-in) - (mutex-lock! *db-access-mutex*) - (let ((res (if db-in - db-in - (let ((db (db:open-no-sync-db))) - (set! *no-sync-db* db) - db)))) - (mutex-unlock! *db-access-mutex*) - res)) - -(define (db:no-sync-set db var val) - (sqlite3:execute (db:no-sync-db db) "INSERT OR REPLACE INTO no_sync_metadat (var,val) VALUES (?,?);" var val)) - -(define (db:no-sync-del! db var) - (sqlite3:execute (db:no-sync-db db) "DELETE FROM no_sync_metadat WHERE var=?;" var)) - -(define (db:no-sync-get/default db var default) - (let ((res default)) - (sqlite3:for-each-row - (lambda (val) - (set! res val)) - (db:no-sync-db db) - "SELECT val FROM no_sync_metadat WHERE var=?;" - var) - (if res - (let ((newres (if (string? res) - (string->number res) - #f))) - (if newres - newres - res)) - res))) - (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 -;; -(define (db:no-sync-get-lock db-in keyname) - (let ((db (db:no-sync-db db-in))) - (sqlite3:with-transaction - db - (lambda () - (handle-exceptions - exn - (let ((lock-time (current-seconds))) - (debug:print-info 2 *default-log-port* "db:no-sync-get-lock keyname=" keyname ", lock-time=" lock-time ", exn=" exn) - (sqlite3:execute db "INSERT INTO no_sync_metadat (var,val) VALUES(?,?);" keyname lock-time) - `(#t . ,lock-time)) - `(#f . ,(sqlite3:first-result db "SELECT val FROM no_sync_metadat WHERE var=?;" keyname))))))) - ;; use a global for some primitive caching, it is just silly to ;; re-read the db over and over again for the keys since they never ;; change @@ -5050,15 +4993,49 @@ (let* ((lockdat (db:no-sync-get-lock no-sync-db from-db)) (gotlock (car lockdat)) (locktime (cdr lockdat))) (if gotlock (begin - (file-copy from-db to-db) - (db:no-sync-del! no-sync-db keyname) + (file-copy from-db to-db #t) + (db:no-sync-del! no-sync-db from-db) #t) #f))) +;; sync for filesystem local db writes +;; +(define (db:run-lock-and-sync no-sync-db) + (let* ((tmp-area (common:get-db-tmp-area)) + (dbfiles (glob (conc tmp-area"/.db/*.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*"/.db/"fname)) + (time1 (file-modification-time file)) + (time2 (file-modification-time fulln)) + (changed (> time1 time2)) + (do-cp (cond + ((not (file-exists? fulln)) ;; shouldn't happen, but this might recover + (debug:print-info 0 *default-log-port* "File "fulln" not found! Copying "fname" to "fulln) + #t) + (changed ;; (and changed + ;; (> (- (current-seconds) time1) 3)) ;; if file is changed and three seconds have passed. + #t) + ((and changed *time-to-exit*) ;; last copy + #t) + (else + #f)))) + (if do-cp + (let* ((start-time (current-milliseconds))) + (debug:print-info 0 *default-log-port* "sync file: "file", fname: "fname", time1: "time1", time2: "time2) + (db:lock-and-sync no-sync-db file fulln) + (hash-table-set! sync-durations (conc fname".db") (- (current-milliseconds) start-time))) + #;(debug:print-info 0 *default-log-port* "skipping sync...")))) + dbfiles) + (hash-table->alist sync-durations))) + ;; straight forward copy based sync ;; 1. for each .db fil ;; 2. next if file changed since last sync cycle ;; 2. next if time delta /tmp file to MTRA less than 3 seconds ;; 3. get a lock for the file in nosyncdb @@ -5073,47 +5050,22 @@ (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))) + ;; Sync moved to http-transport keep-running loop (set! *no-sync-db* no-sync-db) ;; make the no sync db available to api calls (debug:print-info 2 *default-log-port* "Periodic copy-based sync thread started. sync is "legacy-sync", tmp-area is "tmp-area) (debug:print-info 3 *default-log-port* "watchdog starting. legacy-sync is "legacy-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.") (let loop () - ;; sync for filesystem local db writes - ;; - (let* ((dbfiles (glob (conc tmp-area"/.db/*.db")))) - (debug:print-info 0 "dbfiles: "dbfiles) - (for-each - (lambda (file) - (let* ((fname (pathname-file file)) - (fulln (conc *top-level*"/.db/"fname)) - (time1 (file-modification-time fname)) - (time2 (file-modification-time fulln)) - (changed (>= time1 time2)) - (do-cp (cond - ((not (file-exists? fulln)) ;; shouldn't happen, but this might recover - (debug:print-info 0 "File "fulln" not found! Copying "fname" to "fulln) - #t) - (changed ;; (and changed - ;; (> (- (current-seconds) time1) 3)) ;; if file is changed and three seconds have passed. - #t) - ((and changed *time-to-exit*) ;; last copy - #t) - (else - #f)))) - (debug:print-info 0 "file: "file", fname: "fname", time1: "time1", time2: "time2) - (if do-cp - (let* ((start-time (current-milliseconds))) - (db:lock-and-sync no-sync-db fname fulln) - (set! sync-duration (- (current-milliseconds) start-time))) - (debug:print-info 0 "skipping sync...")))) - dbfile)) + + ;; run the sync and print out durations + (debug:print-info 0 *default-log-port* "Sync durations: "(db:run-lock-and-sync no-sync-db)) ;; keep going unless time to exit ;; (if (not *time-to-exit*) (let delay-loop ((count 0)) ;;(debug:print-info 13 *default-log-port* "delay-loop top; count="count" pid="(current-process-id)" this-wd-num="this-wd-num" *time-to-exit*="*time-to-exit*) Index: dbfile.scm ================================================================== --- dbfile.scm +++ dbfile.scm @@ -73,10 +73,12 @@ (dbh #f) (stmt-cache (make-hash-table)) (read-only #f)) (define *dbstruct-dbs* #f) +(define *db-access-mutex* (make-mutex)) +(define *no-sync-db* #f) (define (dbfile:run-id->key run-id) (or run-id 'main)) (define (db:safely-close-sqlite3-db db stmt-cache #!key (try-num 3)) @@ -437,10 +439,24 @@ tmpdb)) ;;====================================================================== ;; no-sync.db - small bits of data to be shared between servers ;;====================================================================== + +;; if we are not a server create a db handle. this is not finalized +;; so watch for problems. I'm still not clear if it is needed to manually +;; finalize sqlite3 dbs with the sqlite3 egg. +;; +(define (db:no-sync-db db-in) + (mutex-lock! *db-access-mutex*) + (let ((res (if db-in + db-in + (let ((db (dbfile:open-no-sync-db))) + (set! *no-sync-db* db) + db)))) + (mutex-unlock! *db-access-mutex*) + res)) (define (dbfile:open-no-sync-db dbpath) (let* (;; (dbpath (db:dbfile-path)) (dbname (conc dbpath "/no-sync.db")) (db-exists (file-exists? dbname)) @@ -450,10 +466,53 @@ (begin (sqlite3:execute db "PRAGMA synchronous = 0;") (sqlite3:execute db "CREATE TABLE IF NOT EXISTS no_sync_metadat (var TEXT,val TEXT, CONSTRAINT no_sync_metadat_constraint UNIQUE (var));") (sqlite3:execute db "PRAGMA journal_mode=WAL;"))) db)) + +(define (db:no-sync-set db var val) + (sqlite3:execute (db:no-sync-db db) "INSERT OR REPLACE INTO no_sync_metadat (var,val) VALUES (?,?);" var val)) + +(define (db:no-sync-del! db var) + (sqlite3:execute (db:no-sync-db db) "DELETE FROM no_sync_metadat WHERE var=?;" var)) + +(define (db:no-sync-get/default db var default) + (let ((res default)) + (sqlite3:for-each-row + (lambda (val) + (set! res val)) + (db:no-sync-db db) + "SELECT val FROM no_sync_metadat WHERE var=?;" + var) + (if res + (let ((newres (if (string? res) + (string->number res) + #f))) + (if newres + newres + res)) + res))) + +;; 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 +;; +(define (db:no-sync-get-lock db-in keyname) + (let ((db (db:no-sync-db db-in))) + (sqlite3:with-transaction + db + (lambda () + (handle-exceptions + exn + (let ((lock-time (current-seconds))) + ;; (debug:print-info 2 *default-log-port* "db:no-sync-get-lock keyname=" keyname ", lock-time=" lock-time ", exn=" exn) + (sqlite3:execute db "INSERT INTO no_sync_metadat (var,val) VALUES(?,?);" keyname lock-time) + `(#t . ,lock-time)) + `(#f . ,(sqlite3:first-result db "SELECT val FROM no_sync_metadat WHERE var=?;" keyname))))))) + ;;====================================================================== ;; file utils ;;====================================================================== Index: http-transport.scm ================================================================== --- http-transport.scm +++ http-transport.scm @@ -35,14 +35,17 @@ (declare (uses tasks)) ;; tasks are where stuff is maintained about what is running. (declare (uses server)) ;; (declare (uses daemon)) (declare (uses portlogger)) (declare (uses rmt)) +(declare (uses dbfile)) (include "common_records.scm") (include "db_records.scm") (include "js-path.scm") + +(import dbfile) (require-library stml) (define (http-transport:make-server-url hostport) (if (not hostport) #f @@ -464,11 +467,13 @@ (begin (debug:print 0 *default-log-port* "SERVER: dbprep") (set! *dbstruct-dbs* (db:setup #t)) ;; run-id)) FIXME!!! (set! server-going #t) (debug:print 0 *default-log-port* "SERVER: running, megatest version: " (common:get-full-version)) ;; NOTE: the server is NOT yet marked as running in the log. We do that in the keep-running routine. - (thread-start! *watchdog*))) + (thread-start! *watchdog*)) + (if *no-sync-db* + (db:run-lock-and-sync *no-sync-db*))) ;; when things go wrong we don't want to be doing the various queries too often ;; so we strive to run this stuff only every four seconds or so. (let* ((sync-time (- (current-milliseconds) start-time)) (rem-time (quotient (- 4000 sync-time) 1000))) @@ -490,11 +495,10 @@ (debug:print-info 0 *default-log-port* "WARNING: interface changed, refreshing iface and port info") (set! iface new-iface) (set! port new-port) (if (not *server-id*) (set! *server-id* (server:mk-signature))) - (debug:print-info 0 *default-log-port* (current-seconds)" "(current-directory)" " (current-process-id) (argv)) (debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds) " server-id: " *server-id*) (flush-output *default-log-port*))) ;; Transfer *db-last-access* to last-access to use in checking that we are still alive (mutex-lock! *heartbeat-mutex*) @@ -503,11 +507,10 @@ (if (common:low-noise-print 120 (conc "server running on " iface ":" port)) (begin (if (not *server-id*) (set! *server-id* (server:mk-signature))) - (debug:print 0 *default-log-port* (current-seconds) (current-directory) (current-process-id) (argv)) (debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds) " server-id: " *server-id*) (flush-output *default-log-port*))) (if (common:low-noise-print 60 "dbstats") (begin (debug:print 0 *default-log-port* "Server stats:")