Ruby マルチスレッド

システム上で実行中の各プログラムはプロセスです。各プロセスは1つ以上のスレッドを含みます。

スレッドはプログラム内の単一の順次制御フローです。単一のプログラム内で複数のスレッドを同時に実行して異なる作業を完了することをマルチスレッドと呼びます。

RubyではThreadクラスを使ってマルチスレッドを作成できます。Rubyのスレッドは軽量で、効率的な方法で並列コードを実装できます。

Ruby スレッドの作成

新しいスレッドを起動するには、Thread.new を呼び出すだけです:

#スレッド #1 コード部分 Thread.new { #スレッド #2 実行コード } #スレッド #1 実行コード

例

以下の例は、Rubyプログラムでマルチスレッドを使用する方法を示しています：

例

#!/usr/bin/ruby def func1 i=0 while i<=2 puts "func1 at: #{Time.now}" sleep(2) i=i+1 end end def func2 j=0 while j<=2 puts "func2 at: #{Time.now}" sleep(1) j=j+1 end end puts "Started At #{Time.now}" t1=Thread.new{func1()} t2=Thread.new{func2()} t1.join t2.join puts "End at #{Time.now}"

試してみる »

上記のコードの実行結果は次のとおりです：

Started At Wed May 14 08:21:54 -0700 2014
func1 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:55 -0700 2014
func1 at: Wed May 14 08:21:56 -0700 2014
func2 at: Wed May 14 08:21:56 -0700 2014
func1 at: Wed May 14 08:21:58 -0700 2014
End at Wed May 14 08:22:00 -0700 2014

スレッドのライフサイクル

1. スレッドの作成にはThread.newを使用でき、同じ構文でThread.startまたはThread.forkを使用して、これら3つのメソッドでスレッドを作成することもできます。

2. スレッドを作成した後、起動する必要はありません。スレッドは自動的に実行されます。

3. Threadクラスはスレッドを操作するためのいくつかのメソッドを定義しています。スレッドはThread.new内のコードブロックを実行します。

4. スレッドのコードブロック内の最後の文がスレッドの値です。スレッドのメソッドを介して呼び出すことができます。スレッドが実行を完了するとスレッドの値を返し、そうでない場合はスレッドが実行を完了するまで値を返しません。

5. Thread.currentメソッドは現在のスレッドを表すオブジェクトを返します。Thread.mainメソッドはメインスレッドを返します。

6. Thread.Joinメソッドを使用してスレッドを実行します。このメソッドは、現在のスレッドが実行を完了するまでメインスレッドを一時停止させます。

スレッドの状態

スレッドには5つの状態があります：

スレッドの状態戻り値
実行可能run
スリープSleeping
終了aborting
正常終了false
例外による終了nil

スレッドと例外

あるスレッドで例外が発生し、それがrescueで捕捉されない場合、そのスレッドは通常、警告なしに終了します。しかし、Thread#joinによってそのスレッドを待機している他のスレッドがある場合、待機中のスレッドでも同じ例外が発生します。

begin t = Thread.new do Thread.pass #メインスレッドは確かにjoinを待機しています raise "unhandled exception" end t.join rescue p $! # => "unhandled exception" end

次の3つの方法を使用すると、あるスレッドが例外によって終了したときにインタプリタの実行を中断させることができます。

  • スクリプト起動時に指定-dオプションを指定し、デバッグモードで実行します。
  • 用Thread.abort_on_exceptionフラグを設定します。
  • 使用してThread#abort_on_exception指定されたスレッドにフラグを設定します。

上記の3つの方法のいずれかを使用すると、インタプリタ全体が中断されます。

t = Thread.new { ... } t.abort_on_exception = true

スレッド同期制御

Rubyでは、同期を実装する3つの方法が提供されています。それぞれ次のとおりです：

1. Mutexクラスによるスレッド同期

2. データの受け渡しを管理するQueueクラスによるスレッド同期

3. ConditionVariableを使用した同期制御

Mutexクラスによるスレッド同期の実装

Mutexクラスを使用してスレッド同期制御を実装します。複数のスレッド内で同時に1つのプログラム変数が必要な場合、その変数の一部をlockでロックできます。コードは次のとおりです：

例

#!/usr/bin/ruby require "thread" puts "Synchronize Thread" @num=200 @mutex=Mutex.new def buyTicket(num) @mutex.lock if @num>=num @num=@num-num puts "you have successfully bought #{num} tickets" else puts "sorry,no enough tickets" end @mutex.unlock end ticket1=Thread.new 10 do 10.times do |value| ticketNum=15 buyTicket(ticketNum) sleep 0.01 end end ticket2=Thread.new 10 do 10.times do |value| ticketNum=20 buyTicket(ticketNum) sleep 0.01 end end sleep 1 ticket1.join ticket2.join

試してみる »

上記のコードの実行結果は次のとおりです：

Synchronize Thread
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets

変数をロックするにはlockを使用するだけでなく、try_lockを使用して変数をロックしたり、Mutex.synchronizeを使用して特定の変数へのアクセスを同期したりすることもできます。

データの受け渡しを管理するQueueクラスによるスレッド同期の実装

Queueクラスはスレッドをサポートするキューを表し、キューの末尾へのアクセスを同期できます。異なるスレッドは同じQueueクラスを使用できますが、このキュー内のデータが同期されるかどうかを心配する必要はありません。さらに、SizedQueueクラスを使用するとキューの長さを制限できます。

SizedQueueクラスは、スレッド同期アプリケーションの開発を非常に便利にします。なぜなら、このキューに追加するだけで、スレッドの同期問題を気にする必要がないからです。

古典的な生産者・消費者問題：

例

#!/usr/bin/ruby require "thread" puts "SizedQuee Test" queue = Queue.new producer = Thread.new do 10.times do |i| sleep rand(i) #スレッドを一定時間スリープさせる queue << i puts "#{i} produced" end end consumer = Thread.new do 10.times do |i| value = queue.pop sleep rand(i/2) puts "consumed #{value}" end end consumer.join

試してみる »

プログラムの出力：

SizedQuee Test
0 produced
1 produced
consumed 0
2 produced
consumed 1
consumed 2
3 produced
consumed 34 produced

consumed 4
5 produced
consumed 5
6 produced
consumed 6
7 produced
consumed 7
8 produced
9 produced
consumed 8
consumed 9

スレッド変数

スレッドはプライベート変数を持つことができます。スレッドのプライベート変数は、スレッド作成時にスレッドに書き込まれます。スレッドの範囲内で使用できますが、スレッド外部で共有することはできません。

しかし、スレッドのローカル変数を他のスレッドやメインスレッドからアクセスする必要がある場合はどうすればよいでしょうか？ Rubyでは、名前を指定してスレッド変数を作成できます。これは、スレッドをハッシュのような散列表とみなすものです。[]=でデータを書き込み、[]でデータを読み出します。次のコードを見てみましょう：

例

#!/usr/bin/ruby count = 0 arr = [] 10.times do |i| arr[i] = Thread.new { sleep(rand(0)/10.0) Thread.current["mycount"] = count count += 1 } end arr.each {|t| t.join; print t["mycount"], ", " } puts "count = #{count}"

上記のコードの実行出力結果は次のとおりです：

8, 0, 3, 7, 2, 1, 6, 5, 4, 9, count = 10

メインスレッドはサブスレッドの実行完了を待ち、その後各値をそれぞれ出力します。

スレッドの優先度

スレッドの優先度は、スレッドのスケジューリングに影響を与える主な要因です。他の要因には、CPUの実行時間の長さ、スレッドのグループスケジューリングなどがあります。

Thread.priority メソッドを使用してスレッドの優先度を取得でき、Thread.priority= メソッドを使用してスレッドの優先度を調整できます。

スレッドの優先度はデフォルトで0です。優先度が高いほど、実行が速くなります。

Thread は自身のスコープ内のすべてのデータにアクセスできますが、あるスレッド内で他のスレッドのデータにアクセスする必要がある場合はどうすればよいでしょうか？ Thread クラスは、スレッドデータを相互にアクセスするためのメソッドを提供しています。スレッドを単純に Hash テーブルとして扱うことができ、任意のスレッド内で []= を使用してデータを書き込み、 [] を使用してデータを読み出すことができます。

athr = Thread.new { Thread.current["name"] = "Thread A"; Thread.stop } bthr = Thread.new { Thread.current["name"] = "Thread B"; Thread.stop } cthr = Thread.new { Thread.current["name"] = "Thread C"; Thread.stop } Thread.list.each {|x| puts "#{x.inspect}: #{x["name"]}" }

ご覧のとおり、スレッドを Hash テーブルとして扱い、 [] と []= メソッドを使用することで、スレッド間のデータ共有を実現できます。

スレッドの相互排他

Mutex（Mutal Exclusion = 相互排他ロック）は、マルチスレッドプログラミングにおいて、2つのスレッドが同じ共有リソース（グローバル変数など）に対して同時に読み書きすることを防ぐためのメカニズムです。

Mutaxを使用しない例

例

#!/usr/bin/ruby require 'thread' count1 = count2 = 0 difference = 0 counter = Thread.new do loop do count1 += 1 count2 += 1 end end spy = Thread.new do loop do difference += (count1 - count2).abs end end sleep 1 puts "count1 : #{count1}" puts "count2 : #{count2}" puts "difference : #{difference}"

上記の例の実行出力結果は次のとおりです：

count1 :  9712487
count2 :  12501239
difference : 0

mutexを使用した例

例

#!/usr/bin/ruby require 'thread' mutex = Mutex.new count1 = count2 = 0 difference = 0 counter = Thread.new do loop do mutex.synchronize do count1 += 1 count2 += 1 end end end spy = Thread.new do loop do mutex.synchronize do difference += (count1 - count2).abs end end end sleep 1 mutex.lock puts "count1 : #{count1}" puts "count2 : #{count2}" puts "difference : #{difference}"

上記の例の実行出力結果は次のとおりです：

count1 :  1336406
count2 :  1336406
difference : 0

デッドロック

2つ以上の演算ユニットが、互いに相手の停止を待ってシステムリソースを取得しようとしているが、どちらも先に終了しない場合、この状況をデッドロックと呼びます。

例えば、プロセス p1 がディスプレイを占有し、さらにプリンタを使用しなければならないとします。しかしプリンタはプロセス p2 に占有されており、p2 はディスプレイを使用しなければならない場合、デッドロックが発生します。

Mutex オブジェクトを使用する際は、スレッドのデッドロックに注意する必要があります。

例

#!/usr/bin/ruby require 'thread' mutex = Mutex.new cv = ConditionVariable.new a = Thread.new { mutex.synchronize { puts "A: I have critical section, but will wait for cv" cv.wait(mutex) puts "A: I have critical section again! I rule!" } } puts "(Later, back at the ranch...)" b = Thread.new { mutex.synchronize { puts "B: Now I am critical, but am done with cv" cv.signal puts "B: I am still critical, finishing up" } } a.join b.join

上記の例の出力結果は次のとおりです：

A: I have critical section, but will wait for cv
(Later, back at the ranch...)
B: Now I am critical, but am done with cv
B: I am still critical, finishing up
A: I have critical section again! I rule!

スレッドのクラスメソッド

完全な Thread（スレッド）クラスメソッドは次のとおりです：

番号メソッドの説明
1Thread.abort_on_exception
この値が真の場合、あるスレッドが例外によって終了すると、インタプリタ全体が中断されます。デフォルト値は偽です。つまり、通常の場合、あるスレッドで例外が発生し、その例外がThread#joinなどで検出されないと、そのスレッドは警告なしに終了します。
2Thread.abort_on_exception=
設定するとtrue、スレッドが例外によって終了すると、インタプリタ全体が中断されます。新しい状態を返します。
3Thread.critical
ブール値を返します。
4Thread.critical=
その値がtrueの場合、スレッドの切り替えは行われません。現在のスレッドが停止（stop）しているか、シグナル（signal）による割り込みがある場合、その値は自動的にfalseになります。
5Thread.current
現在実行中のスレッド（現在のスレッド）を返します。
6Thread.exit
現在のスレッドの実行を終了します。現在のスレッドを返します。現在のスレッドが唯一のスレッドである場合、exit(0)を使用してその実行を終了します。
7Thread.fork { block }
Thread.new と同じようにスレッドを生成します。
8Thread.kill( aThread )
スレッドの実行を終了します。
9Thread.list
実行状態または停止状態にある生存スレッドの配列を返します。
10Thread.main
メインスレッドを返します。
11Thread.new( [ arg ]* ) {| args | block }
スレッドを生成し、実行を開始します。引数はブロックにそのまま渡されます。これにより、スレッドを起動すると同時に、そのスレッド固有のローカル変数に値を渡すことができます。
12Thread.pass
他のスレッドに実行権を譲ります。実行中のスレッドの状態は変更せず、他の実行可能なスレッドに制御を渡します（明示的なスレッドスケジューリング）。
13Thread.start( [ args ]* ) {| args | block }
スレッドを生成し、実行を開始します。引数はブロックにそのまま渡されます。これにより、スレッドを起動すると同時に、そのスレッド固有のローカル変数に値を渡すことができます。
14Thread.stop
他のスレッドがrunメソッドでこのスレッドを再び起こすまで、現在のスレッドを停止させます。

スレッドのインスタンスメソッド

次の例では、スレッドのインスタンスメソッド join を呼び出しています：

例

#!/usr/bin/ruby thr = Thread.new do #インスタンス化 puts "In second thread" raise "Raise exception" end thr.join #インスタンスメソッド join を呼び出します

以下は完全なインスタンスメソッドの一覧です：

番号メソッドの説明
1thr[ name ]
スレッド内の name に対応する固有データを取得します。name には文字列またはシンボルを指定できます。name に対応するデータがない場合は nil を返します。
2thr[ name ] =
スレッド内の name に対応する固有データの値を設定します。name には文字列またはシンボルを指定できます。nil を設定した場合、そのスレッド内の対応するデータは削除されます。
3thr.abort_on_exception
ブール値を返します。
4thr.abort_on_exception=
その値が true の場合、あるスレッドが例外によって終了すると、インタプリタ全体が中断されます。
5thr.alive?
スレッドが「生きている」場合は true を返します。
6thr.exit
スレッドの実行を終了します。self を返します。
7thr.join
現在のスレッドを停止させ、self スレッドが終了するまで待機します。self が例外によって終了した場合、現在のスレッドでも同じ例外が発生します。
8thr.key?
name に対応するスレッド固有データが定義されていれば true を返します。
9thr.kill
類似Thread.exit 。
10thr.priority
スレッドの優先度を返します。優先度のデフォルト値は 0 です。この値が大きいほど優先度が高くなります。
11thr.priority=
スレッドの優先度を設定します。負の数に設定することもできます。
12thr.raise( anException )
そのスレッド内で強制的に例外を発生させます。
13thr.run
停止（stop）状態のスレッドを再起動します。wakeup とは異なり、即座にスレッドの切り替えを行います。死んだスレッドに対してこのメソッドを使用すると、ThreadError 例外が発生します。
14thr.safe_level
self のセキュリティレベルを返します。現在のスレッドの safe_level は $SAFE と同じです。
15thr.status
生存スレッドの状態を文字列 “run”、“sleep”、“aborting” で表します。スレッドが正常終了した場合は false を返します。例外により終了した場合は nil を返します。
16thr.stop?
スレッドが終了状態（dead）または停止（stop）状態にある場合は true を返します。
17thr.value
self スレッドが終了するまで待機し（join と同等）、そのスレッドのブロックの戻り値を返します。スレッドの実行中に例外が発生した場合は、その例外が再度発生します。
18thr.wakeup
停止（stop）状態のスレッドの状態を実行可能状態（run）に変更します。死んだスレッドに対してこのメソッドを実行すると、ThreadError 例外が発生します。
その他の拡張