Как исправить тупик в join() в Ruby

Я работаю в многопоточности в Ruby. Фрагмент кода:

  threads_array = Array.new(num_of_threads)  
  1.upto(num_of_threads) do |i|  

    Thread.abort_on_exception = true
      threads_array[i-1] =  Thread.new {
        catch(:exit) do
          print "s #{i}"
          user_id = nil
          loop do
            user_id = user_ids.pop()
            if user_id == nil
              print "a #{i}"
              Thread.stop()
            end
            dosomething(user_id)
          end
        end
      }
    end
    #puts "after thread"
    threads_array.each {|thread| thread.join}

Я не использую мьютексные блокировки, но захожу в тупик. Это вывод приведенного выше фрагмента кода:

s 2s 6s 8s 1s 11s 7s 10s 14s 16s 21s 24s 5s 26s 3s 19s 20s 23s 4s 28s 9s 12s 18s 22s 29s 30s 27s 13s 17s 15s 25a 4a 10a 3a 6a 21a 24a 16a 9a 18a 5a 28a 20a 2a 22a 11a 29a 8a 14a 23a 26a 1a 19a 7a 12fatal: deadlock detected

Приведенный выше вывод говорит мне, что взаимоблокировка возникает после того, user_ids массив user_ids имеет значение null и происходит с join и stop user_ids.

Что на самом деле происходит и что является решением этой ошибки?

Ответ 1

Простейший код для воспроизведения этой проблемы:

t = Thread.new { Thread.stop }
t.join # => exception in 'join': deadlock detected (fatal)

Thread :: stop → nil

Останавливает выполнение текущего потока, переводит его в состояние "сна" и планирует выполнение другого потока.

Тема # join → thr
Тема # присоединение (ограничение) → thr

Вызывающий поток приостановит выполнение и запустит thr. Не возвращается до тех пор, пока не выйдет thr или пока не пройдут лимитные секунды Если срок истекает, возвращается ноль, в противном случае возвращается thr.

Насколько я понимаю, вы вызываете Thread.join без параметров в потоке и ждете его завершения, но дочерний поток вызывает Thread.stop и переходит в состояние sleep. Это тупиковая ситуация, основной поток ожидает выхода дочернего потока, но дочерний поток спит и не отвечает.

Если вы вызываете join с параметром limit тогда дочерний поток будет прерван после истечения времени ожидания, не вызывая тупиковую ситуацию для вашей программы:

t = Thread.new { Thread.stop }
t.join 1 # => Process finished with exit code 0

Я бы порекомендовал выйти из ваших рабочих потоков после того, как они выполнят работу с Thread.exit или избавиться от бесконечного цикла и нормально дойти до конца потока выполнения, например:

if user_id == nil
  raise StopIteration
end

#or 
if user_id == nil
  Thread.exit
end

Ответ 2

В дополнение к ответу Alex Kliuchnikau, я добавлю, что #join может поднять эту ошибку, когда поток ждет Queue#pop. Простым и сознательным решением является вызов #join с таймаутом.

Это от рубина 2.2.2:

[27] pry(main)> q=Queue.new
=> #<Thread::Queue:0x00000003a39848>
[30] pry(main)> q << "asdggg"
=> #<Thread::Queue:0x00000003a39848>
[31] pry(main)> q << "as"
=> #<Thread::Queue:0x00000003a39848>
[32] pry(main)> t = Thread.new {
[32] pry(main)*   while s = q.pop
[32] pry(main)*     puts s
[32] pry(main)*   end  
[32] pry(main)* }  
asdggg
as
=> #<Thread:[email protected](pry):34 sleep>
[33] pry(main)> q << "asg"
asg
=> #<Thread::Queue:0x00000003a39848>
[34] pry(main)> q << "ashg"
ashg
=> #<Thread::Queue:0x00000003a39848>
[35] pry(main)> t.join
fatal: No live threads left. Deadlock?
from (pry):41:in `join'
[36] pry(main)> t.join(5)
=> nil

Ответ 3

Если я получу ваши намерения правильно, я бы рассмотрел что-то более простое (и, вероятно, безопаснее, users_ids.pop() изнутри потока выглядит страшно для меня):

user_ids = (0..19).to_a
number_of_threads = 3

user_ids \
  .each_slice(user_ids.length / number_of_threads + 1) \
  .map { |slice| 
      Thread.new(slice) { |s| 
        puts s.inspect 
      }
  }.map(&:join)