Tutorial: Parallele Programmierung in PB

[cxAlex]

Das dürfte ein Konflikt mit dem Thread Local Storage (TLS) sein, das wird bei ThreadSafe aktiviert. in ECluster hatte ich schon mal Probleme damit, ich seh mir das mal an.

http://www.purebasic.fr/german/viewtopi ... y&start=20

[cxAlex]

Es sieht so aus als währs ein PB - Bug:

Code: Alles auswählen

; Thread Parameter
Structure Thread_Parameter
  *Title ; Titel
  *Text ; Text
  Flags.i ;Parameter
EndStructure

; Thread Procedure
Procedure MyThread(*Thread.Thread_Parameter)
  Protected Title.s, Text.s
  Title.s = PeekS(*Thread\Title)
  Text.s = PeekS(*Thread\Text)
  MessageRequester(Title, Text, *Thread\Flags)
 
  ; Speicher aufräumen
  FreeMemory(*Thread\Title)
  FreeMemory(*Thread\Text)
  FreeMemory(*Thread)
EndProcedure

; Thread - Starter
Procedure Thread_Launcher(Title.s, Text.s, Flags = 0)
  ; Neue Thread - Parameter anlegen
  Protected *Thread.Thread_Parameter = AllocateMemory(SizeOf(Thread_Parameter))
 
  ; Thread - Paramter schreiben
 
  ; Ich verwende absichtlich keine nativen Strings in der Struktur da man diese
  ; manuell freigeben müsste, daher Speicherblöcke
  *Thread\Title = AllocateMemory(StringByteLength(Title) + SizeOf(Character))
  PokeS(*Thread\Title, Title)
  *Thread\Text = AllocateMemory(StringByteLength(Text) + SizeOf(Character))
  PokeS(*Thread\Text, Text)
  *Thread\Flags = Flags

  Thread = CreateThread(@MyThread(), *Thread)
 
  ProcedureReturn Thread
EndProcedure

; Thread starten
Thread_Launcher("Thread","Ich bin ein Thread")
Thread_Launcher("Thread","Ich bin auch ein Thread")
MessageRequester("Main", "Ich bin das Hauptprogramm")

Ich hab an dem Code nichts geändert außer das ich den Text/Titel nicht direkt mit PeekS() durchschleife sondern über eine Zwischenvariable. Über solche Fehler gibts schon Bugreports im englischem Forum, ich werd den Code dort auch posten, da er anscheinend nur im Zusammenhang mit Threads auftritt.

PS: morgen gibts den nächsten Teil des Tutorials.

[ts-soft]

Ich denke mal, das liegt daran, weil MessageRequester in Threads nicht
zulässig sind! Die gehören zum Mainprocess, wie das Fenster. Evtl. die API
mit hWnd 0 verwenden, aber den MessageRequester nicht in Threads.
IMHO schlechtes Beispiel.

[cxAlex]

Nein, du kannst Messagerequester() und Fensterbefehle IMHO unter Windows in Threads ganz normal verwenden, lediglich unter Linux stellt das ein Problem dar.

Kann mich auch irren, denke aber nicht.

Ich wollte ein Beispiel machen wo man was sieht, da versteht man die Sache schneller.

[ts-soft]

Ein MessageRequester ist vom Design her AppModal, deshalb hat der in
Threads auch nichts zu suchen.
Such Dir lieber etwas unproblematische Beispiele

[KeyKon]

Kurzer zweischeneinwurf: vll sollte man Diskussion und Tutorial in getrennten Threads laufen lassen (klingt komisch mein ich aber so^^)...

[cxAlex]

Yo das währ gut, damit zwischen denn Abschnitten des Tut nicht so viele Posts sind.

Mods:

Könnet ihr bitte alles ab meinem 1. Tutorial-Eintrag in nen separaten Thread trennen?


Danke

[cxAlex]

@ts:

Es hat nichts mit dem MessageRequester zu tun, hier der selbe Fehler komplett ohne MessageRequester im Thread:

Code: Alles auswählen

; Thread Parameter
Structure Thread_Parameter
  *Title ; Titel
  *Text ; Text
  Flags.i ;Parameter
EndStructure

; Thread Procedure
Procedure MyThread(*Thread.Thread_Parameter)
  PeekS(*Thread\Title)
  PeekS(*Thread\Text)
 
  ; Speicher aufräumen
  FreeMemory(*Thread\Title)
  FreeMemory(*Thread\Text)
  FreeMemory(*Thread)
EndProcedure

; Thread - Starter
Procedure Thread_Launcher(Title.s, Text.s, Flags = 0)
  ; Neue Thread - Parameter anlegen
  Protected *Thread.Thread_Parameter = AllocateMemory(SizeOf(Thread_Parameter))
 
  ; Thread - Paramter schreiben
 
  ; Ich verwende absichtlich keine nativen Strings in der Struktur da man diese
  ; manuell freigeben müsste, daher Speicherblöcke
  *Thread\Title = AllocateMemory(StringByteLength(Title) + SizeOf(Character))
  PokeS(*Thread\Title, Title)
  *Thread\Text = AllocateMemory(StringByteLength(Text) + SizeOf(Character))
  PokeS(*Thread\Text, Text)
  *Thread\Flags = Flags

  Thread = CreateThread(@MyThread(), *Thread)
 
  ProcedureReturn Thread
EndProcedure

; Thread starten
Thread_Launcher("Thread","Ich bin ein Thread")
Thread_Launcher("Thread","Ich bin auch ein Thread")
MessageRequester("Main", "Ich bin das Hauptprogramm")

Anscheinen tritt der Fehler immer auf wenn man das Ergebniss von PeekS() nicht in eine Variable schreibt...

[cxAlex]

Thread - Kommunikation

Im 1. Beispiel wurde ein Thread erstellt der sich nach Erfüllung seiner Aufgabe (das Anzeigen der MessageBox) selbst wieder beendet hat.

Nun möchte man aber auch oft das ein Thread permanent im Hintergrund läuft und immer wieder Aufgaben erledigt. Dafür müssen wir mit dem Thread kommunizieren, ihm Signale geben. Das ist die Aufgabe von Semaphoren. Diese kommen natürlich auch in unsere Thread - Struktur:

Wichtig: Für Signalisierung an und von Threads wenn es geht möglichst Semaphoren verwenden.

Code: Alles auswählen

; Thread Parameter
Structure Thread_Parameter
  *Title ; Titel
  *Text ; Text
  Flags.i ;Parameter
  MsgAvaible.i ; Signalsiert einen neuen Msg - Requester
  MsgSend.i ; MsgRequester fertig (Rückmeldung)
  ThreadID.i ; ThreadID
EndStructure

; Thread Procedure
Procedure MyThread(*Thread.Thread_Parameter)
  Protected Title.s, Text.s
  Repeat
    ; Wartet auf Procedure
    WaitSemaphore(*Thread\MsgAvaible)
    
    Title.s = PeekS(*Thread\Title)
    Text.s = PeekS(*Thread\Text)
    MessageRequester(Title, Text, *Thread\Flags)
        
    ; Speicher aufräumen
    FreeMemory(*Thread\Title)
    FreeMemory(*Thread\Text)
    
    ; Bestätigt senden des MessageRequester
    SignalSemaphore(*Thread\MsgSend)
  ForEver
EndProcedure

; Thread - Starter
Procedure Thread_Launcher()
  ; Neue Thread - Parameter anlegen
  Protected *Thread.Thread_Parameter = AllocateMemory(SizeOf(Thread_Parameter))
  
  ; Erstellt Signal Semaphore
  *Thread\MsgAvaible = CreateSemaphore()
  ; Erstellt Rückmeldungs Semaphore (Muss bereits ein Signal enthalten für 1. MsgReq)
  *Thread\MsgSend = CreateSemaphore() : SignalSemaphore(*Thread\MsgSend)
  ; Startet den Thread
  *Thread\ThreadID = CreateThread(@MyThread(), *Thread)
  
  ProcedureReturn *Thread
EndProcedure

; Thread - Starter
Procedure Thread_DoMsg(*Thread.Thread_Parameter, Title.s, Text.s, Flags = 0)
  ; Thread - Paramter schreiben
  
  ; Nur 1. MessageBox/Thread (Wartet auf Rückmeldung vom Thread)
  WaitSemaphore(*Thread\MsgSend)
  
  ; Ich verwende absichtlich keine nativen Strings in der Struktur da man diese
  ; manuell freigeben müsste, daher Speicherblöcke
  *Thread\Title = AllocateMemory(StringByteLength(Title) + SizeOf(Character))
  PokeS(*Thread\Title, Title)
  *Thread\Text = AllocateMemory(StringByteLength(Text) + SizeOf(Character))
  PokeS(*Thread\Text, Text)
  *Thread\Flags = Flags
  
  ; Signalsiert neue Msg
  SignalSemaphore(*Thread\MsgAvaible)
  
EndProcedure

; Thread starten
Thread = Thread_Launcher()


Thread_DoMsg(Thread, "Thread", "Ich bin ein Thread")
Thread_DoMsg(Thread, "Thread", "Ich bin auch ein Thread")
MessageRequester("Main", "Ich bin das Hauptprogramm")

Nun haben wir auch ein neues Verhalten unseres alten MessageRequester - Codes: Nun kann innerhalb des Threads nur 1. MessageRequester aktiv sein, jedoch unabhängig davon ob im Main Thread oder einem anderen Thread ein MessageRequester läuft. Außerdem sehen wir hier bereits eine Rückmeldung vom Thread an die Main Procedure: MsgSend zeigt an das die MsgBox gesendet wurde.

Nun wollen wir ab und zu solche Threads auch wieder beenden. Sucht man in der PB Hilfe findet man dazu KillThread(). Finger weg! Der Befehl beendet den Thread unsauber, und es kann schnell zu IMAs oder sonstwas kommen.

Wichtig: Wenn es geht NIEMALS KillThread() verwenden

Wie machen wir es dann? Natürlich, wir setzen ein Flag in unserer Thread Procedure:

Code: Alles auswählen

; Thread Parameter
Structure Thread_Parameter
  *Title ; Titel
  *Text ; Text
  Flags.i ;Parameter
  MsgAvaible.i ; Signalsiert einen neuen Msg - Requester
  MsgSend.i ; MsgRequester fertig (Rückmeldung)
  DoStop.i ; StopFlag
  ThreadID.i ; ThreadID
EndStructure

; Thread Procedure
Procedure MyThread(*Thread.Thread_Parameter)
  Protected Title.s, Text.s
  Repeat
    ; Wartet auf Procedure
    WaitSemaphore(*Thread\MsgAvaible)
    
    If *Thread\DoStop ; Thread muss sich beenden
      ; Thread räumt auf
      FreeSemaphore(*Thread\MsgAvaible)
      FreeSemaphore(*Thread\MsgSend)
      ; Bestätigt Stop
      SignalSemaphore(*Thread\DoStop)
      ;Thread beendet
      ProcedureReturn
    EndIf
    
    Title.s = PeekS(*Thread\Title)
    Text.s = PeekS(*Thread\Text)
    MessageRequester(Title, Text, *Thread\Flags)
        
    ; Speicher aufräumen
    FreeMemory(*Thread\Title)
    FreeMemory(*Thread\Text)
    
    ; Bestätigt senden des MessageRequester
    SignalSemaphore(*Thread\MsgSend)
  ForEver
EndProcedure

; Thread - Starter
Procedure Thread_Launcher()
  ; Neue Thread - Parameter anlegen
  Protected *Thread.Thread_Parameter = AllocateMemory(SizeOf(Thread_Parameter))
  
  ; Erstellt Signal Semaphore
  *Thread\MsgAvaible = CreateSemaphore()
  ; Erstellt Rückmeldungs Semaphore (Muss bereits ein Signal enthalten für 1. MsgReq)
  *Thread\MsgSend = CreateSemaphore() : SignalSemaphore(*Thread\MsgSend)
  ; Startet den Thread
  *Thread\ThreadID = CreateThread(@MyThread(), *Thread)
  
  ProcedureReturn *Thread
EndProcedure

Procedure Thread_Stop(*Thread.Thread_Parameter)
  ; StopSignal erstellen
  *Thread\DoStop = CreateSemaphore() 
  ; Signal zum Prüfen senden
  SignalSemaphore(*Thread\MsgAvaible)
  ; Auf Bestätigung des Stop warten:
  WaitSemaphore(*Thread\DoStop)
  ; Alles aufräumen was der Thread nicht selbst aufräumen kann
  FreeSemaphore(*Thread\DoStop)
  FreeMemory(*Thread)
EndProcedure

; Thread - Starter
Procedure Thread_DoMsg(*Thread.Thread_Parameter, Title.s, Text.s, Flags = 0)
  ; Thread - Paramter schreiben
  
  ; Nur 1. MessageBox/Thread (Wartet auf Rückmeldung vom Thread)
  WaitSemaphore(*Thread\MsgSend)
  
  ; Ich verwende absichtlich keine nativen Strings in der Struktur da man diese
  ; manuell freigeben müsste, daher Speicherblöcke
  *Thread\Title = AllocateMemory(StringByteLength(Title) + SizeOf(Character))
  PokeS(*Thread\Title, Title)
  *Thread\Text = AllocateMemory(StringByteLength(Text) + SizeOf(Character))
  PokeS(*Thread\Text, Text)
  *Thread\Flags = Flags
  
  ; Signalsiert neue Msg
  SignalSemaphore(*Thread\MsgAvaible)
  
EndProcedure

; Thread starten
Thread = Thread_Launcher()


Thread_DoMsg(Thread, "Thread", "Ich bin ein Thread")
Thread_DoMsg(Thread, "Thread", "Ich bin auch ein Thread")
MessageRequester("Main", "Ich bin das Hauptprogramm")

; Aufräumen:

Thread_Stop(Thread)

Auch hier wieder eine Rückmeldung vom Thread zur Bestätigung des Beenden. Der Thread räumt sich hier selbst auf, nur was der Thread nicht selbst freigeben kann muss die Beenden - Procedure machen.

Auf dieselbe Weise kann man auch Pause/Resume realisieren, währe hier aber mehr oder weniger sinnlos.

Pause/Resume würde sich lohnen wenn der Thread mehrere Aufgaben anstehen lassen würde und diese nach der Reihe verarbeiten, nicht aber bei max. 1 aktiven Aufgabe die man nicht bei der Ausführung Pausieren kann. Das ist genau das was meine JobQueue macht, hierzu wird Stacking usw. benutz, darauf werde ich aber nicht genauer eingehen da das den Rahmen des Tutorials sprengen würde.

Das waren mal die Grundlagen der Thread - Kommunikation, im nächsten Teil werde ich etwas über Thread - Synchronisation schreiben, die Themen überlappen sich ein wenig, dort werde ich dann auch erläutern wann man den Zugriff auf geteilte Ressourcen schützen muss.

*Thread ist in diesem Fall bereits eine geteilte Ressource, jedoch liegt in diesem Fall keine Readers/Writers Problematik o.ä. vor und alle Variablen sind atomic, also muss hier nichts besonders geschützt werden, Erklärungen später im Tutorial.

[cxAlex]

Thread - Synchronisation

In vielen Fällen muss der Thread auf Daten zugreifen die auch vom Haupthread oder einem anderen Thread benötigt werden. Dieser Zugriff muss geschützt werden, bzw. der abwechselnde Zugriff synchronisiert.

Zugriffe auf Atomic - Variablen (x86: Byte,Word,Character,Long,Float; x64: +Double,Quad) müssen nicht geschützt werden. Hierbei kann aber das Readers - Writers Problem auftauchen, hierzu später im Tutorial.

Mal ein kleines Beispiel eines Konflikt beim Zugriff auf eine geteilte Ressource:

Code: Alles auswählen

; Unsere altbekannte Thread-Data Strukture
Structure Thread_Data
  Res.i
EndStructure

Procedure Thread(*Thread.Thread_Data)
  
  ; Schreiben
  For i = 1 To 10000
    WriteStringN(*Thread\Res, "-----------------------------------------------------------------")
  Next
EndProcedure

; Auch den Thread Launcher kennen wir schon
Procedure Thread_Laucher(Res)
  Protected *Thread.Thread_Data = AllocateMemory(SizeOf(Thread_Data))
  *Thread\Res = Res
  ProcedureReturn CreateThread(@Thread(), *Thread)
EndProcedure

File = CreateFile(#PB_Any, GetTemporaryDirectory() + "tmp_123.tmp")
Thread = Thread_Laucher(File)

; Schreiben
For i = 1 To 100
  WriteStringN(File, "-----------------------------------------------------------------")
Next

CloseFile(File)

Vielleicht kein ideales Beispiel aber mir ist einfach nichts anderes eingefallen. In diesem Beispiel versuchen der Haupt und ein NebenThread gleichzeitig in dieselbe Datei zu schreiben. Nun wird die Datei später nicht so aussehen wie gewünscht, Zeilen werden fehlen, Zeilen werden falsch sein, und der Effekt verstärkt sich noch wenn man mehr Threads verwendet:

Code: Alles auswählen

; Unsere altbekannte Thread-Data Strukture
Structure Thread_Data
  Res.i
EndStructure

Procedure Thread(*Thread.Thread_Data)
  
  ; Schreiben
  For i = 1 To 10000
    WriteStringN(*Thread\Res, "-----------------------------------------------------------------")
  Next
EndProcedure

; Auch den Thread Launcher kennen wir schon
Procedure Thread_Laucher(Res)
  Protected *Thread.Thread_Data = AllocateMemory(SizeOf(Thread_Data))
  *Thread\Res = Res
  ProcedureReturn CreateThread(@Thread(), *Thread)
EndProcedure

File = CreateFile(#PB_Any, GetTemporaryDirectory() + "tmp_123.tmp")
Thread1 = Thread_Laucher(File)
Thread2 = Thread_Laucher(File)
Thread3 = Thread_Laucher(File)
Thread4 = Thread_Laucher(File)

; Schreiben
For i = 1 To 100
  WriteStringN(File, "-----------------------------------------------------------------")
Next

CloseFile(File)

Deswegen müsse wir den Zugriff synchronisieren. Hierzu verwenden wir eine Mutex. Eine Mutex ist ein Objekt das immer nur von einem Thread gesperrt werden kann, alle anderen müssen warten bis der Thread die Mutex wieder freigibt.

Ungefähr zu vergleichen mit einem Raum zu dem es nur einen Schlüssel gibt, immer nur eine Person kann rein, dann fällt die Tür wieder ins Schloss und keiner kann rein bis Sie wieder rauskommt und den Schlüssel weitergibt.

Hierbei kann bei unsauberer Programmierung zu einem Death-Lock kommen, auch hierzu später im Tutorial.

Das Unseres Codes würde so gehen, natürlich übergeben wir die Mutext mit unseren Thread - Parametern:

Code: Alles auswählen

; Unsere altbekannte Thread-Data Strukture
Structure Thread_Data
  Res.i
  Mutex.i ; Schutzmutex
EndStructure

Procedure Thread(*Thread.Thread_Data)
  
  LockMutex(*Thread\Mutex) ; Mutex sperren
  ; Schreiben
  For i = 1 To 10000
    WriteStringN(*Thread\Res, "-----------------------------------------------------------------")
  Next
  UnlockMutex(*Thread\Mutex) ; Mutex freigeben
  
  FreeMemory(*Thread)
EndProcedure

; Auch den Thread Launcher kennen wir schon
Procedure Thread_Laucher(Res, Mutex)
  Protected *Thread.Thread_Data = AllocateMemory(SizeOf(Thread_Data))
  *Thread\Res = Res
  *Thread\Mutex = Mutex ; Mutex übergeben
  ProcedureReturn CreateThread(@Thread(), *Thread)
EndProcedure

Mutex = CreateMutex() ; Mutex erstellen
File = CreateFile(#PB_Any, GetTemporaryDirectory() + "tmp_123.tmp")
Thread1 = Thread_Laucher(File, Mutex)
Thread2 = Thread_Laucher(File, Mutex)
Thread3 = Thread_Laucher(File, Mutex)
Thread4 = Thread_Laucher(File, Mutex)

LockMutex(Mutex) ; Mutex sperren
; Schreiben
For i = 1 To 100
  WriteStringN(File, "-----------------------------------------------------------------")
Next
UnlockMutex(Mutex) ; Mutex freigeben

WaitThread(Tread1)
WaitThread(Tread2)
WaitThread(Tread3)
WaitThread(Tread4)


FreeMutex(Mutex)
CloseFile(File)

Natürlich kann die Mutex eine gewaltige Performance-Bremse sein, wenn der Thread die ganze zeit nur auf die Mutex wartet, hier währe man wahrscheinlich mit linearem Schreiben schneller.

Leider gibt es bei Threads keine eierlegende Wollmilchsau, man muss die Synchronisation immer an den Einzelfall anpassen.