Ractor - Ruby 類似 Actor 的並行抽象¶ ↑

Ractor 設計用於提供 Ruby 的並行執行功能，而無需擔心執行緒安全性。

摘要¶ ↑

解釋器程序中的多個 Ractor¶ ↑

您可以建立多個 Ractor，並讓它們並行執行。

• Ractor.new{ expr } 會建立一個新的 Ractor，而 expr 會在平行電腦上並行執行。

• 解釋器會使用第一個 Ractor (稱為主 Ractor) 呼叫。

• 如果主 Ractor 終止，則所有 Ractor 都會收到終止要求，就像執行緒一樣 (如果是主執行緒 (第一個呼叫的 Thread)，Ruby 解釋器會將所有正在執行的執行緒傳送至終止執行)。

• 每個 Ractor 都有一個或多個執行緒。

• Ractor 中的執行緒會共用一個 Ractor 全域鎖定，就像 GIL (MRI 術語中的 GVL)，因此它們無法並行執行 (除非在 C 層級明確釋放 GVL)。不同 Ractor 中的執行緒會並行執行。

• 建立 Ractor 的開銷類似於建立一個 Thread 的開銷。

多個 Ractor 之間的有限共用¶ ↑

與執行緒不同，Ractor 沒有共用所有內容。

• 大多數物件都是不可共用物件，因此您不必擔心因共用而造成的執行緒安全性問題。

• 有些物件是可共用物件。

• 不可變物件：不會參考不可共用物件的凍結物件。

  • i = 123：i 是不可變物件。

  • s = "str".freeze：s 是不可變物件。

  • a = [1, [2], 3].freeze：a 不是不可變物件，因為 a 參考不可共用物件 [2] (未凍結)。

  • h = {c: 物件}.freeze：h 是不可變物件，因為 h 參照 符號 :c 和可共用的 物件 類別物件，而該物件並未凍結。

• 類別/模組物件

• 特殊可共用物件

  • Ractor 物件本身。

  • 還有更多…

Ractor 之間的兩種通訊¶ ↑

Ractor 透過 Ractor 之間的訊息交換進行通訊和同步執行。有兩種訊息交換協定：推播類型（訊息傳遞）和拉取類型。

• 推播類型訊息傳遞：Ractor#send(obj) 和 Ractor.receive() 配對。

• 傳送者 ractor 透過 r.send(obj) 將 obj 傳遞給 ractor r，而接收者 ractor 則使用 Ractor.receive 接收訊息。

• 傳送者知道目的地 Ractor r，而接收者不知道傳送者（接受來自任何 ractor 的所有訊息）。

• 接收者有無限佇列，而傳送者將訊息排入佇列。傳送者不會因為將訊息放入此佇列而封鎖。

• 許多其他基於 Actor 的語言都採用此類型的訊息交換。

• Ractor.receive_if{ filter_expr } 是 Ractor.receive 的變體，用於選擇訊息。

• 拉取類型通訊：Ractor.yield(obj) 和 Ractor#take() 配對。

• 傳送者 ractor 宣告要透過 Ractor.yield(obj) 傳送 obj，而接收者 Ractor 則使用 r.take 取得它。

• 傳送者不知道目的地 Ractor，而接收者知道傳送者 Ractor r。

• 如果沒有另一方，傳送者或接收者將會封鎖。

傳送訊息的複製和移動語意¶ ↑

若要將不可共用物件作為訊息傳送，則會複製或移動物件。

• 複製：使用深度複製。

• 移動：移動成員資格。

• 傳送者在移動物件後無法存取已移動的物件。

• 保證至少只有 1 個 Ractor 可以存取物件。

執行緒安全性¶ ↑

Ractor 有助於撰寫執行緒安全的並行程式，但我們可以使用 Ractor 製作執行緒不安全的程式。

• 良好：共享限制

• 大多數物件都是不可共用的，因此我們無法製作資料競爭和競爭條件程式。

• 可共用物件受到直譯器或鎖定機制的保護。

• 不佳：類別/模組可能會違反此假設

• 為了與舊行為相容，類別和模組可能會引入資料競爭等問題。

• Ruby 程式設計師在多個 Ractor 程式上修改類別/模組物件時應小心。

• 不佳：Ractor 無法解決所有執行緒安全性問題

• 有數個封鎖作業（等待傳送、等待傳送和等待取得），因此您可以製作一個有死結和活結問題的程式。

• 某些可共享的物件可以導入交易（例如 STM）。但是，錯誤使用交易會產生不一致的狀態。

沒有 Ractor 的話，我們需要追蹤所有狀態變異來除錯執行緒安全性問題。有了 Ractor，你可以專注於與 Ractor 共享的可疑程式碼。

建立與終止¶ ↑

Ractor.new¶ ↑

• Ractor.new{ expr } 產生另一個 Ractor。

# Ractor.new with a block creates new Ractor
r = Ractor.new do
  # This block will be run in parallel with other ractors
end

# You can name a Ractor with `name:` argument.
r = Ractor.new name: 'test-name' do
end

# and Ractor#name returns its name.
r.name #=> 'test-name'

給定區塊隔離¶ ↑

Ractor 在給定的區塊中執行給定的 expr。給定的區塊會透過 Proc#isolate 方法從外部範圍隔離（Ruby 使用者尚未公開）。為了防止在 ractor 之間共用不可共享的物件，區塊外部變數、self 和其他資訊會被隔離。

Proc#isolate 會在 Ractor 建立時間（呼叫 Ractor.new 時）呼叫。如果給定的 Proc 物件無法因為外部變數等原因而隔離，將會產生錯誤。

begin
  a = true
  r = Ractor.new do
    a #=> ArgumentError because this block accesses `a`.
  end
  r.take # see later
rescue ArgumentError
end

• 給定區塊的 self 是 Ractor 物件本身。

r = Ractor.new do
  p self.class #=> Ractor
  self.object_id
end
r.take == self.object_id #=> false

傳遞給 Ractor.new() 的引數會變成給定區塊的區塊參數。但是，解釋器不會傳遞參數物件參考，而是將它們傳送為訊息（請參閱下方詳細資訊）。

r = Ractor.new 'ok' do |msg|
  msg #=> 'ok'
end
r.take #=> 'ok'

# almost similar to the last example
r = Ractor.new do
  msg = Ractor.receive
  msg
end
r.send 'ok'
r.take #=> 'ok'

給定區塊的執行結果¶ ↑

給定區塊的傳回值會變成傳送訊息（請參閱下方詳細資訊）。

r = Ractor.new do
  'ok'
end
r.take #=> `ok`

# almost similar to the last example
r = Ractor.new do
  Ractor.yield 'ok'
end
r.take #=> 'ok'

給定區塊中的錯誤會傳播到傳送訊息的接收者。

r = Ractor.new do
  raise 'ok' # exception will be transferred to the receiver
end

begin
  r.take
rescue Ractor::RemoteError => e
  e.cause.class   #=> RuntimeError
  e.cause.message #=> 'ok'
  e.ractor        #=> r
end

Ractor 之間的通訊¶ ↑

Ractor 之間的通訊是透過傳送和接收訊息來達成。有兩種方式可以彼此通訊。

• (1) 訊息傳送/接收

• (1-1) 推送型傳送/接收（傳送者知道接收者）。類似於 Actor 模型。

• (1-2) 拉取型 yield/take（接收者知道傳送者）。

• (2) 使用可共享容器物件

• Ractor::TVar gem (ko1/ractor-tvar)

• 更多？

使用者可以使用 (1) 來控制程式執行時間，但不要使用 (2) 來控制（僅作為臨界區管理）。

對於訊息傳送和接收，有兩種 API 類型：推送型和拉取型。

• (1-1) 傳送/接收（推送型）

• Ractor#send(obj)（Ractor#<<(obj) 為別名）將訊息傳送至 Ractor 的輸入埠。輸入埠連接到無限大小的輸入佇列，因此 Ractor#send 永遠不會封鎖。

• Ractor.receive 從其自己的輸入佇列中取出訊息。如果輸入佇列為空，則 Ractor.receive 呼叫會封鎖。

• Ractor.receive_if{|msg| filter_expr } 是 Ractor.receive 的變體。receive_if 僅接收 filter_expr 為 true 的訊息（因此 Ractor.receive 與 Ractor.receive_if{ true } 相同。

• (1-2) yield/take（拉取類型）

• Ractor.yield(obj) 將訊息傳送至正在透過輸出埠呼叫 Ractor#take 的 Ractor。如果沒有 Ractor 正在等候，則 Ractor.yield(obj) 會封鎖。如果有多個 Ractor 正在等候 Ractor.yield(obj)，則只有一個 Ractor 可以接收訊息。

• Ractor#take 接收由指定 Ractor 的 Ractor.yield(obj) 方法等候的訊息。如果 Ractor 尚未呼叫 Ractor.yield，則 Ractor#take 呼叫會封鎖。

• Ractor.select() 可以等候 take、yield 和 receive 的成功。

• 您可以關閉輸入埠或輸出埠。

• 您可以使用 Ractor#close_incoming 和 Ractor#close_outgoing 關閉它們。

• 如果 Ractor 的輸入埠已關閉，則您無法將訊息 send 至 Ractor。如果 Ractor.receive 因輸入埠已關閉而封鎖，則會引發例外狀況。

• 如果 Ractor 的輸出埠已關閉，則您無法在 Ractor 上呼叫 Ractor#take 和 Ractor.yield。如果 Ractor 因 Ractor#take 或 Ractor.yield 而封鎖，則關閉輸出埠會在這些封鎖 Ractor 上引發例外狀況。

• 當 Ractor 終止時，Ractor 的埠會關閉。

• 有 3 種方法可以將物件傳送為訊息

• (1) 傳送參考：傳送可共用物件時，僅傳送物件的參考（快速）

• (2) 複製物件：透過深度複製物件來傳送不可共用的物件（速度較慢）。請注意，您無法傳送不支援深度複製的物件。某些 T_DATA 物件不受支援。

• (3) 移動物件：傳送具有成員資格的不可共用物件參考。傳送者 Ractor 在移動物件後將無法再存取已移動的物件（會引發例外）。目前的實作會將新物件作為接收者 Ractor 的已移動物件，並將傳送物件的參考複製到已移動的物件。

• 您可以透過 move: 關鍵字、Ractor#send(obj, move: true/false) 和 Ractor.yield(obj, move: true/false)（預設為 false (COPY)）選擇「複製」和「移動」。

傳送/接收埠¶ ↑

每個 Ractor 都有一個輸入埠和輸出埠。輸入埠連接到無限大小的輸入佇列。

Ractor r
                 +-------------------------------------------+
                 | incoming                         outgoing |
                 | port                                 port |
   r.send(obj) ->*->[incoming queue]     Ractor.yield(obj) ->*-> r.take
                 |                |                          |
                 |                v                          |
                 |           Ractor.receive                  |
                 +-------------------------------------------+


Connection example: r2.send obj on r1、Ractor.receive on r2
  +----+     +----+
  * r1 |---->* r2 *
  +----+     +----+


Connection example: Ractor.yield(obj) on r1, r1.take on r2
  +----+     +----+
  * r1 *---->- r2 *
  +----+     +----+

Connection example: Ractor.yield(obj) on r1 and r2,
                    and waiting for both simultaneously by Ractor.select(r1, r2)

  +----+
  * r1 *------+
  +----+      |
              +----> Ractor.select(r1, r2)
  +----+      |
  * r2 *------|
  +----+

r = Ractor.new do
  msg = Ractor.receive # Receive from r's incoming queue
  msg # send back msg as block return value
end
r.send 'ok' # Send 'ok' to r's incoming port -> incoming queue
r.take      # Receive from r's outgoing port

最後一個範例顯示以下 ractor 網路。

  +------+        +---+
  * main |------> * r *---+
  +------+        +---+   |
      ^                   |
      +-------------------+

而且，使用 Ractor.new 的引數可以簡化這段程式碼。

# Actual argument 'ok' for `Ractor.new()` will be sent to created Ractor.
r = Ractor.new 'ok' do |msg|
  # Values for formal parameters will be received from incoming queue.
  # Similar to: msg = Ractor.receive

  msg # Return value of the given block will be sent via outgoing port
end

# receive from the r's outgoing port.
r.take #=> `ok`

Ractor.new 的區塊回傳值¶ ↑

如前所述，Ractor.new 的回傳值（Ractor.new{ expr } 中 expr 的評估值）可以透過 Ractor#take 取得。

Ractor.new{ 42 }.take #=> 42

當區塊回傳值可用時，Ractor 已死亡，因此除了已取用的 Ractor 之外，沒有其他 ractor 可以觸及回傳值，因此任何值都可以透過此通訊路徑傳送，而不會進行任何修改。

r = Ractor.new do
  a = "hello"
  binding
end

r.take.eval("p a") #=> "hello" (other communication path can not send a Binding object directly)

使用 Ractor.select 等待多個 Ractor¶ ↑

您可以使用 Ractor.select(*ractors) 等待多個 Ractor 的 yield。Ractor.select() 的回傳值為 [r, msg]，其中 r 為產生 yield 的 Ractor，而 msg 為產生的 yield 訊息。

等待單一 ractor（與 Ractor.take 相同）

r1 = Ractor.new{'r1'}

r, obj = Ractor.select(r1)
r == r1 and obj == 'r1' #=> true

等待兩個 ractor

r1 = Ractor.new{'r1'}
r2 = Ractor.new{'r2'}
rs = [r1, r2]
as = []

# Wait for r1 or r2's Ractor.yield
r, obj = Ractor.select(*rs)
rs.delete(r)
as << obj

# Second try (rs only contain not-closed ractors)
r, obj = Ractor.select(*rs)
rs.delete(r)
as << obj
as.sort == ['r1', 'r2'] #=> true

Complex 範例

pipe = Ractor.new do
  loop do
    Ractor.yield Ractor.receive
  end
end

RN = 10
rs = RN.times.map{|i|
  Ractor.new pipe, i do |pipe, i|
    msg = pipe.take
    msg # ping-pong
  end
}
RN.times{|i|
  pipe << i
}
RN.times.map{
  r, n = Ractor.select(*rs)
  rs.delete r
  n
}.sort #=> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

多個 Ractor 可以傳送至一個 Ractor。

# Create 10 ractors and they send objects to pipe ractor.
# pipe ractor yield received objects

pipe = Ractor.new do
  loop do
    Ractor.yield Ractor.receive
  end
end

RN = 10
rs = RN.times.map{|i|
  Ractor.new pipe, i do |pipe, i|
    pipe << i
  end
}

RN.times.map{
  pipe.take
}.sort #=> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

待辦事項：目前的 Ractor.select() 有與 select(2) 相同的問題，因此應該改善這個介面。

待辦事項：Go 語言的 select 語法使用輪詢技術來進行公平排程。目前 Ractor.select() 沒有使用此技術。

關閉 Ractor 的埠¶ ↑

• Ractor#close_incoming/outgoing 關閉傳入/傳出埠（類似於 Queue#close）。

• Ractor#close_incoming

• r.send(obj) 其中 r 的傳入埠已關閉，將引發例外狀況。

• 當傳入佇列為空且傳入埠已關閉時，Ractor.receive 會引發例外狀況。如果傳入佇列不為空，它會在沒有例外狀況的情況下取消排隊一個物件。

• Ractor#close_outgoing

• 在已關閉傳出埠的 Ractor 上 Ractor.yield，它將引發例外狀況。

• 對於已關閉傳出埠的 Ractor，Ractor#take 將引發例外狀況。如果 Ractor#take 正在封鎖，它將引發例外狀況。

• 當 Ractor 終止時，埠會自動關閉。

• Ractor 區塊的傳回值將作為 Ractor.yield(ret_val) 讓出，即使實作終止了基於本機的執行緒。

範例（嘗試從已關閉的 Ractor 擷取）

r = Ractor.new do
  'finish'
end
r.take # success (will return 'finish')
begin
  o = r.take # try to take from closed Ractor
rescue Ractor::ClosedError
  'ok'
else
  "ng: #{o}"
end

範例（嘗試傳送至已關閉（終止）的 Ractor）

r = Ractor.new do
end

r.take # wait terminate

begin
  r.send(1)
rescue Ractor::ClosedError
  'ok'
else
  'ng'
end

當多個 Ractor 正在等待 Ractor.yield() 時，Ractor#close_outgoing 會透過引發例外狀況（ClosedError）取消所有封鎖。

透過複製傳送訊息¶ ↑

如果 obj 是無法共用的物件，則 Ractor#send(obj) 或 Ractor.yield(obj) 會深入複製 obj。

obj = 'str'.dup
r = Ractor.new obj do |msg|
  # return received msg's object_id
  msg.object_id
end

obj.object_id == r.take #=> false

有些物件不支援複製其值，並會引發例外狀況。

obj = Thread.new{}
begin
  Ractor.new obj do |msg|
    msg
  end
rescue TypeError => e
  e.message #=> #<TypeError: allocator undefined for Thread>
else
  'ng' # unreachable here
end

透過移動傳送訊息¶ ↑

Ractor#send(obj, move: true) 或 Ractor.yield(obj, move: true) 將 obj 移至目的地的 Ractor。如果來源 Ractor 觸及已移動的物件（例如，呼叫 obj.foo() 等方法），這將會是一個錯誤。

# move with Ractor#send
r = Ractor.new do
  obj = Ractor.receive
  obj << ' world'
end

str = 'hello'
r.send str, move: true
modified = r.take #=> 'hello world'

# str is moved, and accessing str from this Ractor is prohibited

begin
  # Error because it touches moved str.
  str << ' exception' # raise Ractor::MovedError
rescue Ractor::MovedError
  modified #=> 'hello world'
else
  raise 'unreachable'
end

# move with Ractor.yield
r = Ractor.new do
  obj = 'hello'
  Ractor.yield obj, move: true
  obj << 'world'  # raise Ractor::MovedError
end

str = r.take
begin
  r.take
rescue Ractor::RemoteError
  p str #=> "hello"
end

有些物件不支援移動，並且會引發例外狀況。

r = Ractor.new do
  Ractor.receive
end

r.send(Thread.new{}, move: true) #=> allocator undefined for Thread (TypeError)

為了達到對已移動物件的存取禁止，使用類別替換技術來實作它。

可共用物件¶ ↑

下列物件可共用。

• 不可變物件

• 小整數、部分符號、true、false、nil（在內部也稱為 SPECIAL_CONST_P() 物件）

• 凍結的原生物件

  • Numeric 物件：Float、Complex、Rational、大整數（在內部稱為 T_BIGNUM）

  • 所有符號。

• 凍結的 String 和 Regexp 物件（它們的實例變數應僅參考可共用的物件）

• Class、Module 物件（在內部稱為 T_CLASS、T_MODULE 和 T_ICLASS）

• Ractor 和其他關注同步的特殊物件。

實作：現在可共用的物件（RVALUE）具有 FL_SHAREABLE 標記。此標記可以延遲新增。

為了建立可共用的物件，提供了 Ractor.make_shareable(obj) 方法。在這種情況下，嘗試透過凍結 obj 和遞迴可遍歷的物件來建立可共用性。此方法接受 copy: 關鍵字（預設值為 false）。Ractor.make_shareable(obj, copy: true) 嘗試建立 obj 的深度拷貝，並使拷貝的物件可共用。

語言變更以在 Ractor 之間隔離不可共用物件¶ ↑

為了在 Ractor 之間隔離不可共用物件，我們在多 Ractor Ruby 程式中引入了額外的語言語意。

請注意，在不使用 Ractor 的情況下，不需要這些額外的語意（與 Ruby 2 完全相容）。

全域變數¶ ↑

只有主 Ractor（在解釋器啟動時建立的 Ractor）可以存取全域變數。

$gv = 1
r = Ractor.new do
  $gv
end

begin
  r.take
rescue Ractor::RemoteError => e
  e.cause.message #=> 'can not access global variables from non-main Ractors'
end

請注意，某些特殊全域變數是 ractor 本地的，例如 $stdin、$stdout、$stderr。有關更多詳細資訊，請參閱 [Bug #17268]。

可共用物件的實例變數¶ ↑

如果參考值是可共用物件，則可以從非主 Ractor 取得類別/模組的實例變數。

class C
  @iv = 1
end

p Ractor.new do
  class C
     @iv
  end
end.take #=> 1

否則，只有主 Ractor 可以存取可共用物件的實例變數。

class C
  @iv = [] # unshareable object
end

Ractor.new do
  class C
    begin
      p @iv
    rescue Ractor::IsolationError
      p $!.message
      #=> "can not get unshareable values from instance variables of classes/modules from non-main Ractors"
    end

    begin
      @iv = 42
    rescue Ractor::IsolationError
      p $!.message
      #=> "can not set instance variables of classes/modules by non-main Ractors"
    end
  end
end.take

shared = Ractor.new{}
shared.instance_variable_set(:@iv, 'str')

r = Ractor.new shared do |shared|
  p shared.instance_variable_get(:@iv)
end

begin
  r.take
rescue Ractor::RemoteError => e
  e.cause.message #=> can not access instance variables of shareable objects from non-main Ractors (Ractor::IsolationError)
end

請注意，類別/模組物件的實例變數在 Ractor 上也遭到禁止。

Class 變數¶ ↑

只有主 Ractor 可以存取類別變數。

class C
  @@cv = 'str'
end

r = Ractor.new do
  class C
    p @@cv
  end
end


begin
  r.take
rescue => e
  e.class #=> Ractor::IsolationError
end

常數¶ ↑

只有主 Ractor 可以讀取參考不可共用物件的常數。

class C
  CONST = 'str'
end
r = Ractor.new do
  C::CONST
end
begin
  r.take
rescue => e
  e.class #=> Ractor::IsolationError
end

只有主 Ractor 可以定義常數，用於參考不可共用物件。

class C
end
r = Ractor.new do
  C::CONST = 'str'
end
begin
  r.take
rescue => e
  e.class #=> Ractor::IsolationError
end

若要建立支援多個 ractor 的函式庫，常數應僅參考可共用物件。

TABLE = {a: 'ko1', b: 'ko2', c: 'ko3'}

在此情況下，TABLE 參考不可共用的 Hash 物件。因此，其他 ractor 無法參考 TABLE 常數。若要使其可共用，我們可以使用 Ractor.make_shareable()，如下所示。

TABLE = Ractor.make_shareable( {a: 'ko1', b: 'ko2', c: 'ko3'} )

為了簡化，Ruby 3.0 引入了新的 shareable_constant_value 指令。

# shareable_constant_value: literal

TABLE = {a: 'ko1', b: 'ko2', c: 'ko3'}
#=> Same as: TABLE = Ractor.make_shareable( {a: 'ko1', b: 'ko2', c: 'ko3'} )

shareable_constant_value 指令接受下列模式（說明使用範例：CONST = expr）

• none：不執行任何動作。與 CONST = expr 相同

• literal

• 如果 expr 包含文字，則替換為 CONST = Ractor.make_shareable(expr)。

• 否則：替換為 CONST = expr.tap{|o| raise unless Ractor.shareable?(o)}。

• experimental_everything：替換為 CONST = Ractor.make_shareable(expr)。

• experimental_copy：替換為 CONST = Ractor.make_shareable(expr, copy: true)。

除了 none 模式（預設）外，保證已指定常數僅參考可共用物件。

請參閱 doc/syntax/comments.rdoc 以取得更多詳細資料。

實作注意事項¶ ↑

• 每個 Ractor 都有自己的執行緒，這表示每個 Ractor 至少有 1 個原生執行緒。

• 每個 Ractor 都有自己的 ID（rb_ractor_t::pub::id）。

• 在偵錯模式下，所有不可共用物件都標記有目前的 Ractor ID，並在 VM 中檢查以偵測不可共用物件外洩（從不同的 Ractor 存取物件）。

範例¶ ↑

Actor 模型中的傳統 Ring 範例¶ ↑

RN = 1_000
CR = Ractor.current

r = Ractor.new do
  p Ractor.receive
  CR << :fin
end

RN.times{
  r = Ractor.new r do |next_r|
    next_r << Ractor.receive
  end
}

p :setup_ok
r << 1
p Ractor.receive

Fork-join¶ ↑

def fib n
  if n < 2
    1
  else
    fib(n-2) + fib(n-1)
  end
end

RN = 10
rs = (1..RN).map do |i|
  Ractor.new i do |i|
    [i, fib(i)]
  end
end

until rs.empty?
  r, v = Ractor.select(*rs)
  rs.delete r
  p answer: v
end

工作者池¶ ↑

require 'prime'

pipe = Ractor.new do
  loop do
    Ractor.yield Ractor.receive
  end
end

N = 1000
RN = 10
workers = (1..RN).map do
  Ractor.new pipe do |pipe|
    while n = pipe.take
      Ractor.yield [n, n.prime?]
    end
  end
end

(1..N).each{|i|
  pipe << i
}

pp (1..N).map{
  _r, (n, b) = Ractor.select(*workers)
  [n, b]
}.sort_by{|(n, b)| n}

管線¶ ↑

# pipeline with yield/take
r1 = Ractor.new do
  'r1'
end

r2 = Ractor.new r1 do |r1|
  r1.take + 'r2'
end

r3 = Ractor.new r2 do |r2|
  r2.take + 'r3'
end

p r3.take #=> 'r1r2r3'

# pipeline with send/receive

r3 = Ractor.new Ractor.current do |cr|
  cr.send Ractor.receive + 'r3'
end

r2 = Ractor.new r3 do |r3|
  r3.send Ractor.receive + 'r2'
end

r1 = Ractor.new r2 do |r2|
  r2.send Ractor.receive + 'r1'
end

r1 << 'r0'
p Ractor.receive #=> "r0r1r2r3"

監督¶ ↑

# ring example again

r = Ractor.current
(1..10).map{|i|
  r = Ractor.new r, i do |r, i|
    r.send Ractor.receive + "r#{i}"
  end
}

r.send "r0"
p Ractor.receive #=> "r0r10r9r8r7r6r5r4r3r2r1"

# ring example with an error

r = Ractor.current
rs = (1..10).map{|i|
  r = Ractor.new r, i do |r, i|
    loop do
      msg = Ractor.receive
      raise if /e/ =~ msg
      r.send msg + "r#{i}"
    end
  end
}

r.send "r0"
p Ractor.receive #=> "r0r10r9r8r7r6r5r4r3r2r1"
r.send "r0"
p Ractor.select(*rs, Ractor.current) #=> [:receive, "r0r10r9r8r7r6r5r4r3r2r1"]
r.send "e0"
p Ractor.select(*rs, Ractor.current)
#=>
#<Thread:0x000056262de28bd8 run> terminated with exception (report_on_exception is true):
Traceback (most recent call last):
        2: from /home/ko1/src/ruby/trunk/test.rb7:in `block (2 levels) in <main>'
        1: from /home/ko1/src/ruby/trunk/test.rb:7:in `loop'
/home/ko1/src/ruby/trunk/test.rb:9:in `block (3 levels) in <main>': unhandled exception
Traceback (most recent call last):
        2: from /home/ko1/src/ruby/trunk/test.rb7:in `block (2 levels) in <main>'
        1: from /home/ko1/src/ruby/trunk/test.rb:7:in `loop'
/home/ko1/src/ruby/trunk/test.rb:9:in `block (3 levels) in <main>': unhandled exception
        1: from /home/ko1/src/ruby/trunk/test.rb21:in `<main>'
<internal:ractor>:69:in `select': thrown by remote Ractor. (Ractor::RemoteError)

# resend non-error message

r = Ractor.current
rs = (1..10).map{|i|
  r = Ractor.new r, i do |r, i|
    loop do
      msg = Ractor.receive
      raise if /e/ =~ msg
      r.send msg + "r#{i}"
    end
  end
}

r.send "r0"
p Ractor.receive #=> "r0r10r9r8r7r6r5r4r3r2r1"
r.send "r0"
p Ractor.select(*rs, Ractor.current)
[:receive, "r0r10r9r8r7r6r5r4r3r2r1"]
msg = 'e0'
begin
  r.send msg
  p Ractor.select(*rs, Ractor.current)
rescue Ractor::RemoteError
  msg = 'r0'
  retry
end

#=> <internal:ractor>:100:in `send': The incoming-port is already closed (Ractor::ClosedError)
# because r == r[-1] is terminated.

# ring example with supervisor and re-start

def make_ractor r, i
  Ractor.new r, i do |r, i|
    loop do
      msg = Ractor.receive
      raise if /e/ =~ msg
      r.send msg + "r#{i}"
    end
  end
end

r = Ractor.current
rs = (1..10).map{|i|
  r = make_ractor(r, i)
}

msg = 'e0' # error causing message
begin
  r.send msg
  p Ractor.select(*rs, Ractor.current)
rescue Ractor::RemoteError
  r = rs[-1] = make_ractor(rs[-2], rs.size-1)
  msg = 'x0'
  retry
end

#=> [:receive, "x0r9r9r8r7r6r5r4r3r2r1"]