Module: Parallel

Defined in:
lib/parallel.rb,
lib/parallel/version.rb,
lib/parallel/serializer.rb

Defined Under Namespace

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

Constant Summary collapse

Stop = 
Object.new.freeze
VERSION =

rubocop:disable Naming/ConstantName

Version = '2.1.0'

Class Method Summary collapse

Class Method Details

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

Returns:

  • (Boolean) 


256
257
258
259
 
# File 'lib/parallel.rb', line 256

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) 


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

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 



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

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

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



261
262
263
 
# File 'lib/parallel.rb', line 261

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

.filter_map ⇒ Object 



320
321
322
 
# File 'lib/parallel.rb', line 320

def filter_map(...)
  map(...).compact
end

.flat_map ⇒ Object 



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

def flat_map(...)
  map(...).flatten(1)
end

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



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

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 



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/parallel.rb', line 225

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 



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
301
302
303
304
305
306
307
308
309
310
 
# File 'lib/parallel.rb', line 265

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.include?('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 



312
313
314
 
# File 'lib/parallel.rb', line 312

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

.physical_processor_count ⇒ Object

Number of physical processor cores on the current system.



325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
 
# File 'lib/parallel.rb', line 325

def physical_processor_count
  @physical_processor_count ||= begin
    ppc =
      case RbConfig::CONFIG["target_os"]
      when /darwin[12]/
        IO.popen("/usr/sbin/sysctl -n hw.physicalcpu").read.to_i
      when /linux/
        cores = {} # unique physical ID / core ID combinations
        phy = 0
        File.read("/proc/cpuinfo").scan(/^physical id.*|^core id.*/) do |ln|
          if ln.start_with?("physical")
            phy = ln[/\d+/]
          elsif ln.start_with?("core")
            cid = "#{phy}:#{ln[/\d+/]}"
            cores[cid] = true unless cores[cid]
          end
        end
        cores.count
      when /mswin|mingw/
        physical_processor_count_windows
      else
        processor_count
      end
    # fall back to logical count if physical info is invalid
    ppc > 0 ? ppc : processor_count
  end
end

.processor_count ⇒ Object

Number of processors seen by the OS or value considering CPU quota if the process is inside a cgroup, used for process scheduling



355
356
357
 
# File 'lib/parallel.rb', line 355

def processor_count
  @processor_count ||= Integer(ENV['PARALLEL_PROCESSOR_COUNT'] || available_processor_count)
end

.worker_number ⇒ Object 



359
360
361
 
# File 'lib/parallel.rb', line 359

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



364
365
366
 
# File 'lib/parallel.rb', line 364

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