blob: 59c8486f41c0517812dfc2febe2cbd60361cb82a (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
# frozen_string_literal: true
require "drb"
require "drb/unix"
module ActiveSupport
module Testing
class Parallelization # :nodoc:
class Server
include DRb::DRbUndumped
def initialize
@queue = Queue.new
end
def record(reporter, result)
reporter.synchronize do
reporter.record(result)
end
end
def <<(o)
@queue << o
end
def pop; @queue.pop; end
end
@after_fork_hooks = []
def self.after_fork_hook(&blk)
@after_fork_hooks << blk
end
def self.after_fork_hooks
@after_fork_hooks
end
@run_cleanup_hooks = []
def self.run_cleanup_hook(&blk)
@run_cleanup_hooks << blk
end
def self.run_cleanup_hooks
@run_cleanup_hooks
end
def initialize(queue_size)
@queue_size = queue_size
@queue = Server.new
@pool = []
@url = DRb.start_service("drbunix:", @queue).uri
end
def after_fork(worker)
self.class.after_fork_hooks.each do |cb|
cb.call(worker)
end
end
def run_cleanup(worker)
self.class.run_cleanup_hooks.each do |cb|
cb.call(worker)
end
end
def start
@pool = @queue_size.times.map do |worker|
fork do
DRb.stop_service
after_fork(worker)
queue = DRbObject.new_with_uri(@url)
while job = queue.pop
klass = job[0]
method = job[1]
reporter = job[2]
result = Minitest.run_one_method(klass, method)
queue.record(reporter, result)
end
run_cleanup(worker)
end
end
end
def <<(work)
@queue << work
end
def shutdown
@queue_size.times { @queue << nil }
@pool.each { |pid| Process.waitpid pid }
end
end
end
end
|
