Index: rmtmod.scm ================================================================== --- rmtmod.scm +++ rmtmod.scm @@ -43,17 +43,17 @@ ;; return the handle struct for sending queries to a specific database ;; - initializes the connection object if this is the first access ;; - finds the "captain" and asks who to talk to for the given dbfname ;; - establishes the connection to the current dbowner ;; -(define (rmt:connect alldat dbfname dbtype) +#;(define (rmt:connect alldat dbfname dbtype) (let* ((ulexdat (or (alldat-ulexdat alldat) (rmt:setup-ulex alldat)))) (ulex:connect ulexdat dbfname dbtype))) ;; setup the remote calls -(define (rmt:setup-ulex alldat) +#;(define (rmt:setup-ulex alldat) (let* ((udata (ulex:setup))) ;; establish connection to ulex (alldat-ulexdat-set! alldat udata) ;; register all needed procs (ulex:register-handler udata 'ping cmod:get-full-version) ;; override ping with get-full-version (ulex:register-handler udata 'login cmod:get-full-version) ;; force setup of the connection @@ -61,11 +61,11 @@ udata)) ;; set up a connection to the current owner of the dbfile associated with rid ;; then send the query to that dbfile owner and wait for a response. ;; -(define (rmt:send-receive alldat cmd rid params #!key (attemptnum 1)(area-dat #f)) ;; start attemptnum at 1 so the modulo below works as expected +#;(define (rmt:send-receive alldat cmd rid params #!key (attemptnum 1)(area-dat #f)) ;; start attemptnum at 1 so the modulo below works as expected (let* (;; (alldat *alldat*) (areapath (alldat-areapath alldat)) (dbtype (if (or (not rid)(< rid 1)) ;; this is the criteria for "main.db" "main" "runs")) (dbfname (if (equal? dbtype "main") Index: ulex/ulex.scm ================================================================== --- ulex/ulex.scm +++ ulex/ulex.scm @@ -25,11 +25,12 @@ ;; ;;====================================================================== (use mailbox) -(module ulex * +(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 @@ -38,10 +39,83 @@ 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 @@ -50,36 +124,39 @@ ;; called before connecting to a db using connect. ;; ;; find or become the captain ;; setup and return a ulex object ;; -(define (setup #!optional (udata-in #f)) - (let* ((udata (or udata-in (make-udat))) - (cpkts (get-all-captain-pkts udata)) ;; read captain pkts - (captn (get-winning-pkt cpkts))) - ;; check to see if our own server is started and start one if not - (if (not (udat-serv-listener udata))(start-server-find-port udata)) - (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) - (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) - (setup))))) - (begin - (setup-as-captain udata) ;; this saves the thread to captain-thread and starts the thread - (setup))))) +(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 @@ -212,64 +289,10 @@ (define (get-all-ips) (map ip->string (vector->list (hostinfo-addresses (host-information (current-hostname)))))) -;; 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) - (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 - ) - (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)) @@ -308,30 +331,26 @@ ;;====================================================================== ;; NB// This needs to be started in a thread ;; ;; setup to be a captain -;; - start server +;; - local server MUST be started already ;; - create pkt ;; - start server port handler ;; (define (setup-as-captain udata) - (if (start-server-find-port udata) ;; puts the server in 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)) - (begin - (print "ERROR: failed to start server.") + (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) @@ -424,17 +443,23 @@ ;; 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. ;; -(define (start-server-find-port udata #!optional (port 4242)) - (handle-exceptions - exn - (if (< port 65535) - (start-server-find-port udata (+ port 1)) - #f) - (connect-server udata port))) +;; 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]])