Megatest

Diff
Login

Differences From Artifact [88c5ce4192]:

To Artifact [2906a4f6cb]:


22
23
24
25
26
27
28

29





















30
31
32
33
34
35
36
;;   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

 *






















(import scheme
	chicken.base
	chicken.file
	chicken.time
	chicken.condition
	chicken.string







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







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
;;   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
    (
     
     run-listener     ;; (run-listener handler-proc [port]) => uconn

     ;; NOTE: handler-proc params;
     ;;       (handler-proc rem-host-port qrykey cmd params)

     send-receive     ;; (send-receive uconn host-port cmd data)

     ;; NOTE: cmd can be any plain text symbol except for these;
     ;;         'ping 'ack 'goodbye 'response
     
     set-work-handler ;; (set-work-handler proc)

     wait-and-close   ;; (wait-and-close uconn)

     ;; needed to get the interface:port that was automatically found
     udat-port
     udat-host-port
     
     ;; for testing only
     ;; pp-uconn
     )

(import scheme
	chicken.base
	chicken.file
	chicken.time
	chicken.condition
	chicken.string
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
  (peers  (make-hash-table)) ;; host:port->peer
  ;; work handling
  (work-queue (make-queue))
  (work-proc  #f)            ;; set by user
  (cnum       0)             ;; cookie number
  (mboxes     (make-hash-table))
  (avail-cmboxes '())  ;; list of (<cookie> . <mbox>) for re-use



  ) 

;; struct for keeping track of others we are talking to
;;
(defstruct pdat
  (host-port  #f)
  (conns      '()) ;; list of pcon structs, pop one off when calling the peer
  )


;; struct for peer connections, keep track of expiration etc.
;;
(defstruct pcon
  (inp #f)
  (oup #f)
  (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59)
  (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes
  )


;;======================================================================
;; listener
;;======================================================================
;; 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.







>
>
>


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







86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

103
104
105
106
107
108
109
110
111

112
113
114
115
116
117
118
119
  (peers  (make-hash-table)) ;; host:port->peer
  ;; work handling
  (work-queue (make-queue))
  (work-proc  #f)            ;; set by user
  (cnum       0)             ;; cookie number
  (mboxes     (make-hash-table))
  (avail-cmboxes '())  ;; list of (<cookie> . <mbox>) for re-use
  ;; threads
  (cmd-thread #f)
  (work-queue-thread #f)
  ) 

;; ;; struct for keeping track of others we are talking to
;; ;;
;; (defstruct pdat
;;   (host-port  #f)
;;   (conns      '()) ;; list of pcon structs, pop one off when calling the peer

;;   )
;; 
;; ;; struct for peer connections, keep track of expiration etc.
;; ;;
;; (defstruct pcon
;;   (inp #f)
;;   (oup #f)
;;   (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59)
;;   (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes

;;   )

;;======================================================================
;; listener
;;======================================================================
;; 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.
109
110
111
112
113
114
115





















116
117
118
119
120
121
122
  ;; sockaddr-address, sockaddr-port, sockaddr->string
  (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]])
	 (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname)))
    (udat-port-set!      uconn port)
    (udat-host-port-set! uconn (conc addr":"port))
    (udat-socket-set!    uconn tlsn)
    uconn))






















;;======================================================================
;; peers and connections
;;======================================================================

;; send structured data to recipient
;;







>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>







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
162
163
164
165
166
167
168
  ;; sockaddr-address, sockaddr-port, sockaddr->string
  (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]])
	 (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname)))
    (udat-port-set!      uconn port)
    (udat-host-port-set! uconn (conc addr":"port))
    (udat-socket-set!    uconn tlsn)
    uconn))

;; 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 #f))
  (let* ((uconn (make-udat)))
    (udat-work-proc-set! uconn handler-proc)
    (if (setup-listener uconn port-suggestion)
	(let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop"))
	       (th2 (make-thread (lambda ()(process-work-queue uconn)) "Ulex work queue processor")))
	  (thread-start! th1)
	  (thread-start! th2)
	  (udat-cmd-thread-set! uconn th1)
	  (udat-work-queue-thread-set! uconn th2)
	  (print "cmd loop and process workers started")
	  uconn)
	(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)))

;;======================================================================
;; peers and connections
;;======================================================================

;; send structured data to recipient
;;
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

;; add a proc to the cmd list, these are done symetrically (i.e. in all instances)
;; so that the proc can be dereferenced remotely
;;
(define (set-work-handler uconn proc)
  (udat-work-proc-set! uconn proc))

;; run-listener does all the work of starting a listener in a thread
;; it then returns control
;;
(define (run-listener handler-proc)
  (let* ((uconn (make-udat)))
    (udat-work-proc-set! uconn handler-proc)
    (if (setup-listener uconn)
	(let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop"))
	       (th2 (make-thread (lambda ()(process-work-queue uconn)) "Ulex work queue processor")))
	  (thread-start! th1)
	  (thread-start! th2)
	  (print "cmd loop and process workers started")
	  uconn)
	(begin
	  (print "ERROR: run-listener called without proper setup.")
	  (exit)))))

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

;; rdata is (rem-host-port qrykey cmd params)
					     
(define (add-to-work-queue uconn rdata)







<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<







273
274
275
276
277
278
279

















280
281
282
283
284
285
286

;; add a proc to the cmd list, these are done symetrically (i.e. in all instances)
;; so that the proc can be dereferenced remotely
;;
(define (set-work-handler uconn proc)
  (udat-work-proc-set! uconn proc))


















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

;; rdata is (rem-host-port qrykey cmd params)
					     
(define (add-to-work-queue uconn rdata)
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423

;; (map ip->string (vector->list 
;; 		   (hostinfo-addresses
;; 		    (host-information (current-hostname))))))


)

(import ulex trace big-chicken srfi-18 test matchable system-information)
(trace-call-sites #t)
(trace
 ;; ulex-handler
 ;; send
 ;; add-to-work-queue
 )
 
(define (handler-proc rem-host-port qrykey cmd params)
  (print "handler-proc "rem-host-port" "qrykey" "cmd" "params)
  (case cmd
    ((ping) 'pong)
    ((calc)  (eval (with-input-from-string params read)))
    ((print)
     (print "params="params)
     params)
    ((reflect) `(,rem-host-port ,qrykey ,cmd ,params))
    (else  `(data ,data))))

(define uconn (run-listener handler-proc))

(pp-uconn uconn)

;; super basic loop back test
(define res #f)
(define targhost (conc (get-host-name)":"4242))
(define th1 (make-thread (lambda ()
			   (test #f 10 (send-receive uconn targhost 'calc "(+ 5 5)"))
			   (set! res (send-receive uconn targhost 'ping '()))
			   (test #f 'pong (send-receive uconn targhost 'ping '()))
			   )))

(thread-start! th1)
(thread-join! th1)
		
(print "All done")
(print "Received "res)








<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
407
408
409
410
411
412
413








































;; (map ip->string (vector->list 
;; 		   (hostinfo-addresses
;; 		    (host-information (current-hostname))))))


)