[Ruby] スレッド数固定で処理を行う

お仕事で、数百台のサーバに対して、多少時間のかかる処理をする必要があった。

1 つずつ実行していては、さすがに日が暮れてしまう。
そうなるとスレッドで並列実行したいが、対象のサーバの分、数百個スレッドを生成すると逆に重くなってしまう。

というわけで、スレッド数に上限を設けて実行した、という話。

雛形はこんな感じ。

#!/usr/bin/ruby -Ku
# -*- coding: utf-8 -*-
#
# 20120619
# thread template
#

require 'thread'


def job(arg)
  sleep rand(0)
  print "#{arg}\n"
end


def main
  # スレッド数上限
  thread_max = 3

  # キュー
  jobqueue = Queue.new
  (1..10).each do |n|
    jobqueue.push(n)
  end

  # スレッドで処理
  threads = []
  thread_max.times do
    threads << Thread.start do
      while !jobqueue.empty?      # (*1)
        var = jobqueue.pop        # (*2)
        job(var)
      end
    end
  end

  # スレッド完了待ち
  threads.each {|t| t.join}
end

### main ###
main

jobqueue にジョブを追加して、thread_max 数のスレッドで jobqueue から取り出しながら処理する。
複数のレジがあって、客の行列が 1 つあるイメージでしょうか。

実際は、キューに対象の IP を push して、thread_max = 30 にする感じで使いました。

while !jobqueue.empty? してから pop しているのがポイントといえばポイント。
while var = jobqueue.pop とすると、pop でキューが空のときスレッドが sleep してしまい、以下のように、deadlock してしまう。

% ./test_bad.rb
1
2
5
6
4
8
7
3
9
10
deadlock 0x7fcca4541b90: sleep:-  - ./test_bad.rb:31
deadlock 0x7fcca455f370: sleep:J(0x7fcca4541b90) (main) - ./test_bad.rb:38
deadlock 0x7fcca45418c0: sleep:-  - ./test_bad.rb:31
deadlock 0x7fcca4541a28: sleep:-  - ./test_bad.rb:31
./test_bad.rb:31: [BUG] Segmentation fault
ruby 1.8.7 (2011-06-30 patchlevel 352) [x86_64-linux]

zsh: abort      ./test_bad.rb

• 追記 (2012/06/25)

このコードだとまだ問題がある事に後から気づいた。
*1 でキューが空かどうか確認してから、*2 での pop までの間に、他のスレッドが pop をすると、やはり deadlock してしまう。
Mutex を使うなどして排他制御するといいだろう。