Megatest

Diff
Login

Differences From Artifact [f3cc459c57]:

To Artifact [a4804c006e]:


16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

31
32
33






34
35
36
37
38
39
40
;;     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/>.
;;
;;======================================================================

(use srfi-69 posix)

(declare (unit api))
(declare (uses rmt))
(declare (uses db))
(declare (uses dbmod))
(declare (uses dbfile))
(declare (uses tasks))


(import dbmod)
(import dbfile)







;; allow these queries through without starting a server
;;
(define api:read-only-queries
  '(get-key-val-pairs
    get-var
    get-keys







<
<






>



>
>
>
>
>
>







16
17
18
19
20
21
22


23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
;;     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/>.
;;
;;======================================================================



(declare (unit api))
(declare (uses rmt))
(declare (uses db))
(declare (uses dbmod))
(declare (uses dbfile))
(declare (uses tasks))
(declare (uses tcp-transportmod))

(import dbmod)
(import dbfile)
(import tcp-transportmod)

(use srfi-69
     posix
     matchable
     s11n)

;; allow these queries through without starting a server
;;
(define api:read-only-queries
  '(get-key-val-pairs
    get-var
    get-keys
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227











































228
229

230
231
232
233
234
235
236

    ;; TASKS
    tasks-add
    tasks-set-state-given-param-key
    ))

(define *db-write-mutexes* (make-hash-table))

;; 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)
  (db:open-no-sync-db) ;; sets *no-sync-db*
;;   (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 ", exn=" exn)
;;      (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 (> *api-process-request-count* 200)
      (begin
	(if (common:low-noise-print 30 "too many threads")
	    (debug:print 0 *default-log-port* "WARNING: "*api-process-request-count*" threads, potential overload, adding 0.5 sec delay."))
	(thread-sleep! 0.5) ;; take a nap
	))
   (cond
    ((not (vector? dat))                    ;; it is an error to not receive a vector
     (vector #f (vector #f "remote must be called with a vector")))
    #;((> *api-process-request-count* 200) ;; 20)
     (debug:print 0 *default-log-port* "WARNING: api:execute-requests received an overloaded message.")
     (set! *server-overloaded* #t)
     (vector #f (vector #f 'overloaded))) ;; the inner vector is what gets returned. nope, don't know why. please refactor!
    (else  
     (let* ((cmd-in            (vector-ref dat 0))
            (cmd               (if (symbol? cmd-in)
				   cmd-in
				   (string->symbol cmd-in)))
            (params            (vector-ref dat 1))
	    (run-id            (if (null? params)
				   0
				   (car params)))
	    (write-mutex       (if (hash-table-exists? *db-write-mutexes* run-id)
				   (hash-table-ref *db-write-mutexes* run-id)
				   (let* ((newmutex (make-mutex)))
				     (hash-table-set! *db-write-mutexes* run-id newmutex)
				     newmutex)))
            (start-t           (current-milliseconds))
            (readonly-mode     (dbr:dbstruct-read-only dbstruct))
            (readonly-command  (member cmd api:read-only-queries))
            (writecmd-in-readonly-mode (and readonly-mode (not readonly-command))))
       (if (not readonly-command)
	   (mutex-lock! write-mutex))
       (let* ((tmppath    (dbr:dbstruct-tmppath  dbstruct))
	      (clean-run-id (cond
			     ((number? run-id)   run-id)
			     ((equal? run-id #f) "main")
			     (else               "other")))
	      (crumbfile  (dbfile:wait-for-qif tmppath clean-run-id (cons cmd params)))
	      (res    
               (if writecmd-in-readonly-mode
                   (conc "attempt to run write command "cmd" on a read-only database")
		   (api:dispatch-request dbstruct cmd run-id params))))
	 (delete-file* crumbfile)
	 (if (not readonly-command)
	     (mutex-unlock! write-mutex))
	 
	 ;; save all stats
	 (let ((delta-t (- (current-milliseconds)
			   start-t))
	       (modified-cmd (if (eq? cmd 'general-call)
				 (string->symbol (conc "general-call-" (car params)))
				 cmd)))
	   (hash-table-set! *db-api-call-time* modified-cmd
			    (cons delta-t (hash-table-ref/default *db-api-call-time* modified-cmd '()))))
	 (if writecmd-in-readonly-mode
             (begin
               #;(common:telemetry-log (conc "api-out:"(->string cmd))
               payload: `((params . ,params)
               (ok-res . #t)))
	       (vector #f res))
             (begin
               #;(common:telemetry-log (conc "api-out:"(->string cmd))
               payload: `((params . ,params)
               (ok-res . #f)))
               (vector #t res))))))))












































(define (api:dispatch-request dbstruct cmd run-id params)

  (case cmd
    ;;===============================================
    ;; READ/WRITE QUERIES
    ;;===============================================

    ((get-keys-write)                        (db:get-keys dbstruct)) ;; force a dummy "write" query to force server; for debug in -repl
    







|







|
|
|
|
|
|







|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>


>







145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285

    ;; TASKS
    tasks-add
    tasks-set-state-given-param-key
    ))

(define *db-write-mutexes* (make-hash-table))
(define *server-signature* #f)
;; 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)
  (db:open-no-sync-db) ;; sets *no-sync-db*
  ;;   (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 ", exn=" exn)
  ;;      (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 (> *api-process-request-count* 200)
      (begin
	(if (common:low-noise-print 30 "too many threads")
	    (debug:print 0 *default-log-port* "WARNING: "*api-process-request-count*" threads, potential overload, adding 0.5 sec delay."))
	(thread-sleep! 0.5) ;; take a nap
	))
  (cond
   ((not (vector? dat))                    ;; it is an error to not receive a vector
    (vector #f (vector #f "remote must be called with a vector")))
   #;((> *api-process-request-count* 200) ;; 20)
   (debug:print 0 *default-log-port* "WARNING: api:execute-requests received an overloaded message.")
   (set! *server-overloaded* #t)
   (vector #f (vector #f 'overloaded))) ;; the inner vector is what gets returned. nope, don't know why. please refactor!
   (else  
    (let* ((cmd-in            (vector-ref dat 0))
           (cmd               (if (symbol? cmd-in)
				  cmd-in
				  (string->symbol cmd-in)))
           (params            (vector-ref dat 1))
	   (run-id            (if (null? params)
				  0
				  (car params)))
	   (write-mutex       (if (hash-table-exists? *db-write-mutexes* run-id)
				  (hash-table-ref *db-write-mutexes* run-id)
				  (let* ((newmutex (make-mutex)))
				    (hash-table-set! *db-write-mutexes* run-id newmutex)
				    newmutex)))
           (start-t           (current-milliseconds))
           (readonly-mode     (dbr:dbstruct-read-only dbstruct))
           (readonly-command  (member cmd api:read-only-queries))
           (writecmd-in-readonly-mode (and readonly-mode (not readonly-command))))
      (if (not readonly-command)
	  (mutex-lock! write-mutex))
      (let* ((tmppath    (dbr:dbstruct-tmppath  dbstruct))
	     (clean-run-id (cond
			    ((number? run-id)   run-id)
			    ((equal? run-id #f) "main")
			    (else               "other")))
	     (crumbfile  (dbfile:wait-for-qif tmppath clean-run-id (cons cmd params)))
	     (res    
              (if writecmd-in-readonly-mode
                  (conc "attempt to run write command "cmd" on a read-only database")
		  (api:dispatch-request dbstruct cmd run-id params))))
	(delete-file* crumbfile)
	(if (not readonly-command)
	    (mutex-unlock! write-mutex))
	
	;; save all stats
	(let ((delta-t (- (current-milliseconds)
			  start-t))
	      (modified-cmd (if (eq? cmd 'general-call)
				(string->symbol (conc "general-call-" (car params)))
				cmd)))
	  (hash-table-set! *db-api-call-time* modified-cmd
			   (cons delta-t (hash-table-ref/default *db-api-call-time* modified-cmd '()))))
	(if writecmd-in-readonly-mode
            (begin
              #;(common:telemetry-log (conc "api-out:"(->string cmd))
              payload: `((params . ,params)
              (ok-res . #t)))
	      (vector #f res))
            (begin
              #;(common:telemetry-log (conc "api-out:"(->string cmd))
              payload: `((params . ,params)
              (ok-res . #f)))
              (vector #t res))))))))

;; indat is (cmd run-id params meta)
;;
;; WARNING: Do not print anything in the lambda of this function as it
;;          reads/writes to current in/out port
;;
(define (api:tcp-dispatch-request-make-handler dbstruct) ;; cmd run-id params)
  (assert *toppath* "FATAL: api:tcp-dispatch-request-make-handler called but *toppath* not set.")
  (if (not *server-signature*)
      (set! *server-signature* (tt:mk-signature *toppath*)))
  (lambda ()
    (let* ((indat      (deserialize))
	   (newcount   (+ *api-process-request-count* 1))
	   (delay-wait (if (> newcount 10)
			   (- newcount 10)
			   0)))
      (set! *api-process-request-count* newcount)
      (set! *db-last-access* (current-seconds))
      (match indat
	((cmd run-id params meta)
	 (let* ((status  (cond
			  ;; ((> newcount 30) 'busy)
			  ;; ((> newcount 15) 'loaded)
			  (else 'ok)))
		(errmsg  (case status
			   ((busy)   (conc "Server overloaded, "newcount" threads in flight"))
			   ((loaded) (conc "Server loaded, "newcount" threads in flight"))
			   (else     #f)))
		(result  (case status
			   ((busy)  (- newcount 29))
			   ((loaded) #f)
			   (else
			    (case cmd
			      ((ping) *server-signature*)
			      (else
			       (api:dispatch-request dbstruct cmd run-id params))))))
		(meta   `((wait . ,delay-wait)))
		(payload (list status errmsg result meta)))
	   (set! *api-process-request-count* (- *api-process-request-count* 1))
	   (serialize payload)))
	(else
	 (assert #f "FATAL: failed to deserialize indat "indat))))))
       

(define (api:dispatch-request dbstruct cmd run-id params)
  (db:open-no-sync-db)
  (case cmd
    ;;===============================================
    ;; READ/WRITE QUERIES
    ;;===============================================

    ((get-keys-write)                        (db:get-keys dbstruct)) ;; force a dummy "write" query to force server; for debug in -repl