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
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
|
(close-output-port oup)
res)))))) ;; res will always be 'ack
;; 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* ((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)))
(if (eq? (send uconn host-port qrykey cmd data) 'ack)
(let* ((mbox-timeout-secs 120) ;; timeout)
(mbox-timeout-result 'MBOX_TIMEOUT)
(res (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result))
(mbox-receive-time (current-milliseconds)))
(if (eq? res 'MBOX_TIMEOUT)
#f ;; convert to raising exception?
res))
(begin
(print "ERROR: Communication failed?")
#f)))) ;; #f means failed to communicate
;;======================================================================
;; 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)
;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params)
(let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
(case cmd
;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
((ping)
;; (print "Got Ping!")
(add-to-work-queue uconn rdat)
'ack)
((goodbye)
;; just clear out references to the caller
(add-to-work-queue uconn rdat)
'ack)
((response) ;; this is a result from remote processing, send it as mail ...
(if mbox
|
>
>
>
>
|
|
|
|
|
|
>
>
|
|
|
>
|
>
>
|
|
|
|
|
|
|
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
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
|
(close-output-port oup)
res)))))) ;; res will always be 'ack
;; 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)
(cond
((member cmd '(ping goodbye)) ;; these are immediate
(send uconn host-port 'ping cmd data))
(else
(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)))
(if (eq? (send uconn host-port qrykey cmd data) 'ack)
(let* ((mbox-timeout-secs (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?
(if (eq? res 'MBOX_TIMEOUT)
(begin
(print "WARNING: mbox timed out for query "cmd", with data "data)
#f) ;; convert to raising exception?
res))
(begin
(print "ERROR: Communication failed?")
#f)))))) ;; #f means failed to communicate
;;======================================================================
;; 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)
;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params)
(let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
(case cmd
;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
((ping)
;; (print "Got Ping!")
;; (add-to-work-queue uconn rdat)
'ack)
((goodbye)
;; just clear out references to the caller
(add-to-work-queue uconn rdat)
'ack)
((response) ;; this is a result from remote processing, send it as mail ...
(if mbox
|