| ︙ | | | ︙ | |
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
|
;; Call this to start the actual server
;;
(define *db:process-queue-mutex* (make-mutex))
(define (http-transport:run hostn run-id server-id)
(debug:print 2 "Attempting to start the server ...")
(let* ((db #f) ;; (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily
(hostname (get-host-name))
(ipaddrstr (let ((ipstr (if (string=? "-" hostn)
;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
(server:get-best-guess-address hostname)
#f)))
(if ipstr ipstr hostn))) ;; hostname)))
(start-port (portlogger:open-run-close portlogger:find-port))
(link-tree-path (configf:lookup *configdat* "setup" "linktree")))
;; (set! db *inmemdb*)
(debug:print-info 0 "portlogger recommended port: " start-port)
(root-path (if link-tree-path
link-tree-path
(current-directory))) ;; WARNING: SECURITY HOLE. FIX ASAP!
(handle-directory spiffy-directory-listing)
(handle-exception (lambda (exn chain)
(signal (make-composite-condition
(make-property-condition
|
|
|
|
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
|
;; Call this to start the actual server
;;
(define *db:process-queue-mutex* (make-mutex))
(define (http-transport:run hostn run-id server-id)
(debug:print 2 #f "Attempting to start the server ...")
(let* ((db #f) ;; (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily
(hostname (get-host-name))
(ipaddrstr (let ((ipstr (if (string=? "-" hostn)
;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
(server:get-best-guess-address hostname)
#f)))
(if ipstr ipstr hostn))) ;; hostname)))
(start-port (portlogger:open-run-close portlogger:find-port))
(link-tree-path (configf:lookup *configdat* "setup" "linktree")))
;; (set! db *inmemdb*)
(debug:print-info 0 #f "portlogger recommended port: " start-port)
(root-path (if link-tree-path
link-tree-path
(current-directory))) ;; WARNING: SECURITY HOLE. FIX ASAP!
(handle-directory spiffy-directory-listing)
(handle-exception (lambda (exn chain)
(signal (make-composite-condition
(make-property-condition
|
| ︙ | | | ︙ | |
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
(http-transport:try-start-server run-id ipaddrstr start-port server-id)))
;; This is recursively run by http-transport:run until sucessful
;;
(define (http-transport:try-start-server run-id ipaddrstr portnum server-id)
(let ((config-hostname (configf:lookup *configdat* "server" "hostname"))
(tdbdat (tasks:open-db)))
(debug:print-info 0 "http-transport:try-start-server run-id=" run-id " ipaddrsstr=" ipaddrstr " portnum=" portnum " server-id=" server-id " config-hostname=" config-hostname)
(handle-exceptions
exn
(begin
(print-error-message exn)
(if (< portnum 64000)
(begin
(debug:print 0 "WARNING: attempt to start server failed. Trying again ...")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(debug:print 0 "exn=" (condition->list exn))
(portlogger:open-run-close portlogger:set-failed portnum)
(debug:print 0 "WARNING: failed to start on portnum: " portnum ", trying next port")
(thread-sleep! 0.1)
;; get_next_port goes here
(http-transport:try-start-server run-id
ipaddrstr
(portlogger:open-run-close portlogger:find-port)
server-id))
(begin
(tasks:server-force-clean-run-record (db:delay-if-busy tdbdat) run-id ipaddrstr portnum " http-transport:try-start-server")
(print "ERROR: Tried and tried but could not start the server"))))
;; any error in following steps will result in a retry
(set! *server-info* (list ipaddrstr portnum))
(tasks:server-set-interface-port
(db:delay-if-busy tdbdat)
server-id
ipaddrstr portnum)
(debug:print 0 "INFO: Trying to start server on " ipaddrstr ":" portnum)
;; This starts the spiffy server
;; NEED WAY TO SET IP TO #f TO BIND ALL
;; (start-server bind-address: ipaddrstr port: portnum)
(if config-hostname ;; this is a hint to bind directly
(start-server port: portnum bind-address: (if (equal? config-hostname "-")
ipaddrstr
config-hostname))
(start-server port: portnum))
;; (portlogger:open-run-close portlogger:set-port portnum "released")
(tasks:server-force-clean-run-record (db:delay-if-busy tdbdat) run-id ipaddrstr portnum " http-transport:try-start-server")
(debug:print 1 "INFO: server has been stopped"))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;;======================================================================
;; C L I E N T S
|
|
|
|
|
|
|
|
|
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
(http-transport:try-start-server run-id ipaddrstr start-port server-id)))
;; This is recursively run by http-transport:run until sucessful
;;
(define (http-transport:try-start-server run-id ipaddrstr portnum server-id)
(let ((config-hostname (configf:lookup *configdat* "server" "hostname"))
(tdbdat (tasks:open-db)))
(debug:print-info 0 #f "http-transport:try-start-server run-id=" run-id " ipaddrsstr=" ipaddrstr " portnum=" portnum " server-id=" server-id " config-hostname=" config-hostname)
(handle-exceptions
exn
(begin
(print-error-message exn)
(if (< portnum 64000)
(begin
(debug:print 0 #f "WARNING: attempt to start server failed. Trying again ...")
(debug:print 0 #f " message: " ((condition-property-accessor 'exn 'message) exn))
(debug:print 0 #f "exn=" (condition->list exn))
(portlogger:open-run-close portlogger:set-failed portnum)
(debug:print 0 #f "WARNING: failed to start on portnum: " portnum ", trying next port")
(thread-sleep! 0.1)
;; get_next_port goes here
(http-transport:try-start-server run-id
ipaddrstr
(portlogger:open-run-close portlogger:find-port)
server-id))
(begin
(tasks:server-force-clean-run-record (db:delay-if-busy tdbdat) run-id ipaddrstr portnum " http-transport:try-start-server")
(print "ERROR: Tried and tried but could not start the server"))))
;; any error in following steps will result in a retry
(set! *server-info* (list ipaddrstr portnum))
(tasks:server-set-interface-port
(db:delay-if-busy tdbdat)
server-id
ipaddrstr portnum)
(debug:print 0 #f "INFO: Trying to start server on " ipaddrstr ":" portnum)
;; This starts the spiffy server
;; NEED WAY TO SET IP TO #f TO BIND ALL
;; (start-server bind-address: ipaddrstr port: portnum)
(if config-hostname ;; this is a hint to bind directly
(start-server port: portnum bind-address: (if (equal? config-hostname "-")
ipaddrstr
config-hostname))
(start-server port: portnum))
;; (portlogger:open-run-close portlogger:set-port portnum "released")
(tasks:server-force-clean-run-record (db:delay-if-busy tdbdat) run-id ipaddrstr portnum " http-transport:try-start-server")
(debug:print 1 #f "INFO: server has been stopped"))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;;======================================================================
;; C L I E N T S
|
| ︙ | | | ︙ | |
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
|
(define (http-transport:inc-requests-count)
(mutex-lock! *http-mutex*)
(set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*))
;; Use this opportunity to slow things down iff there are too many requests in flight
(if (> *http-requests-in-progress* 5)
(begin
(debug:print-info 0 "Whoa there buddy, ease up...")
(thread-sleep! 1)))
(mutex-unlock! *http-mutex*))
(define (http-transport:dec-requests-count proc)
(mutex-lock! *http-mutex*)
(proc)
(set! *http-requests-in-progress* (- *http-requests-in-progress* 1))
(mutex-unlock! *http-mutex*))
(define (http-transport:dec-requests-count-and-close-all-connections)
(set! *http-requests-in-progress* (- *http-requests-in-progress* 1))
(let loop ((etime (+ (current-seconds) 5))) ;; give up in five seconds
(if (> *http-requests-in-progress* 0)
(if (> etime (current-seconds))
(begin
(thread-sleep! 0.05)
(loop etime))
(debug:print 0 "ERROR: requests still in progress after 5 seconds of waiting. I'm going to pass on cleaning up http connections"))
(close-all-connections!)))
(set! *http-connections-next-cleanup* (+ (current-seconds) 10))
(mutex-unlock! *http-mutex*))
(define (http-transport:inc-requests-and-prep-to-close-all-connections)
(mutex-lock! *http-mutex*)
(set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*)))
;; Send "cmd" with json payload "params" to serverdat and receive result
;;
(define (http-transport:client-api-send-receive run-id serverdat cmd params #!key (numretries 3))
(let* ((fullurl (if (vector? serverdat)
(http-transport:server-dat-get-api-req serverdat)
(begin
(debug:print 0 "FATAL ERROR: http-transport:client-api-send-receive called with no server info")
(exit 1))))
(res #f)
(success #t)
(sparams (db:obj->string params transport: 'http)))
;; (condition-case
;; handle-exceptions
;; exn
;; (if (> numretries 0)
;; (begin
;; (mutex-unlock! *http-mutex*)
;; (thread-sleep! 1)
;; (handle-exceptions
;; exn
;; (debug:print 0 "WARNING: closing connections failed. Server at " fullurl " almost certainly dead")
;; (close-all-connections!))
;; (debug:print 0 "WARNING: Failed to communicate with server, trying again, numretries left: " numretries)
;; (http-transport:client-api-send-receive run-id serverdat cmd sparams numretries: (- numretries 1)))
;; (begin
;; (mutex-unlock! *http-mutex*)
;; (tasks:kill-server-run-id run-id)
;; #f))
;; (begin
(debug:print-info 11 "fullurl=" fullurl ", cmd=" cmd ", params=" params ", run-id=" run-id "\n")
;; set up the http-client here
(max-retry-attempts 1)
;; consider all requests indempotent
(retry-request? (lambda (request)
#f))
;; send the data and get the response
;; extract the needed info from the http data and
;; process and return it.
(let* ((send-recieve (lambda ()
(mutex-lock! *http-mutex*)
;; (condition-case (with-input-from-request "http://localhost"; #f read-lines)
;; ((exn http client-error) e (print e)))
(set! res (vector
success
(db:string->obj
(handle-exceptions
exn
(begin
(set! success #f)
(debug:print 0 "WARNING: failure in with-input-from-request to " fullurl ".")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(hash-table-delete! *runremote* run-id)
;; Killing associated server to allow clean retry.")
;; (tasks:kill-server-run-id run-id) ;; better to kill the server in the logic that called this routine?
(mutex-unlock! *http-mutex*)
;;; (signal (make-composite-condition
;;; (make-property-condition 'commfail 'message "failed to connect to server")))
;;; "communications failed"
|
|
|
|
|
|
|
|
|
|
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
|
(define (http-transport:inc-requests-count)
(mutex-lock! *http-mutex*)
(set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*))
;; Use this opportunity to slow things down iff there are too many requests in flight
(if (> *http-requests-in-progress* 5)
(begin
(debug:print-info 0 #f "Whoa there buddy, ease up...")
(thread-sleep! 1)))
(mutex-unlock! *http-mutex*))
(define (http-transport:dec-requests-count proc)
(mutex-lock! *http-mutex*)
(proc)
(set! *http-requests-in-progress* (- *http-requests-in-progress* 1))
(mutex-unlock! *http-mutex*))
(define (http-transport:dec-requests-count-and-close-all-connections)
(set! *http-requests-in-progress* (- *http-requests-in-progress* 1))
(let loop ((etime (+ (current-seconds) 5))) ;; give up in five seconds
(if (> *http-requests-in-progress* 0)
(if (> etime (current-seconds))
(begin
(thread-sleep! 0.05)
(loop etime))
(debug:print 0 #f "ERROR: requests still in progress after 5 seconds of waiting. I'm going to pass on cleaning up http connections"))
(close-all-connections!)))
(set! *http-connections-next-cleanup* (+ (current-seconds) 10))
(mutex-unlock! *http-mutex*))
(define (http-transport:inc-requests-and-prep-to-close-all-connections)
(mutex-lock! *http-mutex*)
(set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*)))
;; Send "cmd" with json payload "params" to serverdat and receive result
;;
(define (http-transport:client-api-send-receive run-id serverdat cmd params #!key (numretries 3))
(let* ((fullurl (if (vector? serverdat)
(http-transport:server-dat-get-api-req serverdat)
(begin
(debug:print 0 #f "FATAL ERROR: http-transport:client-api-send-receive called with no server info")
(exit 1))))
(res #f)
(success #t)
(sparams (db:obj->string params transport: 'http)))
;; (condition-case
;; handle-exceptions
;; exn
;; (if (> numretries 0)
;; (begin
;; (mutex-unlock! *http-mutex*)
;; (thread-sleep! 1)
;; (handle-exceptions
;; exn
;; (debug:print 0 #f "WARNING: closing connections failed. Server at " fullurl " almost certainly dead")
;; (close-all-connections!))
;; (debug:print 0 #f "WARNING: Failed to communicate with server, trying again, numretries left: " numretries)
;; (http-transport:client-api-send-receive run-id serverdat cmd sparams numretries: (- numretries 1)))
;; (begin
;; (mutex-unlock! *http-mutex*)
;; (tasks:kill-server-run-id run-id)
;; #f))
;; (begin
(debug:print-info 11 #f "fullurl=" fullurl ", cmd=" cmd ", params=" params ", run-id=" run-id "\n")
;; set up the http-client here
(max-retry-attempts 1)
;; consider all requests indempotent
(retry-request? (lambda (request)
#f))
;; send the data and get the response
;; extract the needed info from the http data and
;; process and return it.
(let* ((send-recieve (lambda ()
(mutex-lock! *http-mutex*)
;; (condition-case (with-input-from-request "http://localhost"; #f read-lines)
;; ((exn http client-error) e (print e)))
(set! res (vector
success
(db:string->obj
(handle-exceptions
exn
(begin
(set! success #f)
(debug:print 0 #f "WARNING: failure in with-input-from-request to " fullurl ".")
(debug:print 0 #f " message: " ((condition-property-accessor 'exn 'message) exn))
(hash-table-delete! *runremote* run-id)
;; Killing associated server to allow clean retry.")
;; (tasks:kill-server-run-id run-id) ;; better to kill the server in the logic that called this routine?
(mutex-unlock! *http-mutex*)
;;; (signal (make-composite-condition
;;; (make-property-condition 'commfail 'message "failed to connect to server")))
;;; "communications failed"
|
| ︙ | | | ︙ | |
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
|
#f))
(th1 (make-thread send-recieve "with-input-from-request"))
(th2 (make-thread time-out "time out")))
(thread-start! th1)
(thread-start! th2)
(thread-join! th1)
(thread-terminate! th2)
(debug:print-info 11 "got res=" res)
(if (vector? res)
(if (vector-ref res 0)
res
(begin ;; note: this code also called in nmsg-transport - consider consolidating it
(debug:print 0 "ERROR: error occured at server, info=" (vector-ref res 2))
(debug:print 0 " client call chain:")
(print-call-chain (current-error-port))
(debug:print 0 " server call chain:")
(pp (vector-ref res 1) (current-error-port))
(signal (vector-ref result 0))))
(signal (make-composite-condition
(make-property-condition
'timeout
'message "nmsg-transport:client-api-send-receive-raw timed out talking to server")))))))
|
|
|
|
|
|
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
|
#f))
(th1 (make-thread send-recieve "with-input-from-request"))
(th2 (make-thread time-out "time out")))
(thread-start! th1)
(thread-start! th2)
(thread-join! th1)
(thread-terminate! th2)
(debug:print-info 11 #f "got res=" res)
(if (vector? res)
(if (vector-ref res 0)
res
(begin ;; note: this code also called in nmsg-transport - consider consolidating it
(debug:print 0 #f "ERROR: error occured at server, info=" (vector-ref res 2))
(debug:print 0 #f " client call chain:")
(print-call-chain (current-error-port))
(debug:print 0 #f " server call chain:")
(pp (vector-ref res 1) (current-error-port))
(signal (vector-ref result 0))))
(signal (make-composite-condition
(make-property-condition
'timeout
'message "nmsg-transport:client-api-send-receive-raw timed out talking to server")))))))
|
| ︙ | | | ︙ | |
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
|
#f))
(define (http-transport:server-dat-update-last-access vec)
(if (vector? vec)
(vector-set! vec 5 (current-seconds))
(begin
(print-call-chain (current-error-port))
(debug:print 0 "ERROR: call to http-transport:server-dat-update-last-access with non-vector!!"))))
;;
;; connect
;;
(define (http-transport:client-connect iface port)
(let* ((api-url (conc "http://" iface ":" port "/api"))
(api-uri (uri-reference (conc "http://" iface ":" port "/api")))
(api-req (make-request method: 'POST uri: api-uri))
(server-dat (vector iface port api-uri api-url api-req (current-seconds))))
server-dat))
;; run http-transport:keep-running in a parallel thread to monitor that the db is being
;; used and to shutdown after sometime if it is not.
;;
(define (http-transport:keep-running server-id run-id)
;; if none running or if > 20 seconds since
;; server last used then start shutdown
;; This thread waits for the server to come alive
(debug:print-info 0 "Starting the sync-back, keep alive thread in server for run-id=" run-id)
(let* ((tdbdat (tasks:open-db))
(server-start-time (current-seconds))
(server-info (let loop ((start-time (current-seconds))
(changed #t)
(last-sdat "not this"))
(let ((sdat #f))
(thread-sleep! 0.01)
(debug:print-info 0 "Waiting for server alive signature")
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if (and sdat
(not changed)
(> (- (current-seconds) start-time) 2))
sdat
(begin
(debug:print-info 0 "Still waiting, last-sdat=" last-sdat)
(sleep 4)
(if (> (- (current-seconds) start-time) 120) ;; been waiting for two minutes
(begin
(debug:print 0 "ERROR: transport appears to have died, exiting server " server-id " for run " run-id)
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id "failed to start, never received server alive signature")
(exit))
(loop start-time
(equal? sdat last-sdat)
sdat)))))))
(iface (car server-info))
(port (cadr server-info))
|
|
|
|
|
|
|
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
|
#f))
(define (http-transport:server-dat-update-last-access vec)
(if (vector? vec)
(vector-set! vec 5 (current-seconds))
(begin
(print-call-chain (current-error-port))
(debug:print 0 #f "ERROR: call to http-transport:server-dat-update-last-access with non-vector!!"))))
;;
;; connect
;;
(define (http-transport:client-connect iface port)
(let* ((api-url (conc "http://" iface ":" port "/api"))
(api-uri (uri-reference (conc "http://" iface ":" port "/api")))
(api-req (make-request method: 'POST uri: api-uri))
(server-dat (vector iface port api-uri api-url api-req (current-seconds))))
server-dat))
;; run http-transport:keep-running in a parallel thread to monitor that the db is being
;; used and to shutdown after sometime if it is not.
;;
(define (http-transport:keep-running server-id run-id)
;; if none running or if > 20 seconds since
;; server last used then start shutdown
;; This thread waits for the server to come alive
(debug:print-info 0 #f "Starting the sync-back, keep alive thread in server for run-id=" run-id)
(let* ((tdbdat (tasks:open-db))
(server-start-time (current-seconds))
(server-info (let loop ((start-time (current-seconds))
(changed #t)
(last-sdat "not this"))
(let ((sdat #f))
(thread-sleep! 0.01)
(debug:print-info 0 #f "Waiting for server alive signature")
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if (and sdat
(not changed)
(> (- (current-seconds) start-time) 2))
sdat
(begin
(debug:print-info 0 #f "Still waiting, last-sdat=" last-sdat)
(sleep 4)
(if (> (- (current-seconds) start-time) 120) ;; been waiting for two minutes
(begin
(debug:print 0 #f "ERROR: transport appears to have died, exiting server " server-id " for run " run-id)
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id "failed to start, never received server alive signature")
(exit))
(loop start-time
(equal? sdat last-sdat)
sdat)))))))
(iface (car server-info))
(port (cadr server-info))
|
| ︙ | | | ︙ | |
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
|
((sync-failed)(cond
((> bad-sync-count 10) ;; time to give up
(http-transport:server-shutdown server-id port))
(else ;; (> bad-sync-count 0) ;; we've had a fail or two, delay and loop
(thread-sleep! 5)
(loop count server-state (+ bad-sync-count 1)))))
((exn)
(debug:print 0 "ERROR: error from sync code other than 'sync-failed. Attempting to gracefully shutdown the server")
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id " http-transport:keep-running crashed")
(exit)))
(set! sync-time (- (current-milliseconds) start-time))
(set! rem-time (quotient (- 4000 sync-time) 1000))
(debug:print 4 "SYNC: time= " sync-time ", rem-time=" rem-time)
(if (and (<= rem-time 4)
(> rem-time 0))
(thread-sleep! rem-time)
(thread-sleep! 4))) ;; fallback for if the math is changed ...
;;
|
|
|
|
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
|
((sync-failed)(cond
((> bad-sync-count 10) ;; time to give up
(http-transport:server-shutdown server-id port))
(else ;; (> bad-sync-count 0) ;; we've had a fail or two, delay and loop
(thread-sleep! 5)
(loop count server-state (+ bad-sync-count 1)))))
((exn)
(debug:print 0 #f "ERROR: error from sync code other than 'sync-failed. Attempting to gracefully shutdown the server")
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id " http-transport:keep-running crashed")
(exit)))
(set! sync-time (- (current-milliseconds) start-time))
(set! rem-time (quotient (- 4000 sync-time) 1000))
(debug:print 4 #f "SYNC: time= " sync-time ", rem-time=" rem-time)
(if (and (<= rem-time 4)
(> rem-time 0))
(thread-sleep! rem-time)
(thread-sleep! 4))) ;; fallback for if the math is changed ...
;;
|
| ︙ | | | ︙ | |
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
|
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if (or (not (equal? sdat (list iface port)))
(not server-id))
(begin
(debug:print-info 0 "interface changed, refreshing iface and port info")
(set! iface (car sdat))
(set! port (cadr sdat))))
;; Transfer *last-db-access* to last-access to use in checking that we are still alive
(mutex-lock! *heartbeat-mutex*)
(set! last-access *last-db-access*)
(mutex-unlock! *heartbeat-mutex*)
;; (debug:print 11 "last-access=" last-access ", server-timeout=" server-timeout)
;;
;; no_traffic, no running tests, if server 0, no running servers
;;
;; (let ((wait-on-running (configf:lookup *configdat* "server" b"wait-on-running"))) ;; wait on running tasks (if not true then exit on time out)
;;
(let* ((hrs-since-start (/ (- (current-seconds) server-start-time) 3600))
(adjusted-timeout (if (> hrs-since-start 1)
(- server-timeout (inexact->exact (round (* hrs-since-start 60)))) ;; subtract 60 seconds per hour
server-timeout)))
(if (common:low-noise-print 120 "server timeout")
(debug:print-info 0 "Adjusted server timeout: " adjusted-timeout))
(if (and *server-run*
(> (+ last-access server-timeout)
(current-seconds)))
(begin
(if (common:low-noise-print 120 "server continuing")
(debug:print-info 0 "Server continuing, seconds since last db access: " (- (current-seconds) last-access)))
;;
;; Consider implementing some smarts here to re-insert the record or kill self is
;; the db indicates so
;;
;; (if (tasks:server-am-i-the-server? tdb run-id)
;; (tasks:server-set-state! tdb server-id "running"))
;;
(loop 0 server-state bad-sync-count))
(http-transport:server-shutdown server-id port))))))
(define (http-transport:server-shutdown server-id port)
(let ((tdbdat (tasks:open-db)))
(debug:print-info 0 "Starting to shutdown the server.")
;; need to delete only *my* server entry (future use)
(set! *time-to-exit* #t)
(if *inmemdb* (db:sync-touched *inmemdb* *run-id* force-sync: #t))
;;
;; start_shutdown
;;
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "shutting-down")
(portlogger:open-run-close portlogger:set-port port "released")
(thread-sleep! 5)
(debug:print-info 0 "Max cached queries was " *max-cache-size*)
(debug:print-info 0 "Number of cached writes " *number-of-writes*)
(debug:print-info 0 "Average cached write time "
(if (eq? *number-of-writes* 0)
"n/a (no writes)"
(/ *writes-total-delay*
*number-of-writes*))
" ms")
(debug:print-info 0 "Number non-cached queries " *number-non-write-queries*)
(debug:print-info 0 "Average non-cached time "
(if (eq? *number-non-write-queries* 0)
"n/a (no queries)"
(/ *total-non-write-delay*
*number-non-write-queries*))
" ms")
(debug:print-info 0 "Server shutdown complete. Exiting")
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id " http-transport:keep-running complete")
(exit)))
;; all routes though here end in exit ...
;;
;; start_server?
;;
(define (http-transport:launch run-id)
(let* ((tdbdat (tasks:open-db)))
(set! *run-id* run-id)
(if (args:get-arg "-daemonize")
(begin
(daemon:ize)
(if *alt-log-file* ;; we should re-connect to this port, I think daemon:ize disrupts it
(begin
(current-error-port *alt-log-file*)
(current-output-port *alt-log-file*)))))
(if (server:check-if-running run-id)
(begin
(debug:print 0 "INFO: Server for run-id " run-id " already running")
(exit 0)))
(let loop ((server-id (tasks:server-lock-slot (db:delay-if-busy tdbdat) run-id))
(remtries 4))
(if (not server-id)
(if (> remtries 0)
(begin
(thread-sleep! 2)
(loop (tasks:server-lock-slot (db:delay-if-busy tdbdat) run-id)
(- remtries 1)))
(begin
;; since we didn't get the server lock we are going to clean up and bail out
(debug:print-info 2 "INFO: server pid=" (current-process-id) ", hostname=" (get-host-name) " not starting due to other candidates ahead in start queue")
(tasks:server-delete-records-for-this-pid (db:delay-if-busy tdbdat) " http-transport:launch")
))
(let* ((th2 (make-thread (lambda ()
(debug:print-info 0 "Server run thread started")
(http-transport:run
(if (args:get-arg "-server")
(args:get-arg "-server")
"-")
run-id
server-id)) "Server run"))
(th3 (make-thread (lambda ()
(debug:print-info 0 "Server monitor thread started")
(http-transport:keep-running server-id run-id))
"Keep running")))
(thread-start! th2)
(thread-sleep! 0.25) ;; give the server time to settle before starting the keep-running monitor.
(thread-start! th3)
(set! *didsomething* #t)
(thread-join! th2)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
|
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if (or (not (equal? sdat (list iface port)))
(not server-id))
(begin
(debug:print-info 0 #f "interface changed, refreshing iface and port info")
(set! iface (car sdat))
(set! port (cadr sdat))))
;; Transfer *last-db-access* to last-access to use in checking that we are still alive
(mutex-lock! *heartbeat-mutex*)
(set! last-access *last-db-access*)
(mutex-unlock! *heartbeat-mutex*)
;; (debug:print 11 #f "last-access=" last-access ", server-timeout=" server-timeout)
;;
;; no_traffic, no running tests, if server 0, no running servers
;;
;; (let ((wait-on-running (configf:lookup *configdat* "server" b"wait-on-running"))) ;; wait on running tasks (if not true then exit on time out)
;;
(let* ((hrs-since-start (/ (- (current-seconds) server-start-time) 3600))
(adjusted-timeout (if (> hrs-since-start 1)
(- server-timeout (inexact->exact (round (* hrs-since-start 60)))) ;; subtract 60 seconds per hour
server-timeout)))
(if (common:low-noise-print 120 "server timeout")
(debug:print-info 0 #f "Adjusted server timeout: " adjusted-timeout))
(if (and *server-run*
(> (+ last-access server-timeout)
(current-seconds)))
(begin
(if (common:low-noise-print 120 "server continuing")
(debug:print-info 0 #f "Server continuing, seconds since last db access: " (- (current-seconds) last-access)))
;;
;; Consider implementing some smarts here to re-insert the record or kill self is
;; the db indicates so
;;
;; (if (tasks:server-am-i-the-server? tdb run-id)
;; (tasks:server-set-state! tdb server-id "running"))
;;
(loop 0 server-state bad-sync-count))
(http-transport:server-shutdown server-id port))))))
(define (http-transport:server-shutdown server-id port)
(let ((tdbdat (tasks:open-db)))
(debug:print-info 0 #f "Starting to shutdown the server.")
;; need to delete only *my* server entry (future use)
(set! *time-to-exit* #t)
(if *inmemdb* (db:sync-touched *inmemdb* *run-id* force-sync: #t))
;;
;; start_shutdown
;;
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "shutting-down")
(portlogger:open-run-close portlogger:set-port port "released")
(thread-sleep! 5)
(debug:print-info 0 #f "Max cached queries was " *max-cache-size*)
(debug:print-info 0 #f "Number of cached writes " *number-of-writes*)
(debug:print-info 0 #f "Average cached write time "
(if (eq? *number-of-writes* 0)
"n/a (no writes)"
(/ *writes-total-delay*
*number-of-writes*))
" ms")
(debug:print-info 0 #f "Number non-cached queries " *number-non-write-queries*)
(debug:print-info 0 #f "Average non-cached time "
(if (eq? *number-non-write-queries* 0)
"n/a (no queries)"
(/ *total-non-write-delay*
*number-non-write-queries*))
" ms")
(debug:print-info 0 #f "Server shutdown complete. Exiting")
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id " http-transport:keep-running complete")
(exit)))
;; all routes though here end in exit ...
;;
;; start_server?
;;
(define (http-transport:launch run-id)
(let* ((tdbdat (tasks:open-db)))
(set! *run-id* run-id)
(if (args:get-arg "-daemonize")
(begin
(daemon:ize)
(if *alt-log-file* ;; we should re-connect to this port, I think daemon:ize disrupts it
(begin
(current-error-port *alt-log-file*)
(current-output-port *alt-log-file*)))))
(if (server:check-if-running run-id)
(begin
(debug:print 0 #f "INFO: Server for run-id " run-id " already running")
(exit 0)))
(let loop ((server-id (tasks:server-lock-slot (db:delay-if-busy tdbdat) run-id))
(remtries 4))
(if (not server-id)
(if (> remtries 0)
(begin
(thread-sleep! 2)
(loop (tasks:server-lock-slot (db:delay-if-busy tdbdat) run-id)
(- remtries 1)))
(begin
;; since we didn't get the server lock we are going to clean up and bail out
(debug:print-info 2 #f "INFO: server pid=" (current-process-id) ", hostname=" (get-host-name) " not starting due to other candidates ahead in start queue")
(tasks:server-delete-records-for-this-pid (db:delay-if-busy tdbdat) " http-transport:launch")
))
(let* ((th2 (make-thread (lambda ()
(debug:print-info 0 #f "Server run thread started")
(http-transport:run
(if (args:get-arg "-server")
(args:get-arg "-server")
"-")
run-id
server-id)) "Server run"))
(th3 (make-thread (lambda ()
(debug:print-info 0 #f "Server monitor thread started")
(http-transport:keep-running server-id run-id))
"Keep running")))
(thread-start! th2)
(thread-sleep! 0.25) ;; give the server time to settle before starting the keep-running monitor.
(thread-start! th3)
(set! *didsomething* #t)
(thread-join! th2)
|
| ︙ | | | ︙ | |
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
|
(print "LOGIN_FAILED")
(exit 1)))))
(define (http-transport:server-signal-handler signum)
(signal-mask! signum)
(handle-exceptions
exn
(debug:print " ... exiting ...")
(let ((th1 (make-thread (lambda ()
(thread-sleep! 1))
"eat response"))
(th2 (make-thread (lambda ()
(debug:print 0 "ERROR: Received ^C, attempting clean exit. Please be patient and wait a few seconds before hitting ^C again.")
(thread-sleep! 3) ;; give the flush three seconds to do it's stuff
(debug:print 0 " Done.")
(exit 4))
"exit on ^C timer")))
(thread-start! th2)
(thread-start! th1)
(thread-join! th2))))
;;======================================================================
|
|
|
|
|
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
|
(print "LOGIN_FAILED")
(exit 1)))))
(define (http-transport:server-signal-handler signum)
(signal-mask! signum)
(handle-exceptions
exn
(debug:print 0 #f " ... exiting ...")
(let ((th1 (make-thread (lambda ()
(thread-sleep! 1))
"eat response"))
(th2 (make-thread (lambda ()
(debug:print 0 #f "ERROR: Received ^C, attempting clean exit. Please be patient and wait a few seconds before hitting ^C again.")
(thread-sleep! 3) ;; give the flush three seconds to do it's stuff
(debug:print 0 #f " Done.")
(exit 4))
"exit on ^C timer")))
(thread-start! th2)
(thread-start! th1)
(thread-join! th2))))
;;======================================================================
|
| ︙ | | | ︙ | |