Class: Bootsnap::CLI::WorkerPool

Inherits:
Object
  • Object
show all
Defined in:
lib/bootsnap/cli/worker_pool.rb

Defined Under Namespace

Classes: Inline, Worker

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size:, jobs: {}) ⇒ WorkerPool

Returns a new instance of WorkerPool.



84
85
86
87
88
89
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 84

def initialize(size:, jobs: {})
  @size = size
  @jobs = jobs
  @queue = Queue.new
  @pids = []
end

Class Method Details

.create(size:, jobs:) ⇒ Object 



7
8
9
10
11
12
13
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 7

def create(size:, jobs:)
  if size > 0 && Process.respond_to?(:fork)
    new(size: size, jobs: jobs)
  else
    Inline.new(jobs: jobs)
  end
end

Instance Method Details

#dispatch_loop ⇒ Object 



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 99

def dispatch_loop
  loop do
    case job = @queue.pop
    when nil
      @workers.each do |worker|
        worker.write([:exit])
        worker.close
      end
      return true
    else
      unless @workers.sample.write(job, block: false)
        free_worker.write(job)
      end
    end
  end
end

#free_worker ⇒ Object 



116
117
118
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 116

def free_worker
  IO.select(nil, @workers)[1].sample
end

#push(*args) ⇒ Object 



120
121
122
123
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 120

def push(*args)
  @queue.push(args)
  nil
end

#shutdown ⇒ Object 



125
126
127
128
129
130
131
132
133
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 125

def shutdown
  @queue.close
  @dispatcher_thread.join
  @workers.each do |worker|
    _pid, status = Process.wait2(worker.pid)
    return status.exitstatus unless status.success?
  end
  nil
end

#spawn ⇒ Object 



91
92
93
94
95
96
97
 
# File 'lib/bootsnap/cli/worker_pool.rb', line 91

def spawn
  @workers = @size.times.map { Worker.new(@jobs) }
  @workers.each(&:spawn)
  @dispatcher_thread = Thread.new { dispatch_loop }
  @dispatcher_thread.abort_on_exception = true
  true
end