Làm chủ Fibers và Enumerators trong Ruby: Kỹ thuật Flow Control Nâng cao

Ruby có nhiều cách thực hiện iteration — loops, blocks và enumerators. Hầu hết Ruby programmers ít nhất familiar với loops và blocks nhưng Enumerator và Fiber thường ở trong bóng tối. Trong edition này của Ruby Magic, tác giả khách Julik soi sáng Enumerable và Fiber để giải thích flow controlling enumerables và turning blocks inside out.

Suspending Blocks và Chained Iteration

Chúng ta đã discuss Enumerator trong edition trước của Ruby Magic, nơi mô tả cách return Enumerator từ #each method của bạn và nó có thể dùng cho gì. Use case rộng hơn cho Enumerator và Fiber là chúng có thể "suspending block" mid-flight. Không chỉ block given cho #each hoặc entire call đến #each, mà bất kỳ block nào!

Đây là construct rất powerful, có thể dùng implement shims cho methods hoạt động bằng dùng blocks như bridge đến callers expect sequential calls thay vì taking block. Ví dụ, tưởng tượng chúng ta muốn open database handle và read mỗi item đã retrieve:

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

Block API tuyệt vời vì nó potentially perform cleanup khi block terminated. Tuy nhiên, một số consumer muốn làm việc với database như sau:

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

Trong thực tế, nghĩa là chúng ta muốn "suspending" execution của block "just for now" và carry on sau trong block. Vì vậy, caller take over flow control thay vì nó ở trong tay callee (method performing block).

Chaining Iterators

Một trong những use case phổ biến nhất của pattern này là chaining multiple iterators together. Khi làm vậy, methods chúng ta quen dùng cho iteration (như #each), return Enumerator object thay vào đó, mà chúng ta dùng để "grab" values mà block gửi cho chúng ta dùng yield statement:

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

Enumerators sau đó có thể chained cho phép perform operations như "any iteration nhưng với index". Trong ví dụ này, chúng ta gọi #map trên range để lấy Enumerable object. Chúng ta sau đó chain #with_index để iterate over range với index:

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

Điều này rất useful, đặc biệt nếu system của bạn dùng events. Ruby cung cấp built-in method wrapping bất kỳ method nào với Enumerator generator, cho phép accomplish chính xác điều này. Tưởng tượng chúng ta muốn "pull" rows one by one từ with_each_row_of_result, thay vì method yield chúng ta.

@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 # block đã kết thúc và cursor empty, cleanup đã xảy ra end end

Nếu chúng ta tự implement, nó có thể ra như sau:

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

Turning Blocks Inside Out (Lật ngược Block)

Rails cho phép assign response body cũng là Enumerator. Nó sẽ call next trên Enumerator chúng ta assign as response body và expect returned value là string — sẽ written vào Rack response. Ví dụ, chúng ta có thể return call đến #each method của Range as Rails response body:

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

Đây là cái tôi gọi turning a block inside out. Bản chất, nó là control flow helper cho phép "freeze time" trong block (hoặc loop, cũng là block trong Ruby) mid-flight.

Tuy nhiên, Enumerators có limiting property làm chúng hơi kém useful. Tưởng tượng chúng ta muốn làm như sau:

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

Hãy wrap nó với enumerator, và write vào nó

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

Mọi thứ works great. Tuy nhiên, có hitch — làm sao chúng ta báo cho enumerator rằng chúng ta done writing, để nó có thể "finish" block, close file và exit? Điều này sẽ perform số bước quan trọng — ví dụ resource cleanup (file sẽ closed), cũng như ensure all buffered writes flushed to disk. Chúng ta có access đến File object, và có thể close nó ourselves, nhưng chúng ta muốn enumerator manage closing cho chúng ta; chúng ta phải để enumerator proceed past block.

Rào cản khác là đôi khi chúng ta muốn pass arguments của cái đang xảy ra trong suspended block. Tưởng tượng chúng ta có block-accepting method với semantics sau:

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

nhưng trong calling code chúng ta muốn dùng như sau:

writable = write_file_through_encryptor(file_name) writable << "Some data" # ...later on writable << "Some more data" writable.finish

Ideally, chúng ta wrap method call vào structure cho phép trick sau:

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

What if chúng ta wrap writes như sau?

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)

Trong trường hợp này, chúng ta sẽ dùng :terminate như magic value báo cho method nó có thể finish block và return. Đây là nơi Enumerator không really help vì chúng ta không thể pass arguments cho Enumerator#next. Nếu có thể, chúng ta sẽ làm:

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

Ruby's Fibers Xuất Hiện

Đây chính xác là gì Fibers permit. Fiber cho phép bạn accept arguments on each reentry, nên chúng ta có thể implement wrapper như sau:

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

Đây là cách nó hoạt động: Khi bạn lần đầu gọi .resume trên deferred_writable, nó enter fiber và đi all the way đến first Fiber.yield statement hoặc đến end của outermost Fiber block, whichever comes first. Khi bạn gọi Fiber.yield, nó give bạn back control. Nhớ Enumerator? Block sẽ bị suspended, và lần sau bạn gọi .resume, argument to resume trở thành data_to_write mới.

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 

Vì vậy, trong Fiber, code flow started trên first call to Fiber#resume, suspended tại first call to Fiber.yield, và sau đó continued trên subsequent calls to Fiber#resume, với return value của Fiber.yield là arguments to resume. Code tiếp tục chạy từ điểm Fiber.yield cuối cùng được gọi.

Đây là quirk của Fibers là initial arguments cho fiber sẽ được pass cho bạn như block arguments, không qua return value của Fiber.yield.

Với điều đó, chúng ta biết bằng cách pass special argument cho resume, chúng ta có thể decide trong Fiber có nên stop hay không. Hãy thử:

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

Có số tình huống facilities này rất useful. Vì Fiber chứa suspended block code có thể manually resumed, Fibers có thể dùng implement event reactors và deal với concurrent operations trong single thread. Chúng lightweight, nên bạn có thể implement server dùng Fibers bằng assign single client cho single Fiber và switch giữa Fiber objects as necessary.

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 có additional standard library gọi fiber cho phép explicitly transfer control từ fiber này sang fiber khác, có thể bonus facility cho uses này.

Controlling Data Emission Rates (Kiểm soát Tốc độ Phát dữ liệu)

Use case great khác cho fibers và enumerators nảy sinh khi bạn muốn control rate mà Ruby block emit data. Ví dụ, trong zip_tricks chúng ta support block use sau as primary way dùng library:

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

Chúng ta do đó allow "push" control on part của code tạo ZIP archive, và impossible control amount data nó output và how often. Nếu chúng ta muốn write ZIP in chunks of, say, 5 MB — limitation trên AWS S3 object storage — chúng ta phải tạo custom output_io object somehow "refuse" accept << method calls khi segment cần split off vào S3 multipart part. Chúng ta có thể, tuy nhiên, invert control và make it "pull". Chúng ta vẫn dùng same block để write big CSV file, nhưng chúng ta sẽ resume và halt nó based on output nó provide. Chúng ta do đó make use sau possible:

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`

Điều này cho phép control rate mà ZIP file generator emit data.

Enumerator và Fiber do đó là control flow mechanism để turn "push" blocks thành "pull" objects accept method calls.

Chỉ có một pitfall với Fibers và Enumerators — nếu bạn có gì đó như ensure trong block, hoặc gì đó cần làm sau block complete, nó bây giờ up to caller call bạn đủ lần. In a way, nó comparable to constraints bạn có khi dùng Promises trong JavaScript.

Kết luận

Đây kết thúc look của chúng ta vào flow-controlled enumerables trong Ruby. Dọc theo đường, Julik soi sáng similarities và differences giữa Enumerable và Fiber classes, và dive vào examples nơi caller determine flow of data. Chúng ta cũng học về Fiber additional magic cho phép pass arguments on each block reentry. Happy flow-controlling!

Để nhận steady dose of magic, subscribe Ruby Magic và chúng ta sẽ deliver monthly edition thẳng vào inbox bạn.