38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
srfi-18
posix
matchable
s11n
typed-records)
;; QUEUE METHOD
(define (api:tcp-dispatch-request-make-handler-new dbstruct) ;; cmd run-id params)
(api:tcp-dispatch-request-make-handler-core dbstruct api:dispatch-request))
;; 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-old 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 (indat)
(api:register-thread (current-thread))
(let* ((result
(let* ((numthreads (api:get-count-threads-alive))
(delay-wait (if (> numthreads 10)
(- numthreads 10)
0))
(normal-proc (lambda (cmd run-id params)
(case cmd
((ping) *server-signature*)
(else
(api:dispatch-request dbstruct cmd run-id params))))))
(set! *api-process-request-count* numthreads)
(set! *db-last-access* (current-seconds))
;; (if (not (eq? numthreads numthreads))
;; (begin
;; (api:remove-dead-or-terminated)
;; (let ((threads-now (api:get-count-threads-alive)))
;; (debug:print 0 *default-log-port* "WARNING: numthreads="numthreads", numthreads="numthreads", remaining="threads-now)
;; (set! numthreads threads-now))))
(match indat
((cmd run-id params meta)
(let* ((start-t (current-milliseconds))
(db-ok (let* ((dbfname (dbmod:run-id->dbfname run-id))
(ok (equal? dbfname (dbr:dbstruct-dbfname dbstruct))))
(case cmd
((ping) #t) ;; we are fine
(else
(assert ok "FATAL: database file and run-id not aligned.")))))
(ttdat *server-info*)
(server-state (tt-state ttdat))
(maxthreads 20) ;; make this a parameter?
(status (cond
((and (> numthreads maxthreads)
(> (random 100) 70)) ;; allow a 30% probability to go through so we can figure out what is going wrong in main.db server.
'busy)
;; ((> numthreads 5) 'loaded) ;; this gets transmitted to the client which calls tt:backoff-incr to slow stuff down.
(else 'ok)))
(errmsg (case status
((busy) (conc "Server overloaded, "numthreads" threads in flight"))
((loaded) (conc "Server loaded, "numthreads" threads in flight"))
(else #f)))
(result (case status
((busy)
(if (eq? cmd 'ping)
(normal-proc cmd run-id params)
;; numthreads must be greater than 5 for busy
(* 0.1 (- numthreads maxthreads)) ;; was 15 - return a number for the remote to delay
)) ;; (- numthreads 29)) ;; call back in as many seconds
((loaded)
(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)))
;; (cmd run-id params meta)
(db:add-stats cmd run-id params (- (current-milliseconds) start-t))
payload))
(else
(assert #f "FATAL: failed to deserialize indat "indat))))))
;; (set! *api-process-request-count* (- *api-process-request-count* 1))
;; (serialize payload)
(api:unregister-thread (current-thread))
result)))
(define api:tcp-dispatch-request-make-handler api:tcp-dispatch-request-make-handler-new) ;; choose -old or -new
|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
|
38
39
40
41
42
43
44
|
srfi-18
posix
matchable
s11n
typed-records)
|