Module: TeamEffort

Defined in:
lib/team_effort.rb,
lib/team_effort/version.rb

Constant Summary collapse

VERSION = 
"1.1.1"

Class Method Summary collapse

Class Method Details

.wait_for_completion(pids, arg_sets, completed_count, max_count, progress_proc, previous_proc_result) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/team_effort.rb', line 36

def self.wait_for_completion(pids, arg_sets, completed_count, max_count, progress_proc, previous_proc_result)
  pid, status = Process.wait2
  pids_index = pids.index(pid)
  if pids_index
    pids.delete_at pids_index
    arg_set = arg_sets.delete_at pids_index
    raise "TeamEffort child process failed when processing > #{arg_set} <" if !status.success?
  end

  if progress_proc
    progress_proc_args = [completed_count += 1, max_count]
    progress_proc_args << previous_proc_result if progress_proc.arity == 3
    proc_result = progress_proc.call(*progress_proc_args)
  end

  [pids, arg_sets, completed_count, proc_result]
end

.work(enumerable, max_process_count = 4, progress_proc: nil) ⇒ Object 



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

def self.work(enumerable, max_process_count = 4, progress_proc: nil)
  pids = []
  arg_sets = []
  previous_proc_result = nil

  max_count = enumerable.count
  completed_count = 0

  enumerable.each do |arg_set|
    while pids.size == max_process_count
      pids, arg_sets, completed_count, previous_proc_result = wait_for_completion(pids, arg_sets, completed_count, max_count, progress_proc, previous_proc_result)
    end

    pids << fork do
      begin
        yield arg_set
        exit! 0
      rescue => e
        $stderr.puts "TeamEffort child process error"
        $stderr.puts e
        $stderr.puts caller
        exit! 1
      end
    end
    arg_sets << arg_set
  end

  while !pids.empty?
    pids, arg_sets, completed_count, previous_proc_result = wait_for_completion(pids, arg_sets, completed_count, max_count, progress_proc, previous_proc_result)
  end
end