Changes In Branch interleaved-queries Through [81e546a994] Excluding Merge-Ins
This is equivalent to a diff from 8c476e8627 to 81e546a994
2012-11-16
| ||
01:19 | (no comment) check-in: d1245270ea user: matt tags: interleaved-queries | |
2012-11-15
| ||
23:42 | Partially complete, just taking a snapshot check-in: 81e546a994 user: matt tags: interleaved-queries | |
01:44 | Bit's 'n pieces check-in: d0462389b2 user: matt tags: interleaved-queries | |
2012-11-12
| ||
19:50 | Cherrypicked the fix to building for deploy check-in: e4ac93792c user: matt tags: trunk | |
00:04 | Interleaving query enablement, server side coded check-in: 4980ca9f98 user: matt tags: interleaved-queries | |
2012-11-06
| ||
09:44 | Testing ordering of loading zmq, fixes to deploy script check-in: 8c476e8627 user: mrwellan tags: trunk | |
06:21 | Added deploy script check-in: af55ffc7d7 user: matt tags: trunk | |
Modified db.scm from [108dfe0472] to [c6e157479f].
︙ | ︙ | |||
9 10 11 12 13 14 15 | ;; PURPOSE. ;;====================================================================== ;;====================================================================== ;; Database access ;;====================================================================== | | | | | 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | ;; PURPOSE. ;;====================================================================== ;;====================================================================== ;; Database access ;;====================================================================== (require-extension (srfi 18) extras tcp) ;; rpc) ;; (import (prefix rpc rpc:)) (use sqlite3 srfi-1 posix regex regex-case srfi-69 csv-xml s11n md5 message-digest) (import (prefix sqlite3 sqlite3:)) (use zmq) (declare (unit db)) (declare (uses common)) (declare (uses keys)) |
︙ | ︙ | |||
1179 1180 1181 1182 1183 1184 1185 | (define (cdb:use-non-blocking-mode proc) (set! *client-non-blocking-mode* #t) (let ((res (proc))) (set! *client-non-blocking-mode* #f) res)) ;; params = 'target cached remparams | | | > > > | | < | < < | | | > | | > > | | | | 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 | (define (cdb:use-non-blocking-mode proc) (set! *client-non-blocking-mode* #t) (let ((res (proc))) (set! *client-non-blocking-mode* #f) res)) ;; params = 'target cached remparams (define (cdb:client-call zmq-sockets . params) (debug:print-info 11 "cdb:client-call zmq-sockets=" zmq-sockets " params=" params) (let* ((push-socket (vector-ref zmq-sockets 0)) (sub-socket (vector-ref zmq-sockets 1)) (query-id (conc (server:get-client-signature) "-" (message-digest-string (md5-primitive) (conc params)))) (zdat (db:obj->string (vector query-id params))) ;; (with-output-to-string (lambda ()(serialize params)))) (res #f) (get-res (lambda () (db:string->obj (if *client-non-blocking-mode* (receive-message* sub-socket) (receive-message sub-socket)))))) (send-message push-socket zdat) (let loop ((res (get-res))) (if res res (begin (thread-sleep! 0.5) (get-res)))))) (define (cdb:set-verbosity zmq-socket val) (cdb:client-call zmq-socket 'set-verbosity #f val)) (define (cdb:login zmq-sockets keyval signature) (cdb:client-call zmq-sockets 'login #t keyval megatest-version signature)) (define (cdb:logout zmq-socket keyval signature) (cdb:client-call zmq-socket 'logout #t keyval signature)) (define (cdb:num-clients zmq-socket) (cdb:client-call zmq-socket 'numclients #t)) |
︙ | ︙ |
Modified server.scm from [c2beddecc4] to [01f2819585].
1 2 3 4 5 6 7 8 9 10 | ;; 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. | | | > > > > > > > > > > > > > > > > > > > > > > > > > | > > > > | | | > > > > > > > | > > > | | | | | | | | | < < | | | | > | > | > > > > | < < < < | < < < < > > > | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 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 152 153 154 155 156 157 | ;; 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. (require-extension (srfi 18) extras tcp s11n) ;; (import (prefix rpc rpc:)) (use sqlite3 srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest) (import (prefix sqlite3 sqlite3:)) (use zmq) (declare (unit server)) (declare (uses common)) (declare (uses db)) (declare (uses tests)) (declare (uses tasks)) ;; tasks are where stuff is maintained about what is running. (include "common_records.scm") (include "db_records.scm") ;; Transition to pub --> sub with pull <-- push ;; ;; 1. client sends request to server via push to the pull port ;; 2. server puts request in queue or processes immediately as appropriate ;; 3. server puts responses from completed requests into pub port ;; ;; TODO ;; ;; Done Tested ;; [x] [ ] 1. Add columns pullport pubport to servers table ;; [x] [ ] 2. Add rm of monitor.db if older than 11/12/2012 ;; [x] [ ] 3. Add create of pullport and pubport with finding of available ports ;; [ ] [ ] 4. Add client compose of request ;; [ ] [ ] - name of client: testname/itempath-test_id-hostname ;; [ ] [ ] - name of request: callname, params ;; [ ] [ ] - request key: f(clientname, callname, params) ;; [ ] [ ] 5. Add processing of subscription hits ;; [ ] [ ] - done when get key ;; [ ] [ ] - return results ;; [ ] [ ] 6. Add timeout processing ;; [ ] [ ] - after 60 seconds ;; [ ] [ ] i. check server alive, connect to new if necessary ;; [ ] [ ] ii. resend request ;; [ ] [ ] 7. Turn self ping back on (define (server:make-server-url hostport) (if (not hostport) #f (conc "tcp://" (car hostport) ":" (cadr hostport)))) (define *server-loop-heart-beat* (current-seconds)) (define *heartbeat-mutex* (make-mutex)) (define (server:self-ping server-info) ;; server-info: server-id interface pullport pubport (let ((iface (list-ref server-info 1)) (pullport (list-ref server-info 2)) (pubport (list-ref server-info 3))) (server:client-connect iface pullport pubport) (let loop () (thread-sleep! 2) (cdb:client-call *runremote* 'ping #t) (debug:print 4 "server:self-ping - I'm alive on " iface ":" pullport "/" pubport "!") (mutex-lock! *heartbeat-mutex*) (set! *server-loop-heart-beat* (current-seconds)) (mutex-unlock! *heartbeat-mutex*) (loop)))) (define-inline (zmqsock:get-pub dat)(vector-ref dat 0)) (define-inline (zmqsock:get-pull dat)(vector-ref dat 1)) (define-inline (zmqsock:set-pub! dat s)(vector-set! dat s 0)) (define-inline (zmqsock:set-pull! dat s)(vector-set! dat s 0)) (define (server:run hostn) (debug:print 0 "Attempting to start the server ...") (if (not *toppath*) (if (not (setup-for-run)) (begin (debug:print 0 "ERROR: cannot find megatest.config, cannot start server, exiting") (exit)))) (let* ((zmq-sdat1 #f) (zmq-sdat2 #f) (pull-socket #f) (pub-socket #f) (p1 #f) (p2 #f) (zmq-sockets-dat #f) (iface (if (string=? "-" hostn) "*" ;; (get-host-name) hostn)) (hostname (get-host-name)) (ipaddrstr (let ((ipstr (if (string=? "-" hostn) (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".") #f))) (if ipstr ipstr hostname)))) (set! zmq-sockets-dat (server:setup-ports ipaddrstr (if (args:get-arg "-port") (string->number (args:get-arg "-port")) (+ 5000 (random 1001))))) (set! zmq-sdat1 (car zmq-sockets-dat)) (set! pull-socket (cadr zmq-sdat1)) ;; (iface s port) (set! p1 (caddr zmq-sdat1)) (set! zmq-sdat2 (cadr zmq-sockets-dat)) (set! pub-socket (cadr zmq-sdat2)) (set! p2 (caddr zmq-sdat2)) (set! *cache-on* #t) ;; (set! th1 (make-thread (lambda () ;; (server:self-ping ipaddrstr actual-port)))) ;; (thread-start! th1) ;; what to do when we quit ;; (on-exit (lambda () (if (and *toppath* *server-info*) (begin (open-run-close tasks:server-deregister-self tasks:open-db ipaddrstr p1 p2)) (let loop () (let ((queue-len 0)) (thread-sleep! (random 5)) (mutex-lock! *incoming-mutex*) (set! queue-len (length *incoming-data*)) (mutex-unlock! *incoming-mutex*) (if (> queue-len 0) (begin (debug:print-info 0 "Queue not flushed, waiting ...") (loop)))))))) ;; The heavy lifting ;; (let loop () (let* ((rawmsg (receive-message* pull-socket)) (params (db:string->obj rawmsg)) ;; (with-input-from-string rawmsg (lambda ()(deserialize)))) (res #f)) (debug:print-info 12 "server=> received params=" params) (set! res (cdb:cached-access params)) (debug:print-info 12 "server=> processed res=" res) ;; need address here ;; (send-message zmq-socket (db:obj->string res)) (if (not *time-to-exit*) (loop) (begin (open-run-close tasks:server-deregister-self tasks:open-db #f) (db:write-cached-data) (exit) |
︙ | ︙ | |||
145 146 147 148 149 150 151 | (mutex-unlock! *heartbeat-mutex*) ;; The logic here is that if the server loop gets stuck blocked in working ;; we don't want to update our heartbeat (set! pulse (- (current-seconds) server-loop-heartbeat)) (debug:print-info 2 "Heartbeat period is " pulse " seconds on " (cadr server-info) ":" (caddr server-info) ", last db access is " (- (current-seconds) *last-db-access*) " seconds ago") (if (> pulse 15) ;; must stay less than 10 seconds (begin | | | 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | (mutex-unlock! *heartbeat-mutex*) ;; The logic here is that if the server loop gets stuck blocked in working ;; we don't want to update our heartbeat (set! pulse (- (current-seconds) server-loop-heartbeat)) (debug:print-info 2 "Heartbeat period is " pulse " seconds on " (cadr server-info) ":" (caddr server-info) ", last db access is " (- (current-seconds) *last-db-access*) " seconds ago") (if (> pulse 15) ;; must stay less than 10 seconds (begin (open-run-close tasks:server-deregister tasks:open-db (cadr server-info) pullport: (caddr server-info)) (debug:print 0 "ERROR: Heartbeat failed, committing servercide") (exit)) (open-run-close tasks:server-update-heartbeat tasks:open-db (car server-info))) ;; (if ;; (or (> numrunning 0) ;; stay alive for two days after last access (if (> (+ *last-db-access* ;; (* 48 60 60) ;; 48 hrs ;; 60 ;; one minute |
︙ | ︙ | |||
170 171 172 173 174 175 176 | (set! *time-to-exit* #t) (open-run-close tasks:server-deregister-self tasks:open-db (get-host-name)) (thread-sleep! 1) (debug:print-info 0 "Max cached queries was " *max-cache-size*) (debug:print-info 0 "Server shutdown complete. Exiting") (exit))))))) | | | | > | > > > > > > > | | | | | | | | | > > > > < | | > > > > > > > > > > > > > > > > > > | | | > | | < < < < < < < < < < < < | 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 337 338 | (set! *time-to-exit* #t) (open-run-close tasks:server-deregister-self tasks:open-db (get-host-name)) (thread-sleep! 1) (debug:print-info 0 "Max cached queries was " *max-cache-size*) (debug:print-info 0 "Server shutdown complete. Exiting") (exit))))))) (define (server:find-free-port-and-open iface s port stype #!key (trynum 50)) (let ((s (if s s (make-socket stype))) (p (if (number? port) port 5555)) (old-handler (current-exception-handler))) (handle-exceptions exn (begin (debug:print 0 "Failed to bind to port " p ", trying next port") (debug:print 0 " EXCEPTION: " ((condition-property-accessor 'exn 'message) exn)) ;; (old-handler) ;; (print-call-chain) (if (> trynum 0) (server:find-free-port-and-open iface s (+ p 1) trynum: (- trynum 1)) (debug:print-info 0 "Tried ports up to " p " but all were in use. Please try a different port range by starting the server with parameter \" -port N\" where N is the starting port number to use")) (exit)) ;; To exit or not? That is the question. (let ((zmq-url (conc "tcp://" iface ":" p))) (debug:print 0 "Trying to start server on " zmq-url) (bind-socket s zmq-url) (list iface s port))))) (define (server:setup-ports ipaddrstr startport) (let* ((s1 (server:find-free-port-and-open ipaddrstr #f startport 'pub)) (p1 (caddr s1)) (s2 (server:find-free-port-and-open ipaddrstr #f (+ 1 (if p1 p1 (+ startport 1))) 'pull)) (p2 (caddr s2))) (set! *runremote* #f) (debug:print 0 "Server started on " ipaddrstr " ports " p1 " and " p2) (mutex-lock! *heartbeat-mutex*) (set! *server-info* (open-run-close tasks:server-register tasks:open-db (current-process-id) ipaddrstr p1 p2 0 'live)) (mutex-unlock! *heartbeat-mutex*) (list s1 s2))) (define (server:mk-signature) (message-digest-string (md5-primitive) (with-output-to-string (lambda () (write (list (current-directory) (argv))))))) (define (server:get-client-signature) (if *my-client-signature* *my-client-signature* (let ((sig (server:mk-signature))) (set! *my-client-signature* sig) *my-client-signature*))) ;; (define (server:client-socket-connect iface port #!key (context #f)(type 'req)(subscriptions '())) (debug:print-info 3 "client-connect " iface ":" port) (let ((connect-ok #f) (zmq-socket (if context (make-socket type context) (make-socket type))) (conurl (server:make-server-url (list iface port)))) (if (socket? zmq-socket) (begin ;; first apply subscriptions (for-each (lambda (subscription) (socket-options-set! zmq-socket 'subscribe subscription)) subscriptions) (connect-socket zmq-socket conurl) zmq-socket) #f))) (define (server:client-login zmq-sockets) (cdb:login zmq-sockets *toppath* (server:get-client-signature))) (define (server:client-logout zmq-socket) (let ((ok (and (socket? zmq-socket) (cdb:logout zmq-socket *toppath* (server:get-client-signature))))) ;; (close-socket zmq-socket) ok)) (define (server:client-connect iface pullport pubport) (let* ((push-socket (server:client-socket-connect iface pullport 'push)) (sub-socket (server:client-socket-connect iface pubport 'sub subscriptions: (list (server:get-client-signature) "all"))) (zmq-sockets (vector push-socket sub-socket)) (login-res #f)) (set! login-res (server:client-login zmq-sockets)) (if (and (not (null? login-res)) (car login-res)) (begin (debug:print-info 2 "Logged in and connected to " iface ":" pullport "/" pubport ".") (set! *runremote* zmq-socket) #t) (begin (debug:print-info 2 "Failed to login or connect to " conurl) (set! *runremote* #f) #f)))) ;; Do all the connection work, start a server if not already running (define (server:client-setup #!key (numtries 50)) (if (not *toppath*) (if (not (setup-for-run)) (begin (debug:print 0 "ERROR: failed to find megatest.config, exiting") (exit)))) (let ((hostinfo (open-run-close tasks:get-best-server tasks:open-db))) (if hostinfo (let ((host (list-ref hostinfo 0)) (iface (list-ref hostinfo 1)) (pullport (list-ref hostinfo 2)) (pubport (list-ref hostinfo 3))) (debug:print-info 2 "Setting up to connect to " hostinfo) (handle-exceptions exn (begin ;; something went wrong in connecting to the server. In this scenario it is ok ;; to try again (debug:print 0 "ERROR: Failed to open a connection to the server at: " hostinfo) (debug:print 0 " EXCEPTION: " ((condition-property-accessor 'exn 'message) exn)) (debug:print 0 " perhaps jobs killed with -9? Removing server records") (open-run-close tasks:server-deregister tasks:open-db host pullport: pullport) (server:client-setup (- numtries 1)) #f) (server:client-connect iface pullport pubport))) (if (> numtries 0) (let ((exe (car (argv)))) (debug:print-info 1 "No server available, attempting to start one...") (process-run exe (list "-server" "-" "-debug" (conc *verbosity*))) (sleep 5) ;; give server time to start ;; we are starting a server, do not try again! That can lead to ;; recursively starting many processes!!! |
︙ | ︙ | |||
302 303 304 305 306 307 308 | (debug:print-info 1 "Waiting for the server to come online before starting heartbeat") (thread-sleep! 2) (mutex-lock! *heartbeat-mutex*) (set! server-info *server-info* ) (mutex-unlock! *heartbeat-mutex*) (if (not server-info)(loop))) (debug:print 1 "Server alive, starting self-ping") | | > | 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 | (debug:print-info 1 "Waiting for the server to come online before starting heartbeat") (thread-sleep! 2) (mutex-lock! *heartbeat-mutex*) (set! server-info *server-info* ) (mutex-unlock! *heartbeat-mutex*) (if (not server-info)(loop))) (debug:print 1 "Server alive, starting self-ping") (server:self-ping server-info))) "Self ping")) (th2 (make-thread (lambda () (server:run (args:get-arg "-server"))) "Server run")) (th3 (make-thread (lambda () (server:keep-running)) "Keep running"))) (set! *client-non-blocking-mode* #t) (thread-start! th1) (thread-start! th2) |
︙ | ︙ | |||
325 326 327 328 329 330 331 | exn (debug:print " ... exiting ...") (let ((th1 (make-thread (lambda () (if (not *received-response*) (receive-message* *runremote*))) ;; flush out last call if applicable "eat response")) (th2 (make-thread (lambda () | | | 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 | exn (debug:print " ... exiting ...") (let ((th1 (make-thread (lambda () (if (not *received-response*) (receive-message* *runremote*))) ;; flush out last call if applicable "eat response")) (th2 (make-thread (lambda () (debug:print 0 "ERROR: Received ^C, attempting clean exit. Please be patient and wait a few seconds before hitting ^C again.") (thread-sleep! 3) ;; give the flush three seconds to do it's stuff (debug:print 0 " Done.") (exit 4)) "exit on ^C timer"))) (thread-start! th2) (thread-start! th1) (thread-join! th2)))) |
︙ | ︙ |
Modified tasks.scm from [53e3d6e8bd] to [3c212dbf03].
︙ | ︙ | |||
20 21 22 23 24 25 26 | ;;====================================================================== ;; Tasks db ;;====================================================================== (define (tasks:open-db) (let* ((dbpath (conc *toppath* "/monitor.db")) | | > > > > > > > > | 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | ;;====================================================================== ;; Tasks db ;;====================================================================== (define (tasks:open-db) (let* ((dbpath (conc *toppath* "/monitor.db")) (exists (if (file-exists? dbpath) ;; BUGGISHNESS: Remove this code in six months. Today is 11/13/2012 (if (< (file-change-time dbpath) 1352851396.0) (begin (debug:print 0 "NOTE: removing old db file " dbpath) (delete-file dbpath) #f) #t) #f)) (mdb (sqlite3:open-database dbpath)) ;; (never-give-up-open-db dbpath)) (handler (make-busy-timeout 36000))) (sqlite3:set-busy-handler! mdb handler) (sqlite3:execute mdb (conc "PRAGMA synchronous = 0;")) (if (not exists) (begin (sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS tasks_queue (id INTEGER PRIMARY KEY, |
︙ | ︙ | |||
50 51 52 53 54 55 56 | hostname TEXT, username TEXT, CONSTRAINT monitors_constraint UNIQUE (pid,hostname));") (sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS servers (id INTEGER PRIMARY KEY, pid INTEGER, interface TEXT, hostname TEXT, | | > | | | > | | > | | | | | | | | | | | 58 59 60 61 62 63 64 65 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 | hostname TEXT, username TEXT, CONSTRAINT monitors_constraint UNIQUE (pid,hostname));") (sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS servers (id INTEGER PRIMARY KEY, pid INTEGER, interface TEXT, hostname TEXT, pullport INTEGER, pubport INTEGER, start_time TIMESTAMP, priority INTEGER, state TEXT, mt_version TEXT, heartbeat TIMESTAMP, CONSTRAINT servers_constraint UNIQUE (pid,hostname,pullport,pubport));") (sqlite3:execute 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));") )) mdb)) ;;====================================================================== ;; Server and client management ;;====================================================================== ;; state: 'live, 'shutting-down, 'dead (define (tasks:server-register mdb pid interface pullport pubport priority state) (sqlite3:execute mdb "INSERT OR REPLACE INTO servers (pid,hostname,pullport,pubport,start_time,priority,state,mt_version,heartbeat,interface) VALUES(?, ?, ?, ?, strftime('%s','now'), ?, ?, ?, strftime('%s','now'),?);" pid (get-host-name) pullport pubport priority (conc state) megatest-version interface) (list (tasks:server-get-server-id mdb (get-host-name) pullport pid) interface pullport pubport)) ;; NB// two servers with same pid on different hosts will be removed from the list if pid: is used! (define (tasks:server-deregister mdb hostname #!key (pullport #f)(pid #f)) (debug:print-info 11 "server-deregister " hostname ", pullport " pullport ", pid " pid) (if pid ;; (sqlite3:execute mdb "DELETE FROM servers WHERE pid=?;" pid) (sqlite3:execute mdb "UPDATE servers SET state='dead' WHERE pid=?;" pid) (if pullport ;; (sqlite3:execute mdb "DELETE FROM servers WHERE hostname=? AND port=?;" hostname port) (sqlite3:execute mdb "UPDATE servers SET state='dead' WHERE hostname=? AND pullport=?;" hostname pullport) (debug:print 0 "ERROR: tasks:server-deregister called with neither pid nor port specified")))) (define (tasks:server-deregister-self mdb hostname) (tasks:server-deregister mdb hostname pid: (current-process-id))) (define (tasks:server-get-server-id mdb hostname pullport pid) (let ((res #f)) (sqlite3:for-each-row (lambda (id) (set! res id)) mdb (if (and hostname pid) "SELECT id FROM servers WHERE hostname=? AND pid=?;" "SELECT id FROM servers WHERE hostname=? AND pullport=?;") hostname (if pid pid pullport)) res)) (define (tasks:server-update-heartbeat mdb server-id) (sqlite3:execute mdb "UPDATE servers SET heartbeat=strftime('%s','now') WHERE id=?;" server-id)) ;; alive servers keep the heartbeat field upto date with seconds every 6 or so seconds (define (tasks:server-alive? mdb server-id #!key (hostname #f)(pullport #f)(pid #f)) (let* ((server-id (if server-id server-id (tasks:server-get-server-id mdb hostname pullport pid))) (heartbeat-delta 99e9)) (sqlite3:for-each-row (lambda (delta) (set! heartbeat-delta delta)) mdb "SELECT strftime('%s','now')-heartbeat FROM servers WHERE id=?;" server-id) (< heartbeat-delta 10))) |
︙ | ︙ | |||
156 157 158 159 160 161 162 | ;; ping each server in the db and return first found that responds. ;; remove any others. will not necessarily remove all! (define (tasks:get-best-server mdb) (let ((res '()) (best #f)) (sqlite3:for-each-row | | | | | | | > | | | | | | | > > > > > > > > > | 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 | ;; ping each server in the db and return first found that responds. ;; remove any others. will not necessarily remove all! (define (tasks:get-best-server mdb) (let ((res '()) (best #f)) (sqlite3:for-each-row (lambda (id hostname interface pullport pubport pid) (set! res (cons (list hostname interface pullport pubport pid) res)) (debug:print-info 2 "Found existing server " hostname ":" pullport " registered in db")) mdb "SELECT id,hostname,interface,pullport,pubport,pid FROM servers WHERE state='live' AND mt_version=? ORDER BY start_time ASC LIMIT 1;" megatest-version) ;; (print "res=" res) (if (null? res) #f (let loop ((hed (car res)) (tal (cdr res))) ;; (print "hed=" hed ", tal=" tal) (let* ((host (list-ref hed 0)) (iface (list-ref hed 1)) (pullport (list-ref hed 2)) (pubport (list-ref hed 3)) (pid (list-ref hed 4)) (alive (open-run-close tasks:server-alive? tasks:open-db #f hostname: host pullport: pullport))) (if alive (begin (debug:print-info 2 "Found an existing, alive, server " host ", " pullport " and " pubport ".") (list host iface pullport pubport)) (begin (debug:print-info 1 "Marking " host ":" pullport " as dead in server registry.") (if port (open-run-close tasks:server-deregister tasks:open-db host pullport: pullport) (open-run-close tasks:server-deregister tasks:open-db host pid: pid)) (if (null? tal) #f (loop (car tal)(cdr tal)))))))))) (define (tasks:mark-server hostname pullport pid state) (if port (open-run-close tasks:server-deregister tasks:open-db hostname port: port) (open-run-close tasks:server-deregister tasks:open-db hostname pid: pid))) ;; NOTE: NOT PORTED TO WORK WITH pullport/pubport (define (tasks:kill-server status hostname port pid) (debug:print-info 1 "Removing defunct server record for " hostname ":" port) (if port (open-run-close tasks:server-deregister tasks:open-db hostname port: port) (open-run-close tasks:server-deregister tasks:open-db hostname pid: pid)) (if status ;; #t means alive (begin |
︙ | ︙ | |||
214 215 216 217 218 219 220 | (thread-sleep! 5) ;; give it five seconds to die peacefully then do a brutal kill (process-signal pid signal/kill)) (debug:print 0 "WARNING: Can't kill frozen server on remote host " hostname)))))) (define (tasks:get-all-servers mdb) (let ((res '())) (sqlite3:for-each-row | | | | | 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 | (thread-sleep! 5) ;; give it five seconds to die peacefully then do a brutal kill (process-signal pid signal/kill)) (debug:print 0 "WARNING: Can't kill frozen server on remote host " hostname)))))) (define (tasks:get-all-servers mdb) (let ((res '())) (sqlite3:for-each-row (lambda (id pid hostname interface pullport pubport start-time priority state mt-version) (set! res (cons (vector id pid hostname interface pullport pubport start-time priority state mt-version) res))) mdb "SELECT id,pid,hostname,interface,pullport,pubport,start_time,priority,state,mt_version FROM servers ORDER BY start_time DESC;") res)) ;;====================================================================== ;; Tasks and Task monitors ;;====================================================================== |
︙ | ︙ |
Modified utils/installall.sh from [e461189985] to [6bb115f66a].
|
| | | | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 | #! /bin/env bash set -x # Copyright 2007-2010, 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. echo You may need to do the following first: echo sudo apt-get install libreadline-dev echo sudo apt-get install libwebkitgtk-dev echo sudo apt-get install libmotif3 -OR- set KTYPE=26g4 echo KTYPE can be 26, 26g4, or 32 echo KTYPE=$KTYPE echo You are using PREFIX=$PREFIX echo You are using proxy="$proxy" echo echo "Set additional_libpath to help find gtk or other libraries, don't forget a leading :" |
︙ | ︙ | |||
266 267 268 269 270 271 272 | if [[ -e ${ZEROMQ}${zpatchlev}.tar.gz ]] ; then tar xfz ${ZEROMQ}.tar.gz cd ${ZEROMQ} ln -s $PREFIX/include/uuid src # LDFLAGS=-L$PREFIX/lib ./configure --prefix=$PREFIX | | > | 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 | if [[ -e ${ZEROMQ}${zpatchlev}.tar.gz ]] ; then tar xfz ${ZEROMQ}.tar.gz cd ${ZEROMQ} ln -s $PREFIX/include/uuid src # LDFLAGS=-L$PREFIX/lib ./configure --prefix=$PREFIX ./configure --enable-static --prefix=$PREFIX --with-uuid=$PREFIX LDFLAGS="-L$PREFIX/lib" CPPFLAGS="-fPIC -I$PREFIX/include" LIBS="-lgcc" # --disable-shared CPPFLAGS="-fPIC # LDFLAGS="-L/usr/lib64 -L$PREFIX/lib" ./configure --enable-static --prefix=$PREFIX make make install CSC_OPTIONS="-I$PREFIX/include -L$CSCLIBS" chicken-install $PROX zmq # CSC_OPTIONS="-I$PREFIX/include -L$CSCLIBS" chicken-install $PROX -deploy -prefix $DEPLOYTARG zmq fi |
︙ | ︙ |