您的位置:首頁 > 手機教程 > 蘋果教程 > Ruby3多線程并行Ractor使用方法詳解

Ruby3多線程并行Ractor使用方法詳解

2022-04-18 12:49:04 來源:易采站長站 作者:

Ruby3多線程并行Ractor使用方法詳解

Ruby 3 Ractor官方手冊:https://github.com/ruby/ruby/blob/master/doc/ractor.md

在Ruby3之前,使用Thread來創建新的線程,但這種方式創建的多線程是并發而非并行的,MRI有一個全局解釋器鎖GIL來控制同一時刻只能有一個線程在執行:

# main Thread

t1 = Thread.new do 
  # new Thread
  sleep 3
end
t1.join

Ruby3通過Ractor(Ruby Actor,Actor模型通過消息傳遞的方式來修改狀態)支持真正的多線程并行,多個Ractor之間可并行獨立運行。

# main Ractor

# 創建一個可與main Ractor并行運行的Ractor
r = Ractor.new do
  sleep 2
  Ractor.yield "hello"
end

puts r.take

需注意,每個Ractor中至少有一個原生Ruby線程,但每個Ractor內部都擁有獨立的GIL,使得Ractor內部在同一時刻最多只能有一個線程在運行。從這個角度來看,Ractor實際上是解釋器線程,每個解釋器線程擁有一個全局解釋器鎖。

如果main Ractor退出,則其他Ractor也會收到退出信號,就像main Thread退出時,其他Thread也會退出一樣。

創建Ractor

使用Ractor.new創建一個Ractor實例,創建實例時需指定一個語句塊,該語句塊中的代碼會在該Ractor中運行。

r = Ractor.new do
  puts "new Ractor"
end

可在new方法的參數上為該Ractor實例指定名稱:

r = Ractor.new(name: "ractor1") do
  puts "new Ractor"
end

puts r.name  # ractor 1

new方法也可指定其他參數,這些參數必須在name參數之前,且這些參數將直接原樣傳遞給語句塊參數:

arr = [11, 22, 33]
r = Ractor.new(arr, name: "r1") do |arr|
  puts "arr"
end
sleep 1

關于new的參數,稍后還會有解釋。

可使用Ractor.current獲取當前的Ractor實例,使用Ractor.count獲取當前存活的Ractor實例數量。

Ractor之間傳遞消息

Ractor傳遞消息的方式分兩種:

Push方式:向某個特定的Ractor實例推送消息,可使用r.send(Msg)或別名r << Msg向該Ractor實例傳送消息,并在該Ractor實例內部使用Ractor.receive或別名Ractor.recv或它們的同名私有方法來接收推送進來的消息 Ractor還提供了Ractor.receive_if {expr}方法,表示只在expr為true時才接收消息,receive等價于receive_if {true}Pull方式:從某個特定的Ractor實例拉取消息,可在該Ractor實例內部使用Ractor.yield向外傳送消息,并在需要的地方使用r.take獲取傳輸出來的消息Ractor.new的語句塊返回值,相當于Ractor.yield,它也可被r.take接收

因此,對于Push方式,要求知道消息傳遞的目標Ractor,對于Pull方式,要求知道消息的來源Ractor。

# yield + take
r = Ractor.new {Ractor.yield "hello"}
puts r.take

# send + receive
r1 = Ractor.new do
  # Ractor.receive或Ractor.recv
  # 或同名私有方法:receive、recv
  puts Ractor.receive
end
r1.send("hello")
r1.take    # 本次take取得r1語句塊的返回值,即puts的返回值nil

使用new方法創建Ractor實例時,可指定new的參數,這些參數會被原樣傳遞給Ractor的語句塊參數。

arr = [11, 22, 33]
r = Ractor.new(arr) { |arr| ...}

實際上,new的參數等價于在Ractor語句塊的開頭使用了Ractor.receive接收消息:

r = Ractor.new 'ok' { |msg| msg }
r.take #=> 'ok'

# 基本等價于
r = Ractor.new do
  msg = Ractor.receive
  msg
end
r.send 'ok'
r.take #=> 'ok'

消息端口

Ractor之間傳遞消息時,實際上是通過Ractor的消息端口進行傳遞的。

每個Ractor都有自己的incoming port和outgoing port:

incoming port:是該Ractor接收消息的端口,r.sendRactor.receive使用該端口 每個incoming port都連接到一個大小不限的隊列上r.send傳入的消息都會寫入該隊列,由于該隊列大小不限,因此r.send從不阻塞Ractor.receive從該隊列彈出消息,當隊列為空時,Ractor.receive被阻塞直到新消息出現可使用r.close_incoming關閉incoming port,關閉該端口后,r.send將直接報錯,Ractor.receive將先從隊列中取數據,當隊列為空后,再調用Ractor.receive將報錯outgoing port:是該Ractor向外傳出消息的端口,Ractor.yieldr.take使用該端口Ractor.yield或Ractor語句塊返回時,消息從outgoing port流出當沒有r.take接收消息時,r內部的Ractor.yield將被阻塞當r內部沒有Ractor.yield時,r.take將被阻塞Ractor.yield從outgoing port傳出的消息可被任意多個r.take等待,但只有一個r.take可獲取到該消息可使用r.close_outgoing關閉outgoing port,關閉該端口后,再調用r.takeRactor.yield將直接報錯。如果r.take正被阻塞(等待Ractor.yield傳出消息),關閉outgoing port操作將取消所有等待中的take并報錯

Ractor.select等待消息

可使用Ractor.select(r1,r2,r3...)等待一個或多個Ractor實例outgoing port上的消息(因此,select主要用于等待Ractor.yield的消息),等待到第一個消息后立即返回。

Ractor.select的返回值格式為[r, obj],其中:

r表示等待到的那個Ractor實例obj表示接收到的消息對象

例如:

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

通常來說,會使用Ractor.select來輪詢等待多個Ractor實例的消息,通用化的處理流程參考如下:

# 充當管道功能的Ractor:接收消息并發送出去,并不斷循環
pipe = Ractor.new do
  loop do
    Ractor.yield Ractor.receive
  end
end

RN = 10
# rs變量保存了10個Ractor實例
# 每個Ractor實例都從管道pipe中取一次消息然后由本Ractor發送出去
rs = RN.times.map{|i|
  Ractor.new pipe, i do |pipe, i|
    msg = pipe.take
    msg # ping-pong
  end
}
# 向管道中發送10個數據
RN.times{|i| pipe << i}

# 輪詢等待10個Ractor實例的outgoing port
# 每等待成功一次,從rs中刪除所等待到的Ractor實例,
# 然后繼續等待剩下的Ractor實例
RN.times.map{
  r, n = Ractor.select(*rs)
  rs.delete r
  n
}.sort #=> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

此外,Ractor.select除了可等待消息外,也可以用來yield傳遞消息,更多用法參考官方手冊:Ractor.select。

Ractor并行時如何避免競態

多個Ractor之間是可并行運行的,為了避免Ractor之間傳遞數據時出現競態問題,Ractor采取了一些措施:

對于不可變對象,它們可直接在Ractor之間共享,此時傳遞它們的引用對于可變對象,它們不可直接在Ractor之間共享,此時傳遞數據時,默認先按字節逐字節拷貝,然后后傳遞副本也可以顯式指定移動數據,將某份數據從Ractor1移動到另一個Ractor2中,即轉移數據的所有權(參考Rust的所有權規則),轉移所有權后,原始所有者Ractor中將無法再訪問該數據

傳遞可共享對象:傳遞引用

可共享的對象:自動傳遞它們的引用,效率高

不可變對象可在Ractor之間直接共享(如Integer、symbol、true/false、nil),如:i=123:i是可共享的s="str".freeze:s是可共享的h={c: Object}.freeze:h是可共享的,因為Object是一個類對象,類對象是可共享的a=[1,[2],3].freeze:a不可共享,因為凍結后仍然包含可變的[2]Class/Module對象,即類對象自身和模塊對象自身是可共享的Ractor對象自身是可共享的

例如:

i = 33
r = Ractor.new do
  m = recv
  puts m.object_id
end

r.send(i)  # 傳遞i
r.take     # 等待Ractor執行結束(語句塊返回)
puts i.object_id  # i傳遞后仍然可用
=begin
67
67
=end

值得注意的是,Ractor對象是可共享的,因此可將某個Ractor實例傳遞給另一個Ractor實例。例如:

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

RN = 10
rs = RN.times.map{|i|
  # pipe是一個Ractor實例,這里作為參數傳遞給其他的Ractor實例
  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之間傳遞不可共享的對象時,默認會傳遞deep-copy后的副本,即按字節拷貝的方式拷貝該對象的每一個字節。這種方式效率較低。

例如:

arr = [11, 22, 33]  # 數組是可變的,不可共享
r = Ractor.new do
  m = recv
  puts "copied: #{m.object_id}"
end

r.send(arr)  # 傳遞數組,此時將逐字節拷貝數組
r.take
puts "origin: #{arr.object_id}"

=begin
copied: 60
origin: 80
=end

從結果看,兩個Ractor內的arr不是同一個對象。

需注意,對于全局唯一的對象來說(比如數值、nil、false、true、symbol),逐字節拷貝時并不會拷貝它們。例如:

arr = %i[lang action sub]
r = Ractor.new do
  m = recv
  puts "copied: #{m.object_id}, #{m[0].object_id}, #{m[1].object_id}"
end

r.send(arr)
r.take
puts "origin: #{arr.object_id}, #{arr[0].object_id}, #{arr[1].object_id}"

=begin
copied: 60, 80, 1046748
origin: 100, 80, 1046748
=end

注意,Thread對象無法拷貝,因此無法在Ractor之間傳遞。

轉移數據所有權

還可以讓r.send(msg, move: true)Ractor.yield(msg, move: true)傳遞數據時,明確表示要移動而非拷貝數據,即轉移數據的所有權(從原來的所有者Ractor實例轉移到目標Ractor實例)。

無論是可共享還是不可共享的對象,都可以轉移所有權,只不過轉移可共享對象的所有權沒有意義,因為轉移之后,原所有者仍然擁有所有權。

因此,通常只對不可共享的數據來轉移所有權,轉移所有權后,原所有者將無法訪問該數據。

str = "hello"
puts str.object_id
r = Ractor.new do
  m = recv
  puts m.object_id
end

r.send(str, move: true)  # 轉移str的所有權
r.take
#puts str.object_id  # 轉移所有權后再訪問str,將報錯

=begin
60
80
=end

值得注意的是,移動的本質是內存拷貝,它底層也一樣是逐字節拷貝原始數據的過程,所以移動傳遞數據的效率和傳遞副本數據的效率是類似的。移動傳遞和傳遞副本的區別之處在于所有權,移動傳遞后,原所有者Ractor實例將無法訪問該數據,而拷貝傳遞方式則允許原所有者訪問。

注意,Thread對象無法轉移所有權,因此無法在Ractor之間傳遞。

不可共享變成可共享:Ractor.make_shareable

對于不可共享的數據obj,可通過Ractor.make_shareable(obj)方法將其轉變為可共享的數據,默認轉變的方式是逐層次地遞歸凍結obj。也可指定額外的參數Ractor.make_shareable(obj, copy: true),此時將深拷貝obj得其副本,再讓副本(逐層遞歸凍結)轉變為可共享數據。

例如:

arr = %w[lang action sub]
puts arr.object_id
r = Ractor.new do
  m = recv
  puts m.object_id
end

r.send(Ractor.make_shareable(arr))
r.take
puts arr.object_id
puts arr.frozen?

輸出:

60
60
60
true

示例

工作者線程池:

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:

# 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'

更多關于Ruby3多線程并行Ractor使用方法請查看下面的相關鏈接

如有侵權,請聯系QQ:279390809 電話:15144810328

相關文章

色七七影院_香港三级台湾三级在线播放_男人放进女人阳道猛进猛出