Ractor超入門

## はじめに

この記事は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`はそれぞれ並行して実行されます。

```ruby
Ractor.new do
  5.times do
    puts :hello
  end
end

5.times do
    puts :world
end
```

このコードを実行すると以下のような結果になります。

```bash
world
helloworld

hello
world
helloworld

helloworld

hello
```

このようにそれぞれの処理を並行して実行できます。

また`Ractor`は別の`Ractor`へとオブジェクトを送受信し、同期しながら実行することもできます。同期の方法としては`push型`と`pull型`の二つがあります。

たとえば`push型`の場合は以下のようなコードになります。

```ruby
r1 = Ractor.new do
    :hoge
end

r2 = Ractor.new do
    puts :fuga, Ractor.recv
end

r2.send(r1.take)

r2.take
# => :fuga, :hoge
```

`Ractor`では`send`メソッドを使い、別の`Ractor`へとオブジェクトを送ることができます。上記のコードだと

```ruby
r2.send(r1.take)
```

の部分で`r2`へと送信しています。

送信されたオブジェクトは`Ractor`内で`Ractor.recv`で受け取ることができ

```ruby
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型`は以下のようなコードになります。

```ruby
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`メソッドが実行されるまで処理を待ちます。

```ruby
r1 = Ractor.new 42 do |arg|
    Ractor.yield arg
end
```

また`Ractor.new`には別の`Ractor`を渡すことができるため以下のように書くことができます。

```ruby
r2 = Ractor.new r1 do |r1|
    r1.take
end
```

これで`r1`が引数として受け取った`42`を`r2`の中で受け取ることができます。

最後に`puts r2.take`で`42`を受け取って出力しています。

`pull型`はこういった流れになります。

ざっくりと解説すると

- `push型`: `Ractor#send` + `Ractor.recv`
- `pull型`: `Ractor.yield` + `Ractor#take`

という感じです。

より詳細な`Ractor`の解説に関しては下記のリンクを参照していただければと思います。

- [A proposal of new concurrency model for Ruby 3](http://rubykaigi.org/2016/presentations/ko1.html)
- [Guild Prototype](https://rubykaigi.org/2018/presentations/ko1.html)
- [Ruby向け並列化機構Guildの試作](https://atdot.net/~ko1/activities/sasada_ipsj_pro_120.pdf)
- [Guild → Ractor](http://atdot.net/~ko1/activities/2020_ruby3summit.pdf)
- [[JA] Ractor report / Koichi Sasada ko1](https://youtu.be/40t8EPpnujg)
- [https://github.com/ko1/ruby/blob/ractor/ractor.ja.md](https://github.com/ko1/ruby/blob/ractor/ractor.ja.md)
- [Ractor - Ruby's Actor-like concurrent abstraction](https://github.com/ruby/ruby/blob/master/doc/ractor.md)
- [Ractor: a proposal for a new concurrent abstraction without thread-safety issues](https://bugs.ruby-lang.org/issues/17100)


## Ractorのコード
### Ractorの生成

`Ractor`は`Ractor.new`でブロックに実行したい処理を書きます。

```ruby
Ractor.new do
  # このブロックが並行に実行される
end
```

このブロック内の処理が並行実行されます。

つまり、以下のようなコードの場合

```ruby
Ractor.new do
    10.times do
        puts :hoge
    end
end

10.times do
    puts :fuga
end
```

`:hoge`と`:fuga`がそれぞれ並行に出力されます。

また`Ractor.new`に実行したい処理をブロックとして渡しますので以下のように書くこともできます。

```ruby
Ractor.new{
    10.times{
        puts :hoge
    }
}
```

またキーワード引数`name`を使って名前を付けることもでき、`Ractor#name`で名前をを受け取ることもできます。

```ruby
r = Ractor.new name: 'r1' do
    puts :hoge
end

p r.name
# => "r1"
```

これにより、どの`Ractor`で処理が実行されているかを確認することもできそうです。

### Ractorへ引数を渡す

`Ractor.new`に引数を渡すことでブロック内にオブジェクトを渡すことができます。

```ruby
r = Ractor.new :hoge do |a|
    p a
end

r.take
# => :hoge
```

このように引数を経由してオブジェクトを渡すことができます。


また複数の引数を渡すこともできます

```ruby
r = Ractor.new :hoge, :fuga do |a, b|
    p a
    p b
end

r.take
# => fuga
# => hoge
```

このように`Array`を渡すこともできます。

```ruby
r = Ractor.new [:hoge, :fuga] do |a|
    p a.inspect
end

r.take
# => "[:hoge, :fuga]"
```

ちなみに、`|a|`を`|a, b|`に変更すると

```ruby
r = Ractor.new [:hoge, :fuga] do |a, b|
    p a
    p b
end

r.take
# => :hoge
# => :fuga
```

という出力結果になります。これは`a, b = [:hoge, :fuga]`と同じ挙動と解釈されているようです。

また`Hash`の場合は

```ruby
r = Ractor.new({:hoge => 42, :fuga => 21}) do |a|
    p a
    p a[:hoge]
end

r.take
# => {:hoge=>42, :fuga=>21}
# => 42
```

と出力されます。
ちなみに、`Ractor.new`の後に`()`で括っていないと`SyntaxError`になるので注意が必要です。

```ruby
r = Ractor.new({:hoge => 42, :fuga => 21}) do |a|
    p a
    p a[:hoge]
end

r.take
# => SyntaxError
```


### Ractorでの返り値

Ractorでは実行されるブロック内の返り値を`take`メソッドで受け取ることができます。

```ruby
r = Ractor.new do
    :hoge
end

p r.take
# => :hoge
```

ちなみに、ブロック内で`return`をすると`LocalJumpError`となるようです。

```ruby
r = Ractor.new do
    return :fuga
    :hoge
end

p r.take
# => LocalJumpError
```

### Ractor内での例外

Ractor内での例外は以下のようにして受け取ることができます。

```ruby
r = Ractor.new do
    raise 'error'
end

begin
    r.take
rescue Ractor::RemoteError => e
    p e.message
end
```

ちなみに、Ractor内でも例外処理を書くことはできます。

```ruby
r = Ractor.new name: 'r1' do
    begin
        raise 'error'
    rescue => e
        p e.message
    end
end

r.take
```

またドキュメントによるとRactorのブロック内から返ってきた値を受け取る領域で例外をキャッチできるようです。
つまり以下のようなコードも書くことができます。

```ruby

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
# => "thrown by remote Ractor."
```

### Ractorでの並行実行
#### 簡単な例

こんな感じでRactorで並行実行を行うことができます。

```ruby
Ractor.new do
    3.times do
        puts 42
    end
end

3.times do
    puts 21
end
```

実行すると`42`と`21`という出力がそれぞればらばらに表示されます。

#### ちょっとした例

以下のような感じで複数の`worker`を`Ractor`で生成し、それを`pipe`で経由して値を渡し、結果をまとめることができます。

```ruby
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}
# => 0 ~ 999 までの数値が素数かどうかの結果を出力
```

このコードでは10個の`worker`を生成し、`pipe`を経由してそれぞれの`worker`にオブジェクトを渡しています。
また、受け取ったオブジェクトを`Ractor.yield [n, n.prime?]`で返しています。

こんな感じで`worker`を複数作り、`pipe`経由で処理させたり、結果を受け取ることができます。

#### よしなにworkerなどを生成して処理させるクラスを書いてみる

先ほどのコードだと`worker`内の処理が後々で大きくなることもありそうだったので、以下のように`worker`をよしなに生成してくれるクラスを書いてみました。

```ruby
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`で実行させています。

実際に使うケースとしてはこんな感じです。

```ruby
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
    }
}
# => a ~ z までが並行して出力される
```

~~このクラスはあとで`gem`にでもしてみようかと思います。~~

`gem`にしてみました(rubygemsにはpushしていませんが……)

[S-H-GAMELINKS/rorker](https://github.com/S-H-GAMELINKS/rorker)

例えば、僕の住んでいる島根県浜田市のAED位置情報をRactorを使って処理してみます。
なお、島根県のAED位置情報は島根県が公開しているオープンデータを使用させていただきました。この場を借りて感謝を申し上げます。

[島根県 オープンデータカタログサイト](https://shimane-opendata.jp/)

```ruby
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`を使って引数を受け取ることもできます。

```ruby
r = Ractor.new :hoge do
    puts _1
end

r.take
# => hoge
```

ちなみに、複数の引数でも動作します。

```ruby
r = Ractor.new :hoge, :hoge do
    puts _1
    puts _2
end

r.take
# => hoge
# => fuga
```

複数渡した場合は渡された順番通りに`_1`から`_9`に渡されるみたいです。

ちなみに、Hashを渡した場合はこんな感じになります。

```ruby
r = Ractor.new ({hoge: 1, fuga: 2}) do
    _1.map do |key, value|
        p ":#{key} => #{value}"
    end
end

r.take
# => ":hoge => 1"
# => ":fuga => 2"
```

`=>`を使ったHashも同様の結果になりました

```ruby
r = Ractor.new({:hoge => 1, :fuga => 2}) do
    _1.map do |key, value|
        p ":#{key} => #{value}"
    end
end

r.take
# => ":hoge => 1"
# => ":fuga => 2"
```

ただし、Arrayの場合は少し挙動が異なります

```ruby
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
#=> 1
#=> Integer
#=> 2
#=> Integer
#=> 3
#=> Integer
```

どうやら通常通り複数の引数を渡した時のようにArrayの先頭から順番に渡されるようです。
おそらくは以下のように解釈されているのではないかと思います。

```ruby
_1, _2, _3 = [1, 2, 3]
```

ちなみに、`Numbered Parameter`で受け取れる数より大きな`Array`を渡すと

```ruby
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
#=> 1
#=> 2
#=> 3
#=> 4
#=> 5
#=> 6
#=> 7
#=> 8
#=> 9
```

このように取得できるのは`Numbered Parameter`が受け取れる範囲までのようです


`Ractor`内で`Numbered Parameter`を使う場合は`Hash`を引数に渡した際か

```ruby
r = Ractor.new ({hoge: 1, fuga: 2}) do |hash|
    hash.map do
        p ":#{_1} => #{_2}"
    end
end

r.take
":hoge => 1"
":fuga => 2"
```

またはいくつかの引数を渡した時に省略して書きたいときに使うことになりそうです

```ruby
r = Ractor.new :hoge, :fuga do
    p _1
    p _2
end

r.take
# => :hoge
# => :fuga
```

## おわりに

この記事を読んで`Ractor`について興味を持って頂ければ幸いです。
今後も`Ractor`を使って試したコードを追加していこうと思います

## 参考

- [A proposal of new concurrency model for Ruby 3](http://rubykaigi.org/2016/presentations/ko1.html)
- [Guild Prototype](https://rubykaigi.org/2018/presentations/ko1.html)
- [Ruby向け並列化機構Guildの試作](https://atdot.net/~ko1/activities/sasada_ipsj_pro_120.pdf)
- [Guild → Ractor](http://atdot.net/~ko1/activities/2020_ruby3summit.pdf)
- [[JA] Ractor report / Koichi Sasada ko1](https://youtu.be/40t8EPpnujg)
- [https://github.com/ko1/ruby/blob/ractor/ractor.ja.md](https://github.com/ko1/ruby/blob/ractor/ractor.ja.md)
- [Ractor - Ruby's Actor-like concurrent abstraction](https://github.com/ruby/ruby/blob/master/doc/ractor.md)
- [https://gist.github.com/niku/c19da11edf0b97470af27844b44d12fa](https://gist.github.com/niku/c19da11edf0b97470af27844b44d12fa)
- [Ractor: a proposal for a new concurrent abstraction without thread-safety issues](https://bugs.ruby-lang.org/issues/17100)