Megatest

all : ulex.pdf ulex.png

ulex.pdf : ulex.dot
	dot -Tpdf ulex.dot -o ulex.pdf

ulex.png : ulex.dot
	dot -Tpng ulex.dot -o ulex.png

//  Copyright 2006-2017, Matthew Welland.
// 
// This file is part of Megatest.
// 
//     Megatest is free software: you can redistribute it and/or modify
//     it under the terms of the GNU General Public License as published by
//     the Free Software Foundation, either version 3 of the License, or
//     (at your option) any later version.
// 
//     Megatest is distributed in the hope that it will be useful,
//     but WITHOUT ANY WARRANTY; without even the implied warranty of
//     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//     GNU General Public License for more details.
// 
//     You should have received a copy of the GNU General Public License
//     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.

digraph G {

  // graph[center=true, margin=0.2, nodesep=0.1, ranksep=0.3]

  layout=neato;
  // layout=fdp;
  // overlap=scalexy; //false, compress, ...
  overlap=scalexy;
  // sep="+1"; // 0.1, +1
  sep="-0.1";
 
  user_program [label="user program"];
  
  subgraph cluster_1 {
    node [style=filled,shape=oval];
    label = "caller";
    color=brown;

    send_receive [label="(send-receive uconn\n   host-port qrykey cmd data)"];
    send [label="(send uconn host-port\n    qrykey cmd data)"];
    ulex_cmd_loopcaller [label="(ulex-cmd-loop uconn)"];
    ulex_handlercaller [label="(ulex-handler uconn rdat)"];
    mailbox [label="mailbox\n\nrdat\n...",shape=box];
    
    send_receive -> send;
    ulex_cmd_loopcaller -> ulex_handlercaller;
    ulex_handlercaller -> mailbox;
    mailbox -> send_receive;
  }

  subgraph cluster_2 {
    node [shape=oval];
    label = "listener";
    color=green;

    ulex_cmd_loop [label="(ulex-cmd-loop uconn)"];
    ulex_handler [label="(ulex-handler \nuconn rdat)"];
    add_to_work_queue [label="(add-to-work-queue\n  uconn rdat)"];
    queue [label="queue\n\nrdat\n...",shape=box];
    process_work_queue [label="(process-work-queue uconn)"];
    do_work [label="(do-work uconn rdat)\nrdat: '(rem-host-port qrykey cmd params)"];
    user_proc [label="(proc rem-host-port\n    qrykey cmd params)\n;; proc supplied by user"];
    sendlis [label="(send uconn host-port\n    qrykey 'response result)"];
    
    ulex_cmd_loop -> ulex_handler [label="rdat"];
    ulex_handler -> add_to_work_queue [label="rdat"];

    add_to_work_queue -> queue [label="rdat"];

    subgraph cluster_3 {
      label = "remote work";
      color=blue;
      
      queue -> process_work_queue [label="rdat"];
      process_work_queue -> do_work [label="rdat"];
      do_work -> user_proc; // [label="rdat: '(rem-host-port\n   qrykey cmd params)"];
    }
  }

  user_proc -> sendlis;
  user_program -> send_receive;
  send_receive -> user_program;
  
  send -> ulex_cmd_loop [label="rdat: '(host-port\n  qrykey cmd data)"];
  sendlis -> ulex_cmd_loopcaller [label="rdat: '(host-port qrykey\n  'response result)"];
  ulex_handler -> send [label="'ack"];
  ulex_handlercaller -> sendlis [label="'ack"];
  
}


// 	check_available_queue       -> remove_entries_over_10s_old;
// 	remove_entries_over_10s_old -> set_available [label="num_avail < 3"];
// 	remove_entries_over_10s_old -> exit [label="num_avail > 2"];
// 
// 	set_available               -> delay_2s;
// 	delay_2s          -> check_place_in_queue;
// 
// 	check_place_in_queue        -> "http:transport-launch" [label="at head"];
// 	check_place_in_queue        -> exit [label="not at head"];
// 
// 	"client:login"              -> "server:shutdown" [label="login failed"];
// 	"server:shutdown"           -> exit;	
// 
// 	subgraph cluster_2 {
// 		"http:transport-launch"       -> "http:transport-run";
// 		"http:transport-launch"       -> "http:transport-keep-running";
// 
// 		"http:transport-keep-running" -> "tests running?";
// 		"tests running?"              -> "client:login" [label=yes];
// 		"tests running?"              -> "server:shutdown" [label=no];
// 		"client:login"                -> delay_5s [label="login ok"];
// 		delay_5s                      -> "http:transport-keep-running";
// 	}
// 
	// start_server -> "server_running?";
	// "server_running?" -> set_available [label="no"];
	// "server_running?" -> delay_2s [label="yes"];
	// delay_2s -> "still_running?";
	// "still_running?" -> ping_server [label=yes];
	// "still_running?" -> set_available [label=no];
	// ping_server -> exit [label=alive];
	// ping_server -> remove_server_record [label=dead];
	// remove_server_record -> set_available;
	// set_available -> avail_delay [label="delay 3s"];
	// avail_delay -> "first_in_queue?";
	// 
	// "first_in_queue?" -> set_running [label=yes];
	// set_running -> get_next_port -> handle_requests;
	// "first_in_queue?" -> "dead_entry_in_queue?" [label=no];
	// "dead_entry_in_queue?" -> "server_running?" [label=no];
	// "dead_entry_in_queue?" -> "remove_dead_entries" [label=yes];
	// remove_dead_entries -> "server_running?";
	// 
	// handle_requests -> start_shutdown [label="no traffic\nno running tests"];
	// handle_requests -> shutdown_request;
	// start_shutdown -> shutdown_delay;
	// shutdown_request -> shutdown_delay;
	// shutdown_delay -> exit;

	  (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)
  (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
	      (begin
		(mailbox-send! mbox params) ;; params here is our result
		'ack)
	      (begin
		(print "ERROR: received result but no associated mbox for cookie "qrykey)
		#f)))
	 (else
	  ;; (print "Got generic request: "cmd)
	  (add-to-work-queue uconn rdat)
	  'ack))))
    (else
     (print "BAD DATA? controldat=" rdat)
     'ack) ;; send ack anyway?
    ))

;; given an already set up uconn start the cmd-loop
;;
(define (ulex-cmd-loop uconn)
  (let* ((serv-listener (udat-socket uconn)))
    (let loop ((state 'start))
      (let-values (((inp oup)(tcp-accept serv-listener)))
	(let* ((rdat  (read inp)) ;; '(my-host-port qrykey cmd params)
	       (resp  (ulex-handler uconn rdat)))
	  (if resp (write resp oup))
	  (close-input-port inp)
	  (close-output-port oup))
	(loop state)))))

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

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

(define (do-work uconn rdat)
  (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change
    ;; put this following into a do-work procedure
    (match rdat
      ((rem-host-port qrykey cmd params)
       (let* ((result (proc rem-host-port qrykey cmd params)))
	 ;; send 'response as cmd and result as params
	 (send uconn rem-host-port qrykey 'response result))) ;; could check for ack
      (else
       (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params")))))
     
     
(define (process-work-queue uconn) 
  (let ((wqueue (udat-work-queue uconn))
	(proc   (udat-work-proc  uconn)))
    (let loop ()
      (if (queue-empty? wqueue)
	  (thread-sleep! 0.1)
	  (let ((rdat (queue-remove! wqueue)))
	    (do-work uconn rdat)))
      (loop))))

;; below was to enable re-use of connections. This seems non-trivial so for
;; now lets open on each call
;;
;; ;; given host-port get or create peer struct
;; ;;