オブジェクトをそのままで送受信するPipe

Ruby

IPCの中でもPipeはTCPとかのネットワーク系の基礎になったもので、結構使いやすくて個人的に好きな機能です。
特にスレッドを使うと、スレッド間通信用に多用します。
自分でリングバッファとか作るのも1つの手なのですが、自前でこういうの作るとブロッキングが出来なくsleepとかいれざるを得ないのでちょっと厄介なので特別な理由がない場合使いません。


さて、これを今回Rubyで使おうと思ったのですがRubyの場合IO#pipeで作りwrite/readを使うので文字列でしか通信できないのがネック。
送受信したいのはあくまでもRubyのオブジェクト。Cだと、どうせ同一プロセス内なので型を統一してポインタをwrite、readしたらキャストすればOKなのですがこの技が使えない。
http://docs.ruby-lang.org/ja/2.1.0/class/IO.html#S_PIPE
http://docs.ruby-lang.org/ja/2.1.0/class/IO.html#I_WRITE
http://docs.ruby-lang.org/ja/2.1.0/class/IO.html#I_READ


諦めるのも嫌だし仕方がないので、悩んだのですがMarshalを使うことにして動作するのを確認しました。
http://docs.ruby-lang.org/ja/2.1.0/class/Marshal.html

class ObjctPipe   
  def initialize  
    pipe = IO.pipe
    @rp = pipe[0]
    @wp = pipe[1]
  end

  def write(obj)
    objs = Marshal.dump(obj)

    len = objs.length
    lens = [len].pack('L!')
    @wp.syswrite(lens)

    @wp.syswrite(objs)
  end

  def read
    lens = @rp.sysread(8)
    len = lens.unpack('L!')[0]
    objs = @rp.sysread(len) 
    obj = Marshal.load(objs)
    return obj
  end
end


# 使い方はこうなります
pipe = ObjctPipe.new

# read側はスレッドで行う
Thread.new do
  while true
    obj = pipe.read
  end
end

# write側
while true
  # データを作成しておく
  pipe.write("なにかobject")
end


いくつかポイント

  • write側スレッドとread側スレッドで同じ'pipe'オブジェクトを使いますが、Mutexは使っていない
    • これは、実質同じデータにアクセスしないのでスレッド特有の問題は発生しないからです
  • 送信するobjectをMarshalしておく
    • 当然Marshalの制限がかかります
  • objectを送信する前にオブジェクトサイズを送る
    • サイズはpackで'unsigned long'にしておき、'オブジェクトサイズ'自体の大きさを固定しておく
    • これはread時のバッファやブロッキング動作が絡むからです
  • write/readではなくsyswrite/sysreadを使う
    • これは文字コードが絡んできてデータによってはwriteでエラーになるから


Rubyってobj.__id__でオブジェクトIDは作れるのですが、IDからオブジェクトの復活方法が分からないのでこの方法を取りました。
Marshalやpack/unpackが入るのでCに比べて無駄が多いのですが、まあ許容範囲内でしょう。
意外と便利に使えています。