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
|
;; (debug:print-info 0 "Queue not flushed, waiting ...")
;; (loop))))))))
;; The heavy lifting
;;
;; make-vector-record cdb packet client-sig qtype immediate query-sig params qtime
;;
(let loop ((queue-lst '()))
(let* ((rawmsg (receive-message* pull-socket))
(packet (db:string->obj rawmsg))
(qtype (cdb:packet-get-qtype packet)))
(debug:print-info 12 "server=> received packet=" packet)
(if (not (member qtype '(sync ping)))
(begin
(mutex-lock! *heartbeat-mutex*)
(set! *last-db-access* (current-seconds))
(mutex-unlock! *heartbeat-mutex*)))
(if #t ;; (cdb:packet-get-immediate packet) ;; process immediately or put in queue
(begin
(open-run-close db:process-queue #f pub-socket (cons packet queue-lst))
(loop '()))
(loop (cons packet queue-lst)))))))
;; run zmq-transport:keep-running in a parallel thread to monitor that the db is being
;; used and to shutdown after sometime if it is not.
;;
(define (zmq-transport:keep-running)
;; if none running or if > 20 seconds since
;; server last used then start shutdown
;; This thread waits for the server to come alive
(let* ((server-info (let loop ()
(let ((sdat #f))
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if sdat sdat
(begin
(sleep 4)
(loop))))))
(iface (cadr server-info))
(pullport (caddr server-info))
(pubport (cadddr server-info)) ;; id interface pullport pubport)
(zmq-sockets (zmq-transport:client-connect iface pullport pubport))
(last-access 0))
(debug:print-info 11 "heartbeat started for zmq server on " iface " " pullport " " pubport)
(let loop ((count 0))
(thread-sleep! 4) ;; no need to do this very often
;; NB// sync currently does NOT return queue-length
(let ((queue-len (cdb:client-call zmq-sockets 'sync #t 1)))
;; (print "Server running, count is " count)
(if (< count 1) ;; 3x3 = 9 secs aprox
(loop (+ count 1)))
;; NOTE: Get rid of this mechanism! It really is not needed...
(open-run-close tasks:server-update-heartbeat tasks:open-db (car server-info))
|
>
|
|
|
|
|
|
|
>
|
>
|
|
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
|
;; (debug:print-info 0 "Queue not flushed, waiting ...")
;; (loop))))))))
;; The heavy lifting
;;
;; make-vector-record cdb packet client-sig qtype immediate query-sig params qtime
;;
(debug:print-info 11 "Server setup complete, start listening for messages")
(let loop ((queue-lst '()))
(let* ((rawmsg (receive-message* pull-socket))
(packet (db:string->obj rawmsg))
(qtype (cdb:packet-get-qtype packet)))
(debug:print-info 12 "server=> received packet=" packet)
(if (not (member qtype '(sync ping)))
(begin
(mutex-lock! *heartbeat-mutex*)
(set! *last-db-access* (current-seconds))
(mutex-unlock! *heartbeat-mutex*)))
(if #t ;; (cdb:packet-get-immediate packet) ;; process immediately or put in queue
(begin
(open-run-close db:process-queue #f pub-socket (cons packet queue-lst))
(loop '()))
(loop (cons packet queue-lst)))))))
;; run zmq-transport:keep-running in a parallel thread to monitor that the db is being
;; used and to shutdown after sometime if it is not.
;;
(define (zmq-transport:keep-running)
;; if none running or if > 20 seconds since
;; server last used then start shutdown
;; This thread waits for the server to come alive
(let* ((server-info (let loop ()
(let ((sdat #f))
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if sdat sdat
(begin
(debug:print 12 "WARNING: server not started yet, waiting few seconds before trying again")
(sleep 4)
(loop))))))
(iface (cadr server-info))
(pullport (caddr server-info))
(pubport (cadddr server-info)) ;; id interface pullport pubport)
;; (zmq-sockets (zmq-transport:client-connect iface pullport pubport))
(last-access 0))
(debug:print-info 11 "heartbeat started for zmq server on " iface " " pullport " " pubport)
(let loop ((count 0))
(thread-sleep! 4) ;; no need to do this very often
;; NB// sync currently does NOT return queue-length
;; GET REAL QUEUE LENGTH FROM THE VARIABLE
(let ((queue-len 0)) ;; FOR NOW DO NOT DO THIS (cdb:client-call zmq-sockets 'sync #t 1)))
;; (print "Server running, count is " count)
(if (< count 1) ;; 3x3 = 9 secs aprox
(loop (+ count 1)))
;; NOTE: Get rid of this mechanism! It really is not needed...
(open-run-close tasks:server-update-heartbeat tasks:open-db (car server-info))
|
231
232
233
234
235
236
237
238
239
240
241
242
243
244
|
tasks:open-db
(current-process-id)
ipaddrstr p1
0
'live
'zmq
pubport: p2))
(mutex-unlock! *heartbeat-mutex*)
(list s1 s2)))
(define (zmq-transport:mk-signature)
(message-digest-string (md5-primitive)
(with-output-to-string
(lambda ()
|
>
|
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
tasks:open-db
(current-process-id)
ipaddrstr p1
0
'live
'zmq
pubport: p2))
(debug:print-info 11 "*server-info* set to " *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(list s1 s2)))
(define (zmq-transport:mk-signature)
(message-digest-string (md5-primitive)
(with-output-to-string
(lambda ()
|