| ︙ | | | ︙ | |
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
(declare (uses db))
(declare (uses tests))
(declare (uses tasks)) ;; tasks are where stuff is maintained about what is running.
(declare (uses server))
;; (declare (uses daemon))
(declare (uses portlogger))
(declare (uses rmt))
(include "common_records.scm")
(include "db_records.scm")
(include "js-path.scm")
(require-library stml)
(define (http-transport:make-server-url hostport)
(if (not hostport)
#f
(conc "http://" (car hostport) ":" (cadr hostport))))
|
>
>
>
>
|
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
(declare (uses db))
(declare (uses tests))
(declare (uses tasks)) ;; tasks are where stuff is maintained about what is running.
(declare (uses server))
;; (declare (uses daemon))
(declare (uses portlogger))
(declare (uses rmt))
(declare (uses dbfile))
(declare (uses commonmod))
(include "common_records.scm")
(include "db_records.scm")
(include "js-path.scm")
(import dbfile commonmod)
(require-library stml)
(define (http-transport:make-server-url hostport)
(if (not hostport)
#f
(conc "http://" (car hostport) ":" (cadr hostport))))
|
| ︙ | | | ︙ | |
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
;; This is were we set up the database connections
(let* (($ (request-vars source: 'both))
(dat ($ 'dat))
(res #f))
(cond
((equal? (uri-path (request-uri (current-request)))
'(/ "api"))
(send-response body: (api:process-request *dbstruct-db* $) ;; the $ is the request vars proc
headers: '((content-type text/plain)))
(mutex-lock! *heartbeat-mutex*)
(set! *db-last-access* (current-seconds))
(mutex-unlock! *heartbeat-mutex*))
((equal? (uri-path (request-uri (current-request)))
'(/ ""))
(send-response body: (http-transport:main-page)))
|
|
|
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
;; This is were we set up the database connections
(let* (($ (request-vars source: 'both))
(dat ($ 'dat))
(res #f))
(cond
((equal? (uri-path (request-uri (current-request)))
'(/ "api"))
(send-response body: (api:process-request *dbstruct-dbs* $) ;; the $ is the request vars proc
headers: '((content-type text/plain)))
(mutex-lock! *heartbeat-mutex*)
(set! *db-last-access* (current-seconds))
(mutex-unlock! *heartbeat-mutex*))
((equal? (uri-path (request-uri (current-request)))
'(/ ""))
(send-response body: (http-transport:main-page)))
|
| ︙ | | | ︙ | |
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
;; 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
;;======================================================================
|
|
|
|
|
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
;; 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
;;======================================================================
|
| ︙ | | | ︙ | |
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
|
(if (debug:debug-mode 1)
(debug:print-info 0 *default-log-port* "couldn't talk to server, trying again ...")
(begin
(debug:print 0 *default-log-port* "WARNING: failure in with-input-from-request to " fullurl ".")
(debug:print 0 *default-log-port* " message: " msg ", exn=" exn)
(debug:print 0 *default-log-port* " cmd: " cmd " params: " params " key:" (or server-id "thekey"))
(debug:print 0 *default-log-port* " call-chain: " call-chain)))
(if runremote
(remote-conndat-set! runremote #f))
;;; (signal (make-composite-condition
;;; (make-property-condition 'commfail 'message "failed to connect to server")))
;;; "communications failed"
(db:obj->string #f))
(with-input-from-request ;; was dat
fullurl
(list (cons 'key (or server-id "thekey"))
(cons 'cmd cmd)
(cons 'params sparams))
read-string))
transport: 'http)
0)) ;; added this speculatively
(close-all-connections!)
(mutex-unlock! *http-mutex*)
))
(time-out (lambda ()
(thread-sleep! 45)
(debug:print 0 *default-log-port* "WARNING: send-receive took more than 45 seconds!!")
#f))
(th1 (make-thread send-recieve "with-input-from-request"))
|
>
>
|
|
|
|
|
>
|
|
|
|
|
|
|
|
|
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
|
(if (debug:debug-mode 1)
(debug:print-info 0 *default-log-port* "couldn't talk to server, trying again ...")
(begin
(debug:print 0 *default-log-port* "WARNING: failure in with-input-from-request to " fullurl ".")
(debug:print 0 *default-log-port* " message: " msg ", exn=" exn)
(debug:print 0 *default-log-port* " cmd: " cmd " params: " params " key:" (or server-id "thekey"))
(debug:print 0 *default-log-port* " call-chain: " call-chain)))
(set! *runremote* #f)
(set! runremote #f)
;; (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"
(close-all-connections!)
(db:obj->string #f))
(with-input-from-request ;; was dat
fullurl
(list (cons 'key (or server-id "thekey"))
(cons 'cmd cmd)
(cons 'params sparams))
read-string))
transport: 'http)
0)) ;; added this speculatively
;; Shouldn't this be a call to the managed call-all-connections stuff above?
;; (close-all-connections!) ;; BUG? WHY IS THIS HERE? Are we failing to reuse connections?
(mutex-unlock! *http-mutex*)
))
(time-out (lambda ()
(thread-sleep! 45)
(debug:print 0 *default-log-port* "WARNING: send-receive took more than 45 seconds!!")
#f))
(th1 (make-thread send-recieve "with-input-from-request"))
|
| ︙ | | | ︙ | |
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
|
(let ((api-dat (http-transport:server-dat-get-api-uri server-dat)))
(handle-exceptions
exn
(begin
(print-call-chain *default-log-port*)
(debug:print-error 0 *default-log-port* " closing connection failed with error: " ((condition-property-accessor 'exn 'message) exn) ", exn=" exn))
(close-connection! api-dat)
;;(close-idle-connections!)
#t))
#f)))
(define (make-http-transport:server-dat)(make-vector 6))
(define (http-transport:server-dat-get-iface vec) (vector-ref vec 0))
(define (http-transport:server-dat-get-port vec) (vector-ref vec 1))
|
|
|
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
|
(let ((api-dat (http-transport:server-dat-get-api-uri server-dat)))
(handle-exceptions
exn
(begin
(print-call-chain *default-log-port*)
(debug:print-error 0 *default-log-port* " closing connection failed with error: " ((condition-property-accessor 'exn 'message) exn) ", exn=" exn))
(close-connection! api-dat)
(close-idle-connections!)
#t))
#f)))
(define (make-http-transport:server-dat)(make-vector 6))
(define (http-transport:server-dat-get-iface vec) (vector-ref vec 0))
(define (http-transport:server-dat-get-port vec) (vector-ref vec 1))
|
| ︙ | | | ︙ | |
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
|
;;
(define (http-transport:client-connect iface port server-id)
(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-id)))
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* ((sdat #f)
(tmp-area (common:get-db-tmp-area))
(started-file (conc tmp-area "/.server-started"))
(server-start-time (current-seconds))
(server-info (let loop ((start-time (current-seconds))
(changed #t)
(last-sdat "not this"))
(begin ;; let ((sdat #f))
|
>
>
>
>
|
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
|
;;
(define (http-transport:client-connect iface port server-id)
(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-id)))
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* ((sdat #f)
(no-sync-db (db:open-no-sync-db))
(tmp-area (common:get-db-tmp-area))
(started-file (conc tmp-area "/.server-started"))
(server-start-time (current-seconds))
(server-info (let loop ((start-time (current-seconds))
(changed #t)
(last-sdat "not this"))
(begin ;; let ((sdat #f))
|
| ︙ | | | ︙ | |
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
|
(debug:print 0 *default-log-port* "Failed to create " started-file ", exn=" exn)
(with-output-to-file started-file (lambda ()(print (current-process-id)))))
(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 #t)) ;; 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))
|
>
|
|
>
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
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
|
(debug:print 0 *default-log-port* "Failed to create " started-file ", exn=" exn)
(with-output-to-file started-file (lambda ()(print (current-process-id)))))
(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-dbs*
(begin
(debug:print 0 *default-log-port* "SERVER: dbprep")
(set! *dbstruct-dbs* (db:setup #t)) ;; run-id)) FIXME!!!
(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*)
)
(if (and no-sync-db
(common:low-noise-print 5 "sync-all")) ;; cheesy way to reduce frequency of running sync :)
(begin
(debug:print 0 *default-log-port* "keep-running calling db:all-db-sync at " (time->string (seconds->local-time) "%H:%M:%S"))
;; This is tougher than it seems - have to deal with multiple dbs
;; (db:process-transaction-queue *dbstruct-dbs*)
(db:all-db-sync *dbstruct-dbs*)
;; (db:do-sync no-sync-db)
;; (db:run-lock-and-sync *no-sync-db*)
)
)
)
;; 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))
|
| ︙ | | | ︙ | |
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
|
(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)
(if (not *server-id*)
(set! *server-id* (server:mk-signature)))
(debug:print 0 *default-log-port* (current-seconds) (current-directory) (current-process-id) (argv))
(set! *server-id* (server:mk-signature)))
(debug:print 0 *default-log-port* (current-seconds) (current-directory) (current-process-id) (argv))
(debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds) " server-id: " *server-id*)
(flush-output *default-log-port*)))
(if (not *server-overloaded*)
(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)
(begin
|
<
|
|
|
|
>
>
>
>
>
>
>
>
>
>
>
>
>
|
|
>
>
>
|
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
|
(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)
(if (not *server-id*)
(set! *server-id* (server:mk-signature)))
(debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds) " server-id: " *server-id*)
(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
(if (not *server-id*)
(set! *server-id* (server:mk-signature)))
(debug:print 0 *default-log-port* (current-seconds) (current-directory) (current-process-id) (argv))
(debug:print 0 *default-log-port* "SERVER STARTED: " iface ":" port " AT " (current-seconds) " server-id: " *server-id*)
(flush-output *default-log-port*)))
(if (common:low-noise-print 60 "dbstats")
(begin
(debug:print 0 *default-log-port* "Server stats:")
(db:print-current-query-stats)))
(let* ((hrs-since-start (/ (- (current-seconds) server-start-time) 3600)))
(cond
#;((and *server-run*
(> (- (current-seconds) server-start-time) 420)) ;; let's try server replacement
;; ((adj-proc-load . 0.056875) (adj-core-load . 0.11375) (1m-load . 0.91) (5m-load . 0.77) (15m-load . 1.0) (proc . 16) (core . 8) (phys . 1))
(let* ((loaddat (common:get-normalized-cpu-load #f))
(adj-proc-load (alist-ref 'adj-proc-load loaddat))
(adj-core-load (alist-ref 'adj-core-load loaddat))
(adj-load (max adj-proc-load adj-core-load)))
(if (< adj-load 2) ;; reduce chance of runaway
(server:run *toppath*))
(db:all-db-sync *dbstruct-dbs*)
(thread-sleep! 30)
(http-transport:server-shutdown port)))
((and *server-run*
(> (+ last-access server-timeout)
(current-seconds)))
(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)))
(handle-exceptions
exn
(debug:print 0 *default-log-port* "ERROR: Failed to change timestamp on log file " server-log-file ". Are you out of space on that disk? exn=" exn)
(if (and (< (- (current-seconds) server-start-time) 600) ;; run for ten minutes for experiment, 3600 thereafter
(not *server-overloaded*))
(change-file-times server-log-file curr-time curr-time)
(if (common:low-noise-print 120 "start new server")
(server:kind-run *toppath*) ;; server:kind-run uses [servers] numservers
)))))
(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)
(begin
|
| ︙ | | | ︙ | |