Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Concurrent Programming with Celluloid (MWRC 2012)

tarcieri
March 16, 2012

Concurrent Programming with Celluloid (MWRC 2012)

Threads versus events: which should you choose? How about both? In this talk you’ll learn about the Celluloid concurrency framework, which combines OOP and the Actor Model to give you concurrent Ruby objects. You’ll also learn about how Celluloid lets you combine blocking I/O and asynchronous evented I/O, offering you all the benefits of EventMachine without the restrictions of a single event loop. The talk will also provide a brief introduction to DCell, a distributed extension to Celluloid.

tarcieri

March 16, 2012
Tweet

More Decks by tarcieri

Other Decks in Technology

Transcript

  1. Parallel Blocking I/O But only one Ruby thread at a

    time :( YARV Threads != Multicore
  2. Sidekiq What if 1 Sidekiq process could do the work

    of 20 Resque processes? http://mperham.github.com/sidekiq/
  3. listener = Actor::TCP.listen(HOST, PORT, :filter => :line) puts "Listening on

    #{HOST}:#{PORT}" # The main loop handles incoming connections loop do # Spawn a new actor for each incoming connection Actor.spawn(listener.accept) do |sock| puts "#{sock.remote_addr}:#{sock.remote_port} connected" # Connection handshaking begin sock.write "Please enter a nickname:" nickname = sock.read server << T[:register, Actor.current, nickname] # Flip the socket into asynchronous "active" mode # This means the Actor can receive messages from # the socket alongside other events. sock.controller = Actor.current sock.active = :once # Main message loop loop do Actor.receive do |filter| filter.when(T[:tcp, sock]) do |_, _, message| server << T[:say, Actor.current, message] sock.active = :once end
  4. WAT

  5. listener = Actor::TCP.listen(HOST, PORT, :filter => :line) puts "Listening on

    #{HOST}:#{PORT}" # The main loop handles incoming connections loop do # Spawn a new actor for each incoming connection Actor.spawn(listener.accept) do |sock| puts "#{sock.remote_addr}:#{sock.remote_port} connected" # Connection handshaking begin sock.write "Please enter a nickname:" nickname = sock.read server << T[:register, Actor.current, nickname] # Flip the socket into asynchronous "active" mode # This means the Actor can receive messages from # the socket alongside other events. sock.controller = Actor.current sock.active = :once # Main message loop loop do Actor.receive do |filter| filter.when(T[:tcp, sock]) do |_, _, message| server << T[:say, Actor.current, message] sock.active = :once end
  6. WTF is T? filter.when(T[:tcp, sock]) do |_, _, message| server

    << T[:say, Actor.current, message] sock.active = :once end
  7. YES

  8. require 'thread' class ConcurrentNestedHash def initialize @outer = {} @mutex

    = Mutex.new end def [](*keys) @mutex.synchronize { keys.inject(@outer) { |h,k| h[k] } } end def []=(*args) @mutex.synchronize do value = args.pop raise ArgumentError, "wrong number of arguments (1 for 2)" if args.empty? key = args.pop hash = args.inject(@outer) { |h,k| h[k] ||= {} } hash[key] = value end end def inspect; @mutex.synchronize { super }; end end
  9. >> h = ConcurrentNestedHash.new => #<ConcurrentNestedHash:0x007f99ed735f08 @outer={}, @mutex=#<Mutex: 0x007f99ed735e90>> >>

    h[:foo, :bar, :baz] = 42 => 42 >> h => #<ConcurrentNestedHash:0x007f99ed735f08 @outer={:foo=>{:bar=>{:baz=>42}}}, @mutex=#<Mutex:0x007f99ed735e90>> >> h[:foo, :bar, :baz] => 42
  10. -require 'thread' +require 'celluloid' class ConcurrentNestedHash + include Celluloid def

    initialize @outer = {} - @mutex = Mutex.new end def [](*keys) - @mutex.synchronize { keys.inject(@outer) { |h,k| h[k] } } + keys.inject(@outer) { |h,k| h[k] } end def []=(*args) - @mutex.synchronize do value = args.pop raise ArgumentError, "wrong number of arguments (1 for 2)" if args.empty? key = args.pop hash = args.inject(@outer) { |h,k| h[k] ||= {} } hash[key] = value - end end - - def inspect; @mutex.synchronize { super }; end end
  11. require 'celluloid' class ConcurrentNestedHash include Celluloid def initialize @outer =

    {} end def [](*keys) keys.inject(@outer) { |h,k| h[k] } end def []=(*args) value = args.pop raise ArgumentError, "wrong number of arguments (1 for 2)" if args.empty? key = args.pop hash = args.inject(@outer) { |h,k| h[k] ||= {} } hash[key] = value end end
  12. “I thought of objects being like biological cells and/or individual

    computers on a network, only able to communicate with messages” - Alan Kay, creator of Smalltalk, on the meaning of "object oriented programming"
  13. =

  14. Concurrent Objects &DOOHU &HOOXORLG $FWRU3UR[\ &$// 5HFHLYHU &HOOXORLG 0DLOER[ 5(63216(

    &HOOXORLG 0DLOER[ &HOOXORLG$FWRU &HOOXORLG&DOO &HOOXORLG5HVSRQVH
  15. # Class methods added to classes which include Celluloid module

    ClassMethods # Create a new actor def new(*args, &block) proxy = Actor.new(allocate).proxy proxy._send_(:initialize, *args, &block) proxy end ...
  16. EXTREME Late Binding &DOOHU &HOOXORLG $FWRU3UR[\ &$// 5HFHLYHU &HOOXORLG 0DLOER[

    5(63216( &HOOXORLG 0DLOER[ &HOOXORLG$FWRU &HOOXORLG&DOO &HOOXORLG5HVSRQVH
  17. >> h = ConcurrentNestedHash.new => #<Celluloid::Actor(ConcurrentNestedHash:0x3ff3b952df7c) @outer={}> >> h.inspect =>

    “#<Celluloid::Actor(ConcurrentNestedHash:0x3ff3b952df7c) @outer={}>” Synchronous Calls
  18. Futures &DOOHU &HOOXORLG $FWRU3UR[\ 5HFHLYHU &HOOXORLG 0DLOER[ &HOOXORLG$FWRU &HOOXORLG&DOO &$//

    &HOOXORLG 0DLOER[ )8785( &HOOXORLG )XWXUH &HOOXORLG5HVSRQVH 9$/8(" 9$/8(
  19. >> h = ConcurrentNestedHash.new => #<Celluloid::Actor(ConcurrentNestedHash:0x3ff3b952df7c) @outer={}> >> future =

    h.future :inspect => #<Celluloid::Future:0x3ff3b953821a> >> 41 + 1 # roflscale computation => 42 >> future.value => “#<Celluloid::Actor(ConcurrentNestedHash:0x3ff3b952df7c) @outer={}>” Futures
  20. >> h = ConcurrentNestedHash.new => #<Celluloid::Actor(ConcurrentNestedHash:0x3ff3b952df7c) @outer={}> >> future =

    h.future :inspect => #<Celluloid::Future:0x3ff3b953821a> >> 41 + 1 # roflscale computation => 42 >> future.value => “#<Celluloid::Actor(ConcurrentNestedHash:0x3ff3b952df7c) @outer={}>” Futures
  21. require 'celluloid' class Person include Celluloid def name self.class.to_s end

    def greet(interested_party) "Hello #{interested_party.name}, I'm #{name}" end end class Joe < Person; end class Mike < Person def greet(other) super << "\n" << other.greet(current_actor) end end mike = Mike.new joe = Joe.new puts mike.greet(joe)
  22. require 'celluloid' class Person include Celluloid def name self.class.to_s end

    def greet(interested_party) "Hello #{interested_party.name}, I'm #{name}" end end class Joe < Person; end class Mike < Person def greet(other) super << "\n" << other.greet(current_actor) end end mike = Mike.new joe = Joe.new puts mike.greet(joe)
  23. 0LNH -RH 0LNHJUHHW -RHJUHHW 0LNHQDPH  J U H H

    W  + H OO R  0 L N H    J U H H W  0 L N H  6XVSHQGHG Multitasking Fibers
  24. Call Benchmark Core i7 2.0GHz (OS X 10.7.3) Rubinius YARV

    16815/s (60µs) - JRuby 1.6.7 20899/s (50µs) - JRuby HEAD 15260/s (65µs) - rbx HEAD 9367/s (107µs) - Ruby 1.9.3
  25. Fault Tolerance •Supervisors & Supervision Trees •“Fail early”, restart in

    a clean state •Do what Erlang does •No seriously, do what Erlang does
  26. Evented IO •Large numbers of connections (>1000) •Mostly idle connections

    •Mostly IO-bound problems •Websockets are an ideal case
  27. nio4r •Quasi-inspired by Java NIO •Smallest API possible •libev C

    extension for CRuby/rbx •Java extension for JRuby •Pure Ruby version too! http://github.com/tarcieri/nio4r/
  28. Celluloid::IO::TCPSocket •Uses fibered I/O •“Duck-type” of ::TCPSocket •Evented inside Celluloid::IO

    actors •Blocking IO elsewhere (Ruby Threads, normal Celluloid actors)
  29. class EchoServer include Celluloid::IO def initialize(host, port) puts "*** Starting

    echo server on #{host}:#{port}" # Since we included Celluloid::IO, we're actually making a # Celluloid::IO::TCPServer here @server = TCPServer.new(host, port) run! end def finalize @server.close if @server end def run loop { handle_connection! @server.accept } end def handle_connection(socket) _, port, host = socket.peeraddr puts "*** Received connection from #{host}:#{port}" loop { socket.write socket.readpartial(4096) } rescue EOFError puts "*** #{host}:#{port} disconnected" socket.close end end supervisor = EchoServer.supervise("127.0.0.1", 1234) trap("INT") { supervisor.terminate; exit } sleep
  30. Hello World Benchmark # httperf --num-conns=50 --num-calls=1000 Ruby Version Throughput

    Latency ------------ ---------- ------- JRuby HEAD 5650 reqs/s (0.2 ms/req) Ruby 1.9.3 5263 reqs/s (0.2 ms/req) JRuby 1.6.7 4303 reqs/s (0.2 ms/req) rbx HEAD 2288 reqs/s (0.4 ms/req)
  31. Hello World Comparison Web Server Throughput Latency ---------- ---------- -------

    Goliath (0.9.4) 2058 reqs/s (0.5 ms/req) Thin (1.2.11) 7502 reqs/s (0.1 ms/req) Node.js (0.6.5) 11735 reqs/s (0.1 ms/req)