Module: Parallel

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

Defined Under Namespace

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

Constant Summary collapse

Stop = 
Object.new.freeze
VERSION =

rubocop:disable Naming/ConstantName

Version = '1.27.0'

Class Method Summary collapse

Class Method Details

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

Returns:

  • (Boolean) 


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

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) 


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

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 



234
235
236
 
# File 'lib/parallel.rb', line 234

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

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



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

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

.filter_map ⇒ Object 



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

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

.flat_map ⇒ Object 



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

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

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



228
229
230
231
232
 
# File 'lib/parallel.rb', line 228

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 



212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/parallel.rb', line 212

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 



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

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 



299
300
301
 
# File 'lib/parallel.rb', line 299

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.



312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
 
# File 'lib/parallel.rb', line 312

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



342
343
344
 
# File 'lib/parallel.rb', line 342

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

.worker_number ⇒ Object 



346
347
348
 
# File 'lib/parallel.rb', line 346

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



351
352
353
 
# File 'lib/parallel.rb', line 351

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