| ︙ | | |
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
-
+
|
(define *heartbeat-mutex* (make-mutex))
;;======================================================================
;; S E R V E R
;;======================================================================
(define (nmsg-transport:run dbstruct hostn run-id server-id #!key (retrynum 1000))
(debug:print 2 #f "Attempting to start the server ...")
(debug:print 2 *default-log-port* "Attempting to start the server ...")
(let* ((start-port (portlogger:open-run-close portlogger:find-port))
(server-thread (make-thread (lambda ()
(nmsg-transport:try-start-server dbstruct run-id start-port server-id))
"server thread"))
(tdbdat (tasks:open-db)))
(thread-start! server-thread)
(thread-sleep! 0.1)
|
| ︙ | | |
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
|
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
|
-
+
-
+
-
+
-
+
|
(tasks:server-set-state! (db:delay-if-busy tdbdat) server-id "running")
(thread-start! (make-thread
(lambda ()(nmsg-transport:keep-running server-id run-id))
"keep running"))
(thread-join! server-thread))
(if (> retrynum 0)
(begin
(debug:print 0 #f "WARNING: Failed to connect to server (self) on host " hostn ":" start-port ", trying again.")
(debug:print 0 *default-log-port* "WARNING: Failed to connect to server (self) on host " hostn ":" start-port ", trying again.")
(tasks:server-delete-record (db:delay-if-busy tdbdat) server-id "failed to start, never received server alive signature")
(portlogger:open-run-close portlogger:set-failed start-port)
(nmsg-transport:run dbstruct hostn run-id server-id))
(begin
(debug:print 0 #f "ERROR: could not find an open port to start server on. Giving up")
(debug:print 0 *default-log-port* "ERROR: could not find an open port to start server on. Giving up")
(exit 1))))))
(define (nmsg-transport:try-start-server dbstruct run-id portnum server-id)
(let ((repsoc (nn-socket 'rep)))
(nn-bind repsoc (conc "tcp://*:" portnum))
(let loop ((msg-in (nn-recv repsoc)))
(let* ((dat (db:string->obj msg-in transport: 'nmsg)))
(debug:print 0 #f "server, received: " dat)
(debug:print 0 *default-log-port* "server, received: " dat)
(let ((result (api:execute-requests dbstruct dat)))
(debug:print 0 #f "server, sending: " result)
(debug:print 0 *default-log-port* "server, sending: " result)
(nn-send repsoc (db:obj->string result transport: 'nmsg)))
(loop (nn-recv repsoc))))))
;; all routes though here end in exit ...
;;
(define (nmsg-transport:launch run-id)
(let* ((tdbdat (tasks:open-db))
|
| ︙ | | |
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
-
+
|
(dat (vector "ping" our-key))
(result (condition-case
(nmsg-transport:client-api-send-receive-raw req dat timeout: timeout)
((timeout)(set! success #f) #f)))
(key (if success
(vector-ref result 1)
#f)))
(debug:print 0 #f "success=" success ", key=" key ", expected-key=" expected-key ", equal? " (equal? key expected-key))
(debug:print 0 *default-log-port* "success=" success ", key=" key ", expected-key=" expected-key ", equal? " (equal? key expected-key))
(if (and success
(or (not expected-key) ;; just getting a reply is good enough then
(equal? key expected-key)))
(if return-socket
req
(begin
(if (not socket)(nn-close req)) ;; don't want a side effect of closing socket if handed it
|
| ︙ | | |
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
-
-
+
+
-
+
|
(if success (thread-terminate! timeout)))
;; raise timeout error if timed out
(if success
(if (and (vector? result)
(vector-ref result 0)) ;; did it fail at the server?
result ;; nope, all good
(begin
(debug:print 0 #f "ERROR: error occured at server, info=" (vector-ref result 2))
(debug:print 0 #f " client call chain:")
(debug:print 0 *default-log-port* "ERROR: error occured at server, info=" (vector-ref result 2))
(debug:print 0 *default-log-port* " client call chain:")
(print-call-chain (current-error-port))
(debug:print 0 #f " server call chain:")
(debug:print 0 *default-log-port* " server call chain:")
(pp (vector-ref result 1) (current-error-port))
(signal (vector-ref result 0))))
(signal (make-composite-condition
(make-property-condition 'timeout 'message "nmsg-transport:client-api-send-receive-raw timed out talking to server"))))))
;; run nmsg-transport:keep-running in a parallel thread to monitor that the db is being
;; used and to shutdown after sometime if it is not.
|
| ︙ | | |
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
|
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
|
-
+
-
+
-
+
|
;;======================================================================
;; DO NOT USE
;;
(define (nmsg-transport:client-signal-handler signum)
(handle-exceptions
exn
(debug:print 0 #f " ... exiting ...")
(debug:print 0 *default-log-port* " ... exiting ...")
(let ((th1 (make-thread (lambda ()
(if (not *received-response*)
(receive-message* *runremote*))) ;; flush out last call if applicable
"eat response"))
(th2 (make-thread (lambda ()
(debug:print 0 #f "ERROR: Received ^C, attempting clean exit. Please be patient and wait a few seconds before hitting ^C again.")
(debug:print 0 *default-log-port* "ERROR: Received ^C, attempting clean exit. Please be patient and wait a few seconds before hitting ^C again.")
(thread-sleep! 3) ;; give the flush three seconds to do it's stuff
(debug:print 0 #f " Done.")
(debug:print 0 *default-log-port* " Done.")
(exit 4))
"exit on ^C timer")))
(thread-start! th2)
(thread-start! th1)
(thread-join! th2))))
|