Index: Makefile ================================================================== --- Makefile +++ Makefile @@ -27,11 +27,11 @@ all : $(PREFIX)/bin/.$(ARCHSTR) mtest # add dboard mtut and tcmt back later # module source files MSRCFILES = autoload.scm dbi.scm ducttape-lib.scm pkts.scm stml2.scm \ - cookie.scm mutils.scm mtargs.scm apimod.scm \ + cookie.scm mutils.scm mtargs.scm apimod.scm ulex.scm \ configfmod.scm commonmod.scm dbmod.scm rmtmod.scm \ debugprint.scm mtver.scm csv-xml.scm servermod.scm \ hostinfo.scm adjutant.scm processmod.scm testsmod.scm \ itemsmod.scm keysmod.scm mtmod.scm rmtmod.scm \ tasksmod.scm pgdb.scm launchmod.scm runsmod.scm \ @@ -99,19 +99,23 @@ mofiles/launchmod.o : mofiles/rmtmod.o mofiles/servermod.o mofiles/mtmod.o : mofiles/debugprint.o mofiles/portloggermod.o : mofiles/tasksmod.o mofiles/rmtmod.o : mofiles/apimod.o mofiles/rmtmod.o : mofiles/commonmod.o mofiles/portloggermod.o -mofiles/rmtmod.o : mofiles/itemsmod.o # mofiles/clientmod.o +mofiles/rmtmod.o : mofiles/itemsmod.o mofiles/ulex.o mofiles/runsmod.o : mofiles/rmtmod.o mofiles/archivemod.o mofiles/servermod.o : mofiles/commonmod.o mofiles/stml2.o : mofiles/cookie.o mofiles/dbi.o mofiles/tasksmod.o : mofiles/pgdb.o mofiles/dbmod.o mofiles/testsmod.o : mofiles/commonmod.o mofiles/testsmod.o : mofiles/itemsmod.o mofiles/rmtmod.o mofiles/tasksmod.o + +# split modules. Note: we can switch between ulex and ulex simple. +mofiles/ulex.o : ulex/ulex.scm ulex-simple/ulex.scm dashboard.o megatest.o : db_records.scm megatest-fossil-hash.scm + ADTLSCR=mt_laststep mt_runstep mt_ezstep HELPERS=$(addprefix $(PREFIX)/bin/,$(ADTLSCR)) DEPLOYHELPERS=$(addprefix deploytarg/,$(ADTLSCR)) MTESTHASH=$(shell fossil info|grep checkout:| awk '{print $$2}') Index: TODO ================================================================== --- TODO +++ TODO @@ -18,10 +18,13 @@ TODO ==== Loose ends ---------- + +15:09:29 error in calling find-and-mark-incomplete for run-id 5, exn=# + might be related to initial conditions in the db. (no run entry in runs table?). . -list-servers not correct . move *remotedat* into bigdata . add back server stats on exit (look in rmt:run in rmtmod.scm) Index: apimod.scm ================================================================== --- apimod.scm +++ apimod.scm @@ -29,11 +29,11 @@ ( api:run-server-process api:start-server api:dispatch-cmd api:execute-requests -api:process-request +;; api:process-request ) (import scheme chicken.base chicken.process-context.posix @@ -175,12 +175,13 @@ (seconds->year-work-week/day-time-fname (current-seconds)) "-"cleandbname".log")) (logf2 (conc logd "/server-" (seconds->year-work-week/day-time-fname (current-seconds)) "-"cleandbname"-")) - (cmd (conc "nbfake megatest -server - -area "apath - " -db "dbname" -autolog "logf2))) + (cmd (conc "nbfake megatest -server - -area "apath" -db "dbname) + ;; " -autolog "logf2 ;; the side log did not help. Ended up with two logs and the pid in the name was not that useful. + )) (if (not (directory-exists? logd)) (create-directory logd #t)) (system (conc "NBFAKE_LOG="logf" "cmd)))) ;; special function to get server @@ -374,11 +375,11 @@ ((general-call) (let ((stmtname (car params)) (run-id (cadr params)) (realparams (cddr params))) (db:general-call dbstruct stmtname run-id realparams))) ((sdb-qry) (apply sdb:qry params)) - ((ping) (current-process-id)) + ((ping) `(#t ,(current-process-id) ,(cadr params))) ;; (current-process-id)) ((get-changed-record-ids) (apply db:get-changed-record-ids dbstruct params)) ((get-run-record-ids) (apply db:get-run-record-ids dbstruct params)) ;; TESTMETA ((testmeta-get-record) (apply db:testmeta-get-record dbstruct params)) @@ -420,11 +421,11 @@ ;; api:process-request ;; db:* ;; ;; NB// Runs on the server as part of the server loop ;; -(define (api:process-request dbstruct indat) ;; the $ is the request vars proc +#;(define (api:process-request dbstruct indat) ;; the $ is the request vars proc (let* ((cmd-in (alist-ref 'cmd indat)) ;; ($ 'cmd)) (cmd (if (string? cmd-in)(string->symbol cmd-in) cmd-in)) (params (alist-ref 'params indat)) (key (alist-ref 'key indat)) ;; TODO - add this back ;; (doprint (apply common:low-noise-print 10 params)) Index: build-assist/README ================================================================== --- build-assist/README +++ build-assist/README @@ -1,28 +1,12 @@ -Here is how I like to install chicken for building Megatest. - -This guide assumes you have the Megatest fossil and are in the build-assist directory and -that you have the opensrc fossil with uv synced: - -fossil clone https://www.kiatoa.com/fossils/megatest -fossil clone https://www.kiatoa.com/fossils/opensrc;cd opensrc;fossil uv sync - -Make a build directory and go to it: - -mkdir build;cd build - -Make a destination directory and set PREFIX - -export PREFIX=/opt/chicken/5.3.0; mkdir -p $PREFIX - -Get chicken: - -wget https://code.call-cc.org/releases/5.3.0/chicken-5.3.0.tar.gz - -Extract, build, and install chicken: - -tar xf chicken-5.3.0.tar.gz; cd chicken-5.3.0; make PLATFORM=linux PREFIX=$PREFIX install; cd .. - -Install all needed eggs. -for egg in $(cat ../ck5-egg.list);do echo $egg;ck5 chicken-install $egg;done - -Now run the script ../iup-compile.sh for remaining instructions +README for IUP + + IUP is a portable toolkit for building graphical user interfaces. It offers a configuration API in three basic languages: C, Lua and LED. IUP's purpose is to allow a program to be executed in different systems without any modification, therefore it is highly portable. Its main advantages are: + * high performance, due to the fact that it uses native interface elements. + * fast learning by the user, due to the simplicity of its API. + + Build instructions and usage are available in the IUP documentation. + + For complete information, visit IUP's web site at http://www.tecgraf.puc-rio.br/iup + or access its documentation in the HTML folder. + +(end of README) Index: build-assist/ck5-eggs.list ================================================================== --- build-assist/ck5-eggs.list +++ build-assist/ck5-eggs.list @@ -1,48 +0,0 @@ -csm -address-info -ansi-escape-sequences -apropos -base64 -crypt -csv-abnf -directory-utils -dot-locking -filepath -fmt -format -http-client -itemsmod -json -linenoise -mailbox -md5 -message-digest -nanomsg -postgresql -queues -regex -regex-case -rfc3339 -s11n -sha1 -simple-exceptions -slice -sparse-vectors -spiffy -spiffy-directory-listing -spiffy-request-vars -sql-de-lite -sqlite3 -sql-null -srfi-1 -srfi-13 -srfi-19 -sxml-modifications -sxml-serializer -sxml-transforms -system-information -tcp6 -test -typed-records -uri-common -z3 ADDED build-assist/ck5-full-eggs.list Index: build-assist/ck5-full-eggs.list ================================================================== --- /dev/null +++ build-assist/ck5-full-eggs.list @@ -0,0 +1,49 @@ +csm +address-info +ansi-escape-sequences +apropos +base64 +breadline +crypt +csv-abnf +directory-utils +dot-locking +filepath +fmt +format +http-client +itemsmod +json +linenoise +mailbox +md5 +message-digest +nanomsg +postgresql +queues +regex +regex-case +rfc3339 +s11n +sha1 +simple-exceptions +slice +sparse-vectors +spiffy +spiffy-directory-listing +spiffy-request-vars +sql-de-lite +sqlite3 +sql-null +srfi-1 +srfi-13 +srfi-19 +sxml-modifications +sxml-serializer +sxml-transforms +system-information +tcp6 +test +typed-records +uri-common +z3 Index: build-assist/iup-compile.sh ================================================================== --- build-assist/iup-compile.sh +++ build-assist/iup-compile.sh @@ -15,12 +15,11 @@ echo " cp *.a *.so $PREFIX/lib" echo " cp include/*.h $PREFIX/include" echo " 4. run the chicken-install like this:" echo "If you use a wrapper (e.g. ck5) to create the chicken environment:" -echo "CSC_OPTIONS=\"-I$PREFIX/include -I$PREFIX/include/im -I$PREFIX/include/cd -I$PREFIX/include/iup -L$PREFIX/lib -C -std=gnu99\" ck5 chicken-install iup -feature disable-iup-matrixex" +echo "CSC_OPTIONS=\"-I$PREFIX/include -I$PREFIX/include/im -I$PREFIX/include/cd -I$PREFIX/include/iup -L$PREFIX/lib -C -std=gnu99\" ck5 chicken-install iup canvas-draw -feature disable-iup-matrixex" echo "else:" -echo "CSC_OPTIONS=\"-I$PREFIX/include -I$PREFIX/include/im -I$PREFIX/include/cd -I$PREFIX/include/iup -L$PREFIX/lib -C -std=gnu99\" chicken-install iup" -echo "Then repeat for canvas-draw" +echo "CSC_OPTIONS=\"-I$PREFIX/include -I$PREFIX/include/im -I$PREFIX/include/cd -I$PREFIX/include/iup -L$PREFIX/lib -C -std=gnu99\" chicken-install iup canvas-draw" # (export PREFIX=/home/matt/data/buildall/ck5.2;CSC_OPTIONS="-I/home/matt/data/buildall/ck5.2/include -I/home/matt/data/buildall/ck5.2/include/im -I/home/matt/data/buildall/ck5.2/include/cd -I/home/matt/data/buildall/ck5.2/include/iup -L/home/matt/data/buildall/ck5.2/lib -C -std=gnu99" ck5 chicken-install iup -feature disable-iup-matrixex) Index: commonmod.scm ================================================================== --- commonmod.scm +++ commonmod.scm @@ -386,15 +386,15 @@ *db-access-mutex* *db-transaction-mutex* *db-cache-path* *db-with-db-mutex* *db-api-call-time* +*didsomething* *no-sync-db* *my-signature* *transport-type* *logged-in-clients* -*server-info* *server-run* *run-id* *server-kind-run* *home-host* *heartbeat-mutex* @@ -925,10 +925,11 @@ ;; config files. Allows communicating between confgs ;; (define *user-hash-data* (make-hash-table)) (define *db-keys* #f) +(define *didsomething* #f) (define *pkts-info* (make-hash-table)) ;; store stuff like the last parent here (define *configinfo* #f) ;; raw results from setup, includes toppath and table from megatest.config (define *runconfigdat* #f) ;; run configs data (define *configdat* #f) ;; megatest.config data ==> moved to configfmod @@ -974,12 +975,10 @@ ;; replaced by *rmt:remote* ;; (define *runremote* #f) ;; if set up for server communication this will hold ;; (define *max-cache-size* 0) (define *logged-in-clients* (make-hash-table)) ;; (define *server-id* #f) -(define *server-info* #f) ;; good candidate for easily convert to non-global -;; (define *time-to-exit* #f) (define *server-run* #t) (define *run-id* #f) (define *server-kind-run* (make-hash-table)) (define *home-host* #f) ;; (define *total-non-write-delay* 0) Index: dashboard.scm ================================================================== --- dashboard.scm +++ dashboard.scm @@ -98,10 +98,11 @@ typed-records sparse-vectors format srfi-4 srfi-14 + srfi-18 ) ;; (include "common_records.scm") ;; (include "db_records.scm") ;; (include "run_records.scm") @@ -3754,7 +3755,12 @@ ;; ease debugging by loading ~/.dashboardrc (let ((debugcontrolf (get-debugcontrolf))) (if debugcontrolf (load debugcontrolf))) -(main) +(import srfi-18) + +(thread-join! + (thread-start! + (make-thread main "main"))) + Index: dbmod.scm ================================================================== --- dbmod.scm +++ dbmod.scm @@ -3445,14 +3445,12 @@ run-id)))) ;; NOT IN (SELECT id FROM runs WHERE state='deleted');") ;; NEW BEHAVIOR: Look only at single run with run-id ;; ;; (define (db:get-running-stats dbstruct run-id) -(define (db:get-count-tests-running-for-run-id dbstruct run-id) ;; fastmode) - (let* ((qry ;; (if fastmode - ;; "SELECT count(id) FROM tests WHERE state in ('RUNNING','LAUNCHED','REMOTEHOSTSTART') AND run_id=? LIMIT 1;" - "SELECT count(id) FROM tests WHERE state in ('RUNNING','LAUNCHED','REMOTEHOSTSTART') AND run_id=?;")) ;; ) +(define (db:get-count-tests-running-for-run-id dbstruct run-id) + (let* ((qry "SELECT count(id) FROM tests WHERE state in ('RUNNING','LAUNCHED','REMOTEHOSTSTART') AND run_id=?;")) ;; ) (db:with-db dbstruct run-id #f (lambda (db) Index: debugprint.scm ================================================================== --- debugprint.scm +++ debugprint.scm @@ -6,10 +6,12 @@ ;;(import scheme chicken data-structures extras files ports) (import scheme chicken.base chicken.string + chicken.time + chicken.time.posix chicken.port chicken.process-context chicken.process-context.posix (prefix mtargs args:) @@ -21,11 +23,11 @@ ;; debug stuff ;;====================================================================== (define verbosity (make-parameter '())) (define *default-log-port* (current-error-port)) -(define debug:print-logger (make-parameter #f)) ;; se to a proc to call on every logging print +(define debug:print-logger (make-parameter #f)) ;; set to a proc to call on every logging print (define (debug:setup) (let ((debugstr (or (args:get-arg "-debug") (args:get-arg "-debug-noprop") (get-environment-variable "MT_DEBUG_MODE")))) @@ -109,17 +111,25 @@ (if (debug:print-logger) ;; NOTE: turn params into string to keep things simple for now ((debug:print-logger)(conc "REMOTE ("(get-host-name)", pid="(current-process-id)") " (string-intersperse (map conc params) " ") "; " (string-intersperse (command-line-arguments) " "))))) -(define (debug:print n e . params) +(define debug:enable-timestamp (make-parameter #t)) + +(define (debug:timestamp) + (if (debug:enable-timestamp) + (conc (time->string + (seconds->local-time (current-seconds)) "%H:%M:%S") " ") + "")) + + (define (debug:print n e . params) (if (debug:debug-mode n) (with-output-to-port (or e (current-error-port)) (lambda () ;; (if *logging* ;; (db:log-event (apply conc params)) - (apply print params) + (apply print (debug:timestamp) params) (debug:handle-remote-logging params) ))) #t ;; only here to make remote stuff happy. It'd be nice to fix that ... ) @@ -126,31 +136,31 @@ (define (debug:print-error n e . params) ;; normal print (if (debug:debug-mode n) (with-output-to-port (if (port? e) e (current-error-port)) (lambda () - (apply print "ERROR: " params) + (apply print "ERROR: " (debug:timestamp) params) (debug:handle-remote-logging (cons "ERROR: " params)) ))) ;; pass important messages to stderr (if (and (eq? n 0)(not (eq? e (current-error-port)))) (with-output-to-port (current-error-port) (lambda () - (apply print "ERROR: " params) + (apply print "ERROR: " (debug:timestamp) params) )))) (define (debug:print-info n e . params) (if (debug:debug-mode n) (with-output-to-port (if (port? e) e (current-error-port)) (lambda () - (apply print "INFO: (" n ") " params) ;; res) + (apply print "INFO: (" n ") "(debug:timestamp) params) ;; res) (debug:handle-remote-logging (cons "INFO: " params)) )))) (define (debug:print-warn n e . params) (if (debug:debug-mode n) (with-output-to-port (if (port? e) e (current-error-port)) (lambda () - (apply print "WARN: (" n ") " params) ;; res) + (apply print "WARN: (" n ") " (debug:timestamp) params) ;; res) (debug:handle-remote-logging (cons "WARN: " params)) )))) ) Index: launchmod.scm ================================================================== --- launchmod.scm +++ launchmod.scm @@ -1888,11 +1888,11 @@ ;; 0 RUNNING ==> this is actually the first condition, should not get here (define (runs:end-of-run-check run-id ) (let* ((not-completed-cnt (rmt:get-not-completed-cnt run-id)) (running-cnt (rmt:get-count-tests-running-for-run-id run-id)) - (all-test-launched (rmt:get-var run-id (conc "lunch-complete-" run-id))) + (all-test-launched (rmt:get-var run-id (conc "launch-complete-" run-id))) (current-state (rmt:get-run-state run-id)) (current-status (rmt:get-run-status run-id))) ;;get-vars run-id to query metadata table to check if all completed. if all-test-launched = yes then only not-completed-cnt = 0 means everyting is completed if no entry found in the table do nothing (debug:print 0 *default-log-port* "Running test cnt :" running-cnt) (rmt:set-state-status-and-roll-up-run run-id current-state current-status) Index: megatest.scm ================================================================== --- megatest.scm +++ megatest.scm @@ -156,16 +156,25 @@ rmtmod runsmod servermod tasksmod testsmod - + + ulex ) - + +;; ;; ulex parameters +;; (work-method 'direct) +;; (return-method 'direct) + + ;; ulex parameters + (work-method 'mailbox) + (return-method 'mailbox) + ;; fake out readline usage of toplevel-command (define (toplevel-command . a) #f) - +(define *didsomething* #f) (define *db* #f) ;; this is only for the repl, do not use in general!!!! ;; (include "common_records.scm") ;; (include "key_records.scm") ;; (include "db_records.scm") @@ -319,11 +328,11 @@ (if (not (directory-exists? log-dir)) (system (conc "mkdir -p " log-dir))) (open-output-file logpath)) (exn () (debug:print-error 0 *default-log-port* "Could not open log file for write: "logpath-in) - (define *didsomething* #t) + (set! *didsomething* #t) (exit 1)))) ;; Disabled help items ;; -rollup : (currently disabled) fill run (set by :runname) with latest test(s) ;; from prior runs with same keys @@ -503,18 +512,10 @@ Version " megatest-version ", built from " megatest-fossil-hash )) (define (main) (make-and-init-bigdata) - - ;; set up the functions in http transport - (hash-table-set! *http-functions* 'api:process-request api:process-request) - (hash-table-set! *http-functions* 'http-transport:main-page http-transport:main-page) - (hash-table-set! *http-functions* 'http-transport:show-jquery http-transport:show-jquery) - (hash-table-set! *http-functions* 'http-transport:html-test-log http-transport:html-test-log) - (hash-table-set! *http-functions* 'http-transport:html-dboard http-transport:html-dboard) - ;; load the ~/.megatestrc file, put (use trace)(trace-call-sites #t)(trace function-you-want-to-trace) in this file ;; (let ((debugcontrolf (conc (get-environment-variable "HOME") "/.megatestrc"))) (if (common:file-exists? debugcontrolf) (load debugcontrolf))) @@ -828,12 +829,10 @@ (if (args:get-arg "-version") (begin (print (common:version-signature)) ;; (print megatest-version) (exit))) - (define *didsomething* #f) - ;; Overall exit handling setup immediately ;; (if (or (args:get-arg "-process-reap")) ;; (args:get-arg "-runtests") ;; (args:get-arg "-execute") @@ -1067,15 +1066,17 @@ (else (pp data)))))) (if out-file (close-output-port out-port)) (exit) ;; yes, bending the rules here - need to exit since this is a utility )) - - (if (args:get-arg "-ping") + + ;; disabled for now + + #;(if (args:get-arg "-ping") (let* ((server-id (string->number (args:get-arg "-ping"))) ;; extract run-id (i.e. no ":" (host:port (args:get-arg "-ping"))) - (server:ping (or server-id host:port) #f do-exit: #t))) + (server-ready? (or server-id host:port) #f do-exit: #t))) ;;====================================================================== ;; Capture, save and manipulate environments ;;====================================================================== @@ -2638,11 +2639,11 @@ ;;(debug:print-info 13 *default-log-port* "thread-join! watchdog") ;; join the watchdog thread if it has been thread-start!ed (it may not have been started in the case of a server that never enters running state) ;; (symbols returned by thread-state: created ready running blocked suspended sleeping terminated dead) ;; TODO: for multiple areas, we will have multiple watchdogs; and multiple threads to manage - (let* ((watchdog (bdat-watchdog *bdat*))) + #;(let* ((watchdog (bdat-watchdog *bdat*))) (if (thread? watchdog) (case (thread-state watchdog) ((ready running blocked sleeping terminated dead) (thread-join! watchdog))))) @@ -2660,9 +2661,13 @@ (else (exit 3))))) ) ) -(import megatest-main) -(import commonmod) -(main) +(import megatest-main commonmod) +(import srfi-18) + +(thread-join! + (thread-start! + (make-thread main))) + Index: pgdb.scm ================================================================== --- pgdb.scm +++ pgdb.scm @@ -12,11 +12,11 @@ chicken.condition chicken.string chicken.sort list-utils - postgresql + ;; postgresql srfi-1 srfi-69 typed-records (prefix dbi dbi:) Index: rmtmod.scm ================================================================== --- rmtmod.scm +++ rmtmod.scm @@ -16,10 +16,14 @@ ;; You should have received a copy of the GNU General Public License ;; along with Megatest. If not, see . ;;====================================================================== +;; generate entries for ~/.megatestrc with the following +;; +;; grep define ../rmt.scm | grep rmt: |perl -pi -e 's/\(define\s+\((\S+)\W.*$/\1/'|sort -u + (declare (unit rmtmod)) (declare (uses apimod)) (declare (uses commonmod)) (declare (uses configfmod)) @@ -30,10 +34,11 @@ (declare (uses mtver)) (declare (uses pgdb)) (declare (uses portloggermod)) (declare (uses servermod)) (declare (uses tasksmod)) +(declare (uses ulex)) (module rmtmod * (import scheme @@ -63,11 +68,11 @@ ;; http-client ;; intarweb matchable md5 message-digest - nng ;; nanomsg + ;; nng ;; nanomsg (prefix base64 base64:) (prefix sqlite3 sqlite3:) regex s11n ;; spiffy @@ -96,69 +101,51 @@ portloggermod (prefix mtargs args:) servermod stml2 tasksmod - ) - -(defstruct alldat - (areapath #f) - (ulexdat #f) - ) - - -;; (require-extension (srfi 18) extras tcp s11n) -;; -;; -;; (use srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest posix-extras) -;; -;; (use spiffy uri-common intarweb http-client spiffy-request-vars intarweb spiffy-directory-listing) -;; + + ulex + ) + ;; Configurations for server ;; (tcp-buffer-size 2048) ;; (max-connections 2048) -;; info about me as a server +;; info about me as a listener and my connections to db servers +;; stored (for now) in *db-serv-info* ;; (defstruct servdat (host #f) (port #f) (uuid #f) - (rep #f) (dbfile #f) - (api-url #f) - (api-uri #f) - (api-req #f) + (uconn #f) ;; this is the listener *FOR THIS PROCESS* + (mode #f) (status 'starting) (trynum 0) ;; count the number of ports we've tried + (conns (make-hash-table)) ;; apath/dbname => conndat ) + +(define *db-serv-info* (make-servdat)) (define (servdat->url sdat) (conc (servdat-host sdat)":"(servdat-port sdat))) -;; generate entries for ~/.megatestrc with the following +;; db servers contact info ;; -;; grep define ../rmt.scm | grep rmt: |perl -pi -e 's/\(define\s+\((\S+)\W.*$/\1/'|sort -u - -(defstruct remotedat - (conns (make-hash-table)) ;; apath/dbname => conndat - ) - (defstruct conndat (apath #f) (dbname #f) (fullname #f) (hostport #f) (ipaddr #f) (port #f) - (socket #f) (srvpkt #f) (srvkey #f) (lastmsg 0) - (expires 0) - (inport #f) - (outport #f)) + (expires 0)) (define *srvpktspec* `((server (host . h) (port . p) (servkey . k) @@ -168,43 +155,37 @@ ;;====================================================================== ;; S U P P O R T F U N C T I O N S ;;====================================================================== -;; replaces *runremote* -(define *remotedat* (make-remotedat)) - -;; -> http://abc.com:900/ -;; -(define (conndat->uri conn entrypoint) - (conc "http://"(conndat-ipaddr conn)":"(conndat-port conn)"/"entrypoint)) - ;; set up the api proc, seems like there should be a better place for this? -(define api-proc (make-parameter conc)) -(api-proc api:process-request) +;; +;; IS THIS NEEDED ANYMORE? TODO - REMOVE IF POSSIBLE +;; +;; (define api-proc (make-parameter conc)) +;; (api-proc api:execute-requests) ;; do we have a connection to apath dbname and ;; is it not expired? then return it ;; ;; else setup a connection ;; ;; if that fails, return '(#f "some reason") ;; NB// convert to raising an exception ;; (define (rmt:get-conn remdat apath dbname) - (let* ((fullname (db:dbname->path apath dbname)) - (conn (hash-table-ref/default (remotedat-conns remdat) fullname #f))) - (if (and conn - (< (current-seconds) (conndat-expires conn))) - conn - #f))) - -(define (rmt:find-main-server apath dbname) + (let* ((fullname (db:dbname->path apath dbname))) + (hash-table-ref/default (servdat-conns remdat) fullname #f))) + +(define (rmt:drop-conn remdat apath dbname) + (let* ((fullname (db:dbname->path apath dbname))) + (hash-table-delete! (servdat-conns remdat) fullname))) + +(define (rmt:find-main-server uconn apath dbname) (let* ((pktsdir (get-pkts-dir apath)) (all-srvpkts (get-all-server-pkts pktsdir *srvpktspec*)) - ;; (dbpath (conc apath "/" dbname)) (viable-srvs (get-viable-servers all-srvpkts dbname))) - (get-the-server apath viable-srvs))) + (get-the-server uconn apath viable-srvs))) (define *connstart-mutex* (make-mutex)) (define *last-main-start* 0) @@ -214,35 +195,53 @@ ;; connections for other servers happens by requesting from main ;; ;; TODO: This is unnecessarily re-creating the record in the hash table ;; (define (rmt:open-main-connection remdat apath) - (let* ((fullpath (db:dbname->path apath "/.db/main.db")) - (conns (remotedat-conns remdat)) - (conn (hash-table-ref/default conns fullpath #f))) ;; TODO - create call for this + (let* ((fullpath (db:dbname->path apath ".db/main.db")) + (conns (servdat-conns remdat)) + (conn (rmt:get-conn remdat apath ".db/main.db")) ;; (hash-table-ref/default conns fullpath #f)) ;; TODO - create call for this + (start-rmt:run (lambda () + (let* ((th1 (make-thread (lambda ()(rmt:run (get-host-name))) "non-db mode server"))) + (thread-start! th1) + (thread-sleep! 1) + (let loop ((count 0)) + (assert (< count 30) "FATAL: responder failed to initialize in rmt:open-main-connection") + (if (or (not *db-serv-info*) + (not (servdat-uconn *db-serv-info*))) + (begin + (thread-sleep! 1) + (loop (+ count 1))) + (begin + (servdat-mode-set! *db-serv-info* 'non-db) + (servdat-uconn *db-serv-info*))))))) + (myconn (servdat-uconn *db-serv-info*))) (cond + ((not myconn) + (start-rmt:run) + (rmt:open-main-connection remdat apath)) ((and conn ;; conn is NOT a socket, just saying ... (< (current-seconds) (conndat-expires conn))) #t) ;; we are current and good to go - we'll deal elsewhere with a server that was killed or died ((and conn (>= (current-seconds)(conndat-expires conn))) (debug:print-info 0 *default-log-port* "connection to "fullpath" server expired. Reconnecting.") - (if (conndat-socket conn) - (nng-close! (conndat-socket conn))) - (hash-table-set! conns fullpath #f) ;; clean up + (rmt:drop-conn remdat apath ".db/main.db") ;; (rmt:open-main-connection remdat apath)) (else ;; Below we will find or create and connect to main + (debug:print-info 0 *default-log-port* "rmt:open-main-connection - starting from scratch") (let* ((dbname (db:run-id->dbname #f)) - (the-srv (rmt:find-main-server apath dbname)) + (the-srv (rmt:find-main-server myconn apath dbname)) (start-main-srv (lambda () ;; call IF there is no the-srv found (mutex-lock! *connstart-mutex*) (if (> (- (current-seconds) *last-main-start*) 5) ;; at least four seconds since last attempt to start main server (begin (api:run-server-process apath dbname) (set! *last-main-start* (current-seconds)) - (thread-sleep! 1))) + (thread-sleep! 1)) + (thread-sleep! 0.25)) (mutex-unlock! *connstart-mutex*) (rmt:open-main-connection remdat apath) ;; TODO: Add limit to number of tries ))) (if (not the-srv) ;; have server, try connecting to it (start-main-srv) @@ -255,52 +254,53 @@ (new-the-srv (make-conndat apath: apath dbname: dbname fullname: fullpath hostport: srv-addr - socket: (open-nn-connection srv-addr) + ;; socket: (open-nn-connection srv-addr) - TODO - open ulex connection? ipaddr: ipaddr port: port srvpkt: the-srv srvkey: srvkey ;; generated by rmt:get-signature on the server side lastmsg: (current-seconds) - expires: (+ (current-seconds) 60) ;; this needs to be gathered during the ping + expires: (+ (current-seconds) + (server:expiration-timeout) + -2) ;; this needs to be gathered during the ping ))) (hash-table-set! conns fullpath new-the-srv))) #t))))) -;; NB// remdat is a remotedat struct +;; NB// sinfo is a servdat struct ;; -(define (rmt:general-open-connection remdat apath dbname #!key (num-tries 5)) +(define (rmt:general-open-connection sinfo apath dbname #!key (num-tries 5)) (assert (not (equal? dbname ".db/main.db")) "ERROR: general-open-connection should never be called with main as the db") - (let* ((mdbname (db:run-id->dbname #f)) + (let* ((mdbname ".db/main.db") ;; (db:run-id->dbname #f)) TODO: put this back to the lookup when stable (fullname (db:dbname->path apath dbname)) - (conns (remotedat-conns remdat)) - (mconn (rmt:get-conn remdat apath mdbname))) - (if (and mconn + (conns (servdat-conns sinfo)) + (mconn (rmt:get-conn sinfo apath ".db/main.db")) + (dconn (rmt:get-conn sinfo apath dbname))) + #;(if (and mconn (not (debug:print-logger))) (begin (debug:print-info 0 *default-log-port* "Turning on logging to main, look in logs dir for main log.") (debug:print-logger rmt:log-to-main))) (cond - ((or (not mconn) ;; no channel open to main? - (< (conndat-expires mconn)(+ (current-seconds) 2))) ;; restablish connection if less than 2 seconds on the lease - (if mconn ;; previously opened - clean up NB// consolidate this with the similar code in open main above - (begin - (debug:print-info 0 *default-log-port* "Clearing out connection to main that has expired.") - (nng-close! (conndat-socket mconn)) - (hash-table-set! conns fullname #f))) - (rmt:open-main-connection remdat apath) - (rmt:general-open-connection remdat apath mdbname)) - ((not (rmt:get-conn remdat apath dbname)) ;; no channel open to dbname? - (let* ((res (rmt:send-receive-real remdat apath mdbname 'get-server `(,apath ,dbname)))) + ((and mconn + dconn + (< (current-seconds)(conndat-expires dconn))) + #t) ;; good to go + ((not mconn) ;; no channel open to main? open it... + (rmt:open-main-connection sinfo apath) + (rmt:general-open-connection sinfo apath dbname num-tries: (- num-tries 1))) + ((not dconn) ;; no channel open to dbname? + (let* ((res (rmt:send-receive-real sinfo apath mdbname 'get-server `(,apath ,dbname)))) (case res ((server-started) (if (> num-tries 0) (begin (thread-sleep! 2) - (rmt:general-open-connection remdat apath dbname num-tries: (- num-tries 1))) + (rmt:general-open-connection sinfo apath dbname num-tries: (- num-tries 1))) (begin (debug:print-error 0 *default-log-port* "Failed to start servers needed or open channel to "apath", "dbname) (exit 1)))) (else (if (list? res) ;; server has been registered and the info was returned. pass it on. @@ -317,23 +317,24 @@ fullname (make-conndat apath: apath dbname: dbname hostport: (conc host":"port) - socket: (open-nn-connection (conc host":"port)) + ;; socket: (open-nn-connection (conc host":"port)) ;; TODO - open ulex connection? ipaddr: ipaddr port: port srvkey: servkey lastmsg: (current-seconds) - expires: (+ (current-seconds) 60)))) + expires: (+ (current-seconds) + (server:expiration-timeout) + -2)))) (else (debug:print-info 0 *default-log-port* "return data from starting server did not match host port servkey pid ipaddr apath dbname " res))) res) (begin (debug:print-info 0 *default-log-port* "Unexpected result: " res) res))))))) - #t)) ;;====================================================================== ;; FOR DEBUGGING SET TO #t @@ -342,67 +343,49 @@ (define *dbstruct* (make-dbr:dbstruct)) ;; Defaults to current area ;; (define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f)) - ;; (if (not *remotedat*)(set! *remotedat* (make-remotedat))) (let* ((apath *toppath*) - (remdat *remotedat*) - (conns (remotedat-conns remdat)) ;; just checking that remdat is a remotedat + (sinfo *db-serv-info*) (dbname (db:run-id->dbname rid))) (if *localmode* - (let* ((dbdat (dbr:dbstruct-get-dbdat *dbstruct* dbname)) - (indat `((cmd . ,cmd)(params . ,params)))) - (api:process-request *dbstruct* indat) - ;; (api:process-request dbdat indat) - ) + (api:execute-requests *dbstruct* cmd params) (begin - (rmt:open-main-connection remdat apath) - (if rid (rmt:general-open-connection remdat apath dbname)) - (rmt:send-receive-real remdat apath dbname cmd params))))) - -#;(define (rmt:send-receive-setup conn) - (if (not (conndat-inport conn)) - (let-values (((i o) (tcp-connect (conndat-ipaddr conn) - (conndat-port conn)))) - (conndat-inport-set! conn i) - (conndat-outport-set! conn o)))) - + (rmt:open-main-connection sinfo apath) + (if rid (rmt:general-open-connection sinfo apath dbname)) + #;(if (not (member cmd '(log-to-main))) + (debug:print-info 0 *default-log-port* "rmt:send-receive "cmd" params="params)) + (rmt:send-receive-real sinfo apath dbname cmd params))))) + ;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed ;; sometime in the future ;; -(define (rmt:send-receive-real remdat apath dbname cmd params) - (let* ((conn (rmt:get-conn remdat apath dbname))) - (assert conn "FATAL: rmt:send-receive-real called without the needed channels opened") - (assert (conndat-socket conn) "FATAL: rmt:send-receive-real called without the channel socket opened.") - (let* ((soc (conndat-socket conn)) - (key #f) - (host (conndat-ipaddr conn)) - (port (conndat-port conn)) - (payload `((cmd . ,cmd) - (key . ,(conndat-srvkey conn)) - (params . ,params))) - (res (send-receive-nn soc ;; (open-send-receive-nn (conc host":"port) - (sexpr->string payload)))) - (if (member res '("#")) ;; TODO - fix this in string->sexpr - #f - (string->sexpr res))))) +(define (rmt:send-receive-real sinfo apath dbname cmd params) + (assert (not (eq? 'primordial (thread-name (current-thread)))) "FATAL: Do not call rmt:send-receive-real in the primodial thread.") + (let* ((cdat (rmt:get-conn sinfo apath dbname))) + (assert cdat "FATAL: rmt:send-receive-real called without the needed channels opened") + (let* ((uconn (servdat-uconn sinfo)) ;; get the interface to ulex + ;; then send-receive using the ulex layer to host-port stored in cdat + (res (send-receive uconn (conndat-hostport cdat) cmd params)) + #;(th1 (make-thread (lambda () + (set! res (send-receive uconn (conndat-hostport cdat) cmd params))) + "send-receive thread"))) + ;; (thread-start! th1) + ;; (thread-join! th1) ;; gratuitious thread stuff is so that mailbox is not used in primordial thead + ;; since we accessed the server we can bump the expires time up + (conndat-expires-set! cdat (+ (current-seconds) + (server:expiration-timeout) + -2)) ;; two second margin for network time misalignments etc. + res))) ;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed ;; sometime in the future. ;; ;; Purpose - call the main.db server and request a server be started ;; for the given area path and dbname ;; -;; (define (rmt:send-receive-server-start remdat apath dbname) -;; (let* ((conn (rmt:get-conn remdat apath dbname))) -;; (assert conn "FATAL: Unable to connect to db "apath"/"dbname) -;; #;(let* ((res (with-input-from-request -;; (conndat->uri conn "api") -;; `((params . (,apath ,dbname))) -;; read-string))) -;; (string->sexpr res)))) (define (rmt:print-db-stats) (let ((fmtstr "~40a~7-d~9-d~20,2-f")) ;; "~20,2-f" (debug:print 18 *default-log-port* "DB Stats, "(seconds->year-week/day-time (current-seconds))"\n=====================") (debug:print 18 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg")) @@ -454,11 +437,11 @@ (rmt:send-receive 'kill-server #f (list run-id))) (define (rmt:start-server run-id) (rmt:send-receive 'start-server #f (list run-id))) -(define (rmt:get-server-info apath dbname) +(define (rmt:server-info apath dbname) (rmt:send-receive 'get-server-info #f (list apath dbname))) ;;====================================================================== ;; M I S C ;;====================================================================== @@ -1522,21 +1505,21 @@ (not (equal? (substring (->string megatest-version) 0 4) (substring (conc (common:get-last-run-version)) 0 4)))) ;; host and port are used to ensure we are remove proper records (define (rmt:server-shutdown host port) - (let ((dbfile (servdat-dbfile *server-info*))) + (let ((dbfile (servdat-dbfile *db-serv-info*))) (debug:print-info 0 *default-log-port* "dbfile is "dbfile) (if dbfile (let* ((am-server (args:get-arg "-server")) (dbfile (args:get-arg "-db")) (apath *toppath*) - (remdat *remotedat*)) ;; foundation for future fix + #;(sinfo *remotedat*)) ;; foundation for future fix (if *dbstruct-db* (let* ((dbdat (db:get-dbdat *dbstruct-db* apath dbfile)) (db (dbr:dbdat-db dbdat)) - (inmem (dbr:dbdat-db dbdat)) + (inmem (dbr:dbdat-db dbdat)) ;; WRONG ) ;; do a final sync here (debug:print-info 0 *default-log-port* "Doing final sync for "apath" "dbfile" at "(current-seconds)) (db:sync-inmem->disk *dbstruct-db* apath dbfile force-sync: #t) ;; let's finalize here @@ -1551,24 +1534,24 @@ (debug:print-info 0 *default-log-port* "Db was never opened, no cleanup to do.")) (if (not am-server) (debug:print-info 0 *default-log-port* "I am not a server, should NOT get here!") (if (string-match ".*/main.db$" dbfile) (let ((pkt-file (conc (get-pkts-dir *toppath*) - "/" (servdat-uuid *server-info*) + "/" (servdat-uuid *db-serv-info*) ".pkt"))) (debug:print-info 0 *default-log-port* "removing pkt "pkt-file) (delete-file* pkt-file) (debug:print-info 0 *default-log-port* "Releasing lock (if any) for "dbfile ", host "host", port "port) (db:with-lock-db - (servdat-dbfile *server-info*) + (servdat-dbfile *db-serv-info*) (lambda (dbh dbfile) (db:release-lock dbh dbfile host port)))) ;; I'm not the server - should not have a lock to remove - (let* ((sdat *server-info*) ;; we have a run-id server + (let* ((sdat *db-serv-info*) ;; we have a run-id server (host (servdat-host sdat)) (port (servdat-port sdat)) (uuid (servdat-uuid sdat)) - (res (rmt:deregister-server remdat *toppath* host port uuid dbfile))) + (res (rmt:deregister-server *db-serv-info* *toppath* host port uuid dbfile))) (debug:print-info 0 *default-log-port* "deregistered-server, res="res) (debug:print-info 0 *default-log-port* "deregistering server "host":"port" with uuid "uuid) ))))))) (define (std-exit-procedure) @@ -1584,16 +1567,20 @@ (if (and no-hurry (debug:debug-mode 18)) (rmt:print-db-stats)) (let ((th1 (make-thread (lambda () ;; thread for cleaning up, give it five seconds (let* ((start-time (current-seconds))) - (if (and *server-info* - *unclean-shutdown*) - (begin - (debug:print-info 0 *default-log-port* "Unclean server exit, calling server-shtudown") - (rmt:server-shutdown (servdat-host *server-info*) - (servdat-port *server-info*)))) + (if *db-serv-info* + (let* ((host (servdat-host *db-serv-info*)) + (port (servdat-port *db-serv-info*))) + (debug:print-info 0 *default-log-port* "Shutting down server/responder.") + ;; + ;; TODO - add flushing/waiting on the work queue + ;; + (rmt:server-shutdown host port) + (portlogger:open-run-close portlogger:set-port port "released"))) + (debug:print-info 0 *default-log-port* "Shutdown activities completed in "(- (current-seconds) start-time)" seconds")) ;; (if *dbstruct-db* (db:close-all *dbstruct-db*)) ;; one second allocated #;(if (bdat-task-db *bdat*) ;; TODO: Check that this is correct for task db (let ((db (cdr (bdat-task-db *bdat*)))) (if (sqlite3:database? db) @@ -1635,15 +1622,14 @@ ;; NOTE: This is NOT called directly from clients as not all transports support a client running ;; in the same process as the server. ;; ;; conn is a conndat record ;; -(define (server:ping conn #!key (do-exit #f)) - (let* ((req (conndat-socket conn)) - (srvkey (conndat-srvkey conn)) +#;(define (server:ping uconn #!key (do-exit #f)) + (let* ((srvkey (conndat-srvkey uconn)) (msg (sexpr->string '(ping ,srvkey)))) - (send-receive-nn req msg))) ;; (server-ready? host port server-id)) + (send-receive uconn 'ping msg))) ;; (server-ready? host port server-id)) ;;====================================================================== ;; http-transportmod.scm contents moved here ;;====================================================================== @@ -1656,118 +1642,59 @@ ;; S E R V E R ;; ====================================================================== (define (http-get-function fnkey) (hash-table-ref/default *http-functions* fnkey (lambda () "nothing here yet"))) + +(define *rmt:run-mutex* (make-mutex)) +(define *rmt:run-flag* #f) ;; Main entry point to start a server. was start-server (define (rmt:run hostn) - ;; ;; Configurations for server - ;; (tcp-buffer-size 2048) - ;; (max-connections 2048) - (debug:print 2 *default-log-port* "PID: "(current-process-id)". Attempting to start the server ...") - (let* ((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))) ".") - (server:get-best-guess-address hostname) - #f))) - (if ipstr ipstr hostn))) ;; hostname))) - (port (portlogger:open-run-close portlogger:find-port)) - ;; (link-tree-path (common:get-linktree)) - ;; (tmp-area (common:get-db-tmp-area)) - #;(start-file (conc tmp-area "/.server-start"))) - (debug:print-info 0 *default-log-port* "portlogger recommended port: " port) - (if *server-info* - (begin - (servdat-host-set! *server-info* ipaddrstr) - (servdat-port-set! *server-info* port) - (servdat-status-set! *server-info* 'trying-port) - (servdat-trynum-set! *server-info* (+ (servdat-trynum *server-info*) 1))) - (set! *server-info* (make-servdat host: ipaddrstr port: port))) - (let* ((rep (rmt:try-start-server ipaddrstr port))) - (let loop ((instr (nng-recv rep))) - (let* ((data (string->sexpr instr)) - (res (case data - ((quit) 'quit) - (else (api:process-request *dbstruct-db* data)))) - (resdat (sexpr->string res))) - (if (not (eq? res 'quit)) - (begin - (set! *db-last-access* (current-seconds)) - (nng-send rep resdat) - (loop (nng-recv rep))))))) - (debug:print-info 0 *default-log-port* "After server, should never see this") - ;; server exit stuff here - (let* ((portnum (servdat-port *server-info*)) - (host (servdat-host *server-info*))) - (portlogger:open-run-close portlogger:set-port portnum "released") - (if (not (equal? (get-host-name) host)) - (debug:print-info 0 *default-log-port* "Server shutdown called for host "host", but we are on "(get-host-name)) - (rmt:server-shutdown host portnum)) - ;; (bdat-time-to-exit-set! *bdat* #t) ;; tell on-exit to be fast as we've already cleaned up - (portlogger:open-run-close portlogger:set-port port "released") ;; done in rmt:run - ;; (debug:print-info 0 *default-log-port* "Max cached queries was " *max-cache-size*) - ;; (debug:print-info 0 *default-log-port* "Number of cached writes " *number-of-writes*) - ;; (debug:print-info 0 *default-log-port* "Average cached write time " - ;; (if (eq? *number-of-writes* 0) - ;; "n/a (no writes)" - ;; (/ *writes-total-delay* - ;; *number-of-writes*)) - ;; " ms") - ;; (debug:print-info 0 *default-log-port* "Number non-cached queries " *number-non-write-queries*) - ;; (debug:print-info 0 *default-log-port* "Average non-cached time " - ;; (if (eq? *number-non-write-queries* 0) - ;; "n/a (no queries)" - ;; (/ *total-non-write-delay* - ;; *number-non-write-queries*)) - ;; " ms") - - (db:print-current-query-stats) - (debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting") - ))) - -(define (rmt:try-start-server ipaddrstr portnum) - (if *server-info* ;; update the server info as we might be trying next port - (begin - (servdat-host-set! *server-info* ipaddrstr) - (servdat-port-set! *server-info* portnum) - (servdat-status-set! *server-info* 'trying-port) - (servdat-trynum-set! *server-info* - (+ (servdat-trynum *server-info*) 1))) - (set! *server-info* (make-servdat host: ipaddrstr port: portnum))) - (debug:print-info 0 *default-log-port* "rmt:try-start-server time=" - (seconds->time-string (current-seconds)) - " ipaddrsstr=" ipaddrstr - " portnum=" portnum) - (assert (servdat? *server-info*) "FATAL: Must always have *server-info* properly set up by here.") - (servdat-status-set! *server-info* 'starting) - (servdat-port-set! *server-info* portnum) - (if (not (servdat-rep *server-info*)) - (let ((rep (make-rep-socket))) - (servdat-rep-set! *server-info* rep) - (socket-set! rep 'nng/recvtimeo 2000))) - (let* ((rep (servdat-rep *server-info*))) - (debug:print 0 *default-log-port* "INFO: Trying to start server on " ipaddrstr ":" portnum) - (handle-exceptions - exn - (begin - (print-error-message exn) - (if (< portnum 64000) - (begin - (debug:print 0 *default-log-port* "WARNING: attempt to start server failed. Trying again ...") - (debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn)) - (debug:print 5 *default-log-port* "exn=" (condition->list exn)) - (portlogger:open-run-close portlogger:set-failed portnum) - (debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port") - ;; (thread-sleep! 0.1) - (rmt:try-start-server ipaddrstr - (portlogger:open-run-close portlogger:find-port))) - (begin - (print "ERROR: Tried and tried but could not start the server, stopping at port "portnum)))) - (nng-listen rep (conc "tcp://*:" portnum)) - rep))) + (mutex-lock! *rmt:run-mutex*) + (if *rmt:run-flag* + (begin + (debug:print-warn 0 *default-log-port* "rmt:run already running.") + (mutex-unlock! *rmt:run-mutex*)) + (begin + (set! *rmt:run-flag* #t) + (mutex-unlock! *rmt:run-mutex*) + ;; ;; Configurations for server + ;; (tcp-buffer-size 2048) + ;; (max-connections 2048) + (debug:print 2 *default-log-port* "PID: "(current-process-id)". Attempting to start the server ...") + (if (and *db-serv-info* + (servdat-uconn *db-serv-info*)) + (let* ((uconn (servdat-uconn *db-serv-info*))) + (wait-and-close uconn)) + (let* ((port (portlogger:open-run-close portlogger:find-port)) + (handler-proc (lambda (rem-host-port qrykey cmd params) ;; + (set! *db-last-access* (current-seconds)) + (assert (list? params) "FATAL: handler called with non-list params") + (assert (args:get-arg "-server") "FATAL: handler called on non-server side. cmd="cmd", params="params) + (debug:print 0 *default-log-port* "handler call: "cmd", params="params) + (api:execute-requests *dbstruct-db* cmd params)))) + ;; (api:process-request *dbstuct-db* + (if (not *db-serv-info*) + (set! *db-serv-info* (make-servdat host: hostn port: port))) + (let* ((uconn (run-listener handler-proc port)) + (rport (udat-port uconn))) ;; the real port + (servdat-host-set! *db-serv-info* hostn) + (servdat-port-set! *db-serv-info* rport) + (servdat-uconn-set! *db-serv-info* uconn) + (wait-and-close uconn) + (db:print-current-query-stats) + ))) + (let* ((host (servdat-host *db-serv-info*)) + (port (servdat-port *db-serv-info*)) + (mode (or (servdat-mode *db-serv-info*) + "non-db"))) + ;; server exit stuff here + ;; (rmt:server-shutdown host port) - always do in on-exit + ;; (portlogger:open-run-close portlogger:set-port port "released") ;; moved to on-exit + (debug:print-info 0 *default-log-port* "Server "host":"port" mode "mode"shutdown complete. Exiting") + )))) ;;====================================================================== ;; S E R V E R U T I L I T I E S ;;====================================================================== @@ -1824,17 +1751,18 @@ ;; (define (get-lock-db sdat dbfile host port) (assert host "FATAL: get-lock-db called with host not set.") (assert port "FATAL: get-lock-db called with port not set.") (let* ((dbh (db:open-run-db dbfile db:initialize-db)) ;; open-run-db creates a standard db with schema used by all situations - (res (db:get-iam-server-lock dbh dbfile host port))) + (res (db:get-iam-server-lock dbh dbfile host port)) + (uconn (servdat-uconn sdat))) ;; res => list then already locked, check server is responsive ;; => #t then sucessfully got the lock ;; => #f reserved for future use as to indicate something went wrong (match res ((owner_pid owner_host owner_port event_time) - (if (server-ready? owner_host owner_port "abc") + (if (server-ready? uconn (conc owner_host":"owner_port) "abc") #f ;; locked by someone else (begin ;; locked by someone dead and gone (debug:print 0 *default-log-port* "WARNING: stale lock - have to steal it. This may fail.") (db:steal-lock-db dbh dbfile port)))) (#t #t) ;; placeholder so that we don't touch res if it is #t @@ -1864,11 +1792,14 @@ "ERROR: get-pkts-dir called without *toppath* set. Exiting.") (let* ((pdir (conc effective-toppath "/.meta/srvpkts"))) (if (file-exists? pdir) pdir (begin - (create-directory pdir #t) + (handle-exceptions ;; this exception handler should NOT be needed but ... + exn + pdir + (create-directory pdir #t)) pdir))))) ;; given a pkts dir read ;; (define (get-all-server-pkts pktsdir-in pktspec) @@ -1884,18 +1815,20 @@ (define (server-address srv-pkt) (conc (alist-ref 'host srv-pkt) ":" (alist-ref 'port srv-pkt))) -(define (server-ready? host port key) ;; server-address is host:port - (let* ((data (sexpr->string `((cmd . ping) - (key . ,key) - (params . ())))) - (res (open-send-receive-nn (conc host ":" port) data))) - (if res - (string->sexpr res) - res))) +(define (server-ready? uconn host-port key) ;; server-address is host:port + (let* ((params `((cmd . ping)(key . ,key))) + (data `((cmd . ping) + (key . ,key) + (params . ,params))) ;; I don't get it. + (res (send-receive uconn host-port 'ping data))) + (if (eq? res 'ack) ;; yep, likely it is who we want on the other end + res + #f))) +;; (begin (debug:print-info 0 *default-log-port* "server-ready? => "res) #f)))) ; from the pkts return servers associated with dbpath ;; NOTE: Only one can be alive - have to check on each ;; in the list of pkts returned ;; @@ -1908,17 +1841,18 @@ (loop (cdr tail) (if (equal? dbpath (alist-ref 'dbpath spkt)) (cons spkt res) res)))))) -(define (remove-pkts-if-not-alive serv-pkts) +(define (remove-pkts-if-not-alive uconn serv-pkts) (filter (lambda (pkt) (let* ((host (alist-ref 'host pkt)) (port (alist-ref 'port pkt)) + (host-port (conc host":"port)) (key (alist-ref 'servkey pkt)) (pktz (alist-ref 'Z pkt)) - (res (server-ready? host port key))) + (res (server-ready? uconn host-port key))) (if res res (let* ((pktsdir (get-pkts-dir *toppath*)) (pktpath (conc pktsdir"/"pktz".pkt"))) (debug:print 0 *default-log-port* "WARNING: pkt with no server "pktpath) @@ -1926,21 +1860,22 @@ #f)))) serv-pkts)) ;; from viable servers get one that is alive and ready ;; -(define (get-the-server apath serv-pkts) +(define (get-the-server uconn apath serv-pkts) (let loop ((tail serv-pkts)) (if (null? tail) #f (let* ((spkt (car tail)) (host (alist-ref 'ipaddr spkt)) (port (alist-ref 'port spkt)) + (host-port (conc host":"port)) (dbpth (alist-ref 'dbpath spkt)) (srvkey (alist-ref 'Z spkt)) ;; (alist-ref 'srvkey spkt)) (addr (server-address spkt))) - (if (server-ready? host port srvkey) + (if (server-ready? uconn host-port srvkey) spkt (loop (cdr tail))))))) ;; am I the "first" in line server? I.e. my D card is smallest ;; use Z card as tie breaker @@ -1977,24 +1912,24 @@ ;;====================================================================== ;; if .db/main.db check the pkts ;; (define (rmt:wait-for-server pkts-dir db-file server-key) - (let* ((sdat *server-info*)) + (let* ((sdat *db-serv-info*)) (let loop ((start-time (current-seconds)) (changed #t) (last-sdat "not this")) (begin ;; let ((sdat #f)) (thread-sleep! 0.01) (debug:print-info 0 *default-log-port* "Waiting for server alive signature") (mutex-lock! *heartbeat-mutex*) - (set! sdat *server-info*) + (set! sdat *db-serv-info*) (mutex-unlock! *heartbeat-mutex*) (if (and sdat (not changed) (> (- (current-seconds) start-time) 2)) - (begin + (let* ((uconn (servdat-uconn sdat))) (servdat-status-set! sdat 'iface-stable) (debug:print-info 0 *default-log-port* "Received server alive signature, now attempting to lock in server") ;; create a server pkt in *toppath*/.meta/srvpkts ;; TODO: @@ -2010,17 +1945,17 @@ (servdat-host sdat) db-file)) ;; (set! *my-signature* (servdat-uuid sdat)) ;; replace with Z, no, stick with proper key ;; now read pkts and see if we are a contender (let* ((all-pkts (get-all-server-pkts pkts-dir *srvpktspec*)) (viables (get-viable-servers all-pkts db-file)) - (alive (remove-pkts-if-not-alive viables)) + (alive (remove-pkts-if-not-alive uconn viables)) (best-srv (get-best-candidate alive db-file)) (best-srv-key (if best-srv (alist-ref 'servkey best-srv) #f)) (i-am-srv (equal? best-srv-key server-key)) (delete-pkt (lambda () (let* ((pktfile (conc (get-pkts-dir *toppath*) - "/" (servdat-uuid *server-info*) + "/" (servdat-uuid *db-serv-info*) ".pkt"))) (debug:print-info 0 *default-log-port* "Attempting to remove bogus pkt file "pktfile) (delete-file* pktfile))))) ;; remove immediately instead of waiting for on-exit (debug:print 0 *default-log-port* "best-srv-key: "best-srv-key", server-key: "server-key", i-am-srv: "i-am-srv) ;; am I the best-srv, compare server-keys to know @@ -2053,37 +1988,36 @@ (exit)) (loop start-time (equal? sdat last-sdat) sdat)))))))) -(define (rmt:register-server remdat apath iface port server-key dbname) - (remotedat-conns remdat) ;; just checking types - (rmt:open-main-connection remdat apath) ;; we need a channel to main.db - (rmt:send-receive-real remdat apath ;; params: host port servkey pid ipaddr dbpath +(define (rmt:register-server sinfo apath iface port server-key dbname) + (servdat-conns sinfo) ;; just checking types + (rmt:open-main-connection sinfo apath) ;; we need a channel to main.db + (rmt:send-receive-real sinfo apath ;; params: host port servkey pid ipaddr dbpath (db:run-id->dbname #f) 'register-server `(,iface ,port ,server-key ,(current-process-id) ,iface ,apath ,dbname))) -(define (rmt:get-count-servers remdat apath) - (remotedat-conns remdat) ;; just checking types - (rmt:open-main-connection remdat apath) ;; we need a channel to main.db - (rmt:send-receive-real remdat apath ;; params: host port servkey pid ipaddr dbpath +(define (rmt:get-count-servers sinfo apath) + (servdat-conns sinfo) ;; just checking types + (rmt:open-main-connection sinfo apath) ;; we need a channel to main.db + (rmt:send-receive-real sinfo apath ;; params: host port servkey pid ipaddr dbpath (db:run-id->dbname #f) 'get-count-servers `(,apath))) (define (rmt:get-servers-info apath) (rmt:send-receive 'get-servers-info #f `(,apath))) -(define (rmt:deregister-server remdat apath iface port server-key dbname) - (remotedat-conns remdat) ;; just checking types - (rmt:open-main-connection remdat apath) ;; we need a channel to main.db - (rmt:send-receive-real remdat apath ;; params: host port servkey pid ipaddr dbpath +(define (rmt:deregister-server db-serv-info apath iface port server-key dbname) + (rmt:open-main-connection db-serv-info apath) ;; we need a channel to main.db + (rmt:send-receive-real db-serv-info apath ;; params: host port servkey pid ipaddr dbpath (db:run-id->dbname #f) 'deregister-server `(,iface ,port ,server-key ,(current-process-id) @@ -2090,23 +2024,23 @@ ,iface ,apath ,dbname))) (define (rmt:wait-for-stable-interface #!optional (num-tries-allowed 100)) - ;; wait until *server-info* stops changing + ;; wait until *db-serv-info* stops changing (let* ((stime (current-seconds))) (let loop ((last-host #f) (last-port #f) (tries 0)) - (let* ((curr-host (and *server-info* (servdat-host *server-info*))) - (curr-port (and *server-info* (servdat-port *server-info*)))) - ;; first we verify port and interface, update *server-info* in need be. + (let* ((curr-host (and *db-serv-info* (servdat-host *db-serv-info*))) + (curr-port (and *db-serv-info* (servdat-port *db-serv-info*)))) + ;; first we verify port and interface, update *db-serv-info* in need be. (cond ((> tries num-tries-allowed) (debug:print 0 *default-log-port* "rmt:keep-running, giving up after trying for several minutes.") (exit 1)) - ((not *server-info*) + ((not *db-serv-info*) (thread-sleep! 0.25) (loop curr-host curr-port (+ tries 1))) ((or (not last-host)(not last-port)) (debug:print 0 *default-log-port* "rmt:keep-running, still no interface, tries="tries) (thread-sleep! 0.25) @@ -2119,16 +2053,16 @@ ((< (- (current-seconds) stime) 1) ;; keep up the looping until at least 3 seconds have passed (thread-sleep! 0.5) (loop curr-host curr-port (+ tries 1))) (else (rmt:get-signature) ;; sets *my-signature* as side effect - (servdat-status-set! *server-info* 'interface-stable) + (servdat-status-set! *db-serv-info* 'interface-stable) (debug:print 0 *default-log-port* "SERVER STARTED: " curr-host ":" curr-port " AT " (current-seconds) " server signature: " *my-signature* - " with "(servdat-trynum *server-info*)" port changes") + " with "(servdat-trynum *db-serv-info*)" port changes") (flush-output *default-log-port*) #t)))))) ;; run rmt:keep-running in a parallel thread to monitor that the db is being ;; used and to shutdown after sometime if it is not. @@ -2137,70 +2071,70 @@ ;; if none running or if > 20 seconds since ;; server last used then start shutdown ;; This thread waits for the server to come alive (debug:print-info 0 *default-log-port* "Starting the sync-back, keep alive thread in server") - (let* ((remdat *remotedat*) + (let* ((sinfo *db-serv-info*) (server-start-time (current-seconds)) (pkts-dir (get-pkts-dir)) (server-key (rmt:get-signature)) ;; This servers key (is-main (equal? (args:get-arg "-db") ".db/main.db")) (last-access 0) (server-timeout (server:expiration-timeout)) (shutdown-server-sequence (lambda (host port) - (set! *unclean-shutdown* #f) + (set! *unclean-shutdown* #f) ;; Should not be needed anymore (debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id)) - (rmt:server-shutdown host port) - (portlogger:open-run-close portlogger:set-port port "released") + ;; (rmt:server-shutdown host port) -- called in on-exit + ;; (portlogger:open-run-close portlogger:set-port port "released") called in on-exit (exit))) (timed-out? (lambda () (<= (+ last-access server-timeout) (current-seconds))))) - (servdat-dbfile-set! *server-info* (args:get-arg "-db")) + (servdat-dbfile-set! *db-serv-info* (args:get-arg "-db")) ;; main and run db servers have both got wait logic (could/should merge it) (if is-main (rmt:wait-for-server pkts-dir dbname server-key) (rmt:wait-for-stable-interface)) ;; this is our forever loop - (let* ((iface (servdat-host *server-info*)) - (port (servdat-port *server-info*))) + (let* ((iface (servdat-host *db-serv-info*)) + (port (servdat-port *db-serv-info*)) + (uconn (servdat-uconn *db-serv-info*))) (let loop ((count 0) (bad-sync-count 0) (start-time (current-milliseconds))) (if (and (not is-main) (common:low-noise-print 60 "servdat-status")) - (debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *server-info*))) + (debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *db-serv-info*))) - ;; set up the database handle (mutex-lock! *heartbeat-mutex*) + ;; set up the database handle (if (not *dbstruct-db*) ;; no db opened yet, open the db and register with main if appropriate - (let ((watchdog (bdat-watchdog *bdat*))) - (debug:print 0 *default-log-port* "SERVER: dbprep") - (db:setup dbname) ;; sets *dbstruct-db* as side effect - (servdat-status-set! *server-info* 'db-opened) - ;; IFF I'm not main, call into main and register self - (if (not is-main) - (let ((res (rmt:register-server remdat - *toppath* iface port - server-key dbname))) - (if res ;; we are the server - (servdat-status-set! *server-info* 'have-interface-and-db) - ;; now check that the db locker is alive, clear it out if not - (let* ((serv-info (rmt:get-server-info *toppath* dbname))) - (match serv-info - ((host port servkey pid ipaddr apath dbpath) - (if (not (server-ready? host port servkey)) - (begin - (debug:print-info 0 *default-log-port* "Server registered but not alive. Removing and trying again.") - (rmt:deregister-server remdat apath host port servkey dbpath) ;; servkey pid ipaddr apath dbpath) - - (loop (+ count 1) bad-sync-count start-time)))) - (else - (debug:print 0 *default-log-port* "We are not the server for "dbname", exiting. Server info is: "serv-info) - (exit))))))) - (debug:print 0 *default-log-port* - "SERVER: running, db "dbname" opened, megatest version: " + (let ((watchdog (bdat-watchdog *bdat*))) + (debug:print 0 *default-log-port* "SERVER: dbprep") + (db:setup dbname) ;; sets *dbstruct-db* as side effect + (servdat-status-set! *db-serv-info* 'db-opened) + ;; IFF I'm not main, call into main and register self + (if (not is-main) + (let ((res (rmt:register-server sinfo + *toppath* iface port + server-key dbname))) + (if res ;; we are the server + (servdat-status-set! *db-serv-info* 'have-interface-and-db) + ;; now check that the db locker is alive, clear it out if not + (let* ((serv-info (rmt:server-info *toppath* dbname))) + (match serv-info + ((host port servkey pid ipaddr apath dbpath) + (if (not (server-ready? uconn (conc host":"port) servkey)) + (begin + (debug:print-info 0 *default-log-port* "Server registered but not alive. Removing and trying again.") + (rmt:deregister-server sinfo apath host port servkey dbpath) ;; servkey pid ipaddr apath dbpath) + (loop (+ count 1) bad-sync-count start-time)))) + (else + (debug:print 0 *default-log-port* "We are not the server for "dbname", exiting. Server info is: "serv-info) + (exit))))))) + (debug:print 0 *default-log-port* + "SERVER: running, db "dbname" opened, megatest version: " (common:get-full-version)) ;; start the watchdog ;; is this really needed? @@ -2213,11 +2147,11 @@ (thread-start! watchdog)) (debug:print-info 0 *default-log-port* "Not starting watchdog thread (in state "(thread-state watchdog)")")) (debug:print 0 *default-log-port* "ERROR: *watchdog* not setup, cannot start it.")) #;(loop (+ count 1) bad-sync-count start-time) )) - + (debug:print-info 0 *default-log-port* "syncing "*toppath*" "dbname" at "(current-seconds)) (db:sync-inmem->disk *dbstruct-db* *toppath* dbname force-sync: #t) (mutex-unlock! *heartbeat-mutex*) @@ -2227,17 +2161,17 @@ (let* ((sync-time (- (current-milliseconds) start-time)) (rem-time (quotient (- 4000 sync-time) 1000))) (if (and (<= rem-time 4) (> rem-time 0)) (thread-sleep! rem-time))) + + ;; Transfer *db-last-access* to last-access to use in checking that we are still alive + (set! last-access *db-last-access*) (if (< count 1) ;; 3x3 = 9 secs aprox (loop (+ count 1) bad-sync-count (current-milliseconds))) - ;; Transfer *db-last-access* to last-access to use in checking that we are still alive - (set! last-access *db-last-access*) - (if (common:low-noise-print 60 "dbstats") (begin (debug:print 0 *default-log-port* "Server stats:") (db:print-current-query-stats))) (let* ((hrs-since-start (/ (- (current-seconds) server-start-time) 3600))) @@ -2249,11 +2183,11 @@ (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access)) (shutdown-server-sequence (get-host-name) port)) ((and *server-run* (or (not (timed-out?)) (if is-main ;; do not exit if there are other servers (keep main open until all others gone) - (> (rmt:get-count-servers remdat *toppath*) 1) + (> (rmt:get-count-servers sinfo *toppath*) 1) #f))) (if (common:low-noise-print 120 "server continuing") (debug:print-info 0 *default-log-port* "Server continuing, seconds since last db access: " (- (current-seconds) last-access))) (loop 0 bad-sync-count (current-milliseconds))) (else @@ -2260,12 +2194,17 @@ (set! *unclean-shutdown* #f) (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access)) (shutdown-server-sequence (get-host-name) port) #;(debug:print-info 0 *default-log-port* "Sending 'quit to server, received: " (open-send-receive-nn (conc iface":"port) ;; do this here and not in server-shutdown - (sexpr->string 'quit))) - ))))))) + (sexpr->string 'quit)))))))))) + +(define (rmt:get-reasonable-hostname) + (let* ((inhost (or (args:get-arg "-server") "-"))) + (if (equal? inhost "-") + (get-host-name) + inhost))) ;; Call this to start the actual server ;; ;; all routes though here end in exit ... ;; @@ -2273,28 +2212,24 @@ ;; (define (rmt:server-launch dbname) (debug:print-info 0 *default-log-port* "Entered rmt:server-launch") (let* ((th2 (make-thread (lambda () (debug:print-info 0 *default-log-port* "Server run thread started") - (rmt:run (if (args:get-arg "-server") - (args:get-arg "-server") - "-") - )) "Server run")) + (rmt:run (rmt:get-reasonable-hostname))) + "Server run")) (th3 (make-thread (lambda () (debug:print-info 0 *default-log-port* "Server monitor thread started") - (rmt:keep-running dbname) - "Keep running")))) + (if (args:get-arg "-server") + (rmt:keep-running dbname))) + "Keep running"))) (thread-start! th2) (thread-sleep! 0.252) ;; give the server time to settle before starting the keep-running monitor. (thread-start! th3) (set! *didsomething* #t) (thread-join! th2) - (thread-join! th3) - ;; (exit)) - ) - #f - ) + (thread-join! th3)) + #f) ;; Generate a unique signature for this process, used at both client and ;; server side (define (rmt:mk-signature) (message-digest-string (md5-primitive) @@ -2326,36 +2261,22 @@ ;(print "Output: " inl) (set! ret #t)) (loop (read-line inp))))))) ret)) -;;start a server, returns the connection -;; -(define (start-nn-server portnum ) - (let ((rep (make-rep-socket))) ;; (nn-socket 'rep))) - (socket-set! rep 'nng/recvtimeo 2000) - (handle-exceptions ;; why have exception handler here? - exn - (let ((emsg ((condition-property-accessor 'exn 'message) exn))) - (print "ERROR: Failed to start server \"" emsg "\"") - (exit 1)) - - (nng-dial #;nn-bind rep (conc "tcp://*:" portnum))) - rep)) - -(define (open-nn-connection host-port) +#;(define (open-nn-connection host-port) (let ((req (make-req-socket)) (uri (conc "tcp://" host-port))) (nng-dial req uri) (socket-set! req 'nng/recvtimeo 2000) req)) -(define (send-receive-nn req msg) +#;(define (send-receive-nn req msg) (nng-send req msg) (nng-recv req)) -(define (close-nn-connection req) +#;(define (close-nn-connection req) (nng-close! req)) ;; ;; open connection to server, send message, close connection ;; ;; ;; (define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds @@ -2390,11 +2311,11 @@ ;; (thread-start! th1) ;; (thread-start! th2) ;; (thread-join! th1) ;; res)))) ;; -(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds +#;(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds (let ((req (make-req-socket)) (uri (conc "tcp://" host-port)) (res #f)) (handle-exceptions exn Index: runsmod.scm ================================================================== --- runsmod.scm +++ runsmod.scm @@ -499,11 +499,11 @@ ;; run the run prehook if there are no tests yet run for this run: ;; (runs:run-pre-hook run-id) ;; mark all test launched flag as false in the meta table - (rmt:set-var run-id (conc "lunch-complete-" run-id) "no") + (rmt:set-var run-id (conc "launch-complete-" run-id) "no") (debug:print-info 1 *default-log-port* "Setting end-of-run to no") (let* ((config-reruns (let ((x (configf:lookup *configdat* "setup" "reruns"))) (if x (string->number x) #f))) (config-rerun-cnt (if config-reruns config-reruns @@ -1709,11 +1709,11 @@ (else (debug:print-info 4 *default-log-port* "cond branch - " "rtq-9") (debug:print-info 4 *default-log-port* "Exiting loop with...\n hed=" hed "\n tal=" tal "\n reruns=" reruns)) ))) ;; end loop on sorted test names ;; this is the point where everything is launched and now you can mark the run in metadata table as all launched - (rmt:set-var run-id (conc "lunch-complete-" run-id) "yes") + (rmt:set-var run-id (conc "launch-complete-" run-id) "yes") ;; now *if* -run-wait we wait for all tests to be done ;; Now wait for any RUNNING tests to complete (if in run-wait mode) ;; (if (runs:dat-load-mgmt-function runsdat)((runs:dat-load-mgmt-function runsdat))) (thread-sleep! 10) ;; I think there is a race condition here. Let states/statuses settle @@ -2556,11 +2556,12 @@ ;; Extract out stuff needed in most or many calls ;; here then call proc (let* ((keyvals (keys:target->keyval keys target))) (proc target runname keys keyvals))) ;; (if db (sqlite3:finalize! db)) - (set! *didsomething* #t)))))) + (set! *didsomething* #t) + ))))) ;;====================================================================== ;; Lock/unlock runs ;;====================================================================== ADDED rununit.sh Index: rununit.sh ================================================================== --- /dev/null +++ rununit.sh @@ -0,0 +1,5 @@ +#!/bin/bash + +rm tests/*log tests/simplerun/logs/* + +script -c 'ck5 make unit' Index: tests/simplerun/Makefile ================================================================== --- tests/simplerun/Makefile +++ tests/simplerun/Makefile @@ -1,3 +1,5 @@ cleanup : - killall mtest -v -9;rm -rf .meta .db + killall mtest dboard -v -9 || true + rm -rf *.log *.bak NB* logs/* .meta .db ../simpleruns/* lt + ADDED tests/simplerun/debug.scm Index: tests/simplerun/debug.scm ================================================================== --- /dev/null +++ tests/simplerun/debug.scm @@ -0,0 +1,58 @@ +(import big-chicken trace rmtmod apimod dbmod ulex srfi-18) +(trace-call-sites #t) +(trace + ;; db:get-tests-for-run + ;; rmt:general-open-connection + ;; rmt:open-main-connection + ;; rmt:drop-conn + ;; rmt:send-receive + ;; rmt:log-to-main + ) + + +(module junk + * + + (import big-chicken rmtmod apimod dbmod srfi-18) + +(define (make-run-id) + (let* ((s (conc (current-process-id))) + (l (string-length s))) + (string->number (substring s (- l 3) l)) + )) + +(define (run) + (let* ((th1 (make-thread + (lambda () + (let loop ((r 0) + (i 1) + (s 0)) ;; sum + (let ((start-time (current-milliseconds)) + (run-id (+ r (make-run-id)))) + (rmt:register-test run-id "test1" (conc "item_" i)) + (let* ((qry-time (- (current-milliseconds) start-time)) + (tot-query-time (+ qry-time s)) + (avg-query-time (* 1.0 (/ tot-query-time i)))) + (if (> qry-time 500) + (print "WARNING: rmt:register-test took more than 500ms, "qry-time"ms, i="i", avg-query-time="avg-query-time)) + (if (eq? (modulo i 100) 0) + (print "For run-id="run-id", "(rmt:get-keys-write)" num tests registered="i)) + (if (< i 500) + (loop r (+ i 1) tot-query-time) + (if (< r 100) + (let* ((start-time (current-milliseconds))) + (print "rmt:get-keys "(rmt:get-keys)" in "(- (current-milliseconds) start-time)) + (print "Got "(length (rmt:get-tests-for-run run-id "%" '() '() 0 #f #f #f #f #f 0 #f))" tests for run "run-id) + (print "Average query time: "avg-query-time) + (loop (+ r 1) 0 tot-query-time)))))))) + ))) + (thread-start! th1) + (thread-join! th1))) + +) + +(import junk) +(run) + + + Index: tests/simplerun/megatest.config ================================================================== --- tests/simplerun/megatest.config +++ tests/simplerun/megatest.config @@ -21,10 +21,13 @@ [setup] # Adjust max_concurrent_jobs to limit how much you load your machines max_concurrent_jobs 50 +[server] +timeout 3600 + # Uncomment this to make the in-mem db into a disk based db (slower but good for debug) # be aware that some unit tests will fail with this due to persistent data # # tmpdb /tmp @@ -35,15 +38,15 @@ [validvalues] state start end completed # Job tools are more advanced ways to control how your jobs are launched [jobtools] -useshell yes +# useshell yes launcher nbfake # You can override environment variables for all your tests here [env-override] EXAMPLE_VAR example value # As you run more tests you may need to add additional disks, the names are arbitrary but must be unique [disks] disk0 #{getenv MT_RUN_AREA_HOME}/../simpleruns Index: tests/tests.scm ================================================================== --- tests/tests.scm +++ tests/tests.scm @@ -22,13 +22,17 @@ chicken.string chicken.process-context chicken.file chicken.pretty-print commonmod + ulex ) (define test-work-dir (current-directory)) + +(work-method 'mailbox) ;; threads, direct, mailbox +(return-method 'mailbox) ;; polling, mailbox, direct ;; given list of lists ;; ( ( msg expected param1 param2 ...) ;; ( ... ) ) ;; apply test to all @@ -49,12 +53,22 @@ ;; (for-each ;; (lambda (file) ;; (print "Loading " file) ;; (load file)) ;; files)) + +(define-syntax run-in-thread + (syntax-rules () + ((_ body ...) + (let ((th1 (make-thread (lambda () + body ...) + "the thread"))) + (thread-start! th1) + (thread-join! th1))))) + (let* ((unit-test-name (list-ref (argv) 4)) (fname (conc "../unittests/" unit-test-name ".scm"))) (if (file-exists? fname) (load fname) (print "ERROR: Unit test " unit-test-name " not found in unittests directory"))) Index: tests/unittests/all-rmt.scm ================================================================== --- tests/unittests/all-rmt.scm +++ tests/unittests/all-rmt.scm @@ -91,13 +91,15 @@ ;; (list (list "localhost" #t (get-host-name)) ;; (list "not-a-host" #t "not-a-host" )) ;; post-proc: pair?) ;; ;; (test #f #t (list? (rmt:get-changed-record-ids 0))) -;; -(test #f #f (begin (runs:update-all-test_meta #f) #f)) - +;; + +(run-in-thread + + ;; (test #f #f (begin (runs:update-all-test_meta #f) #f)) (test #f '("test1" "test2")(sort (alist-ref "tagtwo" (rmt:get-tests-tags) equal?) string<=?)) (test #f '() (rmt:get-key-val-pairs 0)) (test #f '("SYSTEM" "RELEASE") (rmt:get-keys)) (test #f '("SYSTEM" "RELEASE") (rmt:get-keys-write)) ;; dummy query to force server start (test #f '() (rmt:get-key-vals 1)) @@ -146,11 +148,11 @@ (test #f '()(rmt:get-prev-run-ids 1)) (test #f #t (begin (rmt:lock/unlock-run 1 #t #f "mikey") #t)) (test #f "JUSTFINE" (rmt:get-run-status 1)) (test #f #t (begin (rmt:set-run-status 1 "NOTFINE" msg: "A message") #t)) (test #f #t (begin (rmt:update-run-event_time 1) #t)) - +) ;; (rmt:get-runs-by-patt keys runnamepatt targpatt offset limit fields last-runs-update) ;; fields of #f uses default ;; (let ((keys (rmt:get-keys)) (rnp "%") ;; run name patt (tpt "%/%")) ;; target patt Index: tests/unittests/basicserver.scm ================================================================== --- tests/unittests/basicserver.scm +++ tests/unittests/basicserver.scm @@ -21,21 +21,23 @@ ;; Run like this: ;; ;; ./rununittest.sh server 1;(cd simplerun;megatest -stop-server 0) (import rmtmod trace http-client apimod dbmod - launchmod srfi-69) + launchmod srfi-69 ulex system-information) (trace-call-sites #t) (trace + ;; get-the-server ;; db:get-dbdat ;; rmt:find-main-server ;; rmt:send-receive-real ;; rmt:send-receive ;; sexpr->string ;; server-ready? ;; rmt:register-server + ;; api:run-server-process ;; rmt:open-main-connection ;; rmt:general-open-connection ;; rmt:get-conny ;; common:watchdog ;; rmt:find-main-server @@ -43,38 +45,43 @@ ;; get-viable-servers ;; get-best-candidate ;; api:run-server-process ;; rmt:run ;; rmt:try-start-server - ) - -(test #f #t (remotedat? (let ((r (make-remotedat))) - (set! *remotedat* r) - r))) -(test #f #f (rmt:get-conn *remotedat* *toppath* ".db/main.db")) -(test #f #f (rmt:find-main-server *toppath* ".db/main.db")) -(test #f #t (rmt:open-main-connection *remotedat* *toppath*)) -(pp (hash-table->alist (remotedat-conns *remotedat*))) -(test #f #t (conndat? (rmt:get-conn *remotedat* *toppath* ".db/main.db"))) - -(define *main* (rmt:get-conn *remotedat* *toppath* ".db/main.db")) - -;; (for-each (lambda (tdat) -;; (test #f tdat (loop-test (rmt:conn-ipaddr *main*) -;; (rmt:conn-port *main*) tdat))) -;; (list 'a -;; '(a "b" 123 1.23 ))) -(test #f #t (rmt:send-receive 'ping #f 'hello)) - -(define *db* (db:setup ".db/main.db")) - -;; these let me cut and paste from source easily -(define apath *toppath*) -(define dbname ".db/2.db") -(define remote *remotedat*) + ;; + ;; ulex + ;; + ;; wait-and-close + ;; run-listener + ) + + +(test #f #t (servdat? (let ((s (make-servdat))) + (set! *servdat* s) + s))) +(test #f #f (rmt:get-conn *servdat* *toppath* ".db/main.db")) +(test #f #f (rmt:find-main-server *servdat* *toppath* ".db/main.db")) +(define th1 (make-thread (lambda () + (rmt:run (get-host-name))) + "rmt:run thread")) +(thread-start! th1) +(thread-sleep! 0.5) ;; give things some time to get going +;; switch to *db-serv-info* instead of *servdat* +(define *uconn* (servdat-uconn *db-serv-info*)) +(print "*uconn*: " *uconn*) +(test #f #t (ulex-listener? (servdat-uconn *db-serv-info*))) +(test #f #t (string? (udat-host-port *uconn*))) + +(run-in-thread + (test #f 'ack (server-ready? *uconn* (udat-host-port *uconn*) (servdat-uuid *db-serv-info*)))) + +(test #f #t (rmt:open-main-connection *db-serv-info* *toppath*)) +;; (pp (hash-table->alist (remotedat-conns *db-serv-info*))) +(test #f #t (conndat? (rmt:get-conn *db-serv-info* *toppath* ".db/main.db"))) + +(define remote *db-serv-info*) (define keyvals '(("SYSTEM" "a")("RELEASE" "b"))) -(test #f '() (string->sexpr "()")) -(test #f 'server-started (api:execute-requests *db* 'get-server (list *toppath* ".db/2.db"))) -(set! *dbstruct-db* #f) +(run-in-thread + (test #f (map car keyvals) (rmt:get-keys))) (exit) Index: tests/unittests/server.scm ================================================================== --- tests/unittests/server.scm +++ tests/unittests/server.scm @@ -53,21 +53,21 @@ (define *db* (db:setup ".db/main.db")) ;; these let me cut and paste from source easily (define apath *toppath*) (define dbname ".db/2.db") -(define remote *remotedat*) +(define remote *db-serv-info*) (define keyvals '(("SYSTEM" "a")("RELEASE" "b"))) (test #f #t (rmt:open-main-connection remote apath)) -(test #f #t (conndat? (rmt:get-conn *remotedat* *toppath* ".db/main.db"))) -(test #f #t (conndat? (rmt:get-conn *remotedat* *toppath* ".db/main.db"))) -(test #f ".db/2.db" (list-ref (rmt:send-receive-real *remotedat* *toppath* ".db/main.db" 'get-server `(,apath ,dbname)) - 6)) +(test #f #t (conndat? (rmt:get-conn *db-serv-info* *toppath* ".db/main.db"))) +(test #f #t (conndat? (rmt:get-conn *db-serv-info* *toppath* ".db/main.db"))) +(test #f 'server-started (rmt:send-receive-real *db-serv-info* *toppath* ".db/main.db" + 'start-server `(,apath ,dbname))) (thread-sleep! 2) -(test #f #t (rmt:general-open-connection *remotedat* *toppath* ".db/2.db")) +(test #f #t (rmt:general-open-connection *db-serv-info* *toppath* ".db/2.db")) ;; (let loop ((end-time (+ (current-seconds) 61))) (test #f #t (list? (rmt:get-servers-info *toppath*))) (test #f '("SYSTEM" "RELEASE") (rmt:get-keys)) @@ -75,16 +75,16 @@ ;; (print "Got here.") (test #f 1 (rmt:send-receive 'register-run 1 (list keyvals "run1" "new" "n/a" "justme" #f))) (test #f 2 (rmt:register-run keyvals "run2" "new" "n/a" "justme" #f)) - ;; (test #f 2 (rmt:deregister-server *remotedat* *toppath* iface port server-key dbname + ;; (test #f 2 (rmt:deregister-server *db-serv-info* *toppath* iface port server-key dbname - (test #f 2 (rmt:get-count-servers *remotedat* *toppath*)) + (test #f 2 (rmt:get-count-servers *db-serv-info* *toppath*)) (test #f "run2" (rmt:get-run-name-from-id 2)) (test #f #f (rmt:send-receive 'get-test-info-by-id 2 '(2 1))) (test #f #t (rmt:general-call 'update-cpuload-diskfree 2 1.5 1e6 1)) ;; (if (< (current-seconds) end-time)(loop end-time))) (exit) ADDED ulex-simple/ulex.scm Index: ulex-simple/ulex.scm ================================================================== --- /dev/null +++ ulex-simple/ulex.scm @@ -0,0 +1,440 @@ +;; ulex: Distributed sqlite3 db +;;; +;; Copyright (C) 2018-2021 Matt Welland +;; Redistribution and use in source and binary forms, with or without +;; modification, is permitted. +;; +;; THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS +;; OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +;; WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +;; ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE +;; LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +;; CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT +;; OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR +;; BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +;; LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +;; (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE +;; USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH +;; DAMAGE. + +;;====================================================================== +;; ABOUT: +;; See README in the distribution at https://www.kiatoa.com/fossils/ulex +;; NOTES: +;; Why sql-de-lite and not say, dbi? - performance mostly, then simplicity. +;; +;;====================================================================== + +(module ulex + ( + + ;; NOTE: looking for the handler proc - find the run-listener :) + + run-listener ;; (run-listener handler-proc [port]) => uconn + + ;; NOTE: handler-proc params; + ;; (handler-proc rem-host-port qrykey cmd params) + + send-receive ;; (send-receive uconn host-port cmd data) + + ;; NOTE: cmd can be any plain text symbol except for these; + ;; 'ping 'ack 'goodbye 'response + + set-work-handler ;; (set-work-handler proc) + + wait-and-close ;; (wait-and-close uconn) + + ulex-listener? + + ;; needed to get the interface:port that was automatically found + udat-port + udat-host-port + + ;; for testing only + ;; pp-uconn + ) + +(import scheme + chicken.base + chicken.file + chicken.time + chicken.condition + chicken.string + chicken.sort + chicken.pretty-print + + address-info + mailbox + matchable + ;; queues + regex + regex-case + s11n + srfi-1 + srfi-18 + srfi-4 + srfi-69 + system-information + tcp6 + typed-records + ) + +;; udat struct, used by both caller and callee +;; instantiated as uconn by convention +;; +(defstruct udat + ;; the listener side + (port #f) + (host-port #f) + (socket #f) + ;; the peers + (peers (make-hash-table)) ;; host:port->peer + ;; work handling + (work-queue (make-mailbox)) + (work-proc #f) ;; set by user + (cnum 0) ;; cookie number + (mboxes (make-hash-table)) ;; for the replies + (avail-cmboxes '()) ;; list of ( . ) for re-use + ;; threads + (numthreads 50) + (cmd-thread #f) + (work-queue-thread #f) + ) + +;; ;; struct for keeping track of others we are talking to +;; ;; +;; (defstruct pdat +;; (host-port #f) +;; (conns '()) ;; list of pcon structs, pop one off when calling the peer +;; ) +;; +;; ;; struct for peer connections, keep track of expiration etc. +;; ;; +;; (defstruct pcon +;; (inp #f) +;; (oup #f) +;; (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59) +;; (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes +;; ) + +;;====================================================================== +;; listener +;;====================================================================== + +;; is uconn a ulex connector (listener) +;; +(define (ulex-listener? uconn) + (udat? uconn)) + +;; create a tcp listener and return a populated udat struct with +;; my port, address, hostname, pid etc. +;; return #f if fail to find a port to allocate. +;; +;; if udata-in is #f create the record +;; if there is already a serv-listener return the udata +;; +(define (setup-listener uconn #!optional (port 4242)) + (handle-exceptions + exn + (if (< port 65535) + (setup-listener uconn (+ port 1)) + #f) + (connect-listener uconn port))) + +(define (connect-listener uconn port) + ;; (tcp-listener-socket LISTENER)(socket-name so) + ;; sockaddr-address, sockaddr-port, sockaddr->string + (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]]) + (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname))) + (udat-port-set! uconn port) + (udat-host-port-set! uconn (conc addr":"port)) + (udat-socket-set! uconn tlsn) + uconn)) + +;; run-listener does all the work of starting a listener in a thread +;; it then returns control +;; +(define (run-listener handler-proc #!optional (port-suggestion 4242)) + (let* ((uconn (make-udat))) + (udat-work-proc-set! uconn handler-proc) + (if (setup-listener uconn port-suggestion) + (let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop")) + #;(th2 (make-thread (lambda ()(process-work-queue uconn)) "Ulex work queue processor"))) + (tcp-buffer-size 2048) + ;; (max-connections 2048) + (thread-start! th1) + #;(thread-start! th2) + (udat-cmd-thread-set! uconn th1) + #;(udat-work-queue-thread-set! uconn th2) + (print "cmd loop and process workers started") + uconn) + (assert #f "ERROR: run-listener called without proper setup.")))) + +(define (wait-and-close uconn) + (thread-join! (udat-cmd-thread uconn)) + (tcp-close (udat-socket uconn))) + +;;====================================================================== +;; peers and connections +;;====================================================================== + +(define *send-mutex* (make-mutex)) + +;; send structured data to recipient +;; +;; NOTE: qrykey is what was called the "cookie" previously +;; +;; retval tells send to expect and wait for return data (one line) and return it or time out +;; this is for ping where we don't want to necessarily have set up our own server yet. +;; +;; NOTE: see below for beginnings of code to allow re-use of tcp connections +;; - I believe (without substantial evidence) that re-using connections will +;; be beneficial ... +;; +(define (send udata host-port qrykey cmd params) + (mutex-lock! *send-mutex*) + (let* ((my-host-port (udat-host-port udata)) ;; remote will return to this + (isme #f #;(equal? host-port my-host-port)) ;; calling myself? + ;; dat is a self-contained work block that can be sent or handled locally + (dat (list my-host-port qrykey cmd params))) + (if isme + (ulex-handler udata dat) ;; no transmission needed + (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC? + exn + #f + (let-values (((inp oup)(tcp-connect host-port))) + (let ((res (if (and inp oup) + (begin + (serialize dat oup) + (deserialize inp)) ;; yes, we always want an ack + (begin + (print "ERROR: send called but no receiver has been setup. Please call setup first!") + #f)))) + (close-input-port inp) + (close-output-port oup) + (mutex-unlock! *send-mutex*) + res)))))) ;; res will always be 'ack + +;; send a request to the given host-port and register a mailbox in udata +;; wait for the mailbox data and return it +;; +(define (send-receive uconn host-port cmd data) + (cond + ((member cmd '(ping goodbye)) ;; these are immediate + (send uconn host-port 'ping cmd data)) + (else + (let* ((cmbox (get-cmbox uconn)) ;; would it be better to keep a stack of mboxes to reuse? + (qrykey (car cmbox)) + (mbox (cdr cmbox)) + (mbox-time (current-milliseconds)) + (sres (send uconn host-port qrykey cmd data))) ;; short res + sres)))) + +;;====================================================================== +;; responder side +;;====================================================================== + +;; take a request, rdat, and if not immediate put it in the work queue +;; +;; Reserved cmds; ack ping goodbye response +;; +(define (ulex-handler uconn rdat) + (assert (list? rdat) "FATAL: ulex-handler give rdat as not list") + (match rdat ;; (string-split controldat) + ((rem-host-port qrykey cmd params) + ;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params) + (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f))) + (case cmd + ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack) + ((ping) + ;; (print "Got Ping!") + ;; (add-to-work-queue uconn rdat) + 'ack) + (else + (do-work uconn rdat))))) + (else + (print "BAD DATA? controldat=" rdat) + 'ack) ;; send ack anyway? + )) + +;; given an already set up uconn start the cmd-loop +;; +(define (ulex-cmd-loop uconn) + (let* ((serv-listener (udat-socket uconn))) + (let loop ((state 'start)) + (let-values (((inp oup)(tcp-accept serv-listener))) + (let* ((rdat (deserialize inp)) ;; '(my-host-port qrykey cmd params) + (resp (ulex-handler uconn rdat))) + (if resp (serialize resp oup)) + (close-input-port inp) + (close-output-port oup)) + (loop state))))) +;;(define (ulex-cmd-loop uconn) +;; (let* ((serv-listener (udat-socket uconn)) +;; ;; (old-listener (lambda () +;; ;; (let loop ((state 'start)) +;; ;; (let-values (((inp oup)(tcp-accept serv-listener))) +;; ;; (let* ((rdat (deserialize inp)) ;; '(my-host-port qrykey cmd params) +;; ;; (resp (ulex-handler uconn rdat))) +;; ;; (if resp (serialize resp oup)) +;; ;; (close-input-port inp) +;; ;; (close-output-port oup)) +;; ;; (loop state))))) +;; (server (make-tcp-server +;; serv-listener +;; (lambda () +;; (let* ((rdat (deserialize )) ;; '(my-host-port qrykey cmd params) +;; (resp (ulex-handler uconn rdat))) +;; (if resp (serialize resp) resp)))))) +;; (server))) + +;; add a proc to the cmd list, these are done symetrically (i.e. in all instances) +;; so that the proc can be dereferenced remotely +;; +(define (set-work-handler uconn proc) + (udat-work-proc-set! uconn proc)) + +;;====================================================================== +;; work queues - this is all happening on the listener side +;;====================================================================== + +;; rdat is (rem-host-port qrykey cmd params) + +(define (add-to-work-queue uconn rdat) + #;(queue-add! (udat-work-queue uconn) rdat) + (mailbox-send! (udat-work-queue uconn) rdat)) + +(define (do-work uconn rdat) + (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change + ;; put this following into a do-work procedure + (match rdat + ((rem-host-port qrykey cmd params) + (let* ((start-time (current-milliseconds)) + (result (proc rem-host-port qrykey cmd params)) + (end-time (current-milliseconds)) + (run-time (- end-time start-time))) + result)) + (else + (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params") + #f)))) + +(define (process-work-queue uconn) + (let ((wqueue (udat-work-queue uconn)) + (proc (udat-work-proc uconn)) + (numthr (udat-numthreads uconn))) + (let loop ((thnum 1) + (threads '())) + (let ((thlst (cons (make-thread (lambda () + (let work-loop () + (let ((rdat (mailbox-receive! wqueue 24000 'MBOX_TIMEOUT))) + (do-work uconn rdat)) + (work-loop))) + (conc "work thread " thnum)) + threads))) + (if (< thnum numthr) + (loop (+ thnum 1) + thlst) + (begin + (print "ULEX: Starting "(length thlst)" worker threads.") + (map thread-start! thlst) + (print "ULEX: Threads started. Joining all.") + (map thread-join! thlst))))))) + +;; below was to enable re-use of connections. This seems non-trivial so for +;; now lets open on each call +;; +;; ;; given host-port get or create peer struct +;; ;; +;; (define (udat-get-peer uconn host-port) +;; (or (hash-table-ref/default (udat-peers uconn) host-port #f) +;; ;; no peer, so create pdat and init it +;; +;; ;; NEED stack of connections, pop and use; inp, oup, +;; ;; creation_time (remove and create new if over 24hrs old +;; ;; +;; (let ((pdat (make-pdat host-port: host-port))) +;; (hash-table-set! (udat-peers uconn) host-port pdat) +;; pdat))) +;; +;; ;; is pcon alive +;; +;; ;; given host-port and pdat get a pcon +;; ;; +;; (define (pdat-get-pcon pdat host-port) +;; (let loop ((conns (pdat-conns pdat))) +;; (if (null? conns) ;; none? make and return - do NOT add - it will be pushed back on list later +;; (init-pcon (make-pcon)) +;; (let* ((conn (pop conns))) +;; +;; ;; given host-port get a pcon struct +;; ;; +;; (define (udat-get-pcon + +;;====================================================================== +;; misc utils +;;====================================================================== + +(define (make-cookie uconn) + (let ((newcnum (+ (udat-cnum uconn) 1))) + (udat-cnum-set! uconn newcnum) + (conc (udat-host-port uconn) ":" + newcnum))) + +;; cookie/mboxes + +;; we store each mbox with a cookie ( . ) +;; +(define (get-cmbox uconn) + (if (null? (udat-avail-cmboxes uconn)) + (let ((cookie (make-cookie uconn)) + (mbox (make-mailbox))) + (hash-table-set! (udat-mboxes uconn) cookie mbox) + `(,cookie . ,mbox)) + (let ((cmbox (car (udat-avail-cmboxes uconn)))) + (udat-avail-cmboxes-set! uconn (cdr (udat-avail-cmboxes uconn))) + cmbox))) + +(define (put-cmbox uconn cmbox) + (udat-avail-cmboxes-set! uconn (cons cmbox (udat-avail-cmboxes uconn)))) + +(define (pp-uconn uconn) + (pp (udat->alist uconn))) + + +;;====================================================================== +;; network utilities +;;====================================================================== + +;; NOTE: Look at address-info egg as alternative to some of this + +(define (rate-ip ipaddr) + (regex-case ipaddr + ( "^127\\..*" _ 0 ) + ( "^(10\\.0|192\\.168)\\..*" _ 1 ) + ( else 2 ) )) + +;; Change this to bias for addresses with a reasonable broadcast value? +;; +(define (ip-pref-less? a b) + (> (rate-ip a) (rate-ip b))) + +(define (get-my-best-address) + (let ((all-my-addresses (get-all-ips))) + (cond + ((null? all-my-addresses) + (get-host-name)) ;; no interfaces? + ((eq? (length all-my-addresses) 1) + (car all-my-addresses)) ;; only one to choose from, just go with it + (else + (car (sort all-my-addresses ip-pref-less?)))))) + +(define (get-all-ips-sorted) + (sort (get-all-ips) ip-pref-less?)) + +(define (get-all-ips) + (map address-info-host + (filter (lambda (x) + (equal? (address-info-type x) "tcp")) + (address-infos (get-host-name))))) + +) ADDED ulex-trials/Makefile Index: ulex-trials/Makefile ================================================================== --- /dev/null +++ ulex-trials/Makefile @@ -0,0 +1,8 @@ +ulex-test : ulex-test.scm ../ulex/ulex.scm + csc ulex-test.scm + +test : ulex-test + for x in $$(seq 9);do export NBFAKE_LOG=NBFAKE_$$x;sleep 1;nbfake ./ulex-test run 828$$x;echo $$cmd;$$cmd;done + +clean : + rm -f ulex-test .runners/* NBFAKE* ADDED ulex-trials/server-one.scm Index: ulex-trials/server-one.scm ================================================================== --- /dev/null +++ ulex-trials/server-one.scm @@ -0,0 +1,61 @@ +(import tcp-server format (chicken tcp) (chicken io) (chicken string) (prefix sqlite3 sqlite3:) sql-de-lite srfi-18 simple-exceptions mailbox s11n) +(let* ((work-mailbox (make-mailbox)) + (notify-mailbox (make-mailbox)) +(th1 (make-thread (lambda () + +((make-tcp-server + (tcp-listen 6505) + (lambda () + (let* ((db (sqlite3:open-database "test.db")) + (rec-data (deserialize))) + (sqlite3:set-busy-handler! db (sqlite3:make-busy-timeout 136000)) + ;;(exec (sql db "INSERT INTO entries (received,send) VALUES (?,?);") "something" (conc "Server One Response: " "something else")) + (sqlite3:execute db "INSERT INTO entries (received,send) VALUES (?,?);" "something" (conc "Server One Response: " "something else")) + (mailbox-send! work-mailbox rec-data) + (format (current-error-port) (conc rec-data)) + (write-line (conc "Response to: " (conc rec-data))) + ;;(close-database db) + ))) +#t)) +"receive")) +(th2 (make-thread (lambda () + (print "Jeff is here") + (let loop ((entries 0)) + (thread-sleep! 0.01) + (print "Preparding to send entries" entries) + (handle-exceptions exn (begin (print "Had an issue: " (message exn))(thread-sleep! 10)) + (define-values (i o) (tcp-connect "localhost" 6504)) + (serialize (list "localhost:6505" "from-server-one") o) + (print (read-line i)) + (close-input-port i) + (close-output-port o)) + (loop (+ entries 1)))) "send")) +(th3 (make-thread (lambda () + (print "In mailbox thread") + (let loop2 ((entries2 0)) + (print "Processing: " (mailbox-receive! work-mailbox)) + (mailbox-send! notify-mailbox (list 'ack)) + (thread-sleep! 0.01) + (loop2 1))) "processing")) +(th4 (make-thread (lambda () + (print "In notify-mailbox thread") + (let loop3 ((entries3 0)) + (print "Notifying: " (mailbox-receive! notify-mailbox)) + (handle-exceptions exn (begin (print "Had an issue: " (message exn))(thread-sleep! 10)) + (define-values (i o) (tcp-connect "localhost" 6504)) + (serialize (list 'ack "from-server-one") o) + (print (read-line i)) + (close-input-port i) + (close-output-port o)) + ;;(thread-sleep! 1) + (loop3 1))) "notify")) + +) +(thread-start! th1) +(thread-start! th2) +(thread-start! th3) +(thread-start! th4) +(thread-join! th2) +) + +(print "Done here") ADDED ulex-trials/server-two.scm Index: ulex-trials/server-two.scm ================================================================== --- /dev/null +++ ulex-trials/server-two.scm @@ -0,0 +1,61 @@ +(import tcp-server format (chicken tcp) (chicken io) (chicken string) (prefix sqlite3 sqlite3:) sql-de-lite srfi-18 simple-exceptions mailbox s11n) +(let* ((work-mailbox (make-mailbox)) + (notify-mailbox (make-mailbox)) +(th1 (make-thread (lambda () + +((make-tcp-server + (tcp-listen 6504) + (lambda () + (let* ((db (sqlite3:open-database "test.db")) + (rec-data (deserialize))) + (sqlite3:set-busy-handler! db (sqlite3:make-busy-timeout 136000)) + ;;(exec (sql db "INSERT INTO entries (received,send) VALUES (?,?);") "something" (conc "Server One Response: " "something else")) + (sqlite3:execute db "INSERT INTO entries (received,send) VALUES (?,?);" "something" (conc "Server One Response: " "something else")) + (mailbox-send! work-mailbox rec-data) + (format (current-error-port) (conc rec-data)) + (write-line (conc "Response to: " (conc rec-data))) + ;;(close-database db) + ))) +#t)) +"receive")) +(th2 (make-thread (lambda () + (print "Jeff is here") + (let loop ((entries 0)) + (thread-sleep! 0.01) + (print "Preparding to send entries" entries) + (handle-exceptions exn (begin (print "Had an issue: " (message exn))(thread-sleep! 10)) + (define-values (i o) (tcp-connect "localhost" 6505)) + (serialize (list "localhost:6505" "from-server-two") o) + (print (read-line i)) + (close-input-port i) + (close-output-port o)) + (loop (+ entries 1)))) "send")) +(th3 (make-thread (lambda () + (print "In mailbox thread") + (let loop2 ((entries2 0)) + (print "Processing: " (mailbox-receive! work-mailbox)) + (mailbox-send! notify-mailbox (list 'ack)) + (thread-sleep! 0.01) + (loop2 1))) "processing")) +(th4 (make-thread (lambda () + (print "In notify-mailbox thread") + (let loop3 ((entries3 0)) + (print "Notifying: " (mailbox-receive! notify-mailbox)) + (handle-exceptions exn (begin (print "Had an issue: " (message exn))(thread-sleep! 10)) + (define-values (i o) (tcp-connect "localhost" 6505)) + (serialize (list 'ack "from-server-two") o) + (print (read-line i)) + (close-input-port i) + (close-output-port o)) + ;;(thread-sleep! 1) + (loop3 1))) "notify")) + +) +(thread-start! th1) +(thread-start! th2) +(thread-start! th3) +(thread-start! th4) +(thread-join! th2) +) + +(print "Done here") ADDED ulex-trials/ulex-test.scm Index: ulex-trials/ulex-test.scm ================================================================== --- /dev/null +++ ulex-trials/ulex-test.scm @@ -0,0 +1,100 @@ +(include "../ulex/ulex.scm") + +(module ulex-test * + +(import scheme + (chicken io) + (chicken base) + (chicken time) + (chicken file) + (chicken file posix) + (chicken string) + (chicken process-context) + (chicken process-context posix) + miscmacros +;; nng + srfi-18 + srfi-69 + test + matchable + typed-records + system-information + directory-utils + + ulex + ) + +(define help "Usage: ulex-test COMMAND + where COMMAND is one of: + run host:port : start test server - start several in same dir +") + +(define (call uconn msg addr) + (print "Sent: "msg", received: " + (send-receive uconn addr 'hello msg))) + +;; start => hello 0 +;; hello 0 => hello 1 +;; hello 1 => hello 2 +;; ... +;; hello 11 => 'done +;; +(define (process-message mesg) + (let ((parts (string-split mesg))) + (match + parts + ((msg c) + (let ((count (string->number c))) + (if (> count 10) + 'done + (conc msg " " (if count count 0))))) + ((msg) + (conc msg " 0")) + (else + "hello 0")))) + +(define (main) + (match + (command-line-arguments) + ((run myport) + ;; start listener + ;; put myaddr into file by host-pid in .runners + ;; for 1 minute + ;; get all in .runners + ;; call each with a message + ;; + (let* ((port (string->number myport)) + (endtimes (+ (current-seconds) 20)) ;; run for 20 seconds + (handler (lambda (rem-host-port qrykey cmd params) + (process-message params))) + (uconn (run-listener handler myport)) + (rfile (conc ".runners/"(get-host-name)"-"(current-process-id)))) + (if (not (and (file-exists? ".runners") + (directory? ".runners"))) + (create-directory ".runners" #t)) + (with-output-to-file rfile + (lambda () + (print myport))) + (let loop ((entries '())) + (if (> (current-seconds) endtimes) + (begin + (delete-file* rfile) + (sleep 1) + (exit)) + (if (null? entries) + (loop (glob ".runners/*")) + (let* ((entry (car entries)) + (destaddr (with-input-from-file entry read-line))) + (call uconn (conc "hello-from-"myport"to-"destaddr) destaddr) + ;; (thread-sleep! 0.025) + (loop (cdr entries)))))))) + ((cmd)(print "ERROR: command "cmd", not recognised.\n\n"help)) + (else + (print help)))) + +) ;; end module + +(import ulex-test) +(main) + + Index: ulex.scm ================================================================== --- ulex.scm +++ ulex.scm @@ -17,8 +17,8 @@ ;; along with Megatest. If not, see . ;;====================================================================== (declare (unit ulex)) -(declare (uses pkts)) (include "ulex/ulex.scm") +;; (include "ulex-simple/ulex.scm") ADDED ulex/Makefile Index: ulex/Makefile ================================================================== --- /dev/null +++ ulex/Makefile @@ -0,0 +1,9 @@ +all : ulex.pdf ulex.png + +ulex.pdf : ulex.dot + dot -Tpdf ulex.dot -o ulex.pdf + +ulex.png : ulex.dot + dot -Tpng ulex.dot -o ulex.png + + DELETED ulex/ulex-prev-orig.scm Index: ulex/ulex-prev-orig.scm ================================================================== --- ulex/ulex-prev-orig.scm +++ /dev/null @@ -1,2252 +0,0 @@ -;; ulex: Distributed sqlite3 db -;;; -;; Copyright (C) 2018 Matt Welland -;; Redistribution and use in source and binary forms, with or without -;; modification, is permitted. -;; -;; THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS -;; OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED -;; WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE -;; ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE -;; LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR -;; CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT -;; OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR -;; BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF -;; LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -;; (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE -;; USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH -;; DAMAGE. - -;;====================================================================== -;; ABOUT: -;; See README in the distribution at https://www.kiatoa.com/fossils/ulex -;; NOTES: -;; Why sql-de-lite and not say, dbi? - performance mostly, then simplicity. -;; -;;====================================================================== - -(use mailbox) - -(module ulex - * - -(import scheme posix chicken data-structures ports extras files mailbox) -(import srfi-18 pkts matchable regex - typed-records srfi-69 srfi-1 - srfi-4 regex-case - (prefix sqlite3 sqlite3:) - foreign - tcp6 - ;; ulex-netutil - hostinfo - ) - -;; make it a global? Well, it is local to area module - -(define *captain-pktspec* - `((captain (host . 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 - ))) - -;; struct for keeping track of our world - -(defstruct udat - ;; captain info - (captain-address #f) - (captain-host #f) - (captain-port #f) - (captain-pid #f) - (captain-lease 0) ;; time (unix epoc) seconds when the lease is up - (ulex-dir (conc (get-environment-variable "HOME") "/.ulex")) - (cpkts-dir (conc (get-environment-variable "HOME") "/.ulex/pkts")) - (cpkt-spec *captain-pktspec*) - ;; this processes info - (my-cpkt-key #f) ;; put Z card here when I create a pkt for myself as captain - (my-address #f) - (my-hostname #f) - (my-port #f) - (my-pid (current-process-id)) - (my-dbs '()) - ;; server and handler thread - (serv-listener #f) ;; this processes server info - (handler-thread #f) - (mboxes (make-hash-table)) ;; key => mbox - ;; other servers - (peers (make-hash-table)) ;; host-port => peer record - (dbowners (make-hash-table)) ;; dbfile => host-port - (handlers (make-hash-table)) ;; dbfile => proc - ;; (outgoing-conns (make-hash-table)) ;; host:port -> conn - (work-queue (make-queue)) ;; most stuff goes here - ;; (fast-queue (make-queue)) ;; super quick stuff goes here (e.g. ping) - (busy #f) ;; is either of the queues busy, use to switch between queuing tasks or doing immediately - ;; app info - (appname #f) - (dbtypes (make-hash-table)) ;; this should be an alist but hash is easier. dbtype => [ initproc syncproc ] - ;; cookies - (cnum 0) ;; cookie num - ) - -;;====================================================================== -;; NEW APPROACH -;;====================================================================== - -;; start-server-find-port ;; gotta have a server port ready from the very begining - -;; udata - all the connection info, captain, server, ulex db etc. MUST BE PASSED IN -;; dbpath - full path and filename of the db to talk to or a symbol naming the db? -;; callname - the remote call to execute -;; params - parameters to pass to the remote call -;; -(define (remote-call udata dbpath dbtype callname . params) - (start-server-find-port udata) ;; ensure we have a local server - (find-or-setup-captain udata) - ;; look at connect, process-request, send, send-receive - (let-values (((cookie-key host-port)(get-db-owner udata dbpath dbtype))) - (send-receive udata host-port callname cookie-key params))) - -;;====================================================================== -;; KEY FUNCTIONS - THESE ARE TOO BE EXPOSED AND USED -;;====================================================================== - -;; connection setup and management functions - -;; This is the basic setup command. Must always be -;; called before connecting to a db using connect. -;; -;; find or become the captain -;; setup and return a ulex object -;; -(define (find-or-setup-captain udata) - ;; see if we already have a captain and if the lease is ok - (if (and (udat-captain-address udata) - (udat-captain-port udata) - (< (current-seconds) (udat-captain-lease udata))) - udata - (let* ((cpkts (get-all-captain-pkts udata)) ;; read captain pkts - (captn (get-winning-pkt cpkts))) - (if captn - (let* ((port (alist-ref 'port captn)) - (host (alist-ref 'host captn)) - (ipaddr (alist-ref 'ipaddr captn)) - (pid (alist-ref 'pid captn)) - (Z (alist-ref 'Z captn))) - (udat-captain-address-set! udata ipaddr) - (udat-captain-host-set! udata host) - (udat-captain-port-set! udata port) - (udat-captain-pid-set! udata pid) - (udat-captain-lease-set! udata (+ (current-seconds) 10)) - (let-values (((success pingtime)(ping udata (conc ipaddr ":" port)))) - (if success - udata - (begin - (print "Found unreachable captain at " ipaddr ":" port ", removing pkt") - (remove-captain-pkt udata captn) - (find-or-setup-captain udata)))) - (begin - (setup-as-captain udata) ;; this saves the thread to captain-thread and starts the thread - (find-or-setup-captain udata))))))) - -;; connect to a specific dbfile -;; - if already connected - return the dbowner host-port -;; - ask the captain who to talk to for this db -;; - put the entry in the dbowners hash as dbfile => host-port -;; -(define (connect udata dbfname dbtype) - (or (hash-table-ref/default (udat-dbowners udata) dbfname #f) - (let-values (((success dbowner-host-port)(get-db-owner udata dbfname dbtype))) - (if success - (begin - ;; just clobber the record, this is the new data no matter what - (hash-table-set! (udat-dbowners udata) dbfname dbowner-host-port) - dbowner-host-port) - #f)))) - -;; returns: success pingtime -;; -;; NOTE: causes the callee to store the info on this host along with the dbs this host currently owns -;; -(define (ping udata host-port) - (let* ((start (current-milliseconds)) - (cookie (make-cookie udata)) - (dbs (udat-my-dbs udata)) - (msg (string-intersperse dbs " ")) - (res (send udata host-port 'ping cookie msg retval: #t)) - (delta (- (current-milliseconds) start))) - (values (equal? res cookie) delta))) - -;; returns: success pingtime -;; -;; NOTE: causes all references to this worker to be wiped out in the -;; callee (ususally the captain) -;; -(define (goodbye-ping udata host-port) - (let* ((start (current-milliseconds)) - (cookie (make-cookie udata)) - (dbs (udat-my-dbs udata)) - (res (send udata host-port 'goodbye cookie "nomsg" retval: #t)) - (delta (- (current-milliseconds) start))) - (values (equal? res cookie) delta))) - -(define (goodbye-captain udata) - (let* ((host-port (udat-captain-host-port udata))) - (if host-port - (goodbye-ping udata host-port) - (values #f -1)))) - -(define (get-db-owner udata dbname dbtype) - (let* ((host-port (udat-captain-host-port udata))) - (if host-port - (let* ((cookie (make-cookie udata)) - (msg #f) ;; (conc dbname " " dbtype)) - (params `(,dbname ,dbtype)) - (res (send udata host-port 'db-owner cookie msg - params: params retval: #t))) - (match (string-split res) - ((retcookie owner-host-port) - (values (equal? retcookie cookie) owner-host-port)))) - (values #f -1)))) - -;; called in ulex-handler to dispatch work, called on the workers side -;; calls (proc params data) -;; returns result with cookie -;; -;; pdat is the info of the caller, used to send the result data -;; prockey is key into udat-handlers hash dereferencing a proc -;; procparam is a first param handed to proc - often to do further derefrencing -;; NOTE: params is intended to be a list of strings, encoding on data -;; is up to the user but data must be a single line -;; -(define (process-request udata pdat dbname cookie prockey procparam data) - (let* ((dbrec (ulex-open-db udata dbname)) ;; this will be a dbconn record, looks for in udata first - (proc (hash-table-ref udata prockey))) - (let* ((result (proc dbrec procparam data))) - result))) - -;; remote-request - send to remote to process in process-request -;; uconn comes from a call to connect and can be used instead of calling connect again -;; uconn is the host-port to call -;; we send dbname to the worker so they know which file to open -;; data must be a string with no newlines, it will be handed to the proc -;; at the remote site unchanged. It is up to the user to encode/decode it's contents -;; -;; rtype: immediate, read-only, normal, low-priority -;; -(define (remote-request udata uconn rtype dbname prockey procparam data) - (let* ((cookie (make-cookie udata))) - (send-receive udata uconn rtype cookie data `(,prockey procparam)))) - -(define (ulex-open-db udata dbname) - #f) - - -;;====================================================================== -;; Ulex db -;; -;; - track who is captain, lease expire time -;; - track who owns what db, lease -;; -;;====================================================================== - -;; -;; -(define (ulex-dbfname) - (let ((dbdir (conc (get-environment-variable "HOME") "/.ulex"))) - (if (not (file-exists? dbdir)) - (create-directory dbdir #t)) - (conc dbdir "/network.db"))) - -;; always goes in ~/.ulex/network.db -;; role is captain, adjutant, node -;; -(define (ulexdb-setup) - (let* ((dbfname (ulex-dbfname)) - (have-db (file-exists? dbfname)) - (db (sqlite3:open-database dbfname))) - (sqlite3:set-busy-handler! db (sqlite3:make-busy-timeout 136000)) - (sqlite3:execute db "PRAGMA synchronous = 0;") - (if (not have-db) - (sqlite3:with-transaction - db - (lambda () - (for-each - (lambda (stmt) - (if stmt (sqlite3:execute db stmt))) - `("CREATE TABLE IF NOT EXISTS nodes - (id INTEGER PRIMARY KEY, - role TEXT NOT NULL, - host TEXT NOT NULL, - port TEXT NOT NULL, - ipadr TEXT NOT NULL, - pid INTEGER NOT NULL, - zcard TEXT NOT NULL, - regtime INTEGER DEFAULT (strftime('%s','now')), - lease_thru INTEGER DEFAULT (strftime('%s','now')), - last_update INTEGER DEFAULT (strftime('%s','now')));" - "CREATE TRIGGER IF NOT EXISTS update_nodes_trigger AFTER UPDATE ON nodes - FOR EACH ROW - BEGIN - UPDATE nodes SET last_update=(strftime('%s','now')) - WHERE id=old.id; - END;" - "CREATE TABLE IF NOT EXISTS dbs - (id INTEGER PRIMARY KEY, - dbname TEXT NOT NULL, - dbfile TEXT NOT NULL, - dbtype TEXT NOT NULL, - host_port TEXT NOT NULL, - regtime INTEGER DEFAULT (strftime('%s','now')), - lease_thru INTEGER DEFAULT (strftime('%s','now')), - last_update INTEGER DEFAULT (strftime('%s','now')));" - "CREATE TRIGGER IF NOT EXISTS update_dbs_trigger AFTER UPDATE ON dbs - FOR EACH ROW - BEGIN - UPDATE dbs SET last_update=(strftime('%s','now')) - WHERE id=old.id; - END;"))))) - db)) - -(define (get-host-port-lease db dbfname) - (sqlite3:fold-row - (lambda (rem host-port lease-thru) - (list host-port lease-thru)) - #f db "SELECT host_port,lease_thru FROM dbs WHERE dbfile = ?" dbfname)) - -(define (register-captain db host ipadr port pid zcard #!key (lease 20)) - (let* ((dbfname (ulex-dbfname)) - (host-port (conc host ":" port))) - (sqlite3:with-transaction - db - (lambda () - (match (get-host-port-lease db dbfname) - ((host-port lease-thru) - (if (> (current-seconds) lease-thru) - (begin - (sqlite3:execute db "UPDATE dbs SET host_port=?,lease_thru=? WHERE dbname=?" - (conc host ":" port) - (+ (current-seconds) lease) - dbfname) - #t) - #f)) - (#f (sqlite3:execute db "INSERT INTO dbs (dbname,dbfile,dbtype,host_port,lease_thru) VALUES (?,?,?,?,?)" - "captain" dbfname "captain" host-port (+ (current-seconds) lease))) - (else (print "ERROR: Unrecognised result from fold-row") - (exit 1))))))) - -;;====================================================================== -;; network utilities -;;====================================================================== - -(define (rate-ip ipaddr) - (regex-case ipaddr - ( "^127\\..*" _ 0 ) - ( "^(10\\.0|192\\.168)\\..*" _ 1 ) - ( else 2 ) )) - -;; Change this to bias for addresses with a reasonable broadcast value? -;; -(define (ip-pref-less? a b) - (> (rate-ip a) (rate-ip b))) - - -(define (get-my-best-address) - (let ((all-my-addresses (get-all-ips)) - ;;(all-my-addresses-old (vector->list (hostinfo-addresses (hostname->hostinfo (get-host-name))))) - ) - (cond - ((null? all-my-addresses) - (get-host-name)) ;; no interfaces? - ((eq? (length all-my-addresses) 1) - (car all-my-addresses)) ;; only one to choose from, just go with it - - (else - (car (sort all-my-addresses ip-pref-less?))) - ;; (else - ;; (ip->string (car (filter (lambda (x) ;; take any but 127. - ;; (not (eq? (u8vector-ref x 0) 127))) - ;; all-my-addresses)))) - - ))) - -(define (get-all-ips-sorted) - (sort (get-all-ips) ip-pref-less?)) - -(define (get-all-ips) - (map ip->string (vector->list - (hostinfo-addresses - (host-information (current-hostname)))))) - -(define (udat-my-host-port udata) - (if (and (udat-my-address udata)(udat-my-port udata)) - (conc (udat-my-address udata) ":" (udat-my-port udata)) - #f)) - -(define (udat-captain-host-port udata) - (if (and (udat-captain-address udata)(udat-captain-port udata)) - (conc (udat-captain-address udata) ":" (udat-captain-port udata)) - #f)) - -(define (udat-get-peer udata host-port) - (hash-table-ref/default (udat-peers udata) host-port #f)) - -;; struct for keeping track of others we are talking to - -(defstruct peer - (addr-port #f) - (hostname #f) - (pid #f) - ;; (inp #f) - ;; (oup #f) - (dbs '()) ;; list of databases this peer is currently handling - ) - -(defstruct work - (peer-dat #f) - (handlerkey #f) - (qrykey #f) - (data #f) - (start (current-milliseconds))) - -#;(defstruct dbowner - (pdat #f) - (last-update (current-seconds))) - -;;====================================================================== -;; Captain functions -;;====================================================================== - -;; NB// This needs to be started in a thread -;; -;; setup to be a captain -;; - local server MUST be started already -;; - create pkt -;; - start server port handler -;; -(define (setup-as-captain udata) - (if (create-captain-pkt udata) - (let* ((my-addr (udat-my-address udata)) - (my-port (udat-my-port udata)) - (th (make-thread (lambda () - (ulex-handler-loop udata)) "Captain handler"))) - (udat-handler-thread-set! udata th) - (udat-captain-address-set! udata my-addr) - (udat-captain-port-set! udata my-port) - (thread-start! th)) - (begin - (print "ERROR: failed to create captain pkt") - #f))) - -;; given a pkts dir read -;; -(define (get-all-captain-pkts udata) - (let* ((pktsdir (let ((d (udat-cpkts-dir udata))) - (if (file-exists? d) - d - (begin - (create-directory d #t) - d)))) - (all-pkt-files (glob (conc pktsdir "/*.pkt"))) - (pkt-spec (udat-cpkt-spec udata))) - (map (lambda (pkt-file) - (read-pkt->alist pkt-file pktspec: pkt-spec)) - all-pkt-files))) - -;; sort by D then Z, return one, choose the oldest then -;; differentiate if needed using the Z key -;;l -(define (get-winning-pkt pkts) - (if (null? pkts) - #f - (car (sort pkts (lambda (a b) - (let ((ad (string->number (alist-ref 'D a))) - (bd (string->number (alist-ref 'D b)))) - (if (eq? a b) - (let ((az (alist-ref 'Z a)) - (bz (alist-ref 'Z b))) - (string>=? az bz)) - (> ad bd)))))))) - -;; put the host, ip, port and pid into a pkt in -;; the captain pkts dir -;; - assumes user has already fired up a server -;; which will be in the udata struct -;; -(define (create-captain-pkt udata) - (if (not (udat-serv-listener udata)) - (begin - (print "ERROR: create-captain-pkt called with out a listener") - #f) - (let* ((pktdat `((port . ,(udat-my-port udata)) - (host . ,(udat-my-hostname udata)) - (ipaddr . ,(udat-my-address udata)) - (pid . ,(udat-my-pid udata)))) - (pktdir (udat-cpkts-dir udata)) - (pktspec (udat-cpkt-spec udata)) - ) - (udat-my-cpkt-key-set! - udata - (write-alist->pkt - pktdir - pktdat - pktspec: pktspec - ptype: 'captain)) - (udat-my-cpkt-key udata)))) - -;; remove pkt associated with captn (the Z key .pkt) -;; -(define (remove-captain-pkt udata captn) - (let ((Z (alist-ref 'Z captn)) - (cpktdir (udat-cpkts-dir udata))) - (delete-file* (conc cpktdir "/" Z ".pkt")))) - -;; call all known peers and tell them to delete their info on the captain -;; thus forcing them to re-read pkts and connect to a new captain -;; call this when the captain needs to exit and if an older captain is -;; detected. Due to delays in sending file meta data in NFS multiple -;; captains can be initiated in a "Storm of Captains", book soon to be -;; on Amazon -;; -(define (drop-captain udata) - (let* ((peers (hash-table-keys (udat-peers udata))) - (cookie (make-cookie udata))) - (for-each - (lambda (host-port) - (send udata host-port 'dropcaptain cookie "nomsg" retval: #t)) - peers))) - -;;====================================================================== -;; server primitives -;;====================================================================== - -(define (make-cookie udata) - (let ((newcnum (+ (udat-cnum udata) 1))) - (udat-cnum-set! udata newcnum) - (conc (udat-my-address udata) ":" - (udat-my-port udata) "-" - (udat-my-pid udata) "-" - newcnum))) - -;; create a tcp listener and return a populated udat struct with -;; my port, address, hostname, pid etc. -;; return #f if fail to find a port to allocate. -;; -;; if udata-in is #f create the record -;; if there is already a serv-listener return the udata -;; -(define (start-server-find-port udata-in #!optional (port 4242)) - (let ((udata (or udata-in (make-udat)))) - (if (udat-serv-listener udata) ;; TODO - add check that the listener is alive and ready? - udata - (handle-exceptions - exn - (if (< port 65535) - (start-server-find-port udata (+ port 1)) - #f) - (connect-server udata port))))) - -(define (connect-server udata port) - ;; (tcp-listener-socket LISTENER)(socket-name so) - ;; sockaddr-address, sockaddr-port, sockaddr->string - (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]]) - (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname))) - (udat-my-address-set! udata addr) - (udat-my-port-set! udata port) - (udat-my-hostname-set! udata (get-host-name)) - (udat-serv-listener-set! udata tlsn) - udata)) - -(define (get-peer-dat udata host-port #!optional (hostname #f)(pid #f)) - (let* ((pdat (or (udat-get-peer udata host-port) - (handle-exceptions ;; ERROR - MAKE THIS EXCEPTION HANDLER MORE SPECIFIC - exn - #f - (let ((npdat (make-peer addr-port: host-port))) - (if hostname (peer-hostname-set! npdat hostname)) - (if pid (peer-pid-set! npdat pid)) - npdat))))) - pdat)) - -;; send structured data to recipient -;; -;; NOTE: qrykey is what was called the "cookie" previously -;; -;; retval tells send to expect and wait for return data (one line) and return it or time out -;; this is for ping where we don't want to necessarily have set up our own server yet. -;; -(define (send udata host-port handler qrykey data - #!key (hostname #f)(pid #f)(params '())(retval #f)) - (let* ((my-host-port (udat-my-host-port udata)) - (isme (equal? host-port my-host-port)) ;; am I calling - ;; myself? - (dat (list - handler ;; " " - my-host-port ;; " " - (udat-my-pid udata) ;; " " - qrykey - params ;;(if (null? params) "" (conc " " - ;;(string-intersperse params " "))) - ))) - ;; (print "send isme is " (if isme "true!" "false!") ", - ;; my-host-port: " my-host-port ", host-port: " host-port) - (if isme - (ulex-handler udata dat data) - (handle-exceptions ;; ERROR - MAKE THIS EXCEPTION HANDLER MORE - ;; SPECIFIC - exn - #f - (let-values (((inp oup)(tcp-connect host-port))) - ;; - ;; CONTROL LINE: - ;; handlerkey host:port pid qrykey params ... - ;; - (let ((res - (if (and inp oup) - (let* () - (if my-host-port - (begin - (write dat oup) - (write data oup) ;; send as sexpr - ;; (print "Sent dat: " dat " data: " data) - (if retval - (read inp) - #t)) - (begin - (print "ERROR: send called but no receiver has been setup. Please call setup first!") - #f)) - ;; NOTE: DO NOT BE TEMPTED TO LOOK AT ANY DATA ON INP HERE! - ;; (there is a listener for handling that) - ) - #f))) ;; #f means failed to connect and send - (close-input-port inp) - (close-output-port oup) - res)))))) - -;; send a request to the given host-port and register a mailbox in udata -;; wait for the mailbox data and return it -;; -(define (send-receive udata host-port handler qrykey data #!key (hostname #f)(pid #f)(params '())(timeout 20)) - (let ((mbox (make-mailbox)) - (mbox-time (current-milliseconds)) - (mboxes (udat-mboxes udata))) - (hash-table-set! mboxes qrykey mbox) - (if (send udata host-port handler qrykey data hostname: hostname pid: pid params: params) - (let* ((mbox-timeout-secs timeout) - (mbox-timeout-result 'MBOX_TIMEOUT) - (res (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result)) - (mbox-receive-time (current-milliseconds))) - (hash-table-delete! mboxes qrykey) - (if (eq? res 'MBOX_TIMEOUT) - #f - res)) - #f))) ;; #f means failed to communicate - -;; -(define (ulex-handler udata controldat data) - (print "controldat: " controldat " data: " data) - (match controldat ;; (string-split controldat) - ((handlerkey host-port pid qrykey params ...) - ;; (print "handlerkey: " handlerkey " host-port: " host-port " pid: " pid " qrykey: " qrykey " params: " params) - (case handlerkey ;; (string->symbol handlerkey) - ((ack)(print "Got ack!")) - ((ping) ;; special case - return result immediately on the same connection - (let* ((proc (hash-table-ref/default (udat-handlers udata) 'ping #f)) - (val (if proc (proc) "gotping")) - (peer (make-peer addr-port: host-port pid: pid)) - (dbshash (udat-dbowners udata))) - (peer-dbs-set! peer params) ;; params for ping is list of dbs owned by pinger - (for-each (lambda (dbfile) - (hash-table-set! dbshash dbfile host-port)) ;; WRONG? - params) ;; register each db in the dbshash - (if (not (hash-table-exists? (udat-peers udata) host-port)) - (hash-table-set! (udat-peers udata) host-port peer)) ;; save the details of this caller in peers - qrykey)) ;; End of ping - ((goodbye) - ;; remove all traces of the caller in db ownership etc. - (let* ((peer (hash-table-ref/default (udat-peers udata) host-port #f)) - (dbs (if peer (peer-dbs peer) '())) - (dbshash (udat-dbowners udata))) - (for-each (lambda (dbfile)(hash-table-delete! dbshash dbfile)) dbs) - (hash-table-delete! (udat-peers udata) host-port) - qrykey)) - ((dropcaptain) - ;; remove all traces of the captain - (udat-captain-address-set! udata #f) - (udat-captain-host-set! udata #f) - (udat-captain-port-set! udata #f) - (udat-captain-pid-set! udata #f) - qrykey) - ((rucaptain) ;; remote is asking if I'm the captain - (if (udat-my-cpkt-key udata) "yes" "no")) - ((db-owner) ;; given a db name who do I send my queries to - ;; look up the file in handlers, if have an entry ping them to be sure - ;; they are still alive and then return that host:port. - ;; if no handler found or if the ping fails pick from peers the oldest that - ;; is managing the fewest dbs - (match params - ((dbfile dbtype) - (let* ((owner-host-port (hash-table-ref/default (udat-dbowners udata) dbfile #f))) - (if owner-host-port - (conc qrykey " " owner-host-port) - (let* ((pdat (or (hash-table-ref/default (udat-peers udata) host-port #f) ;; no owner - caller gets to own it! - (make-peer addr-port: host-port pid: pid dbs: `(,dbfile))))) - (hash-table-set! (udat-peers udata) host-port pdat) - (hash-table-set! (udat-dbowners udata) dbfile host-port) - (conc qrykey " " host-port))))) - (else (conc qrykey " BADDATA")))) - ;; for work items: - ;; handler is one of; immediate, read-only, read-write, high-priority - ((immediate read-only normal low-priority) ;; do this work immediately - ;; host-port (caller), pid (caller), qrykey (cookie), params <= all from first line - ;; data => a single line encoded however you want, or should I build json into it? - (print "handlerkey=" handlerkey) - (let* ((pdat (get-peer-dat udata host-port))) - (match params ;; dbfile prockey procparam - ((dbfile prockey procparam) - (case handlerkey - ((immediate read-only) - (process-request udata pdat dbfile qrykey prockey procparam data)) - ((normal low-priority) ;; split off later and add logic to support low priority - (add-to-work-queue udata pdat dbfile qrykey prockey procparam data)) - (else - #f))) - (else - (print "INFO: params=" params " handlerkey=" handlerkey " controldat=" controldat) - #f)))) - (else - ;; (add-to-work-queue udata (get-peer-dat udata host-port) handlerkey qrykey data) - #f))) - (else - (print "BAD DATA? controldat=" controldat " data=" data) - #f)));; handles the incoming messages and dispatches to queues - -;; -(define (ulex-handler-loop udata) - (let* ((serv-listener (udat-serv-listener udata))) - ;; data comes as two lines - ;; handlerkey resp-addr:resp-port hostname pid qrykey [dbpath/dbfile.db] - ;; data - (let loop ((state 'start)) - (let-values (((inp oup)(tcp-accept serv-listener))) - (let* ((controldat (read inp)) - (data (read inp)) - (resp (ulex-handler udata controldat data))) - (if resp (write resp oup)) - (close-input-port inp) - (close-output-port oup)) - (loop state))))) - -;; add a proc to the handler list, these are done symetrically (i.e. in all instances) -;; so that the proc can be dereferenced remotely -;; -(define (register-handler udata key proc) - (hash-table-set! (udat-handlers udata) key proc)) - - -;;====================================================================== -;; work queues -;;====================================================================== - -(define (add-to-work-queue udata peer-dat handlerkey qrykey data) - (let ((wdat (make-work peer-dat: peer-dat handlerkey: handlerkey qrykey: qrykey data: data))) - (if (udat-busy udata) - (queue-add! (udat-work-queue udata) wdat) - (process-work udata wdat)) ;; passing in wdat tells process-work to first process the passed in wdat - )) - -(define (do-work udata wdat) - #f) - -(define (process-work udata #!optional wdat) - (if wdat (do-work udata wdat)) ;; process wdat - (let ((wqueue (udat-work-queue udata))) - (if (not (queue-empty? wqueue)) - (let loop ((wd (queue-remove! wqueue))) - (do-work udata wd) - (if (not (queue-empty? wqueue)) - (loop (queue-remove! wqueue))))))) - -;;====================================================================== -;; Generic db handling -;; setup a inmem db instance -;; open connection to on-disk db -;; sync on-disk db to inmem -;; get lock in on-disk db for dbowner of this db -;; put sync-proc, init-proc, on-disk handle, inmem handle in dbconn stuct -;; return the stuct -;;====================================================================== - -(defstruct dbconn - (fname #f) - (inmem #f) - (conn #f) - (sync #f) ;; sync proc - (init #f) ;; init proc - (lastsync (current-seconds)) - ) - -(defstruct dbinfo - (initproc #f) - (syncproc #f)) - -;; open inmem and disk database -;; init with initproc -;; return db struct -;; -;; appname; megatest, ulex or something else. -;; -(define (setup-db-connection udata fname-in appname dbtype) - (let* ((is-ulex (eq? appname 'ulex)) - (dbinf (if is-ulex ;; ulex is a built-in special case - (make-dbinfo initproc: ulexdb-init syncproc: ulexdb-sync) - (hash-table-ref/default (udat-dbtypes udata) dbtype #f))) - (initproc (dbinfo-initproc dbinf)) - (syncproc (dbinfo-syncproc dbinf)) - (fname (if is-ulex - (conc (udat-ulex-dir udata) "/ulex.db") - fname-in)) - (inmem-db (open-and-initdb udata #f 'inmem (dbinfo-initproc dbinf))) - (disk-db (open-and-initdb udata fname 'disk (dbinfo-initproc dbinf)))) - (make-dbconn inmem: inmem-db conn: disk-db sync: syncproc init: initproc))) - -;; dest='inmem or 'disk -;; -(define (open-and-initdb udata filename dest init-proc) - (let* ((inmem (eq? dest 'inmem)) - (dbfile (if inmem - ":INMEM:" - filename)) - (dbexists (if inmem #t (file-exists? dbfile))) - (db (sqlite3:open-database dbfile))) - (sqlite3:set-busy-handler! db (sqlite3:make-busy-timeout 136000)) - (if (not dbexists) - (init-proc db)) - db)) - - -;;====================================================================== -;; Previous Ulex db stuff -;;====================================================================== - -(define (ulexdb-init db inmem) - (sqlite3:with-transaction - db - (lambda () - (for-each - (lambda (stmt) - (if stmt (sqlite3:execute db stmt))) - `("CREATE TABLE IF NOT EXISTS processes - (id INTEGER PRIMARY KEY, - host TEXT NOT NULL, - ipadr TEXT NOT NULL, - port INTEGER NOT NULL, - pid INTEGER NOT NULL, - regtime INTEGER DEFAULT (strftime('%s','now')), - last_update INTEGER DEFAULT (strftime('%s','now')));" - (if inmem - "CREATE TRIGGER IF NOT EXISTS update_proces_trigger AFTER UPDATE ON processes - FOR EACH ROW - BEGIN - UPDATE processes SET last_update=(strftime('%s','now')) - WHERE id=old.id; - END;" - #f)))))) - -;; open databases, do initial sync -(define (ulexdb-sync dbconndat udata) - #f) - - -) ;; END OF ULEX - - -;;; ;;====================================================================== -;;; ;; D E B U G H E L P E R S -;;; ;;====================================================================== -;;; -;;; (define (dbg> . args) -;;; (with-output-to-port (current-error-port) -;;; (lambda () -;;; (apply print "dbg> " args)))) -;;; -;;; (define (debug-pp . args) -;;; (if (get-environment-variable "ULEX_DEBUG") -;;; (with-output-to-port (current-error-port) -;;; (lambda () -;;; (apply pp args))))) -;;; -;;; (define *default-debug-port* (current-error-port)) -;;; -;;; (define (sdbg> fn stage-name stage-start stage-end start-time . message) -;;; (if (get-environment-variable "ULEX_DEBUG") -;;; (with-output-to-port *default-debug-port* -;;; (lambda () -;;; (apply print "ulex:" fn " " stage-name " took " (- (if stage-end stage-end (current-milliseconds)) stage-start) " ms. " -;;; (if start-time -;;; (conc "total time " (- (current-milliseconds) start-time) -;;; " ms.") -;;; "") -;;; message -;;; ))))) - -;;====================================================================== -;; M A C R O S -;;====================================================================== -;; iup callbacks are not dumping the stack, this is a work-around -;; - -;; Some of these routines use: -;; -;; http://www.cs.toronto.edu/~gfb/scheme/simple-macros.html -;; -;; Syntax for defining macros in a simple style similar to function definiton, -;; when there is a single pattern for the argument list and there are no keywords. -;; -;; (define-simple-syntax (name arg ...) body ...) -;; -;; -;; (define-syntax define-simple-syntax -;; (syntax-rules () -;; ((_ (name arg ...) body ...) -;; (define-syntax name (syntax-rules () ((name arg ...) (begin body ...))))))) -;; -;; (define-simple-syntax (catch-and-dump proc procname) -;; (handle-exceptions -;; exn -;; (begin -;; (print-call-chain (current-error-port)) -;; (with-output-to-port (current-error-port) -;; (lambda () -;; (print ((condition-property-accessor 'exn 'message) exn)) -;; (print "Callback error in " procname) -;; (print "Full condition info:\n" (condition->list exn))))) -;; (proc))) -;; -;; -;;====================================================================== -;; R E C O R D S -;;====================================================================== - -;;; ;; information about me as a server -;;; ;; -;;; (defstruct area -;;; ;; about this area -;;; (useportlogger #f) -;;; (lowport 32768) -;;; (server-type 'auto) ;; auto=create up to five servers/pkts, main=create pkts, passive=no pkt (unless there are no pkts at all) -;;; (conn #f) -;;; (port #f) -;;; (myaddr (get-my-best-address)) -;;; pktid ;; get pkt from hosts table if needed -;;; pktfile -;;; pktsdir -;;; dbdir -;;; (dbhandles (make-hash-table)) ;; fname => list-of-dbh, NOTE: Should really never need more than one? -;;; (mutex (make-mutex)) -;;; (rtable (make-hash-table)) ;; registration table of available actions -;;; (dbs (make-hash-table)) ;; filename => random number, used for choosing what dbs I serve -;;; ;; about other servers -;;; (hosts (make-hash-table)) ;; key => hostdat -;;; (hoststats (make-hash-table)) ;; key => alist of fname => ( qcount . qtime ) -;;; (reqs (make-hash-table)) ;; uri => queue -;;; ;; work queues -;;; (wqueues (make-hash-table)) ;; fname => qdat -;;; (stats (make-hash-table)) ;; fname => totalqueries -;;; (last-srvup (current-seconds)) ;; last time we updated the known servers -;;; (cookie2mbox (make-hash-table)) ;; map cookie for outstanding request to mailbox of awaiting call -;;; (ready #f) -;;; (health (make-hash-table)) ;; ipaddr:port => num failed pings since last good ping -;;; ) -;;; -;;; ;; host stats -;;; ;; -;;; (defstruct hostdat -;;; (pkt #f) -;;; (dbload (make-hash-table)) ;; "dbfile.db" => queries/min -;;; (hostload #f) ;; normalized load ( 5min load / numcpus ) -;;; ) -;;; -;;; ;; dbdat -;;; ;; -;;; (defstruct dbdat -;;; (dbh #f) -;;; (fname #f) -;;; (write-access #f) -;;; (sths (make-hash-table)) ;; hash mapping query strings to handles -;;; ) -;;; -;;; ;; qdat -;;; ;; -;;; (defstruct qdat -;;; (writeq (make-queue)) -;;; (readq (make-queue)) -;;; (rwq (make-queue)) -;;; (logq (make-queue)) ;; do we need a queue for logging? yes, if we use sqlite3 db for logging -;;; (osshort (make-queue)) -;;; (oslong (make-queue)) -;;; (misc (make-queue)) ;; used for things like ping-full -;;; ) -;;; -;;; ;; calldat -;;; ;; -;;; (defstruct calldat -;;; (ctype 'dbwrite) -;;; (obj #f) ;; this would normally be an SQL statement e.g. SELECT, INSERT etc. -;;; (rtime (current-milliseconds))) -;;; -;;; ;; make it a global? Well, it is local to area module -;;; -;;; (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 -;;; ))) -;;; -;;; ;; work item -;;; ;; -;;; (defstruct witem -;;; (rhost #f) ;; return host -;;; (ripaddr #f) ;; return ipaddr -;;; (rport #f) ;; return port -;;; (servkey #f) ;; the packet representing the client of this workitem, used by final send-message -;;; (rdat #f) ;; the request - usually an sql query, type is rdat -;;; (action #f) ;; the action: immediate, dbwrite, dbread,oslong, osshort -;;; (cookie #f) ;; cookie id for response -;;; (data #f) ;; the data payload, i.e. parameters -;;; (result #f) ;; the result from processing the data -;;; (caller #f)) ;; the calling peer according to rpc itself -;;; -;;; (define (trim-pktid pktid) -;;; (if (string? pktid) -;;; (substring pktid 0 4) -;;; "nopkt")) -;;; -;;; (define (any->number num) -;;; (cond -;;; ((number? num) num) -;;; ((string? num) (string->number num)) -;;; (else num))) -;;; -;;; (use trace) -;;; (trace-call-sites #t) -;;; -;;; ;;====================================================================== -;;; ;; D A T A B A S E H A N D L I N G -;;; ;;====================================================================== -;;; -;;; ;; look in dbhandles for a db, return it, else return #f -;;; ;; -;;; (define (get-dbh acfg fname) -;;; (let ((dbh-lst (hash-table-ref/default (area-dbhandles acfg) fname '()))) -;;; (if (null? dbh-lst) -;;; (begin -;;; ;; (print "opening db for " fname) -;;; (open-db acfg fname)) ;; Note that the handles get put back in the queue in the save-dbh calls -;;; (let ((rem-lst (cdr dbh-lst))) -;;; ;; (print "re-using saved connection for " fname) -;;; (hash-table-set! (area-dbhandles acfg) fname rem-lst) -;;; (car dbh-lst))))) -;;; -;;; (define (save-dbh acfg fname dbdat) -;;; ;; (print "saving dbh for " fname) -;;; (hash-table-set! (area-dbhandles acfg) fname (cons dbdat (hash-table-ref/default (area-dbhandles acfg) fname '())))) -;;; -;;; ;; open the database, if never before opened init it. put the handle in the -;;; ;; open db's hash table -;;; ;; returns: the dbdat -;;; ;; -;;; (define (open-db acfg fname) -;;; (let* ((fullname (conc (area-dbdir acfg) "/" fname)) -;;; (exists (file-exists? fullname)) -;;; (write-access (if exists -;;; (file-write-access? fullname) -;;; (file-write-access? (area-dbdir acfg)))) -;;; (db (sqlite3:open-database fullname)) -;;; (handler (sqlite3:make-busy-timeout 136000)) -;;; ) -;;; (sqlite3:set-busy-handler! db handler) -;;; (sqlite3:execute db "PRAGMA synchronous = 0;") -;;; (if (not exists) ;; need to init the db -;;; (if write-access -;;; (let ((isql (get-rsql acfg 'dbinitsql))) ;; get the init sql statements -;;; ;; (sqlite3:with-transaction -;;; ;; db -;;; ;; (lambda () -;;; (if isql -;;; (for-each -;;; (lambda (sql) -;;; (sqlite3:execute db sql)) -;;; isql))) -;;; (print "ERROR: no write access to " (area-dbdir acfg)))) -;;; (make-dbdat dbh: db fname: fname write-access: write-access))) -;;; -;;; ;; This is a low-level command to retrieve or to prepare, save and return a prepared statment -;;; ;; you must extract the db handle -;;; ;; -;;; (define (get-sth db cache stmt) -;;; (if (hash-table-exists? cache stmt) -;;; (begin -;;; ;; (print "Reusing cached stmt for " stmt) -;;; (hash-table-ref/default cache stmt #f)) -;;; (let ((sth (sqlite3:prepare db stmt))) -;;; (hash-table-set! cache stmt sth) -;;; ;; (print "prepared stmt for " stmt) -;;; sth))) -;;; -;;; ;; a little more expensive but does all the tedious deferencing - only use if you don't already -;;; ;; have dbdat and db sitting around -;;; ;; -;;; (define (full-get-sth acfg fname stmt) -;;; (let* ((dbdat (get-dbh acfg fname)) -;;; (db (dbdat-dbh dbdat)) -;;; (sths (dbdat-sths dbdat))) -;;; (get-sth db sths stmt))) -;;; -;;; ;; write to a db -;;; ;; acfg: area data -;;; ;; rdat: request data -;;; ;; hdat: (host . port) -;;; ;; -;;; ;; (define (dbwrite acfg rdat hdat data-in) -;;; ;; (let* ((dbname (car data-in)) -;;; ;; (dbdat (get-dbh acfg dbname)) -;;; ;; (db (dbdat-dbh dbdat)) -;;; ;; (sths (dbdat-sths dbdat)) -;;; ;; (stmt (calldat-obj rdat)) -;;; ;; (sth (get-sth db sths stmt)) -;;; ;; (data (cdr data-in))) -;;; ;; (print "dbname: " dbname " acfg: " acfg " rdat: " (calldat->alist rdat) " hdat: " hdat " data: " data) -;;; ;; (print "dbdat: " (dbdat->alist dbdat)) -;;; ;; (apply sqlite3:execute sth data) -;;; ;; (save-dbh acfg dbname dbdat) -;;; ;; #t -;;; ;; )) -;;; -;;; (define (finalize-all-db-handles acfg) -;;; (let* ((dbhandles (area-dbhandles acfg)) ;; dbhandles is hash of fname ==> dbdat -;;; (num 0)) -;;; (for-each -;;; (lambda (area-name) -;;; (print "Closing handles for " area-name) -;;; (let ((dbdats (hash-table-ref/default dbhandles area-name '()))) -;;; (for-each -;;; (lambda (dbdat) -;;; ;; first close all statement handles -;;; (for-each -;;; (lambda (sth) -;;; (sqlite3:finalize! sth) -;;; (set! num (+ num 1))) -;;; (hash-table-values (dbdat-sths dbdat))) -;;; ;; now close the dbh -;;; (set! num (+ num 1)) -;;; (sqlite3:finalize! (dbdat-dbh dbdat))) -;;; dbdats))) -;;; (hash-table-keys dbhandles)) -;;; (print "FINALIZED " num " dbhandles"))) -;;; -;;; ;;====================================================================== -;;; ;; W O R K Q U E U E H A N D L I N G -;;; ;;====================================================================== -;;; -;;; (define (register-db-as-mine acfg dbname) -;;; (let ((ht (area-dbs acfg))) -;;; (if (not (hash-table-ref/default ht dbname #f)) -;;; (hash-table-set! ht dbname (random 10000))))) -;;; -;;; (define (work-queue-add acfg fname witem) -;;; (let* ((work-queue-start (current-milliseconds)) -;;; (action (witem-action witem)) ;; NB the action is the index into the rdat actions -;;; (qdat (or (hash-table-ref/default (area-wqueues acfg) fname #f) -;;; (let ((newqdat (make-qdat))) -;;; (hash-table-set! (area-wqueues acfg) fname newqdat) -;;; newqdat))) -;;; (rdat (hash-table-ref/default (area-rtable acfg) action #f))) -;;; (if rdat -;;; (queue-add! -;;; (case (calldat-ctype rdat) -;;; ((dbwrite) (register-db-as-mine acfg fname)(qdat-writeq qdat)) -;;; ((dbread) (register-db-as-mine acfg fname)(qdat-readq qdat)) -;;; ((dbrw) (register-db-as-mine acfg fname)(qdat-rwq qdat)) -;;; ((oslong) (qdat-oslong qdat)) -;;; ((osshort) (qdat-osshort qdat)) -;;; ((full-ping) (qdat-misc qdat)) -;;; (else -;;; (print "ERROR: no queue for " action ". Adding to dbwrite queue.") -;;; (qdat-writeq qdat))) -;;; witem) -;;; (case action -;;; ((full-ping)(qdat-misc qdat)) -;;; (else -;;; (print "ERROR: No action " action " was registered")))) -;;; (sdbg> "work-queue-add" "queue-add" work-queue-start #f #f) -;;; #t)) ;; for now, simply return #t to indicate request got to the queue -;;; -;;; (define (doqueue acfg q fname dbdat dbh) -;;; ;; (print "doqueue: " fname) -;;; (let* ((start-time (current-milliseconds)) -;;; (qlen (queue-length q))) -;;; (if (> qlen 1) -;;; (print "Processing queue of length " qlen)) -;;; (let loop ((count 0) -;;; (responses '())) -;;; (let ((delta (- (current-milliseconds) start-time))) -;;; (if (or (queue-empty? q) -;;; (> delta 400)) ;; stop working on this queue after 400ms have passed -;;; (list count delta responses) ;; return count, delta and responses list -;;; (let* ((witem (queue-remove! q)) -;;; (action (witem-action witem)) -;;; (rdat (witem-rdat witem)) -;;; (stmt (calldat-obj rdat)) -;;; (sth (full-get-sth acfg fname stmt)) -;;; (ctype (calldat-ctype rdat)) -;;; (data (witem-data witem)) -;;; (cookie (witem-cookie witem))) -;;; ;; do the processing and save the result in witem-result -;;; (witem-result-set! -;;; witem -;;; (case ctype ;; action -;;; ((noblockwrite) ;; blind write, no ack of success returned -;;; (apply sqlite3:execute sth data) -;;; (sqlite3:last-insert-rowid dbh)) -;;; ((dbwrite) ;; blocking write -;;; (apply sqlite3:execute sth data) -;;; #t) -;;; ((dbread) ;; TODO: consider breaking this up and shipping in pieces for large query -;;; (apply sqlite3:map-row (lambda x x) sth data)) -;;; ((full-ping) 'full-ping) -;;; (else (print "Not ready for action " action) #f))) -;;; (loop (add1 count) -;;; (if cookie -;;; (cons witem responses) -;;; responses)))))))) -;;; -;;; ;; do up to 400ms of processing on each queue -;;; ;; - the work-queue-processor will allow the max 1200ms of work to complete but it will flag as overloaded -;;; ;; -;;; (define (process-db-queries acfg fname) -;;; (if (hash-table-exists? (area-wqueues acfg) fname) -;;; (let* ((process-db-queries-start-time (current-milliseconds)) -;;; (qdat (hash-table-ref/default (area-wqueues acfg) fname #f)) -;;; (queue-sym->queue (lambda (queue-sym) -;;; (case queue-sym ;; lookup the queue from qdat given a name (symbol) -;;; ((wqueue) (qdat-writeq qdat)) -;;; ((rqueue) (qdat-readq qdat)) -;;; ((rwqueue) (qdat-rwq qdat)) -;;; ((misc) (qdat-misc qdat)) -;;; (else #f)))) -;;; (dbdat (get-dbh acfg fname)) -;;; (dbh (if (dbdat? dbdat)(dbdat-dbh dbdat) #f)) -;;; (nowtime (current-seconds))) -;;; ;; handle the queues that require a transaction -;;; ;; -;;; (map ;; -;;; (lambda (queue-sym) -;;; ;; (print "processing queue " queue-sym) -;;; (let* ((queue (queue-sym->queue queue-sym))) -;;; (if (not (queue-empty? queue)) -;;; (let ((responses -;;; (sqlite3:with-transaction ;; todo - catch exceptions... -;;; dbh -;;; (lambda () -;;; (let* ((res (doqueue acfg queue fname dbdat dbh))) ;; this does the work! -;;; ;; (print "res=" res) -;;; (match res -;;; ((count delta responses) -;;; (update-stats acfg fname queue-sym delta count) -;;; (sdbg> "process-db-queries" "sqlite3-transaction" process-db-queries-start-time #f #f) -;;; responses) ;; return responses -;;; (else -;;; (print "ERROR: bad return data from doqueue " res))) -;;; ))))) -;;; ;; having completed the transaction, send the responses. -;;; ;; (print "INFO: sending " (length responses) " responses.") -;;; (let loop ((responses-left responses)) -;;; (cond -;;; ((null? responses-left) #t) -;;; (else -;;; (let* ((witem (car responses-left)) -;;; (response (cdr responses-left))) -;;; (call-deliver-response acfg (witem-ripaddr witem)(witem-rport witem) -;;; (witem-cookie witem)(witem-result witem))) -;;; (loop (cdr responses-left)))))) -;;; ))) -;;; '(wqueue rwqueue rqueue)) -;;; -;;; ;; handle misc queue -;;; ;; -;;; ;; (print "processing misc queue") -;;; (let ((queue (queue-sym->queue 'misc))) -;;; (doqueue acfg queue fname dbdat dbh)) -;;; ;; .... -;;; (save-dbh acfg fname dbdat) -;;; #t ;; just to let the tests know we got here -;;; ) -;;; #f ;; nothing processed -;;; )) -;;; -;;; ;; run all queues in parallel per db but sequentially per queue for that db. -;;; ;; - process the queues every 500 or so ms -;;; ;; - allow for long running queries to continue but all other activities for that -;;; ;; db will be blocked. -;;; ;; -;;; (define (work-queue-processor acfg) -;;; (let* ((threads (make-hash-table))) ;; fname => thread -;;; (let loop ((fnames (hash-table-keys (area-wqueues acfg))) -;;; (target-time (+ (current-milliseconds) 50))) -;;; ;;(if (not (null? fnames))(print "Processing for these databases: " fnames)) -;;; (for-each -;;; (lambda (fname) -;;; ;; (print "processing for " fname) -;;; ;;(process-db-queries acfg fname)) -;;; (let ((th (hash-table-ref/default threads fname #f))) -;;; (if (and th (not (member (thread-state th) '(dead terminated)))) -;;; (begin -;;; (print "WARNING: worker thread for " fname " is taking a long time.") -;;; (print "Thread is in state " (thread-state th))) -;;; (let ((th1 (make-thread (lambda () -;;; ;; (catch-and-dump -;;; ;; (lambda () -;;; ;; (print "Process queries for " fname) -;;; (let ((start-time (current-milliseconds))) -;;; (process-db-queries acfg fname) -;;; ;; (thread-sleep! 0.01) ;; need the thread to take at least some time -;;; (hash-table-delete! threads fname)) ;; no mutexes? -;;; fname) -;;; "th1"))) ;; )) -;;; (hash-table-set! threads fname th1) -;;; (thread-start! th1))))) -;;; fnames) -;;; ;; (thread-sleep! 0.1) ;; give the threads some time to process requests -;;; ;; burn time until 400ms is up -;;; (let ((now-time (current-milliseconds))) -;;; (if (< now-time target-time) -;;; (let ((delta (- target-time now-time))) -;;; (thread-sleep! (/ delta 1000))))) -;;; (loop (hash-table-keys (area-wqueues acfg)) -;;; (+ (current-milliseconds) 50))))) -;;; -;;; ;;====================================================================== -;;; ;; S T A T S G A T H E R I N G -;;; ;;====================================================================== -;;; -;;; (defstruct stat -;;; (qcount-avg 0) ;; coarse running average -;;; (qtime-avg 0) ;; coarse running average -;;; (qcount 0) ;; total -;;; (qtime 0) ;; total -;;; (last-qcount 0) ;; last -;;; (last-qtime 0) ;; last -;;; (dbs '()) ;; list of db files handled by this node -;;; (when 0)) ;; when the last query happened - seconds -;;; -;;; -;;; (define (update-stats acfg fname bucket duration numqueries) -;;; (let* ((key fname) ;; for now do not use bucket. Was: (conc fname "-" bucket)) ;; lazy but good enough -;;; (stats (or (hash-table-ref/default (area-stats acfg) key #f) -;;; (let ((newstats (make-stat))) -;;; (hash-table-set! (area-stats acfg) key newstats) -;;; newstats)))) -;;; ;; when the last query happended (used to remove the fname from the active list) -;;; (stat-when-set! stats (current-seconds)) -;;; ;; last values -;;; (stat-last-qcount-set! stats numqueries) -;;; (stat-last-qtime-set! stats duration) -;;; ;; total over process lifetime -;;; (stat-qcount-set! stats (+ (stat-qcount stats) numqueries)) -;;; (stat-qtime-set! stats (+ (stat-qtime stats) duration)) -;;; ;; coarse average -;;; (stat-qcount-avg-set! stats (/ (+ (stat-qcount-avg stats) numqueries) 2)) -;;; (stat-qtime-avg-set! stats (/ (+ (stat-qtime-avg stats) duration) 2)) -;;; -;;; ;; here is where we add the stats for a given dbfile -;;; (if (not (member fname (stat-dbs stats))) -;;; (stat-dbs-set! stats (cons fname (stat-dbs stats)))) -;;; -;;; )) -;;; -;;; ;;====================================================================== -;;; ;; S E R V E R S T U F F -;;; ;;====================================================================== -;;; -;;; ;; this does NOT return! -;;; ;; -;;; (define (find-free-port-and-open acfg) -;;; (let ((port (or (area-port acfg) 3200))) -;;; (handle-exceptions -;;; exn -;;; (begin -;;; (print "INFO: cannot bind to port " (rpc:default-server-port) ", trying next port") -;;; (area-port-set! acfg (+ port 1)) -;;; (find-free-port-and-open acfg)) -;;; (rpc:default-server-port port) -;;; (area-port-set! acfg port) -;;; (tcp-read-timeout 120000) -;;; ;; ((rpc:make-server (tcp-listen port)) #t) -;;; (tcp-listen (rpc:default-server-port) -;;; )))) -;;; -;;; ;; register this node by putting a packet into the pkts dir. -;;; ;; look for other servers -;;; ;; contact other servers and compile list of servers -;;; ;; there are two types of server -;;; ;; main servers - dashboards, runners and dedicated servers - need pkt -;;; ;; passive servers - test executers, step calls, list-runs - no pkt -;;; ;; -;;; (define (register-node acfg hostip port-num) -;;; ;;(mutex-lock! (area-mutex acfg)) -;;; (let* ((server-type (area-server-type acfg)) ;; auto, main, passive (no pkt created) -;;; (best-ip (or hostip (get-my-best-address))) -;;; (mtdir (area-dbdir acfg)) -;;; (pktdir (area-pktsdir acfg))) ;; conc mtdir "/.server-pkts"))) -;;; (print "Registering node " best-ip ":" port-num) -;;; (if (not mtdir) ;; require a home for this node to put or find databases -;;; #f -;;; (begin -;;; (if (not (directory? pktdir))(create-directory pktdir)) -;;; ;; server is started, now create pkt if needed -;;; (print "Starting server in " server-type " mode with port " port-num) -;;; (if (member server-type '(auto main)) ;; TODO: if auto, count number of servers registers, if > 3 then don't put out a pkt -;;; (begin -;;; (area-pktid-set! acfg -;;; (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! acfg (conc pktdir "/" (area-pktid acfg) ".pkt")))) -;;; (area-port-set! acfg port-num) -;;; #;(mutex-unlock! (area-mutex acfg)))))) -;;; -;;; (define *cookie-seqnum* 0) -;;; (define (make-cookie key) -;;; (set! *cookie-seqnum* (add1 *cookie-seqnum*)) -;;; ;;(print "MAKE COOKIE CALLED -- on "servkey"-"*cookie-seqnum*) -;;; (conc key "-" *cookie-seqnum*) -;;; ) -;;; -;;; ;; dispatch locally if possible -;;; ;; -;;; (define (call-deliver-response acfg ipaddr port cookie data) -;;; (if (and (equal? (area-myaddr acfg) ipaddr) -;;; (equal? (area-port acfg) port)) -;;; (deliver-response acfg cookie data) -;;; ((rpc:procedure 'response ipaddr port) cookie data))) -;;; -;;; (define (deliver-response acfg cookie data) -;;; (let ((deliver-response-start (current-milliseconds))) -;;; (thread-start! (make-thread -;;; (lambda () -;;; (let loop ((tries-left 5)) -;;; ;;(print "TOP OF DELIVER_RESPONSE LOOP; triesleft="tries-left) -;;; ;;(pp (hash-table->alist (area-cookie2mbox acfg))) -;;; (let* ((mbox (hash-table-ref/default (area-cookie2mbox acfg) cookie #f))) -;;; (cond -;;; ((eq? 0 tries-left) -;;; (print "ulex:deliver-response: I give up. Mailbox never appeared. cookie="cookie) -;;; ) -;;; (mbox -;;; ;;(print "got mbox="mbox" got data="data" send.") -;;; (mailbox-send! mbox data)) -;;; (else -;;; ;;(print "no mbox yet. look for "cookie) -;;; (thread-sleep! (/ (- 6 tries-left) 10)) -;;; (loop (sub1 tries-left)))))) -;;; ;; (debug-pp (list (conc "ulex:deliver-response took " (- (current-milliseconds) deliver-response-start) " ms, cookie=" cookie " data=") data)) -;;; (sdbg> "deliver-response" "mailbox-send" deliver-response-start #f #f cookie) -;;; ) -;;; (conc "deliver-response thread for cookie="cookie)))) -;;; #t) -;;; -;;; ;; action: -;;; ;; immediate - quick actions, no need to put in queues -;;; ;; dbwrite - put in dbwrite queue -;;; ;; dbread - put in dbread queue -;;; ;; oslong - os actions, e.g. du, that could take a long time -;;; ;; osshort - os actions that should be quick, e.g. df -;;; ;; -;;; (define (request acfg from-ipaddr from-port servkey action cookie fname params) ;; std-peer-handler -;;; ;; NOTE: Use rpc:current-peer for getting return address -;;; (let* ((std-peer-handler-start (current-milliseconds)) -;;; ;; (raw-data (alist-ref 'data dat)) -;;; (rdat (hash-table-ref/default -;;; (area-rtable acfg) action #f)) ;; this looks up the sql query or other details indexed by the action -;;; (witem (make-witem ripaddr: from-ipaddr ;; rhost: from-host -;;; rport: from-port action: action -;;; rdat: rdat cookie: cookie -;;; servkey: servkey data: params ;; TODO - rename data to params -;;; caller: (rpc:current-peer)))) -;;; (if (not (equal? servkey (area-pktid acfg))) -;;; `(#f . ,(conc "I don't know you servkey=" servkey ", pktid=" (area-pktid acfg))) ;; immediately return this -;;; (let* ((ctype (if rdat -;;; (calldat-ctype rdat) ;; is this necessary? these should be identical -;;; action))) -;;; (sdbg> "std-peer-handler" "immediate" std-peer-handler-start #f #f) -;;; (case ctype -;;; ;; (dbwrite acfg rdat (cons from-ipaddr from-port) data))) -;;; ((full-ping) `(#t "ack to full ping" ,(work-queue-add acfg fname witem) ,cookie)) -;;; ((response) `(#t "ack from requestor" ,(deliver-response acfg fname params))) -;;; ((dbwrite) `(#t "db write submitted" ,(work-queue-add acfg fname witem) ,cookie)) -;;; ((dbread) `(#t "db read submitted" ,(work-queue-add acfg fname witem) ,cookie )) -;;; ((dbrw) `(#t "db read/write submitted" ,cookie)) -;;; ((osshort) `(#t "os short submitted" ,cookie)) -;;; ((oslong) `(#t "os long submitted" ,cookie)) -;;; (else `(#f "unrecognised action" ,ctype))))))) -;;; -;;; ;; Call this to start the actual server -;;; ;; -;;; ;; start_server -;;; ;; -;;; ;; mode: ' -;;; ;; handler: proc which takes pktrecieved as argument -;;; ;; -;;; -;;; (define (start-server acfg) -;;; (let* ((conn (find-free-port-and-open acfg)) -;;; (port (area-port acfg))) -;;; (rpc:publish-procedure! -;;; 'delist-db -;;; (lambda (fname) -;;; (hash-table-delete! (area-dbs acfg) fname))) -;;; (rpc:publish-procedure! -;;; 'calling-addr -;;; (lambda () -;;; (rpc:current-peer))) -;;; (rpc:publish-procedure! -;;; 'ping -;;; (lambda ()(real-ping acfg))) -;;; (rpc:publish-procedure! -;;; 'request -;;; (lambda (from-addr from-port servkey action cookie dbname params) -;;; (request acfg from-addr from-port servkey action cookie dbname params))) -;;; (rpc:publish-procedure! -;;; 'response -;;; (lambda (cookie res-dat) -;;; (deliver-response acfg cookie res-dat))) -;;; (area-ready-set! acfg #t) -;;; (area-conn-set! acfg conn) -;;; ((rpc:make-server conn) #f)));; ((tcp-listen (rpc:default-server-port)) #t) -;;; -;;; -;;; (define (launch acfg) ;; #!optional (proc std-peer-handler)) -;;; (print "starting launch") -;;; (update-known-servers acfg) ;; gotta do this on every start (thus why limit number of publicised servers) -;;; #;(let ((original-handler (current-exception-handler))) ;; is th -;;; (lambda (exception) -;;; (server-exit-procedure) -;;; (original-handler exception))) -;;; (on-exit (lambda () -;;; (shutdown acfg))) ;; (finalize-all-db-handles acfg))) -;;; ;; set up the rpc handler -;;; (let* ((th1 (make-thread -;;; (lambda ()(start-server acfg)) -;;; "server thread")) -;;; (th2 (make-thread -;;; (lambda () -;;; (print "th2 starting") -;;; (let loop () -;;; (work-queue-processor acfg) -;;; (print "work-queue-processor crashed!") -;;; (loop))) -;;; "work queue thread"))) -;;; (thread-start! th1) -;;; (thread-start! th2) -;;; (let loop () -;;; (thread-sleep! 0.025) -;;; (if (area-ready acfg) -;;; #t -;;; (loop))) -;;; ;; attempt to fix my address -;;; (let* ((all-addr (get-all-ips-sorted))) ;; could use (tcp-addresses conn)? -;;; (let loop ((rem-addrs all-addr)) -;;; (if (null? rem-addrs) -;;; (begin -;;; (print "ERROR: Failed to figure out the ip address of myself as a server. Giving up.") -;;; (exit 1)) ;; BUG Changeme to raising an exception -;;; -;;; (let* ((addr (car rem-addrs)) -;;; (good-addr (handle-exceptions -;;; exn -;;; #f -;;; ((rpc:procedure 'calling-addr addr (area-port acfg)))))) -;;; (if good-addr -;;; (begin -;;; (print "Got good-addr of " good-addr) -;;; (area-myaddr-set! acfg good-addr)) -;;; (loop (cdr rem-addrs))))))) -;;; (register-node acfg (area-myaddr acfg)(area-port acfg)) -;;; (print "INFO: Server started on " (area-myaddr acfg) ":" (area-port acfg)) -;;; ;; (update-known-servers acfg) ;; gotta do this on every start (thus why limit number of publicised servers) -;;; )) -;;; -;;; (define (clear-server-pkt acfg) -;;; (let ((pktf (area-pktfile acfg))) -;;; (if pktf (delete-file* pktf)))) -;;; -;;; (define (shutdown acfg) -;;; (let (;;(conn (area-conn acfg)) -;;; (pktf (area-pktfile acfg)) -;;; (port (area-port acfg))) -;;; (if pktf (delete-file* pktf)) -;;; (send-all "imshuttingdown") -;;; ;; (rpc:close-all-connections!) ;; don't know if this is actually needed -;;; (finalize-all-db-handles acfg))) -;;; -;;; (define (send-all msg) -;;; #f) -;;; -;;; ;; given a area record look up all the packets -;;; ;; -;;; (define (get-all-server-pkts acfg) -;;; (let ((all-pkt-files (glob (conc (area-pktsdir acfg) "/*.pkt")))) -;;; (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")) -;;; -;;; #;(define (get-my-best-address) -;;; (let ((all-my-addresses (get-all-ips))) ;; (vector->list (hostinfo-addresses (hostname->hostinfo (get-host-name)))))) -;;; (cond -;;; ((null? all-my-addresses) -;;; (get-host-name)) ;; no interfaces? -;;; ((eq? (length all-my-addresses) 1) -;;; (ip->string (car all-my-addresses))) ;; only one to choose from, just go with it -;;; (else -;;; (ip->string (car (filter (lambda (x) ;; take any but 127. -;;; (not (eq? (u8vector-ref x 0) 127))) -;;; all-my-addresses))))))) -;;; -;;; ;; whoami? I am my pkt -;;; ;; -;;; (define (whoami? acfg) -;;; (hash-table-ref/default (area-hosts acfg)(area-pktid acfg) #f)) -;;; -;;; ;;====================================================================== -;;; ;; "Client side" operations -;;; ;;====================================================================== -;;; -;;; (define (safe-call call-key host port . params) -;;; (handle-exceptions -;;; exn -;;; (begin -;;; (print "Call " call-key " to " host ":" port " failed") -;;; #f) -;;; (apply (rpc:procedure call-key host port) params))) -;;; -;;; ;; ;; 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 (ping acfg host port) -;;; (let* ((myaddr (area-myaddr acfg)) -;;; (myport (area-port acfg)) -;;; (start-time (current-milliseconds)) -;;; (res (if (and (equal? myaddr host) -;;; (equal? myport port)) -;;; (real-ping acfg) -;;; ((rpc:procedure 'ping host port))))) -;;; (cons (- (current-milliseconds) start-time) -;;; res))) -;;; -;;; ;; returns ( ipaddr port alist-fname=>randnum ) -;;; (define (real-ping acfg) -;;; `(,(area-myaddr acfg) ,(area-port acfg) ,(get-host-stats acfg))) -;;; -;;; ;; is the server alive AND the queues processing? -;;; ;; -;;; #;(define (full-ping acfg servpkt) -;;; (let* ((start-time (current-milliseconds)) -;;; (res (send-message acfg servpkt '(full-ping) 'full-ping))) -;;; (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 acfg -;;; ;; return the number of responsive servers found -;;; ;; -;;; ;; DO NOT VERIFY THAT THE SERVER IS ALIVE HERE. This is called at times where the current server is not yet alive and cannot ping itself -;;; ;; -;;; (define (update-known-servers acfg) -;;; ;; readll all pkts -;;; ;; foreach pkt; if it isn't me ping the server; if alive, add to hosts hash, else rm the pkt -;;; (let* ((start-time (current-milliseconds)) -;;; (all-pkts (delete-duplicates -;;; (append (get-all-server-pkts acfg) -;;; (hash-table-values (area-hosts acfg))))) -;;; (hostshash (area-hosts acfg)) -;;; (my-id (area-pktid acfg)) -;;; (pktsdir (area-pktsdir acfg)) ;; needed to remove pkts from non-responsive servers -;;; (numsrvs 0) -;;; (delpkt (lambda (pktsdir sid) -;;; (print "clearing out server " sid) -;;; (delete-file* (conc pktsdir "/" sid ".pkt")) -;;; (hash-table-delete! hostshash sid)))) -;;; (area-last-srvup-set! acfg (current-seconds)) -;;; (for-each -;;; (lambda (servpkt) -;;; (if (list? servpkt) -;;; ;; (pp servpkt) -;;; (let* ((shost (alist-ref 'ipaddr servpkt)) -;;; (sport (any->number (alist-ref 'port servpkt))) -;;; (res (handle-exceptions -;;; exn -;;; (begin -;;; ;; (print "INFO: bad server on " shost ":" sport) -;;; #f) -;;; (ping acfg shost sport))) -;;; (sid (alist-ref 'Z servpkt)) ;; Z code is our name for the server -;;; (url (conc shost ":" sport)) -;;; ) -;;; #;(if (or (not res) -;;; (null? res)) -;;; (begin -;;; (print "STRANGE: ping of " url " gave " res))) -;;; -;;; ;; (print "Got " res " from " shost ":" sport) -;;; (match res -;;; ((qduration . payload) -;;; ;; (print "Server pkt:" (alist-ref 'ipaddr servpkt) ":" (alist-ref 'port servpkt) -;;; ;; (if payload -;;; ;; "Success" "Fail")) -;;; (match payload -;;; ((host port stats) -;;; ;; (print "From " host ":" port " got stats: " stats) -;;; (if (and host port stats) -;;; (let ((url (conc host ":" port))) -;;; (hash-table-set! hostshash sid servpkt) -;;; ;; store based on host:port -;;; (hash-table-set! (area-hoststats acfg) sid stats)) -;;; (print "missing data from the server, not sure what that means!")) -;;; (set! numsrvs (+ numsrvs 1))) -;;; (#f -;;; (print "Removing pkt " sid " due to #f from server or failed ping") -;;; (delpkt pktsdir sid)) -;;; (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 -;;; (let ((bad-pings (hash-table-ref/default (area-health acfg) url 0))) -;;; (if (> bad-pings 1) ;; two bad pings - remove pkt -;;; (begin -;;; (print "INFO: " bad-pings " bad responses from " url ", deleting pkt " sid) -;;; (delpkt pktsdir sid)) -;;; (begin -;;; (print "INFO: " bad-pings " bad responses from " shost ":" sport " not deleting pkt yet") -;;; (hash-table-set! (area-health acfg) -;;; url -;;; (+ (hash-table-ref/default (area-health acfg) url 0) 1)) -;;; )) -;;; )))) -;;; ;; servpkt is not actually a pkt? -;;; (begin -;;; (print "Bad pkt " servpkt)))) -;;; all-pkts) -;;; (sdbg> "update-known-servers" "end" start-time #f #f " found " numsrvs -;;; " servers, pkts: " (map (lambda (p) -;;; (alist-ref 'Z p)) -;;; all-pkts)) -;;; numsrvs)) -;;; -;;; (defstruct srvstat -;;; (numfiles 0) ;; number of db files handled by this server - subtract 1 for the db being currently looked at -;;; (randnum #f) ;; tie breaker number assigned to by the server itself - applies only to the db under consideration -;;; (pkt #f)) ;; the server pkt -;;; -;;; ;;(define (srv->srvstat srvpkt) -;;; -;;; ;; Get the server best for given dbname and key -;;; ;; -;;; ;; NOTE: key is not currently used. The key points to the kind of query, this may be useful for directing read-only queries. -;;; ;; -;;; (define (get-best-server acfg dbname key) -;;; (let* (;; (servers (hash-table-values (area-hosts acfg))) -;;; (servers (area-hosts acfg)) -;;; (skeys (sort (hash-table-keys servers) string>=?)) ;; a stable listing -;;; (start-time (current-milliseconds)) -;;; (srvstats (make-hash-table)) ;; srvid => srvstat -;;; (url (conc (area-myaddr acfg) ":" (area-port acfg)))) -;;; ;; (print "scores for " dbname ": " (map (lambda (k)(cons k (calc-server-score acfg dbname k))) skeys)) -;;; (if (null? skeys) -;;; (if (> (update-known-servers acfg) 0) -;;; (get-best-server acfg dbname key) ;; some risk of infinite loop here, TODO add try counter -;;; (begin -;;; (print "ERROR: no server found!") ;; since this process is also a server this should never happen -;;; #f)) -;;; (begin -;;; ;; (print "in get-best-server with skeys=" skeys) -;;; (if (> (- (current-seconds) (area-last-srvup acfg)) 10) -;;; (begin -;;; (update-known-servers acfg) -;;; (sdbg> "get-best-server" "update-known-servers" start-time #f #f))) -;;; -;;; ;; for each server look at the list of dbfiles, total number of dbs being handled -;;; ;; and the rand number, save the best host -;;; ;; also do a delist-db for each server dbfile not used -;;; (let* ((best-server #f) -;;; (servers-to-delist (make-hash-table))) -;;; (for-each -;;; (lambda (srvid) -;;; (let* ((server (hash-table-ref/default servers srvid #f)) -;;; (stats (hash-table-ref/default (area-hoststats acfg) srvid '(())))) -;;; ;; (print "stats: " stats) -;;; (if server -;;; (let* ((dbweights (car stats)) -;;; (srvload (length (filter (lambda (x)(not (equal? dbname (car x)))) dbweights))) -;;; (dbrec (alist-ref dbname dbweights equal?)) ;; get the pair with fname . randscore -;;; (randnum (if dbrec -;;; dbrec ;; (cdr dbrec) -;;; 0))) -;;; (hash-table-set! srvstats srvid (make-srvstat numfiles: srvload randnum: randnum pkt: server)))))) -;;; skeys) -;;; -;;; (let* ((sorted (sort (hash-table-values srvstats) -;;; (lambda (a b) -;;; (let ((numfiles-a (srvstat-numfiles a)) -;;; (numfiles-b (srvstat-numfiles b)) -;;; (randnum-a (srvstat-randnum a)) -;;; (randnum-b (srvstat-randnum b))) -;;; (if (< numfiles-a numfiles-b) ;; Note, I don't think adding an offset works here. Goal was only move file handling to a different server if it has 2 less -;;; #t -;;; (if (and (equal? numfiles-a numfiles-b) -;;; (< randnum-a randnum-b)) -;;; #t -;;; #f)))))) -;;; (best (if (null? sorted) -;;; (begin -;;; (print "ERROR: should never be null due to self as server.") -;;; #f) -;;; (srvstat-pkt (car sorted))))) -;;; #;(print "SERVER(" url "): " dbname ": " (map (lambda (srv) -;;; (let ((p (srvstat-pkt srv))) -;;; (conc (alist-ref 'ipaddr p) ":" (alist-ref 'port p) -;;; "(" (srvstat-numfiles srv)","(srvstat-randnum srv)")"))) -;;; sorted)) -;;; best)))))) -;;; -;;; ;; send out an "I'm about to exit notice to all known servers" -;;; ;; -;;; (define (death-imminent acfg) -;;; '()) -;;; -;;; ;;====================================================================== -;;; ;; U L E X - T H E I N T E R E S T I N G S T U F F ! ! -;;; ;;====================================================================== -;;; -;;; ;; register a handler -;;; ;; NOTES: -;;; ;; dbinitsql is reserved for a list of sql statements for initializing the db -;;; ;; dbinitfn is reserved for a db init function, if exists called after dbinitsql -;;; ;; -;;; (define (register acfg key obj #!optional (ctype 'dbwrite)) -;;; (let ((ht (area-rtable acfg))) -;;; (if (hash-table-exists? ht key) -;;; (print "WARNING: redefinition of entry " key)) -;;; (hash-table-set! ht key (make-calldat obj: obj ctype: ctype)))) -;;; -;;; ;; usage: register-batch acfg '((key1 . sql1) (key2 . sql2) ... ) -;;; ;; NB// obj is often an sql query -;;; ;; -;;; (define (register-batch acfg ctype data) -;;; (let ((ht (area-rtable acfg))) -;;; (map (lambda (dat) -;;; (hash-table-set! ht (car dat)(make-calldat obj: (cdr dat) ctype: ctype))) -;;; data))) -;;; -;;; (define (initialize-area-calls-from-specfile area specfile) -;;; (let* ((callspec (with-input-from-file specfile read ))) -;;; (for-each (lambda (group) -;;; (register-batch -;;; area -;;; (car group) -;;; (cdr group))) -;;; callspec))) -;;; -;;; ;; get-rentry -;;; ;; -;;; (define (get-rentry acfg key) -;;; (hash-table-ref/default (area-rtable acfg) key #f)) -;;; -;;; (define (get-rsql acfg key) -;;; (let ((cdat (get-rentry acfg key))) -;;; (if cdat -;;; (calldat-obj cdat) -;;; #f))) -;;; -;;; -;;; -;;; ;; blocking call: -;;; ;; client server -;;; ;; ------ ------ -;;; ;; call() -;;; ;; send-message() -;;; ;; nmsg-send() -;;; ;; nmsg-receive() -;;; ;; nmsg-respond(ack,cookie) -;;; ;; ack, cookie -;;; ;; mbox-thread-wait(cookie) -;;; ;; nmsg-send(client,cookie,result) -;;; ;; nmsg-respond(ack) -;;; ;; return result -;;; ;; -;;; ;; reserved action: -;;; ;; 'immediate -;;; ;; 'dbinitsql -;;; ;; -;;; (define (call acfg dbname action params #!optional (count 0)) -;;; (let* ((call-start-time (current-milliseconds)) -;;; (srv (get-best-server acfg dbname action)) -;;; (post-get-start-time (current-milliseconds)) -;;; (rdat (hash-table-ref/default (area-rtable acfg) action #f)) -;;; (myid (trim-pktid (area-pktid acfg))) -;;; (srvid (trim-pktid (alist-ref 'Z srv))) -;;; (cookie (make-cookie myid))) -;;; (sdbg> "call" "get-best-server" call-start-time #f call-start-time " from: " myid " to server: " srvid " for " dbname " action: " action " params: " params " rdat: " rdat) -;;; (print "INFO: call to " (alist-ref 'ipaddr srv) ":" (alist-ref 'port srv) " from " (area-myaddr acfg) ":" (area-port acfg) " for " dbname) -;;; (if (and srv rdat) ;; need both to dispatch a request -;;; (let* ((ripaddr (alist-ref 'ipaddr srv)) -;;; (rsrvid (alist-ref 'Z srv)) -;;; (rport (any->number (alist-ref 'port srv))) -;;; (res-full (if (and (equal? ripaddr (area-myaddr acfg)) -;;; (equal? rport (area-port acfg))) -;;; (request acfg ripaddr rport (area-pktid acfg) action cookie dbname params) -;;; (safe-call 'request ripaddr rport -;;; (area-myaddr acfg) -;;; (area-port acfg) -;;; #;(area-pktid acfg) -;;; rsrvid -;;; action cookie dbname params)))) -;;; ;; (print "res-full: " res-full) -;;; (match res-full -;;; ((response-ok response-msg rem ...) -;;; (let* ((send-message-time (current-milliseconds)) -;;; ;; (match res-full -;;; ;; ((response-ok response-msg) -;;; ;; (response-ok (car res-full)) -;;; ;; (response-msg (cadr res-full) -;;; ) -;;; ;; (res (take res-full 3))) ;; ctype == action, TODO: converge on one term <<=== what was this? BUG -;;; ;; (print "ulex:call: send-message took " (- send-message-time post-get-start-time) " ms params=" params) -;;; (sdbg> "call" "send-message" post-get-start-time #f call-start-time) -;;; (cond -;;; ((not response-ok) #f) -;;; ((member response-msg '("db read submitted" "db write submitted")) -;;; (let* ((cookie-id (cadddr res-full)) -;;; (mbox (make-mailbox)) -;;; (mbox-time (current-milliseconds))) -;;; (hash-table-set! (area-cookie2mbox acfg) cookie-id mbox) -;;; (let* ((mbox-timeout-secs 20) -;;; (mbox-timeout-result 'MBOX_TIMEOUT) -;;; (res (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result)) -;;; (mbox-receive-time (current-milliseconds))) -;;; (hash-table-delete! (area-cookie2mbox acfg) cookie-id) -;;; (sdbg> "call" "mailbox-receive" mbox-time #f call-start-time " from: " myid " to server: " srvid " for " dbname) -;;; ;; (print "ulex:call mailbox-receive took " (- mbox-receive-time mbox-time) "ms params=" params) -;;; res))) -;;; (else -;;; (print "Unhandled response \""response-msg"\"") -;;; #f)) -;;; ;; depending on what action (i.e. ctype) is we will block here waiting for -;;; ;; all the data (mechanism to be determined) -;;; ;; -;;; ;; if res is a "working on it" then wait -;;; ;; wait for result -;;; ;; mailbox thread wait on -;;; -;;; ;; if res is a "can't help you" then try a different server -;;; ;; if res is a "ack" (e.g. for one-shot requests) then return res -;;; )) -;;; (else -;;; (if (< count 10) -;;; (let* ((url (conc (alist-ref 'ipaddr srv) ":" (alist-ref 'port srv)))) -;;; (thread-sleep! 1) -;;; (print "ERROR: Bad result from " url ", dbname: " dbname ", action: " action ", params: " params ". Trying again in 1 second.") -;;; (call acfg dbname action params (+ count 1))) -;;; (begin -;;; (error (conc "ERROR: " count " tries, still have improper response res-full=" res-full))))))) -;;; (begin -;;; (if (not rdat) -;;; (print "ERROR: action " action " not registered.") -;;; (if (< count 10) -;;; (begin -;;; (thread-sleep! 1) -;;; (area-hosts-set! acfg (make-hash-table)) ;; clear out all known hosts -;;; (print "ERROR: no server found, srv=" srv ", trying again in 1 seconds") -;;; (call acfg dbname action params (+ count 1))) -;;; (begin -;;; (error (conc "ERROR: no server found after 10 tries, srv=" srv ", giving up.")) -;;; #;(error "No server available")))))))) -;;; -;;; -;;; ;;====================================================================== -;;; ;; U T I L I T I E S -;;; ;;====================================================================== -;;; -;;; ;; get a signature for identifing this process -;;; ;; -;;; (define (get-process-signature) -;;; (cons (get-host-name)(current-process-id))) -;;; -;;; ;;====================================================================== -;;; ;; S Y S T E M S T U F F -;;; ;;====================================================================== -;;; -;;; ;; get normalized cpu load by reading from /proc/loadavg and -;;; ;; /proc/cpuinfo return all three values and the number of real cpus -;;; ;; and the number of threads returns alist '((adj-cpu-load -;;; ;; . normalized-proc-load) ... etc. keys: adj-proc-load, -;;; ;; adj-core-load, 1m-load, 5m-load, 15m-load -;;; ;; -;;; (define (get-normalized-cpu-load) -;;; (let ((res (get-normalized-cpu-load-raw)) -;;; (default `((adj-proc-load . 2) ;; there is no right answer -;;; (adj-core-load . 2) -;;; (1m-load . 2) -;;; (5m-load . 0) ;; causes a large delta - thus causing default of throttling if stuff goes wrong -;;; (15m-load . 0) -;;; (proc . 1) -;;; (core . 1) -;;; (phys . 1) -;;; (error . #t)))) -;;; (cond -;;; ((and (list? res) -;;; (> (length res) 2)) -;;; res) -;;; ((eq? res #f) default) ;; add messages? -;;; ((eq? res #f) default) ;; this would be the #eof -;;; (else default)))) -;;; -;;; (define (get-normalized-cpu-load-raw) -;;; (let* ((actual-host (get-host-name))) ;; #f is localhost -;;; (let ((data (append -;;; (with-input-from-file "/proc/loadavg" read-lines) -;;; (with-input-from-file "/proc/cpuinfo" read-lines) -;;; (list "end"))) -;;; (load-rx (regexp "^([\\d\\.]+)\\s+([\\d\\.]+)\\s+([\\d\\.]+)\\s+.*$")) -;;; (proc-rx (regexp "^processor\\s+:\\s+(\\d+)\\s*$")) -;;; (core-rx (regexp "^core id\\s+:\\s+(\\d+)\\s*$")) -;;; (phys-rx (regexp "^physical id\\s+:\\s+(\\d+)\\s*$")) -;;; (max-num (lambda (p n)(max (string->number p) n)))) -;;; ;; (print "data=" data) -;;; (if (null? data) ;; something went wrong -;;; #f -;;; (let loop ((hed (car data)) -;;; (tal (cdr data)) -;;; (loads #f) -;;; (proc-num 0) ;; processor includes threads -;;; (phys-num 0) ;; physical chip on motherboard -;;; (core-num 0)) ;; core -;;; ;; (print hed ", " loads ", " proc-num ", " phys-num ", " core-num) -;;; (if (null? tal) ;; have all our data, calculate normalized load and return result -;;; (let* ((act-proc (+ proc-num 1)) -;;; (act-phys (+ phys-num 1)) -;;; (act-core (+ core-num 1)) -;;; (adj-proc-load (/ (car loads) act-proc)) -;;; (adj-core-load (/ (car loads) act-core)) -;;; (result -;;; (append (list (cons 'adj-proc-load adj-proc-load) -;;; (cons 'adj-core-load adj-core-load)) -;;; (list (cons '1m-load (car loads)) -;;; (cons '5m-load (cadr loads)) -;;; (cons '15m-load (caddr loads))) -;;; (list (cons 'proc act-proc) -;;; (cons 'core act-core) -;;; (cons 'phys act-phys))))) -;;; result) -;;; (regex-case -;;; hed -;;; (load-rx ( x l1 l5 l15 ) (loop (car tal)(cdr tal)(map string->number (list l1 l5 l15)) proc-num phys-num core-num)) -;;; (proc-rx ( x p ) (loop (car tal)(cdr tal) loads (max-num p proc-num) phys-num core-num)) -;;; (phys-rx ( x p ) (loop (car tal)(cdr tal) loads proc-num (max-num p phys-num) core-num)) -;;; (core-rx ( x c ) (loop (car tal)(cdr tal) loads proc-num phys-num (max-num c core-num))) -;;; (else -;;; (begin -;;; ;; (print "NO MATCH: " hed) -;;; (loop (car tal)(cdr tal) loads proc-num phys-num core-num)))))))))) -;;; -;;; (define (get-host-stats acfg) -;;; (let ((stats-hash (area-stats acfg))) -;;; ;; use this opportunity to remove references to dbfiles which have not been accessed in a while -;;; (for-each -;;; (lambda (dbname) -;;; (let* ((stats (hash-table-ref stats-hash dbname)) -;;; (last-access (stat-when stats))) -;;; (if (and (> last-access 0) ;; if zero then there has been no access -;;; (> (- (current-seconds) last-access) 10)) ;; not used in ten seconds -;;; (begin -;;; (print "Removing " dbname " from stats list") -;;; (hash-table-delete! stats-hash dbname) ;; remove from stats hash -;;; (stat-dbs-set! stats (hash-table-keys stats)))))) -;;; (hash-table-keys stats-hash)) -;;; -;;; `(,(hash-table->alist (area-dbs acfg)) ;; dbname => randnum -;;; ,(map (lambda (dbname) ;; dbname is the db name -;;; (cons dbname (stat-when (hash-table-ref stats-hash dbname)))) -;;; (hash-table-keys stats-hash)) -;;; (cpuload . ,(get-normalized-cpu-load))))) -;;; #;(stats . ,(map (lambda (k) ;; create an alist from the stats data -;;; (cons k (stat->alist (hash-table-ref (area-stats acfg) k)))) -;;; (hash-table-keys (area-stats acfg)))) -;;; -;;; #;(trace -;;; ;; assv -;;; ;; cdr -;;; ;; caar -;;; ;; ;; cdr -;;; ;; call -;;; ;; finalize-all-db-handles -;;; ;; get-all-server-pkts -;;; ;; get-normalized-cpu-load -;;; ;; get-normalized-cpu-load-raw -;;; ;; launch -;;; ;; nmsg-send -;;; ;; process-db-queries -;;; ;; receive-message -;;; ;; std-peer-handler -;;; ;; update-known-servers -;;; ;; work-queue-processor -;;; ) -;;; -;;; ;;====================================================================== -;;; ;; netutil -;;; ;; move this back to ulex-netutil.scm someday? -;;; ;;====================================================================== -;;; -;;; ;; #include -;;; ;; #include -;;; ;; #include -;;; ;; #include -;;; -;;; (foreign-declare "#include \"sys/types.h\"") -;;; (foreign-declare "#include \"sys/socket.h\"") -;;; (foreign-declare "#include \"ifaddrs.h\"") -;;; (foreign-declare "#include \"arpa/inet.h\"") -;;; -;;; ;; get IP addresses from ALL interfaces -;;; (define get-all-ips -;;; (foreign-safe-lambda* scheme-object () -;;; " -;;; -;;; // from https://stackoverflow.com/questions/17909401/linux-c-get-default-interfaces-ip-address : -;;; -;;; -;;; C_word lst = C_SCHEME_END_OF_LIST, len, str, *a; -;;; // struct ifaddrs *ifa, *i; -;;; // struct sockaddr *sa; -;;; -;;; struct ifaddrs * ifAddrStruct = NULL; -;;; struct ifaddrs * ifa = NULL; -;;; void * tmpAddrPtr = NULL; -;;; -;;; if ( getifaddrs(&ifAddrStruct) != 0) -;;; C_return(C_SCHEME_FALSE); -;;; -;;; // for (i = ifa; i != NULL; i = i->ifa_next) { -;;; for (ifa = ifAddrStruct; ifa != NULL; ifa = ifa->ifa_next) { -;;; if (ifa->ifa_addr->sa_family==AF_INET) { // Check it is -;;; // a valid IPv4 address -;;; tmpAddrPtr = &((struct sockaddr_in *)ifa->ifa_addr)->sin_addr; -;;; char addressBuffer[INET_ADDRSTRLEN]; -;;; inet_ntop(AF_INET, tmpAddrPtr, addressBuffer, INET_ADDRSTRLEN); -;;; // printf(\"%s IP Address %s\\n\", ifa->ifa_name, addressBuffer); -;;; len = strlen(addressBuffer); -;;; a = C_alloc(C_SIZEOF_PAIR + C_SIZEOF_STRING(len)); -;;; str = C_string(&a, len, addressBuffer); -;;; lst = C_a_pair(&a, str, lst); -;;; } -;;; -;;; // else if (ifa->ifa_addr->sa_family==AF_INET6) { // Check it is -;;; // // a valid IPv6 address -;;; // tmpAddrPtr = &((struct sockaddr_in6 *)ifa->ifa_addr)->sin6_addr; -;;; // char addressBuffer[INET6_ADDRSTRLEN]; -;;; // inet_ntop(AF_INET6, tmpAddrPtr, addressBuffer, INET6_ADDRSTRLEN); -;;; //// printf(\"%s IP Address %s\\n\", ifa->ifa_name, addressBuffer); -;;; // len = strlen(addressBuffer); -;;; // a = C_alloc(C_SIZEOF_PAIR + C_SIZEOF_STRING(len)); -;;; // str = C_string(&a, len, addressBuffer); -;;; // lst = C_a_pair(&a, str, lst); -;;; // } -;;; -;;; // else { -;;; // printf(\" not an IPv4 address\\n\"); -;;; // } -;;; -;;; } -;;; -;;; freeifaddrs(ifa); -;;; C_return(lst); -;;; -;;; ")) -;;; -;;; ;; Change this to bias for addresses with a reasonable broadcast value? -;;; ;; -;;; (define (ip-pref-less? a b) -;;; (let* ((rate (lambda (ipstr) -;;; (regex-case ipstr -;;; ( "^127\\." _ 0 ) -;;; ( "^(10\\.0|192\\.168\\.)\\..*" _ 1 ) -;;; ( else 2 ) )))) -;;; (< (rate a) (rate b)))) -;;; -;;; -;;; (define (get-my-best-address) -;;; (let ((all-my-addresses (get-all-ips)) -;;; ;;(all-my-addresses-old (vector->list (hostinfo-addresses (hostname->hostinfo (get-host-name))))) -;;; ) -;;; (cond -;;; ((null? all-my-addresses) -;;; (get-host-name)) ;; no interfaces? -;;; ((eq? (length all-my-addresses) 1) -;;; (car all-my-addresses)) ;; only one to choose from, just go with it -;;; -;;; (else -;;; (car (sort all-my-addresses ip-pref-less?))) -;;; ;; (else -;;; ;; (ip->string (car (filter (lambda (x) ;; take any but 127. -;;; ;; (not (eq? (u8vector-ref x 0) 127))) -;;; ;; all-my-addresses)))) -;;; -;;; ))) -;;; -;;; (define (get-all-ips-sorted) -;;; (sort (get-all-ips) ip-pref-less?)) -;;; -;;; - ADDED ulex/ulex.dot Index: ulex/ulex.dot ================================================================== --- /dev/null +++ ulex/ulex.dot @@ -0,0 +1,136 @@ +// Copyright 2006-2017, Matthew Welland. +// +// This file is part of Megatest. +// +// Megatest is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Megatest is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with Megatest. If not, see . + +digraph G { + + // graph[center=true, margin=0.2, nodesep=0.1, ranksep=0.3] + + layout=neato; + // layout=fdp; + // overlap=scalexy; //false, compress, ... + overlap=scalexy; + // sep="+1"; // 0.1, +1 + sep="-0.1"; + + user_program [label="user program"]; + + subgraph cluster_1 { + node [style=filled,shape=oval]; + label = "caller"; + color=brown; + + send_receive [label="(send-receive uconn\n host-port qrykey cmd data)"]; + send [label="(send uconn host-port\n qrykey cmd data)"]; + ulex_cmd_loopcaller [label="(ulex-cmd-loop uconn)"]; + ulex_handlercaller [label="(ulex-handler uconn rdat)"]; + mailbox [label="mailbox\n\nrdat\n...",shape=box]; + + send_receive -> send; + ulex_cmd_loopcaller -> ulex_handlercaller; + ulex_handlercaller -> mailbox; + mailbox -> send_receive; + } + + subgraph cluster_2 { + node [shape=oval]; + label = "listener"; + color=green; + + ulex_cmd_loop [label="(ulex-cmd-loop uconn)"]; + ulex_handler [label="(ulex-handler \nuconn rdat)"]; + add_to_work_queue [label="(add-to-work-queue\n uconn rdat)"]; + queue [label="queue\n\nrdat\n...",shape=box]; + process_work_queue [label="(process-work-queue uconn)"]; + do_work [label="(do-work uconn rdat)\nrdat: '(rem-host-port qrykey cmd params)"]; + user_proc [label="(proc rem-host-port\n qrykey cmd params)\n;; proc supplied by user"]; + sendlis [label="(send uconn host-port\n qrykey 'response result)"]; + + ulex_cmd_loop -> ulex_handler [label="rdat"]; + ulex_handler -> add_to_work_queue [label="rdat"]; + + add_to_work_queue -> queue [label="rdat"]; + + subgraph cluster_3 { + label = "remote work"; + color=blue; + + queue -> process_work_queue [label="rdat"]; + process_work_queue -> do_work [label="rdat"]; + do_work -> user_proc; // [label="rdat: '(rem-host-port\n qrykey cmd params)"]; + } + } + + user_proc -> sendlis; + user_program -> send_receive; + send_receive -> user_program; + + send -> ulex_cmd_loop [label="rdat: '(host-port\n qrykey cmd data)"]; + sendlis -> ulex_cmd_loopcaller [label="rdat: '(host-port qrykey\n 'response result)"]; + ulex_handler -> send [label="'ack"]; + ulex_handlercaller -> sendlis [label="'ack"]; + +} + + +// check_available_queue -> remove_entries_over_10s_old; +// remove_entries_over_10s_old -> set_available [label="num_avail < 3"]; +// remove_entries_over_10s_old -> exit [label="num_avail > 2"]; +// +// set_available -> delay_2s; +// delay_2s -> check_place_in_queue; +// +// check_place_in_queue -> "http:transport-launch" [label="at head"]; +// check_place_in_queue -> exit [label="not at head"]; +// +// "client:login" -> "server:shutdown" [label="login failed"]; +// "server:shutdown" -> exit; +// +// subgraph cluster_2 { +// "http:transport-launch" -> "http:transport-run"; +// "http:transport-launch" -> "http:transport-keep-running"; +// +// "http:transport-keep-running" -> "tests running?"; +// "tests running?" -> "client:login" [label=yes]; +// "tests running?" -> "server:shutdown" [label=no]; +// "client:login" -> delay_5s [label="login ok"]; +// delay_5s -> "http:transport-keep-running"; +// } +// + // start_server -> "server_running?"; + // "server_running?" -> set_available [label="no"]; + // "server_running?" -> delay_2s [label="yes"]; + // delay_2s -> "still_running?"; + // "still_running?" -> ping_server [label=yes]; + // "still_running?" -> set_available [label=no]; + // ping_server -> exit [label=alive]; + // ping_server -> remove_server_record [label=dead]; + // remove_server_record -> set_available; + // set_available -> avail_delay [label="delay 3s"]; + // avail_delay -> "first_in_queue?"; + // + // "first_in_queue?" -> set_running [label=yes]; + // set_running -> get_next_port -> handle_requests; + // "first_in_queue?" -> "dead_entry_in_queue?" [label=no]; + // "dead_entry_in_queue?" -> "server_running?" [label=no]; + // "dead_entry_in_queue?" -> "remove_dead_entries" [label=yes]; + // remove_dead_entries -> "server_running?"; + // + // handle_requests -> start_shutdown [label="no traffic\nno running tests"]; + // handle_requests -> shutdown_request; + // start_shutdown -> shutdown_delay; + // shutdown_request -> shutdown_delay; + // shutdown_delay -> exit; ADDED ulex/ulex.pdf Index: ulex/ulex.pdf ================================================================== --- /dev/null +++ ulex/ulex.pdf cannot compute difference between binary files ADDED ulex/ulex.png Index: ulex/ulex.png ================================================================== --- /dev/null +++ ulex/ulex.png cannot compute difference between binary files Index: ulex/ulex.scm ================================================================== --- ulex/ulex.scm +++ ulex/ulex.scm @@ -1,8 +1,8 @@ ;; ulex: Distributed sqlite3 db ;;; -;; Copyright (C) 2018 Matt Welland +;; Copyright (C) 2018-2021 Matt Welland ;; Redistribution and use in source and binary forms, with or without ;; modification, is permitted. ;; ;; THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS ;; OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED @@ -24,11 +24,42 @@ ;; Why sql-de-lite and not say, dbi? - performance mostly, then simplicity. ;; ;;====================================================================== (module ulex - * + * + #;( + + ;; NOTE: looking for the handler proc - find the run-listener :) + + run-listener ;; (run-listener handler-proc [port]) => uconn + + ;; NOTE: handler-proc params; + ;; (handler-proc rem-host-port qrykey cmd params) + + send-receive ;; (send-receive uconn host-port cmd data) + + ;; NOTE: cmd can be any plain text symbol except for these; + ;; 'ping 'ack 'goodbye 'response + + set-work-handler ;; (set-work-handler proc) + + wait-and-close ;; (wait-and-close uconn) + + ulex-listener? + + ;; needed to get the interface:port that was automatically found + udat-port + udat-host-port + + ;; for testing only + ;; pp-uconn + + ;; parameters + work-method ;; parameter; 'threads, 'mailbox, 'limited, 'direct + return-method ;; parameter; 'mailbox, 'polling, 'direct + ) (import scheme chicken.base chicken.file chicken.time @@ -38,13 +69,14 @@ chicken.pretty-print address-info mailbox matchable - queues + ;; queues regex regex-case + s11n srfi-1 srfi-18 srfi-4 srfi-69 system-information @@ -61,36 +93,63 @@ (host-port #f) (socket #f) ;; the peers (peers (make-hash-table)) ;; host:port->peer ;; work handling - (work-queue (make-queue)) - (work-proc #f) ;; set by user - (cnum 0) ;; cookie number - (mboxes (make-hash-table)) - (avail-cmboxes '()) ;; list of ( . ) for re-use - ) - -;; struct for keeping track of others we are talking to -;; -(defstruct pdat - (host-port #f) - (conns '()) ;; list of pcon structs, pop one off when calling the peer - ) - -;; struct for peer connections, keep track of expiration etc. -;; -(defstruct pcon - (inp #f) - (oup #f) - (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59) - (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes - ) + (work-queue (make-mailbox)) + (work-proc #f) ;; set by user + (cnum 0) ;; cookie number + (mboxes (make-hash-table)) ;; for the replies + (avail-cmboxes '()) ;; list of ( . ) for re-use + ;; threads + (numthreads 10) + (cmd-thread #f) + (work-queue-thread #f) + (num-threads-running 0) + ) + +;; Parameters + +;; work-method: +(define work-method (make-parameter 'mailbox)) +;; mailbox - all rdat goes through mailbox +;; threads - all rdat immediately executed in new thread +;; direct - no queuing +;; + +;; return-method, return the result to waiting send-receive: +(define return-method (make-parameter 'mailbox)) +;; mailbox - create a mailbox and use it for passing returning results to send-receive +;; polling - put the result in a hash table keyed by qrykey and send-receive can poll it for result +;; direct - no queuing, result is passed back in single tcp connection +;; + +;; ;; struct for keeping track of others we are talking to +;; ;; +;; (defstruct pdat +;; (host-port #f) +;; (conns '()) ;; list of pcon structs, pop one off when calling the peer +;; ) +;; +;; ;; struct for peer connections, keep track of expiration etc. +;; ;; +;; (defstruct pcon +;; (inp #f) +;; (oup #f) +;; (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59) +;; (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes +;; ) ;;====================================================================== ;; listener ;;====================================================================== + +;; is uconn a ulex connector (listener) +;; +(define (ulex-listener? uconn) + (udat? uconn)) + ;; create a tcp listener and return a populated udat struct with ;; my port, address, hostname, pid etc. ;; return #f if fail to find a port to allocate. ;; ;; if udata-in is #f create the record @@ -111,14 +170,43 @@ (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname))) (udat-port-set! uconn port) (udat-host-port-set! uconn (conc addr":"port)) (udat-socket-set! uconn tlsn) uconn)) + +;; run-listener does all the work of starting a listener in a thread +;; it then returns control +;; +(define (run-listener handler-proc #!optional (port-suggestion 4242)) + (let* ((uconn (make-udat))) + (udat-work-proc-set! uconn handler-proc) + (if (setup-listener uconn port-suggestion) + (let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop")) + (th2 (make-thread (lambda () + (case (work-method) + ((mailbox limited) + (process-work-queue uconn)))) + "Ulex work queue processor"))) + (tcp-buffer-size 2048) + ;; (max-connections 2048) + (thread-start! th1) + (thread-start! th2) + (udat-cmd-thread-set! uconn th1) + (udat-work-queue-thread-set! uconn th2) + (print "cmd loop and process workers started, listening on "(udat-host-port uconn)".") + uconn) + (assert #f "ERROR: run-listener called without proper setup.")))) + +(define (wait-and-close uconn) + (thread-join! (udat-cmd-thread uconn)) + (tcp-close (udat-socket uconn))) ;;====================================================================== ;; peers and connections ;;====================================================================== + +(define *send-mutex* (make-mutex)) ;; send structured data to recipient ;; ;; NOTE: qrykey is what was called the "cookie" previously ;; @@ -131,155 +219,237 @@ ;; (define (send udata host-port qrykey cmd params) (let* ((my-host-port (udat-host-port udata)) ;; remote will return to this (isme #f #;(equal? host-port my-host-port)) ;; calling myself? ;; dat is a self-contained work block that can be sent or handled locally - (dat (list my-host-port qrykey cmd params)) - ) - (if isme - (ulex-handler udata dat) ;; no transmission needed - (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC? - exn - #f + (dat (list my-host-port qrykey cmd params))) + (cond + (isme (ulex-handler udata dat)) ;; no transmission needed + (else + (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC? + exn + #f + (begin + ;; (mutex-lock! *send-mutex*) (let-values (((inp oup)(tcp-connect host-port))) (let ((res (if (and inp oup) (begin - (write dat oup) - (read inp)) ;; yes, we always want an ack + (serialize dat oup) + (deserialize inp)) (begin (print "ERROR: send called but no receiver has been setup. Please call setup first!") #f)))) (close-input-port inp) (close-output-port oup) - res)))))) ;; res will always be 'ack - + ;; (mutex-unlock! *send-mutex*) + res)))))))) ;; res will always be 'ack unless return-method is direct + ;; send a request to the given host-port and register a mailbox in udata ;; wait for the mailbox data and return it ;; (define (send-receive uconn host-port cmd data) - (let* ((cmbox (get-cmbox uconn)) ;; would it be better to keep a stack of mboxes to reuse? - (qrykey (car cmbox)) - (mbox (cdr cmbox)) - (mbox-time (current-milliseconds))) - (if (eq? (send uconn host-port qrykey cmd data) 'ack) - (let* ((mbox-timeout-secs 120) ;; timeout) - (mbox-timeout-result 'MBOX_TIMEOUT) - (res (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result)) - (mbox-receive-time (current-milliseconds))) - (if (eq? res 'MBOX_TIMEOUT) - #f ;; convert to raising exception? - res)) - (begin - (print "ERROR: Communication failed?") - #f)))) ;; #f means failed to communicate + (cond + ((member cmd '(ping goodbye)) ;; these are immediate + (send uconn host-port 'ping cmd data)) + ((eq? (work-method) 'direct) + ;; the result from send will be the actual result, not an 'ack + (send uconn host-port 'direct cmd data)) + (else + (case (return-method) + ((polling) + (let* ((qrykey (make-cookie uconn)) + (sres (send uconn host-port qrykey cmd data))) + (case sres + ((ack) + (let loop ((start-time (current-milliseconds))) + (if (> (current-milliseconds)(+ start-time 10000)) ;; ten seconds timeout + (begin + (print "ULEX ERROR: timed out waiting for response from "host-port", "cmd" "data) + #f) + (let* ((result (hash-table-ref/default (udat-mboxes uconn) qrykey #f))) ;; NOTE: we are re-using mboxes hash + (if result ;; result is '(status . result-data) or #f for nothing yet + (begin + (hash-table-delete! (udat-mboxes uconn) qrykey) + (cdr result)) + (begin + (thread-sleep! 0.01) + (loop start-time))))))) + (else + (print "ULEX ERROR: Communication failed? sres="sres) + #f)))) + ((mailbox) + (let* ((cmbox (get-cmbox uconn)) ;; would it be better to keep a stack of mboxes to reuse? + (qrykey (car cmbox)) + (mbox (cdr cmbox)) + (mbox-time (current-milliseconds)) + (sres (send uconn host-port qrykey cmd data))) ;; short res + (if (eq? sres 'ack) + (let* ((mbox-timeout-secs 120 #;(if (eq? 'primordial (thread-name (current-thread))) + #f + 120)) ;; timeout) + (mbox-timeout-result 'MBOX_TIMEOUT) + (res (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result)) + (mbox-receive-time (current-milliseconds))) + ;; (put-cmbox uconn cmbox) ;; reuse mbox and cookie. is it worth it? + (hash-table-delete! (udat-mboxes uconn) qrykey) + (if (eq? res 'MBOX_TIMEOUT) + (begin + (print "WARNING: mbox timed out for query "cmd", with data "data", waiting for response from "host-port".") + + ;; here it might make sense to clean up connection records and force clean start? + ;; NO. The progam using ulex needs to do the reset. Right thing here is exception + + #f) ;; convert to raising exception? + res)) + (begin + (print "ERROR: Communication failed? Got "sres) + #f)))) + (else + (print "ULEX ERROR: unrecognised return-method "(return-method)".") + #f))))) ;;====================================================================== ;; responder side ;;====================================================================== -;; take a request, rdata, and if not immediate put it in the work queue +;; take a request, rdat, and if not immediate put it in the work queue ;; ;; Reserved cmds; ack ping goodbye response ;; -(define (ulex-handler uconn rdata) - (match rdata ;; (string-split controldat) +(define (ulex-handler uconn rdat) + (assert (list? rdat) "FATAL: ulex-handler give rdat as not list") + (match rdat ;; (string-split controldat) ((rem-host-port qrykey cmd params) ;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params) - (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f))) - (case cmd - ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack) - ((ping) - ;; (print "Got Ping!") - (add-to-work-queue uconn rdata) - 'ack) - ((goodbye) - ;; just clear out references to the caller - (add-to-work-queue uconn rdata) - 'ack) - ((response) ;; this is a result from remote processing, send it as mail ... - (if mbox - (begin - (mailbox-send! mbox params) ;; params here is our result - 'ack) - (begin - (print "ERROR: received result but no associated mbox for cookie "qrykey) - #f))) - (else - ;; (print "Got generic request: "cmd) - (add-to-work-queue uconn rdata) - 'ack)))) - (else - (print "BAD DATA? controldat=" rdata) - 'ack) ;; send ack anyway? - )) + (case cmd + ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack) + ((ping) + ;; (print "Got Ping!") + ;; (add-to-work-queue uconn rdat) + 'ack) + ((goodbye) + ;; just clear out references to the caller. NOT COMPLETE + (add-to-work-queue uconn rdat) + 'ack) + ((response) ;; this is a result from remote processing, send it as mail ... + (case (return-method) + ((polling) + (hash-table-set! (udat-mboxes uconn) qrykey (cons 'ok params)) + 'ack) + ((mailbox) + (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f))) + (if mbox + (begin + (mailbox-send! mbox params) ;; params here is our result + 'ack) + (begin + (print "ERROR: received result but no associated mbox for cookie "qrykey) + 'no-mbox-found)))) + (else (print "ULEX ERROR: unrecognised return-method "(return-method)) + 'bad-return-method))) + (else ;; generic request - hand it to the work queue + (add-to-work-queue uconn rdat) + 'ack))) + (else + (print "ULEX ERROR: bad rdat "rdat) + 'bad-rdat))) ;; given an already set up uconn start the cmd-loop ;; (define (ulex-cmd-loop uconn) - (let* ((serv-listener (udat-socket uconn))) - (let loop ((state 'start)) - (let-values (((inp oup)(tcp-accept serv-listener))) - (let* ((rdat (read inp)) - (resp (ulex-handler uconn rdat))) - (if resp (write resp oup)) - (close-input-port inp) - (close-output-port oup)) - (loop state))))) + (let* ((serv-listener (udat-socket uconn)) + (listener (lambda () + (let loop ((state 'start)) + (let-values (((inp oup)(tcp-accept serv-listener))) + (let* ((rdat (deserialize inp)) ;; '(my-host-port qrykey cmd params) + (resp (ulex-handler uconn rdat))) + (if resp (serialize resp oup)) + (close-input-port inp) + (close-output-port oup)) + (loop state)))))) + ;; start N of them + (let loop ((thnum 0) + (threads '())) + (if (< thnum 100) + (let* ((th (make-thread listener (conc "listener" thnum)))) + (thread-start! th) + (loop (+ thnum 1) + (cons th threads))) + (map thread-join! threads))))) ;; add a proc to the cmd list, these are done symetrically (i.e. in all instances) ;; so that the proc can be dereferenced remotely ;; (define (set-work-handler uconn proc) (udat-work-proc-set! uconn proc)) -;; run-listener does all the work of starting a listener in a thread -;; it then returns control -;; -(define (run-listener handler-proc) - (let* ((uconn (make-udat))) - (udat-work-proc-set! uconn handler-proc) - (if (setup-listener uconn) - (let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop")) - (th2 (make-thread (lambda ()(process-work-queue uconn)) "Ulex work queue processor"))) - (thread-start! th1) - (thread-start! th2) - (print "cmd loop and process workers started") - uconn) - (begin - (print "ERROR: run-listener called without proper setup.") - (exit))))) - ;;====================================================================== ;; work queues - this is all happening on the listener side ;;====================================================================== -;; rdata is (rem-host-port qrykey cmd params) +;; rdat is (rem-host-port qrykey cmd params) -(define (add-to-work-queue uconn rdata) - (queue-add! (udat-work-queue uconn) rdata)) - -(define (do-work uconn rdata) +(define (add-to-work-queue uconn rdat) + #;(queue-add! (udat-work-queue uconn) rdat) + (case (work-method) + ((threads) + (thread-start! (make-thread (lambda () + (do-work uconn rdat)) + "worker thread"))) + ((mailbox) + (mailbox-send! (udat-work-queue uconn) rdat)) + ((direct) + (do-work uconn rdat)) + (else + (print "ULEX ERROR: work-method "(work-method)" not recognised, using mailbox.") + (mailbox-send! (udat-work-queue uconn) rdat)))) + +;; move the logic to return the result somewhere else? +;; +(define (do-work uconn rdat) (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change ;; put this following into a do-work procedure - (match rdata + (match rdat ((rem-host-port qrykey cmd params) - (let* ((result (proc rem-host-port qrykey cmd params))) - ;; send 'response as cmd and result as params - (send uconn rem-host-port qrykey 'response result))) ;; could check for ack + (let* ((start-time (current-milliseconds)) + (result (proc rem-host-port qrykey cmd params)) + (end-time (current-milliseconds)) + (run-time (- end-time start-time))) + (case (work-method) + ((direct) result) + (else + (print "ULEX: work "cmd", "params" done in "run-time" ms") + ;; send 'response as cmd and result as params + (send uconn rem-host-port qrykey 'response result) ;; could check for ack + (print "ULEX: response sent back to "rem-host-port" in "(- (current-milliseconds) end-time)))))) + (MBOX_TIMEOUT #f) (else - (print "ERROR: rdata "rdata", did not match rem-host-port qrykey cmd params"))))) - - + (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params"))))) + +;; NEW APPROACH: +;; (define (process-work-queue uconn) (let ((wqueue (udat-work-queue uconn)) - (proc (udat-work-proc uconn))) - (let loop () - (if (queue-empty? wqueue) - (thread-sleep! 0.1) - (let ((rdata (queue-remove! wqueue))) - (do-work uconn rdata))) - (loop)))) + (proc (udat-work-proc uconn)) + (numthr (udat-numthreads uconn))) + (let loop ((thnum 1) + (threads '())) + (let ((thlst (cons (make-thread (lambda () + (let work-loop () + (let ((rdat (mailbox-receive! wqueue 24000 'MBOX_TIMEOUT))) + (do-work uconn rdat)) + (work-loop))) + (conc "work thread " thnum)) + threads))) + (if (< thnum numthr) + (loop (+ thnum 1) + thlst) + (begin + (print "ULEX: Starting "(length thlst)" worker threads.") + (map thread-start! thlst) + (print "ULEX: Threads started. Joining all.") + (map thread-join! thlst))))))) ;; below was to enable re-use of connections. This seems non-trivial so for ;; now lets open on each call ;; ;; ;; given host-port get or create peer struct @@ -374,50 +544,6 @@ (map address-info-host (filter (lambda (x) (equal? (address-info-type x) "tcp")) (address-infos (get-host-name))))) -;; (map ip->string (vector->list -;; (hostinfo-addresses -;; (host-information (current-hostname)))))) - - -) - -(import ulex trace big-chicken srfi-18 test matchable system-information) -(trace-call-sites #t) -(trace - ;; ulex-handler - ;; send - ;; add-to-work-queue - ) - -(define (handler-proc rem-host-port qrykey cmd params) - (print "handler-proc "rem-host-port" "qrykey" "cmd" "params) - (case cmd - ((ping) 'pong) - ((calc) (eval (with-input-from-string params read))) - ((print) - (print "params="params) - params) - ((reflect) `(,rem-host-port ,qrykey ,cmd ,params)) - (else `(data ,data)))) - -(define uconn (run-listener handler-proc)) - -(pp-uconn uconn) - -;; super basic loop back test -(define res #f) -(define targhost (conc (get-host-name)":"4242)) -(define th1 (make-thread (lambda () - (test #f 10 (send-receive uconn targhost 'calc "(+ 5 5)")) - (set! res (send-receive uconn targhost 'ping '())) - (test #f 'pong (send-receive uconn targhost 'ping '())) - ))) - -(thread-start! th1) -(thread-join! th1) - -(print "All done") -(print "Received "res) - +)