Changes In Branch interleaved-queries Through [d0462389b2] Excluding Merge-Ins
This is equivalent to a diff from 8c476e8627 to d0462389b2
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
| ||
22:34 | Added outline for interleaved queries check-in: a72841295a user: matt tags: interleaved-queries | |
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 [b0135e31f1].
︙ | ︙ | |||
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* zmq-socket) (receive-message zmq-socket)))))) (send-message zmq-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 [129e174847].
︙ | ︙ | |||
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 | (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") (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 iface port) (let ((zsocket (server:client-connect iface port))) (let loop () (thread-sleep! 2) (cdb:client-call zsocket 'ping #t) (debug:print 4 "server:self-ping - I'm alive on " iface ":" port "!") (mutex-lock! *heartbeat-mutex*) (set! *server-loop-heart-beat* (current-seconds)) (mutex-unlock! *heartbeat-mutex*) (loop)))) (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)))) | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | > | > > > > | | | | | | | | < < | | | | > > > > | > > | | < < < < | < < < < | 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 | (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 (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 iface port) (let ((zsocket (server:client-connect iface port))) (let loop () (thread-sleep! 2) (cdb:client-call zsocket 'ping #t) (debug:print 4 "server:self-ping - I'm alive on " iface ":" port "!") (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) (zmq-socket1 #f) (zmq-socket2 #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-socket-dat)) (set! zmq-socket1 (car zmq-sdat1)) (set! p1 (caddr zmq-sdat1)) (set! zmq-sdat2 (cadr zmq-socket-dat)) (set! zmq-socket2 (car 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* zmq-socket1)) (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) (send-message zmq-socket (db:obj->string res)) (if (not *time-to-exit*) (loop) (begin |
︙ | ︙ | |||
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))))))) | | | | > | > > > > > > > | | | | | | | | | | | | | | > | > > | | > > | > | 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 | (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 ipadrstr startport) (let* ((s1 (server:find-free-port-and-open ipadrstr #f startport 'pub)) (p1 (caddr s1)) (s2 (server:find-free-port-and-open ipadrstr #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) iface p 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-connect iface port #!key (context #f)(type 'req)) (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 (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)) ;; 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 port: port) (server:client-setup (- numtries 1)) #f) (let* ((push-socket (server:client-connect iface pullport 'push)) (sub-socket (server:client-connect iface pubport 'sub)) (zmq-sockets (vector push-socket sub-socket)) (login-res #f) ;; (connect-ok (conurl (server:make-server-url (list iface port)))) (socket-option-set! sub-socket 'subscribe (server:get-client-signature)) (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 " conurl) (set! *runremote* zmq-socket) #t) (begin (debug:print-info 2 "Failed to login or connect to " conurl) (set! *runremote* #f) |
︙ | ︙ |
Modified tasks.scm from [53e3d6e8bd] to [90e00b3daa].
︙ | ︙ | |||
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 | ;;====================================================================== ;; 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 (delete-file dbpath) #f) #t) #t)) (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, | | > | 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | 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,port));") (sqlite3:execute mdb "CREATE TABLE IF NOT EXISTS clients (id INTEGER PRIMARY KEY, |
︙ | ︙ | |||
160 161 162 163 164 165 166 | (let ((res '()) (best #f)) (sqlite3:for-each-row (lambda (id hostname interface port pid) (set! res (cons (list hostname interface port pid) res)) (debug:print-info 2 "Found existing server " hostname ":" port " registered in db")) mdb | | | | > | | | | | | | > > > > > > > > > | 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 | (let ((res '()) (best #f)) (sqlite3:for-each-row (lambda (id hostname interface port pid) (set! res (cons (list hostname interface port pid) res)) (debug:print-info 2 "Found existing server " hostname ":" port " 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 |
︙ | ︙ |
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 |
︙ | ︙ |