Class: Parallel

Inherits:
Object
  • Object
show all
Defined in:
lib/parallel.rb

Constant Summary collapse

VERSION = 
File.read( File.join(File.dirname(__FILE__),'..','VERSION') ).strip

Class Method Summary collapse

Class Method Details

.each(array, options = {}, &block) ⇒ Object 



33
34
35
36
 
# File 'lib/parallel.rb', line 33

def self.each(array, options={}, &block)
  map(array, options.merge(:preserve_results => false), &block)
  array
end

.each_with_index(array, options = {}, &block) ⇒ Object 



38
39
40
 
# File 'lib/parallel.rb', line 38

def self.each_with_index(array, options={}, &block)
  each(array, options.merge(:with_index => true), &block)
end

.in_processes(options = {}, &block) ⇒ Object 



22
23
24
25
26
27
28
29
30
31
 
# File 'lib/parallel.rb', line 22

def self.in_processes(options = {}, &block)
  count, options = extract_count_from_options(options)
  count ||= processor_count
  preserve_results = (options[:preserve_results] != false)

  pipes, pids = fork_and_start_writing(count, :preserve_results => preserve_results, &block)
  out = read_from_pipes(pipes)
  pids.each { |pid| Process.wait(pid) }
  out.map{|x| deserialize(x) } if preserve_results
end

.in_threads(options = {:count => 2}) ⇒ Object 



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
 
# File 'lib/parallel.rb', line 6

def self.in_threads(options={:count => 2})
  count, options = extract_count_from_options(options)

  out = []
  threads = []

  count.times do |i|
    threads[i] = Thread.new do
      out[i] = yield(i)
    end
  end

  threads.each{|t| t.join }
  out
end

.map(array, options = {}) ⇒ Object 



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
 
# File 'lib/parallel.rb', line 42

def self.map(array, options = {})
  array = array.to_a if array.is_a?(Range)

  if options[:in_threads]
    method = :in_threads
    size = options[method]
  else
    method = :in_processes
    size = options[method] || processor_count
  end

  # work in #{size} threads that use threads/processes
  results = []
  current = -1

  in_threads(size) do
    # as long as there are more items, work on one of them
    loop do
      index = Thread.exclusive{ current+=1 }
      break if index >= array.size
      results[index] = *send(method, options.merge(:count => 1)) do
        args = [array[index]]
        args << index if options[:with_index]
        yield *args
      end
    end
  end

  results
end

.map_with_index(array, options = {}, &block) ⇒ Object 



73
74
75
 
# File 'lib/parallel.rb', line 73

def self.map_with_index(array, options={}, &block)
  map(array, options.merge(:with_index => true), &block)
end

.processor_count ⇒ Object 



77
78
79
80
81
82
83
84
 
# File 'lib/parallel.rb', line 77

def self.processor_count
  case RUBY_PLATFORM
  when /darwin/
    `hwprefs cpu_count`.to_i
  when /linux/
    `cat /proc/cpuinfo | grep processor | wc -l`.to_i
  end
end