Parallel

Run any code in parallel Processes(> use all CPUs), Threads(> speedup blocking operations), or Ractors(> use all CPUs).
Best suited for map-reduce or e.g. parallel downloads/uploads.

Install

gem install parallel

Usage

# 2 CPUs -> work in 2 processes (a,b + c)results=Parallel.map(['a','b','c'])do |one_letter|
SomeClass.expensive_calculation(one_letter)end# 3 Processes -> finished after 1 runresults=Parallel.map(['a','b','c'],in_processes: 3){ |one_letter| SomeClass.expensive_calculation(one_letter)}# 3 Threads -> finished after 1 runresults=Parallel.map(['a','b','c'],in_threads: 3){ |one_letter| SomeClass.expensive_calculation(one_letter)}# 3 Ractors -> finished after 1 runresults=Parallel.map(['a','b','c'],in_ractors: 3,ractor: [SomeClass,:expensive_calculation])

Same can be done with each

Parallel.each(['a','b','c']){ |one_letter| ... }

or each_with_index, map_with_index, flat_map

Produce one item at a time with lambda (anything that responds to .call) or Queue.

items=[1,2,3]Parallel.each(->{items.pop || Parallel::Stop}){ |number| ... }

Also supports any? or all?

Parallel.any?([1,2,3,4,5,6,7]){ |number| number == 4}# => trueParallel.all?([1,2,nil,4,5]){ |number| number != nil}# => false

Processes/Threads are workers, they grab the next piece of work when they finish.

Processes

• Speedup through multiple CPUs
• Speedup for blocking operations
• Variables are protected from change
• Extra memory used
• Child processes are killed when your main process is killed through Ctrl+c or kill -2

Threads

• Speedup for blocking operations
• Variables can be shared/modified
• No extra memory used

Ractors

• Ruby 3.0+ only
• Speedup for blocking operations
• No extra memory used
• Very fast to spawn
• Experimental and unstable
• start and finish hooks are called on main thread
• Variables must be passed in Parallel.map([1,2,3].map { |i| [i, ARGV, local_var] }, ...
• use Ractor.make_shareable to pass in global objects

ActiveRecord

Connection Lost

• Multithreading needs connection pooling, forks need reconnects
• Adjust connection pool size in config/database.yml when multithreading

# reproducibly fixes things (spec/cases/map_with_ar.rb)Parallel.each(User.all,in_processes: 8)do |user|
user.update_attribute(:some_attribute,some_value)endUser.connection.reconnect!# maybe helps: explicitly use connection poolParallel.each(User.all,in_threads: 8)do |user|
ActiveRecord::Base.connection_pool.with_connectiondouser.update_attribute(:some_attribute,some_value)endend# maybe helps: reconnect once inside every forkParallel.each(User.all,in_processes: 8)do |user|
@reconnected ||= User.connection.reconnect! || trueuser.update_attribute(:some_attribute,some_value)end

NameError: uninitialized constant

A race happens when ActiveRecord models are autoloaded inside parallel threads in environments that lazy-load, like development, test, or migrations.

To fix, autoload classes before the parallel block with either require '<modelname>' or ModelName.class.

Break

Parallel.map([1,2,3])do |i|
raiseParallel::Break# -> stops after all current items are finishedend

Parallel.map([1,2,3]){ |i| raiseParallel::Break,iifi == 2} == 2

Kill

Only use if whatever is executing in the sub-command is safe to kill at any point

Parallel.map([1,2,3])do |x|
raiseParallel::Killifx == 1# -> stop all sub-processes, killing them instantlysleep100# Do stuffend

Progress / ETA

# gem install ruby-progressbarParallel.map(1..50,progress: "Doing stuff"){sleep1}# Doing stuff | ETA: 00:00:02 | ==================== | Time: 00:00:10

Use :finish or :start hook to get progress information.

• :start has item and index
• :finish has item, index, and result

They are called on the main process and protected with a mutex. (To just get the index, use the more performant Parallel.each_with_index)

Parallel.map(1..100,finish: ->(item,i,result){ ... dosomething ... }){sleep1}

Set finish_in_order: true to call the :finish hook in the order of the input (will take longer to see initial output).

Parallel.map(1..9,finish: ->(item,i,result){puts"#{item} ok"},finish_in_order: true){sleeprand}

Worker number

Use Parallel.worker_number to determine the worker slot in which your task is running.

Parallel.each(1..5,in_processes: 2){ |i| puts"Item: #{i}, Worker: #{Parallel.worker_number}"}Item: 1,Worker: 1Item: 2,Worker: 0Item: 3,Worker: 1Item: 4,Worker: 0Item: 5,Worker: 1

Dynamically generating jobs

Example: wait for work to arrive or sleep

queue=[]Thread.new{loop{queue << rand(100);sleep2}}# job producerParallel.map(Proc.new{queue.pop},in_processes: 3){ |f| f ? puts("#{f} received") : sleep(1)}

Processes vs Wire serializer: security risk for hardened environments

Worker processes talk to the parent over an anonymous pipe using Marshal by default.

If you've hardened your host against ptrace//proc/<pid>/mem access (e.g. ptrace_scope >= 2) and want to also close the /proc/<pid>/fd/<n> pipe-reopen vector, use the HMAC serializer.

It length-prefixes and HMAC-SHA256 signs each message with a per-worker secret generated before fork, so a same-UID attacker that reopens the pipe can't inject a forged Marshal payload into the parent (which would be RCE).

Raises SecurityError on mismatch (not a StandardError).

Parallel.map(items,in_processes: 2,serializer: Parallel::Serializer::Hmac.new){ ... }

Tips

• [Benchmark/Test] Disable threading/forking with in_threads: 0 or in_processes: 0, to run the same code with different setups
• [Isolation] Do not reuse previous worker processes: isolation: true
• [Stop all processes with an alternate interrupt signal] 'INT' (from ctrl+c) is caught by default. Catch 'TERM' (from kill) with interrupt_signal: 'TERM'
• [Process count via ENV] PARALLEL_PROCESSOR_COUNT=16 will use 16 instead of the number of processors detected. This is used to reconfigure a tool using parallel without inserting custom logic.
• [Process count] parallel uses a number of processors seen by the OS for process count by default. If you want to use a value considering CPU quota, please add concurrent-ruby to your Gemfile.

TODO

• Replace Signal trapping with simple rescue Interrupt handler

Authors

Contributors

• Przemyslaw Wroblewski
• TJ Holowaychuk
• Masatomo Nakano
• Fred Wu
• mikezter
• Jeremy Durham
• Nick Gauthier
• Andrew Bowerman
• Byron Bowerman
• Mikko Kokkonen
• brian p o'rourke
• [Norio Sato]
• Neal Stewart
• Jurriaan Pruis
• Rob Worley
• Tasveer Singh
• Joachim
• yaoguai
• Bartosz Dziewoński
• yaoguai
• Guillaume Hain
• Adam Wróbel
• Matthew Brennan
• Brendan Dougherty
• Daniel Finnie
• Philip M. White
• Arlan Jaska
• Sean Walbran
• Nathan Broadbent
• Yuki Inoue
• Takumasa Ochi
• Shai Coleman
• Earlopain

Michael Grosser
michael@grosser.it
License: MIT