<p>I couldn't find a decent ThreadPool implementation for Ruby, so I wrote mine (based partly on code from here: <a href="http://web.archive.org/web/20081204101031/http://snippets.dzone.com:80/posts/show/3276" rel="nofollow noreferrer">http://web.archive.org/web/20081204101031/http://snippets.dzone.com:80/posts/show/3276</a> , but changed to wait/signal and other implementation for ThreadPool shutdown. However after some time of running (having 100 threads and handling about 1300 tasks), it dies with deadlock on line 25 - it waits for a new job there. Any ideas, why it might happen?</p>

<pre><code>require 'thread'
begin
  require 'fastthread'
rescue LoadError
  $stderr.puts "Using the ruby-core thread implementation"
end 

class ThreadPool
  class Worker
    def initialize(callback)
      @mutex = Mutex.new
      @cv = ConditionVariable.new
      @callback = callback
      @mutex.synchronize {@running = true}
      @thread = Thread.new do
        while @mutex.synchronize {@running}
          block = get_block
          if block
            block.call
            reset_block
            # Signal the ThreadPool that this worker is ready for another job
            @callback.signal
          else
            # Wait for a new job
            @mutex.synchronize {@cv.wait(@mutex)} # &lt;=== Is this line 25?
          end
        end
      end
    end

    def name
      @thread.inspect
    end

    def get_block
      @mutex.synchronize {@block}
    end

    def set_block(block)
      @mutex.synchronize do
        raise RuntimeError, "Thread already busy." if @block
        @block = block
        # Signal the thread in this class, that there's a job to be done
        @cv.signal
      end
    end

    def reset_block
      @mutex.synchronize {@block = nil}
    end

    def busy?
      @mutex.synchronize {!@block.nil?}
    end

    def stop
      @mutex.synchronize {@running = false}
      # Signal the thread not to wait for a new job
      @cv.signal
      @thread.join
    end
  end

  attr_accessor :max_size

  def initialize(max_size = 10)
    @max_size = max_size
    @workers = []
    @mutex = Mutex.new
    @cv = ConditionVariable.new
  end

  def size
    @mutex.synchronize {@workers.size}
  end

  def busy?
    @mutex.synchronize {@workers.any? {|w| w.busy?}}
  end

  def shutdown
    @mutex.synchronize {@workers.each {|w| w.stop}}
  end
  alias :join :shutdown

  def process(block=nil,&amp;blk)
    block = blk if block_given?
    while true
      @mutex.synchronize do
         worker = get_worker 
         if worker
           return worker.set_block(block)
         else
           # Wait for a free worker
           @cv.wait(@mutex)
         end
      end
    end
  end

  # Used by workers to report ready status
  def signal
    @cv.signal
  end

  private
  def get_worker
    free_worker || create_worker
  end

  def free_worker
    @workers.each {|w| return w unless w.busy?}; nil
  end

  def create_worker
    return nil if @workers.size &gt;= @max_size
    worker = Worker.new(self)
    @workers &lt;&lt; worker
    worker
  end
end
</code></pre>

[{'answer_id': 82034, 'author': 'mweerden', 'author_id': 4285, 'author_profile': 'https://Stackoverflow.com/users/4285', 'pm_score': 1, 'selected': False, 'text': "<p>I'm slightly biased here, but I would suggest modelling this in some process language and model check it. Freely available tools are, for example, the mCRL2 toolset (using a ACP-based language), the Mobility Workbench (pi-calculus) and Spin (PROMELA).</p>\n\n<p>Otherwise I would suggest removing every bit of code that is not essential to the problem and finding a minimal case where the deadlock occurs. I doubt that it the 100 threads and 1300 tasks are essential to get a deadlock. With a smaller case you can probably just add some debug prints which provide enough information the solve the problem.</p>\n"}, {'answer_id': 82061, 'author': 'PierreBdR', 'author_id': 7136, 'author_profile': 'https://Stackoverflow.com/users/7136', 'pm_score': 1, 'selected': False, 'text': "<p>Ok, the problem seems to be in your ThreadPool#signal method. What may happen is:</p>\n\n<p>1 - All your worker are busy and you try to process a new job</p>\n\n<p>2 - line 90 gets a nil worker</p>\n\n<p>3 - a worker get freed and signals it, but the signal is lost as the ThreadPool is not waiting for it</p>\n\n<p>4 - you fall on line 95, waiting even though there is a free worker.</p>\n\n<p>The error here is that you can signal a free worker even when nobody is listening. This ThreadPool#signal method should be:</p>\n\n<pre><code>def signal\n     @mutex.synchronize { @cv.signal }\nend\n</code></pre>\n\n<p>And the problem is the same in the Worker object. What might happen is:</p>\n\n<p>1 - The Worker just completed a job</p>\n\n<p>2 - It checks (line 17) if there is a job waiting: there isn't</p>\n\n<p>3 - The thread pool send a new job and signals it ... but the signal is lost</p>\n\n<p>4 - The worker wait for a signal, even though it is marked as busy</p>\n\n<p>You should put your initialize method as:</p>\n\n<pre><code>def initialize(callback)\n  @mutex = Mutex.new\n  @cv = ConditionVariable.new\n  @callback = callback\n  @mutex.synchronize {@running = true}\n  @thread = Thread.new do\n    @mutex.synchronize do\n      while @running\n        block = get_block\n        if block\n          @mutex.unlock\n          block.call\n          @mutex.lock\n          reset_block\n          # Signal the ThreadPool that this worker is ready for another job\n          @callback.signal\n        else\n          # Wait for a new job\n          @cv.wait(@mutex)\n        end\n      end\n    end\n  end\nend\n</code></pre>\n\n<p>Next, the Worker#get_block and Worker#reset_block methods should not be synchronized anymore. That way, you cannot have a block assigned to a worker between the test for a block and the wait for a signal.</p>\n"}, {'answer_id': 82777, 'author': 'PierreBdR', 'author_id': 7136, 'author_profile': 'https://Stackoverflow.com/users/7136', 'pm_score': 5, 'selected': True, 'text': '<p>Ok, so the main problem with the implementation is: how to make sure no signal is lost and avoid dead locks ?</p>\n\n<p>In my experience, this is REALLY hard to achieve with condition variables and mutex, but easy with semaphores. It so happens that ruby implement an object called Queue (or SizedQueue) that should solve the problem. Here is my suggested implementation:</p>\n\n<pre><code>require \'thread\'\nbegin\n  require \'fasttread\'\nrescue LoadError\n  $stderr.puts "Using the ruby-core thread implementation"\nend\n\nclass ThreadPool\n  class Worker\n    def initialize(thread_queue)\n      @mutex = Mutex.new\n      @cv = ConditionVariable.new\n      @queue = thread_queue\n      @running = true\n      @thread = Thread.new do\n        @mutex.synchronize do\n          while @running\n            @cv.wait(@mutex)\n            block = get_block\n            if block\n              @mutex.unlock\n              block.call\n              @mutex.lock\n              reset_block\n            end\n            @queue &lt;&lt; self\n          end\n        end\n      end\n    end\n\n    def name\n      @thread.inspect\n    end\n\n    def get_block\n      @block\n    end\n\n    def set_block(block)\n      @mutex.synchronize do\n        raise RuntimeError, "Thread already busy." if @block\n        @block = block\n        # Signal the thread in this class, that there\'s a job to be done\n        @cv.signal\n      end\n    end\n\n    def reset_block\n      @block = nil\n    end\n\n    def busy?\n      @mutex.synchronize { !@block.nil? }\n    end\n\n    def stop\n      @mutex.synchronize do\n        @running = false\n        @cv.signal\n      end\n      @thread.join\n    end\n  end\n\n  attr_accessor :max_size\n\n  def initialize(max_size = 10)\n    @max_size = max_size\n    @queue = Queue.new\n    @workers = []\n  end\n\n  def size\n    @workers.size\n  end\n\n  def busy?\n    @queue.size &lt; @workers.size\n  end\n\n  def shutdown\n    @workers.each { |w| w.stop }\n    @workers = []\n  end\n\n  alias :join :shutdown\n\n  def process(block=nil,&amp;blk)\n    block = blk if block_given?\n    worker = get_worker\n    worker.set_block(block)\n  end\n\n  private\n\n  def get_worker\n    if !@queue.empty? or @workers.size == @max_size\n      return @queue.pop\n    else\n      worker = Worker.new(@queue)\n      @workers &lt;&lt; worker\n      worker\n    end\n  end\n\nend\n</code></pre>\n\n<p>And here is a simple test code:</p>\n\n<pre><code>tp = ThreadPool.new 500\n(1..1000).each { |i| tp.process { (2..10).inject(1) { |memo,val| sleep(0.1); memo*val }; print "Computation #{i} done. Nb of tasks: #{tp.size}\\n" } }\ntp.shutdown\n</code></pre>\n'}, {'answer_id': 2063180, 'author': 'Filipe Miguel Fonseca', 'author_id': 112744, 'author_profile': 'https://Stackoverflow.com/users/112744', 'pm_score': 3, 'selected': False, 'text': '<p>You can try the <a href="http://gemcutter.org/gems/work_queue" rel="nofollow noreferrer">work_queue</a> gem, designed to coordinate work between a producer and a pool of worker threads.</p>\n'}, {'answer_id': 51769759, 'author': 'bking', 'author_id': 7915982, 'author_profile': 'https://Stackoverflow.com/users/7915982', 'pm_score': 1, 'selected': False, 'text': '<p>Top commenter\'s code has helped out so much over the years.  Here it is updated for ruby 2.x and improved with thread identification.  How is that an improvement?  When each thread has an ID, you can compose ThreadPool with an array which stores arbitrary information.  Some ideas:</p>\n\n<ul>\n<li>No array: typical ThreadPool usage.  Even with the GIL it makes threading dead easy to code and very useful for high-latency applications like high-volume web crawling,</li>\n<li>ThreadPool and Array sized to number of CPUs: easy to fork processes to use all CPUs,</li>\n<li>ThreadPool and Array sized to number of resources: e.g., each array element represents one processor across a pool of instances, so if you have 10 instances each with 4 CPUs, the TP can manage work across 40 subprocesses.</li>\n</ul>\n\n<p>With these last two, rather than thinking about threads doing work think about the ThreadPool managing subprocesses that are doing the work.  The management task is lightweight and when combined with subprocesses, who cares about the GIL.  </p>\n\n<p>With this class, you can code up a cluster based MapReduce in about a hundred lines of code!  This code is beautifully short although it can be a bit of a mind-bend to fully grok.  Hope it helps.</p>\n\n<pre><code># Usage:\n#\n#   Thread.abort_on_exception = true # help localize errors while debugging\n#   pool = ThreadPool.new(thread_pool_size)\n#   50.times {|i|\n#     pool.process { ... }\n#     or\n#     pool.process {|id| ... } # worker identifies itself as id\n#   }\n#   pool.shutdown()\n\nclass ThreadPool\n\n  require \'thread\'\n\n  class ThreadPoolWorker\n\n    attr_accessor :id\n\n    def initialize(thread_queue, id)\n      @id = id # worker id is exposed thru tp.process {|id| ... }\n      @mutex = Mutex.new\n      @cv = ConditionVariable.new\n      @idle_queue = thread_queue\n      @running = true\n      @block = nil\n      @thread = Thread.new {\n        @mutex.synchronize {\n          while @running\n            @cv.wait(@mutex) # block until there is work to do\n            if @block\n              @mutex.unlock\n              begin\n                @block.call(@id)\n              ensure\n                @mutex.lock\n              end\n              @block = nil\n            end\n            @idle_queue &lt;&lt; self\n          end\n        }\n      }\n    end\n\n    def set_block(block)\n      @mutex.synchronize {\n        raise RuntimeError, "Thread is busy." if @block\n        @block = block\n        @cv.signal # notify thread in this class, there is work to be done\n      }\n    end\n\n    def busy?\n      @mutex.synchronize { ! @block.nil? }\n    end\n\n    def stop\n      @mutex.synchronize {\n        @running = false\n        @cv.signal\n      }\n      @thread.join\n    end\n\n    def name\n      @thread.inspect\n    end\n  end\n\n\n  attr_accessor :max_size, :queue\n\n  def initialize(max_size = 10)\n    @process_mutex = Mutex.new\n    @max_size = max_size\n    @queue = Queue.new # of idle workers\n    @workers = []      # array to hold workers\n\n    # construct workers\n    @max_size.times {|i| @workers &lt;&lt; ThreadPoolWorker.new(@queue, i) }\n\n    # queue up workers (workers in queue are idle and available to\n    # work).  queue blocks if no workers are available.\n    @max_size.times {|i| @queue &lt;&lt; @workers[i] }\n\n    sleep 1 # important to give threads a chance to initialize\n  end\n\n  def size\n    @workers.size\n  end\n\n  def idle\n    @queue.size\n  end\n\n  # are any threads idle\n\n  def busy?\n    # @queue.size &lt; @workers.size\n    @queue.size == 0 &amp;&amp; @workers.size == @max_size\n  end\n\n  # block until all threads finish\n\n  def shutdown\n    @workers.each {|w| w.stop }\n    @workers = []\n  end\n\n  alias :join :shutdown\n\n  def process(block = nil, &amp;blk)\n    @process_mutex.synchronize {\n      block = blk if block_given?\n      worker = @queue.pop # assign to next worker; block until one is ready\n      worker.set_block(block) # give code block to worker and tell it to start\n    }\n  end\n\n\nend\n</code></pre>\n'}]
['https://Stackoverflow.com/questions/81788', 'https://Stackoverflow.com', 'https://Stackoverflow.com/users/12695/']