如何可靠地从 indy TCP 的输入缓冲区中提取多个数据包？

Question

我有一段代码，当许多数据包几乎同时从同一来源到达时，它大部分时间都会跳过数据包。构建数据包时，会在开头附加一个数据包大小字段，该字段后面是数据包的大小，以字节为单位。

TCP 客户端线程以 10 毫秒的间隔运行。

线程库

unit AgThread11;

interface

uses
  SysUtils, Classes, Windows, Rtti;

type

  TAgThreadMethod1 = procedure of object;
  TAgThreadMethod2 = procedure;

  TAgThread = class ( TThread )

  private
    fInterval       : Cardinal;
    fTerminateEvent : THandle;
    fRun            : Boolean;
    fOnRun1         : TAgThreadMethod1;
    fOnRun2         : TAgThreadMethod2;

  protected
    procedure Execute; override;

  public
    constructor Create
                ( const AOnRun: TAgThreadMethod1; const AInterval: Cardinal; 
                  const ARun: Boolean = True ); overload;
    constructor Create
                ( const AOnRun: TAgThreadMethod2; const AInterval: Cardinal; 
                  const ARun: Boolean = True ); overload;

    destructor  Destroy; override;
    procedure Signal;

    property  Run      : Boolean          read fRun       write fRun;
    property  Interval : Cardinal         read fInterval  write fInterval;
    property  OnRun1   : TAgThreadMethod1 read fOnRun1    write fOnRun1;
    property  OnRun2   : TAgThreadMethod2 read fOnRun2    write fOnRun2;

  end;

implementation

constructor TAgThread.Create 
            ( const AOnRun: TAgThreadMethod1; const AInterval: Cardinal; 
              const ARun: Boolean = True );
begin

  fTerminateEvent := CreateEvent ( nil, TRUE, FALSE, nil );
  fInterval := AInterval;
  fRun      := ARun;
  fOnRun1   := AOnRun;
  inherited Create ( False );

end;

constructor TAgThread.Create 
            ( const AOnRun: TAgThreadMethod2; const AInterval: Cardinal;  
              const ARun: Boolean = True );
begin

  fTerminateEvent := CreateEvent ( nil, TRUE, FALSE, nil );
  fInterval := AInterval;
  fRun      := ARun;
  fOnRun2   := AOnRun;
  inherited Create ( False );

end;

destructor  TAgThread.Destroy;
begin

  Terminate;
  Signal;
  WaitFor;
  inherited;

end;

procedure TAgThread.Signal;
begin
  SetEvent ( FTerminateEvent );
end;

procedure TAgThread.Execute;
begin

  while not Terminated do
  begin

    if fRun then
      if      Assigned ( fOnRun1 ) then fOnRun1
      else if Assigned ( fOnRun2 ) then fOnRun2;
    WaitForSingleObject ( FTerminateEvent, fInterval );

  end;

end;

end.

TCP 线程代码

procedure TForm1.THEX_TCP;
var
  Buffer  : TBytes;
  MsgSize : Integer;
begin

  if TCPClient.IOHandler.CheckForDataOnSource then
  begin

    while TCPClient.IOHandler.InputBuffer.Size >= 4 do
    begin

      fRXCount := fRXCount + 1;
      TCPClient.IOHandler.InputBuffer.ExtractToBytes ( Buffer, 4 );
      Move ( Buffer [0], MsgSize, 4 );
      TCPClient.IOHandler.InputBuffer.ExtractToBytes ( Buffer, MsgSize, False );
      NAT.RecievedNATData ( Buffer ); // Packet Processor

    end;

  end;

end;

如何确保零丢包？

Answer 1

你的TCP读取代码有两个主要问题：

1. 在第二次调用 ExtractToBytes() 之前，您无法确保 InputBuffer 实际上有 MsgSize 可用的字节数。如果您尝试提取比缓冲区中实际更多的字节，ExtractToBytes() 会引发异常。

2. 更重要的是，在每次调用 ExtractToBytes() 之前，您不会将 Buffer 变量的大小重新调整为 0。在第一次循环迭代中第一次调用之后，Buffer 的长度为 4 个字节。如果该消息的大小小于 4 个字节，那么您将在 Buffer 的末尾留下随机字节，这些字节将被传递给您的解析器，并且可能会破坏其逻辑。但更糟糕的是，如果缓冲区中有另一个消息大小，您的下一个循环迭代会第三次调用 ExtractToBytes() 并将这 4 个字节追加到现有 Buffer 内容的末尾，not 替换您假设的内容（ExtractToBytes() 的 AAppend 参数默认为 True）。因此，您最终会将前一个消息数据中的 4 个字节复制到您的 MsgSize 变量中，而不是刚刚提取的新 4 个字节，因此您在下一次 ExtractToBytes() 调用中使用了损坏的 MsgSize 值。

因为您的数据包是以长度为前缀的，所以您根本不需要使用CheckForDataOnSource() 或直接访问InputBuffer。使用以下代码，让 Indy 为您完成工作：

procedure TForm1.THEX_TCP;
var
  Buffer  : TBytes;
  MsgSize : Integer;
begin
  MsgSize := TCPClient.IOHandler.ReadLongInt;
  TCPClient.IOHandler.ReadBytes(Buffer, MsgSize);
  Inc(fRXCount);
  NAT.RecievedNATData(Buffer);
end;

默认情况下，这将阻塞调用者，直到有数据可供读取。如果THEX_TCP 需要在没有数据可供读取时退出，请改用以下内容：

procedure TForm1.THEX_TCP;
var
  Buffer  : TBytes;
  MsgSize : Integer;
begin
  if TCPClient.IOHandler.InputBufferIsEmpty then
  begin
    TCPClient.IOHandler.CheckForDataOnSource;
    TCPClient.IOHandler.CheckForDisconnect;
    if TCPClient.IOHandler.InputBufferIsEmpty then Exit;
  end;

  repeat
    MsgSize := TCPClient.IOHandler.ReadLongInt;
    TCPClient.IOHandler.ReadBytes(Buffer, MsgSize);
    Inc(fRXCount);
    NAT.RecievedNATData(Buffer);
    SetLength(Buffer, 0);
  until TCPClient.IOHandler.InputBufferIsEmpty;
end;

这种方法的唯一问题是ReadLongInt() 和ReadBytes() 可能会将更多字节读入InputBuffer，因此如果大量数据在短时间内发送，您的循环可能会运行很长时间的时间。如果您绝对必须一次只读取一个缓冲区并且只处理完整的消​​息，那么请使用以下内容：

procedure TForm1.THEX_TCP;
var
  MsgSizeBuffer: array[0..3] of Byte;
  MsgSize, I : Integer;
  Buffer : TBytes;
begin
  TCPClient.IOHandler.CheckForDataOnSource;
  TCPClient.IOHandler.CheckForDisconnect;

  while TCPClient.IOHandler.InputBuffer.Size >= 4 do
  begin
    // unfortunately, TIdBuffer does not have a way to peek
    // multiple bytes at a time without removing them
    for I := 0 to 3 do
      MsgSizeBuffer[I] := TCPClient.IOHandler.InputBuffer.Peek(I);
    Move(MsgSizeBuffer[0], MsgSize, 4);
    MsgSize := LongInt(GStack.NetworkToHost(LongWord(MsgSize)));

    if TCPClient.IOHandler.InputBuffer.Size < (4+MsgSize) then
      Break;

    TCPClient.IOHandler.InputBuffer.Remove(4);
    TCPClient.IOHandler.InputBuffer.ExtractToBytes(Buffer, MsgSize);
    Inc(fRXCount);
    NAT.RecievedNATData(Buffer);
    SetLength(Buffer, 0);
  end;
end;