Взаимоблокировка в ThreadPool

Question

Я не смог найти достойную реализацию ThreadPool для Ruby, поэтому написал свою (частично основанную на коде отсюда: http://web.archive.org/web/20081204101031/http://snippets.dzone.com:80/posts/show/3276 , но изменен на wait / signal и другую реализацию для завершения работы ThreadPool.Однако после некоторого времени выполнения (имея 100 потоков и обрабатывая около 1300 задач) он останавливается с тупиковой ситуацией в строке 25 - там он ожидает нового задания.Есть какие-нибудь идеи, почему это может произойти?

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)} # <=== 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,&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 >= @max_size
    worker = Worker.new(self)
    @workers << worker
    worker
  end
end

Answer 1

Итак, основная проблема с реализацией заключается в следующем:как убедиться , что сигнал не потерян , и избежать мертвых блокировок ?

По моему опыту, этого ДЕЙСТВИТЕЛЬНО трудно достичь с помощью переменных условий и мьютекса, но легко с помощью семафоров.Так получилось, что ruby реализует объект с именем Queue (или SizedQueue), который должен решить проблему.Вот моя предлагаемая реализация:

require 'thread'
begin
  require 'fasttread'
rescue LoadError
  $stderr.puts "Using the ruby-core thread implementation"
end

class ThreadPool
  class Worker
    def initialize(thread_queue)
      @mutex = Mutex.new
      @cv = ConditionVariable.new
      @queue = thread_queue
      @running = true
      @thread = Thread.new do
        @mutex.synchronize do
          while @running
            @cv.wait(@mutex)
            block = get_block
            if block
              @mutex.unlock
              block.call
              @mutex.lock
              reset_block
            end
            @queue << self
          end
        end
      end
    end

    def name
      @thread.inspect
    end

    def get_block
      @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
      @block = nil
    end

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

    def stop
      @mutex.synchronize do
        @running = false
        @cv.signal
      end
      @thread.join
    end
  end

  attr_accessor :max_size

  def initialize(max_size = 10)
    @max_size = max_size
    @queue = Queue.new
    @workers = []
  end

  def size
    @workers.size
  end

  def busy?
    @queue.size < @workers.size
  end

  def shutdown
    @workers.each { |w| w.stop }
    @workers = []
  end

  alias :join :shutdown

  def process(block=nil,&blk)
    block = blk if block_given?
    worker = get_worker
    worker.set_block(block)
  end

  private

  def get_worker
    if !@queue.empty? or @workers.size == @max_size
      return @queue.pop
    else
      worker = Worker.new(@queue)
      @workers << worker
      worker
    end
  end

end

А вот простой тестовый код:

tp = ThreadPool.new 500
(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" } }
tp.shutdown

Answer 2

Вы можете попробовать рабочая очередь gem, предназначенный для координации работы между производителем и пулом рабочих потоков.

Answer 3

Я здесь немного предвзят, но я бы предложил смоделировать это на каком-нибудь языке процессов и проверить это на модели.Свободно доступными инструментами являются, например, набор инструментов mCRL2 (использующий язык на основе ACP), Mobility Workbench (pi-calculus) и Spin (PROMELA).

В противном случае я бы предложил удалить каждый фрагмент кода, который не является существенным для проблемы, и найти минимальный случай, когда возникает взаимоблокировка.Я сомневаюсь, что 100 потоков и 1300 задач необходимы для того, чтобы зайти в тупик.В меньшем случае вы, вероятно, можете просто добавить несколько отладочных отпечатков, которые предоставляют достаточно информации для решения проблемы.

Answer 4

Хорошо, проблема, похоже, в вашем методе ThreadPool #signal.Что может произойти, так это:

1. Все ваши сотрудники заняты, и вы пытаетесь выполнить новое задание

2 - строка 90 получает нулевого работника

3 - рабочий освобождается и сигнализирует об этом, но сигнал теряется, так как ThreadPool его не ожидает

4 - вы попадаете на линию 95, ожидая, хотя там есть свободный работник.

Ошибка здесь заключается в том, что вы можете подать сигнал свободному работнику, даже если его никто не слушает.Этот метод ThreadPool #signal должен быть:

def signal
     @mutex.synchronize { @cv.signal }
end

И проблема та же в Рабочем объекте.Что может произойти, так это:

1 - Работник только что выполнил задание

2 - Он проверяет (строка 17), есть ли ожидающее задание:там его нет

3 - Пул потоков отправляет новое задание и сигнализирует об этом ...но сигнал потерян

4 - Рабочий ожидает сигнала, даже если он помечен как занятый

Вы должны поместить свой метод инициализации следующим образом:

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

Далее, методы Worker#get_block и Worker#reset_block больше не должны синхронизироваться.Таким образом, вы не можете назначить блок рабочему между тестированием блока и ожиданием сигнала.

Answer 5

Код лучшего комментатора так сильно помогал на протяжении многих лет.Здесь он обновлен для ruby 2.x и улучшен с помощью идентификации потоков.Насколько это улучшило ситуацию?Когда у каждого потока есть идентификатор, вы можете создать ThreadPool с массивом, в котором хранится произвольная информация.Некоторые идеи:

• Нет массива:типичное использование пула потоков.Даже с использованием GIL это упрощает кодирование потоков и очень полезно для приложений с высокой задержкой, таких как сканирование веб-страниц большого объема.,
• Пул потоков и массив размером с количество процессоров:легко разветвляемые процессы для использования всех процессоров,
• Пул потоков и массив размером с количество ресурсов:например, каждый элемент массива представляет один процессор в пуле экземпляров, поэтому, если у вас есть 10 экземпляров, каждый с 4 процессорами, TP может управлять работой в 40 подпроцессорах.

С этими двумя последними вместо того, чтобы думать о потоках, выполняющих работу, подумайте о пуле потоков, управляющем подпроцессами, которые выполняют эту работу.Задача управления невелика, и в сочетании с подпроцессами, кому какое дело до GIL.

С помощью этого класса вы можете создать MapReduce на основе кластера примерно за сотню строк кода!Этот код очень короткий, хотя его полная проработка может быть немного затратной.Надеюсь, это поможет.

# Usage:
#
#   Thread.abort_on_exception = true # help localize errors while debugging
#   pool = ThreadPool.new(thread_pool_size)
#   50.times {|i|
#     pool.process { ... }
#     or
#     pool.process {|id| ... } # worker identifies itself as id
#   }
#   pool.shutdown()

class ThreadPool

  require 'thread'

  class ThreadPoolWorker

    attr_accessor :id

    def initialize(thread_queue, id)
      @id = id # worker id is exposed thru tp.process {|id| ... }
      @mutex = Mutex.new
      @cv = ConditionVariable.new
      @idle_queue = thread_queue
      @running = true
      @block = nil
      @thread = Thread.new {
        @mutex.synchronize {
          while @running
            @cv.wait(@mutex) # block until there is work to do
            if @block
              @mutex.unlock
              begin
                @block.call(@id)
              ensure
                @mutex.lock
              end
              @block = nil
            end
            @idle_queue << self
          end
        }
      }
    end

    def set_block(block)
      @mutex.synchronize {
        raise RuntimeError, "Thread is busy." if @block
        @block = block
        @cv.signal # notify thread in this class, there is work to be done
      }
    end

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

    def stop
      @mutex.synchronize {
        @running = false
        @cv.signal
      }
      @thread.join
    end

    def name
      @thread.inspect
    end
  end


  attr_accessor :max_size, :queue

  def initialize(max_size = 10)
    @process_mutex = Mutex.new
    @max_size = max_size
    @queue = Queue.new # of idle workers
    @workers = []      # array to hold workers

    # construct workers
    @max_size.times {|i| @workers << ThreadPoolWorker.new(@queue, i) }

    # queue up workers (workers in queue are idle and available to
    # work).  queue blocks if no workers are available.
    @max_size.times {|i| @queue << @workers[i] }

    sleep 1 # important to give threads a chance to initialize
  end

  def size
    @workers.size
  end

  def idle
    @queue.size
  end

  # are any threads idle

  def busy?
    # @queue.size < @workers.size
    @queue.size == 0 && @workers.size == @max_size
  end

  # block until all threads finish

  def shutdown
    @workers.each {|w| w.stop }
    @workers = []
  end

  alias :join :shutdown

  def process(block = nil, &blk)
    @process_mutex.synchronize {
      block = blk if block_given?
      worker = @queue.pop # assign to next worker; block until one is ready
      worker.set_block(block) # give code block to worker and tell it to start
    }
  end


end