Megatest

Check-in [c6f20213d4]
Login
Overview
Comment:Rewire inputs and outputs to address tcp-server stuff.
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | v2.0001
Files: files | file ages | folders
SHA1: c6f20213d421d714b31612c7c3c3187f607544a5
User & Date: matt on 2022-01-19 18:58:10
Other Links: branch diff | manifest | tags
Context
2022-01-20
07:35
Add exception handler to decoder and reduce some noise check-in: 5ce7b7ae18 user: matt tags: v2.0001
2022-01-19
18:58
Rewire inputs and outputs to address tcp-server stuff. check-in: c6f20213d4 user: matt tags: v2.0001
13:27
Added inmem check-in: f1db41ee97 user: matt tags: v2.0001
Changes

Modified ulex-simple/ulex.scm from [4e7e7a6994] to [9d24cf68f8].

120
121
122
123
124
125
126

127
128
129
130
131
132
133
;;======================================================================
;; serialization
;;   NOTE: I've had problems with read/write and s11n serialize, deserialize
;;         thus the inefficient method here
;;======================================================================

(define serializing-method (make-parameter 'complex))


;; NOTE: Can remove the regex and base64 encoding for zmq
(define (obj->string obj)
  (case (serializing-method)
    ((complex)
     (string-substitute
      (regexp "=") "_"







>







120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
;;======================================================================
;; serialization
;;   NOTE: I've had problems with read/write and s11n serialize, deserialize
;;         thus the inefficient method here
;;======================================================================

(define serializing-method (make-parameter 'complex))


;; NOTE: Can remove the regex and base64 encoding for zmq
(define (obj->string obj)
  (case (serializing-method)
    ((complex)
     (string-substitute
      (regexp "=") "_"
201
202
203
204
205
206
207


208
209
210

211







212



213



214
215
216
217
218
219
220
;; 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)


	((make-tcp-server
	  (udat-socket uconn)
	  (lambda ()

	    (let* ((rdat  (string->obj (read)) #;(deserialize)) ;; '(my-host-port qrykey cmd params)







		   (resp  (do-work uconn rdat)))



	      (write (obj->string resp)) #;(serialize resp)))))



	(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)))

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







>
>
|
|
|
>
|
>
>
>
>
>
>
>
|
>
>
>
|
>
>
>







202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
;; 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* ((orig-in  (current-input-port))
	       (orig-out (current-output-port)))
	  ((make-tcp-server
	    (udat-socket uconn)
	    (lambda ()
	      (let* ((rdat
		      (string->obj (read))
		      ;; (read in)
		      ;; (deserialize)
		      ) 
		     (resp (let ((tcp-in  (current-input-port))
				 (tcp-out (current-output-port)))
			     (current-input-port orig-in)
			     (current-output-port orig-out)
			     (let ((res (do-work uconn rdat)))
			       (current-input-port tcp-in)
			       (current-output-port tcp-out)
			       res))))
		(write (obj->string resp))
		;; (serialize resp)
		;; (write resp out)
		)))))
	(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)))

;;======================================================================
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
	 (my-host-port (and udata (udat-host-port udata)))          ;; remote will return to this
	 (isme         (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 #;(cons (current-seconds)(current-milliseconds)))))
    (cond
     (isme (do-work udata dat)) ;; no transmission needed
     (else
      (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
       exn
       (begin
         (print "ULEX send-receive: exn="exn)
         (message exn))
       (begin
	 ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
	 (let-values (((inp oup)(tcp-connect host port)))
	   (let ((res (if (and inp oup)
			  (begin
			    (write (obj->string dat) oup) ;; (write dat oup);; (serialize dat oup)


			    (close-output-port oup)
			    (string->obj (read inp))) ;; (deserialize inp))



			  (begin
			    (print "ERROR: send called but no receiver has been setup. Please call setup first!")
			    #f))))
	     ;; (close-output-port oup)
	     (close-input-port inp)
	     ;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
	     res)))))))) ;; res will always be 'ack unless return-method is direct

;;======================================================================
;; work queues - this is all happening on the listener side
;;======================================================================

;; move the logic to return the result somewhere else?
;;
(define (do-work uconn rdat)
  (let* () ;; get it each time - conceivebly it could change
    ;; put this following into a do-work procedure
    (match rdat
      ((rem-host-port qrykey cmd params)
       (case cmd
	 ((ping) 'ping-ack) ;; bypass calling the proc
	 (else
	  (let* ((proc       (udat-work-proc uconn))
		 (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")))))

;;======================================================================
;; misc utils
;;======================================================================

(define (pp-uconn uconn)
  (pp (udat->alist uconn)))







|
|
|
|
|
|




|
>
>

|
>
>
>






|








<
|
|
|
|
|
|
|
|
>
>
|
|
|
|
|
|







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
	 (my-host-port (and udata (udat-host-port udata)))          ;; remote will return to this
	 (isme         (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 #;(cons (current-seconds)(current-milliseconds)))))
    (cond
     (isme (do-work udata dat)) ;; no transmission needed
     (else
      ;; (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
      ;;  exn
      ;;  (begin
      ;;    (print "ULEX send-receive: exn="exn)
      ;;    (message exn))
      ;;  (begin
	 ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
	 (let-values (((inp oup)(tcp-connect host port)))
	   (let ((res (if (and inp oup)
			  (begin
			    (write (obj->string dat) oup)
			    ;; (write dat oup)
			    ;; (serialize dat oup)
			    (close-output-port oup)
			    (string->obj (read inp))
			    ;; (read inp)
			    ;; (deserialize inp)
			    )
			  (begin
			    (print "ERROR: send called but no receiver has been setup. Please call setup first!")
			    #f))))
	     ;; (close-output-port oup)
	     (close-input-port inp)
	     ;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
	     res))))));; )) ;; res will always be 'ack unless return-method is direct

;;======================================================================
;; work queues - this is all happening on the listener side
;;======================================================================

;; move the logic to return the result somewhere else?
;;
(define (do-work uconn rdat)

  ;; put this following into a do-work procedure
  (match rdat
     ((rem-host-port qrykey cmd params)
      (case cmd
	((ping) 'ping-ack) ;; bypass calling the proc
	(else
	 (let* ((proc       (udat-work-proc uconn))
		(start-time (current-milliseconds))
		(result     (with-output-to-port (current-error-port)
			      (lambda ()
				(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"))))

;;======================================================================
;; misc utils
;;======================================================================

(define (pp-uconn uconn)
  (pp (udat->alist uconn)))