Verteilte und Parallele Programmierung

PD Stefan Bosse
Universität Bremen, FB Mathematik & Informatik
SS 2020
Version 2020-07-07

Parallelisierung und Metriken

Parallelität

  • Unterteilung in räumliche und zeitliche Parallelität

  • Parallele Datenverarbeitung bedeutet Partitionierung eines seriellen Programms in eine Vielzahl von Subprogrammen oder Tasks

  • Weitere Unterteilung beider Dimensionen in Abhängigkeit von:

    • Art der Tasks/Algorithmen
    • Ausführungsmodell der Tasks und verwendete Rechnerarchitektur
    • Art und Umfang der Wechselwirkung zwischen Tasks
    • Kontroll- und Datenfluss eines Tasks
Räumliche Parallelität
Die Datenmenge D kann in Teilmengen di ⊃ D zerlegt werden. Die minimal erreichbare Größe der Teilmenge gibt Granularität bei der Parallelisierung wieder. Die Datenmenge D wird durch eine Verarbeitungsstufe in eine neue Datenmenge D' transformiert, die dann von nachfolgenden Verarbeitungsstufen weiter verarbeitet wird. Beispiel : D=Bild ⇒ Glättung ⇒ D' ⇒ Objekterkennung ⇒ D‘’

Parallelität

Zeitliche Parallelität
Zeitliche Parallelität ist vorhanden, wenn eine Folge von gleichartigen Datenmengen D(n) repitierend mit dem gleichen Algorithmus verarbeitet werden → Pipeline-Verfahren

Datenabhängigkeit

  • Räumliche und zeitliche Parallelität führen zu räumlicher und zeitlicher Datenabhängigkeit.

  • Räumliche Datenabhängigkeit findet auf Intra- und Intertaskebene statt.

    • Intrataskebene: Subtasks tauschen Daten aus → Sequenzielle Ausführung. Beispiel: ST1: a=x+y; ST2: b=a+1; → b(a) → ST2(ST1)
    • Intertaskebene: Übertragung von Daten zwischen Tasks in einer Pipeline.

  • Zeitliche Datenabhängigkeit: Ergebnisse aus der Vergangenheit gehen in aktuelle Datenberechnung ein. Beispiel: Bewegungserkennung aus einer Bild-Sequenz.

Datenabhängigkeit

figraumzeitpar

Abb. 1. Räumliche und zeitliche Datenabhängigkeit ⇔ Vertikalen und horizontale Parallelisierung

Rechenzeit

  • Die Rechenzeit ttot für die Ausführung einer Pipeline T1…TM enthält:

    1. Zeit zum Einlesen der Daten I(n),
    2. Zeit für die Ausgabe der Daten und Ergebnisse,
    3. Summe aller Ausführungszeiten der Tasks in der Pipeline, die sich aus Rechen- und Kommuniaktionszeiten zusammensetzten.
\[\begin{gathered} {t_{tot}} = \sum\limits_{i = 1}^m {{\tau _i} + \sum\limits_{i = 1}^{m - 1} {{t_d}({D_{i,i + 1}}} ) + {t_{in}} + {t_{out}}} \hfill \\ {\tau _i} = \max \{ {t_{comp}}({T_{i,j}}({d_j}))|1 \leqslant j \leqslant {n_i}\} + {t_{comm}}({T_i}) \hfill \\ \end{gathered} \]

als die Zeit die benötigt wird, einen Task i unter Berücksichtigung von Datenabhängigkeiten der ni Subtasks T(dj) zu bearbeiten.

Rechenzeit

  • Längste Bearbeitungszeit eines Subtasks bestimmt Bearbeitungszeit des Tasks τi!
  • Ti ist der i-te Task in der horizontalen Pipeline,
  • dj die Teildatenmenge eines Subtasks Ti,i(dj) eines Tasks Ti,
  • tcomp ist die Rechenzeit,
  • tcomm die Intertaskkommunikationszeit,
  • tin und tout die Zeit zum Datentransfer in und aus der Pipeline, und
  • td(Di,i+1) die Datentransferzeit zwischen zwei Tasks.
  • In Vision-Systemen sind die ersten Tasks i.A. low- und mid-level Algorithmen, und die letzten Tasks high-level Algorithmen, die auf den Daten der unteren Ebenen aufbauen. Die einzelnen Datenströme Di können daher von unterschiedlicher Größe und Art sein.

Klassifikation von parallelen Algorithmen

Datenabhängigkeit

Lokal, statisch
Ausgangsdaten (≡Ergebnisse) hängen nur von eng begrenzter kurzreichweitiger Region der Eingangsdaten ab. Die Größe der Eingangsdatenregion ist unveränderlich (statisch). → Kommunikationsbedarf ist gering.
Lokal, dynamisch
Die Größe der Eingangsdaten-Region ist parametrisiert und veränderlich (dynamisch). Z.B. bei mathematischer Faltung von Bildmatrizen ändert sich Größe der Region.
Global, statisch
Die Ausgangsdaten hängen gänzlich vom gesamter Eingangsdatenmenge ab. Abhängigkeit hängt nur von der Größe des Bildes, aber nicht von dessen Inhalt ab. Z.B. Fouriertransformation oder Histogramm-Funktionen. → Kommunikationsbedarf ist groß.
Global, dynamisch
Ausgangsdaten hängen von variierenden Ausschnittsregion der Eingangsdaten ab. Die Berechnung ist vollständig datenabhängig. Z.B. Hough-Transformation.

Klassifikation von parallelen Algorithmen

figpardatab

Abb. 2. Klassifikation nach Kommunikationsbedarf aufgrund Datenabhängigkeit zwischen einzelnen Tasks eines parallelen Programms
  • Weiterhin Klassifikation nach:
    • Eingabedatenabhängigkeit
    • Ergebnisabhängigkeit

Speedup

Bei der Datenverarbeitung gibt es drei Randbedingungen:

  1. Gesamte Rechenzeit → Zeitdimension
  2. Gesamte Ressourcenbelegung → Flächendimension
  3. Bei der Parallelverarbeitung wird eine weitere Dimension hinzugefügt: Anzahl der Verarbeitungseinheiten sN
  • Die Nutzung von Parallelität führt zu einem Performanzgewinn (Speedup) durch Vergleich sequenzielles Programm (N=1) mit parallelen Programm (N Prozessoren):
\[Speedup = S(N) = \frac{Performanz(N)}{Performanz(1)}, S_t(n) = \frac{Rechnenzeit(1)}{Rechenzeit(N)} \]
  • Die sog. Skalierung bei der Parallelisierung ist i.A. nicht linear:
\[0 < S(N) < N \]
  • Kommunikation ist weitere Randbedingung bei der parallelen Datenverarbeitung.

Kosten und Laufzeit

Beispiel: Matrixmultiplikation

FUN matmult(A:ARRAY (p,q), B: ARRAY(p,q)) -> C:ARRAY (p,r)
DEF matmult(A, B) = 
  FOR i = 1 to p DO
    FOR j = 1 to r DO
      C[i,j] <- 0;
      FOR k = 1 to q DO
       C[i,j] <- C[i,j] + A[i,k] * B[k,j]
      END FOR k
    END FOR j
  END FOR i
END

  • Partitionierung kann beliebig erfolgen, da die einzelnen Ergebniswerte Ci,j nicht voneinander abhängen.

  • Datenabhängigkeit des Problems: Die Berechnung eines Ci,j Wertes hängt außerhalb der FOR-k-Schleife von keinem anderen Wert Cn,m mit n ≠ i und m ≠ j ab.

Kosten und Laufzeit

  • Mögliche Partitionierungen der drei For-Schleifen auf N parallel arbeitenden Verarbeitungseinheiten (VE):
  1. Jede VE berechnet einen Ci,j-Wert. D.h. eine VE führt die FOR-k-Schleife für ein gegebenes i und +j* durch.
    • Jede VE benötigt dazu die i-te Zeile von A und die j-te Spalte von B.
    • Keine weitere Datenabhängigkeit!
    • Es werden N=p•r VEs benötigt.
  1. Eine VE berechnet eine Ergebnisspalte, d.h. führt die FOR-k und FOR-j-Schleifen durch.
    • Jede VE benötigt A und eine Spalte von B.
    • Es werden N=p VEs benötigt.

Kosten und Laufzeit

  1. Eine VE führt eine Multiplikation und Addition innerhalb der FOR-K-Schleife durch.
    • Jede VE benötigt einen A und B-Wert.
    • Es werden N=p•r•q VEs benötigt.
    • Jeweils eine VE für einen gegebenen Ci,j-Wert führt die Zusammenführung der Zwischenwerte der FOR-k-VEs durch.
  • Zusammenführung der Ergebnisdaten in den Fällen 1&2 trivial. Im Fall 3 besteht Zusammenführung im wesentlichen in der Summation der Zwischenergebnisse.

Kosten und Laufzeit

figparpartmat

Abb. 3. Mögliche Partitionierungen der Matrixmultiplikation

Kosten und Laufzeit

figparpartcosts

Abb. 4. Kosten und Laufzeitanalyse der verschiedenen Parititioneirung

Kommunikation

  • Berechnung der Kommunikation mit Einheitswerten eines Nachrichtenaustauschs: Message Passing (MP) → Aufwand eine Zelle einer Matrix zu versenden ist MP=1.

figparcomm1

Abb. 5. Kommunikation Fall 1

Kommunikation

figparcomm2

Abb. 6. Kommunikation Fall 2

Kommunikation

figparcomm3

Abb. 7. Kommunikation Fall 3

Kommunikation

  • Bisher konnten Matrizen ganzzahlig auf VEs verteilt werden. In der Realität ist aber p=q=r,n!

  • Bisher wurde angenommen, dass alle VEs mit jeder anderen VE Daten mit einer Distanz/Ausdehnung D=1 austauschen kann. Nur möglich mit vollständig verbundenen Netzwerktopologien unter Verwendung von Kreuzschaltern.

  • Gängige parallele und ökonomische Rechnertopologie: Maschennetz

    • Besteht aus n × n VEs.
    • Jede VE kann mit seinem direkten Nachbarn kommunizieren.
    • Eine Nachricht hat eine maximale Reichweite von (2n-1) VEs, Θ(n).

Kommunikation

  • Bisherige statische Partitionierung resultiert in zu hohen Kommunikationsaufwand in der Verteilung der Matrizen A und B sowie in der Zusammenführung der Matrix C.
    • Dynamisch veränderliche Partitionierung passt sich effizienter an Netzwerktopologie an.
    • Bei Matrixoperationen mit zwei Matrizen (n=p=q=r) ist dafür ein 2n × 2n Netz optimal geeignet.

figmesh2

Abb. 8. Zweidimensionales Maschennetzwerk

Kommunikation

Dynamische Verteilung der Matrizen

  • Das 2n x 2n Netz wird in vier Quadranten der Größe n × n unterteilt.

    • Die Matrizen A und B befinden sich initial im linken unteren und rechten oberen Quadranten
    • Die Ergebnismatrix C wird schließlich im rechten unteren Quadranten zusammen- geführt.

  • Alle VEs die ein A1,j-Element besitzen werden dieses nach rechts verschieben.

  • Alle VEs die ein Bi,1-Element besitzen werden dieses nach unten verschieben.

    • Dieser Vorgang wird für weitere Zeilen (A) und Spalten (B) fortgesetzt.
    • Die einzelnen Elemente von A und B passieren die VEs im C-Quadranten. Die einzelnen C-Werte können mittels Summation berechnet werden.

Kommunikation

figparmatdyn

Abb. 9. Dynamische und überlagerte Verteilung der Matrizen A, B, und C

Kommunikation

  • Die Laufzeit der Matrixmuliplikation auf dem Netz beträgt Θ(n), da n Verschiebungen durchgeführt werden.

  • Die Kosten betragen Θ(n3), vergleichbar mit sequenzieller Ausführung. D.h. diese Partitionierung und Architektur ist kostenoptimal.

  • Das 2n x 2n Netz wird nur in drei Quadranten genutzt. Benötigt werden daher nur 3n2 VEs.

  • Aus Sicht der Rechnerarchitektur ungünstig zu implementieren und nicht generisch, d.h. abhängig vom verwendeten Algorithmus.

  • Reduktion dieses Verfahrens auf n × n Netz möglich

    • Dazu werden die Matrizen A, B und C überlagert, d.h. VE1,1 ⇒ {A1,1,B1,1,C1,1}.

  • Die Zeilenelemente von A werden dann nach vorherigen Ablaufschema gegen den Uhrzeigersinn rotiert (nur horizontal), und die Spaltenelemente von B im Uhrzeigersinn vertikal).

  • Man erhält gleiche asymptotische Grenzwerte für die Laufzeit Θ(n) und Kosten Θ(n3)!

Kommunikation

figparmatdyn2

Abb. 10. Überlagerte Verarbeitung der Matrizen A, B, und C auf einem Maschennetzwerk

Maßzahlen für Parallele Systeme

Berechnungszeit

Die Berechnungszeit Tcomp (computation time) eines Algorithmus als Zeit, die für Rechenoperationen verwendet wird.

Kommunikationszeit

Die Kommunikationszeit Tcomm (communication time) eines Algorithmus als Zeit, die für den Daten- bzw. Nachrichtenaustausch (Sende- und Empfangsoperationen) zwischen Subprogrammen und Verarbeitungseinheiten verwendet wird.

Untätigkeitszeit

Die Untätigkeitszeit Tidle (idle time) eines Systems als Zeit, die mit Warten (auf zu empfangende oder sendende Nachrichten) verbracht wird → Prozessblockierung trägt zur Untätigskeitszeit bei!

Maßzahlen für Parallele Systeme

Es gilt:

\[{T_{tot}} = {T_{comp}} + {T_{comm}} + {T_{idle}} \approx 1/N\sum\limits_{i = 1..n} {{T_{comp,i}} + {T_{comm,i}} + {T_{idle,i}}} \]

figpartime

Maßzahlen für Parallele Systeme

Übertragungszeit
Die Übertragunszeit Tmsg (message time) ist die Zeit, die für das Übertragen einer Nachricht mit der Länge L Datenwörter zwischen zwei Prozessen oder Prozessoren benötigt wird.
  • Sie setzt sich aus einer Startzeit Ts (message startup time) und einer Transferzeit Tw für ein Datenwort zusammen.
  • Es gilt: Tmsg = Ts + L • Tw
  • Voraussetzung: Verbindungsnetz arbeitet konfliktfrei.
Startzeit

Die Startzeit wird durch die Kommunikationshard- und software bestimmt, die zur Initiierung eines Datentransfers benötigt wird, z.B. Overhead des Protokollstacks bei einer Software-Implementierung.

Transferzeit

Die Transferzeit wird durch die Bandbreite des Kommunikationsmediums und zusätzlich bei Software-Implementierung durch den Protokollstack (Datenkopie) bestimmt.

Maßzahlen für Parallele Systeme

Parallelisierungsgrad P
Die maximalze Anzahl von binären Stellen (bits) pro Zeiteinheit (Taktzyklus) die von einer Datenverarbeitungsanlage verarbeitet werden kann.
  • Es gilt: P = W • B
Wortlänge W

Die Wortlänge oder Wortbreite gibt die Anzahl der Bits eines Datenpfades an.

Bitslice-Länge B

Die Bitslice-Länge setzt sich zusammen aus der Anzahl von Verarbeitungseinheiten VE, die parallel ausgeführt werden können, und der Anzahl der Pipeline-Stufen einer VE.

  • Es gilt: B = NVE • NStages

Maßzahlen für Parallele Systeme

figpargrad

Abb. 11. Illustration Parallelisierungsgrad und Beispiel

Maßzahlen für Parallele Systeme

Beschleunigung S und Kosten C
Die Beschleunigung gibt die Steigerung der Verabeitungsgeschwindigkeit bzw. die Re-duzierung der Verarbeitungszeit T an beim Übergang Anzahl Prozessoren N=1 → N>1. Die Kosten C skaliert die Verarbeitungszeit (Komplexitätsklasse) mit den Ressourcen.
\[S(N) = \dfrac{{T(1)}}{{T(N)}},C(N) = T(N)N \]
Effizienz E
Die Effizienz gibt die relative Verbesserung in der Verarbeitungsgeschwindigkeit an, da die Leistungssteigerung S mit der Anzahl der Prozessoren normiert wird.
\[E(N) = \dfrac{{S(N)}}{N},{\text{ mit }}\frac{1}{N} \leqslant E(N) \leqslant 1 \]

Maßzahlen für Parallele Systeme

Mehraufwand R
Bezieht sich auf die Anzahl X der auszuführenden (Einheits-)Operationen des Programms:
\[R(N) = \dfrac{{X(N)}}{{X(1)}} \]
Parallelindex I
Der Parallelindex gibt die Anzahl der parallelen Operationen pro Zeit-/Takteinheit an.
\[I(N) = \dfrac{{X(N)}}{{T(N)}} \]
Auslastung U
Entspricht dem normierten Parallelindex:
\[U(N) = \dfrac{{I(N)}}{{N}} \]

Maßzahlen für Parallele Systeme

Beispiel zur Auslastung

Ein Einprozessorsystem benötigt für die Ausführung von 1000 Operationen genau 1000 (Takt-)Schritte. Ein Multiprozessorsystem mit 4 Prozessoren benötigt dafür 1200 Operationen, die aber insgesamt in 400 Schritten ausgeführt werden können: 

X(1)=1000 und T(1)=1000, X(4)=1200 und T(4)=400 
S(4)=2.5 und E(4)=0.625,I(4)=3 und U(4)=0.75

Im Mittel sind 3 Prozessoren gleichzeitig aktiv, da jeder Prozessor nur zu 75% ausgelastet ist!

Amdahl’s Gesetz

Eine kleine Zahl von sequenziellen Operationen kann den Performanzgewinn durch Parallelisierung signifikant reduzieren.

  • Sequenzieller Anteil der Berechnungszeit T eines Algorithmus in [%]: η

  • Paralleler Anteil dann: (1-η)

  • Kommunikation (Synchronisation) zwischen nebenläufig ausgeführten Tasks oder Verarbeitungseinheiten verursacht immer η>0!

    • Beispiel: Schutz einer geteilten Ressource mit einer Mutex.
    • Beispiel: Datenverteilung über eine Queue

  • Zugriff auf geteilte Ressourcen verursacht immer η>0!

    • Beispiel: Geteilte Ressource Hauptspeicher in einer PRAM

  • Der Kommunikationsanteil ist schwer im Voraus abzuschätzen, und der genaue Wert hängt auch von temporalen Konkurrenzverhalten ab (wie häufig gibt es verlorene Wettbewerbe)!

Amdahl’s Gesetz

  • Es gilt dann für die gesamte Berechnungszeit eines parallelen Systems:
\[T(N,\eta ) = \eta T(1) + \dfrac{{(1 - \eta )T(1)}}{N} \]
  • Daraus lässt sich eine Obergrenze der Beschleunigung S mit zusätzlicher Abhängigkeit von η ableiten:
\[S(N,\eta ) = \dfrac{{T(1)}}{{T(N,\eta )}} \leqslant \dfrac{N}{{(\eta N - 1) + 1}} \]

Amdahl’s Gesetz

Beispiel

Ein Algorithmus mit einem sequenziellen Anteil von η=10% und einem parallelen Teil von 90% wird A.) mit N=10 Prozessoren, und B.) mit N=20 Prozessoren ausgeführt. Die Obergrenze der Beschleunigung S ist dann: 

A.) S(10)=5.26 
B.) S(20)=6.0
  • Verdopplung der Prozessoren erhöht nicht mehr signifikant die Beschleunigung!

Amdahl’s Gesetz

figamdahl

Abb. 12. Beschleunigung S in Abhängigkeit von η und Anzahl der Prozessoren N