Megatest

Check-in [7ce5c6cfb0]
Login
☰Home Timeline Files Branches Tags Tickets Wiki Login
Overview
Comment:rpc server starts now
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | v1.62-rpc
Files: files | file ages | folders
SHA1: 7ce5c6cfb00ab3372eec6b614ddab232df794c00
User & Date: bjbarcla on 2016-12-05 19:38:33
Other Links: branch diff | manifest | tags
Context
2016-12-05
21:57
rpc server starts now check-in: 0b9de1bde2 user: bjbarcla tags: v1.62-rpc
19:38
rpc server starts now check-in: 7ce5c6cfb0 user: bjbarcla tags: v1.62-rpc
15:39
merged v1.63 check-in: be6a92a2af user: bjbarcla tags: v1.62-rpc
Changes
Hide Diffs Side-by-Side Diffs Ignore Whitespace Patch

Modified http-transport.scm from [4d8eecbf3a] to [400911ce2a].

︙
82
83
84
85
86
87
88
89

90
91
92
93
94
95
96

82
83
84
85
86
87
88

89
90
91
92
93
94
95
96








-
+








				      (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))
				   (set! *last-db-access* (current-seconds))
				   (mutex-unlock! *heartbeat-mutex*))
				  ((equal? (uri-path (request-uri (current-request))) 
					   '(/ ""))
				   (send-response body: (http-transport:main-page)))
				  ((equal? (uri-path (request-uri (current-request))) 
					   '(/ "json_api"))
				   (send-response body: (http-transport:main-page)))
︙
424
425
426
427
428
429
430
431

432
433

434
435
436
437
438
439
440

424
425
426
427
428
429
430

431
432

433
434
435
436
437
438
439
440








-
+

-
+








      (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
      ;; Transfer *last-db-access* to last-access to use in checking that we are still alive
      (mutex-lock! *heartbeat-mutex*)
      (set! last-access *db-last-access*)
      (set! last-access *last-db-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)
︙
538
539
540
541
542
543
544
545

546
547
548
549
550
551

552
553
554
555
556
557
558

538
539
540
541
542
543
544

545
546
547
548
549
550

551
552
553
554
555
556
557
558








-
+





-
+








    (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))
    (let loop ((server-id (tasks:server-lock-slot (db:delay-if-busy tdbdat) run-id 'http))
	       (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)
		(loop (tasks:server-lock-slot (db:delay-if-busy tdbdat) run-id 'http)
		      (- 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"))
		))
︙
636
637
638
639
640
641
642
643

644
645
646
647
648
649
650

636
637
638
639
640
641
642

643
644
645
646
647
648
649
650








-
+








	       " ms</td></tr>"
	       "<tr><td>Number non-cached queries</td> <td>"  *number-non-write-queries* "</td></tr>"
	       "<tr><td>Average non-cached time</td>   <td>" (if (eq? *number-non-write-queries* 0)
								 "n/a (no queries)"
								 (/ *total-non-write-delay* 
								    *number-non-write-queries*))
	       " ms</td></tr>"
	       "<tr><td>Last access</td><td>"              (seconds->time-string *db-last-access*) "</td></tr>"
	       "<tr><td>Last access</td><td>"              (seconds->time-string *last-db-access*) "</td></tr>"
	       "</table>")))
    (mutex-unlock! *heartbeat-mutex*)
    res))

(define (http-transport:runs linkpath)
  (conc "<h3>Runs</h3>"
	(string-intersperse
︙

Modified rpc-transport.scm from [261c47f7d8] to [fada9c4b7e].

︙
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

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








-
-
+
+








-
+






-
+














+
+
-
-
-
+
+
+
+








                (if chatty (print " + last try failed with exception- re-throw it >"res-value"<"))
                (abort res-value)); re-raise the exception. TODO: find a way for call-history to show as though from entry to this function
              (begin
                (if chatty (print " + last try failed with exception- return canned failure value >"failure-value"<"))
                failure-value))))))))


(define (rpc-transport:server-shutdown server-id rpc:listener #!key (from-on-exit #f))
  (on-exit (lambda () #t)) ;; turn off on-exit stuff
(define (rpc-transport:server-shutdown server-id rpc:listener ) ;;#!key (from-on-exit #f))
  ;;(on-exit (lambda () #t)) ;; turn off on-exit stuff
  ;;(tcp-close rpc:listener) ;; gotta exit nicely
  ;;(tasks:server-set-state! (db:delay-if-busy (tasks:open-db)) server-id "stopped")


  ;; TODO: (low) the following is extraordinaritly slow.  Maybe we don't even need portlogger for rpc anyway??  the exception-based failover when ports are taken is fast!
  ;;(portlogger:open-run-close portlogger:set-port (rpc:default-server-port) "released")
  
  (set! *time-to-exit* #t)
  (if *inmemdb* (db:sync-touched *inmemdb* *run-id* force-sync: #t))
  ;;(if *inmemdb* (db:sync-touched *inmemdb* *run-id* force-sync: #t))


  (tasks:server-delete-record (db:delay-if-busy (tasks:open-db)) server-id " rpc-transport:keep-running complete")
  

  ;;(BB> "Before (exit) (from-on-exit="from-on-exit")")
  (unless from-on-exit (exit))  ;; sometimes we hang (around) here with 100% cpu.
  ;;(unless from-on-exit (exit))  ;; sometimes we hang (around) here with 100% cpu.
  ;;(BB> "After")
  ;; strace reveals endless:
  ;; getrusage(RUSAGE_SELF, {ru_utime={413, 917868}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 9874}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 13874}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 105880}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 109880}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 201886}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 205886}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 297892}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 301892}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 393898}, ru_stime={0, 60003}, ...}) = 0
  ;; getrusage(RUSAGE_SELF, {ru_utime={414, 397898}, ru_stime={0, 60003}, ...}) = 0
  ;; make a post to chicken-users w/ http://paste.call-cc.org/paste?id=60a4b66a29ccf7d11359ea866db642c970735978


  (if from-on-exit
      ;; avoid above condition!  End current process externally  since 1 in 20 (exit)'s result in hung, 100% cpu zombies. (see above)
      (system (conc  "kill -9  "(current-process-id))))
  ;; (if from-on-exit
  ;;     ;; avoid above condition!  End current process externally  since 1 in 20 (exit)'s result in hung, 100% cpu zombies. (see above)
  
  (system (conc  "kill -9  "(current-process-id)))
  )


;; all routes though here end in exit ...
;;
;; start_server? 
;;
︙
203
204
205
206
207
208
209





































210
211
212
213
214

215
216
217
218
219
220
221

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








+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+





+








  ;;       (current-output-port *alt-log-file*)))

  ;; double check we dont alrady have a running server for this run-id
  (when (server:check-if-running run-id)
    (debug:print 0 *default-log-port* "INFO: Server for run-id " run-id " already running")
    (exit 0))


  ;; clean up dead servers (duped in megatest.scm in -list-servers processing; may want to consolidate into proc)
  (for-each
   (lambda (server)
     (let* ((id         (vector-ref server 0))
            (pid        (vector-ref server 1))
            (hostname   (vector-ref server 2))
            (interface  (vector-ref server 3)) 
            (pullport   (vector-ref server 4))
            (pubport    (vector-ref server 5))
            (start-time (vector-ref server 6))
            (priority   (vector-ref server 7))
            (state      (vector-ref server 8))
            (mt-ver     (vector-ref server 9))
            (last-update (vector-ref server 10)) 
            (transport  (vector-ref server 11))
            (killed     #f)
            (status     (< last-update 20)))

       (if (equal? state "dead")
           (if (> last-update (* 25 60 60)) ;; keep records around for slighly over a day.
               (tasks:server-deregister (db:delay-if-busy (tasks:open-db)) hostname pullport: pullport pid: pid action: 'delete))
           (if (> last-update 20)        ;; Mark as dead if not updated in last 20 seconds
               (tasks:server-deregister (db:delay-if-busy (tasks:open-db)) hostname pullport: pullport pid: pid)))
       ;;(format #t fmtstr id mt-ver pid hostname (conc interface ":" pullport) pubport last-update
       ;;        (if status "alive" "dead") transport)
       ;; (if (or (equal? id sid)
       ;;         (equal? sid 0)) ;; kill all/any
       ;;     (begin
       ;;       (debug:print-info 0 *default-log-port* "Attempting to kill "kill-switch" server with pid " pid)
       ;;       (tasks:kill-server hostname pid kill-switch: kill-switch)))

       )
       
     )
   (tasks:get-all-servers (db:delay-if-busy (tasks:open-db))))
   
  ;; let's get a server-id for this server
  ;;   if at first we do not suceed, try 3 more times.
  (let ((server-id (retry-thunk
                    (lambda () (tasks:server-lock-slot (db:delay-if-busy (tasks:open-db)) run-id 'rpc))
                    chatty: #f
                    final-failure-returns-actual: #t
                    retries: 4)))
    (when (not server-id) ;; dang we couldn't get a server-id.
      ;; 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 (tasks:open-db)) " rpc-transport:launch")
      (exit 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
412
413
414
415



416
417
418






419



420
421
422



423
424
425
426
427
428
429

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








-
-
-
-
-
-
-













-
+
+
+



+
+
+
+
+
+

+
+
+
-
-
-
+
+
+








	 (ipaddrstr       (if (string=? "-" hostn)
			      (server:get-best-guess-address hostname) ;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
			      #f))
	 (portnum         (let ((res (rpc:default-server-port)))  res))
	 (host:port       (conc (if ipaddrstr ipaddrstr hostname) ":" portnum)))


    ;; BB> TODO: remove portlogger!
    ;; if rpc found it needed a different port than portlogger provided, keep portlogger in the loop.
    ;; (when (not (equal? start-port portnum))
    ;;   (BB> "portlogger proffered "start-port" but rpc grabbed "portnum)
    ;;   (portlogger:open-run-close portlogger:set-port start-port "released")
    ;;   (portlogger:open-run-close portlogger:take-port portnum))

    (tasks:server-set-interface-port (db:delay-if-busy (tasks:open-db)) server-id ipaddrstr portnum)

    ;;============================================================
    ;;  activate thread th1 to attach opened tcp port to rpc server
    ;;=============================================================
    (thread-start! th1)
    (set! db *inmemdb*)

    (debug:print 0 *default-log-port* "Server started on " host:port)

    ;; (thread-sleep! 5)

    (if (retry-thunk (lambda ()
                       (rpc-transport:self-test run-id ipaddrstr portnum)))
                       (rpc-transport:self-test run-id ipaddrstr portnum))
                     final-failure-returns-actual: #t
                     )
        (debug:print 0 *default-log-port* "INFO: rpc self test passed!")
        (begin
          (debug:print 0 *default-log-port* "Error: rpc listener did not pass self test.  Shutting down.  On: " host:port)
          (BB> 1)
          (tasks:server-set-state! (db:delay-if-busy (tasks:open-db)) server-id "dead")
          (BB> 2)
          (tcp-close rpc:listener) ;; gotta exit nicely and free up that tcp port
          (BB> 3)
          (rpc-transport:server-shutdown server-id rpc:listener)
          (exit)))
    (mutex-lock! *heartbeat-mutex*)
    (set! *last-db-access* (current-seconds))
    (mutex-unlock! *heartbeat-mutex*)
    
    (on-exit (lambda ()
               (rpc-transport:server-shutdown server-id rpc:listener from-on-exit: #t)))

    ;;(on-exit (lambda ()
    ;;           (rpc-transport:server-shutdown server-id rpc:listener from-on-exit: #t)))
    
    ;; check again for running servers for this run-id in case one has snuck in since we checked last in rpc-transport:launch
    (if (not (equal? server-id (tasks:server-am-i-the-server? (db:delay-if-busy (tasks:open-db)) run-id)));; try to ensure no double registering of servers
        (begin ;; i am not the server, another server snuck in and beat this one to the punch
          (tcp-close rpc:listener) ;; gotta exit nicely and free up that tcp port
          (tasks:server-set-state! (db:delay-if-busy (tasks:open-db)) server-id "collision"))
︙
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

598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615

616
617
618

619
620
621
622
623
624
625
626








+
+
+
+







-
+


-
+








	   (print "LOGIN_OK")
	   (exit 0))
	 (begin
	   (print "LOGIN_FAILED")
	   (exit 1))))))

(define (rpc-transport:self-test run-id host port)
  (if (not host)
      (abort "host not set."))
  (if (not port)
      (abort "port not set."))
  (tcp-buffer-size 0) ;; gotta do this because http-transport undoes it.
  (let* ((testing-res ((rpc:procedure 'testing host port)))
         (login-res ((rpc:procedure 'server:login host port) *toppath*))
         (res (and login-res (equal? testing-res "Just testing"))))
    
    (if login-res
        (begin
          ;;(BB> "Self test PASS.  login-res="login-res" testing-res="testing-res" *toppath*="*toppath*)
          (BB> "Self test PASS.  login-res="login-res" testing-res="testing-res" *toppath*="*toppath*)
          #t)
        (begin
          ;;(BB> "Self test fail.  login-res="login-res" testing-res="testing-res" *toppath*="*toppath*)
          (BB> "Self test fail.  login-res="login-res" testing-res="testing-res" *toppath*="*toppath*)
           
          #f))
    res))

(define (rpc-transport:client-setup run-id server-dat #!key (remtries 10))
  ;;(BB> "entered rpc-transport:client-setup with run-id="run-id" and server-dat="server-dat" and retries="remtries)
  (tcp-buffer-size 0)
︙

Modified server.scm from [3b30831198] to [6aa7bdfd53].

︙
44
45
46
47
48
49
50
51
52
53
54
55
56
57













58
59
60

61
62
63
64
65
66
67

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








-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
+








;; Call this to start the actual server
;;

;; all routes though here end in exit ...
;;
;; start_server
;;
(define (server:launch run-id transport-type)
  (BB> "server:launch fired for run-id="run-id" transport-type="transport-type)
  (case transport-type
    ((http)(http-transport:launch run-id))
    ;;((nmsg)(nmsg-transport:launch run-id))
    ((rpc)  (rpc-transport:launch run-id))
    (else (debug:print-error 0 *default-log-port* "unknown server type " transport-type))))
(define (server:launch run-id transport-type-raw)
  (let ((transport-type
         (cond
          ((string? transport-type-raw) (string->symbol transport-type-raw))
          (else transport-type-raw))))
         
    (BB> "server:launch fired for run-id="run-id" transport-type="transport-type)

    (case transport-type
      ((http)(http-transport:launch run-id))
      ;;((nmsg)(nmsg-transport:launch run-id))
      ((rpc)  (rpc-transport:launch run-id))
      (else (debug:print-error 0 *default-log-port* "unknown server type " transport-type)))))
;;       (else   (debug:print-error 0 *default-log-port* "No known transport set, transport=" transport ", using rpc")
;; 	      (rpc-transport:launch run-id)))))

  
;;======================================================================
;; S E R V E R   U T I L I T I E S 
;;======================================================================

;; Get the transport
(define (server:get-transport)
  (if *transport-type*
︙

Modified tasks.scm from [b8a3c2af2e] to [ae98efc45e].

︙
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

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








-
+



-
+





-
+













-
+








(define (tasks:hostinfo-get-interface   vec)    (vector-ref  vec 1))
(define (tasks:hostinfo-get-port        vec)    (vector-ref  vec 2))
(define (tasks:hostinfo-get-pubport     vec)    (vector-ref  vec 3))
(define (tasks:hostinfo-get-transport   vec)    (vector-ref  vec 4))
(define (tasks:hostinfo-get-pid         vec)    (vector-ref  vec 5))
(define (tasks:hostinfo-get-hostname    vec)    (vector-ref  vec 6))

(define (tasks:server-lock-slot mdb run-id)
(define (tasks:server-lock-slot mdb run-id transport-type)
  (tasks:server-clean-out-old-records-for-run-id mdb run-id " tasks:server-lock-slot")
  (if (< (tasks:num-in-available-state mdb run-id) 4)
      (begin 
	(tasks:server-set-available mdb run-id)
	(tasks:server-set-available mdb run-id transport-type)
	(thread-sleep! (/ (random 1500) 1000)) ;; (thread-sleep! 2) ;; Try removing this. It may not be needed.
	(tasks:server-am-i-the-server? mdb run-id))
      #f))
	
;; register that this server may come online (first to register goes though with the process)
(define (tasks:server-set-available mdb run-id)
(define (tasks:server-set-available mdb run-id transport-type)
  (sqlite3:execute 
   mdb 
   "INSERT INTO servers (pid,hostname,port,pubport,start_time,      priority,state,mt_version,heartbeat,   interface,transport,run_id)
                   VALUES(?, ?,       ?,   ?, strftime('%s','now'), ?,       ?,    ?,-1,?,        ?,        ?);"
   (current-process-id)          ;; pid
   (get-host-name)               ;; hostname
   -1                            ;; port
   -1                            ;; pubport
   (random 1000)                 ;; priority (used a tiebreaker on get-available)
   "available"                   ;; state
   (common:version-signature)    ;; mt_version
   -1                            ;; interface
   ;; (conc (server:get-transport)) ;; transport
   (conc *transport-type*)    ;; transport
   (symbol->string transport-type)    ;; transport
   run-id
   ))

(define (tasks:num-in-available-state mdb run-id)
  (let ((res 0))
    (sqlite3:for-each-row
     (lambda (num-in-queue)
︙

This page was generated in about 0.114s by Fossil 2.24 [452f8e1ffc] 2024-03-26 15:36:00