Changes In Branch v1.6584-nanomsg Through [96feeca725] Excluding Merge-Ins
This is equivalent to a diff from 58eed43d63 to 96feeca725
|
2021-06-07
| ||
| 08:59 | Adjusted receive for new usage (was copied from mtut.scm check-in: f70db69e66 user: matt tags: v1.6584-nanomsg | |
| 08:46 | wip. check-in: 96feeca725 user: matt tags: v1.6584-nanomsg | |
| 06:26 | try nanomsg check-in: 14a50c3c87 user: matt tags: v1.6584-nanomsg | |
|
2021-06-06
| ||
| 23:58 | all effed Leaf check-in: 58eed43d63 user: matt tags: v1.6584-tcp6 | |
| 22:07 | Got all PASS on current tests check-in: f1e43b7b99 user: matt tags: v1.6584-tcp6 | |
Modified rmtmod.scm from [cd536fb107] to [7f26d483f5].
| ︙ | ︙ | |||
60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | directory-utils ;; http-client ;; intarweb matchable md5 message-digest (prefix base64 base64:) (prefix sqlite3 sqlite3:) regex s11n ;; spiffy ;; spiffy-directory-listing ;; spiffy-request-vars srfi-1 srfi-13 srfi-18 srfi-69 stack system-information | > | | 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
directory-utils
;; http-client
;; intarweb
matchable
md5
message-digest
nanomsg
(prefix base64 base64:)
(prefix sqlite3 sqlite3:)
regex
s11n
;; spiffy
;; spiffy-directory-listing
;; spiffy-request-vars
srfi-1
srfi-13
srfi-18
srfi-69
stack
system-information
;; tcp6
typed-records
uri-common
z3
apimod
commonmod
configfmod
|
| ︙ | ︙ | |||
117 118 119 120 121 122 123 124 125 126 127 128 129 130 | ;; info about me as a server ;; (defstruct servdat (host #f) (port #f) (uuid #f) (dbfile #f) (api-url #f) (api-uri #f) (api-req #f) (status 'starting) (trynum 0) ;; count the number of ports we've tried ) | > | 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | ;; info about me as a server ;; (defstruct servdat (host #f) (port #f) (uuid #f) (rep #f) (dbfile #f) (api-url #f) (api-uri #f) (api-req #f) (status 'starting) (trynum 0) ;; count the number of ports we've tried ) |
| ︙ | ︙ | |||
293 294 295 296 297 298 299 |
;; sometime in the future
;;
(define (rmt:send-receive-real remote apath dbname cmd params)
(let* ((conn (rmt:get-conn remote apath dbname)))
(assert conn "FATAL: rmt:send-receive-real called without the needed channels opened")
(pp (rmt:conn->alist conn))
;; (rmt:send-receive-setup conn)
| | | | > > | | | | > > | | | | | | | | 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 |
;; sometime in the future
;;
(define (rmt:send-receive-real remote apath dbname cmd params)
(let* ((conn (rmt:get-conn remote apath dbname)))
(assert conn "FATAL: rmt:send-receive-real called without the needed channels opened")
(pp (rmt:conn->alist conn))
;; (rmt:send-receive-setup conn)
;; (let-values (((i o) (tcp-connect (rmt:conn-ipaddr conn)
;; (rmt:conn-port conn))))
(let* ((key #f)
(host (rmt:conn-ipaddr conn))
(port (rmt:conn-port conn))
(payload `((cmd . ,cmd)
(key . ,(rmt:conn-srvkey conn))
(params . ,params)))
(res (open-send-receive-nn (conc host":"port)
(sexpr->string payload))))
;; begin
;; (write payload o) ;; (rmt:conn-outport conn))
;; (with-input-from-port
;; i ;; (rmt:conn-inport conn)
;; read))))
;; (close-input-port i)
;; (close-output-port o)
(string->sexpr res))))
;; (if (string? res)
;; (string->sexpr res)
;; res))))
;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
|
| ︙ | ︙ | |||
1589 1590 1591 1592 1593 1594 1595 |
(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)))
| | > > > > > > > | | | | | | | | | | | | | | | > > > > > > > > > > > > > > > > > > > > > > | | | | | | | | | | | | | | | | < < < < | < | | 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 |
(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 (nn-recv rep)))
(let* ((data (string->sexpr instr))
(res (api:process-request *dbstruct-db* data))
(resdat (sexpr->string res)))
(set! *db-last-access* (current-seconds))
(nn-send rep resdat)
(loop (nn-recv rep)))))
;; (let oloop ()
;; (let-values (((i o) (tcp-accept l)))
;; ;; (write-line "Hello!" o)
;; (let loop ((indat (read i)))
;; (if (eof-object? indat)
;; (begin
;; (close-input-port i)
;; (close-output-port o)
;; (oloop))
;; (let* ((res (api:process-request *dbstruct-db* indat)))
;; (set! *db-last-access* (current-seconds))
;; (write res o)
;; (loop (read i))))))))
(let* ((portnum (servdat-port *server-info*)))
(portlogger:open-run-close portlogger:set-port portnum "released")
(debug:print 1 *default-log-port* "INFO: server has been stopped"))))
(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)
;; any error in following steps will result in a retry
(if (is-port-in-use portnum)
(begin
(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
(if (not *server-info*)
(set! *server-info* (make-servdat
host: ipaddrstr
port: portnum)))
(servdat-status-set! *server-info* 'starting)
(servdat-port-set! *server-info* portnum)
(if (not (servdat-rep *server-info*))
(servdat-rep-set! *server-info* (nn-socket 'rep)))
(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))))
(nn-bind rep (conc "tcp://*:" portnum))
rep)))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;;======================================================================
;; C L I E N T S
|
| ︙ | ︙ | |||
1794 1795 1796 1797 1798 1799 1800 | all-pkt-files))) (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 | | | | | | < | | | > > > | | | | | | | | | | 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 |
all-pkt-files)))
(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-values (((i o)(handle-exceptions
;; exn
;; (values #f #f)
;; (tcp-connect host port))))
;; (if (and i o)
(let* ((data (sexpr->string `((cmd . ping)
(key . ,key)
(params . ()))))
(res (open-send-receive-nn (conc host ":" port) data)))
(string->sexpr res)))
;; (let ((res (with-input-from-port i
;; read)))
;; (close-output-port o)
;; (close-input-port i)
;; res))
;; (if (string? res)
;; (string->sexpr res)
;; res)))
;; (begin ;; connection failed
;; (debug:print-info 0 *default-log-port* "Server at "host":"port" is not responding.")
;; #f))))
;; (define (loop-test host port data) ;; server-address is host:port
;; ;; ping the server and ask it
;; ;; if it ready
;; ;; (let* ((sdat (servdat-init #f host port #f)))
;; ;; (http-transport:send-receive sdat "abc" 'ping '())))
;; (let* ((payload (sexpr->string data))
;; (res (with-input-from-request
|
| ︙ | ︙ | |||
2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 |
(argv)))))))
(define (rmt:get-signature)
(if *my-signature* *my-signature*
(let ((sig (rmt:mk-signature)))
(set! *my-signature* sig)
*my-signature*)))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;; run ping in separate process, safest way in some cases
;;
| > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 |
(argv)))))))
(define (rmt:get-signature)
(if *my-signature* *my-signature*
(let ((sig (rmt:mk-signature)))
(set! *my-signature* sig)
*my-signature*)))
;;======================================================================
;; Nanomsg transport
;;======================================================================
(define (is-port-in-use port-num)
(let* ((ret #f))
(let-values (((inp oup pid)
(process "netstat" (list "-tulpn" ))))
(let loop ((inl (read-line inp)))
(if (not (eof-object? inl))
(begin
(if (string-search (regexp (conc ":" port-num)) inl)
(begin
;(print "Output: " inl)
(set! ret #t))
(loop (read-line inp)))))))
ret))
;;start a server, returns the connection
;;
(define (start-nn-server portnum )
(let ((rep (nn-socket 'rep)))
(handle-exceptions
exn
(let ((emsg ((condition-property-accessor 'exn 'message) exn)))
(print "ERROR: Failed to start server \"" emsg "\"")
(exit 1))
(nn-bind rep (conc "tcp://*:" portnum)))
rep))
;; open connection to server, send message, close connection
;;
(define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
(let ((req (nn-socket 'req))
(uri (conc "tcp://" host-port))
(res #f)
;; (contacts (alist-ref 'contact attrib))
;; (mode (alist-ref 'mode attrib))
)
(handle-exceptions
exn
(let ((emsg ((condition-property-accessor 'exn 'message) exn)))
;; Send notification
(debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\"" )
#f)
(nn-connect req uri)
;; (print "Connected to the server " )
(nn-send req msg)
;; (print "Request Sent")
(let* ((th1 (make-thread (lambda ()
(let ((resp (nn-recv req)))
(nn-close req)
(set! res (if (equal? resp "ok")
#t
#f))))
"recv thread"))
(th2 (make-thread (lambda ()
(thread-sleep! timeout)
(thread-terminate! th1))
"timer thread")))
(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
(let ((req (nn-socket 'req))
(uri (conc "tcp://" host-port))
(res #f)
;; (contacts (alist-ref 'contact attrib))
;; (mode (alist-ref 'mode attrib))
)
(handle-exceptions
exn
(let ((emsg ((condition-property-accessor 'exn 'message) exn)))
;; Send notification
(debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\"" )
#f)
(nn-connect req uri)
;; (print "Connected to the server " )
(nn-send req msg)
;; (print "Request Sent")
;; receive code here
;;(print (nn-recv req))
(let* ((th1 (make-thread (lambda ()
(let ((resp (nn-recv req)))
(nn-close req)
(print resp)
(set! res (if (equal? resp "ok")
#t
#f))))
"recv thread"))
(th2 (make-thread (lambda ()
(thread-sleep! timeout)
(thread-terminate! th1))
"timer thread")))
(thread-start! th1)
(thread-start! th2)
(thread-join! th1)
res))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;; run ping in separate process, safest way in some cases
;;
|
| ︙ | ︙ |
Modified tests/unittests/server.scm from [245ccd4190] to [fecd3b071a].
| ︙ | ︙ | |||
52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
;; these let me cut and paste from source easily
(define apath *toppath*)
(define dbname ".db/2.db")
(define remote *rmt:remote*)
(define keyvals '(("SYSTEM" "a")("RELEASE" "b")))
(test #f #t (rmt:open-main-connection remote apath))
(test #f #t (rmt:conn? (rmt:get-conn *rmt:remote* *toppath* ".db/main.db")))
(test #f 'server-started (rmt:send-receive-real *rmt:remote* *toppath* ".db/main.db" 'get-server `(,apath ,dbname)))
(thread-sleep! 2)
(test #f #t (list? (rmt:general-open-connection *rmt:remote* *toppath* ".db/2.db")))
(test #f '("SYSTEM" "RELEASE") (rmt:get-keys))
| > | 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 |
;; these let me cut and paste from source easily
(define apath *toppath*)
(define dbname ".db/2.db")
(define remote *rmt:remote*)
(define keyvals '(("SYSTEM" "a")("RELEASE" "b")))
(test #f #t (rmt:open-main-connection remote apath))
(test #f #t (rmt:conn? (rmt:get-conn *rmt:remote* *toppath* ".db/main.db")))
(test #f #t (rmt:conn? (rmt:get-conn *rmt:remote* *toppath* ".db/main.db")))
(test #f 'server-started (rmt:send-receive-real *rmt:remote* *toppath* ".db/main.db" 'get-server `(,apath ,dbname)))
(thread-sleep! 2)
(test #f #t (list? (rmt:general-open-connection *rmt:remote* *toppath* ".db/2.db")))
(test #f '("SYSTEM" "RELEASE") (rmt:get-keys))
|
| ︙ | ︙ |