Probleme mit Threads und Aufgaben-Pool

**endeffects**

Hallo,

ich habe leider ein Problem mit Threads die sich aus dem
Mainthread mit Aufgaben über Synchronize() versorgen sollen.

Code zum Erzeugen der Threads:

markieren

Delphi-Quellcode:

			    for Loop:= 0 to length(ThreadArray) - 1 do

    begin

      ThreadArray[Loop]:= TMyThreads.Create(true);

      ThreadArray[Loop].FIndex:= Loop;

      ThreadArray[Loop].Priority:= tpIdle;

      ThreadArray[Loop].FreeOnTerminate:= True;

      ThreadArray[Loop].Resume;

    end;

Im Thread selbst starte ich dann Synchronize um
aus dem MainThread die Aufgabe zu übergeben:

Code aus dem Thread:

markieren

Delphi-Quellcode:

			    Synchronize(NewJob);

    procedure TMyThreads.NewJob;

    begin

      MainForm.WorkPool(FIndex);

    end;

Sofern noch eine Aufgabe ansteht wird im MainThread
folgende Methode ausgelöst:

Code aus dem MainThread:

markieren

Delphi-Quellcode:

			procedure TMainForm.WorkPool(Index: integer);

begin

    ....

    ThreadArray[Index].FTodo:= TodoList[TodoCounter];

    ....

end;

Ich hab nun schon eine ganze Weile hin und her geprüft
und die Zuweisung ThreadArray[Index].FTodo:= TodoList[TodoCounter];
aus TMainForm.WorkPool wird definitiv ausgeführt.
Trotzdem kommt es vor das diese beim Thread nicht ankommt.

Momentan bin ich ein wenig ratlos, hat Jemand vielleicht
eine Idee woran das liegen könnte?

**alzaimar**

Falls Du D6 benutzt, vergiss Synchronize, das klappt nicht.
Weiterhin solltest Du den Datenaustausch mit Threads irgendwie mit CriticalSections sichern. Anstatt direkt das Feld 'FTodo' zu setzen (ist doch nicht OOP, Mensch), schreib Dir eine Get- und eine Set-Methode im Thread. Den Zugriff auf das Feld fToDo kapselst du mit einer Critical Section.
So ist sichergestellt, das sich lese- und schreiboperationen nicht in die Quere kommen.

Obwohl das bei Dir eigentlich nicht passieren dürfte, da de NewJob Methode im Kontext des Hauptthreads ausgeführt wird...

Wenn ich mir das recht überlege, ist das aber nicht sonderlich cool, wie Du das machst. Ich würde die Todo-Liste erstmal als TThreadList implementieren, weil eben viele Threads darauf zugreifen. Das sollten sie auch gleichzeitig können. Ergo:

zusammenfalten · markieren

Delphi-Quellcode:

			Type

  TJobList = Class

  Private

    fList : TThreadList;

  Public

    Function GetNextJob : TJob;

    Procedure AddJob (aJob : TJob);

    End;

Function TJobList.GetNextJob : TJob;

Var

  l  :TList;

Begin

  l := fList.LockList;

  Try

    if l.Count = 0 Then 

      Result := Nil

    else begin

      Result := l[l.count - 1];

      l.delete (l.count - 1);

      End;

  Finally

     fList.unlockList;

     End;

End;

Procedure TJobList.AddJob (aJob : TJob);

Var

  l  :TList;

Begin

  l := fList.LockList;

  Try

    l.Add (aJob);

  Finally

    fList.unlockList;

    End;

End;

Nun kannst Du die Jobliste füllen (auch während die Threads ackern). Jeder Thread holt sich dann, wenn er fertig ist, den nächsten Job per GetNextJob und macht weiter. Liefert GetNextJob den Wert nil, macht der Thread eben nichts. Das ist ein kleines Problem, weil es nicht so leicht ist, einem Thread zu sagen, er soll nichts machen (also, auch keine CPU verbraten). Man muss ein Synchronisationsobjekt nehmen, ich würde es mit einer Semaphore versuchen. Jedesmal, wenn ein Job per AddJob in die Joblistee reingepackt wird, wird die Sempahore um eins erhöht.
Jeder Thread wartet, bis die Semaphore<>0 ist und holt sich einen Job aus der Liste. Nicht ausprobiert, sollte aber klappen.

**endeffects**

hmm ich hoffe du kannst mir vielleicht noch ein wenig weiter helfen,
so gut kenne ich mich mit objektpascal leider nicht aus.

ich hab jetzt folgenden code in TMainForm aufgenommen
und erhalte an den kommentierten stellen den Fehler:
"Inkompatible Typen: String und Pointer"

zusammenfalten · markieren

Delphi-Quellcode:

			type

  TJob = string;

  TJobList = Class

  Private

    fList : TThreadList;

  Public

    Function GetNextJob : TJob;

    Procedure AddJob (aJob : TJob);

    End;

  TMainForm = class(TForm)

    btnStartThreads: TButton;

    ...

implementation

{$R *.dfm}

Function TJobList.GetNextJob : TJob;

Var

  l  :TList;

Begin

  l := fList.LockList;

  Try

    if l.Count = 0 Then

      Result := Nil

    else              <<<<< Fehler 1

    begin

      Result := l[l.count - 1];   <<<<< Fehler 2

      l.delete (l.count - 1);

      End;

  Finally

     fList.unlockList;

     End;

End;

Procedure TJobList.AddJob (aJob : TJob);

Var

  l  :TList;

Begin

  l := fList.LockList;

  Try

    l.Add (aJob);  <<<<< Fehler 3

  Finally

    fList.unlockList;

    End;

End;

Auch ist mir nicht ganz klar wie ich die
von dir angesprochenen Get und Set-Methoden
im Thread aufbauen soll.

Tut mir ehrlich leid wenn ich mich dabei ein
wenig dumm anstelle.

**alzaimar**

Dein TJob ist ein String, meine Threadlist ist eine Liste von Pointern.... Was kann man da machen? Wir bauen uns eine threadsichere Stringliste.

zusammenfalten · markieren

Delphi-Quellcode:

			uses SyncObjs;

Type

  TJobList = Class

  Private

    fList : TStringlist;

    fCS : TCriticalSecttion;

  Public

    Constructor Create;

    Destructor Destroy; Override;

    Procedure AddJob (aJob : TJob);

    Function GetNextJob : TJob;

    Function JobsAvailable : Boolean;

    End;

Constructor TJobList.Create;

Begin

  fCS := TCriticalSection.Create;

  fList := TStringlist.Create;

End;

Destructor Destroy;

Begin

  fCS.Free;

  fList.Free;

  inherited;

End;

Procedure TJobList.AddJob (aJob : TJob);

Begin

  fCS.Enter;

  Try

    fList.Add (aJob);

  Finally

    fCS.Leave;

    End

End;

Function TJobList.GetNextJob : TJob;

Begin

  fCS.Enter;

  Try

    Result := fList[0];

    fList.Delete (0);

  Finally

    fCS.Leave;

    End

End;

Function TJobList.JobsAvailable : Boolean;

Begin

  fCS.Enter;

  Try

    Result := (fList.Count > 0 );

  Finally

    fCS.Leave;

    End

End;

Wenn Du hier unsicher bist, dann fang doch mit einem Thread erstmal an. Der soll schön im Hintergrund die Jobs nacheinander abarbeiten. Wenn das klappt, dann versuche es mit 2 Threads usw.

**endeffects**

vielen dank für deine geduld, ich hab dein beispiel nun wie folgt in
die main-unit eingebunden, könntest du dir vielleicht kurz anschauen
ob das alles so korrekt ist?

zusammenfalten · markieren

Delphi-Quellcode:

			unit Unit1;

interface

uses

  Windows, Messages, SysUtils, Classes, Controls, Forms, StdCtrls, ComCtrls,

  Unit2, Dialogs, FastStrings, SyncObjs;

type

  TJob = String;

  TJobList = Class

  Private

    fList : TStringlist;

    fCS : TCriticalSection;

  Public

    Constructor Create;

    Destructor Destroy; Override;

    Procedure AddJob (aJob : TJob);

    Function GetNextJob : TJob;

    Function JobsAvailable : Boolean;

    End;

  TMainForm = class(TForm)

    btnStartThreads: TButton;

    ....

    end;

  public

    { Public-Deklarationen }

    pool: TJobList;

var

  MainForm: TMainForm;

implementation

{$R *.dfm}

Constructor TJobList.Create;

Begin

  fCS := TCriticalSection.Create;

  fList := TStringlist.Create;

End;

Destructor TJobList.Destroy;

Begin

  fCS.Free;

  fList.Free;

  inherited;

End;

Procedure TJobList.AddJob (aJob : TJob);

Begin

  fCS.Enter;

  Try

    fList.Add (aJob);

  Finally

    fCS.Leave;

    End

End;

Function TJobList.GetNextJob : TJob;

Begin

  fCS.Enter;

  Try

    Result := fList[0];

    fList.Delete (0);

  Finally

    fCS.Leave; 

    End

End; 

Function TJobList.JobsAvailable : Boolean; 

Begin

  fCS.Enter; 

  Try

    Result := (fList.Count > 0 ); 

  Finally

    fCS.Leave; 

    End

End;

procedure TMainForm.btnStartThreadsClick(Sender: TObject);

....

end;

procedure TMainForm.FormCreate(Sender: TObject);

begin

  Pool:= TJobList.Create;

  Pool.AddJob('test');

  Pool.AddJob('test2');

....

im thread rufe ich dann Job:= MainForm.Pool.GetNextJob; auf,
das klappt alles wunderbar, ich hoffe ich habe soweit alles
richtig eingebunden, GANZ großen dank für deine hilfe

mfg

**endeffects**

ich hab den code oben noch einmal kurz ergänzt, ich hoffe es ist alles richtig

**alzaimar**

Richtig ist, wenn es läuft.

Denk einfach immer dran, bei Thread-Programmierung die gemeinsamen resourcen (Variablen, Klassen etc.) so zu schützen. Desweiteren beschäftige dich mal mit der Synchronisation von Threads (Semaphoren, Mutex, Events).
Bei Dir sollten die Threads einschlafen, sobald die Jobliste leer ist, und wieder aufwachen, sobald die Jobliste gefüllt wird. Weiterhin ist es überflüssig, für N Jobs M Threads (M>N) aufzuwecken. Optimal wäre es, maximal X Threads gleichzeitig zu starten, wobei X bei ca. 15 anzusiedeln ist. Irgendwann habe ich irgendwo mal gelesen das so 15 Threads die praktikable Obergrenze darstellt, weil der Prozess dann nur noch mit dem Umherschalten von Thread-Rechnerzeit beschäftigt ist. Ob das noch so stimmt, weiss ich nicht, kannst ja mal experimentieren.

**endeffects**

nach einigen umbauarbeiten meines alten codes scheint nun alles
wunderbar zu funktionieren - vielen dank für deine hilfe

eine frage habe ich allerdings noch, zur zeit lass ich jeden thread
in einer endlosschleife laufen die als erstes prüft ob noch etwas
zu tun ist und sollte das nicht der fall sein den thread mit sleep
ein paar sekunden einfriert bis die schleife dann erneut durchlaufen wird,
ähnliches veranstalte ich zur zeit im mainthread, hier prüft ein timer
alle paar sekunden ob die jobliste gefüllt werden muss, eine wenig elegante
lösung wie ich finde, hast du vielleicht noch eine anregung wie man das
verbessern könnte?

**alzaimar**

Natürlich habe ich eine andere Idee. Die müsste auch 'besser' sein.
Hintergrund sind 'Semaphoren'. Eigentlich ist so eine Semaphore nix anderes als ein threadsicherer Zähler mit dem man Threads steuert.
Es gibt 4 Windows-API Prozeduren, die hier von Bedeutung sind:
CreateSemaphore erzeugt eine Semaphore mit einem definierten Anfangszustand
ReleaseSemaphore erhöht den Wert um eins,
WaitForSingleObject wartet, bis die Semaphore <>0 ist (und verringert sie dann sofort um 1) und
CloseHandle gibt das Handle wieder an Windows zurück.

Die Idee ist folgende:
Am Anfang ist die Semaphore = 0. Wenn ein Job per AddJob in die Liste kommt, wird die Semphore um eins erhöht.
So weit, so gut. Alle Threads warten nun, bis die Semaphore <> 0 ist und verringern die Semaphore um 1, wenn das der Fall ist. Das ist wichtig. Angenommen, zwei Threads warten auf die Semaphore. Genau 1 Job wird in die Jobliste gepackt, ergo ist die Semaphore=1. Aber: Nur ein Thread (welcher, wissen wir nicht), wird die Nachricht bekommen, das die Semaphore <>0 ist. Der andere Thread kriegt davon nix mit!

Du musst eigentlich nur Folgendes machen:
1. Die TJobList bekommt eine Property "Semaphore : THandle read fHandle".
2. TJobList.Create erzeugt eine Semaphore mit "fHandle := CreateSemaphore (nil,0,32767,nil)". Ich denke, 32767 offene Jobs sollten reichen.
3. TJobList.GetNextJob liefert '' zurück, wenn die Liste leer ist (hatte ich vergessen).
4. TJobList.Destroy gibt das fHandle mit "CloseHandle (fHandle)" wieder frei.
5. Die Execute-Methode der Threads sieht nun so aus:

markieren

Delphi-Quellcode:

			Procedure TMyThread.Execute;

Begin

  While Not Terminated do

    If WaitForSingleObject (aJobList.Semaphore, ccMyThreadTimeout) Then Begin

      aJob := aJobList.GetNextJob;

      if aJob<>'' Then 

        DoExecuteJob (aJob);

      End

End;

ccMyThreadTimeout setzt Du so auf 500 Millisecs. Du kannst auch INFINITE hinschreiben, dann warten sie wirklich *BIS* was zu tun ist, leider bekommst du dann Probleme, sie zu terminieren. Deshalb also dieses Timeout.

Prinzipiell kein Unterschied zu Deiner 'Sleep' Geschichte. WaitForSingleObjects ist aber laut Microsoft der bessere Weg, weil die CPU-Belastung minimal wird. Du kannst das Thread-timeout hochsetzen, wenn dein Programm immer läuft.

**** Nach einer Nacht drüber schlafen, bin ich heute auch etwas schlauer. Natürlich kann man die Threads endlos warten lassen. Dazu muss beim Programmende nur die Semaphore sehr hoch gezählt werden (so hoch wie es threads gibt). Schau Dir die FormDestroy-Methode einfach an...

Bei Deiner Idee, im Hauptprogramm immer mal wieder zu schauen, ob was zu tun ist (um dann die Jobliste zu füllen), kann ich Dir nicht speziell helfen, weil ich nicht weiss, was da genau los ist. Aber, prinzipiell würde ich mich an Deiner Stelle mit den Sempahoren beschäftigen.

Hier ist mein Beispiel-Projekt.

Probleme mit Threads und Aufgaben-Pool

Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Re: Probleme mit Threads und Aufgaben-Pool

Forumregeln

endeffects Registriert seit: 27. Jun 2004 450 Beiträge	#6 Re: Probleme mit Threads und Aufgaben-Pool 17. Mai 2005, 21:24 ich hab den code oben noch einmal kurz ergänzt, ich hoffe es ist alles richtig
	Zitat