Rubyのファイバーと列挙子-ブロックを裏返しにする

Rubyには、ループ、ブロック、列挙子など、反復を実行するさまざまな方法があります。ほとんどのRubyプログラマーは、少なくともループとブロックに精通していますが、Enumerator およびFiber 多くの場合、暗闇にとどまります。今回のRubyMagicでは、ゲスト著者のJulikがEnumerableに光を当てています。 およびFiber フロー制御の列挙型と回転ブロックを裏返しに説明します。

ブロックの一時停止と連鎖反復

以前の版のRubyMagicで列挙子について説明しました。ここでは、Enumeratorを返す方法について説明しました。 自分の#eachから メソッドとその用途。 Enumeratorのさらに広範なユースケース およびFiber 飛行中に「ブロックを一時停止」できるということです。 #eachに与えられたブロックだけではありません または#eachへの呼び出し全体 、しかし任意のブロック！

これは非常に強力な構造であり、ブロックを取得する代わりに順次呼び出しを期待する呼び出し元へのブリッジとしてブロックを使用することによって機能するメソッドのシムを実装するために使用できます。たとえば、データベースハンドルを開いて、取得した各アイテムを読み取りたいとします。

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

ブロックAPIは、ブロックが終了したときにあらゆる種類のクリーンアップを実行する可能性があるため、優れています。ただし、一部のコンシューマーは、次の方法でデータベースを操作したい場合があります。

@cursor = cursor
 
# later:
row = @cursor.next_row
send_row_to_event_stream(row)

実際には、ブロックの実行を「今のところ」「一時停止」し、後でブロック内で続行することを意味します。したがって、呼び出し元は、呼び出し先（ブロックを実行するメソッド）の手にあるのではなく、フロー制御を引き継ぎます。

イテレータの連鎖

このパターンの最も一般的な使用法の1つは、複数のイテレーターをチェーン化することです。そうすると、反復に使用されるメソッド（#eachなど） ）、代わりに列挙子オブジェクトを返します。これを使用して、ブロックがyieldを使用して送信する値を「取得」できます。 ステートメント：

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

その後、列挙子を連鎖できます。 これにより、「インデックスを使用する以外の任意の反復」のような操作を実行できます。この例では、#mapを呼び出しています。 Enumerableを取得する範囲で 物体。次に、#with_indexをチェーンします インデックスを使用して範囲を反復処理するには：

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

これは、特にシステムでイベントを使用する場合に非常に役立ちます。 Rubyには、任意のメソッドを列挙子ジェネレーターでラップするための組み込みメソッドが用意されています。これにより、これを正確に実行できます。 with_each_row_of_resultから行を1つずつ「プル」したいとします。 、私たちにそれらをもたらす方法の代わりに。

@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 # the block has ended and the cursor is empty, the cleanup has taken place
  end
end

これを自分たちで実装した場合、次のようになります。

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

ブロックを裏返しにする

Railsを使用すると、応答本文を列挙子としても割り当てることができます。 nextを呼び出します 列挙子では、応答の本文として割り当て、戻り値が文字列であると想定します。これは、Rack応答に書き出されます。たとえば、#eachへの呼び出しを返すことができます Rails応答本体としてのRangeのメソッド：

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

これは私がブロックを裏返しにすると呼んでいるものです。 本質的に、これは、飛行中のブロック（またはRubyのブロックでもあるループ）で「時間をフリーズ」できるようにする制御フローヘルパーです。

ただし、列挙子には制限プロパティがあり、少し役に立たなくなります。次のようなことをしたいとします：

File.open('output.tmp', 'wb') do |f|
  # Yield file for writing, continuously
  loop { yield(f) }
end

列挙子でラップして、書き込みましょう

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

すべてがうまく機能します。ただし、問題があります。ブロックを「終了」し、ファイルを閉じて終了できるように、書き込みが完了したことを列挙子にどのように伝えるのでしょうか。これにより、リソースのクリーンアップ（ファイルが閉じられる）や、バッファリングされたすべての書き込みがディスクにフラッシュされるようにするなど、いくつかの重要な手順が実行されます。 Fileにアクセスできます 異議を唱え、自分で閉じることはできますが、列挙子に閉じを管理してもらいたいと思います。列挙子をブロックを通過させる必要があります。

もう1つのハードルは、中断されたブロック内で何が起こっているかについての議論を渡したい場合があることです。次のセマンティクスを持つブロック受け入れメソッドがあると想像してください。

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"
# ...later on
writable << "Some more data"
writable.finish

理想的には、メソッド呼び出しを、次のトリックを可能にする構造にラップします。

write_file_through_encryptor(file_name) do |writable|
  loop do
    yield_and_wait_for_next_call(writable)
    # Then we somehow break out of this loop to let the block complete
  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を使用します ブロックを終了して戻ることができることをメソッドに通知する魔法の値として。 これ ここでEnumerator Enumerator#nextに引数を渡すことができないため、実際には役に立ちません。 。できれば、次のことができるでしょう：

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

Rubyのファイバーを入力してください

これはまさにファイバーが許可するものです。ファイバーを使用すると、再入力ごとに引数を受け入れることができます。 、次のようにラッパーを実装できます：

deferred_writable = Fiber.new do |data_to_write_or_termination|
  write_file_through_encryptor(filename) do |f|
     # Here we enter the block context of the fiber, reentry will be to the start of this block
    loop do
      # When we call Fiber.yield our fiber will be suspended—we won't reach the
      # "data_to_write_or_termination = " assignment before our fiber gets resumed
      data_to_write_or_termination = Fiber.yield
    end
  end
end

仕組みは次のとおりです。最初に.resumeを呼び出したとき deferred_writableで 、それはファイバーに入り、最初のFiber.yieldまでずっと行きます ステートメントまたは最も外側のファイバーブロックの最後のいずれか早い方。 Fiber.yieldを呼び出すとき 、それはあなたに制御を戻します。列挙子を覚えていますか？ブロックは一時停止されます 、次に.resumeを呼び出すとき 、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 # Break out of the loop, or...
    write_to_output(data_to_write)       # ...write to the output
    data_to_write = Fiber.yield          # suspend ourselves and wait for the next `resume`
  end
  # We end up here if we break out of the loop above. There is no Fiber.yield
  # statement anywhere, so the Fiber will terminate and become "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

これらの施設が非常に役立つ状況はたくさんあります。ファイバーには、手動で再開できる中断されたコードブロックが含まれているため、ファイバーは、イベントリアクターの実装、および単一スレッド内での同時操作の処理に使用できます。これらは軽量であるため、単一のクライアントを単一のファイバーに割り当て、必要に応じてこれらのファイバーオブジェクトを切り替えることで、ファイバーを使用してサーバーを実装できます。

client_fiber = Fiber.new do |socket|
   loop do
     received_from_client = socket.read_nonblock(10)
     sent_to_client = socket.write_nonblock("OK")
     Fiber.yield # Return control back to the caller and wait for it to call 'resume' on us
   end
end
 
client_fibers << client_fiber
 
# and then in your main webserver loop
client_fibers.each do |client_fiber|
  client_fiber.resume # Receive data from the client if any, and send it an OK
end

Rubyには、fiberと呼ばれる追加の標準ライブラリがあります。 これにより、あるファイバーから別のファイバーに制御を明示的に移すことができます。これは、これらの用途のボーナス機能になります。

データ排出率の管理

Rubyブロックがデータを出力する速度を制御できるようにしたい場合は、ファイバーと列挙子のもう1つの優れた用途があります。たとえば、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アーカイブを作成するコードの部分で「プッシュ」制御を許可し、出力するデータの量と頻度を制御することは不可能です。たとえば、5 MBのチャンクでZIPを記述したい場合（これはAWS S3オブジェクトストレージの制限になります）、カスタムのoutput_ioを作成する必要があります。 どういうわけか<<を受け入れることを「拒否」するオブジェクト セグメントをS3マルチパート部分に分割する必要がある場合にメソッドが呼び出されます。ただし、コントロールを反転して「プル」することはできます。大きな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
 
# At this point nothing has been generated or written yet
enum = output_enum.each # Create an Enumerator
bin_str = enum.next # Let the block generate some binary data and then suspend it
output.write(bin_str) # Our block is suspended and waiting for the next invocation of `next`

これにより、ZIPファイルジェネレータがデータを送信する速度を制御できます。

したがって、列挙子とファイバーは制御フローメカニズム 「プッシュ」ブロックを、メソッド呼び出しを受け入れる「プル」オブジェクトに変換します。

ファイバーと列挙子の落とし穴は1つだけです。ensureのようなものがある場合 あなたのブロック、またはブロックが完了した後に実行する必要がある何かで、今では十分な回数あなたに電話をかけるのは発信者次第です。ある意味で、JavaScriptでPromisesを使用するときの制約に匹敵します。

結論

これで、Rubyのフロー制御された列挙型についての調査は終わりです。その過程で、ジュリックはEnumerableの類似点と相違点に光を当てました。 およびFiber クラス、および呼び出し元がデータのフローを決定した例に飛び込みます。 Fiberについても学びました 各ブロックの再エントリで引数を渡すことができるようにするの追加の魔法。ハッピーフローコントロール！

安定した魔法を手に入れるには、Ruby Magicに登録してください。月刊版が、受信トレイに直接配信されます。