Multi Threads

[_JON_]

Hallo Leute,

Ich benutzte Threads eigentlich recht selten, also meist nur meist nur ein Thread für das Fenster und ein für die Aktion.
Nun wirds aber mal Zeit mich mit dem Thema auseinander zu setzten

Und zwar hab ich folgendes Beispiel:

Eine Datei wird Blockweise gelesen und mit dem Block werden 3 Aufgaben bearbeitet.

Code: Alles auswählen

Define *Buffer, iRead, x
#Buffersize = 8192

Procedure Aufgabe1(*Buffer, BufferSize, iBlock)
  Debug "Aufgabe1 bearbeitet Block: " + iBlock
EndProcedure

Procedure Aufgabe2(*Buffer, BufferSize, iBlock)
  Debug "Aufgabe2 bearbeitet Block: " + iBlock
EndProcedure

Procedure Aufgabe3(*Buffer, BufferSize, iBlock)
  Debug "Aufgabe3 bearbeitet Block: " + iBlock
EndProcedure

*Buffer = AllocateMemory(#Buffersize)
If *Buffer
  If ReadFile(0, "X:\xpbios_509.bin")
    While 1 
      iRead = ReadData(0, *Buffer, #Buffersize)
      If Not iRead : Break : EndIf
      x + 1
      
      Aufgabe1(*Buffer, iRead, x)
      Aufgabe2(*Buffer, iRead, x)
      Aufgabe3(*Buffer, iRead, x)
      
      Debug "============================" + #CRLF$
      
      If iRead < #Buffersize : Break :EndIf
      
    Wend
    CloseFile(0)
  EndIf
  FreeMemory(*Buffer)
EndIf

Jede Aufgabe darf den Block lesen, ihn aber nicht verändern.
Wie mach ich das mit mehreren Threads anstatt normalen Procedere-aufrufen.
Ich will ja kein CreateThread-Wait-Thread für jeden Block, das wäre wohl nicht sehr effizient.

[NicTheQuick]

Meinst du sowas in der Art?

Code: Alles auswählen

Define *Buffer, iRead, x
#Buffersize = 8192

Prototype Aufgabe(*buffer, bufferSize.i, iBlock.i)

Structure info
	proc.Aufgabe
	*buffer
	bufferSize.i
	iBlock.i
EndStructure

Procedure AufgabeThread(*info.info)
	*info\proc(*info\buffer, *info\bufferSize, *info\iBlock)
	FreeStructure(*info)
EndProcedure

Procedure startAufgabe(proc.Aufgabe, *buffer, bufferSize.i, iBlock.i)
	Protected *info.Info = AllocateStructure(info)
	
	With *info
		\proc = proc
		\buffer = *buffer
		\bufferSize = bufferSize
		\iBlock = iBlock
	EndWith
	
	CreateThread(@AufgabeThread(), *info)
EndProcedure

Procedure Aufgabe1(*Buffer, BufferSize, iBlock)
	Debug "Aufgabe1 bearbeitet Block: " + iBlock
EndProcedure

Procedure Aufgabe2(*Buffer, BufferSize, iBlock)
	Debug "Aufgabe2 bearbeitet Block: " + iBlock
EndProcedure

Procedure Aufgabe3(*Buffer, BufferSize, iBlock)
	Debug "Aufgabe3 bearbeitet Block: " + iBlock
EndProcedure

*Buffer = AllocateMemory(#Buffersize)
If *Buffer
	If ReadFile(0, "/home/nicolas/Kartoffelgratin.pdf")
		While 1
			iRead = ReadData(0, *Buffer, #Buffersize)
			If Not iRead : Break : EndIf
			x + 1
			
			startAufgabe(@Aufgabe1(), *Buffer, iRead, x)
			startAufgabe(@Aufgabe2(), *Buffer, iRead, x)
			startAufgabe(@Aufgabe3(), *Buffer, iRead, x)
			
			Debug "============================" + #CRLF$
			
			If iRead < #Buffersize : Break :EndIf
			
		Wend
		CloseFile(0)
	EndIf
	; Zum Testen einfach mal eine Sekunde warten anstatt auf das Beenden der Threads zu warten
	Delay(1000)
	FreeMemory(*Buffer)
EndIf 

Das ist noch nicht ganz sauber. Siehe Delay().

[_JON_]

Hi Nic,

so in der Art habe ich auch erst gedacht, aber verbraucht die ständige Thread Erstellung
nicht zu viel Zeit bei kleiner Blockgröße?

[mhs]

Du kannst auch die Threads vorab erstellen und mit WaitSemaphore warten lassen, bis eine Aktion kommt. Dazu hast du dann eine zentrale (oder für jeden Thread separate) Aufgabenliste in die du die Aufgaben einträgst. Nach dem Eintragen informierst mit du SignalSemaphore alle (oder einen einzelnen) Threads, dass neue Aufgaben vorliegen. Der Thread liest dann die Aufgabe aus der Liste und löscht sie raus (Mutex nicht vergessen).

Alternativ statt dem Löschen kann auch ein Statusfeld einen anderen Status bekommen und der abarbeitende Thread am Ende einen Rückgabewert in die Aufgabe schreiben, der wiederum vom Mainloop gelesen und ausgegeben wird.

[NicTheQuick]

Vergiss meinen Code. Der hat noch ganz andere Fehler. Das hab ich gestern zu schnell geschrieben. Ich bastel mal was neues.

Ist es dir wichtig, dass das ganze auch mit großen Dateien funktioniert, die nicht in den RAM passen? Oder kann ich davon ausgehen, dass die Datei da immer rein passt?

[_JON_]

Das mit Semaphore und Mutex hört sich gut an, allerdings hab ich nirgends ein verständliches Beispiele gefunden.

Ist es dir wichtig, dass das ganze auch mit großen Dateien funktioniert, die nicht in den RAM passen? Oder kann ich davon ausgehen, dass die Datei da immer rein passt?

Ja, block weise lesen ist wichtig, da die Dateien schonmal mehrere hundert MB haben können.

[RSBasic]

Ich find den Beispielcode in der Hilfe sehr verständlich, einfach und nützlich: http://www.purearea.net/pb/german/manua ... phore.html

[NicTheQuick]

Wie soll der Ablauf sein?

Sobald die erste Aufgabe fertig ist, soll sie sich den nächsten Block aus der Datei schnappen und damit weiter machen?
Das heißt später kann es sein, das Aufgabe 1 schon mit Block 100 arbeitet und die Aufgaben 2 und 3 noch bei Block 30 sind?

Oder reicht die einfache Variante?
Dann würden alle Aufgaben als Threads gestartet werden und immer gemeinsam an einem Block arbeiten. Sollte ein Thread schneller sein, wartet er eben auf die anderen.

Im Grunde sind, was den zeitlichen Aspekt angeht, beide Lösungsansätze vermutlich ziemlich gleich schnell. Wobei letzterer den CPU-Cache besser ausnutzen kann, weil mehrere Threads an den selben Daten arbeiten. Und punktet bei schwächeren System mit wenigen CPU-Cores, weil gegen Ende die zunächst langsamen Threads nur noch alleine sind, weil alle anderen schon fertig sind.
Die zweite Version verbraucht außerdem weniger Hauptspeicher, während die erste Version immer die Differenz an Blöcken zwischen schnellster und langsamster Aufgabe in eine Warteschlange hält.

[_JON_]

Oder reicht die einfache Variante?
Dann würden alle Aufgaben als Threads gestartet werden und immer gemeinsam an einem Block arbeiten. Sollte ein Thread schneller sein, wartet er eben auf die anderen.

Ja genau sowas, für jeden Block arbeiten 3 Threads und wenn alle fertig sind wird der nächste Block gelesen
und wieder abgearbeitet.

@RSBasic
Ja das Beispiel in der PB-hilfe ist schon gelungen, allerdings nicht wirklich was ich suche.
Da gehts um 2 Threads die jeweils eine liste sperren und entsperren.

[NicTheQuick]

Dann hier mal meine Lösung zum Problem. Ich hoffe die Kommentare reichen so aus um es zu verstehen.

Code: Alles auswählen

Define *Buffer, iRead, x
#Buffersize = 8192

Prototype Aufgabe(*buffer, bufferSize.i, iBlock.i)

Structure Block
	*buffer       ; Pointer zum Puffer selbst
	size.i        ; Größe des Puffers
	iBlock.i      ; Nummer des Blocks
	waitSema.i    ; Semaphore um auf Befüllen den Blockes zu warten
	doneSema.i    ; Semaphore um auf Beendigung der Verarbeitung zu warten
	allDoneSema.i ; 
	stop.i        ; #True, wenn der Thread sich beenden soll
	List threads.i()
	cThreads.i    ; Anzahl der Threads
EndStructure

Procedure.i newBlock()
	Protected *block.Block = AllocateStructure(Block)
	
	With *block
		\waitSema = CreateSemaphore()
		\doneSema = CreateSemaphore()
		\allDoneSema = CreateSemaphore()
	EndWith
	
	ProcedureReturn *block
EndProcedure

Structure ThreadData
	*block.Block
	aufgabe.Aufgabe
EndStructure

Procedure AufgabeThread(*data.ThreadData)
	With *data\block
		Repeat
			; Warte bis ein neuer Block da ist
			WaitSemaphore(\waitSema)
			
			If \stop
				Break
			EndIf
			
			; Führe die Aufgabe auf dem Block aus
			*data\aufgabe(\buffer, \size, \iBlock)
			
			; Melde dem Hauptthread, dass du fertig bist
			SignalSemaphore(\doneSema)
			
			; Warte auf den Hauptthread, der dir mitteilt, das alle anderen Threads auch fertig sind
			WaitSemaphore(\allDoneSema)
		ForEver
	EndWith
	
	FreeStructure(*data)
EndProcedure

Procedure.i addThread(*block.Block, aufgabe.Aufgabe)
	Protected thread.i
	Protected *threadData.ThreadData = AllocateStructure(ThreadData)
	
	If Not *threadData
		ProcedureReturn #False
	EndIf
	
	With *block
		*threadData\block = *block
		*threadData\aufgabe = aufgabe
		If AddElement(\threads())
			\threads() = CreateThread(@AufgabeThread(), *threadData)
			If Not \threads()
				FreeStructure(*threadData)
				DeleteElement(\threads())
			EndIf
		Else
			FreeStructure(*threadData)
			ProcedureReturn #False
		EndIf
		
		\cThreads = ListSize(\threads())
	EndWith
	
	ProcedureReturn #True
EndProcedure

Procedure.i processBlock(*block.Block, *buffer, bufferSize.i, iBlock.i)
	Protected i.i
	
	With *block
		\buffer = *buffer
		\size = bufferSize
		\iBlock = iBlock
		
		; Sage allen Threads, dass sie ihre Arbeit beginnen sollen
		For i = 1 To \cThreads
			SignalSemaphore(\waitSema)
		Next
		
		; Warte, bis alle Threads fertig sind
		For i = 1 To \cThreads
			WaitSemaphore(\doneSema)
		Next
		
		; Melde allen Threads, das nun alle fertig sind und auf neue Arbeit warten können.
		For i = 1 To \cThreads
			SignalSemaphore(\allDoneSema)
		Next
	EndWith
EndProcedure

Procedure.i destroyBlock(*block.Block)
	With *block
		\stop = #True
		
		; Sage allen Threads, dass sie sich beenden sollen
		For i = 1 To \cThreads
			SignalSemaphore(\waitSema)
		Next
		
		; Warte auf Beendigung der Threads
		ForEach \threads()
			WaitThread(\threads())
		Next
		
		FreeSemaphore(\waitSema)
		FreeSemaphore(\doneSema)
		FreeSemaphore(\allDoneSema)
	EndWith
	
	FreeStructure(*block)
EndProcedure

; ============== Hauptprogramm ===============

Procedure Aufgabe1(*Buffer, BufferSize, iBlock)
	Debug "Aufgabe1 bearbeitet Block: " + iBlock
EndProcedure

Procedure Aufgabe2(*Buffer, BufferSize, iBlock)
	Debug "Aufgabe2 bearbeitet Block: " + iBlock
EndProcedure

Procedure Aufgabe3(*Buffer, BufferSize, iBlock)
	Debug "Aufgabe3 bearbeitet Block: " + iBlock
EndProcedure

Define.Block *block = newBlock()

addThread(*block, @Aufgabe1())
addThread(*block, @Aufgabe2())
addThread(*block, @Aufgabe3())

*buffer = AllocateMemory(#Buffersize)
If Not *buffer
	End
EndIf

If ReadFile(0, "/home/nicolas/Kartoffelgratin.pdf")
	While Not Eof(0)
		iRead = ReadData(0, *buffer, #Buffersize)
		x + 1
		
		processBlock(*block, *buffer, #Buffersize, x)
		
		Debug "============================"
	Wend
	CloseFile(0)
EndIf

FreeMemory(*buffer)

destroyBlock(*block)