11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
(use json format) ;; RADT => purpose of json format??
(declare (unit rmt))
(declare (uses api))
(declare (uses tdb))
(declare (uses http-transport))
;;(declare (uses nmsg-transport))
|
|
>
|
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
(use json format) ;; RADT => purpose of json format??
(declare (unit rmt))
(declare (uses api))
(declare (uses tdb))
(declare (uses http-transport))
(declare (uses rpc-transport))
;;
;; THESE ARE ALL CALLED ON THE CLIENT SIDE!!!
;;
;; ;; For debugging add the following to ~/.megatestrc
;;
;; (require-library trace)
|
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
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
|
(if cinfo
cinfo
;; NB// can cache the answer for server running for 10 seconds ...
;; ;; (and (not (rmt:write-frequency-over-limit? cmd run-id))
(if (tasks:server-running-or-starting? (db:delay-if-busy (tasks:open-db)) run-id)
(client:setup run-id)
#f))))
(define *send-receive-mutex* (make-mutex)) ;; should have separate mutex per run-id
;; RA => e.g. usage (rmt:send-receive 'get-var #f (list varname))
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)) ;; start attemptnum at 1 so the modulo below works as expected
(let* ((transport-type (vector-ref connection-info 6)) ;; BB: assumes all transport-type'-servertdat vector's item 6 ids transport type
(dat (case transport-type ;; BB: replaced *transport-type* global with run-id specific transport-type, item 6 in server-info vector which was populated by *-transport:client-connect with >> (vector iface port api-uri api-url api-req (current-seconds) 'http ) <<
((http)(condition-case
(http-transport:client-api-send-receive run-id connection-info cmd params)
((commfail)(vector #f "communications fail"))
((exn)(vector #f "other fail"))))
;;((rpc) (rpc-transport:client-api-send-receive run-id connection-info cmd params)) ;; BB: let us error out for now
(else
(debug:print-error 0 *default-log-port* "Transport ["
transport "] specified for run-id [" run-id "] is not implemented in rmt:send-receive. Cannot proceed.")
(exit 1))))
(case *transport-type*
((http rpc) res) ;; (db:string->obj res))
;; ((nmsg) res)
)) ;; (vector-ref res 1)))
(begin ;; let ((new-connection-info (client:setup run-id)))
(debug:print 0 *default-log-port* "WARNING: Communication failed, trying call to rmt:send-receive again.")
(case *transport-type*
((http)
(hash-table-delete! *runremote* run-id) ;; don't keep using the same connection
;; NOTE: killing server causes this process to block forever. No idea why. Dec 2.
;; (if (eq? (modulo attemptnum 5) 0)
;; (tasks:kill-server-run-id run-id tag: "api-send-receive-failed"))
;; (mutex-unlock! *send-receive-mutex*) ;; close the mutex here to allow other threads access to communications
(tasks:start-and-wait-for-server (tasks:open-db) run-id 15)
;; (nmsg-transport:client-api-send-receive run-id connection-info cmd param remtries: (- remtries 1))))))
;; no longer killing the server in http-transport:client-api-send-receive
;; may kill it here but what are the criteria?
;; start with three calls then kill server
;; (if (eq? attemptnum 3)(tasks:kill-server-run-id run-id))
;; (thread-sleep! 2)
(rmt:send-receive cmd run-id params attemptnum: (+ attemptnum 1)))))))
;; no connection info? try to start a server, or access locally if no
;; server and the query is read-only
;;
;; Note: The tasks db was checked for a server in starting mode in the rmt:get-connection-info call
;;
(if (and (< attemptnum 15)
(member cmd api:write-queries))
(let ((faststart (configf:lookup *configdat* "server" "faststart")))
(hash-table-delete! *runremote* run-id)
;; (mutex-unlock! *send-receive-mutex*)
(if (and faststart (equal? faststart "no"))
(begin
(tasks:start-and-wait-for-server (db:delay-if-busy (tasks:open-db)) run-id 10)
(thread-sleep! (random 5)) ;; give some time to settle and minimize collison?
(rmt:send-receive cmd rid params attemptnum: (+ attemptnum 1)))
|
>
>
>
>
>
>
>
>
>
>
>
>
<
<
>
|
>
>
>
|
|
>
|
>
|
>
|
>
>
>
>
>
>
>
|
|
>
>
>
>
>
>
|
|
|
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
(if cinfo
cinfo
;; NB// can cache the answer for server running for 10 seconds ...
;; ;; (and (not (rmt:write-frequency-over-limit? cmd run-id))
(if (tasks:server-running-or-starting? (db:delay-if-busy (tasks:open-db)) run-id)
(client:setup run-id)
#f))))
(define (rmt:run-id->transport-type rid)
(let* ((run-id (if rid rid 0))
(connection-info (hash-table-ref/default *runremote* run-id #f)))
;; the nmsg method does the encoding under the hood (the http method should be changed to do this also)
(if connection-info ;; if we already have a connection for this run-id, use that precendent
;; use the server if have connection info
(let* ((transport-type (vector-ref connection-info 6))) ;; BB: assumes all transport-type'-servertdat vector's item 6 ids transport type
transport-type)
;; otherwise pick the global default as preference.
*transport-type*)))
(define *send-receive-mutex* (make-mutex)) ;; should have separate mutex per run-id
;; RA => e.g. usage (rmt:send-receive 'get-var #f (list varname))
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)) ;; start attemptnum at 1 so the modulo below works as expected
;; side-effect: clean out old connections
;; (mutex-lock! *db-multi-sync-mutex*)
(let ((expire-time (- (current-seconds) (server:get-timeout) 10))) ;; don't forget the 10 second margin
(for-each
(lambda (run-id)
(let ((connection (hash-table-ref/default *runremote* run-id #f)))
(if (and (vector? connection)
(< (http-transport:server-dat-get-last-access connection) expire-time)) ;; BB> BBTODO: make this generic, not http transport specific.
(begin
(debug:print-info 0 *default-log-port* "Discarding connection to server for run-id " run-id ", too long between accesses")
(hash-table-delete! *runremote* run-id)))))
(hash-table-keys *runremote*)))
;; (mutex-unlock! *db-multi-sync-mutex*)
;; (mutex-lock! *send-receive-mutex*)
(let* ((run-id (if rid rid 0))
(connection-info (rmt:get-connection-info run-id)))
;; the nmsg method does the encoding under the hood (the http method should be changed to do this also)
(if connection-info
;; use the server if have connection info
(let* ((transport-type (rmt:run-id->transport-type run-id))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Here, we make request to remote server
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(dat (case transport-type ;; BB: replaced *transport-type* global with run-id specific transport-type
((http)(condition-case
(http-transport:client-api-send-receive run-id connection-info cmd params)
((commfail)(vector #f "communications fail"))
((exn)(vector #f "other fail"))))
;;((rpc) (rpc-transport:client-api-send-receive run-id connection-info cmd params)) ;; BB: let us error out for now
(else
(debug:print-error 0 *default-log-port* "(1) Transport [" transport-type
"] specified for run-id [" run-id
"] is not implemented in rmt:send-receive. Cannot proceed.")
(vector #f (conc "transport ["transport-type"] unimplemented")))))
(success (if (vector? dat) (vector-ref dat 0) #f))
(res (if (vector? dat) (vector-ref dat 1) #f)))
(if (vector? connection-info)(http-transport:server-dat-update-last-access connection-info)) ;; BB> BBTODO: make this generic, not http transport specific.
(if success
(begin
;; (mutex-unlock! *send-receive-mutex*)
(case transport-type
((http rpc) res) ;; (db:string->obj res))
(else
(debug:print-error 0 *default-log-port* "(2) Transport [" transport-type
"] specified for run-id [" run-id
"] is not implemented in rmt:send-receive. Cannot proceed. Also unexpected since this branch follows success which would follow a suported transport...")
#f)
;; ((nmsg) res)
)) ;; (vector-ref res 1)))
(begin ;; let ((new-connection-info (client:setup run-id)))
(debug:print 0 *default-log-port* "WARNING: Communication failed, trying call to rmt:send-receive again.")
(case transport-type
((http)
(hash-table-delete! *runremote* run-id) ;; don't keep using the same connection
;; NOTE: killing server causes this process to block forever. No idea why. Dec 2.
;; (if (eq? (modulo attemptnum 5) 0)
;; (tasks:kill-server-run-id run-id tag: "api-send-receive-failed"))
;; (mutex-unlock! *send-receive-mutex*) ;; close the mutex here to allow other threads access to communications
(tasks:start-and-wait-for-server (tasks:open-db) run-id 15)
;; (nmsg-transport:client-api-send-receive run-id connection-info cmd param remtries: (- remtries 1))))))
;; no longer killing the server in http-transport:client-api-send-receive
;; may kill it here but what are the criteria?
;; start with three calls then kill server
;; (if (eq? attemptnum 3)(tasks:kill-server-run-id run-id))
;; (thread-sleep! 2)
(rmt:send-receive cmd run-id params attemptnum: (+ attemptnum 1)))
(else
(debug:print-error 0 *default-log-port* "(3) Transport [" transport-type
"] specified for run-id [" run-id
"] is not implemented in rmt:send-receive. Cannot proceed.")
#f)))))
;; no connection info; try to start a server, or access locally if no
;; server and the query is read-only
;;
;; Note: The tasks db was checked for a server in starting mode in the rmt:get-connection-info call
;;
(if (and (< attemptnum 15)
(member cmd api:write-queries))
(let* ((faststart (configf:lookup *configdat* "server" "faststart")))
(hash-table-delete! *runremote* run-id)
;; (mutex-unlock! *send-receive-mutex*)
(if (and faststart (equal? faststart "no"))
(begin
(tasks:start-and-wait-for-server (db:delay-if-busy (tasks:open-db)) run-id 10)
(thread-sleep! (random 5)) ;; give some time to settle and minimize collison?
(rmt:send-receive cmd rid params attemptnum: (+ attemptnum 1)))
|
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
|
(define (rmt:login run-id)
(rmt:send-receive 'login run-id (list *toppath* megatest-version run-id *my-client-signature*)))
;; This login does no retries under the hood - it acts a bit like a ping.
;; Deprecated for nmsg-transport.
;;
(define (rmt:login-no-auto-client-setup connection-info run-id)
(case *transport-type*
((http)(rmt:send-receive-no-auto-client-setup connection-info 'login run-id (list *toppath* megatest-version run-id *my-client-signature*)))
;;((nmsg)(nmsg-transport:client-api-send-receive run-id connection-info 'login (list *toppath* megatest-version run-id *my-client-signature*)))
))
;; hand off a call to one of the db:queries statements
;; added run-id to make looking up the correct db possible
;;
(define (rmt:general-call stmtname run-id . params)
(rmt:send-receive 'general-call run-id (append (list stmtname run-id) params)))
|
|
|
>
>
>
|
|
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
|
(define (rmt:login run-id)
(rmt:send-receive 'login run-id (list *toppath* megatest-version run-id *my-client-signature*)))
;; This login does no retries under the hood - it acts a bit like a ping.
;; Deprecated for nmsg-transport.
;;
(define (rmt:login-no-auto-client-setup connection-info run-id)
(case (rmt:run-id->transport-type run-id)
((http)(rmt:send-receive-no-auto-client-setup connection-info 'login run-id (list *toppath* megatest-version run-id *my-client-signature*)))
(else
(debug:print-error 0 *default-log-port* "(4) Transport [" *transport-type*
"] specified for run-id [" run-id
"] is not implemented in rmt:send-receive. Cannot proceed.")
(exit 1))))
;; hand off a call to one of the db:queries statements
;; added run-id to make looking up the correct db possible
;;
(define (rmt:general-call stmtname run-id . params)
(rmt:send-receive 'general-call run-id (append (list stmtname run-id) params)))
|