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
|
;; 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)
(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")))
(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)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
<
>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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
|
;; 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)
;; (if (> *api-process-request-count* 50)
;; (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 - no, the napping is moved to the clients via tt:backoff-incr
;; ))
;; (cond
;; ((not (vector? dat)) ;; it is an error to not receive a vector
;; (vector #f (vector #f "remote must be called with a vector")))
;; (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)
|
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
|
(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)) ;; call back in as many seconds
((loaded)
(if (eq? (rmt:transport-mode) 'tcp)
(thread-sleep! 0.5))
(normal-proc cmd run-id params))
(else
(normal-proc cmd run-id params))))
(meta (case cmd
((ping) `((sstate . ,server-state)))
(else `((wait . ,delay-wait)))))
(payload (list status errmsg result meta)))
|
|
|
|
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
|
(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)) ;; call back in as many seconds
((loaded)
;; (if (eq? (rmt:transport-mode) 'tcp)
;; (thread-sleep! 0.5))
(normal-proc cmd run-id params))
(else
(normal-proc cmd run-id params))))
(meta (case cmd
((ping) `((sstate . ,server-state)))
(else `((wait . ,delay-wait)))))
(payload (list status errmsg result meta)))
|