Class: TestQueue::Runner
- Inherits:
-
Object
show all
- Defined in:
- lib/test_queue/runner.rb,
lib/test_queue/runner/sample.rb,
lib/test_queue/runner/cucumber.rb,
lib/test_queue/runner/puppet_lint.rb,
lib/test_queue/runner/minitest.rb,
lib/test_queue/runner/rspec.rb
Defined Under Namespace
Classes: Cucumber, MiniTest, PuppetLint, RSpec, Sample
Instance Attribute Summary collapse
Instance Method Summary
collapse
Constructor Details
#initialize(queue, concurrency = nil, socket = nil, relay = nil) ⇒ Runner
Returns a new instance of Runner.
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
|
# File 'lib/test_queue/runner.rb', line 25
def initialize(queue, concurrency=nil, socket=nil, relay=nil)
raise ArgumentError, 'array required' unless Array === queue
if forced = ENV['TEST_QUEUE_FORCE']
forced = forced.split(/\s*,\s*/)
whitelist = Set.new(forced)
queue = queue.select{ |s| whitelist.include?(s.to_s) }
queue.sort_by!{ |s| forced.index(s.to_s) }
end
@procline = $0
@queue = queue
@suites = queue.inject(Hash.new){ |hash, suite| hash.update suite.to_s => suite }
@workers = {}
@completed = []
@concurrency =
concurrency ||
(ENV['TEST_QUEUE_WORKERS'] && ENV['TEST_QUEUE_WORKERS'].to_i) ||
if File.exists?('/proc/cpuinfo')
File.read('/proc/cpuinfo').split("\n").grep(/processor/).size
elsif RUBY_PLATFORM =~ /darwin/
`/usr/sbin/sysctl -n hw.activecpu`.to_i
else
2
end
@socket =
socket ||
ENV['TEST_QUEUE_SOCKET'] ||
"/tmp/test_queue_#{$$}_#{object_id}.sock"
@relay =
relay ||
ENV['TEST_QUEUE_RELAY']
if @relay == @socket
STDERR.puts "*** Detected TEST_QUEUE_RELAY == TEST_QUEUE_SOCKET. Disabling relay mode."
@relay = nil
elsif @relay
@queue = []
end
end
|
Instance Attribute Details
#concurrency ⇒ Object
Returns the value of attribute concurrency.
23
24
25
|
# File 'lib/test_queue/runner.rb', line 23
def concurrency
@concurrency
end
|
Instance Method Details
#after_fork(num) ⇒ Object
240
241
|
# File 'lib/test_queue/runner.rb', line 240
def after_fork(num)
end
|
#after_fork_internal(num, iterator) ⇒ Object
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
|
# File 'lib/test_queue/runner.rb', line 216
def after_fork_internal(num, iterator)
srand
output = File.open("/tmp/test_queue_worker_#{$$}_output", 'w')
$stdout.reopen(output)
$stderr.reopen($stdout)
$stdout.sync = $stderr.sync = true
$0 = "test-queue worker [#{num}]"
puts
puts "==> Starting #$0 (#{Process.pid}) - iterating over #{iterator.sock}"
puts
after_fork(num)
end
|
#around_filter(suite) ⇒ Object
236
237
238
|
# File 'lib/test_queue/runner.rb', line 236
def around_filter(suite)
yield
end
|
#cleanup_worker ⇒ Object
251
252
|
# File 'lib/test_queue/runner.rb', line 251
def cleanup_worker
end
|
#connect_to_relay ⇒ Object
328
329
330
|
# File 'lib/test_queue/runner.rb', line 328
def connect_to_relay
TCPSocket.new(*@relay.split(':'))
end
|
#distribute_queue ⇒ Object
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
|
# File 'lib/test_queue/runner.rb', line 286
def distribute_queue
return if relay?
remote_workers = 0
until @queue.empty? && remote_workers == 0
if IO.select([@server], nil, nil, 0.1).nil?
reap_worker(false) if @workers.any? else
sock = @server.accept
cmd = sock.gets.strip
case cmd
when 'POP'
if obj = @queue.shift
data = Marshal.dump(obj.to_s)
sock.write(data)
end
when /^SLAVE (\d+) ([\w\.-]+)/
num = $1.to_i
slave = $2
remote_workers += num
STDERR.puts "*** #{num} workers connected from #{slave} after #{Time.now-@start_time}s"
when /^WORKER (\d+)/
data = sock.read($1.to_i)
worker = Marshal.load(data)
worker_completed(worker)
remote_workers -= 1
end
sock.close
end
end
ensure
stop_master
until @workers.empty?
reap_worker
end
end
|
#execute ⇒ Object
79
80
81
82
83
84
85
86
87
88
|
# File 'lib/test_queue/runner.rb', line 79
def execute
$stdout.sync = $stderr.sync = true
@start_time = Time.now
@concurrency > 0 ?
execute_parallel :
execute_sequential
ensure
summarize_internal unless $!
end
|
#execute_parallel ⇒ Object
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
# File 'lib/test_queue/runner.rb', line 145
def execute_parallel
start_master
spawn_workers
distribute_queue
ensure
stop_master
@workers.each do |pid, worker|
Process.kill 'KILL', pid
end
until @workers.empty?
reap_worker
end
end
|
#execute_sequential ⇒ Object
141
142
143
|
# File 'lib/test_queue/runner.rb', line 141
def execute_sequential
exit! run_worker(@queue)
end
|
#prepare(concurrency) ⇒ Object
233
234
|
# File 'lib/test_queue/runner.rb', line 233
def prepare(concurrency)
end
|
#reap_worker(blocking = true) ⇒ Object
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
|
# File 'lib/test_queue/runner.rb', line 261
def reap_worker(blocking=true)
if pid = Process.waitpid(-1, blocking ? 0 : Process::WNOHANG) and worker = @workers.delete(pid)
worker.status = $?
worker.end_time = Time.now
if File.exists?(file = "/tmp/test_queue_worker_#{pid}_output")
worker.output = IO.binread(file)
FileUtils.rm(file)
end
if File.exists?(file = "/tmp/test_queue_worker_#{pid}_stats")
worker.stats = Marshal.load(IO.binread(file))
FileUtils.rm(file)
end
relay_to_master(worker) if relay?
worker_completed(worker)
end
end
|
#relay? ⇒ Boolean
324
325
326
|
# File 'lib/test_queue/runner.rb', line 324
def relay?
!!@relay
end
|
#relay_to_master(worker) ⇒ Object
332
333
334
335
336
337
338
339
340
341
|
# File 'lib/test_queue/runner.rb', line 332
def relay_to_master(worker)
worker.host = Socket.gethostname
data = Marshal.dump(worker)
sock = connect_to_relay
sock.puts("WORKER #{data.bytesize}")
sock.write(data)
ensure
sock.close if sock
end
|
#run_worker(iterator) ⇒ Object
243
244
245
246
247
248
249
|
# File 'lib/test_queue/runner.rb', line 243
def run_worker(iterator)
iterator.each do |item|
puts " #{item.inspect}"
end
return 0 end
|
#spawn_workers ⇒ Object
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
# File 'lib/test_queue/runner.rb', line 196
def spawn_workers
prepare(@concurrency)
@concurrency.times do |i|
num = i+1
pid = fork do
@server.close if @server
iterator = Iterator.new(relay?? @relay : @socket, @suites, method(:around_filter))
after_fork_internal(num, iterator)
ret = run_worker(iterator) || 0
cleanup_worker
exit! ret
end
@workers[pid] = Worker.new(pid, num)
end
end
|
#start_master ⇒ Object
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
|
# File 'lib/test_queue/runner.rb', line 161
def start_master
if relay?
begin
sock = connect_to_relay
sock.puts("SLAVE #{@concurrency} #{Socket.gethostname}")
sock.close
rescue Errno::ECONNREFUSED
STDERR.puts "*** Unable to connect to relay #{@relay}. Aborting.."
exit! 1
end
else
if @socket =~ /^(?:(.+):)?(\d+)$/
address = $1 || '0.0.0.0'
port = $2.to_i
@socket = "#$1:#$2"
@server = TCPServer.new(address, port)
else
FileUtils.rm(@socket) if File.exists?(@socket)
@server = UNIXServer.new(@socket)
end
end
desc = "test-queue master (#{relay?? "relaying to #{@relay}" : @socket})"
puts "Starting #{desc}"
$0 = "#{desc} - #{@procline}"
end
|
#stats ⇒ Object
70
71
72
73
74
75
76
77
|
# File 'lib/test_queue/runner.rb', line 70
def stats
@stats ||=
if File.exists?(file = stats_file)
Marshal.load(IO.binread(file)) || {}
else
{}
end
end
|
#stats_file ⇒ Object
136
137
138
139
|
# File 'lib/test_queue/runner.rb', line 136
def stats_file
ENV['TEST_QUEUE_STATS'] ||
'.test_queue_stats'
end
|
#stop_master ⇒ Object
188
189
190
191
192
193
194
|
# File 'lib/test_queue/runner.rb', line 188
def stop_master
return if relay?
FileUtils.rm_f(@socket) if @socket && @server.is_a?(UNIXServer)
@server.close rescue nil if @server
@socket = @server = nil
end
|
#summarize ⇒ Object
133
134
|
# File 'lib/test_queue/runner.rb', line 133
def summarize
end
|
#summarize_internal ⇒ Object
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
|
# File 'lib/test_queue/runner.rb', line 90
def summarize_internal
puts
puts "==> Summary (#{@completed.size} workers in %.4fs)" % (Time.now-@start_time)
puts
@failures = ''
@completed.each do |worker|
summary, failures = summarize_worker(worker)
@failures << failures if failures
puts " [%2d] %60s %4d suites in %.4fs (pid %d exit %d%s)" % [
worker.num,
summary,
worker.stats.size,
worker.end_time - worker.start_time,
worker.pid,
worker.status.exitstatus,
worker.host && " on #{worker.host.split('.').first}"
]
end
unless @failures.empty?
puts
puts "==> Failures"
puts
puts @failures
end
puts
if @stats
File.open(stats_file, 'wb') do |f|
f.write Marshal.dump(stats)
end
end
summarize
estatus = @completed.inject(0){ |s, worker| s + worker.status.exitstatus }
estatus = 255 if estatus > 255
exit!(estatus)
end
|
#summarize_worker(worker) ⇒ Object
254
255
256
257
258
259
|
# File 'lib/test_queue/runner.rb', line 254
def summarize_worker(worker)
num_tests = ''
failures = ''
[ num_tests, failures ]
end
|
#worker_completed(worker) ⇒ Object
281
282
283
284
|
# File 'lib/test_queue/runner.rb', line 281
def worker_completed(worker)
@completed << worker
puts worker.output if ENV['TEST_QUEUE_VERBOSE'] || worker.status.exitstatus != 0
end
|