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
|
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
|
-
-
+
+
-
+
-
-
+
-
+
-
-
-
-
+
+
+
+
+
+
+
+
+
|
(debug:print-info 1 "db write rate too high, starting a server, count=" count " start=" start " run-id=" run-id " queries-per-second=" queries-per-second)
#t)
#f))))
;; if a server is either running or in the process of starting call client:setup
;; else return #f to let the calling proc know that there is no server available
;;
(define (rmt:get-connection-info run-id)
(let ((cinfo (hash-table-ref/default *runremote* run-id #f)))
(define (rmt:get-connection-info run-id #!key (remote #f))
(let ((cinfo (common:get-remote remote run-id)))
(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)
(client:setup run-id remote: remote)
#f))))
(define *send-receive-mutex* (make-mutex)) ;; should have separate mutex per run-id
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)) ;; start attemptnum at 1 so the modulo below works as expected
(let ((connection (hash-table-ref/default *runremote* run-id #f)))
(let ((connection (common:get-remote remote run-id)))
(if (and (vector? connection)
(< (http-transport:server-dat-get-last-access connection) expire-time))
(begin
(debug:print-info 0 "Discarding connection to server for run-id " run-id ", too long between accesses")
;; SHOULD CLOSE THE CONNECTION HERE
(case *transport-type*
((nmsg)(nn-close (http-transport:server-dat-get-socket
(hash-table-ref *runremote* run-id)))))
(hash-table-delete! *runremote* run-id)))))
(hash-table-keys *runremote*)))
(mutex-unlock! *db-multi-sync-mutex*)
(common:get-remote remote run-id)))))
(common:del-remote! remote run-id)))))
(common:get-remote-all remote)))
(mutex-unlock! *db-multi-sync-mutex*))
(define *send-receive-mutex* (make-mutex)) ;; should have separate mutex per run-id
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(remote #f)) ;; start attemptnum at 1 so the modulo below works as expected
(rmt:discard-old-connections)
;; (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* ((dat (case *transport-type*
|
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
-
+
|
(case *transport-type*
((http) res) ;; (db:string->obj res))
((nmsg) res))) ;; (vector-ref res 1)))
(begin ;; let ((new-connection-info (client:setup run-id)))
(debug:print 0 "WARNING: Communication failed, trying call to rmt:send-receive again.")
;; (case *transport-type*
;; ((nmsg)(nn-close (http-transport:server-dat-get-socket connection-info))))
(hash-table-delete! *runremote* run-id) ;; don't keep using the same connection
(common:del-remote! remote 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))))))
|
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
-
+
|
;; 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)
(common:del-remote! remote 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)))
(begin
|