Changes In Branch v1.6584-nanomsg Through [29dd9489e5] Excluding Merge-Ins
This is equivalent to a diff from 58eed43d63 to 29dd9489e5
2021-06-12
| ||
04:25 | wip check-in: c47b41a610 user: matt tags: v1.6584-nanomsg | |
2021-06-09
| ||
09:02 | wip check-in: 29dd9489e5 user: matt tags: v1.6584-nanomsg | |
2021-06-07
| ||
08:59 | Adjusted receive for new usage (was copied from mtut.scm check-in: f70db69e66 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 commonmod.scm from [f4c84442dd] to [d20588a683].
︙ | ︙ | |||
3647 3648 3649 3650 3651 3652 3653 | ;; timeout is hms string: 1h 5m 3s, default is 10 minutes ;; (define (server:expiration-timeout) (let ((tmo (configf:lookup *configdat* "server" "timeout"))) (if (and (string? tmo) (common:hms-string->seconds tmo)) ;; BUG: hms-string->seconds is broken, if given "10" returns 0. Also, it doesn't belong in this logic unless the string->number is changed below (* 3600 (string->number tmo)) | | | 3647 3648 3649 3650 3651 3652 3653 3654 3655 3656 3657 3658 3659 3660 3661 | ;; timeout is hms string: 1h 5m 3s, default is 10 minutes ;; (define (server:expiration-timeout) (let ((tmo (configf:lookup *configdat* "server" "timeout"))) (if (and (string? tmo) (common:hms-string->seconds tmo)) ;; BUG: hms-string->seconds is broken, if given "10" returns 0. Also, it doesn't belong in this logic unless the string->number is changed below (* 3600 (string->number tmo)) 60))) ;; default is one minute (define (runs:get-mt-env-alist run-id runname target testname itempath) ;;(bb-check-path msg: "runs:set-megatest-env-vars entry") `(("MT_TEST_NAME" . ,testname) ("MT_ITEMPATH" . ,itempath) |
︙ | ︙ |
Modified docs/manual/server.dot from [3e029f5fe5] to [ec783673b9].
︙ | ︙ | |||
15 16 17 18 19 20 21 | // You should have received a copy of the GNU General Public License // along with Megatest. If not, see <http://www.gnu.org/licenses/>. digraph G { subgraph cluster_1 { node [style=filled,shape=box]; | > > > > | | | | | | | | | | | | | | | | | | | | | | < | < < < < < < < < < < < | > > > > > > > > > > > > | | | | | | | | | | | | | | | | 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | // You should have received a copy of the GNU General Public License // along with Megatest. If not, see <http://www.gnu.org/licenses/>. digraph G { subgraph cluster_1 { node [style=filled,shape=box]; rmt:send-receive -> "init-*remote* if needed" -> rmt:general-open-connection -> rmt:send-receive-real; // 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; // // label = "server:launch"; // color=brown; } // client_start_server -> start_server; // handle_requests -> read_write; // read_write -> handle_requests; } |
Modified rmtmod.scm from [cd536fb107] to [d9cd48e79d].
︙ | ︙ | |||
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 ) |
︙ | ︙ | |||
241 242 243 244 245 246 247 | #t) (start-main-srv))) (start-main-srv)))) ;; NB// remote is a rmt:remote struct ;; (define (rmt:general-open-connection remote apath dbname #!key (num-tries 5)) | | > | > | 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 | #t) (start-main-srv))) (start-main-srv)))) ;; NB// remote is a rmt:remote struct ;; (define (rmt:general-open-connection remote apath dbname #!key (num-tries 5)) (let* ((mdbname (db:run-id->dbname #f)) (mconn (rmt:get-conn remote apath mdbname))) (cond ((or (not mconn) ;; no channel open to main? (< (rmt:conn-expires mconn)(+ (current-seconds) 2))) ;; restablish connection if less than 2 seconds on the lease (rmt:open-main-connection remote apath) (thread-sleep! 2) (rmt:general-open-connection remote apath mdbname)) ((not (rmt:get-conn remote apath dbname)) ;; no channel open to dbname? (let* ((res (rmt:send-receive-real remote apath mdbname 'get-server `(,apath ,dbname)))) (case res ((server-started) |
︙ | ︙ | |||
291 292 293 294 295 296 297 | ;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed ;; 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") | | < | | < | | | | < < < | < < < < | < < < | | 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 | ;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed ;; 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") (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)))) (string->sexpr 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 remote apath dbname) ;; (let* ((conn (rmt:get-conn remote apath dbname))) ;; (assert conn "FATAL: Unable to connect to db "apath"/"dbname) ;; #;(let* ((res (with-input-from-request ;; (rmt:conn->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")) (for-each (lambda (cmd) (let ((cmd-dat (hash-table-ref *db-stats* cmd))) (debug:print 18 *default-log-port* (format #f fmtstr cmd (vector-ref cmd-dat 0) (vector-ref cmd-dat 1) (/ (vector-ref cmd-dat 1)(vector-ref cmd-dat 0)))))) (sort (hash-table-keys *db-stats*) (lambda (a b) (> (vector-ref (hash-table-ref *db-stats* a) 0) |
︙ | ︙ | |||
1456 1457 1458 1459 1460 1461 1462 | (bdat-time-to-exit-set! *bdat* #t) #t)))) (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.") (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 (if *server-info* | < < < | | > | > | | | | | | | | | | | | | | | | | 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 | (bdat-time-to-exit-set! *bdat* #t) #t)))) (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.") (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 (if *server-info* (let ((dbfile (servdat-dbfile *server-info*))) (if dbfile (begin ;; do a final sync here (if (string-match ".*/main.db$" dbfile) (let ((pkt-file (conc (get-pkts-dir *toppath*) "/" (servdat-uuid *server-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 for "dbfile) (db:with-lock-db (servdat-dbfile *server-info*) (lambda (dbh dbfile) (db:release-lock dbh dbfile)))) (let* ((sdat *server-info*)) ;; we have a run-id server (rmt:send-receive-real *rmt:remote* *toppath* (db:run-id->dbname #f) 'deregister-server `(,(servdat-uuid sdat) ,(current-process-id) ,(servdat-host sdat) ;; iface ,(servdat-port sdat))))))))) ;; (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) (begin (sqlite3:interrupt! db) (sqlite3:finalize! db #t) (bdat-task-db-set! *bdat* #f))))) #;(http-client#close-idle-connections!) (if (not (eq? *default-log-port* (current-error-port))) (close-output-port *default-log-port*)) (set! *default-log-port* (current-error-port))) "Cleanup db exit thread")) (th2 (make-thread (lambda () (debug:print 4 *default-log-port* "Attempting clean exit. Please be patient and wait a few seconds...") (if no-hurry (begin (thread-sleep! 5)) ;; give the clean up few seconds to do it's stuff (begin (thread-sleep! 2))) (debug:print 4 *default-log-port* " ... done") ) "clean exit"))) (thread-start! th1) (thread-start! th2) (thread-join! th1) ) |
︙ | ︙ | |||
1537 1538 1539 1540 1541 1542 1543 | #f (conc "http://" (car hostport) ":" (cadr hostport)))) ;;====================================================================== ;; S E R V E R ;; ====================================================================== | < < < < < < < < < < < < < < < < < < < < < < < < < | 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 | #f (conc "http://" (car hostport) ":" (cadr hostport)))) ;;====================================================================== ;; 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 hostn) ;; ;; Configurations for server ;; (tcp-buffer-size 2048) ;; (max-connections 2048) (debug:print 2 *default-log-port* "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)) |
︙ | ︙ | |||
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))) | | < < < | | < < < < | > | | | | | > > > > > > > > > > > > > > > > > > > > | | | | | | | | | | | | | | | | < < < < | < | | 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 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 | (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* ((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) (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 | | | | | | < | | | > > > | | | | | | | | | | 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 | 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 |
︙ | ︙ | |||
1968 1969 1970 1971 1972 1973 1974 | (exit)) (loop start-time (equal? sdat last-sdat) sdat)))))))) (define (rmt:register-server remote apath iface port server-key dbname) (rmt:open-main-connection remote apath) ;; we need a channel to main.db | | | > | | | | | | | 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 | (exit)) (loop start-time (equal? sdat last-sdat) sdat)))))))) (define (rmt:register-server remote apath iface port server-key dbname) (rmt:open-main-connection remote apath) ;; we need a channel to main.db (rmt:send-receive-real remote 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 (http-transport:wait-for-stable-interface #!optional (num-tries-allowed 100)) ;; wait until *server-info* stops changing (let* ((stime (current-seconds))) (let loop ((last-host #f) (last-port #f) (tries 0)) |
︙ | ︙ | |||
2042 2043 2044 2045 2046 2047 2048 | ;; this is our forever loop (let* ((iface (servdat-host *server-info*)) (port (servdat-port *server-info*))) (let loop ((count 0) (bad-sync-count 0) (start-time (current-milliseconds))) | | > | 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 | ;; this is our forever loop (let* ((iface (servdat-host *server-info*)) (port (servdat-port *server-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*))) ;; set up the database handle (mutex-lock! *heartbeat-mutex*) (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") |
︙ | ︙ | |||
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 ;; | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 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 | (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 resp))) "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)) |
︙ | ︙ |