schnelle Server Client Verbindung ohne Verluste

[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?

[TomyN]

Evtl. 32k Buffersize?

[jaenicke]

Zitat von AJ_Oldendorf:

Durch das Sleep(1) funktioniert es auch (dass die CPU Auslastung nicht ansteigt), wenn kein Empfang mehr stattfindet.

Besser ist es, wenn du das Sleep nur ausführst, wenn keine Daten vorhanden waren. Dadurch verzögerst du den Empfang nicht, wenn Daten kommen.

Delphi-Quellcode:

procedure TMyTCPServer.OnExecuteHandler(AContext: TIdContext);
var
  Buffer: TIdBytes;
begin
  if AContext.Connection.IOHandler.InputBuffer.Size > 0 then
  begin
    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
  else
    Sleep(1);
end;

Zitat von AJ_Oldendorf:

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?

Darauf hast du keinen direkten Einfluss. Die Pakete werden für den Transport nach Bedarf unterteilt und beim Empfang in den Puffer geschrieben. Man kann dort zwar auch Einstellungen setzen, aber das garantiert dennoch nicht ein bestimmtes Verhalten.

[AJ_Oldendorf]

@jaenicke: Danke für den Hinweis mit dem Sleep.
Beim Empfang muss ich sowieso die Telegramme wieder entsprechend zusammensetzen. Wollte es nur verstanden haben.

EDIT:
Das mit dem Sleep in deiner Variante funktioniert leider nicht. Da geht die CPU Auslastung auch auf 7% hoch nachdem der Empfang fertig ist

[jaenicke]

Ach, Entschuldigung, das ist ja auch verkehrt herum. Ich habe es korrigiert. Die Beschreibung war korrekt, der Quelltext nicht. Das Sleep muss ins Else.