[Logical advise needed] TCP persistent connection custom protocol server

[FearDC 1]

Hi.

 

I'm currently working on a heavily-loaded TCP server in Delphi 12 using latest version of Indy 10, which will run on both Windows64 and Linux64.

 

There are over 3.000 concurrent clients connected to the server, every each of them is a persistent TCP connection using custom protocol - request/response based. Each connection is handshaked with as unique user with unique nickname/ID.

 

These requests are properly validated by several helpers like protocol parser, ban list, protocol feature support, flood detection, etc. So they involve shared memory access, hard-drive access, within and outside Indy classes.

 

The requests can be multiple, they can be one at a time, they can be ten or more at a time, from each client.

 

Main priority for the server is to follow the request/response order, otherwise protocol exchange will get out of sync between each connection and server. Connected users are also notified about connection/disconnection of other users.

 

So the actual logical advises I'm seeking is the correct/best way to perform this protocol synchronization, aswell as serving the requests in correct order. I will supply a basic class implementation of my application and write a few comments inside about which does what.

 

Server class:

type
  TOnUserConnect =    procedure(AConn: TIdContext) of object;
  TOnUserDisconnect = procedure(AConn: TIdContext) of object;
  TOnUserDataIn =     procedure(AConn: TIdContext; const AData: String) of object;
  TOnUserDataOut =    procedure(AConn: TIdContext; const AData: String) of object;

type // server listener
  TMyServer = class(TIdCustomTCPServer)
  protected
    procedure InitComponent; override;

  private
    FOnUserConnect:    TOnUserConnect;
    FOnUserDisconnect: TOnUserDisconnect;
    FOnUserDataIn:     TOnUserDataIn;

    procedure OnSocketConnect(AConn: TIdContext); // note: thread safety
    procedure OnSocketDisconnect(AConn: TIdContext); // note: thread safety
    procedure OnSocketExecute(AConn: TIdContext); // note: thread safety

    procedure DoParseProtocol(AConn: TIdContext; const AData: String);

  public
    LWaitFor: String; // command separator

    FOnUserDataOut:    TOnUserDataOut;

    property OnUserConnect:    TOnUserConnect    read FOnUserConnect    write FOnUserConnect;
    property OnUserDisconnect: TOnUserDisconnect read FOnUserDisconnect write FOnUserDisconnect;
    property OnUserDataIn:     TOnUserDataIn     read FOnUserDataIn     write FOnUserDataIn;
    property OnUserDataOut:    TOnUserDataOut    read FOnUserDataOut    write FOnUserDataOut;
  end;

procedure TMyServer.InitComponent;
begin
  inherited InitComponent;

  Self.ContextClass := TMyClient;

  Self.OnConnect := Self.OnSocketConnect;
  Self.OnDisconnect := Self.OnSocketDisconnect;
  Self.OnExecute := Self.OnSocketExecute;

//  Self.OnUserDataIn is bound below only to show short version,
//  otherwise it is specified in TMyServer parent class which is
//  actual protocol parser, user validator, and main worker

  Self.OnUserDataIn := DoParseProtocol;
end;

procedure TMyServer.OnSocketConnect(AConn: TIdContext);
begin
  TThread.Queue(nil,
    procedure
    begin
      if Assigned(Self.FOnUserConnect) then
        Self.FOnUserConnect(AConn);
    end
  );
end;

procedure TMyServer.OnSocketDisconnect(AConn: TIdContext);
begin
  TThread.Queue(nil,
    procedure
    begin
      if Assigned(Self.FOnUserDisconnect) then
        Self.FOnUserDisconnect(AConn);
    end
  );
end;

procedure TMyServer.OnSocketExecute(AConn: TIdContext);
var
  AData: String;
begin
  AConn.Connection.IOHandler.ReadTimeout := 5000;
  AData := AConn.Connection.IOHandler.WaitFor(LWaitFor);

  TThread.Queue(nil,
    procedure
    begin
      if Assigned(Self.FOnUserDataIn) then
        Self.FOnUserDataIn(AConn, AData);
    end
  );
end;

procedure TMyServer.DoParseProtocol(AConn: TIdContext; const AData: String);
var
  AUser: TMyClient;
begin
  AUser := AConn as TMyClient;

  if AData.Equals('Validate') then begin

    AUser.FNick := '<parsed data>';

//  below code will send response data to user directly - the connection will be written to,
//  but note that it's done inside Thread.Queue() - is this correct, or do i need to
//  step outside and perform the actual write? maybe some kind of internal buffer queue
//  that will write on OnServerExecute() to each connection?

    AUser.SendData('Hello');

  end else if AData.Equals('<bad command>') then begin

//  below code will disconnect the user, also inside Thread.Queue() - is this safe from here?

    AUser.Connection.Disconnect;
 
  end;

//  this is the actual protocol parser, from here user connection will be
//  verified, protocol parsed, responses written, other helper classes
//  will check for bans, shared memory access will be performed, file
//  contents will be read from hard drive and sent back to users,
//  thread-unsafe objects will be used, users will get disconnected,
//  basically all the main load will be performed here

end;

{ ... }

Client class:

type // user connection
  TMyClient = class(TIdServerContext)
  protected
    //

  private
    FNick: String;
    FFeatures: TStrings;

    procedure SendData(const AData: String);

  public
    constructor Create(AConn: TIdTCPConnection; AYarn: TIdYarn; AList: TIdContextThreadList = nil); override;
    destructor Destroy; override;
  end;

constructor TMyClient.Create(AConn: TIdTCPConnection; AYarn: TIdYarn; AList: TIdContextThreadList = nil);
begin
  inherited Create(AConn, AYarn, AList);

  FNick := '';
  FFeatures := TStrings.Create;
end;

destructor TMyClient.Destroy;
begin
  FFeatures.Free;

  inherited Destroy;
end;

procedure TMyClient.SendData(const AData: String);
begin
  if Self.Connection.Connected then begin
    Self.Connection.IOHandler.Write(AData);

    if Assigned((Self.Server as TMyServer).FOnUserDataOut) then
      (Self.Server as TMyServer).FOnUserDataOut(Self, AData);
  end;
end;

//  in this class connection will be written to, disconnected, etc

{ ... }

 

I know about Indys internal write buffering, it will be filled with an amount and written when maximum size is reached - this is exactly what I need, so basically I don't need some kind of internal buffer implementation for outbound data.

 

Thread syncing is required to step outside Indys threads in order to access shared data - what ever that could be. I do that, but I'm not sure if the actual write or disconnect against clients is the right place to perform.

 

Well, I hope you get the logic of my server - if not, please let me know, I will describe further.

 

Any advise is appreciated.

 

Regards.

Edited February 7, 2024 by FearDC

[Remy Lebeau 1659]

53 minutes ago, FearDC said:

Main priority for the server is to follow the request/response order, otherwise protocol exchange will get out of sync between each connection and server. Connected users are also notified about connection/disconnection of other users.

In a strictly request/response model, having the server send unsolicited notifications to the client will simply not work (unless they are delivered on a separate channel).  When a client sends a request and is waiting for a response, it won't know to expect or handle notifications before the response arrives.  And if the client is not waiting for a response, it won't be able to read in notifications at all until it sends its next request.  So, you would have to either:

 

• cache the notifications per-client and deliver them only in solicited responses.  But that means there is possible delay in notifications being delivered.
• make the client poll the server for the status of other clients periodically. Which is usually not desirable, especially with the sheer number of clients you are dealing with.

 

Otherwise, you have to change your protocol to allow for clients to handle unsolicited notifications while waiting for solicited responses, and to handle unsolicited notifications while no request is pending.  This typically requires:

• the client to read from the connection constantly.
• the client to include a unique ID in each request.
• the server to echo that unique ID back in the corresponding response.
• the server to format/identify notifications and responses in a way that allows the client to differentiate between them.

Quote

I know about Indys internal write buffering, it will be filled with an amount and written when maximum size is reached - this is exactly what I need, so basically I don't need some kind of internal buffer implementation for outbound data.

Indy's write buffering is at the byte level when interacting directly with the socket connection.  But you will likely need higher level message buffering at the business logic level instead.  IOW, you will likely need a per-client thread-safe queue for outgoing messages, and then you can push your responses and notifications into that queue as needed.  Then you can have the server's OnExecute event flush the calling client's queue to its connection when it is safe to do so.  I've posted examples of that many times before in various forums.

Edited February 6, 2024 by Remy Lebeau

[FearDC 1]

Thank you for your reply @Remy Lebeau.

 

I'm having hard time to understand what an "unsolicited" notification/response is - kind of "unexpected"?

 

I will provide you with short example of client to server handshake and further communication.

 

Client to server handshake:

Quote

Server > Client1: $Welcome|

Client1 > Server: $MyFeatures Chat Connect|$MyName Client1|

Server > Client1: $MyFeatures Chat Connect|$Hello Client1|

Server > All clients: $ClientOnline Client1|

(or disconnect if nick is taken for example)

Server > Client1: $ChatFrom Server Try another nick|

Client1.Connection.Disconnect;

 

Next part is a free communication where client talks first:

Quote

Client1 > Server: $ConnectTo Client2 Via 1.2.3.4:5|

(private communication between Client1 and Client2)

Server > Client2: $ConnectFrom Client1 Via 1.2.3.4:5|

(or ignore the request if Client1 is not allowed to connect)

 

Client1 > Server: $ChatTo Client2 <Message>|

Server > Client2: $ChatFrom Client1 <Message>|

(or ignore the request if Client1 is not allowed to chat)

 

Client1 > Server: $ChatToAll <Message>|

Server > All clients: $ChatFrom Client1 <Message>|

(or disconnect if Client1 is flooding)

Server > Client1: $ChatFrom Server Stop flooding|

Client1.Connection.Disconnect;

 

Another moment is where server talks first notifying everyone that someone enters or leaves:

Quote

Server > All clients: $ClientOnline Client1|

...

Server > All clients: $ClientOffline Client2|

 

That's it basically.

 

When a client connects, the server creates a separate thread where it handles reads and writes from/to client. It first welcomes the client, exchanges the handshake information and waits for requests from client. Possibly sends public notifications to client.

 

I don't really understand why the above would not work. Or do you mean that client will not be written to while server is reading the client - client thread is frozen and waiting for data from client? In that case there is timeout - is that the delay you are talking about?

 

Sorry, english is not my primary language, so it's not always I understand everything even if I try to translate. Also the logical part is the hardest part for me, even in real life, while writing the code is the easiest. 😛

 

Regards.

[Remy Lebeau 1659]

4 hours ago, FearDC said:

I'm having hard time to understand what an "unsolicited" notification/response is - kind of "unexpected"

Meaning, the server sends something to a client that it did not explicitly ask for.  Such as a notification when another client connects/disconnects, or when a client sends a message to other client.  Things which can happen at any time, outside of the normal request/response flow of "I want something" -> "Here you go" (solicited). "Oh, BTW, here is something else I have for you but you didn't ask for" (unsolicited).

4 hours ago, FearDC said:

Client to server handshake:

That contains a mix of solicited and unsolicited messaging.

 

The new client connects, gets a greeting, says hello, and gets acknowledged.  The client initiated an action and got a direct response before doing anything else.  That is solicited messaging.

 

Then, all of the other clients get notified of the new connection.  They didn't have to send any requests to receive that event, they are just blindly told that it happened, when it happened.  It happened in the background, and they have to passively watch out for it.  That is unsolicited messaging.

4 hours ago, FearDC said:

Next part is a free communication where client talks first

That is unsolicited messaging.  Client1 sends a request to the server (solicited), but then the server notifies Client2/etc (unsolicited).  Client2/etc did not send a request to receive that event, it/they are blindly told when the event happens.

4 hours ago, FearDC said:

Another moment is where server talks first notifying everyone that someone enters or leaves:

Same thing.

 

All that extra messaging that the server does in the background needs to be handled properly.  But if you code your server to just receive a request and send a response, you won't be able to handle that extra messaging at all.

4 hours ago, FearDC said:

When a client connects, the server creates a separate thread where it handles reads and writes from/to client. It first welcomes the client, exchanges the handshake information and waits for requests from client. Possibly sends public notifications to client.

And right there is the problem.  The code you showed doesn't handle that last part.  You can't just write to a given client connection whenever and from wherever you want.  You have to serialize your outgoing messaging.  Think of what would happen if 2 clients happen to connect/disconnect at the same moment and thus need to notify the same target client.  Or if multiple clients send private messages to the same target client at the same moment.  You don't want multiple messages to that target client to overlap, that will corrupt your socket communications.

 

So, you must provide some mechanism to make sure that subsequent messages wait their turn while an earlier message is still being written to the socket.  That could as simple as putting a critical section or other exclusive lock around socket writes.  But, that can cause unwanted blockages in threads that want to write to a socket, so that is where my earlier suggestion to use a per-client queue comes into play instead.  That way, only the thread that actually writes the queue to the socket may block, while other threads are free to continue putting new messages into the queue.

4 hours ago, FearDC said:

do you mean that client will not be written to while server is reading the client - client thread is frozen and waiting for data from client? In that case there is timeout

Well, that is certainly true, if you do the writing in the OnExecute event (as you usually should).  But, if OnExecute never writes, only reads, then you can do the writing from another thread.

4 hours ago, FearDC said:

is that the delay you are talking about?

I didn't say anything about a delay.

[FearDC 1]

Thank you for help @Remy Lebeau.

 

I will figure out some sort of priority outbound buffer with instant flushing.

 

I might ask more questions, but later in that case.

 

Regards.

[FearDC 1]

It seems that I managed to tackle this, atleast there is a working model 😃

 

Each client is waiting for data in its own thread, without blocking main thread.

 

On data arrival it calls main thread to handle the request and send an answer instantly - this is the only blocking operation on main thread, but data is usually very small and gets delivered instantly. Is there a timeout for sending buffer data with blocking sockets? @Remy Lebeau

 

This model allows a synchronized way of ordered communication betweer server and multiple clients in any direction.

 

The only problem I have left it seem, is that some client threads are not terminated - still waiting to read data, when I suddenly stop listening on custom TCP server. I have tried to Connection.Disconnect with locking contexts. Any idea on this one? @Remy Lebeau

[Remy Lebeau 1659]

42 minutes ago, FearDC said:

Each client is waiting for data in its own thread, without blocking main thread.

You get that for free with TIdTCPServer, as each client runs in its own thread.

42 minutes ago, FearDC said:

On data arrival it calls main thread to handle the request and send an answer instantly - this is the only blocking operation on main thread, but data is usually very small and gets delivered instantly.

Sending the answer from the main thread itself is not a good idea. If the send blocks, not only will you block all other requests and answers, but also block the UI, too.  This goes back to my earlier suggestion that you should use a per-client queue for outgoing messages. The main thread can put the answer in the client's queue, and then the server's OnExecute event can send the client's queue when it is safe to do so, ie between reads of incoming messages.

42 minutes ago, FearDC said:

Is there a timeout for sending buffer data with blocking sockets?

By default, no. Blocking sockets wait forever for operations to finish. But, if desired, you can use the TIdContext.Binding.SetSockOpt() method to set a timeout for the SO_SNDTIMEO option on the underlying socket. If a send times out, a failure will be reported back to you. At which point, you don't know the state of the socket, or how many bytes it may have sent before timing out, so the only sane thing you can do is close that socket and let that client reconnect to your server.

42 minutes ago, FearDC said:

This model allows a synchronized way of ordered communication betweer server and multiple clients in any direction.

That may or may not be a good design choice.  For example, if Client1 and Client2 want to send a message to Client3 at the same time, then you would need to synchronize those messages.  But, if Client1 wants to send a message to Client2 while Client3 wants to send a message to Client4 at the same time, then there is no need to synchronize those messages.  So, unless the requests are accessing UI resources, there is not a good reason for the main UI thread to be processing the client requests.

42 minutes ago, FearDC said:

The only problem I have left it seem, is that some client threads are not terminated - still waiting to read data, when I suddenly stop listening on custom TCP server. I have tried to Connection.Disconnect with locking contexts. Any idea on this one?

TIdTCPServer already disconnects active client sockets during its shutdown. But, if you must do it manually, then using Connection.Binding.CloseSocket() would be safer than Connection.Disconnect() especially in a multi-threaded environment.

 

That said, make sure your server's event handlers are exiting properly during server shutdown. This is another reason NOT to use the main UI thread to process requests. If the main thread is blocked shutting down the server then it can't handle subsequent synchronization requests, thus causing a deadlock (client threads are blocked waiting on the main thread, and the main thread is blocked waiting on the client threads).  So, do not perform synchronizations with the main thread during server shutdown, or else perform the shutdown in a separate thread leaving the main thread open to handle synchronizations until the server is fully shutdown.

[FearDC 1]

17 hours ago, Remy Lebeau said:

Sending the answer from the main thread itself is not a good idea. If the send blocks, not only will you block all other requests and answers, but also block the UI, too.  This goes back to my earlier suggestion that you should use a per-client queue for outgoing messages. The main thread can put the answer in the client's queue, and then the server's OnExecute event can send the client's queue when it is safe to do so, ie between reads of incoming messages.

Could you show me an example please?

 

To be more exact, how do I get thread handle of a context?

Edited yesterday at 01:57 PM by FearDC

[Remy Lebeau 1659]

10 hours ago, FearDC said:

Could you show me an example please?

Try something like this (based on your earlier example):

type
  TOnUserConnect =    procedure(AConn: TIdContext) of object;
  TOnUserDisconnect = procedure(AConn: TIdContext) of object;
  TOnUserDataIn =     procedure(AConn: TIdContext; const AData: String) of object;
  TOnUserDataOut =    procedure(AConn: TIdContext; const AData: String) of object;

type // server listener
  TMyServer = class(TIdCustomTCPServer)
  protected
    procedure InitComponent; override;

  private
    FOnUserConnect:    TOnUserConnect;
    FOnUserDisconnect: TOnUserDisconnect;
    FOnUserDataIn:     TOnUserDataIn;
    FOnUserDataOut:    TOnUserDataOut;

    procedure DoConnect(AContext: TIdContext); override; // note: thread safety
    procedure DoDisconnect(AContext: TIdContext); override; // note: thread safety
    function  DoExecute(AContext: TIdContext): Boolean; override; // note: thread safety

    procedure DoParseProtocol(AContext: TIdContext; const AData: String);

  public
    LWaitFor: String; // command separator

    property OnUserConnect:    TOnUserConnect    read FOnUserConnect    write FOnUserConnect;
    property OnUserDisconnect: TOnUserDisconnect read FOnUserDisconnect write FOnUserDisconnect;
    property OnUserDataIn:     TOnUserDataIn     read FOnUserDataIn     write FOnUserDataIn;
    property OnUserDataOut:    TOnUserDataOut    read FOnUserDataOut    write FOnUserDataOut;
  end;

procedure TMyServer.InitComponent;
begin
  inherited InitComponent;
  ContextClass := TMyClient;
end;

procedure TMyServer.DoConnect(AContext: TIdContext);
begin
  if Assigned(FOnUserConnect) then
    TThread.Queue(nil,
      procedure
      begin
        if Assigned(FOnUserConnect) then
          FOnUserConnect(AContext);
      end
    );
end;

procedure TMyServer.DoDisconnect(AContext: TIdContext);
begin
  if Assigned(FOnUserDisconnect) then
    TThread.Queue(nil,
      procedure
      begin
        if Assigned(FOnUserDisconnect) then
          FOnUserDisconnect(AContext);
      end
    );
end;

function TMyServer.DoExecute(AContext: TIdContext): Boolean;
var
  AData: String;
begin
  (AContext as TMyClient).SendQueue;

  AData := AContext.Connection.IOHandler.WaitFor(LWaitFor, True, False, nil, 5000);
  if AData <> '' then
  begin
    TThread.Queue(nil,
      procedure
      begin
        if Assigned(FOnUserDataIn) then
          FOnUserDataIn(AConn, AData);
        DoParseProtocol(AConn, AData);
      end
    );
  end;

  Result := AContext.Connection.Connected;
end;

procedure TMyServer.DoParseProtocol(AConn: TIdContext; const AData: String);
var
  AUser: TMyClient;
begin
  AUser := AConn as TMyClient;

  if AData.Equals('Validate') then begin

    AUser.FNick := '<parsed data>';
    AUser.QueueData('Hello');

  end else if AData.Equals('<bad command>') then begin

    AConn.Binding.CloseSocket;
 
  end;

//  this is the actual protocol parser, from here user connection will be
//  verified, protocol parsed, responses written, other helper classes
//  will check for bans, shared memory access will be performed, file
//  contents will be read from hard drive and sent back to users,
//  thread-unsafe objects will be used, users will get disconnected,
//  basically all the main load will be performed here

end;

...

uses
  ..., IdThreadSafe;

type // user connection
  TMyClient = class(TIdServerContext)
  protected
    //
    FQueue: TIdThreadSafeStringList;
    ...
    procedure SendData(const AData: String);
    procedure SendQueue;

  public
    constructor Create(AConn: TIdTCPConnection; AYarn: TIdYarn; AList: TIdContextThreadList = nil); override;
    destructor Destroy; override;

    procedure QueueData(const AData: String);
  end;

constructor TMyClient.Create(AConn: TIdTCPConnection; AYarn: TIdYarn; AList: TIdContextThreadList = nil);
begin
  inherited Create(AConn, AYarn, AList);
  ...
  FQueue := TIdThreadSafeStringList.Create;
end;

destructor TMyClient.Destroy;
begin
  ...
  FQueue.Free;
  inherited Destroy;
end;

procedure TMyClient.QueueData(const AData: String);
begin
  FQueue.Add(AData);
end;

procedure TMyClient.SendData(const AData: String);
begin
  Connection.IOHandler.Write(AData);

  if Assigned((Server as TMyServer).FOnUserDataOut) then
    TThread.Queue(nil,
      procedure
      begin
        if Assigned((Server as TMyServer).FOnUserDataOut) then
          (Server as TMyServer).FOnUserDataOut(Self, AData);
      end
    );
end;

procedure TMyClient.SendQueue;
var
  QueueList: TStringList;
  SendList: TStringList;
  I: Integer;
begin
  SendList := nil;
  try
    QueueList := FQueue.Lock;
    try
      if QueueList.Count = 0 then Exit;
      SendList := TStringList.Create;
      SendList.Assign(QueueList);
      QueueList.Clear;
    finally
      FQueue.Unlock;
    end;
    for I := 0 to SendList.Count-1 do
      SendData(SendList[I]);
  finally
    SendList.Free;
  end;
end;

...

10 hours ago, FearDC said:

To be more exact, how do I get thread handle of a context?

Why? You don't need the thread handle for this task.

 

But, that being said, it is technically possible if you do need it for some reason - you can type-cast the TIdContext.Yarn property to TIdYarnOfThread, and then use the TIdYarnOfThread.Thread.Handle property.

handle.

[FearDC 1]

Thank you for putting your time on this example, it is very useful!