Megatest

Check-in [d3db3e5ba1]
Login
☰Home Timeline Files Branches Tags Tickets Wiki Login
Overview
Comment:incrementl changes for http transport
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | http-transport
Files: files | file ages | folders
SHA1: d3db3e5ba1ce8dc595acf6a8b532960aca4bd4da
User & Date: mrwellan on 2013-01-15 17:40:58
Other Links: branch diff | manifest | tags
Context
2013-01-15
22:47
Basic server/client working check-in: ed470f76ce user: matt tags: http-transport
17:40
incrementl changes for http transport check-in: d3db3e5ba1 user: mrwellan tags: http-transport
2013-01-14
23:03
second pass http, queue processing remaining check-in: e6b987028d user: matt tags: http-transport
Changes
Hide Diffs Side-by-Side Diffs Ignore Whitespace Patch

Modified db.scm from [644346c603] to [a287f6a261].

︙
12
13
14
15
16
17
18
19

20

21
22
23
24
25
26
27

12
13
14
15
16
17
18

19
20
21
22
23
24
25
26
27
28








-
+

+








;;======================================================================
;; Database access
;;======================================================================

(require-extension (srfi 18) extras tcp) ;;  rpc)
;; (import (prefix rpc rpc:))

(use sqlite3 srfi-1 posix regex regex-case srfi-69 csv-xml s11n md5 message-digest)
(use sqlite3 srfi-1 posix regex regex-case srfi-69 csv-xml s11n md5 message-digest base64)
(import (prefix sqlite3 sqlite3:))
(import (prefix base64 base64:))

(declare (unit db))
(declare (uses common))
(declare (uses keys))
(declare (uses ods))

(include "common_records.scm")
︙
1094
1095
1096
1097
1098
1099
1100
1101
1102












1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120





1121
1122
1123



1124
1125
1126
1127



1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142













1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157

















1158
1159
1160
1161
1162
1163
1164

1095
1096
1097
1098
1099
1100
1101


1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126





1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138



1139
1140
1141















1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154















1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178








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













-
-
-
-
-
+
+
+
+
+



+
+
+

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








;; (define (db:updater)
;;   (debug:print-info 4 "Starting cache processing")
;;   (let loop ()
;;     (thread-sleep! 10) ;; move save time around to minimize regular collisions?
;;     (db:write-cached-data)
;;     (loop)))

(define (db:obj->string obj)(with-output-to-string (lambda ()(serialize obj))))
(define (db:string->obj msg)(with-input-from-string msg (lambda ()(deserialize))))
(define (db:obj->string obj)
  (string-substitute
   (regexp "=") "_"
   (base64:base64-encode (with-output-to-string (lambda ()(serialize obj))))
   #t))

(define (db:string->obj msg)
  (with-input-from-string 
      (base64:base64-decode
       (string-substitute 
	(regexp "_") "=" msg #t))
    (lambda ()(deserialize))))

(define (cdb:use-non-blocking-mode proc)
  (set! *client-non-blocking-mode* #t)
  (let ((res (proc)))
    (set! *client-non-blocking-mode* #f)
    res))
  
;; params = 'target cached remparams
;;
;; make-vector-record cdb packet client-sig qtype immediate query-sig params qtime
;;
(define (cdb:client-call serverdat qtype immediate numretries . params)
  (debug:print-info 11 "cdb:client-call serverdat=" serverdat ", qtype=" qtype ", immediate=" immediate ", numretries=" numretries ", params=" params)
  (handle-exceptions
   exn
   (begin
     (thread-sleep! 5) 
     (if (> numretries 0)(apply cdb:client-call serverdat qtype immediate (- numretries 1) params)))
  ;; (handle-exceptions
  ;;  exn
  ;;  (begin
  ;;    (thread-sleep! 5) 
  ;;    (if (> numretries 0)(apply cdb:client-call serverdat qtype immediate (- numretries 1) params)))
   (let* ((client-sig  (server:get-client-signature))
	  (query-sig   (message-digest-string (md5-primitive) (conc qtype immediate params)))
	  (zdat        (db:obj->string (vector client-sig qtype immediate query-sig params (current-seconds)))) ;; (with-output-to-string (lambda ()(serialize params))))
	  )
     ;; (print "zdat=" zdat)
     (let* (
	  (res  #f)
	  (send-receive (lambda ()
			  (let loop ((res (server:client-send-receive serverdat zdat)))
			    ;; get the sender info
	  (rawdat      (server:client-send-receive serverdat zdat))
	  (tmp         #f))
     (print "Sent " zdat ", received " rawdat)
			    ;; this should match (server:get-client-signature)
			    ;; we will need to process "all" messages here some day
			    ;; now get the actual message
			    (let ((myres (db:string->obj res)))
			      (if (equal? query-sig (vector-ref myres 1))
				  (set! res (vector-ref myres 2))
				  (loop (server:client-send-receive serverdat zdat)))))))
	  (timeout (lambda ()
		     (let loop ((n numretries))
		       (thread-sleep! 15)
		       (if (not res)
			   (if (> numretries 0)
			       (begin
				 (debug:print 2 "WARNING: no reply to query " params ", trying resend")
				 (debug:print-info 11 "re-sending message")
     (set! tmp (db:string->obj newres))
     ;; (if (equal? query-sig (vector-ref myres 1))
     ;; (set! res
     (vector-ref myres 2)
     ;; (loop (server:client-send-receive serverdat zdat)))))))
	  ;; (timeout (lambda ()
	  ;;            (let loop ((n numretries))
	  ;;              (thread-sleep! 15)
	  ;;              (if (not res)
	  ;;       	   (if (> numretries 0)
	  ;;       	       (begin
	  ;;       		 (debug:print 2 "WARNING: no reply to query " params ", trying resend")
	  ;;       		 (debug:print-info 11 "re-sending message")
				 (send-message push-socket zdat)
				 (debug:print-info 11 "message re-sent")
				 (loop (- n 1)))
			       ;; (apply cdb:client-call serverdats qtype immediate (- numretries 1) params))
			       (begin
				 (debug:print 0 "ERROR: cdb:client-call timed out " params ", exiting.")
				 (exit 5))))))))
     (debug:print-info 11 "Starting threads")
     (let ((th1 (make-thread send-receive "send receive"))
	   (th2 (make-thread timeout      "timeout")))
       (thread-start! th1)
       (thread-start! th2)
       (thread-join!  th1)
       (debug:print-info 11 "cdb:client-call returning res=" res)
       res))))
	  ;;       		 (apply cdb:client-call serverdat qtype immediate numretries params)
	  ;;       		 (debug:print-info 11 "message re-sent")
	  ;;       		 (loop (- n 1)))
	  ;;       	       ;; (apply cdb:client-call serverdats qtype immediate (- numretries 1) params))
	  ;;       	       (begin
	  ;;       		 (debug:print 0 "ERROR: cdb:client-call timed out " params ", exiting.")
	  ;;       		 (exit 5))))))))
     ;; (send-receive)
     )))
     ;; (debug:print-info 11 "Starting threads")
     ;; (let ((th1 (make-thread send-receive "send receive"))
     ;;       (th2 (make-thread timeout      "timeout")))
     ;;   (thread-start! th1)
     ;;   (thread-start! th2)
     ;;   (thread-join!  th1)
     ;;   (debug:print-info 11 "cdb:client-call returning res=" res)
     ;;   res))))
  
(define (cdb:set-verbosity serverdat val)
  (cdb:client-call serverdat 'set-verbosity #f *default-numtries* val))

(define (cdb:login serverdat keyval signature)
  (cdb:client-call serverdat 'login #t *default-numtries* keyval megatest-version signature))
︙
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285

1261
1262
1263
1264
1265
1266
1267

























1268
1269
1270
1271
1272
1273
1274








-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-








			       flush
			       sync
			       set-verbosity
			       killserver))

;; not used, intended to indicate to run in calling process
(define db:run-local-queries '()) ;; rollup-tests-pass-fail))








UPDATE DB:PROCESS_QUEUE@@@@

















;; The queue is a list of vectors where the zeroth slot indicates the type of query to
;; apply and the second slot is the time of the query and the third entry is a list of 
;; values to be applied
;;
(define (db:process-queue db pubsock indata)
  (let* ((data       (sort indata (lambda (a b)
︙

Modified server.scm from [a4326a1d4d] to [ccebef113b].

︙
44
45
46
47
48
49
50
51
52
53



54
55
56
57
58

59
60
61
62
63
64
65
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

44
45
46
47
48
49
50



51
52
53
54
55
56
57

58
59
60
61
62
63
64
65
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








-
-
-
+
+
+




-
+








+
+
+



+








-
+












-
+
+


-
-

+




+
-
+
+












+
+
+
+
+
















-
-
-
+
+
+
+
+
+
+








(define (server:run hostn)
  (debug:print 2 "Attempting to start the server ...")
  (if (not *toppath*)
      (if (not (setup-for-run))
	  (begin
	    (debug:print 0 "ERROR: cannot find megatest.config, cannot start server, exiting")
	    (exit))))
  (let* ((iface           (if (string=? "-" hostn)
			      "*" ;; (get-host-name) 
			      hostn))
  (let* (;; (iface           (if (string=? "-" hostn)
	 ;;        	      #f ;; (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 hostn))) ;; hostname)))
	 (start-port    (if (args:get-arg "-port")
			    (string->number (args:get-arg "-port"))
			    (+ 5000 (random 1001)))))
    (set! *cache-on* #t)
    (server:try-start-server ipaddrstr start-port)))


(define (server:main-loop)
  (print "INFO: Exectuing main server loop")
  (access-log "megatest-http.log")
  (server-bind-address #f)
  (define-page (main-page-path)
    (lambda ()
      (with-request-variables (dat)
        (print "Got dat=" dat)
	(let* ((packet (db:string->obj dat))
	       (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*)))
	  (open-run-close db:process-queue #f pub-socket (cons packet queue-lst)))))))
	  (open-run-close db:process-queue-item packet))))))


;; This is recursively run by server:run until sucessful
;;
(define (server:try-start-server ipaddrstr portnum)
  (handle-exceptions
   exn
   (begin
     (print-error-message exn)
     (if (< portnum 9000)
	 (begin 
	   (print "WARNING: failed to start on portnum: " portnum ", trying next port")
	   (sleep 1)
	   (thread-sleep! 0.1)
	   (open-run-close tasks:remove-server-records tasks:open-db)
	   (server:try-start-server ipaddrstr (+ portnum 1)))
	 (print "ERROR: Tried and tried but could not start the server")))
   (print "INFO: Trying to start server on portnum: " portnum)
   
   (set! *runremote* (list ipaddrstr portnum))
   (open-run-close tasks:remove-server-records tasks:open-db)
   (open-run-close tasks:server-register 
		   tasks:open-db 
		   (current-process-id)
		   ipaddrstr portnum 0 'live)
   (print "INFO: Trying to start server on " ipaddrstr ":" portnum)
   (awful-start server:main-loop ip-address: ipaddrstr port: portnum)))
   (awful-start server:main-loop port: portnum) ;; ip-address: ipaddrstr 
   (print "INFO: server has been stopped")))

(define (server:mk-signature)
  (message-digest-string (md5-primitive) 
			 (with-output-to-string
			   (lambda ()
			     (write (list (current-directory)
					  (argv)))))))

;;======================================================================
;; S E R V E R   U T I L I T I E S 
;;======================================================================

(define (server:reply pubsock target query-sig success/fail result)
  (debug:print-info 11 "server:reply target=" target ", result=" result)
  ;; (send-message pubsock target send-more: #t)
  ;; (send-message pubsock 
  (db:obj->string (vector success/fail query-sig result)))

;;======================================================================
;; C L I E N  T S
;;======================================================================

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

;; <html>
;; <head></head>
;; <body>1 Hello, world! Goodbye Dolly</body></html>
;; Send msg to serverdat and receive result
(define (server:client-send-receive serverdat msg)
  (let* ((res (with-input-from-request (conc (server:make-server-url serverdat) "/?dat=" msg) #f read-string))
	 (match (string-search (regexp "<body>(.*)<.body>") (caddr (string-split res "\n")))))
    (cadr match)))
  (let* ((url     (server:make-server-url serverdat))
	 (fullurl (conc url "/?dat=" msg)))
    (print "url=" url ", fullurl=" fullurl)
    (let* ((res   (with-input-from-request fullurl #f read-string)))
      (print "got res=" res)
      (let ((match (string-search (regexp "<body>(.*)<.body>") (caddr (string-split res "\n")))))
	(cadr match)))))

(define (server:client-login serverdat)
  (cdb:login serverdat *toppath* (server:get-client-signature)))

;; Not currently used! But, I think it *should* be used!!!
(define (server:client-logout serverdat)
  (let ((ok (and (socket? serverdat)
︙
189
190
191
192
193
194
195

























































196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214

215
216
217
218
219

220
221
222
223
224
225
226

204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
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








+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+


















-
+




-
+








			(sleep 2) ;; give server time to start
			(if (< count 5)
			    (loop (+ count 1)))))))
	      ;; we are starting a server, do not try again! That can lead to 
	      ;; recursively starting many processes!!!
	      (server:client-setup numtries: 0))
	    (debug:print-info 1 "Too many attempts, giving up")))))

;; 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 *runremote*)
                          (mutex-unlock! *heartbeat-mutex*)
                          (if sdat sdat
                              (begin
                                (sleep 4)
                                (loop))))))
         (iface       (car server-info))
         (port        (cadr server-info))
         (last-access 0))
    ;; (print "Keep-running got server-info " server-info)
    (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 server-info '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
        (mutex-lock! *heartbeat-mutex*)
        (set! last-access *last-db-access*)
        (mutex-unlock! *heartbeat-mutex*)
        (if (> (+ last-access
                  ;; (* 50 60 60)    ;; 48 hrs
                  ;; 60              ;; one minute
                  ;; (* 60 60)       ;; one hour
                  (* 45 60)          ;; 45 minutes, until the db deletion bug is fixed.
                  )
               (current-seconds))
            (begin
              (debug:print-info 2 "Server continuing, seconds since last db access: " (- (current-seconds) last-access))
              (loop 0))
            (begin
              (debug:print-info 0 "Starting to shutdown the server.")
              ;; need to delete only *my* server entry (future use)
              (set! *time-to-exit* #t)
              (open-run-close tasks:server-deregister-self tasks:open-db (get-host-name))
              (thread-sleep! 1)
              (debug:print-info 0 "Max cached queries was " *max-cache-size*)
              (debug:print-info 0 "Server shutdown complete. Exiting")
              (exit)))))))



;; all routes though here end in exit ...
(define (server:launch)
  (if (not *toppath*)
      (if (not (setup-for-run))
	  (begin
	    (debug:print 0 "ERROR: cannot find megatest.config, exiting")
	    (exit))))
  (debug:print-info 2 "Starting the standalone server")
  (let ((hostinfo (open-run-close tasks:get-best-server tasks:open-db)))
    (if hostinfo
	(debug:print-info 2 "NOT starting new server, one is already running on " (car hostinfo) ":" (cadr hostinfo))
	(if *toppath* 
	    (let* ((th2 (make-thread (lambda ()
				       (server:run 
					(if (args:get-arg "-server")
					    (args:get-arg "-server")
					    "-"))) "Server run"))
		   (th3 (make-thread (lambda ()(server:keep-running)) "Keep running"))
		   ;; (th3 (make-thread (lambda ()(server:keep-running)) "Keep running"))
		   )
	      (set! *client-non-blocking-mode* #t)
	      ;; (thread-start! th1)
	      (thread-start! th2)
	      (thread-start! th3)
	      ;; (thread-start! th3)
	      (set! *didsomething* #t)
	      ;; (thread-join! th3)
	      (thread-join! th2)
	      )
	    (debug:print 0 "ERROR: Failed to setup for megatest")))
    (exit)))
︙

Modified tasks.scm from [2c690c7219] to [cbe1edb880].

︙
197
198
199
200
201
202
203



204
205
206
207
208
209
210

197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213








+
+
+








		  (if port
		      (open-run-close tasks:server-deregister tasks:open-db host port: port)
		      (open-run-close tasks:server-deregister tasks:open-db host pid:  pid))
		  (if (null? tal)
		      #f
		      (loop (car tal)(cdr tal))))))))))

(define (tasks:remove-server-records mdb)
  (sqlite3:exec mdb "DELETE FROM servers;"))

(define (tasks:mark-server hostname port pid state)
  (if port
      (open-run-close tasks:server-deregister tasks:open-db hostname port: port)
      (open-run-close tasks:server-deregister tasks:open-db hostname pid:  pid)))


(define (tasks:kill-server status hostname port pid)
︙

This page was generated in about 0.115s by Fossil 2.24 [452f8e1ffc] 2024-03-26 15:36:00