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:

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:

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:

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:

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"

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.

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?

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 :dance:

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. :warn:
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:

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.