RubyのFiberとEnumerator ― 実行中のブロックを裏返す

Rubyには、ループ、ブロック、エニュメレータなど、反復処理を行うためのさまざまな手段が用意されています。多くのRubyプログラマはループとブロックには慣れ親しんでいるものの、EnumeratorやFiberはあまり深く理解されていないことが多いでしょう。今回のRuby Magicでは、ゲスト著者のJulikがEnumerableとFiberに光を当て、フロー制御可能な列挙処理と「ブロックを裏返す」テクニックを解説します。

ブロックの一時停止と連鎖的な反復処理

以前のRuby MagicではEnumeratorについて取り上げ、独自の#eachメソッドからEnumeratorを返す方法とその活用方法を説明しました。EnumeratorとFiberのさらに強力なユースケースは、実行中のブロックを途中で「一時停止」できるという点です。#eachに渡されたブロックだけではなく、任意のブロックを対象にできます。

これは非常に強力な構造で、ブロックを使って動作するメソッドと、ブロックではなく逐次的な呼び出しを期待する呼び出し元との橋渡しとなるラッパー(shim)を実装するのに使えます。例えば、データベースハンドルを開いて、取得した行を1行ずつ読み出したい状況を想像してみてください。

db.with_each_row_of_result(sql_stmt) do |row|
  yield row
end

ブロックAPIは優れた設計です。ブロックが終了するタイミングで、リソース解放などの各種クリーンアップを自動的に実行してくれる可能性があるからです。しかし、利用者の中には次のようにデータベースを扱いたい人もいるかもしれません。

@cursor = cursor
 
# 後ほど:
row = @cursor.next_row
send_row_to_event_stream(row)

つまり、ブロックの実行を「今のところ」一時停止し、後でブロック内の続きから再開したいわけです。こうすることで、フロー制御の主導権が、ブロックを実行するメソッド(callee)の手から呼び出し元(caller)の手へと移ります。

イテレータの連鎖

このパターンで最もよく使われるのが、複数のイテレータを連鎖させる手法です。連鎖させると、普段使っている反復用メソッド(#eachなど)は代わりにEnumeratorオブジェクトを返すようになり、それを使ってyieldステートメントが送り出す値を「受け取れる」ようになります。

range = 1..8
each_enum = range.each # => <Enumerator...>

Enumerator同士は連鎖でき、「任意の反復処理をインデックス付きで行う」といった操作が可能になります。次の例では、Rangeに対して#mapを呼び出してEnumerableオブジェクトを取得し、さらに#with_indexを連鎖させてインデックス付きで範囲を反復しています。

(1..3).map.with_index {|element_n, index| [element_n, index] }
#=> [[1, 0], [2, 1], [3, 2]]

これはイベント駆動のシステムでは特に便利です。Rubyには、任意のメソッドをEnumerator生成用にラップする組み込みメソッドが用意されており、まさにこの目的を簡単に実現できます。例えば、with_each_row_of_resultからyieldされるのを待つのではなく、行を1行ずつ「引き出す(pull)」形に変えたいとしましょう。

@cursor = db.to_enum(:with_each_row_of_result, sql_stmt)
schedule_for_later do
  begin
    row = @cursor.next
    send_row_to_event_stream(row)
  rescue StopIteration # ブロックが終了しカーソルが空になった。クリーンアップも完了している
  end
end

これを自前で実装するとすれば、おそらく次のようになるでしょう。

cursor = Enumerator.new do |yielder|
  db.with_each_row_of_result(sql_stmt) do |row|
    yielder.yield row
  end
end

ブロックを裏返す

Railsでは、レスポンスボディとしてEnumeratorを割り当てることもできます。Railsは、レスポンスボディとして設定されたEnumeratorに対してnextを呼び出し、戻り値が文字列であることを期待します。その文字列はRackレスポンスへ書き出されます。例えば、Rangeの#eachの戻り値をそのままRailsのレスポンスボディとして返せます。

class MyController < ApplicationController
  def index
    response.body = ('a'..'z').each
  end
end

筆者はこれを「ブロックを裏返す」と呼んでいます。本質的には、ブロック(Rubyにおいてループもブロックの一種です)の実行途中で「時間を凍結」できるようにする、制御フローのためのヘルパーなのです。

ただし、Enumeratorには有用性を少し損なう制約があります。例えば、次のようなコードを書きたい場面を考えてみましょう。

File.open('output.tmp', 'wb') do |f|
  # 書き込み用にファイルをyieldし、継続的に使う
  loop { yield(f) }
end

これをEnumeratorで包んで書き込んでみます。

writer_enum = File.to_enum(:open, 'output.tmp', 'wb')
file = en.next
file << data
file << more_data

動作自体は問題ありません。しかし、ここで壁に突き当たります。書き込みが完了したことをenumeratorに伝え、ブロックを「完走」させてファイルを閉じ、終了させるにはどうすればよいのでしょうか? ファイルを閉じる処理では、リソースのクリーンアップや、バッファリングされた書き込み内容のディスクへのフラッシュといった重要なステップが実行されます。Fileオブジェクトには直接アクセスできるため自分で閉じることもできますが、本来はenumerator側にクローズを任せたいものです。そのためには、enumeratorにブロックの先へ進んでもらう必要があります。

もうひとつの課題は、一時停止中のブロック内の状態を引数として渡したい場合があることです。例えば、次のようなセマンティクスを持つブロックを受け取るメソッドがあるとします。

write_file_through_encryptor(file_name) do |writable|
  writable << "Some data"
  writable << "Some more data"
  writable << "Even more data"
end

しかし呼び出し側では、次のように使いたいのです。

writable = write_file_through_encryptor(file_name)
writable << "Some data"
# ...その後
writable << "Some more data"
writable.finish

理想的には、メソッド呼び出しを何らかの構造でラップして、次のようなトリックが実現できるようにしたいところです。

write_file_through_encryptor(file_name) do |writable|
  loop do
    yield_and_wait_for_next_call(writable)
    # そして、なんらかの方法でループを抜けてブロックを完了させる
  end
end

では、書き込みをこのようにラップしてみたらどうでしょうか?

deferred_writable = write_file_through_encryptor(file_name)
deferred_writable.next("Some data")
deferred_writable.next("Some more data")
deferred_writable.next("Even more data")
deferred_writable.next(:terminate)

この場合、:terminateをマジック値として使い、「ブロックを完了してreturnしてよい」ことをメソッドに伝えます。ここで、Enumeratorは役に立ちません。Enumerator#nextには引数を渡せないからです。もし渡せれば、次のように書けるはずでした。

deferred_writable = write_file_through_encryptor(file_name)
deferred_writable.next("Some data")
...
deferred_writable.next(:terminate)

Rubyのファイバーの登場

まさにこれこそが、ファイバー(Fiber)が実現してくれることです。ファイバーは再開のたびに引数を受け取れるため、ラッパーを次のように実装できます。

deferred_writable = Fiber.new do |data_to_write_or_termination|
  write_file_through_encryptor(filename) do |f|
     # ここでファイバーのブロックコンテキストに入る。再開時はこのブロックの先頭から
    loop do
      # Fiber.yieldを呼ぶとファイバーは一時停止する。
      # 「data_to_write_or_termination = 」の代入は、ファイバーが再開されるまで実行されない
      data_to_write_or_termination = Fiber.yield
    end
  end
end

仕組みはこうです。最初にdeferred_writableに対して.resumeを呼ぶと、ファイバーに入り、最初のFiber.yieldステートメント、あるいは最も外側のFiberブロックの終端のうち、先に来た方まで実行が進みます。Fiber.yieldが呼ばれると制御が手元に戻ります。Enumeratorのときと同じく、ブロックは一時停止されます。そして次に.resumeを呼ぶと、その引数が新しいdata_to_writeになります。

deferred_writes = Fiber.new do |data_to_write|
  loop do
    $stderr.puts "Received #{data_to_write} to work with"
    data_to_write = Fiber.yield
  end
end
# => #<Fiber:0x007f9f531783e8>
deferred_writes.resume("Hello") #=> Received Hello to work with
deferred_writes.resume("Goodbye") #=> Received Goodbye to work with

つまりファイバー内部では、コードフローは最初のFiber#resume呼び出しで開始され、最初のFiber.yield呼び出しで一時停止し、以降のFiber#resume呼び出しで続行されます。このときFiber.yieldの戻り値が、resumeに渡した引数となります。コードは、Fiber.yieldが最後に呼ばれた地点から再開されます。

ここにはファイバー特有の癖がひとつあります。ファイバーへの初期引数は、Fiber.yieldの戻り値としてではなく、ブロック引数として渡されるという点です。

この挙動を踏まえると、resumeに特別な引数を渡すことで、ファイバー内で停止すべきかどうかを判断できることが分かります。実際に試してみましょう。

deferred_writes = Fiber.new do |data_to_write|
  loop do
    $stderr.puts "Received #{data_to_write} to work with"
    break if data_to_write == :terminate # ループを抜ける、あるいは...
    write_to_output(data_to_write)       # ...出力へ書き込む
    data_to_write = Fiber.yield          # 自分自身を一時停止し、次のresumeを待つ
  end
  # 上のループを抜けるとここに到達する。以降にFiber.yieldはないため、
  # ファイバーは終了し「dead」状態になる
end
 
deferred_writes.resume("Hello") #=> Received Hello to work with
deferred_writes.resume("Goodbye") #=> Received Goodbye to work with
deferred_writes.resume(:terminate)
deferred_writes.resume("Some more data after close") # FiberError: dead fiber called

こうした仕組みが威力を発揮する場面は多数あります。ファイバーは手動で再開できる中断済みのコード断片を保持できるため、イベントリアクタの実装や、単一スレッド内での並行処理の管理に利用できます。しかも軽量なので、1クライアントに1つのファイバーを割り当て、必要に応じてファイバー間を切り替えることで、ファイバーだけでサーバーを構築することも可能です。

client_fiber = Fiber.new do |socket|
   loop do
     received_from_client = socket.read_nonblock(10)
     sent_to_client = socket.write_nonblock("OK")
     Fiber.yield # 制御を呼び出し元に戻し、resumeされるのを待つ
   end
end
 
client_fibers << client_fiber
 
# メインのWebサーバーループ内で
client_fibers.each do |client_fiber|
  client_fiber.resume # クライアントからのデータを受信し、OKを送信
end

さらにRubyにはfiberという標準ライブラリが用意されており、あるファイバーから別のファイバーへ明示的に制御を移譲できます。上述のような用途では、これが追加の武器になります。

データ出力レートの制御

ファイバーとエニュメレータのもうひとつの優れた使い道は、Rubyのブロックがデータを出力するペースを制御したい場合に生まれます。例えばzip_tricksでは、ライブラリの基本的な使い方として次のようなブロック形式をサポートしています。

ZipTricks::Streamer.open(output_io) do |z|
  z.write_deflated_file("big.csv") do |destination|
   columns.each do |col|
     destination << column
   end
  end
end

この方式では、ZIPアーカイブを生成する側のコードが「プッシュ」型の主導権を持つため、どれだけのデータを、どの頻度で出力するかを制御できません。例えば5MBずつZIPを書き出したい場合(AWS S3オブジェクトストレージの制限など)、セグメントをS3のマルチパートパートとして分割すべきタイミングで<<メソッド呼び出しを「拒否する」カスタムoutput_ioオブジェクトを用意する必要があります。そこで、制御を逆転させて「プル」型に変えます。大きなCSVファイルの書き込みには同じブロックを使いながら、出力される内容に応じてブロックを再開・停止させるのです。これにより、次のような使い方が可能になります。

output_enum = ZipTricks::Streamer.output_enum do |z|
  z.write_deflated_file("big.csv") do |destination|
   columns.each do |col|
     destination << column
   end
  end
end
 
# この時点ではまだ何も生成も書き込みもされていない
enum = output_enum.each # Enumeratorを生成
bin_str = enum.next # ブロックにバイナリデータを生成させ、その場で一時停止させる
output.write(bin_str) # ブロックは一時停止しており、次のnext呼び出しを待っている

これにより、ZIPファイルジェネレータがデータを出力するペースを自由に制御できるようになります。

つまり、EnumeratorとFiberとは、「プッシュ」型のブロックを、メソッド呼び出しを受け付ける「プル」型のオブジェクトへと変換するための制御フローの仕組みなのです。

ファイバーとエニュメレータには落とし穴がひとつあります。ブロック内にensureがある場合や、ブロック完了後に実行すべき処理がある場合、その処理が確実に実行されるかどうかは呼び出し元が十分な回数呼び出してくれるかに委ねられます。ある意味で、JavaScriptのPromiseを使う際の制約に似ていると言えるでしょう。

まとめ

以上で、Rubyにおけるフロー制御可能な列挙処理についての考察は終わりです。この過程でJulikは、EnumerableクラスとFiberクラスの共通点と相違点を明らかにし、呼び出し元がデータの流れを主導する実例を掘り下げました。さらに、ブロックの再入のたびに引数を渡せるというFiberならではの魔法についても学びました。それでは、楽しいフロー制御ライフを!

魔法の定期的な供給をお求めの方は、ぜひRuby Magicを購読してください。毎月の号をあなたの受信箱へ直接お届けします。