Module: Parallel

Extended by:
ProcessorCount
Defined in:
lib/parallel.rb,
lib/parallel/version.rb,
lib/parallel/processor_count.rb

Defined Under Namespace

Modules: ProcessorCount Classes: Break, DeadWorker, ExceptionWrapper, JobFactory, Kill, UndumpableException, UserInterruptHandler, Worker

Constant Summary collapse

Stop = 
Object.new.freeze
VERSION =

rubocop:disable Naming/ConstantName

Version = '1.22.1'

Class Method Summary collapse

Methods included from ProcessorCount

physical_processor_count, processor_count

Class Method Details

.all?(*args, &block) ⇒ Boolean

Returns:

  • (Boolean) 


246
247
248
249
 
# File 'lib/parallel.rb', line 246

def all?(*args, &block)
  raise "You must provide a block when calling #all?" if block.nil?
  !!each(*args) { |*a| raise Kill unless block.call(*a) }
end

.any?(*args, &block) ⇒ Boolean

Returns:

  • (Boolean) 


241
242
243
244
 
# File 'lib/parallel.rb', line 241

def any?(*args, &block)
  raise "You must provide a block when calling #any?" if block.nil?
  !each(*args) { |*a| raise Kill if block.call(*a) }
end

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



237
238
239
 
# File 'lib/parallel.rb', line 237

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

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



251
252
253
 
# File 'lib/parallel.rb', line 251

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

.flat_map(*args, &block) ⇒ Object 



306
307
308
 
# File 'lib/parallel.rb', line 306

def flat_map(*args, &block)
  map(*args, &block).flatten(1)
end

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



231
232
233
234
235
 
# File 'lib/parallel.rb', line 231

def in_processes(options = {}, &block)
  count, options = extract_count_from_options(options)
  count ||= processor_count
  map(0...count, options.merge(in_processes: count), &block)
end

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



215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/parallel.rb', line 215

def in_threads(options = { count: 2 })
  threads = []
  count, = extract_count_from_options(options)

  Thread.handle_interrupt(Exception => :never) do
    Thread.handle_interrupt(Exception => :immediate) do
      count.times do |i|
        threads << Thread.new { yield(i) }
      end
      threads.map(&:value)
    end
  ensure
    threads.each(&:kill)
  end
end

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



255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
 
# File 'lib/parallel.rb', line 255

def map(source, options = {}, &block)
  options = options.dup
  options[:mutex] = Mutex.new

  if options[:in_processes] && options[:in_threads]
    raise ArgumentError, "Please specify only one of `in_processes` or `in_threads`."
  elsif RUBY_PLATFORM =~ (/java/) && !(options[:in_processes])
    method = :in_threads
    size = options[method] || processor_count
  elsif options[:in_threads]
    method = :in_threads
    size = options[method]
  elsif options[:in_ractors]
    method = :in_ractors
    size = options[method]
  else
    method = :in_processes
    if Process.respond_to?(:fork)
      size = options[method] || processor_count
    else
      warn "Process.fork is not supported by this Ruby"
      size = 0
    end
  end

  job_factory = JobFactory.new(source, options[:mutex])
  size = [job_factory.size, size].min

  options[:return_results] = (options[:preserve_results] != false || !!options[:finish])
  add_progress_bar!(job_factory, options)

  result =
    if size == 0
      work_direct(job_factory, options, &block)
    elsif method == :in_threads
      work_in_threads(job_factory, options.merge(count: size), &block)
    elsif method == :in_ractors
      work_in_ractors(job_factory, options.merge(count: size), &block)
    else
      work_in_processes(job_factory, options.merge(count: size), &block)
    end

  return result.value if result.is_a?(Break)
  raise result if result.is_a?(Exception)
  options[:return_results] ? result : source
end

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



302
303
304
 
# File 'lib/parallel.rb', line 302

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

.worker_number ⇒ Object 



310
311
312
 
# File 'lib/parallel.rb', line 310

def worker_number
  Thread.current[:parallel_worker_number]
end

.worker_number=(worker_num) ⇒ Object

TODO: this does not work when doing threads in forks, so should remove and yield the number instead if needed



315
316
317
 
# File 'lib/parallel.rb', line 315

def worker_number=(worker_num)
  Thread.current[:parallel_worker_number] = worker_num
end