Die Programmiersprache Ruby (German Edition)

einer
SizedQueue
können Sie die maximale Größe mit
max
und
max=
abfragen und verändern.
    Weiter oben in diesem Kapitel haben wir gesehen, wie wir dem Modul
Enumerable
eine Concurrent-Map-Methode hinzufügen können. Wir definieren nun eine Methode, die ein paralleles
map
mit einem parallelen
inject
kombiniert. Es erzeugt einen Thread für jedes Element der enumerierbaren Collection und verwendet diesen Thread, um eine abbildende
Proc
anzuwenden. Der von
Proc
zurückgegebene Wert wird einem
Queue
-Objekt hinzugefügt. Ein letzter Thread agiert als Consumer – er entfernt Werte aus der Queue und übergibt sie dem »einfügenden«
Proc
, wenn sie bereit sind.
    Wir bezeichnen diese parallele Injection-Methode als
conject
, und Sie können sie zum Beispiel dafür verwenden, die Summe der Quadrate in einem Array zu berechnen. Beachten Sie, dass ein sequenzieller Algorithmus für so ein einfaches Quadratsummenbeispiel sicherlich schneller wäre:
a = [-2,-1,0,1,2]
mapper = lambda {|x| x*x } # Quadrate berechnen
injector = lambda {|total,x| total+x } # Summe berechnen
a.conject(0, mapper, injector) # => 10
    Der Code für diese Methode
conject
sieht wie folgt aus – beachten Sie die Verwendung eines
Queue
-Objekts und seiner Methoden
enq
und
deq
:
module Enumerable
# Concurrent Inject: erwartet einen initialen Wert und zwei Procs
def conject(initial, mapper, injector)
    # Verwende eine Queue für die Weitergabe von Werten von den Abbildungs-Threads
    # an den Injector-Thread
    q = Queue.new
    count = 0 # Wie viele Elemente?
    each do |item| # Für jedes Element
     Thread.new do # Erzeuge einen neuen Thread
     q.enq(mapper[item]) # Abbilden und abgebildeten Wert in die Queue stecken
     end
     count += 1 # Elemente zählen
    end

    t = Thread.new do # Injector-Thread erzeugen
     x = initial # Beginn mit angegebenem Anfangswert
     while(count > 0) # Einmal über jedes Element laufen
     x = injector[x,q.deq] # Wert auslesen und in inject stecken
     count -= 1 # Herunterzählen
     end
     x # Thread-Wert ist injizierter Wert
    end
    t.value # Auf Injector-Thread warten und Wert zurückgeben
end
end
    9.9.9 Bedingungsvariablen und Queues
    Ein wichtiger Punkt an der Klasse
Queue
ist, dass die Methode
deq
blockieren kann. Normalerweise denken wir beim Blockieren nur an
IO
-Methoden (oder wenn man
join
für einen Thread oder
lock
für ein
Mutex
aufruft). Bei der Multithread-Programmierung kann es allerdings manchmal notwendig sein, dass ein Thread auf die Erfüllung einer bestimmten Bedingung wartet (die nicht der Kontrolle dieses Threads unterliegt). Bei der Klasse
Queue
ist die Bedingung der nicht leere Status der Queue: Wenn die Queue leer ist, muss ein Consumer-Thread warten, bis ein Producer-Thread
enq
aufruft und die Queue nicht mehr leer ist.
    Einen Thread darauf warten zu lassen, dass ein anderer Thread ihm mitteilt, dass er fortfahren kann, wird am besten über eine
ConditionVariable
erreicht. Wie
Queue
ist
ConditionVariable
Teil der Standardbibliothek
thread
. Erzeugen Sie eine
ConditionVariable
mit
ConditionVariable.new
. Lassen Sie einen Thread auf die Bedingungen mit der Methode
wait
warten. Wecken Sie einen wartenden Thread mit
signal
auf. Wecken Sie alle wartenden Threads mit
broadcast
auf. Bei der Verwendung von Bedingungsvariablen gibt es nur einen Punkt, den man beachten muss: Damit alles funktioniert, muss der wartende Thread ein gesperrtes
Mutex
-Objekt an die Methode
wait
übergeben. Dieses
Mutex
wird dann vorübergehend entsperrt, während der Thread wartet, und wieder gesperrt, wenn der Thread aufwacht.
    Wir schließen unseren Rundgang durch die Threads mit einer Hilfsklasse ab, die bei der Multithread-Programmierung nützlich sein kann. Sie hat den Namen
Exchanger
und ermöglicht es zwei Threads, bestimmte Werte zu tauschen. Stellen Sie sich vor, dass wir die Threads
t1
und
t2
und ein
Exchanger
-Objekt
e
haben.
t1
ruft
e.exchange(1)
. Diese Methode blockiert dann (natürlich mit einer
ConditionVariable
), bis
t2 e.exchange(2)
aufruft. Dieser zweite Thread blockiert nicht, sondern gibt einfach
1
zurück – den von
t1
übergebenen Wert. Nachdem nun dieser zweite Thread
exchange
aufgerufen hat, wacht
t1
wieder auf und liefert von der
exchange
-Methode den Wert
2
zurück.
    Die hier gezeigte Implementierung von
Exchanger
ist recht komplex, zeigt aber eine typische Anwendung der Klasse
ConditionVariable
. Interessant an diesem Code ist, dass er zwei
Mutex
-Objekte nutzt. Eines davon wird zum Synchronisieren des Zugriffs auf die