_HandleRead/_HandleWrite/_NewIoData等改为虚方法

[hezihang12]

DelphiCrossSocket目前架构不支持零内存拷贝，因为采用了threadvar类型的buffer读取数据，这个buffer不能放入异步发送io中，因为io未发送完来新的接收io，将会覆盖发送io的内容。
为了绕开匿名函数方法的开销，并实现零内存拷贝需求，提高海量并发转发数据的性能。因此需要自定义内存块池，需要重载 _HandleRead/_HandleWrite/_NewIoData等方法。

[hezihang12]

举例：

{ 零拷贝 IOCP 覆盖层 (仅 Windows)

• PerIoData 池化: 线程本地空闲链表 + 全局备用
• 覆盖 _HandleRead: recv 读入池 buffer
• 覆盖 Send: 存 SrBuf 引用; _FreeIoData 自动 Release
  所有 PerIoData 操作都是 TSrIocpCrossSocket 的方法 (访问 protected TPerIoData)。 }
  unit sr.io.zerocopy;
  {$IFDEF FPC}{$MODE DELPHI}{$ENDIF}
  {$IFDEF WINDOWS}
  interface
  uses
  SysUtils, Classes, Windows,
  Net.Winsock2, Net.CrossSocket.Base, Net.CrossSocket, Net.CrossSocket.Iocp, Net.SocketAPI,
  sr.io.bufferpool;

type
TSrIocpCrossSocket = class(TIocpCrossSocket)
private
function PoolAcquire: PPerIoData;
procedure PoolRelease(P: PPerIoData);
function GetExtSrBuf(P: PPerIoData): TSrBuffer;
procedure SetExtSrBuf(P: PPerIoData; buf: TSrBuffer);
function GetExtNext(P: PPerIoData): PPerIoData;
procedure SetExtNext(P, N: PPerIoData);
function GetExtCreatorTid(P: PPerIoData): TThreadID;
procedure SetExtCreatorTid(P: PPerIoData; tid: TThreadID);
protected
function _NewIoData: PPerIoData; override;
procedure _FreeIoData(const P: PPerIoData); override;
procedure _HandleRead(const APerIoData: PPerIoData); override;
procedure _HandleWrite(const APerIoData: PPerIoData); override;
public
procedure Send(const AConnection: ICrossConnection; const ABuf: Pointer;
const ALen: Integer; const ACallback: TCrossConnectionCallback = nil); override;
{ 预热: 在 worker 线程启动时调用, 预分配 PerIoData 到线程本地链表 }
procedure WarmUpPioPool;
end;

implementation

const
{ 扩展区: SrBuf 指针 + Next 指针 + CreatorTid (用于 Release 时判断同线程/跨线程) }
EXT_TOTAL = SizeOf(Pointer) * 2 + SizeOf(TThreadID);

threadvar
SrPioFreeList: Pointer; { PPerIoData, 但 PPerIoData 不可见, 用 Pointer }
SrPioFreeCount: Integer;

var
SrPioGlobalLock: TRTLCriticalSection;
SrPioGlobalFree: Pointer; { 同上 }

{ ==================== 扩展区访问 (方法内可见 TPerIoData) ==================== }

function TSrIocpCrossSocket.GetExtSrBuf(P: PPerIoData): TSrBuffer;
var pp: ^Pointer;
begin
pp := Pointer(PByte(P) + SizeOf(TPerIoData));
Result := TSrBuffer(pp^);
end;

procedure TSrIocpCrossSocket.SetExtSrBuf(P: PPerIoData; buf: TSrBuffer);
var pp: ^Pointer;
begin
pp := Pointer(PByte(P) + SizeOf(TPerIoData));
pp^ := buf;
end;

function TSrIocpCrossSocket.GetExtNext(P: PPerIoData): PPerIoData;
var pp: ^PPerIoData;
begin
pp := Pointer(PByte(P) + SizeOf(TPerIoData) + SizeOf(Pointer));
Result := pp^;
end;

procedure TSrIocpCrossSocket.SetExtNext(P, N: PPerIoData);
var pp: ^PPerIoData;
begin
pp := Pointer(PByte(P) + SizeOf(TPerIoData) + SizeOf(Pointer));
pp^ := N;
end;

function TSrIocpCrossSocket.GetExtCreatorTid(P: PPerIoData): TThreadID;
var pp: ^TThreadID;
begin
pp := Pointer(PByte(P) + SizeOf(TPerIoData) + SizeOf(Pointer) * 2);
Result := pp^;
end;

procedure TSrIocpCrossSocket.SetExtCreatorTid(P: PPerIoData; tid: TThreadID);
var pp: ^TThreadID;
begin
pp := Pointer(PByte(P) + SizeOf(TPerIoData) + SizeOf(Pointer) * 2);
pp^ := tid;
end;

{ ==================== PerIoData 池 ==================== }

const
PIO_WARMUP_COUNT = 128; { PerIoData 线程预热 + 本地链表上限 }
PIO_GROWTH_COUNT = 128; { PerIoData 批量增长 + 全局搬迁批量 }

function TSrIocpCrossSocket.PoolAcquire: PPerIoData;
var
i, moved: Integer;
p: PPerIoData;
begin
{ 1. 线程本地链表 (无锁) }
if SrPioFreeList <> nil then
begin
Result := PPerIoData(SrPioFreeList);
SrPioFreeList := Pointer(GetExtNext(Result));
Dec(SrPioFreeCount);
end
else
begin
{ 2. 全局备用: 批量搬迁到本地 (减少锁竞争) }
Result := nil;
EnterCriticalSection(SrPioGlobalLock);
try
moved := 0;
while (SrPioGlobalFree <> nil) and (moved < PIO_GROWTH_COUNT) do
begin
p := PPerIoData(SrPioGlobalFree);
SrPioGlobalFree := Pointer(GetExtNext(p));
if Result = nil then
Result := p { 第一个返回给调用方 }
else
begin { 其余搬到线程本地 }
SetExtNext(p, PPerIoData(SrPioFreeList));
SrPioFreeList := Pointer(p);
Inc(SrPioFreeCount);
end;
Inc(moved);
end;
finally
LeaveCriticalSection(SrPioGlobalLock);
end;

if Result = nil then
begin
  { 3. 全局也空: 批量创建 PIO_GROWTH_COUNT 个, 放线程本地 }
  GetMem(Result, SizeOf(TPerIoData) + EXT_TOTAL);
  for i := 1 to PIO_GROWTH_COUNT - 1 do
  begin
    GetMem(p, SizeOf(TPerIoData) + EXT_TOTAL);
    SetExtNext(p, PPerIoData(SrPioFreeList));
    SrPioFreeList := Pointer(p);
    Inc(SrPioFreeCount);
  end;
end;

end;

FillChar(Result^, SizeOf(TPerIoData) + EXT_TOTAL, 0);
System.Initialize(Result^);
SetExtSrBuf(Result, nil);
SetExtCreatorTid(Result, FastThreadId);
end;

procedure TSrIocpCrossSocket.PoolRelease(P: PPerIoData);
var buf: TSrBuffer;
begin
buf := GetExtSrBuf(P);
if buf <> nil then
begin
buf._Release;
SetExtSrBuf(P, nil);
end;
System.Finalize(P^);

{ 同线程: 放回线程本地链表 (无锁, 无上限)
跨线程 (WSASend 完成回调在不同 IO 线程): 放全局备用 (锁)
避免全局 NtripServer 场景下某线程本地链表无限膨胀 }
if GetExtCreatorTid(P) = FastThreadId then
begin
SetExtNext(P, PPerIoData(SrPioFreeList));
SrPioFreeList := Pointer(P);
Inc(SrPioFreeCount);
end
else
begin
EnterCriticalSection(SrPioGlobalLock);
try
SetExtNext(P, PPerIoData(SrPioGlobalFree));
SrPioGlobalFree := Pointer(P);
finally
LeaveCriticalSection(SrPioGlobalLock);
end;
end;
end;

{ ==================== _NewIoData / _FreeIoData ==================== }

function TSrIocpCrossSocket._NewIoData: PPerIoData;
begin
Result := PoolAcquire;
AtomicIncrement(FPerIoDataCount);
end;

procedure TSrIocpCrossSocket._FreeIoData(const P: PPerIoData);
begin
if P = nil then Exit;
PoolRelease(P);
AtomicDecrement(FPerIoDataCount);
end;

{ ==================== _HandleRead ==================== }

procedure TSrIocpCrossSocket._HandleRead(const APerIoData: PPerIoData);
var
LConnection: ICrossConnection;
LRcvd, LError: Integer;
buf: TSrBuffer;
begin
if (APerIoData.CrossData = nil) then
begin
if Assigned(APerIoData.Callback) then APerIoData.Callback(nil, False);
Exit;
end;
LConnection := APerIoData.CrossData as ICrossConnection;
while True do
begin
buf := TSrBufferPool.Instance.Acquire;
LRcvd := TSocketAPI.Recv(LConnection.Socket, buf.DataPtr^, POOL_BUF_SIZE);
if (LRcvd = 0) then begin buf._Release; LConnection.Close; Exit; end;
if (LRcvd < 0) then
begin
LError := GetLastError; buf._Release;
if (LError = WSAEINTR) then Continue
else if (LError = WSAEWOULDBLOCK) or (LError = WSAEINPROGRESS) then Break
else begin LConnection.Close; Exit; end;
end;
buf.Len := LRcvd;
try TriggerReceived(LConnection, buf.DataPtr, LRcvd);
except end;
buf._Release;
if LConnection.IsClosed then Exit;
if (LRcvd < POOL_BUF_SIZE) then Break;
end;
if not _NewReadZero(LConnection) then LConnection.Close;
end;

{ ==================== Send ==================== }

procedure TSrIocpCrossSocket.Send(const AConnection: ICrossConnection;
const ABuf: Pointer; const ALen: Integer; const ACallback: TCrossConnectionCallback);
var
LPerIoData: PPerIoData;
LBytes, LFlags: Cardinal;
buf: TSrBuffer;
begin
LPerIoData := _NewIoData;
LPerIoData.Buffer.DataBuf.buf := ABuf;
LPerIoData.Buffer.DataBuf.len := ALen;
LPerIoData.Action := ioWrite;
LPerIoData.Socket := AConnection.Socket;
LPerIoData.CrossData := AConnection;
LPerIoData.Callback := ACallback;
SetExtSrBuf(LPerIoData, nil);

{ 反算池 buffer; 用魔数验证是否来自池 }
try
buf := BufToSrBuffer(ABuf);
if (buf.Magic = $52424253) and (buf.RefCount > 0) then
begin
SetExtSrBuf(LPerIoData, buf);
buf._AddRef;
end;
except
end;

LFlags := 0; LBytes := 0;
if (WSASend(AConnection.Socket, @LPerIoData.Buffer.DataBuf, 1, LBytes, LFlags,
PWSAOverlapped(LPerIoData), nil) < 0)
and (WSAGetLastError <> WSA_IO_PENDING) then
begin
_FreeIoData(LPerIoData);
if Assigned(ACallback) then ACallback(AConnection, False);
if Assigned(AConnection) then AConnection.Close;
end;
end;

{ ==================== _HandleWrite ==================== }

procedure TSrIocpCrossSocket._HandleWrite(const APerIoData: PPerIoData);
begin
if Assigned(APerIoData.Callback) then
APerIoData.Callback(APerIoData.CrossData as ICrossConnection, True);
end;

procedure TSrIocpCrossSocket.WarmUpPioPool;
var
i: Integer;
p: PPerIoData;
begin
{ 预热: 在 worker 线程上下文调用, 预分配 PIO_WARMUP_COUNT 个 PerIoData 到线程本地 }
for i := 1 to PIO_WARMUP_COUNT do
begin
p := PoolAcquire;
PoolRelease(p);
end;
end;

{ ==================== init/finalize ==================== }

initialization
InitCriticalSection(SrPioGlobalLock);
finalization
{ PerIoData 池的线程本地链表 + 全局链表在进程退出时由 OS 回收。
无法在此调用 GetExtNext (实例方法, 需要 TSrIocpCrossSocket 对象) 遍历释放。
Windows 进程退出时所有堆内存由 OS 统一回收, 无实际泄漏。 }
DoneCriticalSection(SrPioGlobalLock);
{$ENDIF WINDOWS}
end.