Index: api.scm
==================================================================
--- api.scm
+++ api.scm
@@ -114,11 +114,11 @@
;; 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)
+(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))
@@ -147,11 +147,13 @@
(case cmd
;;===============================================
;; READ/WRITE QUERIES
;;===============================================
- ((get-keys-write) (db:get-keys dbstruct)) ;; force a dummy "write" query to force server; for debug in -repl
+ ((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))
@@ -322,11 +324,11 @@
(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))) ;; process the request, resdat = #( flag result )
+ (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*)
@@ -340,5 +342,82 @@
;; (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))))
+
+
Index: common.scm
==================================================================
--- common.scm
+++ common.scm
@@ -126,18 +126,19 @@
(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
+(define *rmt-mutex* (make-mutex)) ;; remote access calls mutex
;; RPC transport
(define *rpc:listener* #f)
;; KEY info
@@ -165,10 +166,25 @@
(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)
Index: rmt.scm
==================================================================
--- rmt.scm
+++ rmt.scm
@@ -85,11 +85,11 @@
;; 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
@@ -335,11 +335,11 @@
(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))))
+ (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