Ruby concurrent, process, thread, gil, eventmachine, celluloid

October 6, 2021By Simo Ruby

Regarding concurrency and parallelism, it happened not long ago. A group of colleagues went to Family Mart to buy lunch, and brought them back. The company only has one microwave oven for heating, and there are two microwave ovens at Family Mart that can be heated. In other words, our party People go to buy lunch together. This is a concurrent program, and then it can be heated in parallel in Family Mart. However, if you take it back to the company, because there is only one microwave oven (single core), it can only be executed one by one in sequence.

Serial execution

Give a range to an array and get the index of a specific number

1 < span class="lnt">2 3 4 5 6 7 

range = 0.. .100_000_000number = 99_999_999puts range.to_a.index(number) ? time ruby sequential.rb99999999ruby sequential.rb 4.04s < span class="n">user 0.56s system 98% cpu  4.667 total

It takes about 5s to execute this code, using 99% CPU (single core).

Parallel execution

By dividing the range and the fork< in the Ruby standard library /code> method, we can use multiple processes to execute the above code.








 1  2  3  4  5  6  7  8  9 10 11 12 13 


range1 = 0< span class="o">...50_000_000range2 =< /span> 50_000_000...100_000_000number = 99_999_999puts "Parent #{Process.< /span>pid}" fork { puts "Child1 #{Process.pid}: #{range1.to_a.index< /span>(number) }" }fork { puts "Child2 #{Process.pid}: #{range2.to_a.index< /span>(number) }" }Process.wait? time ruby parallel.rbParent 5086Child2 5100 : 49999999Child1 5099 span>:ruby parallel. rb 3.73s user 0.43s span> system 192% cpu   2.162 total
 td> 



Under the multi-core CPU, because the range is split in half, the processing time is also faster. (Numbers have There will be some gaps). Process.wait means waiting for all child processes to complete.
Because of the existence of GIL (global interpreter lock), Ruby MRI needs to use the core of the CPU The only way is multi-process.
Unicorn fork Rails main process spawns multiple workers to handle HTTP requests.
The biggest advantage of this multi-process approach is sufficient. Multi-core CPU is used and memory is not shared between processes, so debugging will be easier, but because memory is not shared, it means that memory consumption will become larger. However, starting from Ruby 2.0, using the COW (Copy On Write) function of the system can be Let fork processes share memory data, and only copy data when the data changes.
Threads
Starting from Ruby 1.9, the thread implementation is changed to the system's Native Threads and threads are also scheduled by the operating system. But due to the existence of the GIL, only one thread of a Ruby process can get the GIL running at the same time. The meaning of the existence of GIL is thread safety. To prevent race conditions, it is easier to implement GIL than to achieve thread safety on data structures.
However, in fact, GIL cannot completely prevent race conditions from occurring =. =
Race Condition






 1  2  3  2  3  span> 4  5  6  7  8  9 10 11 12 13 14 15 16 


# threads.rb@executed = falsedef ensure_executed unless @executed puts "executing!" @executed span> = true endend threads = 10.times.map { Thread.new { ensure_executed } }< span class="n">threads.each(&:join) # The main thread waits for the execution of the child thread to finish before continuing to run back, such as p'done' at the back.? ruby threads.rbexecuting!executing!executing!< /span>executing!





Because of all The threads share a @executed variable, and read (unless @executed) and write (@executed = true) operations are not atomic. That is to say, when the value of this @executed is read in a certain thread, the @executed may have been rewritten in another thread.
 < p>GIL and Blocking I/O
If the GIL can't allow multiple threads to run at the same time, then why do multiple threads do it? In fact, there are still high-end places. When When a thread encounters blocking I/O, it will release the GIL to other threads, such as HTTP requests, database queries, disk reads and writes, and even sleep .
 < div class="highlight"> 





1 2 3 4 5 6 7 8 < /pre>   # sleep.rbthreads = 10.times.< span class="n">map do |i| Thread.new< /span> { sleep 1  }endthreads.each(&:join)? time ruby < span class="nb">sleep.rbruby sleep.rb 0.09s user 0.03 span>s system 9% cpu  1.146 total


Ten threads execute sleep It does not need to be executed for 10s. The execution right of the thread will be handed over to sleep. Compared to using the process, the thread is lighter, and you can even run thousands of threads to handle blocking I/ The O operation is also very useful. However, you have to be careful of race-conditions. If you use a mutex to avoid it, you have to pay attention to the occurrence of deadlocks. In addition, the thread Switching between threads is also costly, so if there are too many threads, time will be spent switching threads.
Puma allows multiple threads to be used in each process, and each process has Respective thread pools. Most of the time, you will not encounter the above-mentioned competition problem, because each HTTP request is processed in a different thread.
EventMachine
EventMachine(EM) is a gem that provides event-driven I/O based on the Reactor design pattern. One advantage of using EventMachine is that when processing a large number of IO operations, there is no need to manually handle multithreading. EM can be in one Process multiple HTTP requests in the thread.






 1  2  3  4 < span class="lnt"> 5  6  7  8  9 10 11 12 < /span>13 14 15  16 


# em.rbrequire 'eventmachine'EM.< span class="n">run do EM.add_timer(1)< /span> do puts 'sleeping...' EM.system('sleep 1') { puts "woke up!" } puts 'continuing...' end EM.add_timer(3)< /span> { EM. stop }end? ruby em.rbsleeping...continuing...woke up!
 



EM.system simulates the I/O operation, and passes in a block as a callback, You can continue to respond to other operations (events) while waiting for the callback. However, because the complex system has to handle the success and failure callbacks, and other events and callbacks may be nested inside the callbacks, it is easy to fall into Callback Hell
Fiber
Used so few, grammar mastered and forgotten, grammar mastered and forgotten =. =
Fiber is a new lightweight running unit since Ruby 1.9. Similar to the suspension of threads and resume execution of code, the biggest difference is that threads are executed due to operating system scheduling, while Fiber is handled by the programmer himself.






 1  2  3  4 < span class="lnt"> 5  6  7  8  9 10 11 12 < /span>13 14 15 16 17 18 < /span>19 20 21  22 


# fiber.rbfiber = Fiber.new do # 3. Fiber.yield hand over the right to execute, and return 1 (here)< /span> Fiber.yield  1 # 5. After the execution is complete, return to 2 2end# 1. After the fiber is created, the code inside will not be executed# 2. Only when resume is called will it be executedputs fiber.resume# 4. Go on here...put s fiber.resume# 5. The fiber has hung up, an error is reportedputs fiber.< /span>resume? ruby  fiber.rb12dead fiber called (FiberError)


 


Fiber combined with EventMachine can avoid Callback Hell






 1  2  3  4  5  6  7  8  9  10 


EventMachine .run do page span> = EM:: HttpRequest.new('https://google.ca/'). get page.errback { puts "Google is down" } page.callback { url = 'https://google.ca/search?q=universe.com' about  = EM::HttpRequest.new(url).get about< span class="o">.errback { ...< /span> } about. callback { ... } }end





Rewrite with Fiber





 < pre class="chroma"> 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 

EventMachine.run do < span class="no">Fiber.new { page = http_get(< /span>'http://www.google.com/') if< /span> page.response_header.status == 200 about =< /span> http_get('https://google.ca/search?q= universe.com') # ... else puts "Google is down" end }.resumeenddef http_get(url) current_fiber = Fiber< span class="o">.current http = EM::HttpRequest. new(url).get http.callback { current_fiber.resume(http) } http.errback span> { current_fiber.resume (http) } Fiber.yieldend




 < /div> 
If I/O operations are performed in Fiber, the entire thread will be blocked by this fiber. After all, it is the reason of GIL. In fact, it does not really solve the concurrency problem, and the syntax of Fiber estimates that I will be Forget it.
Celluloid
Celluloid borrows from Erlang and brings a similar Actor model to Ruby. Each include contains  The >Celluloid class has become an Actor with its own thread of execution.


 < tbody> 
 < /tr> 



 1  2  3  4  5  6 < span class="lnt"> 7  8  9 10 11 12 13 14 < /span>15 16 17  18 19 20 21 22 23 24 25  26 


class Sheen include Celluloid def span> initialize(name)  @name = name end def set_status(status) @status = < span class="n">status end def report "#{@name } is #{@status}" endendirb (main):009:0> charlie = Sheen.new "Charlie Sheen"=>  #irb(main):010 :0> charlie.set_status "winning!"=> "winning!"irb(main):011:0> < span class="n">charlie.report=> "Charlie Sheen is winning!"irb(main):012: 0> charlie. span>async.set_status " asynchronously winning!"=> #irb(main):013: 0> charlie.report=> "Charlie Sheen is asynchronously winning!"




Celluloid::Proxy::Async The object will intercept the method call, and then save it to the Actor concurrent object In the call queue, the program can be executed (asynchronously) without waiting for a response. Each concurrent object has its own call queue, and executes the method calls inside one by one in order.
The Actor The biggest difference with Erlang's Actor model is that Erlang variables are immutable, and Ruby does not have this restriction, so the process of message (object) delivery is May be modified, unless freeze?

https://engineering.universe.com/introduction-to-concurrency-models-with-ruby-part- i-550d0dbb970
https://www.jstorimer.com/blogs/workingwithcode/8085491-nobody-understands-the-gil
http://merbist.com/2011/ 02/22/concurrency-in-ruby-explained/
  
  Actors and Threads  from mperham 

http://pra cticingruby.com/articles/gentle-intro-to-actor-based-concurrency
 Original: Big column Ruby concurrency, process, thread, GIL, EventMachine, Celluloid

< p>





1 < /span>2 3 4  5 6 7 


range = 0...100_000_000number = 99_999_999puts range. to_a.index(number)? time  ruby sequential.rb99999999ruby sequential.rb 4.04s user 0< span class="o">.56s system 98% cpu  4. 667 total









1 2 3 4 5 6 7 


range =  0...100_000_000number = 99_999_999puts range.to_a.< span class="n">index(number)? time ruby sequential< /span>.rb99999999 ruby sequential.rb  4.04s user 0.56s system 98% cpu  4.667 total









 1  2  3  4  5  6  7  8  9 10 11 12 13 


range1 = 0...50_000_000range2 = 50_000_000...100_000_000number = 99_999_999puts "Parent #{Process.pid}"fork { puts "Child1 #{Process.pid}: #{range1.to_a.index(number)}" }< /span>fork { puts "Child2 #{Process.pid}: #{range2.to_a.index(number)}" }Process.wait? time ruby parallel.rbParent 5086Child2 5100: 499999 99Child1 5099:ruby parallel.rb  3.73s user 0.43s system 192% cpu  2.162 total





  




 1  2  3  4  5  6  7  8  9 10 11 12 13 


range1 = 0...50_000_000range2 = 50_000_000...100_000_000number = 99_999_999puts "Parent #{Process.pid}"fork { puts "Child1 #{Process.pid}: #{range1.to_a.index(number)}" }fork { puts "Child2 #{Process.pid}: #{range2.to_a.index(number)}" }Process.wait? time ruby parallel.rbParent 5086Child2 5100: 49999999Child1 5099:ruby parallel.rb  3.73s user 0.43s system 192% cpu  2.162 total









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 


# threads.rb@executed = falsedef ensure_executed  unless @executed    puts "executing!"    @executed = true  endendthreads = 10.times.map { Thread.new { ensure_executed } }threads.each(&:join) # 主线程等待子线程执行完毕再继续往后面跑, 比如在后面 p 'done'.? ruby threads.rbexecuting!executing!executing!execu ting!









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 


# threads.rb@executed = falsedef ensure_executed  unless @executed    puts "executing!"    @execu ted = true  endendthreads = 10.times.map { Thread.new { ensure_executed } }threads.each(&:join) # 主线程等待子线程执行完毕再继续往后面跑, 比如在后面 p 'done'.? ruby threads.rbexecuting!executing!executing! executing!









1 2 3 4 5 6 7 8 


# sleep.rbthreads = 10.times.map do |i|  Thread.new { sleep 1 }endthreads.each(&:join)? time ruby sleep.rbruby sleep.rb  0.09s user 0.03s system 9% cpu  1.146 total









1 2 3 4 5 6 7 8 


# sleep.rbthreads = 10.times.map do |i|  Thread.new { sleep 1 }endthreads.each(&:join)? time ruby sleep.rbruby sleep.rb  0.09s user 0.03s system 9% cpu  1.146 total









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 


# em.rbrequire 'eventmachine'EM.run do  EM.add_timer(1) do    puts 'sleeping...'    EM.system('sleep 1') { puts "woke up!" }    puts 'continuing...'  end  EM.add_timer(3) { EM.stop }end? ruby em.rbsleeping...continuing...woke up!









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 


# em.rbrequire 'eventmachine'EM.run do  EM.add_timer(1) do    puts 'sleeping...'    EM.system('sleep 1') { puts "woke up!" }    puts 'continuing...'  end  EM.add_timer(3) { EM.stop }end? ruby em.rbsleeping...continuing...woke up!





  




 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 19 20 21 22 


# fiber.rbfiber = Fiber.new do  # 3. Fiber.yield 交出执行权, 并返回 1 (在这里)  Fiber.yield 1  # 5. 执行完毕, 返回 2  2end# 1. fiber 创建以后不会执行里边的代码# 2. 调用 resume 才会执行puts fiber.resume# 4. Go on here...puts fiber.resume# 5. fiber 已挂, 报错puts fiber.resume? ruby fiber.rb12dead fiber called (FiberError)









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 19 20 21 22 


# fiber.rbfiber = Fiber.new do  # 3. Fibe r.yield 交出执行权, 并返回 1 (在这里)  Fiber.yield 1  # 5. 执行完毕, 返回 2  2end# 1. fiber 创建以后不会执行里边的代码# 2. 调用 resume 才会执行puts fiber.resume# 4. Go on here...puts fiber.resume# 5. fiber 已挂, 报错puts fiber.resume? ruby fiber.rb12dead fiber calle d (FiberError)









 1  2  3  4  5  6  7  8  9 10 


EventMachine.run do  page = EM::HttpRequest.new('https://google.ca/').get  page.errback { puts "Google is down" }  page.callback {    url = 'https://google.ca/search?q=universe.com'    about = EM::HttpRequest.new(url).get    about.errback  { ... }    about.callback { ... }  }end









 1  2  3  4  5  6  7  8  9 10 


EventMachine.run do  page = EM::HttpRequest.new('https://google.ca/').get  page.errback { puts "Google is down" }  page.callback {    url = 'https://google.ca/search?q=universe.com'    about = EM::HttpRequest.new(url).get    about.errback  { ... }    about.callback { ... }  }end









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 


EventMachine.run do  Fiber.new {    page = http_get('http://www.google.com/')    if page.response_header.status == 200      about = http_get('https://google.ca/search?q=universe.com')      # ...    else      puts "Google is down"    end  }.resumeenddef http_get(url)  current_fiber = Fiber.current  http = EM::HttpRequest.new(url).get  http.callback { current_fiber.resume(http) }  http.errback  { current_fiber.resume(http) }  Fiber.yieldend









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 


EventMachine.run do  Fiber.new {    page = http_get('http://www.google.com/')    if page.response_header.status == 200      about = http_get('https://google.ca/search?q=universe.com')      # ...    else      puts "Google is down"    end  }.resumeenddef http_get(url)  current_fiber = Fiber.current  http = EM::HttpRequest.new(url).get  http.callback { current_fiber.resume(http) }  http.errback  { current_fiber.resume(http) }  Fiber.yieldend









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 


class Sheen  include Celluloid  def initialize(name)    @name = name  end  def set_status(status)    @status = status  end  def report    "#{@name} is #{@status}"  endendirb(main):009:0> charlie = Sheen.new "Charlie Sheen"=> #irb(main):010:0> charlie.set_status "winning!"=> "winning!"irb(main):011:0> charlie.report=> "Charlie Sheen is winning!"irb(main):012:0> charlie.async.set_status "asynchronously winning!" => #irb(main):013:0> charlie.report=> "Charlie Sheen is asynchronously winning!"









 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 


class Sheen  include Celluloid  def initialize(name)    @name = name  end  def set_status(status)    @status = status  end  def report    "#{@name} is #{@status}"  endendirb(main):009:0> charlie = Sheen.new "Charlie Sheen"=> #irb(main):010:0> charlie.set_status "winning!"=> "winning!"irb(main):011:0> charlie.report=> "Charlie Sheen is winning!"irb(main):012:0> charlie.async.set_status "asynchronously winning!"=> #irb(main):013:0> charlie.report=> "Charlie Sheen is asynchronously winning!"







 celluloid, concurrent, eventmachine, gil, process, ruby, thread



Post navigation
Ruby – Regular expression, look for strings that include but not starting with strings
Ruby-on-rails – RSPEC can output a value that makes test failure?




Leave a Comment Cancel reply
Your email address will not be published.