Sha256: c257c603a67c8849bb495122461c2b06c4c7d4695c405e2b2a48bdbb002f277b
Contents?: true
Size: 1.82 KB
Versions: 12
Compression:
Stored size: 1.82 KB
Contents
require "timeout"
#
# a simple (really really simple) map-in-parallel multithreaded mapper.
module Array::ParallelMap
def self.timeout(opts, &block)
return yield unless opts[:timeout]
begin
Timeout.timeout(opts[:timeout], &block)
rescue Timeout::Error
end
end
def peach_with_index(opts = {}, &block)
return [] if empty?
threads = []
each_with_index do |data, index|
threads << Thread.new {
Array::ParallelMap.timeout(opts) { yield(data, index) }
}
end
threads.each do |thread| thread.join end
self
end
def peach(opts = {}, &block)
peach_with_index(opts) do |data, index|
yield(data)
end
end
def pmap(opts = {}, &block)
semaphore = Mutex.new
results = []
peach_with_index(opts) do |data, index|
r = yield(data)
semaphore.synchronize { results[index] = r }
end
results
end
end
Array.send :include, Array::ParallelMap
module Array::ParallelMap::Etest
MAX = 1000
def calculate(repeat, method)
(1..repeat).to_a.send(method) do |p|
(1..MAX).inject(0) do |sum, i|
sum + p * i
end
end
end
def calculate_serial(repeat)
calculate(repeat, :map)
end
def calculate_parallel(repeat)
calculate(repeat, :pmap)
end
def test_pmap
serial = calculate_serial(4)
parallel = calculate_parallel(4)
assert_equal(serial, parallel)
end
def test_peach
semaphore = Mutex.new
results = []
array = [ 1, 2, 3, 4]
r = array.peach do |p|
semaphore.synchronize { results << p }
end
assert_equal(array, results.sort)
assert_equal(array, r)
end
def test_pmap_timeout
r = [1, 2].pmap(:timeout => 0.1) do |p|
sleep 0.2
p * p
end
assert_equal([], r)
end
end if VEX_TEST == "base"
Version data entries
12 entries across 12 versions & 1 rubygems