Megatest

Check-in [0a1b205bcf]
Login
Overview
Comment:queued coalesced queries (trying again)
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | v1.64-coalesced-queries
Files: files | file ages | folders
SHA1: 0a1b205bcffe79a1eaa1cf9261c18f8dacaac14e
User & Date: matt on 2017-06-29 23:17:56
Other Links: branch diff | manifest | tags
Context
2017-07-03
21:46
small fixes check-in: 948b22f1a9 user: matt tags: v1.64-coalesced-queries
2017-06-29
23:17
queued coalesced queries (trying again) check-in: 0a1b205bcf user: matt tags: v1.64-coalesced-queries
2017-06-28
14:01
Added missing schema patch for test_rundat check-in: 85fa0e2f14 user: mrwellan tags: v1.64
Changes

Modified api.scm from [1f6842e15f] to [8da0d403fb].

112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
    ))

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







|







112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
    ))

;; 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
145
146
147
148
149
150
151


152
153
154
155
156
157
158
159
             (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)                        (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








>
>
|







145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
             (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

320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
























































































;; 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))) ;; process the request, resdat = #( flag result )
	 (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)))
































































































|

















>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
;; 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))

(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"))
  )

;; 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:queue-mutex queues))
	(let ((dat (vector proc params #f))) ;; #f is placeholder for the result
	  (case qry-type
	    ((read)
	     (api:queue-readq-set!  queues (cons dat (api:queue-readq queues)))
	     (mutex-unlock! (api:queue-mutex queues)(api:queue-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:queue-writeq-set! queues (cons dat (api:queue-writeq queues)))
	     (mutex-unlock! (api:queue-mutex queues)(api:queue-write-cvar queues)) ;; unlock mutex and proceed when condition var is triggered
	     (vector-ref dat 2))
	    (else
	     (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))))
      
	 

Modified common.scm from [fe27a287b6] to [bfed3753ba].

124
125
126
127
128
129
130

131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
;; (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







>







|







124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
;; (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 *queues*            (make-api:queues enable: #t)) ;; set up the queues for coalescing queries
;; (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

Modified rmt.scm from [54fa717fe0] to [36c2c78229].

83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
    ;; 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)))







|







83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
    ;; 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)))
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
(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))))
			       (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))







|







333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
(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))