Verteilte und Parallele Programmierung

Verteilte Programmierung und Systeme

Parallele und Verteilte Architekturen

Cluster Computing

• Ein Computer-Cluster besteht aus miteinander verbundenen eigenständigen Computern, die kooperativ als eine einzige integrierte Computer-Ressource arbeiten.

• In der Vergangenheit haben Computer-Cluster eindrucksvolle Ergebnisse bei der Bewältigung hoher Arbeitslasten mit großen Datenmengen gezeigt.

Cluster Computing

• Ein idealer Cluster mit mehrere Systembilder sollte zu einem Single-System-Image (SSI) zusammengeführt werden.

• Cluster-Entwickler wünschen ein Cluster-Betriebssystem oder eine Middleware, um SSI auf verschiedenen Ebenen zu unterstützen, einschließlich der gemeinsamen Nutzung von CPUs, Arbeitsspeicher und E/A über alle Cluster-Knoten hinweg.

• Ein SSI ist eine durch Software oder Hardware erzeugte Illusion, die eine Sammlung von Ressourcen als eine integrierte Ressource darstellt.

• SSI lässt den Cluster wie eine einzige Maschine für den Benutzer erscheinen. Ein Cluster mit mehreren Systemabbildern ist nichts anderes als eine Sammlung unabhängiger Computer &arr; One Big Virtual Machine!

Cloud Computing

• Cloud Computing bedeutet u.A. die Bereitstellung von virtualisierten Ressourcen von Rechenzentren zu einer Internet-Cloud, die mit Hardware, Software, Speicher, Netzwerk und Services für bezahlte Benutzer zur Verfügung gestellt wird, um ihre Anwendungen auszuführen.

Virtuelle Netzwerke

• Neben der Virtualisierung von Ressourcen wie Speicher und CPUs ist die Virtualisierung und Transformation der Netzwerkstrukturen zentrale Methodik

Partitionierung und Kommunikation

Gustafson Gesetz

• In parallelen Systemen drückt das Amdahlsche Gesetz den maximalen Speedup in Abhängigkeit vom sequenziellen Programmteil (bzw. dessen Ausführung) aus
  • Ein hoher Speedup und daher eine effiziente Ausnutzung (Workload) aller Verarbeitungseinheiten/Rechner ist nur durch Minimierung des sequenziellen Teils η (verursacht durch Kommunikation) möglich

Um bei Verwendung eines großen Clusters eine höhere Effizienz zu erzielen, muss die Größe des Problems so skalieren, dass sie der Cluster-Fähigkeit entspricht!

• Problem bei der Anwendung des Amdahlschen Gesetzes: Ein konstanter Workload wird für den sequenziellen und parallelen Anteil angenommen!

• Dies führt zu dem folgenden Beschleunigungsgesetz, das von John Gustafson (1988) vorgeschlagen wurde und als skalierte Arbeitslastbeschleunigung bezeichnet wird.

Gustafson Gesetz

• Sei W die Arbeitslast in einem bestimmten Programm. Bei Verwendung eines n-Prozessor-Systems skaliert der Benutzer die Arbeitslast zu W*= ηW + (1-η) nW.
  • Nur der parallelisierbare Teil der Auslastung wird im zweiten Ausdruck n-mal skaliert.

• Diese skalierte Arbeitslast W* ist im Wesentlichen die sequentielle Ausführungszeit auf einem einzelnen Prozessor.

• Die parallele Ausführungszeit einer skalierten Arbeitslast W* auf n Prozessoren wird wie folgt durch eine skalierte Arbeitslast-Beschleunigung definiert:

• Amdahls und Gustafsons Gesetze werden bei unterschiedlichen Workloads W angewendet!

Map & Reduce

• Ein Web-Programmiermodell für die skalierbare Datenverarbeitung in großen Clustern über große Datenmengen.

• Das Modell wird häufig in Web-Scale-Search- und Cloud-Computing-Anwendungen eingesetzt.

• Aber auch funktionale und parallele Programmierung macht von MapReduce Methoden Gebrauch

• Methode:

  • Es wird eine Map-Funktion angegeben, um eine Gruppe von Schlüssel/Wert-Zwischenpaaren zu generieren.
  • Dann wird eine Reduce-Funktion auf diesen Paaren angewendet, um alle Zwischenwerte mit demselben Zwischenschlüssel zusammenzuführen.

• MapReduce ist hochgradig skalierbar, um hohe Parallelitätsgrade auf verschiedenen Arbeitsebenen zu erreichen.

Map & Reduce

• Ein typischer MapReduce-Berechnungsprozess kann Terabytes an Daten auf Zehntausenden oder mehr Client-Computern verarbeiten. Hunderte von MapReduce-Programmen können gleichzeitig ausgeführt werden.
  • Tatsächlich werden jeden Tag Tausende von MapReduce-Jobs in Clustern von Google ausgeführt.
  • Das Hadoop Framework bietet für WEB Anwendungen MapReduce Services an → Master-Slave Architektur!

Die Map-Funktion verarbeitet ein Paar (Schlüssel, Wert) und gibt eine Liste von Zwischenpaaren (Schlüssel, Wert) zurück:

Die Reduzierungsfunktion führt alle Zwischenwerte zusammen, die die gleichen Zwischenschlüssel haben:

Map & Reduce

Map & Reduce

MapReduce Phasen

1. Ein Master-Prozess erhält einen Jobdeskriptor, der den auszuführenden MapReduce-Job angibt. Der Jobdeskriptor enthält neben anderen Informationen den Ort der Eingabedaten, auf die unter Verwendung eines verteilten Dateisystems zugegriffen werden kann.

2. Gemäß dem Jobdeskriptor startet der Master eine Anzahl von Mapper- und Reducer-Prozessen auf verschiedenen Maschinen. Gleichzeitig startet es einen Prozess, der die Eingabedaten von seinem Speicherort liest, diese Daten in eine Gruppe von Aufteilungen unterteilt und diese Aufteilungen an verschiedene Zuordner verteilt.

3. Nach dem Empfang seiner Datenpartition führt jeder Zuordnungsvorgang die Zuordnungsfunktion aus (die als Teil des Jobdeskriptors bereitgestellt wird), um eine Liste von Zwischenschlüssel / Wert-Paaren zu erzeugen. Dann werden diese Paare auf der Basis ihrer Schlüssel gruppiert.

Map & Reduce

4. Alle Paare mit den gleichen Schlüsseln sind dem gleichen Reduziervorgang zugeordnet. Daher führt jeder Reduzierprozess die Reduktionsfunktion (definiert durch den Jobdeskriptor) aus, die alle Werte vereinigt, die mit dem gleichen Schlüssel assoziiert sind, um einen möglicherweise kleineren Satz von Werten zu erzeugen.

5. Dann werden die von jedem Reduktionsprozess erzeugten Ergebnisse gesammelt und an einen durch den Jobdeskriptor spezifizierten Ort geliefert, um die endgültigen Ausgabedaten zu bilden.

Map & Reduce

Beispiel

local options = {workers = 2}
local data = {34,35,36,37,38,39,40,41}
  
local function worker (id,set)
  local results = T{}
  for i = 1,#set do
    results:push(fib(set[i]))
  end
  return results
end
  
-- Parallel Processing
local Parallel = require('parallel')
local p = Parallel:new(data,options)
function sum (x,y) return x+y  end
p:time():
  map(worker):
  apply(function (r) print(r:print()) end):
  reduce(sum):
  apply(print):
  time()

MPI

Message Passing Interface: MPI

Ziele und Eigenschaften

• Anwendungsprogrammierschnittstelle (nicht unbedingt für Compiler oder eine Systemimplementierungsbibliothek).

• Effiziente Kommunikation:

  • Vermeidung von Arbeitsspeicher-Arbeitsspeicher Kopien
  • Überlappung von Berechnung und Kommunikation
  • Auslagerung auf Kommunikations-Coprozessoren, soweit verfügbar.

• Implementierungen, die in einer heterogenen Umgebung verwendet werden können (verschiedene Hostplattformen).

• Einfache Einbindung in Programmiersprachen und Bibliotheken und Plattformunabhängigkeit

• Zuverlässige Kommunikation: Nutzer/Programmier muss sich nicht um Kommunikationsfehler kümmern

MPI

API

• Point-to-point communication,
• Datatypes,
• Collective operations,
• Process groups,
• Communication contexts,
• Process topologies,
• Environmental management and inquiry,
• The Info object,
• Process creation and management,
• One-sided communication,
• External interfaces,
• Parallel file I/O,

MPI

Operationen

MPI

Communicator

• Kommunikationswelt mit einer Gruppe aus Prozessen die gemeinsam Nachrichten austauschen können

• Nachrichten können nur innerhalb der Kommunikationswelt ausgetauscht werden

• MPI_COMM_WORLD ist die Standardwelt

MPI

Rank

• Einheitliche Prozessnummer innerhalb einer Kommunikationswelt

• Werden vom System bei der Initialisierung vergeben und werden fortlaufend ab 0 nummeriert

• Rank IDs werden bei der Kommunikation zur Identifikation von Empfänger und Sender verwendet (Kommunikationsendpunkte)

• Rank IDs dienen zur Programmdifferenzierung (if rank==0 then do this else do that)

MPI

Point-to-Point Kommunikation

• Kommunikation zwischen zwei Prozessen

• Quellprozess sendet eine Nachricht (Typ, Daten) an einen Zielprozess unter Angabe der Rank ID

• Kommunikation kann nur innerhalb eines Communicators stattfinden

MPI.send(dest:number,message:{type:string,content:string})

MPI

• Damit der Zielprozess die Nachricht empfangen kann muss er einen Handler für den entsprechenden Nachrichtentyp einrichten:

MPI.recv(type:string,callback:function (message))

MPI

Broadcast Kommunikation

• Ein Quellprozess kann eine Nachricht an alle Prozesse innerhalb eines Communicators senden

MPI.broadcast(message:{type:string,content:string})

RPC

Remote Procedure Call Interface

• Ähnlich MPI
• Klienten-Server Architektur
• Es gibt drei Operationen:
  • getreq → Server; Auf eine Anfrage warten
  • putrep → Server: Eine Annfrage beantworten
  • trans → Klient: Eine Anfrage senden und auf Antwort warten

RPC

LUA API (CSP)

• Die Kommmunikation findet über das IP-UDP/TCP Protokoll statt.
• Daten werden serialisiert und es können beliebige Daten übertragen werden (inkl. seitenfreier Funktionen)

Rpc(options?:{}) → rpc

Erzeugt eine RPC Instanz (für Klient und Server)

rpc:getreq(ip:string,port:number,callback:function (req) → rep)

Server Handlerfunktion (putrep wird implizit mit dem Rückgabewert des Callbackhandlers ausgeführt)

rpc:trans(ip_string,port:number,request:*) → reply:*

Klienten Transaktion

RPC

Beispiel

// Server
require('Csp')
local rpc = Rpc({verbose=2})
rpc:getreq('127.0.0.1',12345,function (req)
  if req.cmd=='iabs' then
    return {
     result=math.sqrt(math.pow(req.x,2)+
                      math.pow(req.y,2)),
     stat='OK'
    }
  else return {stat='EINVALID'} end
end)
loop.start()

// Client
require('Csp')
local rpc = Rpc({verbose=2})
local stat,reply = rpc:trans('127.0.0.1',12345,
                             {cmd='iabs',x=1,y=2})
print(reply)

Sicherheit und Lebendigkeit

• Ein Algorithmus, z.B. die Wahl eines Leaders in einer Prozessgruppe, gelte als sicher wenn maximal ein Leader unter allen Umständen gewählt wird. Dies ist auch der Fall wenn einzelne Prozesse der Prozessgruppe fehlerhaft sind (nicht erreichbar sind oder terminiert sind).

• Ein Algorithmus, z.B. die Wahl eines Leaders in einer Prozessgruppe, gelte als lebendig wenn irgendwann mindestens ein Leader unter allen Umständen gewählt wird.

Robustheit und Fehler

• Angenommen eine Prozessgruppe bestehe aus N Prozessen die in einem Netzwerk verteilt sind.

• Die Netzwerkknoten sind miteinander verbunden.

• Nun wird aufgrund einer technischen Störung das Netzwerk partitioniert (z.B. in zwei getrennte Bereiche geteilt).

Sicherheit und Lebendigkeit

• Der verteilte Algorithmus wird in jeder der Partitionen unabhängig arbeiten. Dann ist das verteilte System zwar noch lebendig (es werden unabhängig zwei Leader in den Gruppenpartitionen gewählt), aber nicht mehr sicher !!!!
  • Die Invariante des Algorithmus ist verletzt worden durch Ausfall/Störung!

Mutualer Ausschluss

Verteilter Algorithmus nach Lamport

LA1: Um einen kritischen Bereich zu erlangen (Mutex Acquire), sendet ein Prozess eine zeitgestempelte Anforderung an jeden anderen Prozess im System und fügt die Anforderung auch in seiner lokalen Queue Q hinzu.

LA2: Wenn ein Prozess eine Anforderung empfängt, wird sie in Q platziert. Wenn sich der Prozess nicht in seinem CS befindet, sendet er eine zeitgestempelte Bestätigung an den Absender. Andernfalls wird das Senden der Bestätigung bis zum Verlassen des CS verzögert (Mutex Release).

LA3: Ein Prozess tritt in seinen CS ein, wenn (1) seine Anfrage vor allen anderen Anfragen (d.h. der Zeitstempel seiner eigenen Anfrage ist kleiner als die Zeitstempel aller anderen Anfragen) in seinem lokalen Q angeordnet ist und (2) Es hat die Antworten von jedem anderen Prozess als Antwort auf seine aktuelle Anfrage erhalten.

LA4: Um die CS zu verlassen, löscht ein Prozess (1) die Anfrage von seiner lokalen Warteschlange Q und (2) sendet eine zeitgestempelte Freigabenachricht an alle anderen Prozesse.

Mutualer Ausschluss

LA5: Wenn ein Prozess eine Freigabenachricht erhält, entfernt er die entsprechende Anforderung aus seiner lokalen Warteschlange Q.

Verteilter Algorithmus nach Ricart–Agrawala’

• Verbesserte Version

RA1: Jeder Prozess, der den Eintritt in seinen CS anfordert (Mutex Acquire), sendet eine zeitgestempelte Anfrage an jeden anderen Prozess im System.

RA2: Ein Prozess, der eine Anforderung empfängt, sendet eine Bestätigung an den Absender zurück, nur wenn

• (1) der Prozess nicht an dem Eintritt in seinen CS interessiert ist (Mutex Acquire) oder
• (2) der Prozess versucht, seine CS zu erlangen, aber sein Zeitstempel größer ist als der des Absenders.
• Wenn sich der Prozess bereits in seinem CS befindet (besitzt den Lock) oder sein Zeitstempel kleiner als der des Absenders ist, puffert er alle Anforderungen bis zum Verlassen des CS.

Mutualer Ausschluss

RA3: Ein Prozess tritt in seine CS ein, wenn er von jedem der verbleibenden (n-1) Prozesse eine Bestätigung erhält.

RA4: Nach dem Verlassen seiner CS muss ein Prozess eine Rückmeldung an jede der ausstehenden Anfragen senden, bevor er eine neue Anfrage macht oder andere Aktionen ausführt.

Mutualer Ausschluss

Token Algorithmen

• Eine andere Klasse verteilter mutualer Ausschlussalgorithmen verwendet das Konzept eines expliziten variablen Tokens, das als eine Erlaubnis für den Eintritt in die CS dient (Mutex Acquire) und von einem anfordernden Prozess durch das Prozesssystem weitergereicht werden kann.

• Immer wenn ein Prozess in seinen CS eintreten möchte, muss er den Token erwerben. Der erste bekannte Algorithmus, der zu dieser Klasse gehört, ist auf Suzuki und Kasami zurückzuführen.

• Da es nur ein Token gibt ist der mutuale Ausschluss (Sicherheit) garantiert. Die Lebendigkeit aber nicht unbedingt (Verlust des Tokens).

Mutualer Ausschluss

Verteilter Konsens

• Ein verteilter Konsensalgorithmus hat das Ziel in einer Gruppe von Prozessen oder Agenten eine gemeinsame Entscheidung zu treffen

• Zentrale Eigenschaften:

  • Zustimmung/Übereinstimmung
  • Terminierung; Lebendigkeit und Deadlockfreiheit
  • Gültigkeit; Robustheit gegenüber Störungen wie fehlerhaften Nachrichten oder Ausfälle von Gruppenteilnehmern

• Beim Konsens kann ein Master-Slave Konzept oder ein Gruppenkonzept mit Leader/Commander und Workern verwendet werden.

  • Beim Master-Slave Konzept kommunizieren nur Slaves mit dem Master
  • Bei Gruppenkonzept (i.A. mit einem Leader) kommunizieren auch alle Gruppenteilnehmer untereinander

• Durch Störung (Fehler oder Absicht) kann es zu fehlerhaften bis hin zu fehlgeschlagenen Konsens kommen.

Verteilter Konsens

• Bedingungen für Interaktive Konsistenz:
  • IC1: Jeder Worker empfängt die gleiche Anweisung vom Leader!
  • IC2: Wenn der Leader fehlerfrei arbeitet, dann empfängt jeder fehlerfreie Worker die Anweisung die der Leader sendete!

Byzantinisches Generalproblem

• Beispiel: In einer Gruppe aus drei Prozessen/Agenten ist einer fehlerhaft bzw. versendet fehlerhafte Nachrichten (durch Störung oder Absicht) mit Anweisungen (schließlich ein Konsensergebnis) [E]

Verteilter Konsens

• Jeder Worker der Nachrichten empfängt ordnet diese nach direkten und indirekten (von Nachbarn)

• Fall (a): Prozess 2 versendet fehlerhafte Nachricht mit Anweisung 0, Prozess 1 empfängt eine direkte Nachricht mit Anweisung 1 und eine indirekte mit (falschen) Inhalt Anweisung 0

  • Bedingung IC1 ist erfüllt. Unter Annahme von Bedingung IC2 wird Worker 1 die direkte Anweisung 1 von Prozess 0 (Commander) auswählen → Konsens wurde gefunden

• Fall (b): Prozess 0 (Leader) versendet an Prozess 1 richtige Nachricht mit Anweisung 1 und falsche Nachricht mit Anweisung 0 an Prozess 1

  • Würde Prozess 1 wieder zur Erfüllung von IC2 eine Entscheidung treffen (Anweisung 1 auswählen), dann wäre IC1 verletzt. Wie auch immer Prozess 1 entscheidet ist entweder IC1 oder IC2 verletzt → Unentscheidbarkeit → Kein Konsens möglich

Verteilter Konsens

Das nicht-signierte Nachrichtenmodell erfüllt die Bedingungen:

1. Nachrichten werden während der Übertragung nicht verändert (aber keine harte Bedingung).
2. Nachrichten können verloren gehen, aber die Abwesenheit von Nachrichten kann erkannt werden.
3. Wenn eine Nachricht empfangen wird (oder ihre Abwesenheit erkannt wird), kennt der Empfänger die Identität des Absenders (oder des vermeintlichen Absenders bei Verlust).

• Algorithmen zur Lösung des Konsensproblems müssen m fehlerhafte Prozesse annehmen (bzw. fehlerhafte Nachrichten)

Der OM(m) Algorithmus

• Ein Algorithmus der einen Konsens erreicht bei Erfüllung der Bedingungen IC1 und IC2 mit bis zu m fehlerhaften Prozesse bei insgesamt n ≥ 3m+1 Prozessen mit nicht signierten (“mündlichen”) Nachrichten.
  1. Leader i sendet einen Wert v ∈ {0, 1} an jeden Worker j ≠ i.
  2. Jeder Worker j akzeptiert den Wert von i als Befehl vom Leader i.

Verteilter Konsens

Verteilter Konsens

Verteilter Konsens

Der Paxos Algorithmus

• Paxos ist ein Algorithmus zur Implementierung von fehlertoleranten Konsensfindungen.

• Er läuft auf einem vollständig verbundenen Netzwerk von n Prozessen und toleriert bis zu m Ausfälle, wobei n ≥ 2m+1 ist.

• Prozesse können abstürzen und Nachrichten können verloren gehen, byzantinische Ausfälle (absichtliche Verfälschung) sind jedoch zumindest in der aktuellen Version ausgeschlossen.

• Der Algorithmus löst das Konsensproblem bei Vorhandensein dieser Fehler auf einem asynchronen System von Prozessen.

• Obwohl die Konsensbedingungen Zustimmung, Gültigkeit und Terminierung sind, garantiert Paxos in erster Linie die Übereinstimmung und Gültigkeit und nicht die Beendigung - es ermöglicht die Möglichkeit der Beendigung nur dann, wenn es ein ausreichend langes Intervall gibt, in dem kein Prozess das Protokoll neu startet.

Verteilter Konsens

• Ein Prozess kann drei verschiedene Rollen wahrnehmen:
  • Antragsteller,
  • Akzeptor und
  • Lerner.

Verteilter Konsens

• Die Antragsteller reichen die vorgeschlagenen Werte im Namen eines Initiators ein,
• die Akzeptoren entscheiden über die Kandidatenwerte für die endgültige Entscheidung und
• die Lernenden sammeln diese Informationen von den Akzeptoren und melden die endgültige Entscheidung dem Initiator zurück.

• Ein Vorschlag, der von einem Antragsteller gesendet wird, ist ein Tupel (v, n), wobei v ein Wert und n eine Sequenznummer ist.

• Wenn es nur einen Akzeptor gibt, der entscheidet, welcher Wert als Konsenswert gewählt wird, dann wäre diese Situation zu einfach. Was passiert, wenn der Akzeptor abstürzt? Um damit umzugehen, gibt es mehrere Akzeptoren.

• Ein Vorschlag muss von mindestens einem Akzeptor bestätigt werden, bevor er für die endgültige Entscheidung in Frage kommt.

Verteilter Konsens

• Die Sequenznummer wird verwendet, um zwischen aufeinander folgenden Versuchen der Protokollanwendung zu unterscheiden.

• Nach Empfang eines Vorschlags mit einer größeren Sequenznummer von einem gegebenen Prozess, verwerfen Akzeptoren die Vorschläge mit niedrigeren Sequenznummern.

• Schließlich akzeptiert ein Akzeptor die Entscheidung der Mehrheit.

Phasen des Paxos Algorithmus

1. Die Vorbereitungsphase
   • Jeder Antragsteller sendet einen Vorschlag (v, n) an jeden Akzeptor
   • Wenn n die größte Sequenznummer eines von einem Akzeptor empfangenen Vorschlags ist, dann sendet er ein ack (n, ⊥, ⊥) an seinen Vorschlager
   • Hat der Akzeptor einen Vorschlag mit einer Sequenznummer n' < n und einem vorgeschlagenen Wert v akzeptiert, antwortet er mit ack(n, v, n').

Verteilter Konsens

2. Aufforderung zur Annahme eines Eingabewertes

   • Wenn ein Antragsteller ack(n, ⊥, ⊥) von einer Mehrheit von Akzeptoren empfängt, sendet er an alle Akzeptoren accept(v, n) und fordert sie auf, diesen Wert zu akzeptieren.
   • Wenn ein Akzeptor in Phase 1 einen ack(n, v, n') an den Antragsteller zurücksendet, muss der Antragsteller den Wert v mit der höchsten Sequenznummer in seiner Anfrage an die Akzeptoren einbeziehen.
   • Ein Akzeptor akzeptiert einen Vorschlag (v, n), sofern er nicht bereits zugesagt hat, Vorschläge mit einer Sequenznummer größer als n zu berücksichtigen.

3. Die endgültige Entscheidung

   • Wenn eine Mehrheit der Akzeptoren einen vorgeschlagenen Wert akzeptiert, wird dies der endgültige Entscheidungswert. Die Akzeptoren senden den akzeptierten Wert an die Lernenden, wodurch sie feststellen können, ob ein Vorschlag von einer Mehrheit von Akzeptoren akzeptiert wurde.