1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
;; Copyright 2006-2012, Matthew Welland.
;;
;; This program is made available under the GNU GPL version 2.0 or
;; greater. See the accompanying file COPYING for details.
;;
;; This program is distributed WITHOUT ANY WARRANTY; without even the
;; implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
;; PURPOSE.
(require-extension (srfi 18) extras tcp s11n)
(use srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest) ;; sqlite3
;; (import (prefix sqlite3 sqlite3:))
(use spiffy uri-common intarweb http-client spiffy-request-vars intarweb spiffy-directory-listing)
;; Configurations for server
(tcp-buffer-size 2048)
(max-connections 2048)
|
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
;; Copyright 2006-2012, Matthew Welland.
;;
;; This program is made available under the GNU GPL version 2.0 or
;; greater. See the accompanying file COPYING for details.
;;
;; This program is distributed WITHOUT ANY WARRANTY; without even the
;; implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
;; PURPOSE.
(require-extension (srfi 18) extras tcp s11n)
(use srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest posix-extras) ;; sqlite3
;; (import (prefix sqlite3 sqlite3:))
(use spiffy uri-common intarweb http-client spiffy-request-vars intarweb spiffy-directory-listing)
;; Configurations for server
(tcp-buffer-size 2048)
(max-connections 2048)
|
| ︙ | | | ︙ | |
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
;;======================================================================
;; 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 *default-log-port* "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)))
|
|
|
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
;;======================================================================
;; Call this to start the actual server
;;
(define *db:process-queue-mutex* (make-mutex))
(define (http-transport:run hostn)
(debug:print 2 *default-log-port* "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)))
|
| ︙ | | | ︙ | |
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
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
|
(send-response body: "hey there!\n"
headers: '((content-type text/plain))))
((equal? (uri-path (request-uri (current-request)))
'(/ "hey"))
(send-response body: "hey there!\n"
headers: '((content-type text/plain))))
(else (continue))))))))
(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 *default-log-port* "http-transport:try-start-server time=" (seconds->time-string (current-seconds)) " 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 *default-log-port* "WARNING: attempt to start server failed. Trying again ...")
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(debug:print 0 *default-log-port* "exn=" (condition->list exn))
(portlogger:open-run-close portlogger:set-failed portnum)
(debug:print 0 *default-log-port* "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")
(tasks:server-set-interface-port
(db:delay-if-busy tdbdat)
server-id
ipaddrstr 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")
|
|
|
|
<
|
|
<
|
<
<
<
<
<
<
|
<
|
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
(send-response body: "hey there!\n"
headers: '((content-type text/plain))))
((equal? (uri-path (request-uri (current-request)))
'(/ "hey"))
(send-response body: "hey there!\n"
headers: '((content-type text/plain))))
(else (continue))))))))
(http-transport:try-start-server ipaddrstr start-port)))
;; This is recursively run by http-transport:run until sucessful
;;
(define (http-transport:try-start-server ipaddrstr portnum)
(let ((config-hostname (configf:lookup *configdat* "server" "hostname")))
(debug:print-info 0 *default-log-port* "http-transport:try-start-server time=" (seconds->time-string (current-seconds)) " ipaddrsstr=" ipaddrstr " portnum=" portnum " config-hostname=" config-hostname)
(handle-exceptions
exn
(begin
(print-error-message exn)
(if (< portnum 64000)
(begin
(debug:print 0 *default-log-port* "WARNING: attempt to start server failed. Trying again ...")
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(debug:print 0 *default-log-port* "exn=" (condition->list exn))
(portlogger:open-run-close portlogger:set-failed portnum)
(debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
(thread-sleep! 0.1)
;; get_next_port goes here
(http-transport:try-start-server ipaddrstr
(portlogger:open-run-close portlogger:find-port)))
(begin
(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))
(debug:print 0 *default-log-port* "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")
(debug:print 1 *default-log-port* "INFO: server has been stopped"))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;;======================================================================
|
| ︙ | | | ︙ | |
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
|
(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 *default-log-port* "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)))
(debug:print-info 11 *default-log-port* "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
|
|
|
|
>
|
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
(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)(area-dat #f))
(let* ((fullurl (if (vector? serverdat)
(http-transport:server-dat-get-api-req serverdat)
(begin
(debug:print 0 *default-log-port* "FATAL ERROR: http-transport:client-api-send-receive called with no server info")
(exit 1))))
(res (vector #f "uninitialized"))
(success #t)
(sparams (db:obj->string params transport: 'http))
(runremote (or area-dat *runremote*)))
(debug:print-info 11 *default-log-port* "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
|
| ︙ | | | ︙ | |
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
(db:string->obj
(handle-exceptions
exn
(begin
(set! success #f)
(debug:print 0 *default-log-port* "WARNING: failure in with-input-from-request to " fullurl ".")
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(if *runremote*
(remote-conndat-set! *runremote* #f))
;; 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"
(db:obj->string #f))
|
|
|
|
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
|
(db:string->obj
(handle-exceptions
exn
(begin
(set! success #f)
(debug:print 0 *default-log-port* "WARNING: failure in with-input-from-request to " fullurl ".")
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(if runremote
(remote-conndat-set! runremote #f))
;; 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"
(db:obj->string #f))
|
| ︙ | | | ︙ | |
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
|
(signal (make-composite-condition
(make-property-condition
'timeout
'message "nmsg-transport:client-api-send-receive-raw timed out talking to server")))))))
;; careful closing of connections stored in *runremote*
;;
(define (http-transport:close-connections run-id)
(let* ((server-dat (if *runremote*
(remote-conndat *runremote*)
|
|
>
|
|
|
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
|
(signal (make-composite-condition
(make-property-condition
'timeout
'message "nmsg-transport:client-api-send-receive-raw timed out talking to server")))))))
;; careful closing of connections stored in *runremote*
;;
(define (http-transport:close-connections #!key (area-dat #f))
(let* ((runremote (or area-dat *runremote*))
(server-dat (if runremote
(remote-conndat runremote)
#f))) ;; (hash-table-ref/default *runremote* run-id #f)))
(if (vector? server-dat)
(let ((api-dat (http-transport:server-dat-get-api-uri server-dat)))
(close-connection! api-dat)
#t)
#f)))
|
| ︙ | | | ︙ | |
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
435
436
437
438
439
440
441
442
443
444
445
446
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
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
|
(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 *default-log-port* "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 *default-log-port* "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))
(begin
(debug:print-info 0 *default-log-port* "Received server alive signature")
sdat)
(begin
(debug:print-info 0 *default-log-port* "Still waiting, last-sdat=" last-sdat)
(sleep 4)
(if (> (- (current-seconds) start-time) 120) ;; been waiting for two minutes
(begin
(debug:print-error 0 *default-log-port* "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))
(last-access 0)
(server-timeout (server:get-timeout))
(server-going #f))
(let loop ((count 0)
(server-state 'available)
(bad-sync-count 0)
(start-time (current-milliseconds)))
;; Removed code is pasted below (keeping it around until we are clear it is not needed).
;; no *dbstruct-db* yet, set running after our first pass through and start the db
(if (eq? server-state 'available)
(let ((new-server-id (tasks:server-am-i-the-server? (db:delay-if-busy tdbdat) run-id))) ;; try to ensure no double registering of servers
(if (equal? new-server-id server-id)
(begin
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "dbprep")
(thread-sleep! 0.5) ;; give some margin for queries to complete before switching from file based access to server based access
(set! *dbstruct-db* (db:setup)) ;; run-id))
(set! server-going #t)
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "running")
(server:write-dotserver *toppath* (conc iface ":" port))
(delete-file* (conc *toppath* "/.starting-server")))
(begin ;; gotta exit nicely
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "collision")
(http-transport:server-shutdown server-id port))))))
;; when things go wrong we don't want to be doing the various queries too often
;; so we strive to run this stuff only every four seconds or so.
(let* ((sync-time (- (current-milliseconds) start-time))
(rem-time (quotient (- 4000 sync-time) 1000)))
(if (and (<= rem-time 4)
(> rem-time 0))
(thread-sleep! rem-time)))
(if (< count 1) ;; 3x3 = 9 secs aprox
(loop (+ count 1) 'running bad-sync-count (current-milliseconds)))
;; Check that iface and port have not changed (can happen if server port collides)
(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 *default-log-port* "interface changed, refreshing iface and port info")
(set! iface (car sdat))
(set! port (cadr sdat))))
;; Transfer *db-last-access* to last-access to use in checking that we are still alive
(mutex-lock! *heartbeat-mutex*)
(set! last-access *db-last-access*)
(mutex-unlock! *heartbeat-mutex*)
;; (debug:print 11 *default-log-port* "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 *default-log-port* "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 *default-log-port* "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 (current-milliseconds)))
(http-transport:server-shutdown server-id port))))))
;; code cut out from above
;;
;; (condition-case
;; ;; (if (and (member (mutex-state *db-sync-mutex*) '(abandoned not-abandoned))
;; ;; (> (- (current-seconds) *db-last-sync*) 5)) ;; if not currently being synced nor recently synced
;; (db:sync-touched *dbstruct-db* *run-id* force-sync: #t) ;; usually done in the watchdog, not here.
;; ((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-error 0 *default-log-port* "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 *default-log-port* "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 ...
(define (http-transport:server-shutdown server-id port)
(let ((tdbdat (tasks:open-db)))
(debug:print-info 0 *default-log-port* "Starting to shutdown the server.")
;;
;; start_shutdown
;;
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "shutting-down")
(set! *time-to-exit* #t) ;; tell on-exit to be fast as we've already cleaned up
(portlogger:open-run-close portlogger:set-port port "released")
(thread-sleep! 5)
(debug:print-info 0 *default-log-port* "Max cached queries was " *max-cache-size*)
(debug:print-info 0 *default-log-port* "Number of cached writes " *number-of-writes*)
(debug:print-info 0 *default-log-port* "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 *default-log-port* "Number non-cached queries " *number-non-write-queries*)
(debug:print-info 0 *default-log-port* "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 *default-log-port* "Server shutdown complete. Exiting")
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id " http-transport:keep-running complete")
;; if the .server file contained :myport then we can remove it
(server:remove-dotserver-file *toppath* port)
(define (http-transport:launch run-id)
(with-output-to-file
(conc *toppath* "/.starting-server")
(lambda ()
(print (current-process-id) " on " (get-host-name))))
(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 (and (server:read-dotserver *toppath*)
(server:check-if-running run-id))
(begin
(debug:print 0 *default-log-port* "INFO: Server for run-id " run-id " already running")
(exit 0))
(begin ;; ok, no server detected, clean out any lingering records
(tasks:server-force-clean-running-records-for-run-id (db:delay-if-busy tdbdat) run-id "notresponding")))
(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 *default-log-port* "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")
(delete-file* (conc *toppath* "/.starting-server"))
))
(let* ((th2 (make-thread (lambda ()
(debug:print-info 0 *default-log-port* "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 *default-log-port* "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)
(exit))))))
;; (define (http:ping run-id host-port)
;; (let* ((server-dat (http-transport:client-connect (car host-port)(cadr host-port)))
;; (login-res (rmt:login-no-auto-client-setup server-dat run-id)))
;; (if (and (list? login-res)
;; (car login-res))
;; (begin
;; (print "LOGIN_OK")
;; (exit 0))
;; (begin
;; (print "LOGIN_FAILED")
;; (exit 1)))))
(define (http-transport:server-signal-handler signum)
(signal-mask! signum)
(handle-exceptions
exn
(debug:print 0 *default-log-port* " ... exiting ...")
(let ((th1 (make-thread (lambda ()
|
|
|
<
|
|
<
|
>
<
<
<
<
<
<
|
<
<
>
|
|
<
<
>
|
<
<
<
|
|
|
|
|
|
|
>
>
|
>
>
>
|
|
<
<
<
<
>
|
|
|
<
>
|
|
<
|
<
<
<
|
<
|
<
|
<
<
<
<
<
<
<
<
>
|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
|
>
|
|
|
>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
>
>
>
<
<
<
|
<
<
<
<
<
<
|
|
|
|
|
|
|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
|
|
|
|
|
|
<
|
|
|
|
|
|
|
|
|
|
|
<
<
<
<
<
<
<
<
<
<
<
<
|
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
435
436
437
438
439
440
441
442
443
444
445
446
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
|
(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)
;; 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 *default-log-port* "Starting the sync-back, keep alive thread in server")
(let* ((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 *default-log-port* "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))
(begin
(debug:print-info 0 *default-log-port* "Received server alive signature")
sdat)
(begin
(debug:print-info 0 *default-log-port* "Still waiting, last-sdat=" last-sdat)
(sleep 4)
(if (> (- (current-seconds) start-time) 120) ;; been waiting for two minutes
(begin
(debug:print-error 0 *default-log-port* "transport appears to have died, exiting server")
(exit))
(loop start-time
(equal? sdat last-sdat)
sdat)))))))
(iface (car server-info))
(port (cadr server-info))
(last-access 0)
(server-timeout (server:get-timeout))
(server-going #f)
(server-log-file (args:get-arg "-log"))) ;; always set when we are a server
(let loop ((count 0)
(server-state 'available)
(bad-sync-count 0)
(start-time (current-milliseconds)))
;; Use this opportunity to sync the tmp db to megatest.db
(if (not server-going) ;; *dbstruct-db*
(begin
(debug:print 0 *default-log-port* "SERVER: dbprep")
(set! *dbstruct-db* (db:setup)) ;; run-id))
(set! server-going #t)
(debug:print 0 *default-log-port* "SERVER: running, megatest version: " (common:get-full-version)) ;; NOTE: the server is NOT yet marked as running in the log. We do that in the keep-running routine.
(thread-start! *watchdog*)))
;; when things go wrong we don't want to be doing the various queries too often
;; so we strive to run this stuff only every four seconds or so.
(let* ((sync-time (- (current-milliseconds) start-time))
(rem-time (quotient (- 4000 sync-time) 1000)))
(if (and (<= rem-time 4)
(> rem-time 0))
(thread-sleep! rem-time)))
(if (< count 1) ;; 3x3 = 9 secs aprox
(loop (+ count 1) 'running bad-sync-count (current-milliseconds)))
;; Check that iface and port have not changed (can happen if server port collides)
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if (not (equal? sdat (list iface port)))
(let ((new-iface (car sdat))
(new-port (cadr sdat)))
(debug:print-info 0 *default-log-port* "WARNING: interface changed, refreshing iface and port info")
(set! iface new-iface)
(set! port new-port)
(debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds))
(flush-output *default-log-port*)))
;; Transfer *db-last-access* to last-access to use in checking that we are still alive
(mutex-lock! *heartbeat-mutex*)
(set! last-access *db-last-access*)
(mutex-unlock! *heartbeat-mutex*)
(if (common:low-noise-print 120 (conc "server running on " iface ":" port))
(begin
(debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds))
(flush-output *default-log-port*)))
(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 *default-log-port* "Adjusted server timeout: " adjusted-timeout))
(cond
((and *server-run*
(> (+ last-access server-timeout)
(current-seconds))
(< (- (current-seconds) server-start-time) 3600)) ;; do not update log or touch log if we've been running for more than one hour.
(if (common:low-noise-print 120 "server continuing")
(debug:print-info 0 *default-log-port* "Server continuing, seconds since last db access: " (- (current-seconds) last-access))
(let ((curr-time (current-seconds)))
(change-file-times server-log-file curr-time curr-time)))
(loop 0 server-state bad-sync-count (current-milliseconds)))
(else
(debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
(http-transport:server-shutdown port)))))))
(define (http-transport:server-shutdown port)
(let ((tdbdat (tasks:open-db)))
;;(BB> "http-transport:server-shutdown called")
(debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id))
;;
;; start_shutdown
;;
;; (tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "shutting-down")
(set! *time-to-exit* #t) ;; tell on-exit to be fast as we've already cleaned up
(portlogger:open-run-close portlogger:set-port port "released")
(thread-sleep! 1)
;; (debug:print-info 0 *default-log-port* "Max cached queries was " *max-cache-size*)
;; (debug:print-info 0 *default-log-port* "Number of cached writes " *number-of-writes*)
;; (debug:print-info 0 *default-log-port* "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 *default-log-port* "Number non-cached queries " *number-non-write-queries*)
;; (debug:print-info 0 *default-log-port* "Average non-cached time "
;; (if (eq? *number-non-write-queries* 0)
;; "n/a (no queries)"
;; (/ *total-non-write-delay*
;; *number-non-write-queries*))
;; " ms")
(db:print-current-query-stats)
(debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting")
(exit)))
;; all routes though here end in exit ...
;;
;; start_server?
;;
(define (http-transport:launch)
(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*)))))
(let* ((th2 (make-thread (lambda ()
(debug:print-info 0 *default-log-port* "Server run thread started")
(http-transport:run
(if (args:get-arg "-server")
(args:get-arg "-server")
"-")
)) "Server run"))
(th3 (make-thread (lambda ()
(debug:print-info 0 *default-log-port* "Server monitor thread started")
(http-transport:keep-running)
"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)
(exit)))
(define (http-transport:server-signal-handler signum)
(signal-mask! signum)
(handle-exceptions
exn
(debug:print 0 *default-log-port* " ... exiting ...")
(let ((th1 (make-thread (lambda ()
|
| ︙ | | | ︙ | |