Megatest

Check-in [58d1011733]
Login
Overview
Comment:made changes to fix tcp-buffer size to 0 so rpc worked. small fight here between http-transport and rpc-transport here... hopefully this patch does the job.
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | rpc-transport
Files: files | file ages | folders
SHA1: 58d101173394e09a56cd84ef1c6ef13892501e8d
User & Date: bjbarcla on 2016-11-02 03:09:14
Other Links: branch diff | manifest | tags
Context
2016-11-03
16:52
another pass to allow distinct per-run-id transports to be used check-in: 140ed85cfb user: bjbarcla tags: rpc-transport
2016-11-02
03:09
made changes to fix tcp-buffer size to 0 so rpc worked. small fight here between http-transport and rpc-transport here... hopefully this patch does the job. check-in: 58d1011733 user: bjbarcla tags: rpc-transport
00:44
Overhauled rpc-transport:launch and rpc-transport:run to account for things introduced to http-transport like inmem db maintenance ; made and used rpc-transport:server-shutdown ; made (more) opinionaltedly encapsulated procedures for task unit check-in: c88c8f26e0 user: bjbarcla tags: rpc-transport
Changes

Modified fs-transport.scm from [59920959a9] to [311c358987].

11
12
13
14
15
16
17
18

19
20
21
22
23
24
25
(require-extension (srfi 18) extras tcp s11n)

(use sqlite3 srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest)
(import (prefix sqlite3 sqlite3:))

(use spiffy uri-common intarweb http-client spiffy-request-vars)

(tcp-buffer-size 2048)


(declare (unit fs-transport))

(declare (uses common))
(declare (uses db))
(declare (uses tests))
(declare (uses tasks)) ;; tasks are where stuff is maintained about what is running.







|
>







11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
(require-extension (srfi 18) extras tcp s11n)

(use sqlite3 srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest)
(import (prefix sqlite3 sqlite3:))

(use spiffy uri-common intarweb http-client spiffy-request-vars)

;;(tcp-buffer-size 2048)
(BB> "HEY TURNING OFF tcp-buffer-size TO TEST FOR RPC SIDE EFFECT>  TURN BACK ON BEFORE PRODUCTION")

(declare (unit fs-transport))

(declare (uses common))
(declare (uses db))
(declare (uses tests))
(declare (uses tasks)) ;; tasks are where stuff is maintained about what is running.

Modified http-transport.scm from [8236b0c25e] to [b16e6277ce].

12
13
14
15
16
17
18

19



20
21
22
23
24
25
26

(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) 

(declare (unit http-transport))

(declare (uses common))
(declare (uses db))
(declare (uses tests))







>
|
>
>
>







12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

(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) ;; this interferes with rpc ; compensating in rpc-transport... so far so good



(max-connections 2048) 

(declare (unit http-transport))

(declare (uses common))
(declare (uses db))
(declare (uses tests))

Modified rpc-transport.scm from [bd4853520b] to [4a03110bb8].

121
122
123
124
125
126
127

128





129
130







131
132
133
134
135
136
137
  (debug:print 2 *default-log-port* "Attempting to start the rpc server ...")
   ;; (trace rpc:publish-procedure!)

  ;;======================================================================
  ;;	  start of publish-procedure section
  ;;======================================================================
  (rpc:publish-procedure! 'server:login server:login) ;; this allows client to validate it is the same megatest instance as the server.  No security here, just making sure we're in the right room.

  (rpc:publish-procedure! 'testing (lambda () "Just testing"))






  ;; BB: BBTODO: publish procedure to receive request from client's rpc:send-receive/rpc-transport:client-api-send-receive call







  
  ;;======================================================================
  ;;	  end of publish-procedure section
  ;;======================================================================










>
|
>
>
>
>
>

|
>
>
>
>
>
>
>







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
  (debug:print 2 *default-log-port* "Attempting to start the rpc server ...")
   ;; (trace rpc:publish-procedure!)

  ;;======================================================================
  ;;	  start of publish-procedure section
  ;;======================================================================
  (rpc:publish-procedure! 'server:login server:login) ;; this allows client to validate it is the same megatest instance as the server.  No security here, just making sure we're in the right room.
  (BB> "published 'testing")
  (rpc:publish-procedure!
   'testing
   (lambda ()
     (BB> "Current-peer=["(rpc:current-peer)"]")
     (BB> "published rpc proc 'testing was invoked")
     "Just testing"))

  ;; procedure to receive arbitrary API request from client's rpc:send-receive/rpc-transport:client-api-send-receive 
  (rpc:publish-procedure! 'rpc-transport:autoremote rpc-transport:autoremote)
  ;; can use this to run most anything at the remote
  (rpc:publish-procedure! 
   'remote:run 
   (lambda (procstr . params)
     (server:autoremote procstr params)))
  
  
  ;;======================================================================
  ;;	  end of publish-procedure section
  ;;======================================================================



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
198
         ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
         ;; rpc:listener is the tcp-listen result from inside the find-free-port-and-open complex.
         ;;   It is our handle on the listening tcp port
         ;;   We will attach this to our rpc server with rpc:make-server in thread th1 .
	 (rpc:listener    (rpc-transport:find-free-port-and-open start-port)) 
	 (th1             (make-thread
			   (lambda ()


			     ((rpc:make-server rpc:listener) #t))

			   "rpc:server"))
         
			   ;; (cute (rpc:make-server rpc:listener) "rpc:server")
			   ;; 'rpc:server))
	 (hostname        (if (string=? "-" hostn)
			      (get-host-name) 
			      hostn))
	 (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))) (BB> "rpc:default-server-port="res" rpc-listener-port="*rpc-listener-port*) res))
	 (host:port       (conc (if ipaddrstr ipaddrstr hostname) ":" portnum)))

    ;; 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))



    ;;============================================================
    ;;  activate thread th1 to attach opened tcp port to rpc server
    ;;=============================================================
    (BB> "Got here before thread start of rpc listener")
    (thread-start! th1)


    (BB> "started rpc server thread th1="th1)
    (set! db *inmemdb*)
o    (tasks:bb-server-set-interface-port server-id ipaddrstr portnum)
    (debug:print 0 *default-log-port* "Server started on " host:port)
    











    (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:bb-server-am-i-the-server? 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







>
>
|
>

|
|
<
|









|
|
|
|
>
>
>





|

<

|



>
>
>
>
>
>
>
>
>
>







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
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
         ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
         ;; rpc:listener is the tcp-listen result from inside the find-free-port-and-open complex.
         ;;   It is our handle on the listening tcp port
         ;;   We will attach this to our rpc server with rpc:make-server in thread th1 .
	 (rpc:listener    (rpc-transport:find-free-port-and-open start-port)) 
	 (th1             (make-thread
			   (lambda ()
                             (BB> "+++ before rpc:make-server "rpc:listener)
                             ;;(cute (rpc:make-server rpc:listener) "rpc:server")
			     ((rpc:make-server rpc:listener) #t)
                             (BB> "--- after rpc:make-server"))
			   "rpc:server"))



         (hostname        (if (string=? "-" hostn)
			      (get-host-name) 
			      hostn))
	 (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))) (BB> "rpc:default-server-port="res" rpc-listener-port="*rpc-listener-port*) res))
	 (host:port       (conc (if ipaddrstr ipaddrstr hostname) ":" portnum)))

    ;; 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:bb-server-set-interface-port server-id ipaddrstr portnum)

    ;;============================================================
    ;;  activate thread th1 to attach opened tcp port to rpc server
    ;;=============================================================
    (BB> "Got here before thread start of rpc listener")
    (thread-start! th1)
    (BB> "started rpc server thread th1="th1)


    (set! db *inmemdb*)

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

    (thread-sleep! 8)
    (BB> "before self test")
    (if (rpc-transport:self-test run-id ipaddrstr portnum)
        (BB> "Pass self-test.")
        (begin
          (print "Error: rpc listener did not pass self test.  Shutting down.")
          (exit)))
    (BB> "after self test")

    
    (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:bb-server-am-i-the-server? 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
317
318
319
320
321
322
323
324
325
326
327
328

329
330


331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346

























347
348
349
350
351
352
353


(define (rpc-transport:find-free-port-and-open port #!key )
  (handle-exceptions
   exn
   (begin
     (print "Failed to bind to port " (rpc:default-server-port) ", trying next port")
     (rpc-transport:find-free-port-and-open (+ port 1)))
   (rpc:default-server-port port)
   (set! *rpc-listener-port* port) ;; a bit paranoid about rpc:default-server-port parameter not changing across threads (as params are wont to do).  keeping this global in my back pocket in case this causes problems
   (set! *rpc-listener-port-bind-timestamp* (current-milliseconds)) ;; may want to test how long it has been since the last bind attempt happened...
   (tcp-read-timeout 240000)

   (BB> "rpc-transport> attempting to bind tcp port "port)
   (tcp-listen (rpc:default-server-port) 10000)))


  
(define (rpc-transport:ping run-id host port)
  (handle-exceptions
   exn
   (begin
     (print "SERVER_NOT_FOUND")
     (exit 1))
   (let ((login-res ((rpc:procedure 'server:login host port) *toppath*)))
     (if (and (list? login-res)
	      (car login-res))
	 (begin
	   (print "LOGIN_OK")
	   (exit 0))
	 (begin
	   (print "LOGIN_FAILED")
	   (exit 1))))))


























(define (rpc-transport:client-setup run-id #!key (remtries 10))
  (if *runremote*
      (begin
	(debug:print-error 0 *default-log-port* "Attempt to connect to server but already connected")
	#f)
      (let* ((host-info (hash-table-ref/default *runremote* run-id #f))) ;; (open-run-close db:get-var #f "SERVER"))







|




>

|
>
>








<
|






>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>







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


(define (rpc-transport:find-free-port-and-open port #!key )
  (handle-exceptions
   exn
   (begin
     (print "Failed to bind to port " (rpc:default-server-port) ", trying next port")
     (rpc-transport:find-free-port-and-open (add1 port)))
   (rpc:default-server-port port)
   (set! *rpc-listener-port* port) ;; a bit paranoid about rpc:default-server-port parameter not changing across threads (as params are wont to do).  keeping this global in my back pocket in case this causes problems
   (set! *rpc-listener-port-bind-timestamp* (current-milliseconds)) ;; may want to test how long it has been since the last bind attempt happened...
   (tcp-read-timeout 240000)
   (tcp-buffer-size 0) ;; gotta do this because http-transport undoes it.
   (BB> "rpc-transport> attempting to bind tcp port "port)
   (tcp-listen (rpc:default-server-port) 10000)
   ;;(tcp-listen (rpc:default-server-port) )
   ))
  
(define (rpc-transport:ping run-id host port)
  (handle-exceptions
   exn
   (begin
     (print "SERVER_NOT_FOUND")
     (exit 1))
   (let ((login-res ((rpc:procedure 'server:login host port) *toppath*)))

     (if login-res
	 (begin
	   (print "LOGIN_OK")
	   (exit 0))
	 (begin
	   (print "LOGIN_FAILED")
	   (exit 1))))))

(define (rpc-transport:self-test run-id host port)
  (BB> "SELF TEST RPC ... *toppath*="*toppath*)
  (BB> "local: [" (server:login *toppath*) "]")
  ;(handle-exceptions
   ;exn
   ;(begin
   ;  (BB> "SERVER_NOT_FOUND")
   ;  #f)
  (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"))))

     (BB> "testing-res = >"testing-res"<")
     (BB> "login-res = >"testing-res"<")
     (if login-res
	 (begin
	   (BB> "LOGIN_OK")
	   #t)
	 (begin
	   (BB> "LOGIN_FAILED")
	   #f))
     (BB> "self test res="res)
     res));)

(define (rpc-transport:client-setup run-id #!key (remtries 10))
  (if *runremote*
      (begin
	(debug:print-error 0 *default-log-port* "Attempt to connect to server but already connected")
	#f)
      (let* ((host-info (hash-table-ref/default *runremote* run-id #f))) ;; (open-run-close db:get-var #f "SERVER"))

Modified server.scm from [b3bc3d6537] to [fc86462d4a].

262
263
264
265
266
267
268
269
270

271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
	     (else       #f))
	   (loop (read-line) inl))))))


;; Client will call this procedure on the server via the low-level transport (http/rpc/etc) to verify its toppath matches the server's toppath.
;; A true result means client and server are associated with same megatest instance, share the same megatest.config, etc...)  A false result means the client should not talk to this server.
(define (server:login toppath)
  (lambda (toppath)
    (set! *last-db-access* (current-seconds))

    (if (equal? *toppath* toppath)
	(begin
	  ;; (debug:print-info 2 *default-log-port* "login successful")
	  #t)
	(begin
	  ;; (debug:print-info 2 *default-log-port* "login failed")
	  #f))))

(define (server:get-timeout)
  (let ((tmo (configf:lookup  *configdat* "server" "timeout")))
    (if (and (string? tmo)
	     (string->number tmo))
	(* 60 60 (string->number tmo))
	;; (* 3 24 60 60) ;; default to three days
	(* 60 1)         ;; default to one minute
	;; (* 60 60 25)      ;; default to 25 hours
	)))








<
|
>
|
|
|
|
|
|
|











262
263
264
265
266
267
268

269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
	     (else       #f))
	   (loop (read-line) inl))))))


;; Client will call this procedure on the server via the low-level transport (http/rpc/etc) to verify its toppath matches the server's toppath.
;; A true result means client and server are associated with same megatest instance, share the same megatest.config, etc...)  A false result means the client should not talk to this server.
(define (server:login toppath)

  (set! *last-db-access* (current-seconds))
  (BB> "server:login ours="*toppath*" theirs="toppath)
  (if (equal? *toppath* toppath)
      (begin
        ;; (debug:print-info 2 *default-log-port* "login successful")
        #t)
      (begin
        ;; (debug:print-info 2 *default-log-port* "login failed")
        #f)))

(define (server:get-timeout)
  (let ((tmo (configf:lookup  *configdat* "server" "timeout")))
    (if (and (string? tmo)
	     (string->number tmo))
	(* 60 60 (string->number tmo))
	;; (* 3 24 60 60) ;; default to three days
	(* 60 1)         ;; default to one minute
	;; (* 60 60 25)      ;; default to 25 hours
	)))