Thrad Pool for arbitarily killing in Ruby

thread_pool.rb

#!/bin/ruby
require 'thread'

class ThreadPool
  def initialize(size, &session)
    @size = size
    @queue = SizedQueue.new(size)
    @map = {}
    @map_mutex = Mutex.new
    session.call(self)
  rescue
    #p "#{Thread.current} rescue #{$!}"
  ensure
    #p "#{Thread.current} shutdown"
    shutdown
  end

  def run(name,&job)
    @queue.push([name,job])
    @map_mutex.synchronize {
      return false if @map[name]
      @map[name] = Thread.start(@queue,@map,@map_mutex) { |q,map,mutex|
        begin
          n,j = q.pop
          j.call
        rescue
        ensure
          mutex.synchronize {
            map.delete(n)
          }
        end
      }
    }
  end

  def kill(name)
    @map_mutex.synchronize {
      @map[name].kill if @map[name] and @map[name].alive?
      @map.delete(name)
      #p @map
    }
  end
  
  def alive?(name)
    return nil unless @map[name]
    @map[name].alive?
  end

  def shutdown
    @map_mutex.synchronize {
      @map.each { |name,th|
        th.kill if th.alive?
        @map.delete(name)
      }
    }
  end

  def join
    @map.each { |name,th|
      th.join
      @map_mutex.synchronize {
        @map.delete(name)
      }
    }
  end
end

if __FILE__ == $0

  ThreadPool.new(100) do |pool|
    10.times do |n|
      job = Proc.new {
        wn = n
        wait = rand(100).to_f/10
        puts "start #{wn}: #{Thread.current} (#{wait} sec)\n"
        sleep(wait)
        puts "stop  #{wn}: #{Thread.current} (#{wait} sec)\n"
      }
      p "#{n}times job : #{job}"
      pool.run("p#{n}",&job)
      sleep(0.01)
    end

    pool.kill("p4")
    pool.kill("p7")

    pool.join
  end

end