Ruby EventMachine and fork

README

Ruby の fork, thread と EventMachine gem を使ったプログラムの例

EventMachine は、 select(2) だか epoll(2) だかを使って非同期処理をしてくれるライブラリ。 C++ で書かれててはやい。

EventMachine.start に host, port と、 EventMachine::Connection を継承したクラスか任意のモジュールを指定する。クラス名を与えるとそのクラスのインスタンスがリクエストを受ける毎に作られる。必要なメソッドをオーバーライドして実装する。モジュール名を与えると、 EventMachine::Connection に MixIn される。こちらもメソッドをオーバーライドしてやる。

fork は Kernel#fork と Process#fork がある。 Process はモジュールでインスタンス化しない。どう違うのかよくわからない。 fork do ... end という書き方もできて、子プロセス毎にブロック内を実行して終了するみたい。これはわかりやすい。 $$ が pid になってる。

Thread は例だけ書いてみたけど、順番に実行しちゃうだけだしぜんぜんわかってない。

# 例の実行
$ ruby event.rb

# 別窓で
$ ruby fork.rb

fork do ... end ブロックで適当に sleep してるので、子プロセスが適当な順番で接続してサーバが動いてるのがデモれる。でもこれが本当の意味で非同期っていうか、複数のクライアントからの同時接続をちゃんとさばけるもんなのか自信ない。

event.rb

#!/opt/local/bin/ruby
require 'rubygems'
require 'eventmachine'

class CharacterCount < EventMachine::Connection
  def post_init
    puts "==> Received a new connection"
  end

  def receive_data(data)
    puts "==> Receive_data data: #{data}"
    send_data "#{data.length} (characters)\r\n"
    close_connection_after_writing
  end
end

EventMachine::run do
  host, port = "127.0.0.1", 3793
  EventMachine.start_server(host, port, CharacterCount)
  puts "==> Now accepting connections on address #{host}, port #{port}"
end

event2.rb

#!/opt/local/bin/ruby
require 'rubygems'
require 'eventmachine'

EventMachine::run do
  host, port = "127.0.0.1", 3793
  EventMachine.start_server(host, port, EventMachine::Connection)
  puts "==> Now accepting connections on address #{host}, port #{port}"
end

fork.rb

#!/opt/local/bin/ruby
require 'socket'

host = '127.0.0.1'
port = 3793

10.times do
  fork do
    s = TCPSocket.open(host, port)
    elapse = $$.to_f % 3 / 10
    sleep elapse
    s.send "send data from #{$$}, sleep #{elapse}", 0
    s.close
  end
end

thread.rb

#!/opt/local/bin/ruby
puts "Test start"
puts "Create thread"
t = Thread.new do
  puts "Start thread"
  sleep 0.5
  puts "End thread"
end

puts "Waiting for the thread to complete"
t.join

puts "Test completed"

thread2.rb

#!/opt/local/bin/ruby
require 'socket'

host = '127.0.0.1'
port = 3793

100.times do
  t = Thread.new do
    sleep 0.5
    s = TCPSocket.open(host, port)
    s.send("send data from #{$$}", 0)
    s.close
  end
  t.join
end