takuma104のRuby/Rails日記

Ruby/EventMachine での非同期サーバを Generator を使って同期サーバっぽく書く

ruby, eventmachine | 04:24 |

Ruby/EventMachine という、非同期サーバがお手軽に書けるライブラリを教えてもらいました。ほかにも、Rev というのもあるようですが、今回試したのは、EventMachine のほう。Ruby/EventMachine は、現在の実装だと select(2) に落ちているようです。Rev はすでに epoll(2) 実装になっているようです。

まず、EventMachine のインストールは、

$ sudo gem install eventmachine

とかでいけます。ネイティブエクステンションとかも勝手にコンパイルとかしてくれます。gems 便利！

簡単な使い方は、Young risk taker.: [Ruby] Ruby/EventMachineでネットワークプログラミング などがあります。このサンプルがちょっと間違っててそのまま動かないので、修正して必要最小限にしたのは以下のようになります。

require 'rubygems'
require 'eventmachine'

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

  def receive_data(data)
    puts "Received 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

これだけで 127.0.0.1 の Port 3793 で、listen を開始して、Ctrl-C とかするまで動き続けるサーバーが出来ます。telnet とかで、3793 に接続して文字入力して改行を押すと、文字数とかが返ってくると思います。

CharacterCount クラスが、EventMachine::Connection クラスを継承していて、EventMachine.start_server にて登録されています。この CharacterCount のインスタンスが、TCPコネクションが張られるたびに自動生成されるようです。EventMachine::Connection のメソッドは、

• post_init が、コネクション貼られたときに呼び出される
• receive_data がデータを受信した時に呼び出される
• データ送信は、send_data です。

のようになります。で、例えば同期サーバ (TCPServer) で以下のような処理を書きたい場合に、

require "socket"

port = 3793
gs = TCPServer.open(port)

loop do
  Thread.start(gs.accept) do |s|
    puts "Now accepting connections on address port #{port}"
    puts s.read(10)
    puts "---"
    puts s.read(20)
    puts "---END"
    s.write(">> OK\r\n")
    s.close
  end
end

これをそのままEventMachineのような非同期サーバにそのまま変更しようと思うと、結構大変です。非同期サーバの作り方として、よくありがちなのが、状態遷移マシンを作って、例えばある状態の時は、10バイトまで読み続けて、ある状態になったら、応答を返してとかですかね。でもそれだと、このまま同期サーバーを非同期サーバに移植しようと思うと大変で気が滅入ります。

ということで、Ruby1.8におけるFiber (Co-routine, microthreadなどの呼び名がありますが、基本思想的には同じものです) 機構の代替えになるような Generator を使って、同期サーバーっぽく書けるようにしてみました。まだまだ改善余地があると思いますが、とりあえず、これで動作しています。

reader_setup の引数ブロックが、Generator 内部になっているので、reader メソッドで、上記の同期サーバーとほぼ同じ書き方が出来ています。*1

ちなみに、Generator は、遅いそうなので、これではあまり高速という話にはならないかもしれません。（ちゃんと計測したりとかはまだしてません）Ruby 1.9 の Fiber でも同様なことは書けると思いますので、速さを求めるなら、Ruby 1.9 の Fiber でも良いかもしれません。

#!/usr/bin/env ruby -KU

require 'rubygems'
require 'eventmachine'
require 'socket'
require 'generator'

class GeneratorSampleConnection < EventMachine::Connection
  def initialize
    @recv_data = ''
    @reader_generator = nil
  end
  
  def post_init
    addr = Socket.unpack_sockaddr_in(get_peername)
    puts "Connected from %s:%d" % [addr[1],addr[0]]
    reader_setup { reader }
  end

  def receive_data(data)
    push(data)
  end

private
  def reader
    puts read_bytes(10)
    puts "---"
    puts read_bytes(20)
    puts "---END"
    send_data ">> OK\r\n"
    close_connection_after_writing
  end

private
  def reader_setup
    Generator.new do |g|
      @reader_generator = g
      yield
    end
  end

  def push(data)
    @recv_data << data
    if @reader_generator.next?
      @reader_generator.next
    end
  end
  
  def pop(size)
    if @recv_data.size >= size
      r = @recv_data[0..size-1]
      @recv_data = @recv_data[size..-1]
      r
    else
      nil
    end
  end  

  def read_bytes(size)
    loop do
      d = pop(size)
      if d
        return d
      else
        remain = size - @recv_data.size
        @reader_generator.yield(remain)
      end
    end
  end
end

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