schnelle Server Client Verbindung ohne Verluste

[jaenicke]

Bei mir läuft das so 1:1 durch. Wenn das bei dir hängt, ist die Frage, wo der Server dann hängt. Da es bei mir nicht passiert, tappe ich da im Dunkeln, da müsstest du mal im Debugger schauen...

[DenkDirNix]

Zitat:

Zitat von jaenicke (Beitrag 1547668)

Bei mir läuft das so 1:1 durch. Wenn das bei dir hängt, ist die Frage, wo der Server dann hängt. Da es bei mir nicht passiert, tappe ich da im Dunkeln, da müsstest du mal im Debugger schauen...

Meine Änderung an Deinem Code ist wie gesagt nur die, dass der Client nur 10 mal 1024 Bytes schickt.

Anbei mal der Stack (kann man den aus der IDE auch direkt als Text speichern ???).
Im OnExecuteHandler werden im ersten Durchlauf 8192 von den vorliegenden 10240 Bytes abgeholt, der zweite Durchlauf für die restlichen 2048 Bytes führt dann zum gezeigten Stack so weit ich ihn verfolgt habe. Leider führt der Befehl "Pause" für einen hängenden Prozess ja nur zu einem leeren Stack.

Der etwas irritierende Parameterwert "-2" im ReadFromSource() ist nicht der Original-Aufrufparameter sondern der vom Code dann auf "IdTimeoutInfinite" geänderte.

[jaenicke]

Ach so, ja, das ist Absicht. Es ging ja um Geschwindigkeitstest, da habe ich den Fall, dass ein Paket oder eine Gruppe von Paketen (8192 Byte) nicht vollständig ankommt, nicht behandelt.

[MyRealName]

Ich habe früher viel RealThinClient genutzt, auch mittlerweile OpenSource. Ist sehr stabil, schnell und einfach zu implementieren

[Kas Ob.]

Zitat:

Zitat von MyRealName (Beitrag 1547700)

Ich habe früher viel RealThinClient genutzt, auch mittlerweile OpenSource. Ist sehr stabil, schnell und einfach zu implementieren

Good choice and great advice, as RTC is the best of them all !

I either do low level using API for extra tweaking or RTC for anything else, also notice with RTC, you don't need to use its components and classes from the palate, but can use the RAW ones for TCP and UDP, they are powerful and comes with threading (pool and sync... etc) out-of-the-box that can be switched from Multithreaded (asynchronous) to Single-thread (synchronous) by flipping a bool.

[mjustin]

Zitat:

Zitat von AJ_Oldendorf (Beitrag 1547641)

Ich darf nur nicht mit ReadLn und WriteLn arbeiten.

Mit Indy darf man auch ReadLn und WriteLn verwenden, ohne Performanceproblem.

Indy verwendet einen 32K (default) großen Buffer für das Empfangen von Daten. Man kann die Buffergröße bei Bedarf auch noch erhöhen (siehe TIdIOHandler.SendBufferSize property). Auch beim Senden wird ein 32k (default) großer Buffer verwendet.

Dass IOHandler.ReadLn / WriteLn schlechtere Performance hat, wäre schon vor Jahren Thema in den einschlägigen Delphi-Foren gewesen, wenn es denn wahr wäre.

[AJ_Oldendorf]

@jaenicke
Ich habe dein Code mal minimal angepasst gemäß deinem Beitrag hier:

https://www.delphipraxis.net/1547639-post18.html

Server:

Delphi-Quellcode:

program Server;

uses
  System.Classes,
  System.SysUtils,
  System.SyncObjs,
  IdTCPServer,
  IdContext,
  IdGlobal,
  System.Generics.Collections,
  System.Diagnostics;

type
  TDataQueue = class
  private
    FQueue: TQueue<TIdBytes>;
    FLock: TCriticalSection;
  public
    constructor Create;
    destructor Destroy; override;
    procedure Enqueue(const Data: TIdBytes);
    function Dequeue: TIdBytes;
  end;

  TProcessingThread = class(TThread)
  private
    FDataQueue: TDataQueue;

    Anz : Integer;
  protected
    procedure Execute; override;
  public
    constructor Create(ADataQueue: TDataQueue);
  end;

  TMyTCPServer = class
  private
    FServer: TIdTCPServer;
    FDataQueue: TDataQueue;
    FProcessingThread: TProcessingThread;
    procedure OnExecuteHandler(AContext: TIdContext);
  public
    constructor Create;
    destructor Destroy; override;
    procedure Start;
    procedure Stop;
  end;

{ TDataQueue }

constructor TDataQueue.Create;
begin
  FQueue := TQueue<TIdBytes>.Create;
  FLock := TCriticalSection.Create;
end;

destructor TDataQueue.Destroy;
begin
  FQueue.Free;
  FLock.Free;
  inherited;
end;

procedure TDataQueue.Enqueue(const Data: TIdBytes);
begin
  FLock.Acquire;
  try
    FQueue.Enqueue(Data);
  finally
    FLock.Release;
  end;
end;

function TDataQueue.Dequeue: TIdBytes;
begin
  FLock.Acquire;
  try
    if FQueue.Count > 0 then
      Result := FQueue.Dequeue
    else
      SetLength(Result, 0);
  finally
    FLock.Release;
  end;
end;

{ TProcessingThread }

constructor TProcessingThread.Create(ADataQueue: TDataQueue);
begin
  FDataQueue := ADataQueue;
  Anz := 0;
  inherited Create(False);
end;

procedure TProcessingThread.Execute;
var
  Data: TIdBytes;
  sw3 : TStopwatch;
  t3 : Int64;
begin
  while not Terminated do
  begin
    Data := FDataQueue.Dequeue;
    if Length(Data) > 0 then
    begin
      Inc(Anz, Length(Data));

      Writeln('Empfangen: ', Length(Data), ' Bytes' + '- Anz: ' + Anz.ToString);

      Writeln('Gesamtlänge: ' + Anz.ToString + ' Bytes');
    end
    else
      Sleep(1);
  end;
end;

{ TMyTCPServer }

constructor TMyTCPServer.Create;
begin
  FDataQueue := TDataQueue.Create;
  FProcessingThread := TProcessingThread.Create(FDataQueue);
  FServer := TIdTCPServer.Create(nil);
  FServer.DefaultPort := 5000;
  FServer.OnExecute := OnExecuteHandler;
end;

destructor TMyTCPServer.Destroy;
begin
  Stop;
  FreeAndNil(FServer);
  FreeAndNil(FProcessingThread);
  FreeAndNil(FDataQueue);
  inherited;
end;

procedure TMyTCPServer.OnExecuteHandler(AContext: TIdContext);
var
  Buffer: TIdBytes;
begin
  //SetLength(Buffer, 61000);  //<- nicht feste größe einlesen
  while AContext.Connection.IOHandler.InputBuffer.Size > 0 do
  begin
    SetLength(Buffer, AContext.Connection.IOHandler.InputBuffer.Size); //<- so viel einlesen wie im Buffer enthalten ist
    AContext.Connection.IOHandler.ReadBytes(Buffer, Length(Buffer), False);
    FDataQueue.Enqueue(Buffer);
  end;
end;

procedure TMyTCPServer.Start;
begin
  FServer.Active := True;
end;

procedure TMyTCPServer.Stop;
begin
  FServer.Active := False;
end;

var
  MyServer: TMyTCPServer;

begin
  try
    MyServer := TMyTCPServer.Create;
    MyServer.Start;
    Writeln('Server läuft auf Port 5000. Drücke Enter zum Beenden.');
    Readln;
    MyServer.Stop;
    FreeAndNil(MyServer);
  except
    on E: Exception do
      Writeln('Fehler: ', E.Message);
  end;
end.

Client:

Delphi-Quellcode:

program Client;

uses
  System.Classes,
  System.SysUtils,
  IdTCPClient,
  IdGlobal,
  System.Diagnostics;

type
  TMyTCPClient = class
  private
    FClient: TIdTCPClient;
  public
    constructor Create;
    destructor Destroy; override;
    procedure Connect(const AHost: string; APort: Integer);
    procedure Disconnect;
    procedure SendData(const Data: TIdBytes);
  end;

{ TMyTCPClient }

constructor TMyTCPClient.Create;
begin
  FClient := TIdTCPClient.Create(nil);
end;

destructor TMyTCPClient.Destroy;
begin
  Disconnect;
  FreeAndNil(FClient);
  inherited;
end;

procedure TMyTCPClient.Connect(const AHost: string; APort: Integer);
begin
  FClient.Host := AHost;
  FClient.Port := APort;
  FClient.ConnectTimeout := 5000; // 5 Sekunden Timeout
  FClient.ReadTimeout := 5000; // 5 Sekunden Timeout für Lesevorgänge
  FClient.Connect;
  //FClient.IOHandler.RecvBufferSize := 32768;
  //FClient.IOHandler.SendBufferSize := 32768;
  //FClient.IOHandler.RecvBufferSize := 61440;
  //FClient.IOHandler.SendBufferSize := 61440;
  Writeln('Verbunden mit ', AHost, ':', APort);
end;

procedure TMyTCPClient.Disconnect;
begin
  if FClient.Connected then
  begin
    FClient.Disconnect;
    Writeln('Verbindung getrennt.');
  end;
end;

procedure TMyTCPClient.SendData(const Data: TIdBytes);
begin
  if FClient.Connected then
  begin
    FClient.IOHandler.Write(Data);
    //Writeln(Now, ' Gesendet: ', Length(Data), ' Bytes');
  end
  else
  begin
    Writeln('Fehler: Nicht verbunden.');
  end;
end;

var
  MyClient: TMyTCPClient;
  TestData: TIdBytes;
  Anz : LongWord;
begin
  try
    MyClient := TMyTCPClient.Create;
    try
      MyClient.Connect('127.0.0.1', 5000);

      var sw3 := TStopwatch.StartNew;
      var t3 : Int64;

      SetLength(TestData, 61000); //1024
      FillChar(TestData[0], Length(TestData), 65);

      Anz := 0;

      for var i := 1 to 20 do
      begin
        Inc(Anz, Length(TestData));

        MyClient.SendData(TestData);
      end;

      t3 := sw3.ElapsedMilliseconds; //Zeitmessung stoppen
      Writeln('Zeitdauer: ' + t3.ToString + ' ms');

      Writeln('Gesamtlänge: ' + Anz.ToString + ' Bytes');

      Readln;
      MyClient.Disconnect;
    finally
      FreeAndNil(MyClient);
    end;
  except
    on E: Exception do
      Writeln('Fehler: ', E.Message);
  end;
end.

Folgendes Erscheinungsbild:

1) Client sendet mit 20 Telegrammen in Summe 1220000 Bytes, wird auch angezeigt
2) Server empfängt in 38 Telegrammen die 1220000 Bytes, wird auch angezeigt
3) CPU Auslastung vom Server steigt danach auf 5-7% ohne das weitere Daten empfangen werden oder vom Client geschickt werden

Beim OnExecuteHandler vom Server habe ich eine kleine Anpassung gemacht (siehe Kommentar).

Frage:
Woran liegt das mit der CPU Auslastung und wie bekommt man das gelöst?
Müsste der Server nicht auch 20 Telegramme á 61000 Bytes empfangen, anstatt 38 Stück?

[AJ_Oldendorf]

@jaenicke
Ich push einfach mal. Hast du dazu eine Lösung/Idee?

Es ist ja so, dass mein Client immer unterschiedlich lange Telegramme schickt.
Die können von 13 Byte bis 61000 Byte lang sein. Damit muss der Server klar kommen und ich dachte, ich kann beim Server dann entsprechend so viel Byte einlesen, wie der InputBuffer lang ist.

[Kas Ob.]

@AJ_Oldendorf i am no expert in Indy at all, as i hate the s*** out of it due to exception raising policy, but while waiting for someone to resolve this for you, let me put few thoughts

1) Make sure you are blocking on the socket means as long you existing the loop then it will return and this will raise the CPU usage up to full Thread/Core.
2) ReadLn is blocking capable and that why it used everywhere.
3) Guess what ?! ReadBytes also blocking capable too.

so try this

Delphi-Quellcode:

procedure TMyTCPServer.OnExecuteHandler(AContext: TIdContext);
var
  Buffer: TIdBytes;
begin
  if not AContext.Connection.Connected then
    Exit;

  AContext.Connection.IOHandler.ReadBytes(Buffer, -1, True); // block and wait ! , while True for append (best practice) in case there is leftover

  if Length(Buffer) > 0 then
    FDataQueue.Enqueue(Buffer);
  {
  //SetLength(Buffer, 61000); //<- nicht feste größe einlesen
  while not AContext.Connection.IOHandler.InputBufferIsEmpty do
  begin
    SetLength(Buffer, AContext.Connection.IOHandler.InputBuffer.Size); //<- so viel einlesen wie im Buffer enthalten ist
    AContext.Connection.IOHandler.ReadBytes(Buffer, Length(Buffer), False);
    FDataQueue.Enqueue(Buffer);
  end;}
end;

On side note this blocking might need timeout adjustment, so adjust AContext.Connection.IOHandler.ReadTimeout to something short, it could be 30 second or even shorter will be fine, this might affect how much concurrent connection you expect, if few then put it 1 second (timeout = 1000) and it will be find.

about the hate for Indy, it comes form where an exception will surprise you, well from almost every line/operation, so i think some try..except is due somewhere in your client and server, but again you need someone else to help with what i wrote (which could be not the optimal) and exception handling..... or
Try somethin else ICS has plenty of examples/samples and it is way more friendly with its events, Indy still valuable and will stay the most protocol stuffed library in pascal.

[AJ_Oldendorf]

ok, ich glaube es verstanden zu haben

Delphi-Quellcode:

while AContext.Connection.IOHandler.InputBuffer.Size > 0 do
begin
  SetLength(Buffer, AContext.Connection.IOHandler.InputBuffer.Size);
  AContext.Connection.IOHandler.ReadBytes(Buffer, Length(Buffer), False);

  FDataQueue.Enqueue(Buffer);
end;
Sleep(1);

Durch das Sleep(1) funktioniert es auch (dass die CPU Auslastung nicht ansteigt), wenn kein Empfang mehr stattfindet.
Nur ist mir nicht klar, warum der Client 20 Telegramme á 61000 Byte abschickt und der Server 38 Stück empfängt wobei die meisten 32768 Byte lang sind und am Ende ein kürzeres. Warum nicht auch 20 Telegramme?