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
|
# File 'lib/parallelize/enumerable_ext.rb', line 3
def self.it enum, num_threads, method, collect_exceptions, &block
raise ArgumentError.new("Block not given") unless block_given?
raise ArgumentError.new("Invalid number of threads") if num_threads < 1
threads = []
enum.each_slice((enum.count{true} / num_threads.to_f).ceil) do |slice|
threads <<
case block.arity
when 2
Thread.new(slice, threads.length) { |my_slice, thread_idx|
my_slice.send(method) { |e| yield e, thread_idx }
}
when 1
Thread.new(slice) { |my_slice|
my_slice.send(method) { |e| yield e }
}
when 0, -1
raise ArgumentError.new("Invalid arity: #{block.arity}") if
RUBY_VERSION !~ /^1.8\./ && block.arity == -1
Thread.new(slice) { |my_slice|
my_slice.send(method) { yield }
}
else
raise ArgumentError.new("Invalid arity: #{block.arity}")
end
end
exceptions = {}
threads.each_with_index do |thr, idx|
begin
thr.join
rescue Exception => e
if collect_exceptions
exceptions[idx] = e
nil
else
raise e
end
end
end
unless exceptions.empty?
raise ParallelException.new(exceptions)
end
if method == :each
threads
elsif method == :map
threads.map(&:value).inject(:+)
end
end
|