Thread Pool

[Christian+]

Da ja inzwischen eigentlich jeder Rechner mehrere Rechenkerne hat bin ich gerade dabei mal zu schauen ob ich nicht einige meiner Entwicklungen etwas beschleunigen kann.
Problem dabei ist dass die meisten Aufgaben nicht lange genug dauern damit sich das Erstellen eines Thread so richtig lohnt oder man hat so viele hat dass es Unsinn wäre für jeden einen Thread zu erstellen.
Um diese Probleme zu umgehen entstand jetzt dieser Quellcode der es erleichtern soll mehr Rechenkerne zu nutzen in dem mehre Threads als Gruppe angelegt werden können die dann nacheinander alle an sie geschickten Aufgaben abarbeiten.
Vielleicht kann es ja noch jemand außer mir brauchen oder hat sogar noch andere gute Ideen für so etwas.

Update: Kleine Code Änderung + Kommentare

Code: Alles auswählen

; -------------------------------------------------------------------------------
; -------------------------------------------------------------------------------
; ThreadPool.pb
; 28-02-2012
; Christian+
; -------------------------------------------------------------------------------
; -------------------------------------------------------------------------------
EnableExplicit

; -------------------------------------------------------------------------------
; RunThreadData Structure
; -------------------------------------------------------------------------------
; Speichert alle vom Thread Pool benötigten Daten.
; -------------------------------------------------------------------------------
Structure RunThreadData
  *ThreadProcedure
  *Parameter
  Semaphore.i
EndStructure

; -------------------------------------------------------------------------------
; ThreadPoolData Structure
; -------------------------------------------------------------------------------
; Speichert alle von den Thread Pool Threads benötigten Daten.
; -------------------------------------------------------------------------------
Structure ThreadPoolData
  Semaphore.i
  Mutex.i
  List Tasks.RunThreadData()
  List Threads.i()
EndStructure

; -------------------------------------------------------------------------------
; RunThread(*ThreadPool.ThreadPoolData)
; -------------------------------------------------------------------------------
; Die vom Thread Pool als Threads verwendete Procedure.
; -------------------------------------------------------------------------------
; *ThreadPool.ThreadPoolData
;     = Zeiger auf den Thread Pool (= die ThreadPoolData Strcuture).
; -------------------------------------------------------------------------------
Procedure RunThread(*ThreadPool.ThreadPoolData)
  Protected *ThreadProcedure, *Parameter
  Protected Semaphore.i
  Repeat
    WaitSemaphore(*ThreadPool\Semaphore)
    LockMutex(*ThreadPool\Mutex)
    FirstElement(*ThreadPool\Tasks())
    *ThreadProcedure = *ThreadPool\Tasks()\ThreadProcedure
    *Parameter = *ThreadPool\Tasks()\Parameter
    Semaphore = *ThreadPool\Tasks()\Semaphore
    DeleteElement(*ThreadPool\Tasks())
    UnlockMutex(*ThreadPool\Mutex)
    If *ThreadProcedure = 0 And *Parameter = 0
      Break
    EndIf
    CallFunctionFast(*ThreadProcedure, *Parameter)
    If Semaphore
      SignalSemaphore(Semaphore)
    EndIf
  ForEver
EndProcedure

; -------------------------------------------------------------------------------
; Run(*ThreadPool.ThreadPoolData, *ThreadProcedure, *Parameter, Semaphore.i = 0)
; -------------------------------------------------------------------------------
; Übergibt die Angegebene Methode an den Thread Pool zum im Hintergrund
; abarbeiten. Ähnlich wie CreateThread() die Proceduren unterliegen
; auch den gleichen Einschränkungen.
; -------------------------------------------------------------------------------
; *ThreadPool.ThreadPoolData
;     = Zeiger auf den Thread Pool (= die ThreadPoolData Strcuture).
; *ThreadProcedure
;     = Zeiger auf die Procedure die als Thread ausgeführt werden soll.
; *Parameter
;     = Zeiger auf Daten wird beim Ausführen der Procedure als Parameter an
;       die Procedure übergeben.
; Semaphore.i = 0
;     = Optionales Semaphore für das ein SignalSemaphore ausgeführt wird wen
;       die Procedure ausgeführt wurde.
; -------------------------------------------------------------------------------
Procedure Run(*ThreadPool.ThreadPoolData, *ThreadProcedure, *Parameter, Semaphore.i = 0)
  LockMutex(*ThreadPool\Mutex)
  LastElement(*ThreadPool\Tasks())
  AddElement(*ThreadPool\Tasks())
  *ThreadPool\Tasks()\ThreadProcedure = *ThreadProcedure
  *ThreadPool\Tasks()\Parameter = *Parameter
  *ThreadPool\Tasks()\Semaphore = Semaphore
  UnlockMutex(*ThreadPool\Mutex)
  SignalSemaphore(*ThreadPool\Semaphore)
EndProcedure

; -------------------------------------------------------------------------------
; CreateThreadPool(*ThreadPool.ThreadPoolData, Number.i)
; -------------------------------------------------------------------------------
; Legt einen neuen ThreadPool an.
; -------------------------------------------------------------------------------
; *ThreadPool.ThreadPoolData
;     = Zeiger auf den Thread Pool (= die ThreadPoolData Strcuture).
; Number
;     = Anzahl an Thread die verwendet werden soll.
; -------------------------------------------------------------------------------
Procedure CreateThreadPool(*ThreadPool.ThreadPoolData, Number.i)
  Protected i.i
  *ThreadPool\Semaphore = CreateSemaphore()
  *ThreadPool\Mutex = CreateMutex()
  For i = 1 To Number
    AddElement(*ThreadPool\Threads())
    *ThreadPool\Threads() = CreateThread(@RunThread(), *ThreadPool)
  Next
EndProcedure

; -------------------------------------------------------------------------------
; CloseThreadPool(*ThreadPool.ThreadPoolData)
; -------------------------------------------------------------------------------
; Beendet einen ThreadPool und wartet dabei bis die Warteschlage abgearbeitet ist.
; -------------------------------------------------------------------------------
; *ThreadPool.ThreadPoolData
;     = Zeiger auf den Thread Pool (= die ThreadPoolData Strcuture).
; -------------------------------------------------------------------------------
Procedure CloseThreadPool(*ThreadPool.ThreadPoolData)
  Protected i.i, Size.i
  Size = ListSize(*ThreadPool\Threads())
  For i = 1 To Size
    Run(*ThreadPool, 0, 0)
  Next
  For i = 1 To Size
    WaitThread(*ThreadPool\Threads())
    DeleteElement(*ThreadPool\Threads(), 1)
  Next
EndProcedure

; -------------------------------------------------------------------------------
; -------------------------------------------------------------------------------



; -------------------------------------------------------------------------------
; Beispiel:
; (Nicht sehr sinnvoll aber es sollte ja möglichst kurz sein.)
; -------------------------------------------------------------------------------

Procedure Thread(*Parameters)
  Protected i.i
  For i = 0 To 1000000
  Next
EndProcedure

Define i.i, Time.i
OpenConsole()


Time = ElapsedMilliseconds()
Define ThreadPool.ThreadPoolData
;Thread Pool mit 4 Threads anlegen
CreateThreadPool(@ThreadPool, 4)
;50 Aufgaben an den Thread Pool senden
For i = 0 To 50
  Run(ThreadPool, @Thread(), 0)
Next
;Thread Pool schließen und Warten bis alle abgearbeitet sind
CloseThreadPool(ThreadPool)
time = ElapsedMilliseconds() - time
PrintN("Mit Threads: " + Str(time) + "ms")


Time.i = ElapsedMilliseconds()
For i = 0 To 50
  Thread(0)
Next
time = ElapsedMilliseconds() - time
PrintN("Ohne Threads: " + Str(time) + "ms")


Input()
End

[7x7]

Warum gehen eigentlich die meisten, die hier einen Code posten, davon aus, dass IHR Code keine Erklärung
oder eingefügte Kommentare braucht?

Sowas schaue ich mir gar nicht erst an.

[bobobo]

Werden die Threads einzelnen Kernen zugewiesen ?

[Christian+]

@7x7
Du hast nicht ganz Unrecht ich werde versuchen mich zu bessern aber wenn du es dir sowieso nicht anschaust brauchst du auch nichts meckern es zwingt dich keiner meinen Code anzuschauen.
@bobobo
Nun darum kümmert sich ja das Betriebssystem es geht ja darum überhaupt erst mal die Arbeit auf so viele Thread zu verteilen dass das Betriebssystem genug Threads hat um alle Kerne des Prozessors auszulasten. Die meisten einfachen Programme auch meine Nutzen viel zu wenige Threads so dass auch auf Rechnern mit 4,6 oder 8 Kernen nichts schneller läuft wenn die CPU gerade Mal bei 25% oder 50% last ist weil es nicht besser parallelisiert werden kann. Mit meinem Code will ich genau das dann verhindern in dem ich z.B. einmal 8 Threads anlege und diesen dann die Aufgaben schicke die sie abarbeiten aber eben mit max. so vielen Kernen wie Threads angelegt wurden und nicht nur mit einem. Der Vorteil bei dem Code ist das ohne das ständige neuanlegen von Threads was gerade bei kleinen Aufgaben im Verhältnis viel Zeit kostet und sich nur für größere Aufgaben anbietet ausgekommen wird und dass man sich nicht drum kümmern muss wie viele Threads man gerade hat man legt einfach nur noch die Anzahl fest und übergibt dann was zu tun ist und es nutzt die Threads so gut wie möglich.

[Danilo]

Vielleicht kann es ja noch jemand außer mir brauchen oder hat sogar noch andere gute Ideen für so etwas.

Wenn ich es richtig gesehen habe, fehlt noch eine Möglichkeit die Ergebnisse der Threads zu bekommen.

Ein alter Code von mir dazu (nur 32bit, CountList() durch ListSize() ersetzen): mehrere Aufgaben parallel abarbeiten lassen

[Nino]

Hallo,

ich kenne mich leider mit Threads kaum aus. Gerade deshalb finde ich diesen "Thread Pool" sehr interessant, weil -- wenn ich es richtig verstanden habe -- ich nur meine Threads in den Pool "hineinzuwerfen" brauche, und dieser und das Betriebssystem sich dann automatisch um eine Multicore-Verarbeitung kümmern. Stimmt das so ungefähr? Das fände ich ziemlich genial!

Allerdings muss ich sagen, dass ich von Deinem Code, Christian+, leider so gut wie nichts verstehe. Daher ist er so wie er im Moment vorliegt für mich nur ein nettes Beispiel, aber leider nicht ernsthaft zu benutzen. Wie soll jemand, der von Threads keine Ahnung hat, den Code ohne jegliche Erklärung verstehen? Und andere Leute wie Danilo, die sich gut mit sowas auskennen, brauchen das eh nicht weil sie sich das selbst schreiben können.

Also wie gesagt, ich finde das ziemlich genial, und ich danke Dir dass Du das hier gepostet hast. Aber es wäre klasse, wenn Du das jetzt noch für die Zielgruppe der "Thread-Dummies" verdaubar und selbständig benutzbar machst.

Ich fände es gut, wenn Du die Bedeutungen der verschiedenen Prozedur-Parameter erklären könntest. Und kann man auch mehrere Threads zugleich in den Thread Pool hineingeben? Wenn ja, wie geht das? Auch z.B. in folgender Situation kann man ja Threads gut gebrauchen: Man hat ein GUI, im Hintergrund läuft ein längerer Prozess (z.B. Kopieren von Dateien):

Code: Alles auswählen

Enumeration
   #Window1
EndEnumeration

Enumeration
   #Button
   #Checkbox
EndEnumeration


Procedure Kopieren()
   Protected i.i
   
   For i = 1 To 1000
      Delay(200)
   Next
EndProcedure   


Define Event.i

OpenWindow(#Window1, 0, 0, 140, 90, "Beispiel", #PB_Window_SystemMenu | #PB_Window_ScreenCentered)
ButtonGadget  (#Button,   20, 10, 90, 25, "Kopieren")
CheckBoxGadget(#Checkbox, 20, 50, 90, 20, "Markiere mich")

Repeat
   Event = WaitWindowEvent()
   Select Event
      Case #PB_Event_Gadget
         Select EventGadget()
            Case #Button
               Kopieren()
            Case #Checkbox
               Debug "Checkbox"
         EndSelect
   EndSelect
Until Event = #PB_Event_CloseWindow

Ohne einen Thread ist das GUI während des längeren Kopiervorgangs nicht benutzbar, weil es nicht auf Ereignisse reagieren kann. Lässt sich auch in so einer Situation der Thread Pool anwenden?

Und was muss man generell bei Verwendung des Thread Pools beachten?
Kann man z.B. auch Sortierroutinen in den Thead Pool hineingeben, oder gerät die Sortierung dabei durcheinander?

Danke und Grüße, Nino

[Christian+]

@Danilo
Einen Rückgabe Wert haben Threads ja auch nicht ich habe dies um dem Code einfach zu halten eingespart mal schauen ob ich da noch was ändere. Wo das Ergebnis hin soll kann ja mit Hilfe des Parameters mitgegeben werden und die Möglichkeit festzustellen ob die Aufgabe schon abgearbeitet wurde oder auch darauf zu warten habe ich ja über einer optionales Semaphore eingebaut. Werde mir aber deine Lösung jetzt mal noch Gründlich anschauen.
@Nino
Ja statt selbst ständig CreateThread() aufzurufen wobei du beachten musst das du es nicht mit zu kleinen Aufgaben machst da dann das Thread erstellen im Verhältnis zu langte dauert übergibst du es einfach an den Thread Pool. Das hat den Vorteil dass auch nur sehr kurz dauernde Aufgaben effizient ausgeführt werden da das ständige neuanlegen von Threads wegfällt außerdem verhindert es in einer Schleife mal 1000 Threads zu generieren was zwar 1000 Kerne nutzen würde aber auf einem DualCore dann auch wieder ehr bremsen würde weil so viel Threads ja auch verwaltet werden müssen und gerade wenn man selbst ein paar Lock/Unlock Mutex braucht kann es bei zu vielen Threads noch ungünstiger werden. Ich nutzte das weil es meist schwer ist mehrere Sinnvolle Threads anzulegen und ich so alles von kurzen bis sehr langen Aufgaben einfach damit abarbeiten kann. Deine Frage wegen Sortierroutinen wenn es eine ist die Threads nutzen kann man auch den Thread Pool dafür benutzen. Der Thread Pool zerlegt ja die einzelnen Prozeduren nicht so dass es für das Ergebnis egal ist sondern führt eben mehrere gleichzeitig aus z.B. wenn du mehrere Arrays sortieren willst denn falls die nicht wirklich sehr große sind wird CreateThread so viel Zeit brauchen das es sich darüber umzusetzen kaum lohnt. Ich werde mal schauen dass ich noch ein weiteres Beispiel erstelle bzw. deines mal anpasse.

[WPö]

Moin!

Christian, ich habe ein sehr seltsames Verhalten gefunden. Bei Deinem Beispiel werden 51x1.000.001 Schleifendurchläufe ausgeführt. Während der Abarbeitung braucht das Programm mit Nebenläufigkeit 13,9s bei Vollast beider Kerne, danach 4,5s bei 55 und 50% Auslastung. Da die Grundlast etwa 2x8% ist, komme ich auf 25,6 mit zu 4 Kernsekunden ohne Nebenläufigkeit.

Auch bei Veränderung auf 2x25.000.001 oder 1.001x50.001 Durchläufe kommt bei mir dasselbe 'raus. Daraus darf man wohl schließen, daß die Verwaltung der Nebenläufigkeit kein (großes) Problem darstellt. Bei größeren Aufgaben wie 2x500.000.001 Durchläufe erhöht sich der Rechenbedarf proportional: 270s zu 93s.

Schau doch bitte, ob Du dasselbe Verhalten produzieren kannst. Warum werden ohne Nebenläufigkeit beide Kerne genutzt und warum läuft dann die Abarbeitung 6mal so schnell? Ach ja, für diesen Test nutzte ich noch die 4.51er Version von Purebasic. Das sollte aber auch keine Schwierigkeit darstellen.

Gruß - WPö

[Christian+]

Also dein Text ist recht schwer zu verstehen aber mal zuerst noch eine paar Fragen.
Hast du den Debugger an? bzw. ThreadSave nicht aktiv? Wenn beides zutrifft sollte das das Problem sein.
Falls ja mach auf jeden Fall ThreadSave rein und da der Debugger viel bremst kann es auch nicht schaden den mal auszumachen.
Und zwei Kerne sollten nur bei der ThreadPool Variante ausgelastet werden oder wenn du noch andere Software laufen hast. Vielleicht auch wenn der Test zu kurz ist. Bei einigen Sekunden Laufzeit erst habe ich bei mir dann konstante Werte die nur noch die +- 5% Abweichungen was sonst halt so gerechnet wird haben.
Werde mal schauen ob ich es nachstellen kann bei mir.

[Kiffi]

Nebenläufigkeit

man kann es auch übertreiben...

Grüße ... Kiffi