Implementieren einer Synchronisierungsbarriere in Ruby

Question

Ich versuche, das Verhalten der CUDA __synchtreads()-Funktion in Ruby zu "replizieren". Genauer gesagt, habe ich eine Reihe von N Threads, die Code ausführen müssen, dann warten alle aufeinander zum Zeitpunkt der Ausführung, bevor sie mit dem Rest ihres Geschäfts fortfahren. Zum Beispiel:

x = 0 

a = Thread.new do 
    x = 1 
    syncthreads() 
end 

b = Thread.new do 
    syncthreads() 
    # x should have been changed 
    raise if x == 0 
end 

[a,b].each { |t| t.join } 

Welche Tools muss ich verwenden, um dies zu erreichen? Ich habe versucht, einen globalen Hash zu verwenden und dann zu schlafen, bis alle Threads ein Flag gesetzt haben, das anzeigt, dass sie mit dem ersten Teil des Codes fertig sind. Ich konnte es nicht richtig funktionieren lassen; es führte zu Hängen und Stillstand. Ich denke, Ich muss eine Kombination von Mutex und ConditionVariable verwenden, aber ich bin unsicher, warum/wie.

Bearbeiten: 50 Aufrufe und keine Antwort! Sieht aus wie ein Kandidat für ein Kopfgeld ...

Answer 1

Lassen Sie uns eine Synchronisationsbarriere implementieren. Es muss wissen, wie viele Threads es verarbeiten wird, n, vorne. Während der ersten n - 1 Anrufe an sync die Barriere wird dazu führen, dass ein aufrufender Thread wartet. Die Rufnummer n weckt alle Threads auf.

class Barrier 
    def initialize(count) 
    @mutex = Mutex.new 
    @cond = ConditionVariable.new 
    @count = count 
    end 

    def sync 
    @mutex.synchronize do 
     @count -= 1 
     if @count > 0 
     @cond.wait @mutex 
     else 
     @cond.broadcast 
     end 
    end 
    end 
end 

ganzer Körper sync ist ein kritischer Abschnitt, das heißt es nicht von zwei Threads gleichzeitig ausgeführt werden kann. Daher der Anruf zu Mutex#synchronize.

Wenn der verringerte Wert von @count positiv ist, wird der Thread eingefroren. Die Übergabe des Mutex als Argument an den Aufruf an ConditionVariable#wait ist entscheidend, um Deadlocks zu verhindern. Es bewirkt, dass der Mutex entsperrt wird, bevor der Thread eingefroren wird.

Ein einfaches Experiment startet 1k Themen und macht sie Elemente zu einem Array hinzufügen. Zuerst fügen sie Nullen hinzu, dann synchronisieren sie und fügen solche hinzu. Das erwartete Ergebnis ist ein sortiertes Array mit 2k Elementen, von denen 1k Nullen und 1k Einsen sind.

mtx = Mutex.new 
arr = [] 
num = 1000 
barrier = Barrier.new num 
num.times.map do 
    Thread.start do 
    mtx.synchronize { arr << 0 } 
    barrier.sync 
    mtx.synchronize { arr << 1 } 
    end 
end .map &:join; 
# Prints true. See it break by deleting `barrier.sync`. 
puts [ 
    arr.sort == arr, 
    arr.count == 2 * num, 
    arr.count(&:zero?) == num, 
    arr.uniq == [0, 1], 
].all? 

als eine Angelegenheit der Tat, es gibt a gem named barrier das ist genau das tut, was ich oben beschrieben.

In einem letzten Hinweis, verwenden Sie keinen Schlaf für Wartezeiten unter solchen Umständen. Es heißt beschäftigt warten und is considered a bad practice.

Answer 2

Es könnte sein, dass die Threads darauf warten, dass sie aufeinander warten. Aber ich denke, dass es sauberer ist, wenn die Threads tatsächlich auf "Midpoint" enden, weil Ihre Frage offensichtlich besagt, dass die Threads die Ergebnisse der anderen am "Mittelpunkt" benötigen. Saubere Design-Lösung wäre es, sie fertig zu stellen, das Ergebnis ihrer Arbeit zu liefern und eine Reihe neuer Threads basierend darauf zu starten.