Megatest

    ))

;; These are called by the server on recipt of /api calls
;;    - keep it simple, only return the actual result of the call, i.e. no meta info here
;;
;;    - returns #( flag result )
;;
(define (api:execute-requests dbstruct dat queues)
  (handle-exceptions
   exn
   (let ((call-chain (get-call-chain)))
     (debug:print 0 *default-log-port* "WARNING: api:execute-requests received an exception from peer, dat=" dat)
     (print-call-chain (current-error-port))
     (debug:print 0 *default-log-port* " message: "  ((condition-property-accessor 'exn 'message) exn))       
     (vector #f (vector exn call-chain dat))) ;; return some stuff for debug if an exception happens

             (if writecmd-in-readonly-mode
                 (conc "attempt to run write command "cmd" on a read-only database")
                 (case cmd
                   ;;===============================================
                   ;; READ/WRITE QUERIES
                   ;;===============================================

                   ((get-keys-write)                  (api:queued-request queues 'write params
									  (lambda ()
									    (db:get-keys dbstruct)))) ;; force a dummy "write" query to force server; for debug in -repl
                   
                   ;; SERVERS
                   ((start-server)                    (apply server:kind-run params))
                   ((kill-server)                     (set! *server-run* #f))

                   ;; TESTS

;; NB// Runs on the server as part of the server loop
;;
(define (api:process-request dbstruct $) ;; the $ is the request vars proc
  (set! *api-process-request-count* (+ *api-process-request-count* 1))
  (let* ((cmd     ($ 'cmd))
	 (paramsj ($ 'params))
	 (params  (db:string->obj paramsj transport: 'http)) ;; incoming data from the POST (or is it a GET?)
	 (resdat  (api:execute-requests dbstruct (vector cmd params) *queues*)) ;; process the request, resdat = #( flag result ), we resort to a global here for the queues.
	 (success (vector-ref resdat 0))
	 (res     (vector-ref resdat 1))) ;; (vector flag payload), get the payload, ignore the flag (why?)
    (if (not success)
	(debug:print 0 *default-log-port* "ERROR: success flag is #f for " cmd " with params " params))
    (if (> *api-process-request-count* *max-api-process-requests*)
	(set! *max-api-process-requests* *api-process-request-count*))
    (set! *api-process-request-count* (- *api-process-request-count* 1))
    ;; This can be here but needs controls to ensure it doesn't run more than every 4 seconds
    ;; (rmt:dat->json-str
    ;;  (if (or (string? res)
    ;;          (list?   res)
    ;;          (number? res)
    ;;          (boolean? res))
    ;;      res 
    ;;      (list "ERROR, not string, list, number or boolean" 1 cmd params res)))))
    (db:obj->string res transport: 'http)))

(define api:queue-mutex (make-mutex))

;; api queued request handler
;;
;; qry-type: read write transaction
;;
(define (api:queued-request queues qry-type params proc)
  ;; add proc to read, write queue or if transaction do it immediately (for now, not sure but might need to process differently.)
  (if queues
      (begin
	(mutex-lock! (api:queues-mutex queues))
	(let ((dat (vector proc params #f))) ;; #f is placeholder for the result
	  (case qry-type
	    ((read)
	     (api:queues-readq-set!  queues (cons dat (api:queues-readq queues)))
	     (mutex-unlock! (api:queues-mutex queues)(api:queues-read-cvar queues)) ;; unlock mutex and proceed when condition var is triggered
	     (vector-ref dat 2)) ;; return the value from the query to the caller
	    ((write)
	     (api:queues-writeq-set! queues (cons dat (api:queues-writeq queues)))
	     (mutex-unlock! (api:queues-mutex queues)(api:queues-write-cvar queues)) ;; unlock mutex and proceed when condition var is triggered
	     (vector-ref dat 2))
	    (else
	     (mutex-unlock! (api:queues-mutex queues))
	     (proc)))))
      (proc)))

;; process queues
;;
(define (api:process-queues queues)
  (mutex-lock (api:queues-mutex queues))
  (let* ((now        (current-milliseconds))
	 (due        (- now 500)) ;; we will process the queue if it has not been processed in 500 ms
	 (reads      (api:queues-readq      queues))
	 (writes     (api:queues-writeq     queues))
	 (last-read  (api:queues-last-read  queues))
	 (last-write (api:queues-last-write queues)))
    (cond
     ((and (>= last-read last-write) ;; nudge the system to toggle between processing the reads and processing the writes
	   (not (null? reads))
	   (> due last-read))
      (db:with-db                    ;; process the procs inside a transaction
       (api:queues-dbstruct queues)
       #f
       #f
       (lambda (db)
	 (sqlite3:with-transaction   ;; the transaction
	  db
	  (lambda ()
	    (for-each
	     (lambda (procdat)
	       (vector-set! procdat 2 ((vector-ref procdat 0)))) ;; set vector 3rd pos to the result of calculating proc
	     reads)))))
      ;; now reset the queue values
      (api:queues-read-set!      queues '())
      (api:queues-last-read-set! queues now)
      (condition-variable-broadcast! (api:queues-read-cvar queues)))
     ((and (not (null? writes))
	   (> due last-write))
      (db:with-db
       (api:queues-dbstruct queues)
       #f
       #f
       (lambda (db)
	 (sqlite3:with-transaction
	  db
	  (lambda ()
	    (for-each
	     (lambda (procdat)
	       (vector-set! procdat 2 ((vector-ref procdat 0))))
	     writes)))))
      ;; now reset the queue values
      (api:queues-write-set!    queues '())
      (api:queues-last-write-set! queues now)
      (condition-variable-broadcast! (api:queues-write-cvar queues))))
    (mutex-unlock (api:queues-mutex queues))))
      
	 

;; (define *server-id*         #f)
(define *server-info*       #f)  ;; good candidate for easily convert to non-global
(define *time-to-exit*      #f)
(define *server-run*        #t)
(define *run-id*            #f)
(define *server-kind-run*   (make-hash-table))
(define *home-host*         #f)

;; (define *total-non-write-delay* 0)
(define *heartbeat-mutex*   (make-mutex))
(define *api-process-request-count* 0)
(define *max-api-process-requests* 0)
(define *server-overloaded*  #f)

;; client
(define *rmt-mutex*         (make-mutex))     ;; remote access calls mutex

;; RPC transport
(define *rpc:listener*      #f)

;; KEY info
(define *target*            (make-hash-table)) ;; cache the target here; target is keyval1/keyval2/.../keyvalN
(define *keys*              (make-hash-table)) ;; cache the keys here

  (conndat           #f)
  (transport         *transport-type*)
  (server-timeout    (server:get-timeout)) ;; default from server:get-timeout
  (force-server      #f)
  (ro-mode           #f)  
  (ro-mode-checked   #f)) ;; flag that indicates we have checked for ro-mode

;; api queued requests structure and global (temporary solution)
(defstruct api:queues
  (enable      #f)
  (dbstruct    #f)                   ;; must be initialized!
  (mutex       (make-mutex))
  (readq      '())
  (writeq     '())
  (last-read   (current-milliseconds))
  (last-write  (current-milliseconds))
  (read-cvar   (make-condition-variable "reads"))
  (write-cvar  (make-condition-variable "writes"))
  )

(define *queues*            (make-api:queues enable: #t)) ;; set up the queues for coalescing queries

;; launching and hosts
(defstruct host
  (reachable    #f)
  (last-update  0)
  (last-used    0)
  (last-cpuload 1))

    ;; DOT MUTEXLOCK -> INIT_RUNREMOTE [label="no remote?"];
    ;; DOT INIT_RUNREMOTE -> MUTEXLOCK;
    ;; ensure we have a record for our connection for given area
    (if (not runremote)                   ;; can remove this one. should never get here.         
	(begin
	  (set! *runremote* (make-remote))
	  (set! runremote   *runremote*))) ;; new runremote will come from this on next iteration

    ;; DOT SET_HOMEHOST; // leaving off - doesn't really add to the clarity
    ;; DOT MUTEXLOCK -> SET_HOMEHOST [label="no homehost?"];
    ;; DOT SET_HOMEHOST -> MUTEXLOCK;
    ;; ensure we have a homehost record
    (if (not (pair? (remote-hh-dat runremote)))  ;; not on homehost
	(thread-sleep! 0.1) ;; since we shouldn't get here, delay a little
	(remote-hh-dat-set! runremote (common:get-homehost)))

(define (rmt:open-qry-close-locally cmd run-id params #!key (remretries 5))
  (let* ((qry-is-write   (not (member cmd api:read-only-queries)))
	 (db-file-path   (db:dbfile-path)) ;;  0))
	 (dbstruct-local (db:setup #t))  ;; make-dbr:dbstruct path:  dbdir local: #t)))
	 (read-only      (not (file-write-access? db-file-path)))
	 (start          (current-milliseconds))
	 (resdat         (if (not (and read-only qry-is-write))
			     (let ((v (api:execute-requests dbstruct-local (vector (symbol->string cmd) params) *queues*)))
			       (handle-exceptions ;; there has been a long history of receiving strange errors from values returned by the client when things go wrong..
				exn               ;;  This is an attempt to detect that situation and recover gracefully
				(begin
				  (debug:print0 *default-log-port* "ERROR: bad data from server " v " message: "  ((condition-property-accessor 'exn 'message) exn))
				  (vector #t '())) ;; should always get a vector but if something goes wrong return a dummy
				(if (and (vector? v)
					 (> (vector-length v) 1))