Overview
Comment: | progress |
---|---|
Downloads: | Tarball | ZIP archive | SQL archive |
Timelines: | family | ancestors | v1.65-multi-db |
Files: | files | file ages | folders |
SHA1: |
adbeb66c05311ae7dd12044e50ffe30a |
User & Date: | matt on 2019-02-09 20:55:38 |
Other Links: | branch diff | manifest | tags |
Context
2019-02-09
| ||
20:55 | progress Leaf check-in: adbeb66c05 user: matt tags: v1.65-multi-db | |
2019-02-07
| ||
10:19 | Tidy up of mtut formating check-in: e52f8b2513 user: matt tags: v1.65-multi-db | |
Changes
Modified db.scm from [095da180ef] to [7f1125265b].
︙ | ︙ | |||
24 25 26 27 28 29 30 | ;; dbstruct vector containing all the relevant dbs like main.db, megatest.db, run.db etc (declare (unit db)) (module db ( | < > | 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | ;; dbstruct vector containing all the relevant dbs like main.db, megatest.db, run.db etc (declare (unit db)) (module db ( db:setup ) (import scheme posix chicken data-structures ports) (use (prefix sqlite3 sqlite3:) (srfi 18) extras tcp stack srfi-1 posix regex regex-case srfi-69 csv-xml s11n md5 message-digest base64 format dot-locking z3 |
︙ | ︙ |
Modified megatest.scm from [e83d84bbae] to [8bf0940777].
︙ | ︙ | |||
40 41 42 43 44 45 46 | (declare (uses runs)) (declare (uses launch)) (declare (uses server)) (declare (uses client)) (declare (uses tests)) (declare (uses genexample)) ;; (declare (uses daemon)) | | > | 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | (declare (uses runs)) (declare (uses launch)) (declare (uses server)) (declare (uses client)) (declare (uses tests)) (declare (uses genexample)) ;; (declare (uses daemon)) (declare (uses db)) ;; (declare (uses dcommon)) (import db) (declare (uses tdb)) (declare (uses mt)) (declare (uses api)) (declare (uses tasks)) ;; only used for debugging. (declare (uses env)) (declare (uses diff-report)) |
︙ | ︙ |
Modified server.scm from [afd098704f] to [9fd201bbc1].
︙ | ︙ | |||
80 81 82 83 84 85 86 87 88 89 | ;; make it a global? Well, it is local to area module (define *area-info* (make-area)) (define *pktspec* `((server (hostname . h) (port . p) (pid . i) ) (data (hostname . h) ;; sender hostname (port . p) ;; sender port | > | | 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | ;; make it a global? Well, it is local to area module (define *area-info* (make-area)) (define *pktspec* `((server (hostname . h) (port . p) (pid . i) (ipaddr . a) ) (data (hostname . h) ;; sender hostname (port . p) ;; sender port (ipaddr . a) ;; sender ip (hostkey . k) ;; sending host key - store info at server under this key (servkey . s) ;; server key - this needs to match at server end or reject the msg (format . f) ;; sb=serialized-base64, t=text, sx=sexpr, j=json (data . d) ;; base64 encoded slln data ))) (define (server:get-mtrah) |
︙ | ︙ | |||
132 133 134 135 136 137 138 | (print "Starting server in " server-type " mode") (if (eq? server-type 'main) (begin (area-pktid-set! *area-info* (write-alist->pkt pktdir `((hostname . ,(get-host-name)) | | | 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | (print "Starting server in " server-type " mode") (if (eq? server-type 'main) (begin (area-pktid-set! *area-info* (write-alist->pkt pktdir `((hostname . ,(get-host-name)) (ipaddr . ,best-ip) (port . ,port-num) (pid . ,(current-process-id))) pktspec: *pktspec* ptype: 'server)) (area-pktfile-set! *area-info* (conc pktdir "/" (area-pktid *area-info*) ".pkt")))) ;; set all the area info in the (area-pktsdir-set! *area-info* pktdir) |
︙ | ︙ | |||
195 196 197 198 199 200 201 | (let ((dat (server:receive rep))) (set! last-msg-time (current-seconds)) ;; (print "received: " pktdat) (if (not (eof-object? dat)) (let ((resdat (proc dat))) (print "Got " dat) (print "Responding with " resdat) | | | > | | | | > > > > > > > > > > > > | > | > > > > > | > > > > > > > > > > > > > > > > > > > > | > > | 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 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 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 | (let ((dat (server:receive rep))) (set! last-msg-time (current-seconds)) ;; (print "received: " pktdat) (if (not (eof-object? dat)) (let ((resdat (proc dat))) (print "Got " dat) (print "Responding with " resdat) (nmsg:send rep (sexpr->string resdat)) ;; (with-output-to-string (lambda ()(write resdat)))) (loop)))))) "message handler")) (th2 (make-thread (lambda () (let loop () (thread-sleep! 10) (if (> (- (current-seconds) last-msg-time) 60) ;; timeout after 60 seconds (begin (print "Waited for 60 seconds and no messages, exiting now.") (exit)) (loop))))))) (thread-start! th1) (thread-start! th2) (thread-join! th1))) ;; get the response ;; (define (server:receive rep) (let ((instr (nmsg:recv rep))) (if (string? instr) (string->sexpr instr) ;; (with-input-from-string instr read) instr))) (define (server:shutdown) (let ((conn (area-conn *area-info*)) (pktf (area-pktfile *area-info*)) (port (area-port *area-info*))) (if conn (begin (if pktf (delete-file* pktf)) (server:send-all "imshuttingdown") (nmsg:close conn) (portlogger:open-run-close portlogger:release-port port))))) (define (server:send-all msg) #f) ;; given a area record look up all the packets ;; (define (server:get-all-server-pkts rec) (let ((all-pkt-files (glob (conc (area-pktsdir rec) "/*.pkt")))) ;; (pktspec (area-pktspec rec))) (map (lambda (pkt-file) (read-pkt->alist pkt-file pktspec: *pktspec*)) all-pkt-files))) #;((Z . "9a0212302295a19610d5796fce0370fa130758e9") (port . "34827") (pid . "28748") (hostname . "zeus") (T . "server") (D . "1549427032.0")) ;; srvpkt is the info for the server we wish to send the message to ;; (define (server:send servpkt data action) (let* ((port (alist-ref 'port servpkt)) (host (alist-ref 'hostname servpkt)) (ip (alist-ref 'ipaddr servpkt)) (hkey (alist-ref 'Z servpkt)) (addr (conc (or ip host) ":" port)) ;; fall back to host if ip not provided (myport (area-port *area-info*)) (myhost (area-myaddr *area-info*)) (mykey (area-pktid *area-info*)) (msg (with-output-to-string (lambda () (write `((hostname . ,myhost) (port . ,myport) (servkey . ,hkey) ;; server looks at this to ensure message is for them (hostkey . ,mykey) (action . ,action) ;; formating of the message (data . ,data)) ;; *pktspec* ;; ptype: 'data)) ))))) ;; (print "msg: " msg) (if (and port host) (string->sexpr ;; begin ;; (print "sending ")(pp msg)(print " to " addr) (nmsg:open-send-receive addr msg)) #f))) (define (server:get-my-best-address) (ip->string (car (filter (lambda (x) (not (eq? (u8vector-ref x 0) 127))) (vector->list (hostinfo-addresses (hostname->hostinfo "zeus"))))))) ;; whoami? I am my pkt ;; (define (server:whoami? area) (hash-table-ref/default (area-hosts area)(area-pktid area) #f)) ;;====================================================================== ;; "Client side" operations ;;====================================================================== ;; convert to/from string / sexpr (define (string->sexpr str) (if (string? str) (with-input-from-string str read) str)) (define (sexpr->string s) (with-output-to-string (lambda ()(write s)))) ;; is the server alive? ;; (define (server:ping servpkt) (let* ((start-time (current-milliseconds)) (res (server:send servpkt 'ping 'immediate))) (cons (- (current-milliseconds) start-time) res))) ;; (equal? res "got ping")))) ;; look up all pkts and get the server id (the hash), port, host/ip ;; store this info in the global struct *area-info* ;; ;; pass in *area-info* as rec ;; (define (server:update-known-servers rec) ;; readll all pkts ;; foreach pkt; if it isn't me ping the server; if alive, add to hosts hash, else rm the pkt (let ((all-pkts (delete-duplicates (append (server:get-all-server-pkts rec) (hash-table-values (area-hosts rec))))) (hostshash (area-hosts rec)) (my-id (area-pktid rec)) (pktsdir (area-pktsdir rec)) ;; needed to remove pkts from non-responsive servers ) (for-each (lambda (servpkt) (let* ((res (server:ping servpkt)) (sid (alist-ref 'Z servpkt)) ;; Z code is our name for the server ) (match res ((qduration . payload) (print "Server pkt:")(pp servpkt) (print "res: ")(pp res) (match payload ((code message) (print "code: " code " message: " message) (if code (hash-table-set! hostshash sid servpkt) (print "got #f from the server, not sure what that means!"))) (else (print "Got ")(pp res)(print " from server ")(pp servpkt) " but response did not match (#f/#t . msg)"))) (else ;; here we delete the pkt - can't reach the server, remove it ;; however this logic is inadequate. we should mark the server as checked ;; and not good, if it happens a second time - then remove the pkt ;; or something similar. I.e. don't be too quick to assume the server is wedged or dead ;; could be it is simply too busy to reply (print "clearing out server " sid) (delete-file* (conc pktsdir "/" sid ".pkt")) (hash-table-delete! hostshash side))))) all-pkts))) ;; send out an "I'm about to exit notice to all known servers" ;; (define (server:imminent-death) '()) |
︙ | ︙ |