Megatest

Check-in [d7b4fe7a7f]
Login
Overview
Comment:Sync up with v2.0001
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | v2.0001-ulex-one-shot
Files: files | file ages | folders
SHA1: d7b4fe7a7fa00bb8d2d8f34ef010754e00639ca5
User & Date: matt on 2022-01-12 16:52:23
Other Links: branch diff | manifest | tags
Context
2022-01-12
17:49
Switched back to tcp6 check-in: 1b8dcc586b user: matt tags: v2.0001-ulex-one-shot
16:52
Sync up with v2.0001 check-in: d7b4fe7a7f user: matt tags: v2.0001-ulex-one-shot
16:40
wip, misc cleanup and reduce some messages. check-in: 20b4054f76 user: matt tags: v2.0001
2022-01-10
12:54
wip check-in: 9ae53d1765 user: matt tags: v2.0001-ulex-one-shot
Changes

Modified apimod.scm from [97d3b608d8] to [37041542cf].

173
174
175
176
177
178
179
180
181



182
183
184
185
186
187
188
173
174
175
176
177
178
179


180
181
182
183
184
185
186
187
188
189







-
-
+
+
+







	 (logd        (conc apath "/logs")) 
	 (logf        (conc logd "/server-launch-";;(current-process-id)
			    (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
;; look up in db
;; if found -> return it

Modified dashboard.scm from [cc72246a09] to [4505f63ba6].

3640
3641
3642
3643
3644
3645
3646
3647

3648
3649
3650
3651
3652
3653
3654
3640
3641
3642
3643
3644
3645
3646

3647
3648
3649
3650
3651
3652
3653
3654







-
+







(define (dashboard:do-update-rundat tabdat)
  (dboard:update-rundat
   tabdat
   (hash-table-ref/default (dboard:tabdat-searchpatts tabdat) "runname" "%")
   (dboard:tabdat-numruns tabdat)
   (hash-table-ref/default (dboard:tabdat-searchpatts tabdat) "test-name" "%/%")
   ;; generate key patterns from the target stored in tabdat
   (let* ((dbkeys (dboard:tabdat-dbkeys tabdat)))
   (let* ((dbkeys  (dboard:tabdat-dbkeys tabdat)))
     (let ((fres   (if (dboard:tabdat-target tabdat)
                       (let ((ptparts (append (dboard:tabdat-target tabdat)(make-list (length dbkeys) "%"))))
                         (map (lambda (k v)(list k v)) dbkeys ptparts))
                       (let ((res '()))
                         (for-each (lambda (key)
                                     (if (not (equal? key "runname"))
                                         (let ((val (hash-table-ref/default (dboard:tabdat-searchpatts tabdat) key #f)))

Modified dbmod.scm from [4d2069b432] to [ac637164a6].

691
692
693
694
695
696
697


698

699
700
701
702
703
704
705
691
692
693
694
695
696
697
698
699

700
701
702
703
704
705
706
707







+
+
-
+







;; ;;     (mutex-unlock! *db-multi-sync-mutex*)
;; ;;     (stack-push! (dbr:dbstruct-dbstack dbstruct) tmpdb)))

;; NOTE: touched logic is disabled/not done
;; sync run to disk if touched
;;
(define (db:sync-inmem->disk dbstruct apath dbfile #!key (force-sync #f))
  (if #f
      (debug:print-info 0 *default-log-port* "syncing "*toppath*" "dbfile" at "(current-seconds))
  #f) ;; disabled
      #f)) ;; disabled
;;   (let* ((dbdat       (db:get-dbdat dbstruct apath dbfile))
;; 	 (dbfullname  (conc apath "/" dbfile))
;; 	 (db          (db:open-run-db dbfullname db:initialize-db)) ;; (dbr:dbdat-db dbdat))
;; 	 (inmem       (dbr:dbdat-inmem dbdat))
;; 	 (start-t     (current-seconds))
;; 	 (last-update (dbr:dbdat-last-write dbdat))
;; 	 (last-sync   (dbr:dbdat-last-sync dbdat)))

Modified rmtmod.scm from [63c06f1ce1] to [c55aca3d24].

2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2136
2137
2138
2139
2140
2141
2142

2143
2144
2145
2146
2147
2148
2149







-







			(debug:print-info 0 *default-log-port* "Starting watchdog thread (in state "(thread-state watchdog)")")
			(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*)
	
	;; when things go wrong we don't want to be doing the various
	;; queries too often so we strive to run this stuff only every
	;; four seconds or so.

Modified ulex-trials/Makefile from [e108f86d9a] to [cec464a43d].

1

2
3
4
5

6
7
8

1
2
3
4

5
6
7
8
-
+



-
+



ulex-test : ulex-test.scm
ulex-test : ulex-test.scm ../ulex/ulex.scm
	csc ulex-test.scm

test : ulex-test
	./ulex-test do-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 .runners/* NBFAKE*

Modified ulex-trials/ulex-test.scm from [563b467581] to [f76ffe0828].

22
23
24
25
26
27
28
29
30

31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

91
92

93
94
95
96
97
98
99
22
23
24
25
26
27
28


29
30
31
































32


























33


34
35
36
37
38
39
40
41







-
-
+


-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-
-
+







	directory-utils

	ulex
	)

(define help "Usage: ulex-test COMMAND
  where COMMAND is one of:
    do-test              : run the basic req/rep test
    run tcp://host:port  : start test server - start several in same dir
    run host:port  : start test server - start several in same dir
")

(define address-tcp-1 "tcp://localhost:5555")
(define address-tcp-2 "tcp://localhost:6666")

(define address-inproc-1 "inproc://local1")
(define address-inproc-2 "inproc://local2")

;;;
;;; Req-Rep
;;;
(define (make-listening-reply-socket address)
  (let ((socket (make-rep-socket)))
    (socket-set! socket 'nng/recvtimeo 2000)
    (nng-listen socket address)
    socket))

(define (make-dialed-request-socket address)
  (let ((socket (make-req-socket)))
    (socket-set! socket 'nng/recvtimeo 2000)
    (nng-dial socket address)
    socket))

(define (req-rep-test address)
  (let ((rep (make-listening-reply-socket address))
        (req (make-dialed-request-socket address)))
    (nng-send req "message 1")
    (nng-recv rep)
    (nng-send rep "message")
    (begin0
     (nng-recv req)
     (nng-close! rep))))

(define (do-test)
(define (call uconn msg addr)
  (test-group "nng"
              (test "tcp req-rep"
                    "message"
                    (req-rep-test address-tcp-1))
              (test "inproc req-rep"
                    "message"
                    (req-rep-test address-inproc-1)))
  (test-exit))

;; this should be run in a thread
(define (run-listener-responder socket myaddr)
  (let loop ((status 'running))
    (let* ((msg (nng-recv socket))
	   (response (process-message msg)))
      (if (not (eq? response 'done))
	  (begin
	    (nng-send socket response)
	    (loop status))))))

(define *channels* (make-hash-table))

(define (call channels msg addr)
  (let* ((csocket (hash-table-ref/default channels addr #f))
	 (socket  (or csocket (make-dialed-request-socket addr))))
    (nng-send socket msg)
    (print "Sent: "msg", received: "(nng-recv socket))
  (print "Sent: "msg", received: "
    (if (not (hash-table-exists? channels addr))
	(hash-table-set! channels addr socket))))
	 (send-receive uconn addr 'hello msg)))

;; start    => hello 0
;; hello 0  => hello 1
;; hello 1  => hello 2
;;  ...
;; hello 11 => 'done
;;
110
111
112
113
114
115
116
117
118

119
120
121
122
123
124

125
126
127





128
129
130
131
132
133
134
135
136
137

138
139
140
141
142
143
144
145
146
147
148
149

150
151
152
153
154
155
156
157
158

159
160
161
52
53
54
55
56
57
58


59
60
61
62
63
64
65
66



67
68
69
70
71




72
73
74
75
76

77

78
79
80
81
82
83
84
85
86
87

88
89
90
91
92
93
94
95
96

97
98
99
100







-
-
+






+
-
-
-
+
+
+
+
+
-
-
-
-





-
+
-










-
+








-
+



      (conc msg " 0"))
     (else
      "hello 0"))))

(define (main)
  (match
   (command-line-arguments)
   (("do-test")(do-test))
   ((run myaddr)
   ((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))
    (let* ((endtimes (+ (current-seconds) 20)) ;; run for 20 seconds
	   (socket (make-listening-reply-socket myaddr))
	   (rfile  (conc ".runners/"(get-host-name)"-"(current-process-id)))
	   (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))))
	   (th1    (make-thread (lambda ()
				  (run-listener-responder socket myaddr)
				  )
				"responder")))
      (if (not (and (file-exists? ".runners")
		    (directory? ".runners")))
	  (create-directory ".runners" #t))
      (with-output-to-file rfile
	(lambda ()
	  (print myaddr)))
	  (print myport)))
      (thread-start! th1)
      (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 *channels* (conc "hello-from-"destaddr)  destaddr)
		  (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 nng-test)
(import ulex-test)
(main)


Modified ulex.scm from [64369b6c76] to [f004a2cedd].

16
17
18
19
20
21
22
23
24


16
17
18
19
20
21
22


23
24







-
-
+
+
;;     You should have received a copy of the GNU General Public License
;;     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.

;;======================================================================

(declare (unit ulex))

;; (include "ulex/ulex.scm")
(include "ulex-simple/ulex.scm")
(include "ulex/ulex.scm")
;; (include "ulex-simple/ulex.scm")

Modified ulex/ulex.scm from [ded9484f4d] to [81b8992868].

22
23
24
25
26
27
28
29


30
31
32
33
34
35
36
22
23
24
25
26
27
28

29
30
31
32
33
34
35
36
37







-
+
+







;;   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)
105
106
107
108
109
110
111
112

113
114
115
116
117
118
119

120
121
122
123
124
125
126
106
107
108
109
110
111
112

113
114
115
116
117
118
119

120
121
122
123
124
125
126
127







-
+






-
+







  (work-queue-thread #f)
  (num-threads-running 0)
  ) 

;; Parameters

;; work-method:
(define work-method (make-parameter 'direct))
(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 'direct))
(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
;; ;;
216
217
218
219
220
221
222
223

224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243



















































244
245
246
247
248

249
250
251
252
253
254
255
256
257









258
259

260
261

262
263
264
265

266
267
268
269
270
271
272
273
274
275
276
277



278
279
280
281
282


283
284
285
286
287
288
289
290

291
292
293
294
295
296


297
298
299

300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320

321
322
323
324
325
326
327
217
218
219
220
221
222
223

224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301









302
303
304
305
306
307
308
309
310


311


312




313












314
315
316





317
318








319






320
321



322








323
324
325
326
327
328
329
330
331
332
333
334

335
336
337
338
339
340
341
342







-
+




















+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+





+
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
-
-
+
-
-
+
-
-
-
-
+
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-
-
-
-
-
+
+
-
-
-
-
-
-
-
-
+
-
-
-
-
-
-
+
+
-
-
-
+
-
-
-
-
-
-
-
-












-
+







;;        - I believe (without substantial evidence) that re-using connections will
;;          be beneficial ...
;;
(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)))
	 (dat          (list my-host-port qrykey cmd params #;(cons (current-seconds)(current-milliseconds)))))
    (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
			     (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)
	      ;; (mutex-unlock! *send-mutex*)
	      res)))))))) ;; res will always be 'ack unless return-method is direct

(define (send-via-polling uconn host-port cmd data)
  (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))))

(define (send-via-mailbox 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))
	 (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))))
  
;; 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* ((start-time (current-milliseconds))
  (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)
	 (result     (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)))
			  (send-via-polling uconn host-port cmd data))
	 (case sres
	   ((ack)
			 ((mailbox) 
	    (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)
			  (send-via-mailbox uconn 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))))
			 (else
			  (print "ULEX ERROR: unrecognised return-method "(return-method)".")
			  #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))
    ;; this is ONLY for development and debugging. It will be removed once Ulex is stable.
    (if (< 5000 (- (current-milliseconds) start-time))
	      (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)))
	(print "ULEX WARNING: round-trip took over 5 seconds; "
	       ;; (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".")

	       cmd", host-port="host-port", data="data))
    result))
		     ;; 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, 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)
    ((rem-host-port qrykey cmd params);; timedata)
     ;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params)
     (case cmd
       ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
       ((ping)
	;; (print "Got Ping!")
	;; (add-to-work-queue uconn rdat)
	'ack)