schnelle Server Client Verbindung ohne Verluste

[mjustin]

Zitat von AJ_Oldendorf:

Server schickt dem Client was, Client kann dem Server antworten. Das ist das Grundprinzip

Das ist ein Protokoll das nicht so ganz zu HTTP (siehe Vorschlag weiter vorher) passt. Der Client muss hier nun warten, bis Daten vom Server kommen, diese verarbeiten, und dann wieder warten. (und eventuell eine Antwort zurücksenden).

Dieses Kommunikationsmuster habe ich mit Indy mal in in einem kleinen Demoprogramm als 'umgekehrte request-response' (Anfrage/Antwort) umgesetzt.

https://mikejustin.wordpress.com/202...ocket-library/, Code ist auf https://github.com/michaelJustin/ind...equestResponse

Die Oberfläche erlaubt wahlweise, dass der Client eine Anfrage an den Server sendet, oder umgekehrt, der Server eine Anfrage an den Client sendet.

Es fehlt zwar etwas Fehlerbehandlung - zum Beispiel Prüfung auf ReadLnTimeout nach Readln - aber es zeigt, dass auch der Server die erste(n) Nachricht(en) senden kann, nachdem die Verbindung hergestellt ist.

Der clientseitige Code läuft in einem Thread und sein Betriebsmodus wird durch ein Flag gesteuert. Entweder wird ein Request gesendet und dann auf die Antwort gewartet, oder auf einen Request gewartet und eine Antwort gesendet.

Delphi-Quellcode:

while not Terminated do
begin
  if Send then
  begin
    // send request to server and receive response
    Send := False;
    Request := 'REQ:' + FClientRequestHolder.Text;
    TCPClient.IOHandler.WriteLn(Request);
    repeat
      Response := TCPClient.IOHandler.ReadLn(IndyTextEncoding_UTF8);
    until Response <> '';
    LogInMainThread(Request, Response);
  end
  else
  begin
    // receive request from server and send response
    Request := TCPClient.IOHandler.ReadLn(IndyTextEncoding_UTF8);
    if StartsStr('REQ:', Request) then
    begin
      Response := 'from client ' + ReverseString(Request);
      TCPClient.IOHandler.WriteLn(Response);
    end;
  end;
end;

An diesem Beispiel fällt die Abwesenheit von CheckForDataOnSource auf. Dieses wird nicht benötigt, da ReadLn solange wartet, bis eine Zeile aus dem Socket gelesen wurde, oder es zu einem Timeout kommt.

[Kas Ob.]

Hi,

@AJ_Oldendorf, there is so much can be written about this, yet don't know where to start, so i will start with what thoughts come to mind (logically ordered or not)
1) You said a lot of data between 20 and 80 packet, ( assuming packet and telegram are the same with in the translation !), see i wrote many servers and they are tested and running in real production, working with more than 400k packet per second saturating almost %50 of the 2.5Gbps connection, yet the CPU never goes above %6, some of traffic is done on MySQL, simple insert and delete but mostly update,
so can TCP solve you need ?, absolutely yes.

2) In older thread you tried to get the MTU and its 1500b effect on the speed, before you go that way, i suggest lets fix you problem by understanding it, which i will try below, but going after low level is white rabbit hole, as yes it is 1500 most the time, yet it will never be like that as this is isolated by the network adapter/driver, and every OS has many option/setting that will affect this behavior, yet it will stay 1500, but you user mode software will handle it differently, i suggest to not care about these deep and low level for now, as they are not your problem, forget about tweaking low level stuff for now, don't waste your time while you problem is in different place.

3) Indy is not slow or slower of any other, all other are exactly like that, the same, all of them will serve you at the same speed, the difference will manifested in aspects you will not need, not in this case/scenario, so no matter what you will use it will serve you well.

4) Once i saw ReadLn and WriteLn, i know for sure this can't perform s***, the server will send line by line and the client will read line by line, this is a huge bottleneck in the code.

5) I see this code

Delphi-Quellcode:

IdTCPClient1.IOHandler.CheckForDataOnSource(10);

if not IdTCPClient1.IOHandler.InputBufferIsEmpty then
begin
  InData := IdTCPClient1.IOHandler.ReadLn(' #~#* ' + EOL, 100, -1, IndyTextEncoding_UTF8);

  if InData <> ''  then
  begin
    //do something with it
  end ;
end ;

and i know for sure this can't be perform (like 4), mixing lines with process in place, you can do this only if you send data very rarely, here you combined most worse approaches in one loop.

What is happening here ? let me give you an example how these act as sticks in wheel:
1) Server send line after line, here Nagle (if enabled) will matter, but in not much as will see below.
2) Client notified to receive
3) Client Read One Single Line, in most and slowest possible way, that handle char by char.
4) Client proceed to process One Single Line.
5) When client extracted one line from the received buffer .. well this part is important, the extraction happened in a buffer in user space owned by you own process because you received it, it could be one line or more than one, it could be one or more MTU splitting many lines, well this fact is not important as the important part is what is happening on the network at this exact moment, when you application issued read line for the first time, read all what it could from the network using API like recv, Indy with ReadLn will only process and grab from the network new recv if there is no new lines in the already received and memory loaded buffers, so in other words and shorter way...
When you are handling line by line the network buffers could be filled from the peer (server here) and this will trigger stop sending ACK, so the server will stop sending lines/buffers, your application handle more and more lines and then request one buffer no matter how much in size, it will read from the network driver/provider..., triggering sending an ACK to server server will send one and wait new ACK, and you client server now locked in very slow recv-process-request, extremely slow.

Suggestions:
1) start with removing WriteLn/ReadLn with something useful and performant, in fact, any other method is faster.
2) Send the data in bulk as much as you can.
3) Read as much as you can on each and every read from the socket, provide a buffer lets say 16kb buffer perform read for 16kb or 64kb, even 8kb will be fast enough as it is default for most applications, the point is when read is needed read what you can and as much as you can, not as much you need.
4) Try ICS may be you will like it, there is an obsession with old and outdated practices like this case with readln/writeln, heck.. i was shocked that TFile is still used by many, it is slow as turtle, use modern practices.
5) I saw another code above like this one by mjustin

Delphi-Quellcode:

while not Terminated do
begin
  if Send then
  begin
    // send request to server and receive response
    Send := False;
    Request := ' REQ: ' + FClientRequestHolder.Text;
    TCPClient.IOHandler.WriteLn(Request);
    repeat
      Response := TCPClient.IOHandler.ReadLn(IndyTextEncoding_UTF8);
    until Response <> '';
    LogInMainThread(Request, Response);
  end
  else
  begin
    // receive request from server and send response
    Request := TCPClient.IOHandler.ReadLn(IndyTextEncoding_UTF8);
    if StartsStr(' REQ: ', Request) then
    begin
      Response := ' from client ' + ReverseString(Request);
      TCPClient.IOHandler.WriteLn(Response);
    end ;
  end ;
end ;

Will comment on that one line "LogInMainThread(Request, Response);" what do you think the network stream behavior will be while the application is synchronizing with main thread, i explained above, it trigger full stop send on server, depleted socket window for this socket and many others, lost the sending momentum that build up with TCP between peers(unlike UDP which start with maximum), this behavior you see when downloading huge files, the first few seconds the speed is low then gradually build up to saturate the allowed bandwidth, as the server start to send more and more without waiting for ACK.

Hope that helps.

[jaenicke]

Wenn es so schnell sein muss, sollten die Daten sofort an einen anderen Thread zur Verarbeitung gegeben werden und dort erst die Pakete extrahiert werden. So bekomme ich eine sehr geringe Latenz. Client:

Delphi-Quellcode:

program Project204;

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

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;
  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
    Writeln('Fehler: Nicht verbunden.');
end;

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

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

       for var i := 1 to 1000000 do
        MyClient.SendData(TestData);

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

Server:

Delphi-Quellcode:

program Project203;

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

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;
  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;
  inherited Create(False);
end;

procedure TProcessingThread.Execute;
var
  Data: TIdBytes;
begin
  while not Terminated do
  begin
    Data := FDataQueue.Dequeue;
    if Length(Data) > 0 then
    begin
      Writeln('Empfangen: ', Length(Data), ' 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, 8192);
  while AContext.Connection.IOHandler.InputBuffer.Size > 0 do
  begin
    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.

[AJ_Oldendorf]

Danke jaenicke und danke an Kas Ob.

Ich habe verstanden, dass TCPIP und Indy mein Grundprinzip locker abdecken können. Ich darf nur nicht mit ReadLn und WriteLn arbeiten. Ich werde versuchen, meine Software umzugestalten (auf getrenntes Senden und Empfangen in Threads) sowie ohne Terminator und TIdBytes sowie immer Maximum lesen. Ich werde berichten...

[mjustin]

Zitat von AJ_Oldendorf:

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.

[DenkDirNix]

Zitat von jaenicke:

Wenn es so schnell sein muss, sollten die Daten sofort an einen anderen Thread zur Verarbeitung gegeben werden und dort erst die Pakete extrahiert werden. So bekomme ich eine sehr geringe Latenz.

Ich habe Deinen Code mal ausprobiert, der Übersichtlichkeit halber sendet mein Client nur 10 Pakete à 1024 Bytes. Der Server empfängt aber nur einmal 8192 Bytes und bleibt dann hängen. Kann da ein Fehler im OnExecuteHandler des Servers vorliegen? Dies ist unabhängig von der Verwendung der Queue.

[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 von 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...

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.

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