Übung 5 - Parallel Lua / CSP Tutorial (Teil 1)

In dieser Übung findet eine erste Einführung in die Programmierung von parallelen Systemen nach dem CCSP Modell. Dabei steht CCSP für "Concurrent Communicating Sequential Processes" und beschreibt die Komposition und Kommunikation von sequenziellen Prozessen mit Konkurrenz. man unterscheidet:

Prozesskonstruktoren
Kommunikationsobjekte

Vorbereitung und Verwendung

Minimalanforderungen: lvm 1.1.18 luv 2.7.5!

Es wird die parallele LuaJit VM lvm mit integrierten Csp Modul für das jeweilige eigene Betriebssystem benötigt
- Momentan steht lvm für die Betriebssysteme Linux 32bit, Linux 64bit, Solaris 32bit, Windows 32bit und MacOS X 64bit zur Verfügung. Unter Microsoft Windows kann es zu funktionalen Abweichungen und Einschränkungen von lvm kommen → lvm, die aber für die Übung nicht relevant sind.
- Lua/Csp Programme können direkt von der Konsole ausgeführt werden:
```
# lvm myprog.lua
```
- Die Programme (oder Teile davon) können ebenso in diesem Notebook ausgeführt werden wenn lvm mit einem WEB Shell Wrapper gestartet wird (lokaler Rechner, Konsole):
```
# lvm weblvm.lua
Service thread 2 started.
[2@1592562880] HTTP server (2) listening to http://0.0.0.0:4610
Monitor thread 4 started.
[4@1592562880] HTTP server (4) listening to http://0.0.0.0:4611
```
- Diese Variante erlaubt das schrittweise Erlernen von Lua/Csp.
- Alle Code Snippets in diesem Notebook haben dann den gleichen Ausführungskontext (d.h. globale Variablen und Funktionen können von den einzelnen Snippets geteilt werden!)
- Dieser geteilte Kontext kann aber auch zu Problemen führen und bei Fehlern in parallelen Prozessen können Geisterprozesse übrig bleiben die nicht beendet werden können
- Ein Neustart des lvm Web Wrappers kann daher hin und wieder notwendig werden!. Das Notebook muss dann nicht neu geladen werden und funktioniert sofort nach dem Neustart von lvm wieder. Nur der globale VM Kontext ist wieder "leer"!
- Die Konsolenausgabe (mittels der print Anweisung) bei der Ausführung eines Programabschnitts wird automatisch weitergeleitet und hier angezeigt. Es kann aber nach einer gewissen Zeit erforderlich werden diese automatische Ausgabeweiterleitung wieder durch Drücken des Refresh Buttons anzustoßen (rechts an der jeweiligen Ausgabekonsole hier im Notebook)
- Jedes Codesnippet hat seine eigene Ausgabekonsole!

Der Programmcode kann in dem oberen Teilfenster eines Snippets verändert werden.
Der Programmcode wird ausgeführt durch Drücken der Playtaste ▸ im unteren Teilfenster (Ausgabekonsole).
Die Ausgabekonsole kann auf der rechten Seite gelöscht werden durch Drücken von ✗.
Hin und wieder kann eine Aktualisierung des Ausgabefensters durch Drücken des Knopfes ↻ erforderlich sein.
Bei parallelen Prozessen werden jeweils neue VM Instanzen in Threads gestartet. Wenn etwas schief läuft kann das Programm hängen und nicht terminieren (also der aktuelle Codeabschnitt). Der Abbruch kann versucht werden durch Drücken des Knopfes ◼.

Eine einfache Textausgabe (1)

print(math.random())

▸

◼

✗

↻

≡

Prozesse

Sequenzielle Prozesse

Sequenzielle Prozesse werden in dem gleichen Thread und der gleichen Koroutine ausgeführt.
Wichtig: Die Seq-Teilprozesse besitzen einen eigenen eingeschränkten Kontext; Bindungen von freien Variablen sind nicht möglich (also außerhalb der Funktion definierte), nur über den geteilten Kontext!

Sequenzielle Prozesse sind in Lua (und jeder anderen imperativen/prozeduralen Programmiersprache) ein Artefakt (jedes Programm und jede Funktion ist bereits ein seq. Prozess)

Ein Scheduling ist nicht möglich, aber Blockierung!

Eine einfache Textausgabe (2)

require 'Csp';
Seq({
  function () 
    print('Hello '..word)
  end,
  function ()
    print('Thank '..word)
  end
},{
  word='shared'
})
print('After Seq..')

▸

◼

✗

↻

≡

Aufgabe 1. Füge in die beiden obigen seq. Prozesse die sleep(millisec) Funktion vor der Konsolenausgabe print ein (wähle für die Verzögerung Werte im Bereich 100-500 ms)

Frage 2. Wie verhält sich das Programm? Füge die Ausgabe der Prozesszeit ein (print(time())) vor und nach dem sleep Aufruf. Alternativ kann anstelle der print die log Anweisung verwendet werden. Diese gibt die aktuelle Prozesszeit in Millisekunden zusammen mit den Argumenten aus.

Lösung.

Die Prozesse werden strikt sequenziell ausgeführt. Die Rechenzeit ist akkumulativ.

Frage 3. Was passiert wenn in einem Prozess ein Fehler auftritt? Also füge z.B. den Aufruf einer nicht definierten Funktion foo(1) ein...

Lösung.

Der gesamte Prozessfluss wird unterbochen.
Es wird ein Fehlersignal "EPROC" erzeugt.

Fehlerbehandlung

Die Fehlerbehandlung (genau genommen die Behandlung von Ausnahmefehlern und Signalen) erfolgt in einem sequenziellen Programm mit der funktionalen Try().catch().Finally() Anweisung (deterministisch)- lvm unterstützt eine nicht Lua-konforme try .. catch(e) .. end Kurzschreibweise.
Die Fehlerbehandlung in parallelen Systemen ist nicht ohne weiteres möglich!
Alle Prozesskonstruktoren werden dann mit einem Fehlersignal "EPROC" abgebrochen (außer es handelt sich um ein Signal):

Benutze nur noch die Großschreibung Try().catch().Finally() um mit der generischen und Fengari Lua VM kompatibel zu sein.

Fehlerbehandlung

require 'Csp';
Try(function () 
  Seq({
    function () 
      print('Hello '..word)
    end,
    function ()
      foo(1) -- nicht definiert, Fehler
      print('Thank '..word)
    end
  },{
    word='shared'
  })
end).catch (function (e)
  print "Caught exception:"
  print(e)
end)

▸

◼

✗

↻

≡

Häufig ist bei der Fehlersuche der genaue Ort und eine Funktionsaufrufverlauf (trace) erforderlich. Diese Ausgabe kann durch die Option debug=true erreicht werden:

Fehlerbehandlung

require 'Csp';
Try(function ()
  Seq({
    function () 
      print('Hello '..word)
    end,
    function ()
      foo(1)
      print('Thank '..word)
    end
  },{
    word='shared',
    debug=true, -- Gibt erweiterten Stacktrace aus
  })
end).catch (function (e)
  print "Caught exception:"
  print(e)
end)

▸

◼

✗

↻

≡

Neben der Behandlung von Ausnahmen (Fehlern) werden Try-Catch Umgebungen auch für die gewollte Behandelung von Signalen verwendet, um zum Beispiel den Programmfluß zu verlassen:

Nutzerdefinierte Signale

require 'Csp';
Try(function ()
  Seq({
    function () 
      print('Hello '..word)
    end,
    function ()
      raise "MySignal"
      print('Thank '..word)
    end
  },{
    word='shared',
  })
end).catch (function (sig)
  print "Got signal:"
  print(sig)
end)

▸

◼

✗

↻

≡

Koroutinen

Koroutinen werden im gleichen Kontext ausgeführt. Jedoch führt die CSP Bibliothek einen lokalen geteilten Kontext ein (indem hier z.B. die print Funktion fehlt und daher durch den geteilten Kontext explizit eingeführt werden muß).

Die CSP Bibliothek stellt verschiedene Prozesskommunikationsmoethoden zur Verfügung. Dabei muss zwischen den Koroutinen (Fibers) und den parallelen Prozessen (Threads) unterschieden werden. I.a. gibt es ein zusätzliches boolesches Argument bei der Erzeugung der Kommunikationsobjekte. Ein Beispiel eines Events ist nachfolgend geziegt.

Events haben zwei Methoden:

ev:await blockiert den aufrufenden Prozess (die Koroutine) bis ein Ereignis eingetreten ist;
ev:wkaeup löst das Ereignis aus.

Eine einfache Textausgabe (3)

require 'Csp';
Co({
  function (id)
    print(id..' starting')
    sleep(500)
    print('Hello '..word)
    ev:wakeup()
  end,
  function (id)
    print(id..' starting')
    -- sleep(500)
    ev:await()
    print('Thank '..word)
  end
},{
  ev = Event(true), -- Event(fiber:boolean) true: Fiber Event, und nicht Thread!
  word='shared',
  print=print
})
print('After Co..')

▸

◼

✗

↻

≡

Frage 4. Ist der Co Prozess synchron? D.h. wird auf die Terminierung aller Teilprozesse gewartet?

Lösung.

Ja, Co verhält sich synchron wie Par, d.g. die Terminierung der Teilprozesses wird abgewartet.

Aufgabe 5. Verschiebe die sleep Operation aus Prozess 1 in Prozess 2 (Kommentar entfernen und oben einfügen). Welches Problem kann bei der Verwendung eines Eventobjekts entstehen?

Lösung.

Das Signal kommt vor dem Await was zu einem Deadlock führen kann.

Parallele Prozesse

Die verwendung des Par Konstruktors erzeugt echte parallele Prozesse.
Parallele Prozesse können nur über definierte IPC Objekte wie Channels kommunizieren
Variablen können von parallelen Prozesse "geteilt" werden, aber nur als Kopie
Gemeinsam geteilte Objekte, Variablen und Funktionen müssen nach der Prozessfunktionsliste als Tabelle übergeben werden

Eine einfache Textausgabe mit Prozesskommunikation (2)

require 'Csp';
log('Creating Par..')
Par({
  function () 
    local word = ch:read();
    log('Hello '..name..' '..word)
  end,
  function ()
    ch:write('World');
    log('Thank '..name)
  end
},{
  ch=Channel(1),
  name='Joe',
})
log('After Par ..')

▸

◼

✗

↻

≡

Frage 6. Was passiert im Prozessfluss wenn der Par Konstruktor mit Fork ausgetauscht wird?

Lösung.

Kommunikationskanäle

Ein Channel wird zwischen mindestens zwei Prozessen verwendet um synchronisiert Daten auszutauschen
Die Daten werden automatisch serialisiert und wieder deserialisiert
Ein Channel wird mit dem Konstruktor Channel(n) erzeugt. Die Puffergröße n gibt die maximale Anzahl zwischengespeicherter Daten an.
Ein Channel mit n=0 kann nur von zwei Prozessen (einem Leser und einem Schreiber) verwendet werden → Rendezvous Protokoll
Es gibt zwei Operationen: channel:write(data) und channel:read()
- Die Leseoperation blockiert den aufrufenden Prozess solange bis wenigstens ein Datenelement in den Kanal geschrieben wird
- Die Schreibeoperation könnte bei einer Obergrenze an Elementen im Channel (die noch nicht abgerufen wurden) blockieren (optional)
- Aber: Wenn das Rendezvous Protokoll (n=0) verwendet wird blockiert die Schreiboperation solange bis der Leseprozess die Leseoperation ausführt (und umgekehrt)

Frage 7. Was läuft bei dem nächsten Beispiel falsch (beachte die Puffergröße vom Komm.kanal)?

Hinweis: Wenn der Par Prozess nicht terminiert, benutze den ◼ Knopf bis die Meldung mit dem Signal INTR erscheint! Ggfs. Konsolenupdate durchführen (rechts, Kreispfeil)

Lösung.

Aufgabe 8. Ändere das Programm so ab dass sich beide Prozesse richtig synchronisieren und der Par Prozess terminiert.

Lösung.

Eine komplexere Textausgabe mit Prozesskommunikation (3)

require 'Csp';
log('Creating Par..')
Par({
  function () 
    local word = ch:read();
    log('Hello '..name..' '..word)
    ch:write('You')
  end,
  function ()
    ch:write('World');
    local word = ch:read();
    log('Thank '..name..' '..word)
  end
},{
  ch=Channel(1),
  name='Joe',
})
log('After Par ..')

▸

◼

✗

↻

≡

Semaphoren

Das wichtigste Kommunikationsobjekt für Prozesse ist der Semaphor als ein geschützter Zähler
Es gibt zwei Operationen:
- semaphore:up() erhöht den Zähler bei jedem Aufruf um eins
- semaphore:down() erniedrigt den Zähler bei jedem Aufruf um eins, aber;
- Die Downoperation kann den Aufrufer blockieren wenn die Semaphoreninvariante verletzt werden würde

Schwache und starke Invariante des Semaphors (Randbedingungen die unter keinen Umständen verletzt werden dürfen):

\[ \text{S.counter}\ge{0}\\ \text{S.counter}=\text{S.init}+\sum\text{S:up}-\sum\text{S:down} \]

Ein Semaphor wird für Produzenten-Konsumenten Systeme und für die Koordination zwischen Prozessen eingesetzt
Ein Semaphor wird mit dem Konstruktor Semaphore(init) erzeugt. Der Startwert gibt den initialen Zählerwert an → die richtige Wahl ist relevant und bestimmt die Anwendung!

Im nächste Beispiel werden Semaphoren für die Synchronisation von parallel ausgeführten Arbeitsprozessen (Worker) mit einem Leitungsprozess (Master) verwendet. Es gibt bei der parallelen und verteilen Datenverarbeitung zwei wesentliche Phasen:

Verteilungsphase (Worker werden gestartet und warten auf Daten/Start der Berechnung bzw. dem Master, Verteilung der Daten)
Zusammenführungs- und Einsammlungsphase (der Master wartet auf die Worker und sammelt die Ergebnisse wieder ein)

Ein einfaches Produzenten-Konsumenten System

require 'Csp';
log('Creating Par..')
Par({
  function ()
    -- Der Master
    -- Distribute
    log('Master: Distributing..')
    for i=1,2 do prod:up() end
    -- Join
    log('Master: Collecting..')
    for i=1,2 do cons:down() end
    log('Master: Done.')
  end,
  function ()
    -- Worker 1
    prod:down()
    log('Worker 1: processing..')
    sleep(500)
    cons:up()
    log('Worker 1: Done.')
  end,
  function ()
    -- Worker 2
    prod:down()
    log('Worker 2: processing..')
    sleep(500)
    cons:up()
    log('Worker 2: Done.')
  end
},{
  prod=Semaphore(0),
  cons=Semaphore(0)
})
log('After Par ..')

▸

◼

✗

↻

≡

Frage 9. Wie ist der zeitliche Ablauf?

Lösung.

Frage 10. Welchen Wert haben die Semaphorenzähler von prod und cons am Ende?

Lösung.

Aufgabe 11. Verändere das Prozesssystem derart dass der Master außerhalb des (nach dem) Worker Pool (also im Hauptprozess) ausgeführt wird. Dazu muss der Fork Prozesskonstruktor verwendet werden. Warum?

Lösung.

Das Deadlock Problem: Dinierende Philosophen

Dinierende Philosophen ist ein paralleles System welches Semaphoren benutzt um physische Ressourcen abzubilden.
- Es gibt N Philosophen (N ungradzahlig)
- Jeder Philosoph kann sich in zwei Zuständen befinden: "Essend" oder "denkend"!
- Gabeln sind geteilte Ressourcen
- Um in den Zustand "Essend" zu gelangen benötigt der Philosoph i zwei Gabeln S_i und S_i+1 (Geschlossener Ring mit Modulo n Indizierung).
- Jede Gabel wird durch eine binäre Semaphore repräsentiert
- Die Philosophen und Gabeln stellen einen Ring dar, d.h. der letzte Philosoph N benötigt die letzte und erste Gabel.

Das DP Problem dient als klassisches Deadlock Problem und das Auftreten von Race Conditions (Wettrennen zwischen Threads) bei geteilten Ressourcen mit konkurrierenden und nebenläufigen (asynchronen) Zugriff

Zwei dinierende Philosophen - der Auftakt!

require 'Csp';
log('Creating Par..')
Par({
  function ()
    log('P1 Start')
    sync:await()
    -- Block Start
    -- sleep(math.floor(math.random()*10))
    log('P1: Thinking')
    fork1:down()
    -- yield()
    fork2:down()
    log('P1: Eating')
    fork1:up()
    fork2:up()
    -- Block End
    log('P1 End')
  end,
  function ()
    log('P2 Start')
    sync:await()
    -- Block Start
    -- sleep(math.floor(math.random()*10))
    log('P2: Thinking')
    fork2:down()
    -- yield()
    fork1:down()
    log('P2: Eating')
    fork2:up()
    fork1:up()
    -- Block End
    log('P1 End')
  end
},{
  sync=Barrier(2),
  fork1=Semaphore(1),
  fork2=Semaphore(1)
})
log('After Par ..')

▸

◼

✗

↻

≡

Die Barriere dient hier nur für einen synchronisierten Start der Philosophenprozesse

Aufgabe 12. Starte eine Reihe (10) von Durchläufen. Ist die Ablaufreihenfolge deterministisch?

Lösung.

Aufgabe. Führe in jedem Phil.prozess eine Zählschleife (for)(1,M) ein die den obigen Anweisungblock (Block) kapselt. Experimentiere mit den auskommentierten random delay (sleep in Millisekunden) und der Threadwechsel mit yield. Führe den Prozess aus, mit z.b. M=20 Durchläufen. Was lässt sich beobachten?

Lösung.

Aufgabe 14. Erweitere nun das parallele Prozesssystem auf N=5 Phils. Lässt sich ein Deadlock feststellen? Die Versuche müssen ggfs. mehrfach wiederholt werden (und nur auf Rechnern mit mehr als einer CPU/Core ist etwas auffälliges zu beobachten).

Lösung.

Frage 15. Was passiert hier? Was führt zum Versagen des parallelen Systems?

Lösung.

Frage 16. Wie kann das Versagen des Systems naiv verhindert werden?

Lösung.

Parallele Matrixberechnung

In der folgenden Aufgabe soll ähnlich wie den parallelen Zellulären Automaten eine Matrixberechnung auf vier Prozesse verteilt und partitioniert werden. Dabei gibt es zwei Phasen:

Es wird die elementweise Addition C=A+B duchgeführt

\[ \hat{{C}}=\hat{{A}}+\hat{{B}},{c}_{{{i},{j}}}={a}_{{{i},{j}}}+{b}_{{{i},{j}}} \]

Es wird die Summe aller Elemente und der Mittelwert berechnet:

\[ \overline{{C}}=\frac{{\sum\hat{{C}}}}{{\left|\hat{{C}}\right|}} \]

mit |M|: Anzahl der Elemente der Matrix M.

Die Matrizen A, B, und C werden hier durch geteilte Speichervariablen implementiert. Dabei ist eine geteilte Matrix ein Objekt mit den zwei Methoden mat:read(row,col) und mat:write(val,row,col).

Kooperative Parallele Matrixberechnung

Aufgabe 17. Implementiere im nachfolgenden Programmcode die Verteilungs- und Zusammenführungsphase der vier Arbeiterprozess durch einen Master(prozess) gemäß obigen Produzenten-Konsumenten Beispiel mit Semaphoren. Beide Phasen der Berechnung sind getrennt auszuführen.

Aufgabe 18. Überprüfe zuvor die Mittelwertberechnung durch direkte Berechnung (in Schleifen):

\[ {m}{e}{a}{n}=\frac{{\sum{\left(\hat{{A}}+\hat{{B}}\right)}}}{{\left|\hat{{A}}\right|}}=?\overline{{C}} \]

Partitionierte Parallele Datenverarbeitung (Zwei Phasen)

require 'Csp';
log('Creating Par..')
local A,B,C,D=
      Matrix:new({100,100},function () return math.random() end),
      Matrix:new({100,100},function () return math.random() end),
      Matrix:new({100,100},0),
      Matrix:new({4},0);
local prod,cons =
      Semaphore( TODO ), -- Hier richtigen Startwert einsetzen 
      Semaphore( TODO )  -- Hier richtigen Startwert einsetzen 
local mean0 = 0;
print(A:read(1,1),B:read(1,1),C:read(1,1))
-- Hier mean0 direkt aus A/B berechnen for i, for j, ...

-- Paritionen beschreiben
local  partitions = {
    -- Matrixpartitionen
    -- Hier Start- und Endindizes der Matrixpartition der Partition eintragen
    { rowStart,colStart,rowEnd,colEnd }, 
    { }, -- usw.
    { },
    { },
}

-- PART 1: C=A+B
Fork({
  function (id)
    -- Worker 1
    local part = paritions[id];
    log(id)
  end,
  function (id)
    -- Worker 2
    log(id)
  end,
  function (id)
    -- Worker 3
    log(id)
  end,
  function (id)
    -- Worker 4
    log(id)
  end,  
},{
  A=A,
  B=B,
  C=C,
  prod=prod,
  cons=cons,
  paritions=paritions,
})
-- Master 1
-- TODO
-- Part 2: mean=sum(C)/(100*100)
local mean=0
Fork({
  function (id)
    -- Worker 1
    log(id)
  end,
  function (id)
    -- Worker 2
    log(id)
  end,
  function (id)
    -- Worker 3
    log(id)
  end,
  function (id)
    -- Worker 4
    log(id)
  end,  
},{
  C=C,
  D=D,
  prod=prod,
  cons=cons,
  partitions = paritions,
})
-- Master 2
-- TODO
print(mean)

▸

◼

✗

↻

≡

Lösung.

Hilfe

Einreichung (Assignment #2025-94697 )

Prüfen

Bewerten (Lehrer)

Created by the NoteBook Compiler Ver. 1.36.4 (c) Dr. Stefan Bosse (Mon Jun 30 2025 17:47:17 GMT+0200 (Central European Summer Time))