Megatest

Diff
Login

Differences From Artifact [a650f389d3]:

To Artifact [67b3f4e15b]:


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

186
187
188
189
190
191
192
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
186
187
188
189







-
+
+














-
-
-
-



















+
+
-
+


-
+
-
-
+
-
-
-
+
-
-
-
-
-

-
-
+
-
-
-
+
+

+
+
+








-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+


-
+



-
+







	 (iface           (if (string=? "-" hostn)
			      "*" ;; (get-host-name) 
			      hostn))
	 (hostname        (get-host-name))
	 (ipaddrstr       (let ((ipstr (if (string=? "-" hostn)
					   (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
					   #f)))
			    (if ipstr ipstr hostname))))
			    (if ipstr ipstr hostname)))
	 (last-run       0))
    (set! zmq-sockets-dat (server:setup-ports ipaddrstr (if (args:get-arg "-port")
							    (string->number (args:get-arg "-port"))
							    (+ 5000 (random 1001)))))

    (set! zmq-sdat1    (car   zmq-sockets-dat))
    (set! pull-socket  (cadr  zmq-sdat1)) ;; (iface s  port)
    (set! p1           (caddr zmq-sdat1))
    
    (set! zmq-sdat2    (cadr  zmq-sockets-dat))
    (set! pub-socket   (cadr  zmq-sdat2))
    (set! p2           (caddr zmq-sdat2))

    (set! *cache-on* #t)

    ;; (set! th1 (make-thread (lambda ()
    ;;     		     (server:self-ping ipaddrstr actual-port))))
    ;; (thread-start! th1)
    
    ;; what to do when we quit
    ;;
    (on-exit (lambda ()
	       (if (and *toppath* *server-info*)
		   (begin
		     (open-run-close tasks:server-deregister-self tasks:open-db ipaddrstr p1 p2))
		   (let loop () 
		     (let ((queue-len 0))
		       (thread-sleep! (random 5))
		       (mutex-lock! *incoming-mutex*)
		       (set! queue-len (length *incoming-data*))
		       (mutex-unlock! *incoming-mutex*)
		       (if (> queue-len 0)
			   (begin
			     (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 ()
    (let loop ((queue-lst '()))
      ;; (print "GOT HERE EH?")
      (let* ((rawmsg (receive-message* pull-socket))
	     (params (db:string->obj rawmsg)) ;; (with-input-from-string rawmsg (lambda ()(deserialize))))
	     (packet (db:string->obj rawmsg)))
	     (res    #f))
	(debug:print-info 12 "server=> received params=" params)
	(debug:print-info 12 "server=> received packet=" packet)
	(set! res (cdb:cached-access params))
	(debug:print-info 12 "server=> processed res=" res)

	(if (cdb:packet-get-immediate packet) ;; process immediately or put in queue
	;; need address here
	;;
	;; (send-message zmq-socket (db:obj->string res))
	(if (not *time-to-exit*)
	    (loop)
	    (begin
	      (open-run-close tasks:server-deregister-self tasks:open-db #f)
	      (db:write-cached-data)
	      (db:process-queue pubsock (cons packet queue))
	      (exit)
	      ))))
    (thread-join! th1)))
	      (loop '()))
	    (loop (cons packet queue)))))))

(define (server:reply pubsock target result)
  (send-message pubsock target send-more: #t)
  (send-message pubsock result))

;; run server:keep-running in a parallel thread to monitor that the db is being 
;; used and to shutdown after sometime if it is not.
;;
(define (server: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)))))))
  (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 (server:client-connect iface pullport pubport)))
	 )
    (let loop ((count 0))
      (thread-sleep! 4) ;; no need to do this very often
      (db:write-cached-data)
      ;;  (let ((queue-len (string->number (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))
      
      ;; (if ;; (or (> numrunning 0) ;; stay alive for two days after last access
      (if (> (+ *last-db-access* 
		;; (* 48 60 60)    ;; 48 hrs
		;; 60              ;; one minute
250
251
252
253
254
255
256
257

258
259
260
261
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
289
290
291
292
293
294


295
296
297
298
299
300
301
247
248
249
250
251
252
253

254
255
256
257
258
259
260
261
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
289
290
291
292
293


294
295
296
297
298
299
300
301
302







-
+









+
-
+



+
+
-
+











-
-
+
+
+








-
-
+
+







  (if *my-client-signature* *my-client-signature*
      (let ((sig (server:mk-signature)))
	(set! *my-client-signature* sig)
	*my-client-signature*)))

;; 
(define (server:client-socket-connect iface port #!key (context #f)(type 'req)(subscriptions '()))
  (debug:print-info 3 "client-connect " iface ":" port)
  (debug:print-info 3 "client-connect " iface ":" port ", type=" type ", subscriptions=" subscriptions)
  (let ((connect-ok #f)
	(zmq-socket (if context 
			(make-socket type context)
			(make-socket type)))
	(conurl     (server:make-server-url (list iface port))))
    (if (socket? zmq-socket)
	(begin
	  ;; first apply subscriptions
	  (for-each (lambda (subscription)
		      (debug:print 2 "Subscribing to " subscription)
		      (socket-options-set! zmq-socket 'subscribe subscription))
		      (socket-option-set! zmq-socket 'subscribe subscription))
		    subscriptions)
	  (connect-socket zmq-socket conurl)
	  zmq-socket)
	(begin
	  (debug:print 0 "ERROR: Failed to open socket to " conurl)
	#f)))
	  #f))))
  
(define (server:client-login zmq-sockets)
  (cdb:login zmq-sockets *toppath* (server:get-client-signature)))

(define (server:client-logout zmq-socket)
  (let ((ok (and (socket? zmq-socket)
		 (cdb:logout zmq-socket *toppath* (server:get-client-signature)))))
    ;; (close-socket zmq-socket)
    ok))

(define (server:client-connect iface pullport pubport)
  (let* ((push-socket (server:client-socket-connect iface pullport 'push))
	 (sub-socket  (server:client-socket-connect iface pubport 'sub
  (let* ((push-socket (server:client-socket-connect iface pullport type: 'push))
	 (sub-socket  (server:client-socket-connect iface pubport
						    type: 'sub
						    subscriptions: (list (server:get-client-signature) "all")))
	 (zmq-sockets (vector push-socket sub-socket))
	 (login-res   #f))
    (set! login-res (server:client-login zmq-sockets))
    (if (and (not (null? login-res))
	     (car login-res))
	(begin
	  (debug:print-info 2 "Logged in and connected to " iface ":" pullport "/" pubport ".")
	  (set! *runremote* zmq-socket)
	  #t)
	  (set! *runremote* zmq-sockets)
	  zmq-sockets)
	(begin
	  (debug:print-info 2 "Failed to login or connect to " conurl)
	  (set! *runremote* #f)
	  #f))))

;; Do all the connection work, start a server if not already running
(define (server:client-setup #!key (numtries 50))
357
358
359
360
361
362
363
364

365

366
367
368
369
370
371



372
373
374
375
376
377
378
358
359
360
361
362
363
364

365

366
367
368
369
370
371

372
373
374
375
376
377
378
379
380
381







-
+
-
+





-
+
+
+







		   ;;      		   (if (not server-info)(loop)))
		   ;;      		 (debug:print 1 "Server alive, starting self-ping")
		   ;;      		 (server:self-ping server-info)
		   ;;      		 ))
		   ;;      	     "Self ping"))
		   (th2 (make-thread (lambda ()
				       (server:run (args:get-arg "-server"))) "Server run"))
		   (th3 (make-thread (lambda ()
		   (th3 (make-thread (lambda ()(server:keep-running)) "Keep running"))
				       (server:keep-running)) "Keep running")))
		   )
	      (set! *client-non-blocking-mode* #t)
	      ;; (thread-start! th1)
	      (thread-start! th2)
	      (thread-start! th3)
	      (set! *didsomething* #t)
	      (thread-join! th3))
	      ;; (thread-join! th3)
	      (thread-join! th2)
	      )
	    (debug:print 0 "ERROR: Failed to setup for megatest")))
    (exit)))

(define (server:client-signal-handler signum)
  (handle-exceptions
   exn
   (debug:print " ... exiting ...")