เชี่ยวชาญ Fibers และ Enumerators ใน Ruby: Advanced Flow Control Techniques

Ruby 有各種執行 iteration 的方式 — loops, blocks 和 enumerators 大多數 Ruby programmers 至少熟悉 loops 和 blocks 但 Enumerator 和 Fiber 經常停留在黑暗中 在這個 Ruby Magic 版本 guest author Julik 發光於 Enumerable 和 Fiber 解釋 flow controlling enumerables 和 turning blocks inside out

Suspending Blocks and Chained Iteration

我們在之前的 Ruby Magic 版本討論過 Enumerator 在那裡我們描述了如何從你自己的 #each 方法返回 Enumerator 和它可以用於什麼 Enumerator 和 Fiber 甚至更廣泛的用例是它們可以 "suspend a block" mid-flight 不僅僅是給 #each 的塊或整個 #each 調用 而是任何塊!

這是一個非常強大的構造 可用於實現 shims 為那些通過使用 blocks 作為橋樑來調用期望 sequential calls 而不是接受塊的 callers 的方法 例如 想像我們想打開 database handle 並讀取我們檢索到的每個 item:

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

Block API 很棒 因為當塊終止時 它可能為我們執行各種 cleanup 然而 某些 consumers 可能想以這種方式與 database 工作:

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

實際上 這意味著我們想 "suspend" 塊的執行 "just for now" 並在塊內稍後繼續 因此 caller 接管 flow control 而不是在 callee (執行塊的方法) 手中

Chaining Iterators

這個 pattern 最常見的用途之一是將多個 iterators chain 在一起 當我們這樣做時 我們習慣用於 iteration 的方法 (像 #each) 返回一個 Enumerator object 我們可以用它來 "grab" 塊通過 yield 語句發送給我們的值:

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

Enumerators 然後可以被 chained 這允許我們執行像 "任何 iteration 但帶有 index" 這樣的操作 在這個例子中 我們在 range 上調用 #map 獲取一個 Enumerable object 然後 chain #with_index 以 index 迭代 range:

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

如果你的系統使用 events 這可能非常有用 Ruby 提供一個 built-in method 用於 wrapping 任何方法與 Enumerator generator 這允許我們精確完成這一點 想像我們想從 with_each_row_of_result "pull" rows 一個接一個 而不是方法 yielding 它們給我們

@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

Turning Blocks Inside Out

Rails 允許我們將 response body 分配為也可以是 Enumerator 它將在作為 response body 分配的 Enumerator 上調用 next 並期望返回值是一個字符串 — 將被寫入 Rack response 例如 我們可以返回 Range 的 #each 方法調用作為 Rails response body:

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

這就是我所謂的 turning a block inside out. 本質上 它是一個 control flow helper 允許我們在塊中 "freeze time" (或循環 在 Ruby 中也是一個塊) mid-flight

然而 Enumerators 有一個限制性質 使它們稍微不那麼有用 想像我們想做像這樣的事:

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

讓我們用 enumerator wrap 它 並寫入它

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

一切運作良好 然而 有一個 hitch — 我們如何告訴 enumerator 我們寫完了 以便它可以 "finish" 塊 關閉文件並退出? 這將執行一些重要步驟 — 例如 resource cleanup (文件將被關閉) 以及確保所有 buffered writes 被 flushed 到磁盤 我們確實有 File 對象的訪問權 並可以自己關閉它 但我們希望 enumerator 為我們管理關閉 我們必須讓 enumerator 繼續超過塊

另一個障礙是有時我們想傳遞參數給 suspended 塊內發生的事情 想像我們有一個接受塊的方法具有以下語義:

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

但在我們的 calling code 我們想像這樣使用:

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

如果我們 wrap 我們的 writes 像這樣呢?

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 作為 magic value 告訴我們的方法它可以完成塊並返回 這 是 Enumerator 不會真正幫助我們的地方 因為我們不能傳遞任何參數給 Enumerator#next 如果我們能 我們將能夠做:

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

Enter Ruby's Fibers

這正是 Fibers 允許的 Fibers 允許你 accept arguments on each reentry 所以我們可以這樣實現我們的 wrapper:

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

這是它如何工作: 當你第一次在 deferred_writable 上調用 .resume 它進入 fiber 並一路到第一個 Fiber.yield 語句或最外層 Fiber 塊的末尾 以先到者為準 當你調用 Fiber.yield 它把控制權交還給你 記住 Enumerator? 塊將被 suspended 下次你調用 .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 內部 代碼流程在第一次調用 Fiber#resume 時 started 在第一次調用 Fiber.yield 時 suspended 然後在後續調用 Fiber#resume 時 continued Fiber.yield 的返回值是 resume 的參數 代碼從上次調用 Fiber.yield 的點繼續運行

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

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

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`