Ruby - 使用互斥锁防止线程过早停止答案

【问题标题】：Ruby - Using a Mutex to keep threads from stopping prematurelyRuby - 使用互斥锁防止线程过早停止
【发布时间】：2014-10-20 13:39:48
【问题描述】：

我正在编写一个 Ruby 应用程序（Linux x86_64 中的 Ruby v2.1.3p242），它将重复处理在线数据并将结果存储在数据库中。为了加快速度，我有多个线程同时运行，并且我一直在研究一种方法来干净地停止所有线程，无论是在命令上还是从线程引发异常时。

问题是在调用Sever.stop 之后，某些线程将继续运行#do_stuff 的多次迭代。它们最终确实会停止，但我会看到几个线程在其余线程停止后运行 10-50 次。

每个线程的互斥体在每次迭代之前都被锁定，之后被解锁。当调用Server.stop 时，代码@mutex.synchronize { kill } 在每个线程上运行。这应该在下一次迭代后立即终止线程，但情况似乎并非如此。

编辑：

代码按原样运行，如果您愿意，请随时对其进行测试。在我的测试中，调用Server.stop 后，所有线程都需要 30 秒到几分钟才能停止。请注意，每次迭代需要 1-3 秒。我使用以下代码来测试代码（在同一目录中使用ruby -I.）：

require 'benchmark'
require 'server'

s = Server.new
s.start
puts Benchmark.measure { s.stop }

代码如下：

server.rb:

require 'server/fetcher_thread'

class Server
  THREADS = 8  

  attr_reader :threads
  def initialize
    @threads = []
  end

  def start
    create_threads
  end

  def stop
    @threads.map {|t| Thread.new { t.stop } }.each(&:join)
    @threads = []
  end

  private

  def create_threads
    THREADS.times do |i|
      @threads << FetcherThread.new(number: i + 1)
    end
  end
end

server/fetcher_thread.rb：

class Server
  class FetcherThread < Thread
    attr_reader :mutex

    def initialize(opts = {})
      @mutex = Mutex.new
      @number = opts[:number] || 0

      super do      
        loop do
          @mutex.synchronize { do_stuff } 
        end
      end
    end

    def stop
      @mutex.synchronize { kill }
    end

    private

    def do_stuff
      debug "Sleeping for #{time_to_sleep = rand * 2 + 1} seconds"
      sleep time_to_sleep
    end

    def debug(message)
      $stderr.print "Thread ##{@number}: #{message}\n"
    end
  end
end

【问题讨论】：

您能否将您的问题简化为可以在不依赖任何外部系统的情况下在纯 Ruby 中重现的问题？
我更改了代码，使其在每次迭代中运行do_stuff。这只是显示一条消息并休眠 1-3 秒。该代码有效并演示了该问题，因此请随时对其进行测试。
谢谢。那应该可以帮助您获得帮助。但是，如果所有这些代码对于从本质上重现您的问题确实是必要的，我会感到惊讶。例如，是否需要 ActiveSupport？
好点。我精简了代码并删除了对 ActiveSupport 和 Singleton 的依赖。问题仍然可以重现。

标签： ruby multithreading mutex

【解决方案1】：

无法保证调用stop 的线程将在循环的下一次迭代之前获取互斥锁。这完全取决于 Ruby 和操作系统调度程序，并且一些操作系统 (including Linux) 没有实现 FIFO 调度算法，但会考虑其他因素来尝试优化性能。

您可以通过避免kill 并使用变量干净地退出循环来使这更可预测。然后，您只需要在访问变量的代码周围包裹互斥锁

class Server
  class FetcherThread < Thread
    attr_reader :mutex

    def initialize(opts = {})
      @mutex = Mutex.new
      @number = opts[:number] || 0

      super do      
        until stopped?
          do_stuff
        end
      end
    end

    def stop
      mutex.synchronize { @stop = true }
    end

    def stopped?
      mutex.synchronize { @stop }
    end

    #...
  end
end

【讨论】：