Как управлять рубиновыми потоками, чтобы они завершили всю свою работу?

У меня есть вычисление, которое можно разделить на независимые единицы, и теперь я имею дело с ним, создавая фиксированное количество потоков, а затем отдавая куски работы, которые должны выполняться в каждом потоке. Итак, в псевдокоде здесь выглядит как

# main thread
work_units.take(10).each {|work_unit| spawn_thread_for work_unit}

def spawn_thread_for(work)
  Thread.new do
    do_some work
    more_work = work_units.pop
    spawn_thread_for more_work unless more_work.nil?
  end
end

В принципе, когда создается начальное число потоков, каждый из них выполняет некоторую работу, а затем продолжает делать вещи из рабочего стека, пока ничего не останется. Все работает нормально, когда я запускаю вещи в irb, но когда я выполняю script с помощью интерпретатора, все не так хорошо работает. Я не уверен, как сделать основной поток до тех пор, пока все работы не будут завершены. Есть ли хороший способ сделать это, или я застрял в выполнении sleep 10 until work_units.empty? в основном потоке

Ответ 1

Если вы измените spawn_thread_for, чтобы сохранить ссылку на созданный Thread, вы можете вызвать Thread#join в потоке, чтобы дождаться завершения:

x = Thread.new { sleep 0.1; print "x"; print "y"; print "z" }
a = Thread.new { print "a"; print "b"; sleep 0.2; print "c" }
x.join # Let the threads finish before
a.join # main thread exits...

дает:

abxyzc

(Украден из документации ri Thread.new. Подробнее см. документацию ri Thread.join.)

Итак, если вы измените spawn_thread_for, чтобы сохранить ссылки на Thread, вы можете присоединиться к ним всем:

(Непроверенный, но должен придать аромат)

# main thread
work_units = Queue.new # and fill the queue...

threads = []
10.downto(1) do
  threads << Thread.new do
    loop do
      w = work_units.pop
      Thread::exit() if w.nil?
      do_some_work(w)
    end
  end
end

# main thread continues while work threads devour work

threads.each(&:join)

Ответ 2

В ruby 1.9 (и 2.0) вы можете использовать ThreadsWait из stdlib для этой цели:

require 'thread'
require 'thwait'

threads = []
threads << Thread.new { }
threads << Thread.new { }
ThreadsWait.all_waits(*threads)

Ответ 3

Кажется, вы копируете то, что предоставляет библиотека Parallel Each (Peach).

Ответ 4

Thread.list.each{ |t| t.join unless t == Thread.current }