Ruby-マルチスレッド

従来のプログラムには単一の実行スレッドがあり、プログラムを構成するステートメントまたは命令は、プログラムが終了するまで順番に実行されます。

マルチスレッドプログラムには、複数の実行スレッドがあります。各スレッド内では、ステートメントは順番に実行されますが、スレッド自体は、たとえばマルチコアCPUで並列に実行される場合があります。多くの場合、単一のCPUマシンでは、複数のスレッドが実際には並列に実行されませんが、並列処理はスレッドの実行をインターリーブすることによってシミュレートされます。

Rubyを使用すると、Threadクラスを使用してマルチスレッドプログラムを簡単に作成できます。Rubyスレッドは、コードの同時実行性を実現するための軽量で効率的な方法です。

Rubyスレッドの作成

新しいスレッドを開始するには、ブロックをThread.newの呼び出しに関連付けるだけです。ブロック内のコードを実行するために新しいスレッドが作成され、元のスレッドはThread.newからすぐに戻り、次のステートメントで実行を再開します-

# Thread #1 is running here
Thread.new {
   # Thread #2 runs this code
}
# Thread #1 runs this code

例

これは、マルチスレッドの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 2008
func1 at: Wed May 14 08:21:54 -0700 2008
func2 at: Wed May 14 08:21:54 -0700 2008
func2 at: Wed May 14 08:21:55 -0700 2008
func1 at: Wed May 14 08:21:56 -0700 2008
func2 at: Wed May 14 08:21:56 -0700 2008
func1 at: Wed May 14 08:21:58 -0700 2008
End at Wed May 14 08:22:00 -0700 2008

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

Thread.newを使用して新しいスレッドが作成されます。同義語のThread.startとThread.forkを使用することもできます。

スレッドの作成後にスレッドを開始する必要はありません。CPUリソースが使用可能になると、スレッドは自動的に実行を開始します。

Threadクラスは、実行中にスレッドを照会および操作するためのいくつかのメソッドを定義します。スレッドは、Thread.newの呼び出しに関連付けられたブロック内のコードを実行してから、実行を停止します。

そのブロックの最後の式の値はスレッドの値であり、Threadオブジェクトのvalueメソッドを呼び出すことで取得できます。スレッドが完了するまで実行された場合、値はスレッドの値をすぐに返します。それ以外の場合、valueメソッドはブロックされ、スレッドが完了するまで戻りません。

クラスメソッドThread.currentは、現在のスレッドを表すThreadオブジェクトを返します。これにより、スレッドは自分自身を操作できます。クラスメソッドThread.mainは、メインスレッドを表すThreadオブジェクトを返します。これは、Rubyプログラムが開始されたときに開始された最初の実行スレッドです。

特定のスレッドのThread.joinメソッドを呼び出すことにより、そのスレッドが終了するのを待つことができます。呼び出し元のスレッドは、指定されたスレッドが終了するまでブロックします。

スレッドと例外

メインスレッドで例外が発生し、どこでも処理されない場合、Rubyインタープリターはメッセージを出力して終了します。メインスレッド以外のスレッドでは、未処理の例外によりスレッドの実行が停止します。

スレッドの場合 t 未処理の例外と別のスレッドが原因で終了します st.joinまたはt.valueを呼び出してから、で発生した例外t スレッドで発生します s。

Thread.abort_on_exceptionがデフォルトの条件であるfalseの場合、未処理の例外は単に現在のスレッドを強制終了し、残りはすべて実行を継続します。

スレッド内の未処理の例外によってインタープリターが終了するようにする場合は、クラスメソッドThread.abort_on_exceptionをtrueに設定します。

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

スレッド変数

スレッドは通常、スレッドの作成時にスコープ内にあるすべての変数にアクセスできます。スレッドのブロックにローカルな変数はスレッドにローカルであり、共有されません。

スレッドクラスは、スレッドローカル変数を作成して名前でアクセスできるようにする特別な機能を備えています。スレッドオブジェクトをハッシュであるかのように扱い、[] =を使用して要素に書き込み、[]を使用して要素を読み戻します。

この例では、各スレッドは、変数countの現在の値をキーmycountを使用してthreadlocal変数に記録します。

#!/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時間を取得します。

優先度=および優先度を使用して、Rubyスレッドオブジェクトの優先度を設定およびクエリできます。新しく作成されたスレッドは、それを作成したスレッドと同じ優先度で開始されます。メインスレッドは優先度0から始まります。

実行を開始する前にスレッドの優先度を設定する方法はありません。ただし、スレッドは、最初に実行するアクションとして、スレッド自体の優先度を上げたり下げたりすることができます。

スレッドの除外

2つのスレッドが同じデータへのアクセスを共有し、少なくとも1つのスレッドがそのデータを変更する場合は、スレッドが一貫性​​のない状態でデータを参照できないように特別な注意を払う必要があります。これはスレッド除外と呼ばれます。

Mutexは、共有リソースへの相互に排他的なアクセスのための単純なセマフォロックを実装するクラスです。つまり、一度に1つのスレッドだけがロックを保持できます。他のスレッドは、ロックが使用可能になるのを順番に待つことを選択する場合もあれば、ロックが使用できないことを示す即時エラーを取得することを選択する場合もあります。

共有データへのすべてのアクセスをミューテックスの制御下に置くことにより、一貫性とアトミック操作を保証します。例を試してみましょう。最初の例はmutaxなしで、2番目の例はmutaxありです-

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 :  1583766
count2 :  1583766
difference : 0

#!/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 :  696591
count2 :  696591
difference : 0

デッドロックの処理

スレッドの除外にミューテックスオブジェクトを使用し始めるときは、デッドロックを回避するように注意する必要があります。デッドロックは、すべてのスレッドが別のスレッドによって保持されているリソースの取得を待機しているときに発生する状態です。すべてのスレッドがブロックされているため、保持しているロックを解放することはできません。また、ロックを解放できないため、他のスレッドはそれらのロックを取得できません。

ここで、条件変数が明らかになります。条件変数は、単にリソースに関連付けられている特定の保護範囲内で使用されているセマフォであるミューテックス。利用できないリソースが必要な場合は、条件変数を待ちます。このアクションにより、対応するミューテックスのロックが解除されます。他のスレッドがリソースが使用可能であることを通知すると、元のスレッドは待機を解除し、同時にクリティカル領域のロックを回復します。

例

#!/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!

スレッドの状態

次の表に示すように、5つの可能な状態に対応する5つの可能な戻り値があります。ステータスメソッドは、スレッドの状態を返します。

スレッドの状態	戻り値
実行可能	実行
睡眠	睡眠
中絶	中絶
正常に終了	false
例外を除いて終了	nil

スレッドクラスメソッド

以下のメソッドはThreadクラスによって提供され、プログラムで使用可能なすべてのスレッドに適用できます。これらのメソッドは、次のようにThreadクラス名を使用して呼び出されます-

Thread.abort_on_exception = true

利用可能なすべてのクラスメソッドの完全なリストは次のとおりです-

シニア番号	方法と説明
1	Thread.abort_on_exception 例外条件でグローバルアボートのステータスを返します。デフォルトはfalseです。trueに設定すると、いずれかのスレッドで例外が発生した場合、すべてのスレッドが中止されます（プロセスはexit（0）になります）。
2	Thread.abort_on_exception= trueに設定すると、例外が発生した場合にすべてのスレッドが中止されます。新しい状態を返します。
3	Thread.critical グローバルスレッドのクリティカル状態のステータスを返します。
4	Thread.critical= グローバルスレッドのクリティカル状態のステータスを設定し、それを返します。trueに設定すると、既存のスレッドのスケジューリングを禁止します。新しいスレッドの作成と実行をブロックしません。特定のスレッド操作（スレッドの停止または強制終了、現在のスレッドでのスリープ、例外の発生など）により、クリティカルセクションにある場合でもスレッドがスケジュールされる場合があります。
5	Thread.current 現在実行中のスレッドを返します。
6	Thread.exit 現在実行中のスレッドを終了し、別のスレッドを実行するようにスケジュールします。このスレッドがすでに強制終了としてマークされている場合、exitはスレッドを返します。これがメインスレッドまたは最後のスレッドである場合は、プロセスを終了します。
7	Thread.fork { block } Thread.newの同義語。
8	Thread.kill( aThread ) 指定されたスレッドを終了させます
9	Thread.list 実行可能または停止中のすべてのスレッドのThreadオブジェクトの配列を返します。糸。
10	Thread.main プロセスのメインスレッドを返します。
11	Thread.new( [ arg ]* ) {| args | block } ブロックで指定された命令を実行するための新しいスレッドを作成し、実行を開始します。Thread.newに渡された引数はすべて、ブロックに渡されます。
12	Thread.pass スレッドスケジューラを呼び出して、実行を別のスレッドに渡します。
13	Thread.start( [ args ]* ) {| args | block } 基本的にThread.newと同じです。ただし、クラスThreadがサブクラス化されている場合、そのサブクラスでstartを呼び出しても、サブクラスのinitializeメソッドは呼び出されません。
14	Thread.stop 現在のスレッドの実行を停止してスリープ状態にし、別のスレッドの実行をスケジュールします。クリティカル状態をfalseにリセットします。

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

これらのメソッドは、スレッドのインスタンスに適用できます。これらのメソッドは、次のようにスレッドのインスタンスを使用して呼び出されます-

#!/usr/bin/ruby

thr = Thread.new do   # Calling a class method new
   puts "In second thread"
   raise "Raise exception"
end
thr.join   # Calling an instance method join

利用可能なすべてのインスタンスメソッドの完全なリストは次のとおりです-

シニア番号	方法と説明
1	thr[ aSymbol ] 属性リファレンス-シンボルまたはaSymbol名のいずれかを使用して、スレッドローカル変数の値を返します。指定された変数が存在しない場合、nilを返します。
2	thr[ aSymbol ] = 属性の割り当て-シンボルまたは文字列のいずれかを使用して、スレッドローカル変数の値を設定または作成します。
3	thr.abort_on_exception thrの例外条件での中止のステータスを返します。デフォルトはfalseです。
4	thr.abort_on_exception= trueに設定すると、thrで例外が発生した場合に、すべてのスレッド（メインプログラムを含む）が中止されます。プロセスは事実上exit（0）になります。
5	thr.alive? thrが実行中またはスリープ中の場合はtrueを返します。
6	thr.exit thrを終了し、実行する別のスレッドをスケジュールします。このスレッドがすでに強制終了としてマークされている場合、exitはスレッドを返します。これがメインスレッドまたは最後のスレッドである場合、プロセスを終了します。
7	thr.join 呼び出し元のスレッドは実行を一時停止し、thrを実行します。thrが終了するまで戻りません。参加していないスレッドは、メインプログラムが終了すると強制終了されます。
8	thr.key? 指定された文字列（またはシンボル）がスレッドローカル変数として存在する場合、trueを返します。
9	thr.kill 同義語Thread.exit。
10	thr.priority thrの優先度を返します。デフォルトはゼロです。優先度の高いスレッドは、優先度の低いスレッドの前に実行されます。
11	thr.priority= thrの優先度を整数に設定します。優先度の高いスレッドは、優先度の低いスレッドの前に実行されます。
12	thr.raise( anException ) thrから例外を発生させます。発信者はthrである必要はありません。
13	thr.run thrを起動し、スケジューリングの対象にします。クリティカルセクションにない場合は、スケジューラを呼び出します。
14	thr.safe_level thrに対して有効な安全レベルを返します。
15	thr.status 戻り値の状態THR：睡眠場合THRは寝たり、I / O上で待機している実行する場合はTHRがあれば、偽を実行しているTHRが正常に終了し、nilの場合THRは例外で終了。
16	thr.stop? thrが死んでいるか眠っている場合はtrueを返します。
17	thr.value thrがThread.joinを介して完了するのを待ち、その値を返します。
18	thr.wakeup thrをスケジューリングの対象としてマークしますが、I / Oでブロックされたままになる可能性があります。