Rubyの並列処理とグローバルインタプリタロックの関係

hotty

2013.12.11

hottyです。
ソニックムーブ Advent Calendar 2013 12/11(水)の記事になります。
今回はRubyの並列処理について書こうと思います。

Rubyの処理系は1.8まではユーザレベルで行うグリーンスレッドでしたが、
1.9からはMRIにYARVという処理系が組み込まれ、
カーネルがスレッドの管理を行うネイティブスレッドになりました。

では、早速ですがThreadクラスで複数のスレッドを作って並列処理を行ってみましょう。

[sourcecode lang="ruby"]
list = ["A", "B", "C", "D"]
io   = File.open("result.log", "w")

list.each do |name|
  thread = Thread.fork(name) do |name|
    for count in 10000.times
      io.puts name
    end
  end
  thread.join
end
[/sourcecode]

[sourcecode lang="ruby"]

list = ["A", "B", "C", "D"]

io = File.open("result.log", "w")

list.each do |name|

thread = Thread.fork(name) do |name|

for count in 10000.times

io.puts name

end

thread.join

end

[/sourcecode]

メインスレッドはforkしたスレッドの終了を待ってくれないので
Thread#joinでforkしたスレッドの終了をメインスレッドが待つようにしています。
一見問題なくそれぞれA〜Dまでの4つのスレッドが並列で動いてそうです。

しかしこれは本当に「並列」なのでしょうか・・・？
もし本当に並列で処理ができていたらどれか4つのスレッドの内どれか1つが停止しても
残りの3つのスレッドは停止せず走り続けるはずです。

ここでスレッドBを10秒間停止してみます。
またわかりやすいように、
イテレータの処理が終わったスレッドはメッセージを出力に出すようにしました。

[sourcecode lang="ruby" highlight="6,10"]
list = ["A", "B", "C", "D"]
io   = File.open("result.log", "w")

list.each do |name|
  thread = Thread.fork(name) do |name|
    sleep(10) if name == "B"
    for count in 10000.times
      io.puts name
    end
    puts "Thread#{name} completed!"
  end
  thread.join
end
[/sourcecode]

[sourcecode lang="ruby" highlight="6,10"]

list = ["A", "B", "C", "D"]

io = File.open("result.log", "w")

list.each do |name|

thread = Thread.fork(name) do |name|

sleep(10) if name == "B"

for count in 10000.times

io.puts name

end

puts "Thread#{name} completed!"

end

thread.join

end

[/sourcecode]

これを do.rbで保存して実行してみます。

[sourcecode lang="bash"]
$ ruby do.rb 
ThreadA completed!
[/sourcecode]

[sourcecode lang="bash"]

$ ruby do.rb

ThreadA completed!

[/sourcecode]

しーーーーーーーーーーーーーーーーーーーーん

————およそ10秒経過後

[sourcecode lang="bash"]
$ ruby do.rb 
ThreadA completed!
ThreadB completed!
ThreadC completed!
ThreadD completed!
[/sourcecode]

[sourcecode lang="bash"]

$ ruby do.rb

ThreadA completed!

ThreadB completed!

ThreadC completed!

ThreadD completed!

[/sourcecode]

にょきにょき！

おわかりでしょうか。
Bスレッドを停止している間、他のC、Dスレッドも停止しているようでした。
という事は・・・、もしや A → B → C → D と1つずつちまちまと実行しているのでしょうか・・・！？
psにHオプションをつけてスレッドの動きを見てみます。

[sourcecode lang="bash"]
$ ruby do.rb 
ThreadA completed!
[/sourcecode]

[sourcecode lang="bash"]

$ ruby do.rb

ThreadA completed!

[/sourcecode]

10秒間のスリープ中です。
おー・・・？
ruby do.rb が 3つほどあります。
スレッドAは役目を終えてdeadしてるので
この3つはプロセスとメインスレッドとスレッドBのようですね。

————およそ10秒経過後

・・・結局プロセスが終了するまでこれ以上スレッドが増える事はありませんでした。
なんという事でしょう。
こっちは並列で処理しているつもりがスレッド1つ1つ処理をしてたとは・・・。
何のためのネイティブスレッドですか！

一体なぜこのような結果になってしまうかというと、
「グローバルインタプリタロック」というロックが原因です。
Rubyでは「グローバルバーチャルマシンロック(以下GVL)」という呼ばれ方をします。

Ruby1.9ではスレッドセーフでないCで拡張したライブラリなどに考慮しているのか
複数のスレッドの同時実行(並列処理)を許可しておらず
実行できるスレッドはGVLを持つスレッドに制限されています。

並列処理は制御されていたんですねー。
どうやらGVLを持ったスレッドBが停止している間
スレッドC、スレッドDはスレッドBの排他ロックにより動けずにいたようです。

じゃあRubyで並列処理もできないの？というとそうではありません。
公式によると、

ただし、IO 関連のブロックする可能性があるシステムコールを行う場合には GVL を解放します。その場合にはスレッドは同時に実行され得ます。
また拡張ライブラリから GVL を操作できるので、複数のスレッドを同時に実行するような拡張ライブラリは作成可能です。

と書いてあります。

複数のスレッドを同時に実行するような拡張ライブラリ・・・？
もしや、Parallelのこと…！？
という事で Parallelを使って書きなおしてみましょう。
ちなみに Parallelは
gem install parallel
で手に入ります。

[sourcecode lang="ruby"]
require "parallel"
list = ["A", "B", "C", "D"]
io   = File.open("result.log", "w")

Parallel.map(list, :in_threads => 4) do |name|
  for count in 10000.times
    io.puts name
  end
  puts "Thread#{name} completed!"
end
[/sourcecode]

[sourcecode lang="ruby"]

require "parallel"

list = ["A", "B", "C", "D"]

io = File.open("result.log", "w")

Parallel.map(list, :in_threads => 4) do |name|

for count in 10000.times

io.puts name

end

puts "Thread#{name} completed!"

end

[/sourcecode]

Threadクラスを使うよりスッキリしてます。
この mapというイテレータ侮れませんね！
では実行してみます。

ちゃんと並列処理させれていますね！
やっぱり Parallelは素晴らしいですね！

以上、Rubyの並列処理のお話でした。

2013/4/23 追記

twitterで間違ってるって話題だよ(早く直せ！)
とブログ担当に言われ確認してみたところ…
joinの位置がおかしいだけでGVL関係ありませんでした(泣)

parallelのソースを読むのを怠ってしまったせいでなんかparallelの中でGVLを操作してるんだろうなーと勝手に思ってました；ｳｯ

つーわけで
例で使用した mapメソッドについてparallelのソースを読んで行っている処理を確認してみました。

①mapメソッドからwork_in_threadsメソッドが呼ばれる。
②work_in_threadsメソッドからin_threadsメソッドがブロック付きで呼ばれる。
③in_threadsメソッドでは mapの第2引数の in_threads => 4 で指定した数字分 Thread.newをブロック付きで呼び出している。
④Thread.new do〜end では yield によりスタックから ②のブロックの内容が評価される。
⑤④の処理の中でwith_instrumentationメソッドをブロック付きで呼び出している。
⑥⑤のブロックでcall_with_indexメソッドを呼び出している。
⑦call_with_indexメソッドでは Parallel.map do〜endの処理(クロージャ化されている)を callで呼び出している。

みたいな流れ。
うむ、全然GVL関係ないですね。

で、in_threadsメソッドでスレッドを全てforkし終わったあとに、ユーザがCtrl+cを押した時にforkしたプロセスやスレッドがkillされるような処理が書いてありました。
その処理でyieldで呼ばれてたのがforkしたスレッドの配列にjoinする処理。
これだ。
つまり、メインスレッドが全てのスレッドをforkし終わったあと joinしなきゃだめってこと。

[sourcecode lang="ruby"]
list = ["A", "B", "C", "D"]
io   = File.open("result.log", "w")
 
list.each do |name|
  thread = Thread.fork(name) do |name|
    for count in 10000.times
      io.puts name
    end
    puts "Thread#{name} completed!"
  end
  thread.join
end
[/sourcecode]

[sourcecode lang="ruby"]

list = ["A", "B", "C", "D"]

io = File.open("result.log", "w")

list.each do |name|

thread = Thread.fork(name) do |name|

for count in 10000.times

io.puts name

end

puts "Thread#{name} completed!"

end

thread.join

end

[/sourcecode]

これの何が間違っているかっていうと、

スレッド1つfork
↓
join
↓
joinしたのでforkしたスレッドの処理が終わるまでメインスレッド停止
↓
次のスレッドをforkするメインスレッドが停止してるんだから当然次のスレッドが作られない（当たり前だ）
↓
結果、スレッド作成 → 作成したスレッド実行 → 作成したスレッド終了 → 次のスレッド作成 → 作成したスレッド実行 → 作成したスレッド終了 → …

という具合に順番に処理を行っていたのです。

だから正しく書くと

[sourcecode lang="ruby"]
list = ["A", "B", "C", "D"]
io   = File.open("result.log", "w")
threads = []

list.each do |name|
  threads << Thread.fork(name) do |name|
    for count in 10000.times
      io.puts name
    end
    puts "Thread#{name} completed!";
  end
end

threads.each do |thread|
  thread.join
end
[/sourcecode]