Overview
Comment: | Merged http-transport to trunk |
---|---|
Downloads: | Tarball | ZIP archive | SQL archive |
Timelines: | family | ancestors | descendants | both | trunk |
Files: | files | file ages | folders |
SHA1: |
6bba674f33d066d866285ab557c2f9f4 |
User & Date: | matt on 2013-01-27 12:22:08 |
Other Links: | manifest | tags |
Context
2013-01-29
| ||
10:15 | Merged multi-transport to trunk, all tests passed check-in: 19f85b577e user: mrwellan tags: trunk | |
2013-01-27
| ||
13:14 | Creating branch for multi transport options, http, rpc, zmq, and network fs check-in: 66763d5399 user: matt tags: multi-transport | |
13:11 | merged trunk into sqlite3-logging Closed-Leaf check-in: 7bd6b6dae6 user: matt tags: sqlite3-logging | |
12:22 | Merged http-transport to trunk check-in: 6bba674f33 user: matt tags: trunk | |
12:17 | Merged zmq-3.2.2 to trunk check-in: 3be673e9be user: matt tags: trunk | |
10:04 | Streamlined db access a little, test4 completes in reasonable time Closed-Leaf check-in: a893c641ca user: matt tags: http-transport | |
Changes
Modified db.scm from [5f49a2d10d] to [4a4c4c2fc7].
︙ | ︙ | |||
12 13 14 15 16 17 18 | ;;====================================================================== ;; Database access ;;====================================================================== (require-extension (srfi 18) extras tcp) ;; rpc) ;; (import (prefix rpc rpc:)) | | | < | 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 | ;;====================================================================== ;; 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 base64) (import (prefix sqlite3 sqlite3:)) (import (prefix base64 base64:)) (declare (unit db)) (declare (uses common)) (declare (uses keys)) (declare (uses ods)) (include "common_records.scm") |
︙ | ︙ | |||
97 98 99 100 101 102 103 | (debug:print 0 " " ((condition-property-accessor 'exn 'message) exn)) (print-call-chain) (thread-sleep! (random 120)) (debug:print-info 0 "trying db call one more time....") (apply open-run-close-no-exception-handling proc idb params)) (apply open-run-close-no-exception-handling proc idb params))) | | > | 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | (debug:print 0 " " ((condition-property-accessor 'exn 'message) exn)) (print-call-chain) (thread-sleep! (random 120)) (debug:print-info 0 "trying db call one more time....") (apply open-run-close-no-exception-handling proc idb params)) (apply open-run-close-no-exception-handling proc idb params))) ;; (define open-run-close open-run-close-exception-handling) (define open-run-close open-run-close-no-exception-handling) (define *global-delta* 0) (define *last-global-delta-printed* 0) (define (open-run-close-measure proc idb . params) (debug:print-info 11 "open-run-close-measure START, idb=" idb ", params=" params) (let* ((start-ms (current-milliseconds)) |
︙ | ︙ | |||
787 788 789 790 791 792 793 | (if currstate (conc "state='" currstate "' AND ") "") (if currstatus (conc "status='" currstatus "' AND ") "") " run_id=? AND testname=? AND NOT (item_path='' AND testname in (SELECT DISTINCT testname FROM tests WHERE testname=? AND item_path != ''));"))) ;;(debug:print 0 "QRY: " qry) (sqlite3:execute db qry run-id newstate newstatus testname testname))) testnames)) | | | | 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 | (if currstate (conc "state='" currstate "' AND ") "") (if currstatus (conc "status='" currstatus "' AND ") "") " run_id=? AND testname=? AND NOT (item_path='' AND testname in (SELECT DISTINCT testname FROM tests WHERE testname=? AND item_path != ''));"))) ;;(debug:print 0 "QRY: " qry) (sqlite3:execute db qry run-id newstate newstatus testname testname))) testnames)) (define (cdb:delete-tests-in-state serverdat run-id state) (cdb:client-call serverdat 'delete-tests-in-state #t *default-numtries* run-id state)) ;; speed up for common cases with a little logic (define (db:test-set-state-status-by-id db test-id newstate newstatus newcomment) (cond ((and newstate newstatus newcomment) (sqlite3:exectute db "UPDATE tests SET state=?,status=?,comment=? WHERE id=?;" newstate newstatus test-id)) ((and newstate newstatus) |
︙ | ︙ | |||
966 967 968 969 970 971 972 | (define (db:test-set-comment db test-id comment) (sqlite3:execute db "UPDATE tests SET comment=? WHERE id=?;" comment test-id)) | | | | | | | | 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 | (define (db:test-set-comment db test-id comment) (sqlite3:execute db "UPDATE tests SET comment=? WHERE id=?;" comment test-id)) (define (cdb:test-set-rundir! serverdat run-id test-name item-path rundir) (cdb:client-call serverdat 'test-set-rundir #t *default-numtries* rundir run-id test-name item-path)) (define (cdb:test-set-rundir-by-test-id serverdat test-id rundir) (cdb:client-call serverdat 'test-set-rundir-by-test-id #t *default-numtries* rundir test-id)) (define (db:test-get-rundir-from-test-id db test-id) (let ((res #f)) ;; (hash-table-ref/default *test-paths* test-id #f))) ;; (if res ;; res ;; (begin (sqlite3:for-each-row (lambda (tpath) (set! res tpath)) db "SELECT rundir FROM tests WHERE id=?;" test-id) ;; (hash-table-set! *test-paths* test-id res) res)) ;; )) (define (cdb:test-set-log! serverdat test-id logf) (if (string? logf)(cdb:client-call serverdat 'test-set-log #f *default-numtries* logf test-id))) ;;====================================================================== ;; Misc. test related queries ;;====================================================================== (define (db:test-get-paths-matching db keynames target fnamepatt #!key (res '())) (let* ((testpatt (if (args:get-arg "-testpatt")(args:get-arg "-testpatt") "%")) |
︙ | ︙ | |||
1102 1103 1104 1105 1106 1107 1108 | ;; (define (db:updater) ;; (debug:print-info 4 "Starting cache processing") ;; (let loop () ;; (thread-sleep! 10) ;; move save time around to minimize regular collisions? ;; (db:write-cached-data) ;; (loop))) | | > > > > > | > > > > > | | | | | | | < < | > > > < < | > | < < < < < < | | > | | | | | | | | | | < > | | | | | | > > | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 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 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 | ;; (define (db:updater) ;; (debug:print-info 4 "Starting cache processing") ;; (let loop () ;; (thread-sleep! 10) ;; move save time around to minimize regular collisions? ;; (db:write-cached-data) ;; (loop))) (define (db:obj->string obj) (string-substitute (regexp "=") "_" (base64:base64-encode (with-output-to-string (lambda ()(serialize obj)))) #t)) (define (db:string->obj msg) (with-input-from-string (base64:base64-decode (string-substitute (regexp "_") "=" msg #t)) (lambda ()(deserialize)))) (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 ;; ;; make-vector-record cdb packet client-sig qtype immediate query-sig params qtime ;; (define (cdb:client-call serverdat qtype immediate numretries . params) (debug:print-info 11 "cdb:client-call serverdat=" serverdat ", qtype=" qtype ", immediate=" immediate ", numretries=" numretries ", params=" params) ;; (handle-exceptions ;; exn ;; (begin ;; (thread-sleep! 5) ;; (if (> numretries 0)(apply cdb:client-call serverdat qtype immediate (- numretries 1) params))) (let* ((client-sig (server:get-client-signature)) (query-sig (message-digest-string (md5-primitive) (conc qtype immediate params))) (zdat (db:obj->string (vector client-sig qtype immediate query-sig params (current-seconds)))) ;; (with-output-to-string (lambda ()(serialize params)))) ) (debug:print-info 11 "zdat=" zdat) (let* ( (res #f) (rawdat (server:client-send-receive serverdat zdat)) (tmp #f)) (debug:print-info 11 "Sent " zdat ", received " rawdat) (set! tmp (db:string->obj rawdat)) ;; (if (equal? query-sig (vector-ref myres 1)) ;; (set! res (vector-ref tmp 2) ;; (loop (server:client-send-receive serverdat zdat))))))) ;; (timeout (lambda () ;; (let loop ((n numretries)) ;; (thread-sleep! 15) ;; (if (not res) ;; (if (> numretries 0) ;; (begin ;; (debug:print 2 "WARNING: no reply to query " params ", trying resend") ;; (debug:print-info 11 "re-sending message") ;; (apply cdb:client-call serverdat qtype immediate numretries params) ;; (debug:print-info 11 "message re-sent") ;; (loop (- n 1))) ;; ;; (apply cdb:client-call serverdats qtype immediate (- numretries 1) params)) ;; (begin ;; (debug:print 0 "ERROR: cdb:client-call timed out " params ", exiting.") ;; (exit 5)))))))) ;; (send-receive) ))) ;; (debug:print-info 11 "Starting threads") ;; (let ((th1 (make-thread send-receive "send receive")) ;; (th2 (make-thread timeout "timeout"))) ;; (thread-start! th1) ;; (thread-start! th2) ;; (thread-join! th1) ;; (debug:print-info 11 "cdb:client-call returning res=" res) ;; res)))) (define (cdb:set-verbosity serverdat val) (cdb:client-call serverdat 'set-verbosity #f *default-numtries* val)) (define (cdb:login serverdat keyval signature) (cdb:client-call serverdat 'login #t *default-numtries* keyval megatest-version signature)) (define (cdb:logout serverdat keyval signature) (cdb:client-call serverdat 'logout #t *default-numtries* keyval signature)) (define (cdb:num-clients serverdat) (cdb:client-call serverdat 'numclients #t *default-numtries*)) (define (cdb:test-set-status-state serverdat test-id status state msg) (if msg (cdb:client-call serverdat 'state-status-msg #t *default-numtries* state status msg test-id) (cdb:client-call serverdat 'state-status #t *default-numtries* state status test-id))) ;; run-id test-name item-path minutes cpuload diskfree tmpfree) (define (cdb:test-rollup-test_data-pass-fail serverdat test-id) (cdb:client-call serverdat 'test_data-pf-rollup #t *default-numtries* test-id test-id test-id test-id)) (define (cdb:pass-fail-counts serverdat test-id fail-count pass-count) (cdb:client-call serverdat 'pass-fail-counts #t *default-numtries* fail-count pass-count test-id)) (define (cdb:tests-register-test serverdat run-id test-name item-path) (let ((item-paths (if (equal? item-path "") (list item-path) (list item-path "")))) (cdb:client-call serverdat 'register-test #t *default-numtries* run-id test-name item-path))) (define (cdb:flush-queue serverdat) (cdb:client-call serverdat 'flush #f *default-numtries*)) (define (cdb:kill-server serverdat) (cdb:client-call serverdat 'killserver #f *default-numtries*)) (define (cdb:roll-up-pass-fail-counts serverdat run-id test-name item-path status) (cdb:client-call serverdat 'immediate #f *default-numtries* open-run-close db:roll-up-pass-fail-counts #f run-id test-name item-path status)) (define (cdb:get-test-info serverdat run-id test-name item-path) (cdb:client-call serverdat 'immediate #f *default-numtries* open-run-close db:get-test-info #f run-id test-name item-path)) (define (cdb:get-test-info-by-id serverdat test-id) (cdb:client-call serverdat 'immediate #f *default-numtries* open-run-close db:get-test-info-by-id #f test-id)) ;; db should be db open proc or #f (define (cdb:remote-run proc db . params) (apply cdb:client-call *runremote* 'immediate #f *default-numtries* open-run-close proc #f params)) (define (db:test-get-logfile-info db run-id test-name) (let ((res #f)) |
︙ | ︙ | |||
1274 1275 1276 1277 1278 1279 1280 | (let* ((data (sort indata (lambda (a b) (< (cdb:packet-get-qtime a)(cdb:packet-get-qtime b)))))) (for-each (lambda (item) (db:process-queue-item db pubsock item)) data))) | | | | | | | | | | | | | | | 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 | (let* ((data (sort indata (lambda (a b) (< (cdb:packet-get-qtime a)(cdb:packet-get-qtime b)))))) (for-each (lambda (item) (db:process-queue-item db pubsock item)) data))) (define (db:process-queue-item db item) (let* ((stmt-key (cdb:packet-get-qtype item)) (qry-sig (cdb:packet-get-query-sig item)) (return-address (cdb:packet-get-client-sig item)) (params (cdb:packet-get-params item)) (query (let ((q (alist-ref stmt-key db:queries))) (if q (car q) #f)))) (debug:print-info 11 "Special queries/requests stmt-key=" stmt-key ", return-address=" return-address ", query=" query ", params=" params) (cond (query (apply sqlite3:execute db query params) (server:reply return-address qry-sig #t #t)) ((member stmt-key db:special-queries) (debug:print-info 11 "Handling special statement " stmt-key) (case stmt-key ((immediate) (let ((proc (car params)) (remparams (cdr params))) ;; we are being handed a procedure so call it (debug:print-info 11 "Running (apply " proc " " remparams ")") (server:reply return-address qry-sig #t (apply proc remparams)))) ((login) (if (< (length params) 3) ;; should get toppath, version and signature (server:reply return-address qry-sig '(#f "login failed due to missing params")) ;; missing params (let ((calling-path (car params)) (calling-vers (cadr params)) (client-key (caddr params))) (if (and (equal? calling-path *toppath*) (equal? megatest-version calling-vers)) (begin (hash-table-set! *logged-in-clients* client-key (current-seconds)) (server:reply return-address qry-sig #t '(#t "successful login"))) ;; path matches - pass! Should vet the caller at this time ... (server:reply return-address qry-sig #f (list #f (conc "Login failed due to mismatch paths: " calling-path ", " *toppath*))))))) ((flush sync) (server:reply return-address qry-sig #t 1)) ;; (length data))) ((set-verbosity) (set! *verbosity* (car params)) (server:reply return-address qry-sig #t '(#t *verbosity*))) ((killserver) (debug:print 0 "WARNING: Server going down in 15 seconds by user request!") (open-run-close tasks:server-deregister tasks:open-db (car *runremote*) pullport: (cadr *runremote*)) (thread-start! (make-thread (lambda ()(thread-sleep! 15)(exit)))) (server:reply return-address qry-sig #t '(#t "exit process started"))) (else ;; not a command, i.e. is a query (debug:print 0 "ERROR: Unrecognised query/command " stmt-key) (server:reply pubsock return-address qry-sig #f 'failed)))) (else (debug:print-info 11 "Executing " stmt-key " for " params) (apply sqlite3:execute (hash-table-ref queries stmt-key) params) (server:reply return-address qry-sig #t #t))))) (define (db:test-get-records-for-index-file db run-id test-name) (let ((res '())) (sqlite3:for-each-row (lambda (id itempath state status run_duration logf comment) (set! res (cons (vector id itempath state status run_duration logf comment) res))) db |
︙ | ︙ |
Modified server.scm from [2caba19210] to [12f5deda72].
︙ | ︙ | |||
9 10 11 12 13 14 15 | ;; PURPOSE. (require-extension (srfi 18) extras tcp s11n) (use sqlite3 srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest) (import (prefix sqlite3 sqlite3:)) | > | > < < < < < < < < < < < < < < < < < < < < < < < < < | > > > > > | | < < < < < < < < < | | | > | < | | | | < < < < < < < < > > > < > | < < | < < < < < | | < | | > > | > > > | < < > | < | | | | | | | | | < | < | | < < < < < < < | > | < > > > | < < | | | | | | | | | | | | | | | | | | | < | | | > | | < < < | < | < > | > | | | | | < < < | | < < | < < < < < | > > > | > > | | < < < < < > | | < | | | < < < | < < > | > > > > | | > > > > > > > > > | | > > > | < < < < < | > > > > | > | > | > | < > | > | > > > > > | | > > > | | | | | > | | | | | | < < < < < | > | | | | | | | < < < < < < < < < < < | | | | | | | | | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > < < < < < < < < < < < < < < | < < < < | | < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < | 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 158 159 160 161 162 163 164 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 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 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 | ;; PURPOSE. (require-extension (srfi 18) extras tcp s11n) (use sqlite3 srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest) (import (prefix sqlite3 sqlite3:)) (use spiffy uri-common intarweb http-client spiffy-request-vars) (tcp-buffer-size 2048) (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") (define (server:make-server-url hostport) (if (not hostport) #f (conc "http://" (car hostport) ":" (cadr hostport)))) (define *server-loop-heart-beat* (current-seconds)) (define *heartbeat-mutex* (make-mutex)) ;;====================================================================== ;; S E R V E R ;;====================================================================== ;; Call this to start the actual server ;; (define *db:process-queue-mutex* (make-mutex)) (define (server:run hostn) (debug:print 2 "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* (;; (iface (if (string=? "-" hostn) ;; #f ;; (get-host-name) ;; hostn)) (db #f) ;; (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily (hostname (get-host-name)) (ipaddrstr (let ((ipstr (if (string=? "-" hostn) (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".") #f))) (if ipstr ipstr hostn))) ;; hostname))) (start-port (if (args:get-arg "-port") (string->number (args:get-arg "-port")) (+ 5000 (random 1001)))) (link-tree-path (config-lookup *configdat* "setup" "linktree"))) (set! *cache-on* #t) (root-path (if link-tree-path link-tree-path (current-directory))) ;; WARNING: SECURITY HOLE. FIX ASAP! ;; Setup the web server and a /ctrl interface ;; (vhost-map `(((* any) . ,(lambda (continue) ;; open the db on the first call (if (not db)(set! db (open-db))) (let* (($ (request-vars source: 'both)) (dat ($ 'dat)) (res #f)) (cond ((equal? (uri-path (request-uri (current-request))) '(/ "hey")) (send-response body: "hey there!\n" headers: '((content-type text/plain)))) ;; This is the /ctrl path where data is handed to the server and ;; responses ((equal? (uri-path (request-uri (current-request))) '(/ "ctrl")) (let* ((packet (db:string->obj dat)) (qtype (cdb:packet-get-qtype packet))) (debug:print-info 12 "server=> received packet=" packet) (if (not (member qtype '(sync ping))) (begin (mutex-lock! *heartbeat-mutex*) (set! *last-db-access* (current-seconds)) (mutex-unlock! *heartbeat-mutex*))) ;; (mutex-lock! *db:process-queue-mutex*) ;; trying a mutex ;; (set! res (open-run-close db:process-queue-item open-db packet)) (set! res (db:process-queue-item db packet)) ;; (mutex-unlock! *db:process-queue-mutex*) (debug:print-info 11 "Return value from db:process-queue-item is " res) (send-response body: (conc "<head>ctrl data</head>\n<body>" res "</body>") headers: '((content-type text/plain))))) (else (continue)))))))) (server:try-start-server ipaddrstr start-port) ;; lite3:finalize! db))) )) ;; (define (server:main-loop) ;; (print "INFO: Exectuing main server loop") ;; (access-log "megatest-http.log") ;; (server-bind-address #f) ;; (define-page (main-page-path) ;; (lambda () ;; (let ((dat ($ "dat"))) ;; ;; (with-request-variables (dat) ;; (debug:print-info 12 "Got dat=" dat) ;; (let* ((packet (db:string->obj dat)) ;; (qtype (cdb:packet-get-qtype packet))) ;; (debug:print-info 12 "server=> received packet=" packet) ;; (if (not (member qtype '(sync ping))) ;; (begin ;; (mutex-lock! *heartbeat-mutex*) ;; (set! *last-db-access* (current-seconds)) ;; (mutex-unlock! *heartbeat-mutex*))) ;; (let ((res (open-run-close db:process-queue-item open-db packet))) ;; (debug:print-info 11 "Return value from db:process-queue-item is " res) ;; res)))))) ;;; (use spiffy uri-common intarweb) ;;; ;;; (root-path "/var/www") ;;; ;;; (vhost-map `(((* any) . ,(lambda (continue) ;;; (if (equal? (uri-path (request-uri (current-request))) ;;; '(/ "hey")) ;;; (send-response body: "hey there!\n" ;;; headers: '((content-type text/plain))) ;;; (continue)))))) ;;; ;;; (start-server port: 12345) ;; This is recursively run by server:run until sucessful ;; (define (server:try-start-server ipaddrstr portnum) (handle-exceptions exn (begin (print-error-message exn) (if (< portnum 9000) (begin (print "WARNING: failed to start on portnum: " portnum ", trying next port") (thread-sleep! 0.1) (open-run-close tasks:remove-server-records tasks:open-db) (server:try-start-server ipaddrstr (+ portnum 1))) (print "ERROR: Tried and tried but could not start the server"))) (set! *runremote* (list ipaddrstr portnum)) (open-run-close tasks:remove-server-records tasks:open-db) (open-run-close tasks:server-register tasks:open-db (current-process-id) ipaddrstr portnum 0 'live) (print "INFO: Trying to start server on " ipaddrstr ":" portnum) ;; This starts the spiffy server (start-server port: portnum) (print "INFO: server has been stopped"))) (define (server:mk-signature) (message-digest-string (md5-primitive) (with-output-to-string (lambda () (write (list (current-directory) (argv))))))) ;;====================================================================== ;; S E R V E R U T I L I T I E S ;;====================================================================== ;; When using zmq this would send the message back (two step process) ;; with spiffy or rpc this simply returns the return data to be returned ;; (define (server:reply return-addr query-sig success/fail result) (debug:print-info 11 "server:reply return-addr=" return-addr ", result=" result) ;; (send-message pubsock target send-more: #t) ;; (send-message pubsock (db:obj->string (vector success/fail query-sig result))) ;;====================================================================== ;; C L I E N T S ;;====================================================================== (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*))) ;; <html> ;; <head></head> ;; <body>1 Hello, world! Goodbye Dolly</body></html> ;; Send msg to serverdat and receive result (define (server:client-send-receive serverdat msg) (let* ((url (server:make-server-url serverdat)) (fullurl (conc url "/ctrl")) ;; (conc url "/?dat=" msg))) (numretries 0)) (handle-exceptions exn (if (< numretries 200) (server:client-send-receive serverdat msg)) (begin (debug:print-info 11 "fullurl=" fullurl "\n") ;; set up the http-client here (max-retry-attempts 100) (retry-request? (lambda (request) (thread-sleep! (/ (if (> numretries 100) 100 numretries) 10)) (set! numretries (+ numretries 1)) #t)) ;; send the data and get the response ;; extract the needed info from the http data and ;; process and return it. (let* ((res (with-input-from-request fullurl ;; #f ;; msg (list (cons 'dat msg)) read-string))) (debug:print-info 11 "got res=" res) (let ((match (string-search (regexp "<body>(.*)<.body>") res))) (debug:print-info 11 "match=" match) (let ((final (cadr match))) (debug:print-info 11 "final=" final) final))))))) (define (server:client-login serverdat) (max-retry-attempts 100) (cdb:login serverdat *toppath* (server:get-client-signature))) ;; Not currently used! But, I think it *should* be used!!! (define (server:client-logout serverdat) (let ((ok (and (socket? serverdat) (cdb:logout serverdat *toppath* (server:get-client-signature))))) ;; (close-socket serverdat) ok)) (define (server:client-connect iface port) (let* ((login-res #f) (serverdat (list iface port))) (set! login-res (server:client-login serverdat)) (if (and (not (null? login-res)) (car login-res)) (begin (debug:print-info 2 "Logged in and connected to " iface ":" port) (set! *runremote* serverdat) serverdat) (begin (debug:print-info 2 "Failed to login or connect to " iface ":" port) (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)) (port (list-ref hostinfo 2)) (pid (list-ref hostinfo 3))) (debug:print-info 2 "Setting up to connect to " hostinfo) (server:client-connect iface port)) ;; ) (if (> numtries 0) (let ((exe (car (argv))) (pid #f)) (debug:print-info 0 "No server available, attempting to start one...") (set! pid (process-run exe (list "-server" "-" "-debug" (if (list? *verbosity*) (string-intersperse *verbosity* ",") (conc *verbosity*))))) ;; (set! pid (process-fork (lambda () ;; (current-input-port (open-input-file "/dev/null")) ;; (current-output-port (open-output-file "/dev/null")) ;; (current-error-port (open-output-file "/dev/null")) ;; (server:launch)))) (let loop ((count 0)) (let ((hostinfo (open-run-close tasks:get-best-server tasks:open-db))) (if (not hostinfo) (begin (debug:print-info 0 "Waiting for server pid=" pid " to start") (sleep 2) ;; give server time to start (if (< count 5) (loop (+ count 1))))))) ;; we are starting a server, do not try again! That can lead to ;; recursively starting many processes!!! (server:client-setup numtries: 0)) (debug:print-info 1 "Too many attempts, giving up"))))) ;; run server:keep-running in a parallel thread to monitor that the db is being ;; used and to shutdown after sometime if it is not. ;; (define (server:keep-running) ;; if none running or if > 20 seconds since ;; server last used then start shutdown ;; This thread waits for the server to come alive (let* ((server-info (let loop () (let ((sdat #f)) (mutex-lock! *heartbeat-mutex*) (set! sdat *runremote*) (mutex-unlock! *heartbeat-mutex*) (if sdat sdat (begin (sleep 4) (loop)))))) (iface (car server-info)) (port (cadr server-info)) (last-access 0) (tdb (tasks:open-db)) (spid (tasks:server-get-server-id tdb #f iface port #f))) (print "Keep-running got server pid " spid ", using iface " iface " and port " port) (let loop ((count 0)) (thread-sleep! 4) ;; no need to do this very often ;; NB// sync currently does NOT return queue-length (let () ;; (queue-len (cdb:client-call server-info 'sync #t 1))) ;; (print "Server running, count is " count) (if (< count 1) ;; 3x3 = 9 secs aprox (loop (+ count 1))) ;; NOTE: Get rid of this mechanism! It really is not needed... (tasks:server-update-heartbeat tdb spid) ;; (if ;; (or (> numrunning 0) ;; stay alive for two days after last access (mutex-lock! *heartbeat-mutex*) (set! last-access *last-db-access*) (mutex-unlock! *heartbeat-mutex*) (if (> (+ last-access ;; (* 50 60 60) ;; 48 hrs ;; 60 ;; one minute ;; (* 60 60) ;; one hour (* 45 60) ;; 45 minutes, until the db deletion bug is fixed. ) (current-seconds)) (begin (debug:print-info 2 "Server continuing, seconds since last db access: " (- (current-seconds) last-access)) (loop 0)) (begin (debug:print-info 0 "Starting to shutdown the server.") ;; need to delete only *my* server entry (future use) (set! *time-to-exit* #t) (tasks:server-deregister-self tdb (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))))))) ;; all routes though here end in exit ... (define (server:launch) (if (not *toppath*) (if (not (setup-for-run)) (begin (debug:print 0 "ERROR: cannot find megatest.config, exiting") (exit)))) (debug:print-info 2 "Starting the standalone server") (let ((hostinfo (open-run-close tasks:get-best-server tasks:open-db))) (debug:print 11 "server:launch hostinfo=" hostinfo) (if hostinfo (debug:print-info 2 "NOT starting new server, one is already running on " (car hostinfo) ":" (cadr hostinfo)) (if *toppath* (let* ((th2 (make-thread (lambda () (server:run (if (args:get-arg "-server") (args:get-arg "-server") "-"))) "Server run")) (th3 (make-thread (lambda ()(server:keep-running)) "Keep running")) ) (thread-start! th2) (thread-start! th3) (set! *didsomething* #t) (thread-join! th2) ) (debug:print 0 "ERROR: Failed to setup for megatest"))) (exit))) (define (server:client-signal-handler signum) (handle-exceptions exn (debug:print " ... exiting ...") (let ((th1 (make-thread (lambda () "") ;; do nothing for now (was 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! 1) ;; give the flush one second to do it's stuff (debug:print 0 " Done.") (exit 4)) "exit on ^C timer"))) (thread-start! th2) (thread-start! th1) (thread-join! th2)))) (define (server:client-launch) (set-signal-handler! signal/int server:client-signal-handler) (if (server:client-setup) (debug:print-info 2 "connected as client") (begin (debug:print 0 "ERROR: Failed to connect as client") (exit)))) |
Modified tasks.scm from [bc19277591] to [7e2c4cdfd8].
︙ | ︙ | |||
58 59 60 61 62 63 64 | 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 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | 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, port 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, 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 port priority state) (debug:print-info 11 "tasks:server-register " pid " " interface " " port " " priority " " state) (sqlite3:execute mdb "INSERT OR REPLACE INTO servers (pid,hostname,port,start_time,priority,state,mt_version,heartbeat,interface) VALUES(?, ?, ?, strftime('%s','now'), ?, ?, ?, strftime('%s','now'),?);" pid (get-host-name) port priority (conc state) megatest-version interface) (list (tasks:server-get-server-id mdb (get-host-name) interface port pid) interface port )) ;; 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 (port #f)(pid #f)(action 'markdead)) (debug:print-info 11 "server-deregister " hostname ", port " port ", pid " pid) (if pid (case action ((delete)(sqlite3:execute mdb "DELETE FROM servers WHERE pid=?;" pid)) (else (sqlite3:execute mdb "UPDATE servers SET state='dead' WHERE pid=?;" pid))) (if port (case action ((delete)(sqlite3:execute mdb "DELETE FROM servers WHERE hostname=? AND port=?;" hostname port)) (else (sqlite3:execute mdb "UPDATE servers SET state='dead' WHERE hostname=? AND port=?;" hostname port))) (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 iface port pid) (let ((res #f)) (sqlite3:for-each-row (lambda (id) (set! res id)) mdb (cond ((and hostname pid) "SELECT id FROM servers WHERE hostname=? AND pid=?;") ((and iface port) "SELECT id FROM servers WHERE interface=? AND port=?;") ((and hostname port) "SELECT id FROM servers WHERE hostname=? AND port=?;") (else (begin (debug:print 0 "ERROR: tasks:server-get-server-id needs (hostname and pid) OR (iface and port) OR (hostname and port)") "SELECT id FROM servers WHERE pid=-999;"))) (if hostname hostname iface)(if pid pid port)) 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 (iface #f)(hostname #f)(port #f)(pid #f)) (let* ((server-id (if server-id server-id (tasks:server-get-server-id mdb hostname iface port 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))) (define (tasks:client-register mdb pid hostname cmdline) (sqlite3:execute mdb "INSERT OR REPLACE INTO clients (server_id,pid,hostname,cmdline,login_time) VALUES(?,?,?,?,strftime('%s','now'));") (tasks:server-get-server-id mdb hostname #f #f pid) pid hostname cmdline) (define (tasks:client-logout mdb pid hostname cmdline) (sqlite3:execute mdb "UPDATE clients SET logout_time=strftime('%s','now') WHERE pid=? AND hostname=? AND cmdline=?;" pid hostname cmdline)) |
︙ | ︙ | |||
169 170 171 172 173 174 175 | ;; 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 | | | | | > | > > > > | | | | | | < | | | | | | | | | | | | | | > > > | < | 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 217 218 219 220 221 222 223 224 225 226 227 228 229 230 | ;; 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 port pid) (set! res (cons (list hostname interface port pid id) res)) (debug:print-info 2 "Found existing server " hostname ":" port " registered in db")) mdb "SELECT id,hostname,interface,port,pid FROM servers WHERE strftime('%s','now')-heartbeat < 10 AND mt_version=? ORDER BY start_time ASC LIMIT 1;" megatest-version) ;; for now we are keeping only one server registered in the db, return #f or first server found (if (null? res) #f (car res)))) ;; BUG: This logic is probably needed unless methodology changes completely... ;; ;; (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)) ;; (port (list-ref hed 2)) ;; (pid (list-ref hed 4)) ;; (alive (open-run-close tasks:server-alive? tasks:open-db #f hostname: host port: port))) ;; (if alive ;; (begin ;; (debug:print-info 2 "Found an existing, alive, server " host ", " port ".") ;; (list host iface port)) ;; (begin ;; (debug:print-info 1 "Marking " host ":" port " as dead in server registry.") ;; (if port ;; (open-run-close tasks:server-deregister tasks:open-db host port: port) ;; (open-run-close tasks:server-deregister tasks:open-db host pid: pid)) ;; (if (null? tal) ;; #f ;; (loop (car tal)(cdr tal)))))))))) (define (tasks:remove-server-records mdb) (sqlite3:execute mdb "DELETE FROM servers;")) (define (tasks:mark-server hostname port 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))) (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 |
︙ | ︙ | |||
240 241 242 243 244 245 246 | (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 | | | | | 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 | (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 port start-time priority state mt-version last-update) (set! res (cons (vector id pid hostname interface port start-time priority state mt-version last-update) res))) mdb "SELECT id,pid,hostname,interface,port,start_time,priority,state,mt_version,strftime('%s','now')-heartbeat AS last_update FROM servers ORDER BY start_time DESC;") res)) ;;====================================================================== ;; Tasks and Task monitors ;;====================================================================== |
︙ | ︙ |
Added testhttp/example-client.scm version [57afe37838].
> > > > > > | 1 2 3 4 5 6 | (use regex http-client) (print (with-input-from-request "http://localhost:8083/?foo=1" #f (lambda () (let ((match (string-search (regexp "<body>(.*)<.body>") (caddr (string-split (read-string) "\n"))))) (cadr match))))) |
Added testhttp/example-server.scm version [1efa10faa7].
> > > > > > > > > > > > > > > > > > > > > > > > > > | 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 | (use spiffy awful) (tcp-buffer-size 2048) (enable-sxml #t) (define (hello-world) (define-page (main-page-path) (lambda () (with-request-variables (foo) foo)))) (define (start-server #!key (portnum 8080)) (handle-exceptions exn (begin (print-error-message exn) (if (< portnum 9000) (begin (print "WARNING: failed to start on portnum: " portnum ", trying next port") (sleep 1) (start-server portnum: (+ portnum 1))) (print "ERROR: Tried and tried but could not start the server"))) (print "INFO: Trying to start server on portnum: " portnum) (awful-start hello-world port: portnum))) (start-server) |
Added testhttp/mockupclient.scm version [fceab37de6].
> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 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 | (use posix) (define cname "Bob") (define runtime 10) (let ((args (argv))) (if (< (length args) 3) (begin (print "Usage: mockupclient clientname runtime") (exit)) (begin (set! cname (cadr args)) (set! runtime (string->number (caddr args)))))) ;; (define start-delay (/ (random 100) 9)) ;; (define runtime (+ 1 (/ (random 200) 2))) (print "Starting client " cname " with runtime " runtime) (include "mockupclientlib.scm") (set! endtime (+ (current-seconds) runtime)) (let loop () (let ((x (random 15)) (varname (list-ref (list "hello" "goodbye" "saluton" "kiaorana")(random 4)))) (case x ;; ((1)(dbaccess cname 'sync "nodat" #f)) ((2 3 4 5)(dbaccess cname 'set varname (random 999))) ((6 7 8 9 10)(print cname ": Get \"" varname "\" " (dbaccess cname 'get varname #f))) (else (thread-sleep! 0.011))) (if (< (current-seconds) endtime) (loop)))) (print "Client " cname " all done!!") |
Added testhttp/mockupclientlib.scm version [6a9bd9de5d].
> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 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 | (define sub (make-socket 'sub)) (define push (make-socket 'push)) (socket-option-set! sub 'subscribe cname) (connect-socket sub "tcp://localhost:5563") (connect-socket push "tcp://localhost:5564") (define (dbaccess cname cmd var val #!key (numtries 1)) (let* ((msg (conc cname ":" cmd ":" (if val (conc var " " val) var))) (res #f) (do-access (lambda () (print "Sending msg: " msg) (send-message push msg) (print "Message " msg " sent") (print "Client " cname " waiting for response to " msg) (print "Client " cname " received address " (receive-message* sub)) (set! res (receive-message* sub))))) (let ((th1 (make-thread do-access "do access")) (th2 (make-thread (lambda () (thread-sleep! 5) (if (not res) (if (> numtries 0) (begin (print "WARNING: access timed out for " cname ", trying again. Trys remaining=" numtries) (dbaccess cname cmd var val numtries: (- numtries 1))) (begin (print "ERROR: dbaccess timed out. Exiting") (exit))))) "timeout thread"))) (thread-start! th1) (thread-start! th2) (thread-join! th1) res))) |
Added testhttp/mockupserver.scm version [aa3c5bcdcf].
> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 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 | ;; pub/sub with envelope address ;; Note that if you don't insert a sleep, the server will crash with SIGPIPE as soon ;; as a client disconnects. Also a remaining client may receive tons of ;; messages afterward. (use srfi-18 sqlite3 spiffy) (define cname "server") (define total-db-accesses 0) (define start-time (current-seconds)) ;; setup the server here (tcp-buffer-size 2048) (server-port 5563) (define (open-db) (let* ((dbpath "mockup.db") (dbexists (file-exists? dbpath)) (db (open-database dbpath)) ;; (never-give-up-open-db dbpath)) (handler (make-busy-timeout 10))) (set-busy-handler! db handler) (if (not dbexists) (for-each (lambda (stmt) (execute db stmt)) (list "PRAGMA SYNCHRONOUS=0;" "CREATE TABLE clients (id INTEGER PRIMARY KEY,name TEXT,num_accesses INTEGER DEFAULT 0);" "CREATE TABLE vars (var TEXT,val TEXT,CONSTRAINT vars_constraint UNIQUE (var));"))) db)) (define cid-cache (make-hash-table)) (define (get-client-id db cname) (let ((cid (hash-table-ref/default cid-cache cname #f))) (if cid cid (begin (execute db "INSERT OR REPLACE INTO clients (name) VALUES(?);" cname) (for-each-row (lambda (id) (set! cid id)) db "SELECT id FROM clients WHERE name=?;" cname) (hash-table-set! cid-cache cname cid) (set! total-db-accesses (+ total-db-accesses 2)) cid)))) (define (count-client db cname) (let ((cid (get-client-id db cname))) (execute db "UPDATE clients SET num_accesses=num_accesses+1 WHERE id=?;" cid) (set! total-db-accesses (+ total-db-accesses 1)) )) (define db (open-db)) ;; (define queuelst '()) ;; (define mx1 (make-mutex)) (define max-queue-len 0) (define (process-queue queuelst) (let ((queuelen (length queuelst))) (if (> queuelen max-queue-len) (set! max-queue-len queuelen)) (for-each (lambda (item) (let ((cname (vector-ref item 1)) (clcmd (vector-ref item 2)) (cdata (vector-ref item 3))) (send-message pub cname send-more: #t) (send-message pub (case clcmd ((sync) (conc queuelen)) ((set) (set! total-db-accesses (+ total-db-accesses 1)) (apply execute db "INSERT OR REPLACE INTO vars (var,val) VALUES (?,?);" (string-split cdata)) "ok") ((get) (set! total-db-accesses (+ total-db-accesses 1)) (let ((res "noval")) (for-each-row (lambda (val) (set! res val)) db "SELECT val FROM vars WHERE var=?;" cdata) res)) (else (conc "unk cmd: " clcmd)))))) queuelst))) (define th1 (make-thread (lambda () (let ((last-run 0)) ;; current-seconds when run last (let loop ((queuelst '())) (let* ((indat (receive-message* pull)) (parts (string-split indat ":")) (cname (car parts)) ;; client name (clcmd (string->symbol (cadr parts))) ;; client cmd (cdata (caddr parts)) ;; client data (svect (vector (current-seconds) cname clcmd cdata))) ;; record for the queue (count-client db cname) (case clcmd ((sync) ;; just process the queue (print "Got sync from " cname) (process-queue (cons svect queuelst)) (loop '())) ((get) (process-queue (cons svect queuelst)) (loop '())) (else (loop (cons svect queuelst)))))))) "server thread")) (include "mockupclientlib.scm") ;; ;; send a sync to the pull port ;; (define th2 (make-thread ;; (lambda () ;; (let ((last-action-time (current-seconds))) ;; (let loop () ;; (thread-sleep! 5) ;; (let ((queuelen (string->number (dbaccess "server" 'sync "nada" #f))) ;; (last-action-delta #f)) ;; (if (> queuelen 1)(set! last-action-time (current-seconds))) ;; (set! last-action-delta (- (current-seconds) last-action-time)) ;; (print "Server: Got queuelen=" queuelen ", last-action-delta=" last-action-delta) ;; (if (< last-action-delta 60) ;; (loop) ;; (print "Server exiting, 25 seconds since last access")))))) ;; "sync thread")) (handle-not-found (thread-start! th1) (thread-start! th2) (thread-join! th2) (let* ((run-time (- (current-seconds) start-time)) (queries/second (/ total-db-accesses run-time))) (print "Server exited! Total db accesses=" total-db-accesses " in " run-time " seconds for " queries/second " queries/second with max queue length of: " max-queue-len)) |
Added testhttp/testclient.scm version [4aba6dc5d5].
> > > > > > > > | 1 2 3 4 5 6 7 8 | (use http-client) (with-input-from-request "http://localhost:12345/hey" ;; #f ;; msg (list (cons 'dat "Testing eh")) read-string) |
Added testhttp/testserver.scm version [02c28bc9cb].
> > > > > > > > > > > > > > > > | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 | (use spiffy uri-common intarweb spiffy-request-vars) (root-path "/var/www") (vhost-map `(((* any) . ,(lambda (continue) (let (($ (request-vars source: 'both))) (print ($ 'dat)) (if (equal? (uri-path (request-uri (current-request))) '(/ "hey")) (send-response body: "hey there!\n" headers: '((content-type text/plain))) (continue))))))) (start-server port: 12345) |
Modified tests/tests.scm from [052cb1980d] to [17571516a2].
︙ | ︙ | |||
77 78 79 80 81 82 83 | ;; S E R V E R ;;====================================================================== (test "setup for run" #t (begin (setup-for-run) (string? (getenv "MT_RUN_AREA_HOME")))) (test "server-register, get-best-server" #t (let ((res #f)) | | | > | < < < < | 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 | ;; S E R V E R ;;====================================================================== (test "setup for run" #t (begin (setup-for-run) (string? (getenv "MT_RUN_AREA_HOME")))) (test "server-register, get-best-server" #t (let ((res #f)) (open-run-close tasks:server-register tasks:open-db 1 "bob" 1234 100 'live) (set! res (open-run-close tasks:get-best-server tasks:open-db)) (number? (cadddr res)))) (test "de-register server" #t (let ((res #f)) (open-run-close tasks:server-deregister tasks:open-db "bob" pullport: 1234) (list? (open-run-close tasks:get-best-server tasks:open-db)))) (define hostinfo #f) (test "get-best-server" #t (let ((dat (open-run-close tasks:get-best-server tasks:open-db))) (set! hostinfo dat) ;; host ip pullport pubport (and (string? (car dat)) (number? (caddr dat))))) (test #f #t (let ((zmq-socket (server:client-connect (cadr hostinfo) (caddr hostinfo) ;; (cadddr hostinfo) ))) (set! *runremote* zmq-socket) (string? (car *runremote*)))) (test #f #t (let ((res (server:client-login *runremote*))) (car res))) (test #f #t (car (cdb:login *runremote* *toppath* *my-client-signature*))) ;;====================================================================== ;; C O N F I G F I L E S ;;====================================================================== (define conffile #f) |
︙ | ︙ |