Megatest

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

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

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