82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
(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")
;; bb- disabling nanomsg
;; SHOULD CLOSE THE CONNECTION HERE
;; (case *transport-type*
;; ((nmsg)(nn-close (http-transport:server-dat-get-socket
;; (hash-table-ref *runremote* run-id)))))
|
<
<
<
<
<
|
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
(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)
|
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
|
;;((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))))
(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))
(if success
(begin
;; (mutex-unlock! *send-receive-mutex*)
(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*
;; ((nmsg)(nn-close (http-transport:server-dat-get-socket connection-info))))
|
|
<
|
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
|
;;((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))))
(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))
;; ((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)
|