Selbstgebauter Threadpool funktioniert nicht

[Neutral General]

Hi,

Habe mir mal gestern mal schnell ne Threadpool Klasse zusammengeschustert. Allerdings gibts da noch einige Probleme.

Delphi-Quellcode:

procedure TThreadPoolThread.GetJob;
begin
  if FPool.FJobs.Count > 0 then
    FJob := FPool.FJobs.Pop
  else
    FJob := nil;
end;

procedure TThreadPoolThread.Execute;
begin
  while not FDead do
  begin
    GetJob;
    if (Assigned(FOnWork)) and (Assigned(FJob)) then
      FOnWork(Self,FJob);
    sleep(1);
  end;
end;

Das ist so ungefähr mein Grundgerüst. Ich dachte mir aber schon ganz zu Anfang, dass ich GetJob wohl synchronisieren muss, weil es schlecht wäre, wenn mehrere Threads gleichzeitig auf die Job-Queue zugreifen wollen.
Bei Synchronize stürzt der Thread ab. Aber soweit ich herausgefunden habe, ist Synchronize auch nicht das richtige in diesem Fall.

Habe es dann mit ner CriticalSection probiert:

Delphi-Quellcode:

procedure TThreadPoolThread.Execute;
var cs: TCriticalSection;
begin
  cs := TCriticalSection.Create;
  while not FDead do
  begin
    cs.Enter;
      GetJob;
    cs.Leave;

    if (Assigned(FOnWork)) and (Assigned(FJob)) then
      FOnWork(Self,FJob);
    
    sleep(1);
  end;
end;

Das funktioniert soweit ich das mitbekommen habe pro Thread 1 mal und bei jedem weiteren mal bekomme ich ne AV.

Weiß jetzt gerade nicht wie ich das machen muss :mrgreen: FJobs ist übrigens eine TObjectQueue (uses contnrs).

Gruß
Neutral General

[Apollonius]

Was du gerade tust ist so, als ob jeder einzelne Mensch für die Toilette sein eigenes besetzt-Schild mitbringt und das dann privat auf besetzt stellt. Das kann nicht funktionieren, oder? Du brauchst für den ganzen Pool eine Critical Section. Außerdem ist dein Polling eine ganz schlechte Idee. Nimm eine Semaphore. Oder wenn du es ganz korrekt machen willst: Verwende IO Completion Ports, da ist das gesamte Management schon eingebaut (und die Anzahl der aktiven Threads automatisch auf die CPU-Anzahl optimiert) und deine Liste kannst du dir dann auch schenken.

[Neutral General]

Hi,

Wie verwendet man die IO Completion Ports APIs denn? Ich blicke da nicht ganz durch.

[Apollonius]

Eigentlich sind diese dafür gedacht, bei asynchronem IO Nachrichten über die Fertigstellung von Operationen zu bekommen. Du kannst aber auch manuell Nachrichten schicken. Die Benutzung ist simpel: Mit CreateIOCompletionPort einen Port erstellen, mit GetQueuedCompletionStatus in den Threadpool-Threads auf Nachrichten warten und von außen mit PostQueuedCompletionStatus Aufträge abschicken. Die Auftragsliste wird automatisch verwaltet, dabei hast du 12 Bytes Speicher pro Auftrag zur Verfügung, das sollte reichen, ohne dass du den Speichermanager bemühen musst.
Das geniale dabei ist, dass wie schon gesagt automatisch die Anzahl der laufenden Threads an die CPU-Zahl angepasst wird - bekanntermaßen sollten diese Zahlen gleich sein, damit weder eine CPU brach liegt noch der Scheduler unnötig belastet wird. Wenn nun ein Threadpool-Thread mit WaitForSingleObject o.Ä. auf ein Event wartet, wird der Completion Port automatisch informiert und ein weiterer Thread wird aus GetQueuedCompletionStatus freigelassen. So ein gutes Scheduling ist sonst absolut unmöglich, weil du nicht merkst, wenn ein Thread anfängt zu warten.

[Neutral General]

Hi,

Habs jetzt folgendermaßen aufgebaut:

Delphi-Quellcode:

procedure TThreadPool.AddJob(AJob: TObject);
begin
  PostQueuedCompletionStatus(FIOComPort,SizeOf(TObject),Cardinal(AJob),nil);
end;

procedure TThreadPoolThread.GetJob;
var bytes: Cardinal;
    over: POverlapped;
    ajob: Cardinal;
begin
  if GetQueuedCompletionStatus(FIOComPort,bytes,AJob,over,0) then
    FJob := TObject(AJob)
  else
    FJob := nil;
end;

procedure TThreadPoolThread.Execute;
begin
  while not FDead do
  begin
    GetJob;
    if (Assigned(FOnWork)) and (Assigned(FJob)) then
      FOnWork(Self,FJob);
    sleep(1);
  end;
end;

Das klappt jetzt prinzipiell schon... Problem ist nur, dass es mehrere Jobs gibt, die ein und diesselbe Objektinstanz repräsentieren, die aber mehr oder weniger gleichzeitig von meinen Threads im Threadpool bearbeitet werden. Und dann kracht es -.-

Wenn du noch irgendwelche Infos dazu brauchst, dann sag mir ruhig Bescheid.

Danke auf jeden Fall für die Hilfe bisher ;)

[Apollonius]

Hör doch bitte mit dem Polling auf. Setze in GetQueuedCompletionStatus INFINITE als letzten Parameter und lass das Sleep(1) weg. Sonst verbrätst du vollkommen unnötig CPU-Zeit. Dann musst du allerdings eine spezielle Stopp-Nachricht einführen, wofür du eines der ungenutzten Informations-Felder verwenden kannst. Ferner steht auch dwNumberOfBytesTransferred zu deiner freien Verfügung, du musst also nicht SizeOf(TObject) dort einsetzen.

Dein Problem, dass mehrere gleiche Jobs gleichzeitig abgearbeitet werden, ist eigentlich keines. Der Threadpool sollte sich um solche Aufgaben nicht kümmern. Wenn die Jobs nicht gleichzeitig abgearbeitet werden können, muss das der Code regeln, der den Threadpool nutzt, nicht der Threadpool selbst. Am besten erledigst du das wohl, indem der Job sich selbst noch einmal in die Liste einfügt, wenn er abgearbeitet wurde.

[Neutral General]

Hi,

Also ich bins nochmal. Hätte jetzt theoretisch ne Idee wie ich mein Problem lösen könnte aber ich glaube es wäre nicht der richtige Weg.

Ich habe das Gefühl, dass ich die Sache mit den IO Completion Ports nicht ganz verstanden habe. Für mich sind das irgendwie nur von Windows verwaltete Warteschlangen.
Wofür dieser letzte Parameter? Anzahl der Threads. Was geschieht denn da im Hintergrund?

[Apollonius]

IO Completion Ports sind Warteschlangen, genau. Der letzte Parameter steuert genau das spezielle Scheduling, was ich weiter oben bereits erläutert habe:

Zitat:

Das geniale dabei ist, dass wie schon gesagt automatisch die Anzahl der laufenden Threads an die CPU-Zahl angepasst wird - bekanntermaßen sollten diese Zahlen gleich sein, damit weder eine CPU brach liegt noch der Scheduler unnötig belastet wird. Wenn nun ein Threadpool-Thread mit WaitForSingleObject o.Ä. auf ein Event wartet, wird der Completion Port automatisch informiert und ein weiterer Thread wird aus GetQueuedCompletionStatus freigelassen. So ein gutes Scheduling ist sonst absolut unmöglich, weil du nicht merkst, wenn ein Thread anfängt zu warten.

Ich denke, dass du für den letzten Parameter Null einsetzen solltest, sodass die Anzahl der laufenden Threads gleich der Anzahl der CPUs ist.
Könntest du dein Problem etwas genauer erläutern?

[Neutral General]

Also es geht um ein Serverprogramm.

Ich habe von TSocketServer (uses Sockets) abgeleitet und möchte jetzt die von den einzelnen Clients einkommenden Daten asynchron verarbeiten. Dafür soll das ServerSocket mit einem Threadpool ausgestattet werden.
Es wird laufend geprüft ob Daten von den einzelnen Clients verfügbar ist. Wenn das der Fall ist, wird das ClientSocket in eine Warteschlange eingereiht und der Threadpool empfängt und verarbeitet diese Daten dann.

Soweit mein Plan. Dafür hatte/wollte ich zuerst meine eigene Threadpool Klasse. Jetzt die Threadpool Klasse in Verbindung mit den IO Completion Ports. Aber ich glaube ich bin da aufgrund eines oder mehrerer Misverständnisse aufm Holzweg.

[Apollonius]

Zunächst einmal würde ich nur einen Threadpool instanziieren, nicht einen für jeden Socket. Dazu solltest du dann auch den Threadpool etwas verändern, sodass du nicht mehr ein OnWork-Ereignis hast, sondern einfach als Job einen Methodenzeiger übergibst (evtl. mit einem weitern Kontext-Parameter), der dann ausgeführt wird. Mit dem zusätzlichen Parameter nutzt du dann auch die 12 Byte voll aus.

Zum zweiten würde ich für jeden ClientSocket ein Flag Queued einführen. Dieses setzt du auf True, wenn du ihn in die Warteschlange stellst, und am Ende der Abarbeitungsroutine wieder auf False. So vermeidest du, dass ein Socket mehrfach in die Warteschlange gesetzt wird. Könntest du noch etwas genauer sagen, wo genau du die Daten abrufst und wie du auf neue Daten prüfst?

[Neutral General]

Zitat:

Zitat von Apollonius

Zunächst einmal würde ich nur einen Threadpool instanziieren, nicht einen für jeden Socket.

Das tue ich auch nicht ;)

Zitat:

Zitat von Apollonius

Dazu solltest du dann auch den Threadpool etwas verändern, sodass du nicht mehr ein OnWork-Ereignis hast, sondern einfach als Job einen Methodenzeiger übergibst (evtl. mit einem weitern Kontext-Parameter), der dann ausgeführt wird. Mit dem zusätzlichen Parameter nutzt du dann auch die 12 Byte voll aus.

In diesem einen OnWork Event werden alle Clientdaten bearbeitet.

Zitat:

Zitat von Apollonius

Zum zweiten würde ich für jeden ClientSocket ein Flag Queued einführen. Dieses setzt du auf True, wenn du ihn in die Warteschlange stellst, und am Ende der Abarbeitungsroutine wieder auf False. So vermeidest du, dass ein Socket mehrfach in die Warteschlange gesetzt wird.

Ja das hatte ich vor, nachdem ich gemerkt habe, dass ich ansonste Probleme bekomme. Habe es aber noch nicht eingebaut.

Zitat:

Zitat von Apollonius

Könntest du noch etwas genauer sagen, wo genau du die Daten abrufst und wie du auf neue Daten prüfst?

Kann dir Quellcode geben. Die ganzen Sockets sind noch alle in Entwicklung und einige Dinge sind nicht so Ideal evtl:

Delphi-Quellcode:

// Wird im Mainthread aufgerufen
procedure TServerSocket.Listen;
var id: Cardinal;
begin
  FClosed := false;
  FSocket.Open;
  BeginThread(nil,0,@TServerSocket_CheckData,Self,0,id);
  BeginThread(nil,0,@TServerSocket_Listen,Self,0,id);
end;

procedure TServerSocket_CheckData(Server: TServerSocket);
var i: Integer;
begin
  with Server do
  begin
    while not FClosed do
    begin
      for i := FClients.Count-1 downto 0 do
        if FClients[i].DataAvailable then
          FThreadPool.AddJob(FClients[i]);
      sleep(1);  
    end;
  end;
end;

// Der Threadpool

procedure TThreadPool.AddJob(AJob: TObject);
begin
  PostQueuedCompletionStatus(FIOComPort,SizeOf(TObject),Cardinal(AJob),nil);
end;

procedure TThreadPoolThread.GetJob;
var bytes: Cardinal;
    over: POverlapped;
    AJob: Cardinal;
begin
  if GetQueuedCompletionStatus(FIOComPort,bytes,AJob,over,0) then
    FJob := TObject(AJob)
  else
    FJob := nil;
end;

procedure TThreadPoolThread.Execute;
begin
  while not FDead do
  begin
    GetJob;
    if (Assigned(FOnWork)) and (Assigned(FJob)) then
      FOnWork(Self,FJob);
    sleep(1);
  end;
end;

Sollte das nicht reichen, dann kann ich notfalls auch die 2 Units anhängen..

[Apollonius]

Häng bitte mal die Units an. So fehlt doch noch ziemlich viel Relevantes.

Warum bist du so ein Freund von Polling? Zumindest beim Threadpool ist eventbasiertes Arbeiten Pflicht. Bei den Sockets sehe ich auch keinen Grund dagegen.

[Neutral General]

Hi,

Bin kein Freund von Polling :mrgreen: Aber ich weiß mir grad nicht anders zu helfen :?

[Apollonius]

Delphi-Quellcode:

procedure TThreadPool.AddJob(AJob: TObject);
begin
  PostQueuedCompletionStatus(FIOComPort,1,Cardinal(AJob),nil);
end;

procedure TThreadPool.Dead;
var i: Integer;
begin
  for i:=1 to FThreads.Count do
    PostQueuedCompletionStatus(FIOComPort,0,0,nil); //"Beenden"-Nachricht schicken
end;

procedure TThreadPoolThread.GetJob;
var bytes: Cardinal;
    over: POverlapped;
    AJob: Cardinal;
begin
  if not GetQueuedCompletionStatus(FIOComPort,bytes,AJob,over,INFINITE) then //!!
    RaiseLastOsError;
  if bytes = 0 then
    Dead
  else
    FJob := TObject(AJob);
end;

procedure TThreadPoolThread.Execute;
begin
  while True do
  begin
    GetJob;
    if FDead then
      Exit;
    if (Assigned(FOnWork)) and (Assigned(FJob)) then
      FOnWork(Self,FJob);
  end;
end;

So pollt wenigstens der Threadpool nicht mehr.

Warum bist du so masochistisch, dich mit der Unit Sockets anzulegen? :mrgreen: Da bist du doch schneller, indem du direkt die Windows-Funktionen benutzt. Schau dir mal die Unit ScktComp an: Da wird gezeigt, wie man ohne zusätzliche Threads über die Winsockets kommunizieren kann. Da gibt es nämlich auch Möglichkeiten, ereignisgesteuert zu arbeiten.

[Neutral General]

Hi,

Danke schonmal für deine Verbesserungsvorschläge. Ich schau mir das gleich mal an.

Wie ich gesehen hab, arbeiten die ScktComp Sockets mit WindowMessages.
Habe vorher die ICS von Overbyte benutzt, die auch mit Messages gearbeitet hat und kam zu dem Schluss, dass "messagebasierte" Sockets für mein Vorhaben ungeeignet sind. Habe Testweise 200 Sockets auf meinen Server losgelassen (damals noch ICS). Wenn alle diese Sockets dem Server alle paar Millisekunden Daten geschickt hat und dann noch neue Clientverbindungen reinkamen, kamen von 200 neuen Verbindungen nur ca. 10 beim Server an.
Ich gehe davon aus, dass es daran lag, dass Windows nichtmehr hinterhergekommen ist mit den Messages (o.ä.).
Daher habe ich alternative Komponenten gesucht. Jedoch keine geeigneten gefunden und habe mich dann schließlich entschlossen mir eigene Sockets auf Basis der Sockets.pas zu bauen. Und bevor ich meinen Threadpool eingebaut hatte, war meine Socketklasse auch die für meinen Fall am besten funktionierendste.

[Apollonius]

Das ist aber kein Grund zu pollen. Es gibt bei den Winsockets zwei Möglichkeiten, ereignisgesteuert zu arbeiten, nämlich mit WSAAsyncSelect und WSAEventSelect. Wenn dabei bei einem Socket etwas anliegt, wird eine Nachricht geschickt bzw. ein Event gesetzt. Wenn dir die Methode mit den Nachrichten nicht schnell genug ist, erstelle dir für jedes Socket ein Event und verwende WSAEventSelect. Wenn du dann allerdings mehr als 64 Sockets hast, brauchst du einen weiteren Thread. Wenn einer dieser Threads dann feststellt, dass ein Ereignis eingetroffen ist, kann er den entsprechenden Job an den Threadpool weiterreichen. Das sollte nicht langsamer sein als die Methode mit dem Polling, aber bei geringer Last deutlich weniger CPU-Zeit verbrennen.

[Neutral General]

Hi,

Ich kenne mich mit der SocketAPI nicht aus, aber ich werde mir das mal anschauen.
Aber du meintest, dass ich für alle 64 Clients nen eigenen Thread brauche (wenn ich dich nicht falsch verstanden habe).
Voraussichtlich werde ich bis zu 2000 Clients haben. Das wären 32 Threads (+ X für andere Aufgaben). Liegt das noch in einem akzeptablen Rahmen?

[sirius]

Zitat:

Zitat von Neutral General

Hi,

Ich kenne mich mit der SocketAPI nicht aus, aber ich werde mir das mal anschauen.
Aber du meintest, dass ich für alle 64 Clients nen eigenen Thread brauche (wenn ich dich nicht falsch verstanden habe).
Voraussichtlich werde ich bis zu 2000 Clients haben. Das wären 32 Threads (+ X für andere Aufgaben). Liegt das noch in einem akzeptablen Rahmen?

Wenn die Threads nicht alle gleichzeitig arbeiten, sondern die meiste Zeit schlafen und wenn du auf die Stackgröße der Threads achtest, ja.
Aber (ohne dein Problem näher zu kennen), du könntest auch mal über mehrere Prozesse nachdenken.

[Apollonius]

Möglicherweise genügt es, wenn du einen eigenen Thread mit einer Nachrichtenschleife erstellst, ihm ein Fenster spendierst und bei diesem dann mit WSAAsyncSelect alle Sockets registrierst, das ist dann einfacher als mit den Events. Aber ganz ehrlich: Ich habe keine Ahnung, wie sich Windows bei so einer Belastung verhält. Versuche es.

[Neutral General]

Naja ich befürchte zwar, dass ich nur belächelt werde, aber es geht mir (der ein oder andere erinnert sich vielleicht noch) um mein Online-RPG. Lange Zeit hat das Projekt geschlafen und im Dezember habe ich es nochmal von vorne begonnen.

Zur Zeit baue ich erstmal ein paar Serverprototypen und teste verschiedene Dinge, weil ich denke, dass eine funktionierende, gute asynchrone Server<->Client Kommunikation eines der wichtigsten aber auch schwersten Dinge werden wird.

Habe es wie gesagt schon mit verschiedenen Serverstrukturen bzw Komponenten versucht. Entweder waren die Komponenten vom Aufbau her nicht für meine Pläne geeignet oder sie waren wie z.B. die ICS zu langsam, bzw. bei vielen Anfragen zu unzuverlässig.

Dann begann ich damit mir auf Basis der Sockets eigene Socketklassen zu schreiben. Und an diesem Punkt bin ich eben jetzt gerade.
Die Sockets, bzw mein Server soll eben mit bis zu 2000 Clients zurecht kommen. Das habe ich mir einfach mal als Ziel gesetzt. Wenn es ein paar weniger werden, dann ist das auch nicht tragisch. Aber es soll eben gut und zuverlässig funktionieren.

[Neutral General]

Hi,

Ich melde mich grad nochmal. Also habe gerade ein kleines Problem. Die function WSAEventSelect existiert nicht in der WinSock.pas des TDEs. Muss ich die selbst importieren oder bin ich blind?

[sirius]

ich würde auf unit winsock tippen.
Edit: :oops: