はじめに
この記事はRuby3で導入される並列・並行処理のための新機能Ractorに興味のある方向けの記事になります。
簡単なRactorを使ったサンプルコードを解説しつつ、理解を深めることができるように書いてみました(ほとんど未来の僕へ向けた記事になっている感じはありますが……)。
また、Ractor自体でどういったことができるのかを調べた結果をまとめた記事でもあります。
そのため後半はRactorで色々遊んだ時のコードをもとに挙動を解説しています。
環境
- Windows10 2004
- WSL2(Ubuntu 18.04)
- ruby 3.0.0dev (2020-09-07T04:29:42Z master 17a27060a7) [x86_64-linux]
Ractorとは？
そもそもRactorについて知らない人もいると思いますので、簡単に紹介します。
RactorとはRuby3で導入される並列・並行処理のための新機能になります。機能自体の提案は数年前からあり、当時はGuildという名前で提案されていました。
しかし、ゲーム業界から「Guildという名前は使っているので、別の名前にしてほしい」という声があり、現在のRactorへと変わりました。
Actorモデルを参考にしているようで、そのためRactor(Ruby’s Actor)という名前に変更されたようです。
Ractorは並行実行の単位であり、それぞれが並行して実行されます。
たとえば以下のコードではputs :helloとputs :helloはそれぞれ並行して実行されます。
Ractor.new do
5.times do
puts :hello
end
end
5.times do
puts :world
end
このコードを実行すると以下のような結果になります。
world
helloworld
hello
world
helloworld
helloworld
hello
このようにそれぞれの処理を並行して実行できます。
またRactorは別のRactorへとオブジェクトを送受信し、同期しながら実行することもできます。同期の方法としてはpush型とpull型の二つがあります。
たとえばpush型の場合は以下のようなコードになります。
r1 = Ractor.new do
:hoge
end
r2 = Ractor.new do
puts :fuga, Ractor.recv
end
r2.send(r1.take)
r2.take
Ractorではsendメソッドを使い、別のRactorへとオブジェクトを送ることができます。上記のコードだと
r2.send(r1.take)
の部分でr2へと送信しています。
送信されたオブジェクトはRactor内でRactor.recvで受け取ることができ
r2 = Ractor.new do
puts :fuga, Ractor.recv
end
r2.send(r1.take)で送られたオブジェクトを受け取ってputsメソッドに渡すことができます。
またRactorが実行された結果を受け取る際にtakeメソッドを使います。
ですのでr1.takeは:hogeを受け取っています。
つまり、r2.send(r1.take)ではr1の実行結果を受け取り、それをr2へと送信しています。
そして、r2内のputs :fuga, Ractor.recvはputs :fuga, :hogeとなり、fugaとhogeがそれぞれ出力されるということです。
これがpush型でのオブジェクトをやり取りしている流れになります。
対してpull型は以下のようなコードになります。
r1 = Ractor.new 42 do |arg|
Ractor.yield arg
end
r2 = Ractor.new r1 do |r1|
r1.take
end
puts r2.take
Ractor.newで渡した引数は|arg|のようにブロック内で使用できる変数として受け取ることができます。
例えば以下のコードはr1でtakeメソッドが実行されるまで処理を待ちます。
r1 = Ractor.new 42 do |arg|
Ractor.yield arg
end
またRactor.newには別のRactorを渡すことができるため以下のように書くことができます。
r2 = Ractor.new r1 do |r1|
r1.take
end
これでr1が引数として受け取った42をr2の中で受け取ることができます。
最後にputs r2.takeで42を受け取って出力しています。
pull型はこういった流れになります。
ざっくりと解説すると
push型: Ractor#send + Ractor.recvpull型: Ractor.yield + Ractor#take
という感じです。
より詳細なRactorの解説に関しては下記のリンクを参照していただければと思います。
Ractorのコード
Ractorの生成
RactorはRactor.newでブロックに実行したい処理を書きます。
Ractor.new do
end
このブロック内の処理が並行実行されます。
つまり、以下のようなコードの場合
Ractor.new do
10.times do
puts :hoge
end
end
10.times do
puts :fuga
end
:hogeと:fugaがそれぞれ並行に出力されます。
またRactor.newに実行したい処理をブロックとして渡しますので以下のように書くこともできます。
Ractor.new{
10.times{
puts :hoge
}
}
またキーワード引数nameを使って名前を付けることもでき、Ractor#nameで名前をを受け取ることもできます。
r = Ractor.new name: 'r1' do
puts :hoge
end
p r.name
これにより、どのRactorで処理が実行されているかを確認することもできそうです。
Ractorへ引数を渡す
Ractor.newに引数を渡すことでブロック内にオブジェクトを渡すことができます。
r = Ractor.new :hoge do |a|
p a
end
r.take
このように引数を経由してオブジェクトを渡すことができます。
また複数の引数を渡すこともできます
r = Ractor.new :hoge, :fuga do |a, b|
p a
p b
end
r.take
このようにArrayを渡すこともできます。
r = Ractor.new [:hoge, :fuga] do |a|
p a.inspect
end
r.take
ちなみに、|a|を|a, b|に変更すると
r = Ractor.new [:hoge, :fuga] do |a, b|
p a
p b
end
r.take
という出力結果になります。これはa, b = [:hoge, :fuga]と同じ挙動と解釈されているようです。
またHashの場合は
r = Ractor.new({:hoge => 42, :fuga => 21}) do |a|
p a
p a[:hoge]
end
r.take
と出力されます。
ちなみに、Ractor.newの後に()で括っていないとSyntaxErrorになるので注意が必要です。
r = Ractor.new({:hoge => 42, :fuga => 21}) do |a|
p a
p a[:hoge]
end
r.take
Ractorでの返り値
Ractorでは実行されるブロック内の返り値をtakeメソッドで受け取ることができます。
r = Ractor.new do
:hoge
end
p r.take
ちなみに、ブロック内でreturnをするとLocalJumpErrorとなるようです。
r = Ractor.new do
return :fuga
:hoge
end
p r.take
Ractor内での例外
Ractor内での例外は以下のようにして受け取ることができます。
r = Ractor.new do
raise 'error'
end
begin
r.take
rescue Ractor::RemoteError => e
p e.message
end
ちなみに、Ractor内でも例外処理を書くことはできます。
r = Ractor.new name: 'r1' do
begin
raise 'error'
rescue => e
p e.message
end
end
r.take
またドキュメントによるとRactorのブロック内から返ってきた値を受け取る領域で例外をキャッチできるようです。
つまり以下のようなコードも書くことができます。
r1 = Ractor.new do
raise 'error'
end
r2 = Ractor.new r1 do |r1|
begin
r1.take
rescue Ractor::RemoteError => e
p e.message
end
end
r2.take
Ractorでの並行実行
簡単な例
こんな感じでRactorで並行実行を行うことができます。
Ractor.new do
3.times do
puts 42
end
end
3.times do
puts 21
end
実行すると42と21という出力がそれぞればらばらに表示されます。
ちょっとした例
以下のような感じで複数のworkerをRactorで生成し、それをpipeで経由して値を渡し、結果をまとめることができます。
require 'prime'
pipe = Ractor.new do
loop do
Ractor.yield Ractor.recv
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}
このコードでは10個のworkerを生成し、pipeを経由してそれぞれのworkerにオブジェクトを渡しています。
また、受け取ったオブジェクトをRactor.yield [n, n.prime?]で返しています。
こんな感じでworkerを複数作り、pipe経由で処理させたり、結果を受け取ることができます。
よしなにworkerなどを生成して処理させるクラスを書いてみる
先ほどのコードだとworker内の処理が後々で大きくなることもありそうだったので、以下のようにworkerをよしなに生成してくれるクラスを書いてみました。
class Ninsoku
def initialize(task, worker_count: 10)
@task = task
@pipe = create_pipe
@workers = create_workers(worker_count)
end
def send(arg)
@pipe.send arg
end
def run
yield Ractor.select(*@workers)
end
def create_pipe
Ractor.new do
loop do
Ractor.yield Ractor.recv
end
end
end
def create_workers(worker_count)
(1..worker_count).map do
Ractor.new @pipe, @task do |pipe, task|
loop do
arg = pipe.take
task.send arg
Ractor.yield task.take
end
end
end
end
end
Ninsoku.newでpipeとworkerを生成しています。またtaskは処理させたい内容をRactorで渡し、workerで実行させています。
実際に使うケースとしてはこんな感じです。
task = Ractor.new do
func = lambda{|n| n.downcase }
loop do
Ractor.yield func.call(Ractor.recv)
end
end
ninsoku = Ninsoku.new(task)
('A'..'Z').each{|i|
ninsoku.send i
}
('A'..'Z').map{
ninsoku.run{|r, n|
puts n
}
}
このクラスはあとでgemにでもしてみようかと思います。
gemにしてみました(rubygemsにはpushしていませんが……)
S-H-GAMELINKS/rorker
例えば、僕の住んでいる島根県浜田市のAED位置情報をRactorを使って処理してみます。
なお、島根県のAED位置情報は島根県が公開しているオープンデータを使用させていただきました。この場を借りて感謝を申し上げます。
島根県 オープンデータカタログサイト
require "rorker"
require "csv"
task = Ractor.new do
func = lambda{|row|
row.map{|value|
if value =~ /浜田市/
row
end
}.compact
}
loop do
Ractor.yield func.call(Ractor.recv)
end
end
rorker = Rorker.new(task)
csv = CSV.read "a.csv"
csv.each do |row|
rorker.send row
end
n = 0
while n < csv.count
rorker.run{|worker, result|
if !result.empty?
puts result
end
}
n += 1
end
こんな感じで読み取ったCSVを一行づつworkerに渡して並行処理で必要なデータをとってくることもできます。
RactorでNumbered Parameterを使う
Ractorでは処理をブロックで渡しますので、Numbered Parameterを使って引数を受け取ることもできます。
r = Ractor.new :hoge do
puts _1
end
r.take
ちなみに、複数の引数でも動作します。
r = Ractor.new :hoge, :hoge do
puts _1
puts _2
end
r.take
複数渡した場合は渡された順番通りに_1から_9に渡されるみたいです。
ちなみに、Hashを渡した場合はこんな感じになります。
r = Ractor.new ({hoge: 1, fuga: 2}) do
_1.map do |key, value|
p ":#{key} => #{value}"
end
end
r.take
=>を使ったHashも同様の結果になりました
r = Ractor.new({:hoge => 1, :fuga => 2}) do
_1.map do |key, value|
p ":#{key} => #{value}"
end
end
r.take
ただし、Arrayの場合は少し挙動が異なります
r = Ractor.new [1, 2, 3] do
puts _1
puts _1.class
puts _2
puts _2.class
puts _3
puts _3.class
end
r.take
どうやら通常通り複数の引数を渡した時のようにArrayの先頭から順番に渡されるようです。
おそらくは以下のように解釈されているのではないかと思います。
_1, _2, _3 = [1, 2, 3]
ちなみに、Numbered Parameterで受け取れる数より大きなArrayを渡すと
r = Ractor.new [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] do
puts _1
puts _2
puts _3
puts _4
puts _5
puts _6
puts _7
puts _8
puts _9
end
r.take
このように取得できるのはNumbered Parameterが受け取れる範囲までのようです
Ractor内でNumbered Parameterを使う場合はHashを引数に渡した際か
r = Ractor.new ({hoge: 1, fuga: 2}) do |hash|
hash.map do
p ":#{_1} => #{_2}"
end
end
r.take
":hoge => 1"
":fuga => 2"
またはいくつかの引数を渡した時に省略して書きたいときに使うことになりそうです
r = Ractor.new :hoge, :fuga do
p _1
p _2
end
r.take
おわりに
この記事を読んでRactorについて興味を持って頂ければ幸いです。
今後もRactorを使って試したコードを追加していこうと思います
参考
