speed exchanges data between TIdTcpServer and TIdTCPClient (like a flood) how to

Question

I have a simple TidTCPServer Working on a console and accepting Data. My problem is when the client Send Stream but having a very high of speed exchange data, The server freeze after 70 lines and the CPU load of the server go to 70%; I don't know how can i resolve without adding a sleep between every send . below an example of Client and Server . Can you help me to resolve this (Server Side) thanks .

program Srv;

{$I Synopse.inc}

uses
  {$IFDEF UNIX}{$IFDEF UseCThreads}
  cthreads,
  {$ENDIF}{$ENDIF}
  Classes, SysUtils, CustApp, Generics.Collections, IdTCPServer, IdCustomTCPServer, IdContext, IdGlobal, Db, mORMot, mORMotSQLite3, IdSync, functions, SynCommons, SynSQLite3Static;

type

  { TMyApplication }
  TMyApplication = class(TCustomApplication)

   var IdTCPServer: TIdTCPServer;

   protected
    procedure DoRun; override;
    procedure ServerOnConnect(AContext: TIdContext);
    procedure ServerOnExecute(AContext: TIdContext);
    function ReceiveStream(AContext: TIdContext;Size:integer; var AStream: TStream);

  public
    constructor Create(TheOwner: TComponent); override;
    destructor Destroy; override;
  end;

  type
    TLog = class(TIdNotify)
    protected
      FMsg: string;
      procedure DoNotify; override;
    public
      class procedure LogMsg(const AMsg: string);
    end;

{ TMyApplication }

    procedure TLog.DoNotify;
    var i:integer;
    begin
     writeln(FMsg);
    end;

    class procedure TLog.LogMsg(const AMsg: string);
    begin
      with TLog.Create do
      try
        FMsg := AMsg;
        Notify;
      except
        Free;
        raise;
      end;
    end;

function TMyApplication.ReceiveStream(AContext: TIdContext; var AStream: TStream)
  : Boolean; overload;
var
  LSize: LongInt;
begin
  Result := True;
  try
    LSize := AContext.Connection.IOHandler.ReadLongInt();
    AContext.Connection.IOHandler.ReadStream(AStream,LSize, False)
    AStream.Seek(0,soFromBeginning);
  except
    Result := False;
  end;
end;      

procedure TMyApplication.ServerOnExecute(AContext: TIdContext);
var AStream:TMemoryStream;
begin

if (Acontext.Connection.IOHandler.InputBufferIsEmpty) then
  begin
    TLog.LogMsg('--: '+random(100000).ToString); //After executing Client this line is displayed 70 time and CPU load is from 40 % to 70%
    AStream:=TMemoryStream.Create;
    try 
      ReceiveStream(AContext,TStream(AStream)); 
      // .. here we use AStream to execute some stuff  
    finally
      Astream.free;
    end;        
end;

procedure TMyApplication.ServerOnConnect(AContext: TIdContext);
begin
 TLog.LogMsg('connect');        
end;

procedure TMyApplication.DoRun;
begin

    IdTCPServer := tIdTCPServer.Create;
    IdTCPServer.ListenQueue := 15;
    IdTCPServer.MaxConnections := 0;
    IdTCPServer.TerminateWaitTime := 5000;
    with IdTCPServer.Bindings.Add
    do begin
      IP   := '0.0.0.0';
      Port := 80;
      IPVersion:=Id_IPv4;
    end;
    IdTCPServer.OnConnect := ServerOnConnect;
    IdTCPServer.OnDisconnect := ServerOnDiconnect;
    IdTCPServer.OnExecute := ServerOnExecute;
    IdTCPServer.Active := True;

  while true do
   begin
    Classes.CheckSynchronize() ;
    sleep(10);
   end;

  readln;        

  Terminate;
end;

constructor TMyApplication.Create(TheOwner: TComponent);
begin

  inherited Create(TheOwner);
  StopOnException := True;

end;


destructor TMyApplication.Destroy;
begin
  IdTCPServer.Free;
  inherited Destroy;

end;

var
  Application: TMyApplication;
begin
  Application := TMyApplication.Create(nil);
  Application.Title := 'My Application';
  Application.Run;
  Application.Free;
end.

Client

function TForm1.SendStream(AClient: TIdTCPClient; AStream: TStream): Boolean; overload;
var
  StreamSize: LongInt;
begin
  try
    Result := True;
    try
      AStream.Seek(0,soFromBeginning);
      StreamSize := (AStream.Size);
      AClient.IOHandler.Write(LongInt(StreamSize));
      AClient.IOHandler.WriteBufferOpen;
      AClient.IOHandler.Write(AStream, 0, False);
      AClient.IOHandler.WriteBufferFlush;
    finally
      AClient.IOHandler.WriteBufferClose;
    end;
  except
    Result := False;
  end;
end;   
    
procedure TForm1.Button1Click(Sender: TObject);
var 
  Packet:TPacket;
  AStream:TMemoryStream;
begin
for i:=0 to 1000 do 
  begin
    Application.ProcessMessages;
    With Packet do
                 begin
                   MX               := random(10000);
                   MY               := random(10000);
                 end;
     AStream:=TMemoryStream.Create;
     try
        AStream.Write(Packet,SizeOf(TPacket));
        SendStream(IdTCPClientCmd,TStream(AStream));
     finally
        AStream.Free;
     end;
  end;

end;                

Answer 1

On the server side, your InputBufferIsEmpty() check is backwards. If the client is sending a lot of data, InputBufferIsEmpty() is likely to become False eventually, which will cause your server code to enter a tight unyielding loop that doesn't actually read anything. Just get rid of the check entirely and let ReceiveStream() block until there is a packet available to read.

Also, why are you setting the server's ListenQueue to 15, but the MaxConnections to 0? MaxConnections=0 will force the server to immediately close every client connection that is accepted, so the OnExecute event will never get a chance to be called.

On the client side, there is no need to destroy and recreate the TMemoryStream on each loop iteration, you should reuse that object.

But more importantly, you are not using write buffering correctly, so either fix that or get rid of it. I would do the latter, as you are sending lots of small packets, so just let TCP's default coalescing handle the buffering for you.

And TIdIOHandler.Write(TStream)/TIdIOHandler.ReadStream() can exchange the stream size for you, you don't need to do that manually.

Try this instead:

Server

program Srv;

{$I Synopse.inc}

uses
  {$IFDEF UNIX}{$IFDEF UseCThreads}
  cthreads,
  {$ENDIF}{$ENDIF}
  Classes, SysUtils, CustApp, Generics.Collections, IdTCPServer, IdCustomTCPServer, IdContext, IdGlobal, Db, mORMot, mORMotSQLite3, IdSync, functions, SynCommons, SynSQLite3Static;

type

  { TMyApplication }
  TMyApplication = class(TCustomApplication)
  var
    IdTCPServer: TIdTCPServer;
  protected
    procedure DoRun; override;
    procedure ServerOnConnect(AContext: TIdContext);
    procedure ServerOnExecute(AContext: TIdContext);
    function ReceiveStream(AContext: TIdContext; Size: Integer; var AStream: TStream);
  public
    constructor Create(TheOwner: TComponent); override;
    destructor Destroy; override;
  end;

  type
    TLog = class(TIdNotify)
    protected
      FMsg: string;
      procedure DoNotify; override;
    public
      class procedure LogMsg(const AMsg: string);
    end;

{ TMyApplication }

procedure TLog.DoNotify;
begin
  WriteLn(FMsg);
end;

class procedure TLog.LogMsg(const AMsg: string);
begin
  with TLog.Create do
  try
    FMsg := AMsg;
    Notify;
  except
    Free;
    raise;
  end;
end;

function TMyApplication.ReceiveStream(AContext: TIdContext; AStream: TStream): Boolean; overload;
begin
  try
    AContext.Connection.IOHandler.ReadStream(AStream, -1, False);
    AStream.Position := 0;
    Result := True;
  except
    Result := False;
  end;
end;      

procedure TMyApplication.ServerOnExecute(AContext: TIdContext);
var
  AStream: TMemoryStream;
begin
  AStream := TMemoryStream.Create;
  try 
    if not ReceiveStream(AContext, AStream) then
    begin
      AContext.Connection.Disconnect;
      Exit;
    end;
    TLog.LogMsg('--: '+random(100000).ToString); //After executing Client this line is displayed 70 time and CPU load is from 40 % to 70%
    // .. here we use AStream to execute some stuff  
  finally
    AStream.Free;
  end;        
end;

procedure TMyApplication.ServerOnConnect(AContext: TIdContext);
begin
  TLog.LogMsg('connect');        
  AContext.Connection.IOHandler.LargeStream := False;
end;

procedure TMyApplication.DoRun;
begin
  IdTCPServer := TIdTCPServer.Create;
  IdTCPServer.ListenQueue := 15;
  IdTCPServer.MaxConnections := 1;
  IdTCPServer.TerminateWaitTime := 5000;
  with IdTCPServer.Bindings.Add do
  begin
    IP   := '0.0.0.0';
    Port := 80;
    IPVersion := Id_IPv4;
  end;
  IdTCPServer.OnConnect := ServerOnConnect;
  IdTCPServer.OnDisconnect := ServerOnDiconnect;
  IdTCPServer.OnExecute := ServerOnExecute;
  IdTCPServer.Active := True;

  while True do
  begin
    Classes.CheckSynchronize();
    Sleep(10);
  end;

  ReadLn;
  Terminate;
end;

constructor TMyApplication.Create(TheOwner: TComponent);
begin
  inherited Create(TheOwner);
  StopOnException := True;
end;

destructor TMyApplication.Destroy;
begin
  IdTCPServer.Free;
  inherited Destroy;
end;

var
  Application: TMyApplication;
begin
  Application := TMyApplication.Create(nil);
  Application.Title := 'My Application';
  Application.Run;
  Application.Free;
end.

Client

function TForm1.SendStream(AClient: TIdTCPClient; AStream: TStream): Boolean; overload;
begin
  try
    AClient.IOHandler.LargeStream := False; // <-- or, set this 1 time after TIdTCPClient.Connect() exits...
    AClient.IOHandler.Write(AStream, 0, True);
    Result := True;
  except
    Result := False;
  end;
end;   
    
procedure TForm1.Button1Click(Sender: TObject);
var 
  Packet: TPacket;
  AStream: TMemoryStream;
  i: Integer;
begin
  AStream := TMemoryStream.Create;
  try
    AStream.Size := SizeOf(TPacket);
    for i := 0 to 1000 do 
    begin
      Application.ProcessMessages;
      with Packet do
      begin
        MX := random(10000);
        MY := random(10000);
      end;
      AStream.Position := 0;
      AStream.Write(Packet, SizeOf(TPacket));
      SendStream(IdTCPClientCmd, AStream);
    end;
  finally
    AStream.Free;
  end;
end;