24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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
|
+
+
+
+
-
+
-
-
+
+
+
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
;; (use trace)
;; (trace nn-bind nn-socket nn-assert nn-recv nn-send thread-terminate! nn-close )
(define port 22022)
;; get needed stuff from commandline
;;
(define queuelen #f)
(define cmd '()) ;; cmd is run to give a count of the queue length => returns number in queue
(define usage "Usage: queuefeeder-server port target_queue_length command
where command is a script or program that gives an integer on stdout of current queue length")
(let ((args (argv)))
(if (> (length args) 2)
(if (> (length args) 3)
(begin
(set! port (cadr args))
(set! cmd (caddr args))) ;; no params supported
(set! port (cadr args))
(set! queuelen (string->number (caddr args)))
(set! cmd (cadddr args))) ;; no params supported
(begin
(print "Usage: queuefeeder-server port command")
(print " where the command gives an integer on stdout indicating the queue load")
(exit))))
(print usage)
(exit))))
(if (not queuelen)
(begin
(print "queuelen must be a number")
(print usage)
(exit)))
(print "Running queue feeder with port=" port ", command=" cmd)
(define rep (nn-socket 'rep))
(print "connecting, got: " (nn-bind rep (conc "tcp://" "*" ":" port)))
(define *current-delay* 0)
(define (exp-droop-calc x targ)
(cond
((> (- x targ) 1) 136) ;; top off at 136 seconds
(else
(let ((res (* 50 (exp (- x targ)))))
(cond
((and (> res 0)(< res 0.01)) 0.01)
((> res 45) 45) ;; cap at 45 seconds
(else res))))))
(define (piecewise-droop-calc x targ)
(let ((top 50))
(cond
((> (- x targ) 0) top) ;; top off at 45 seconds
((> x (- targ top))(+ (* 1 (- x (- targ top)))(/ (- top targ) targ)))
(else (let ((res (/ x targ)))
(if (< res 0.01)
0.01
res))))))
(define (server soc)
(print "server starting")
(let loop ((msg-in (nn-recv soc))
(count 0))
(if (eq? 0 (modulo count 1000))
(print "server received: " msg-in ", count=" count))
|
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
-
+
|
(thread-start! (make-thread (lambda ()
(let ((delay-time (string->number (or (get-environment-variable "QUEUE_CHK_DELAY") "30"))))
(let loop ()
(with-input-from-pipe
cmd
(lambda ()
(let* ((val (read))
(droop-val (if (number? val)(/ val 500) #f)))
(droop-val (if (number? val)(piecewise-droop-calc val queuelen) #f)))
;; val is number of jobs in queue. Use a linear droop of val/40
(mutex-lock! *current-delay-mutex*)
(set! *current-delay* (or droop-val 30)) ;; (/ (or droop-val 100) 50))
(mutex-unlock! *current-delay-mutex*)
(print "droop-val=" droop-val)
(thread-sleep! delay-time))))
(loop))))))
|