Ruby - 반복 할 때 멀티 스레딩 (생성자/소비자 모델)을 사용하면 예기치 않은 결과가 발생합니다.

Question

참고 : DNS 이름을 확인하기 위해 스레딩을 사용하고 있지만 비슷한 동작으로 같은 동작을 재현 할 수 있습니다.

---

(이전에 작동하는) 코드를 표준 단일 스레드 실행에서 다중 스레드로 이동하려고하면 예기치 않은 결과가 나타납니다. 특히, 내 코드는 해시 배열을 반복하고 배열의 각 해시에 키/값 쌍을 추가합니다.

새로운 키/값 쌍이 생성되는 루프의 dns_cname.map 루프에서 문제가 발생하는 것으로 보입니다. 올바른 값을 가진 "external_dns_entry" 키 (즉, DNS로 확인 된 이름이 포함 된 result.name.to_s) 대신 url_nameserver_mapping에있는 많은 다른 서버 중 하나의 이름이 대신 나타납니다.

스레드가 사용 가능 해지고 해시가 업데이트되지 않아서 DNS 확인이 진행되고 있지만이 문제를 추적하는 방법을 알지 못한다고 생각합니다.

문제가있는 결과 : server1에 대한 DNS 확인은 서버 17에 매핑됩니다. 마찬가지로 서버 17은 서버 99에 매핑됩니다. 나머지 해시는 그대로 유지됩니다.

도움이 매우 감사합니다. 대단히 감사드립니다!

여기에 멀티 스레딩은을 사용할 수 없습니다 때 내 코드 이다 (제대로 작동) :

url_nameserver_mapping = { "server1" => "dallasdns.dns.com", 
          "server2" => "portlanddns.dns.com", 
          "server3" => "losangelesdns.dns.com" } 


# Parse the JSON string response from the API into a valid Ruby Hash 
# The net/http GET request is not shown here for brevity but it was stored in 'response' 
unsorted_urls = JSON.parse(response.body) 

# Sort (not sure this is relevant) 
# I left it since my data is being populated to the Hash incorrectly (w/ threading enabled) 
url_properties = unsorted_urls['hostnames']['items'].sort_by { |k| k["server"]} 

url_nameserver_mapping.each do |server,location| 

     dns = Resolv::DNS.new(:nameserver => ['8.8.8.8']) 
     dns_cname = dns.getresources(server, Resolv::DNS::Resource::IN::CNAME) 

     dns_cname.map do |result| 
     # Create a new key/value for each Hash in url_properties Array 
     # Occurs if the server compared matches the value of url['server'] key 
     url_properties.each do |url| 
      url["external_dns_entry"] = result.name.to_s if url['server'] == server 
     end 
     end 

end 

내가 생산자/소비자 스레딩 모델을 구현하기 위해 https://blog.engineyard.cm/2013/ruby-concurrency에서 가이드를 따라 갔다. 여기

는 멀티 스레딩이 (작동하지 않음)를 사용할 수 있습니다 내 적응 코드 입니다 :

require 'thread' 
require 'monitor' 

thread_count = 8 
threads = Array.new(thread_count) 
producer_queue = SizedQueue.new(thread_count) 
threads.extend(MonitorMixin) 
threads_available = threads.new_cond 
sysexit = false 

url_nameserver_mapping = { "server1" => "dallasdns.dns.com", 
          "server2" => "portlanddns.dns.com", 
          "server3" => "losangelesdns.dns.com" } 


unsorted_urls = JSON.parse(response.body) 

url_properties = unsorted_urls['hostnames']['items'].sort_by { |k| k["server"]} 

#################### 
##### Consumer ##### 
#################### 

consumer_thread = Thread.new do 

    loop do 

    break if sysexit && producer_queue.length == 0 
    found_index = nil 

    threads.synchronize do 
     threads_available.wait_while do 
     threads.select { |thread| thread.nil? || 
            thread.status == false || 
            thread["finished"].nil? == false}.length == 0 
     end 
     # Get the index of the available thread 
     found_index = threads.rindex { |thread| thread.nil? || 
               thread.status == false || 
               thread["finished"].nil? == false } 
    end 

    @domain = producer_queue.pop 

     threads[found_index] = Thread.new(@domain) do 

     dns = Resolv::DNS.new(:nameserver => ['8.8.8.8']) 
     dns_cname = dns.getresources(@domain, Resolv::DNS::Resource::IN::CNAME) 

     dns_cname.map do |result| 
      url_properties.each do |url| 
      url["external_dns_entry"] = result.name.to_s if url['server'] == @domain 
      end 
     end 

     Thread.current["finished"] = true 

     # Notify the consumer that another batch of work has been completed 
     threads.synchronize { threads_available.signal } 
     end 
    end 
end 

#################### 
##### Producer ##### 
#################### 

producer_thread = Thread.new do 

    url_nameserver_mapping.each do |server,location| 

    producer_queue << server 

    threads.synchronize do 
     threads_available.signal 
    end 
    end 
    sysexit = true 
end 

# Join on both the producer and consumer threads so the main thread doesn't exit 
producer_thread.join 
consumer_thread.join 

# Join on the child processes to allow them to finish 
threads.each do |thread| 
    thread.join unless thread.nil? 
end 

Answer 1

@domain 모든 스레드에 의해 공유 -이 공유가 문제의 루트입니다 :이 업데이트 될 때 대기열에서 다음 작업 단위를 표시하면 모든 스레드가 그 변경 사항을 볼 수 있습니다.

Thread.new(producer_queue.pop) do |domain| 
    #domain isn't shared with anyone (as long as there 
    #is no local variable called domain in the enclosing scope 
end 

질문에 접하는 것이지만, 이는 실제로 과장된 방식을 선호하는 것처럼 보입니다. 미리 많은 소비자 스레드를 스핀 업 (spin up)하고 작업 대기열에서 직접 읽도록하는 것이 훨씬 쉽습니다.