Module: Enumerable

Defined in:
lib/parallelize/enumerable_ext.rb

Instance Method Summary collapse

Instance Method Details

#peach(num_threads, collect_exceptions = false, &block) ⇒ Array

Divides the Enumerable objects into pieces and execute with multiple threads

Parameters:

  • num_threads (Fixnum) —

    Number of concurrent threads

  • collect_exceptions (Boolean) (defaults to: false) —

    If true, waits for all threads to complete even in case of exception, and throws ParallelException at the end. If false exception is immediately thrown.

Returns:

  • (Array) —

    Threads.

Raises:

  • (ArgumentError) 


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
 
# File 'lib/parallelize/enumerable_ext.rb', line 6

def peach num_threads, collect_exceptions = false, &block
	raise ArgumentError.new("Block not given") unless block_given?
	raise ArgumentError.new("Invalid number of threads") if num_threads < 1

	threads = []
	self.each_slice((self.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.each { |e| yield e, thread_idx }
				}
			when 1
				Thread.new(slice) { |my_slice|
					my_slice.each { |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.each { 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
			else
				raise e
			end
		end
	end

	if exceptions.empty?
		threads
	else
		raise ParallelException.new(exceptions)
	end
end