Delphi-IOCP 学习笔记<六>=====IO内存池和扩展套接字(ClientContext)-阿里云开发者社区

规划下将要出炉的IOCP。

1.将接收IO数据改成内存池。

2.扩展lpCompletionKey: DWORD参数.扩展套接字对象。

3.借鉴java netty思路，使用decode –> handler的思路来处理客户端数据。

//内存池

unit uMemPool;

interface

uses
  JwaWinsock2, Windows, SyncObjs;

const
  MAX_BUFFER_SIZE = 1024;

type
  LPPER_IO_OPERATION_DATA = ^PER_IO_OPERATION_DATA;

  PER_IO_OPERATION_DATA = packed record
    Overlapped: OVERLAPPED;
    IO_TYPE: Cardinal;
    DataBuf: TWSABUF;
    WorkBytes: Cardinal;    //如果是接收，接收的字节数
    WorkFlag: Cardinal;
    pre:LPPER_IO_OPERATION_DATA;
    next:LPPER_IO_OPERATION_DATA;
  end;
  
  TIODataMemPool = class(TObject)
  private
    FCs: TCriticalSection;

    //第一个可用的内存块
    FHead: LPPER_IO_OPERATION_DATA;

    //最后一个可用的内存卡
    FTail: LPPER_IO_OPERATION_DATA;

    //可用的内存个数
    FUseableCount:Integer;

    //正在使用的个数
    FUsingCount:Integer;

    /// <summary>
    ///   将一个内存块添加到尾部
    /// </summary>
    /// <param name="pvIOData"> (LPPER_IO_OPERATION_DATA) </param>
    procedure AddData2Pool(pvIOData:LPPER_IO_OPERATION_DATA);

    /// <summary>
    ///   得到一块可以使用的内存
    /// </summary>
    /// <returns> LPPER_IO_OPERATION_DATA
    /// </returns>
    function getUsableData: LPPER_IO_OPERATION_DATA;

    /// <summary>
    ///   创建一块内存空间
    /// </summary>
    /// <returns> LPPER_IO_OPERATION_DATA
    /// </returns>
    function InnerCreateIOData: LPPER_IO_OPERATION_DATA;

    procedure clearMemBlock(pvIOData:LPPER_IO_OPERATION_DATA);
  public
    class function instance: TIODataMemPool;
    constructor Create;
    destructor Destroy; override;

    //借一块内存
    function borrowIOData: LPPER_IO_OPERATION_DATA;

    //换会一块内存
    procedure giveBackIOData(const pvIOData: LPPER_IO_OPERATION_DATA);
  end;

implementation

var
  __IODATA_instance:TIODataMemPool;

constructor TIODataMemPool.Create;
begin
  inherited Create;
  FCs := TCriticalSection.Create();
  FUseableCount := 0;
  FUsingCount := 0;
end;

destructor TIODataMemPool.Destroy;
begin
  FCs.Free;
  inherited Destroy;
end;

{ TIODataMemPool }

procedure TIODataMemPool.AddData2Pool(pvIOData:LPPER_IO_OPERATION_DATA);
begin
  if FHead = nil then
  begin
    FHead := pvIOData;
    FHead.next := nil;
    FHead.pre := nil;
    FTail := pvIOData;
  end else
  begin
    FTail.next := pvIOData;
    pvIOData.pre := FTail;
    FTail := pvIOData;
  end;
  Inc(FUseableCount);
end;

function TIODataMemPool.InnerCreateIOData: LPPER_IO_OPERATION_DATA;
begin
  Result := LPPER_IO_OPERATION_DATA(GlobalAlloc(GPTR, sizeof(PER_IO_OPERATION_DATA)));
  GetMem(Result.DataBuf.buf, MAX_BUFFER_SIZE);
  Result.DataBuf.len := MAX_BUFFER_SIZE;

  //清理一块内存
  clearMemBlock(Result);
end;

function TIODataMemPool.borrowIOData: LPPER_IO_OPERATION_DATA;
begin
  FCs.Enter;
  try
    Result := getUsableData;
    if Result = nil then
    begin
      //生产一个内存块
      Result := InnerCreateIOData;

      //直接借走<增加使用计数器>
      Inc(FUsingCount);
    end;
  finally
    FCs.Leave;
  end;
end;

procedure TIODataMemPool.clearMemBlock(pvIOData: LPPER_IO_OPERATION_DATA);
begin
  //清理一块内存
  pvIOData.IO_TYPE := 0;
  
  pvIOData.WorkBytes := 0;
  pvIOData.WorkFlag := 0;

  ZeroMemory(@pvIOData.Overlapped, sizeof(OVERLAPPED));
  ZeroMemory(pvIOData.DataBuf.buf, pvIOData.DataBuf.len);
end;

procedure TIODataMemPool.giveBackIOData(const pvIOData:
    LPPER_IO_OPERATION_DATA);
begin
  FCs.Enter;
  try
    //清理内存块
    clearMemBlock(pvIOData);

    //加入到可以使用的内存空间
    AddData2Pool(pvIOData);

    //减少使用计数器
    Dec(FUsingCount);
  finally
    FCs.Leave;
  end;
end;

function TIODataMemPool.getUsableData: LPPER_IO_OPERATION_DATA;
var
  lvPre:LPPER_IO_OPERATION_DATA;
begin
  if FTail = nil then
  begin
    Result := nil;
  end else  
  begin   
    Result := FTail;

    lvPre := FTail.pre;
    if lvPre <> nil then
    begin
      lvPre.next := nil;
      FTail := lvPre;
    end else  //FTail是第一个也是最后一个,只有一个
    begin
      FHead := nil;
      FTail := nil;
    end;  

    Result.next := nil;
    Result.pre := nil;

    Dec(FUseableCount);
    Inc(FUsingCount);
  end;
end;

class function TIODataMemPool.instance: TIODataMemPool;
begin
  Result := __IODATA_instance;
end;


initialization
  __IODATA_instance := TIODataMemPool.Create;

finalization
  if __IODATA_instance <> nil then
  begin
    __IODATA_instance.Free;
    __IODATA_instance := nil;
  end;

end.

//扩展的套接字对象

unit uClientContext;

interface

uses
  Windows, JwaWinsock2, uBuffer, uBufferBuilder;

type
  TClientContext = class(TObject)
  private
    FSocket: TSocket;

    FBuffers: TBufferLink;
  public

    procedure CloseClientSocket;

    constructor Create(ASocket: TSocket);

    procedure AppendBuffer(const buf:WSABUF); overload; 

    function AppendBuffer(buf:PAnsiChar; len:Cardinal): Cardinal; overload;

    function readBuffer(buf:PAnsiChar; len:Cardinal): Cardinal;

    destructor Destroy; override;

    property Socket: TSocket read FSocket;

  end;

implementation

procedure TClientContext.AppendBuffer(const buf:WSABUF);
begin
  FBuffers.AddBuffer(buf.buf, buf.len);
end;

procedure TClientContext.CloseClientSocket;
begin
  if FSocket <> INVALID_SOCKET then
  begin
    closesocket(FSocket);
  end;
end;

constructor TClientContext.Create(ASocket: TSocket);
begin
  inherited Create;
  FSocket := ASocket;
  FBuffers := TBufferLink.Create();
end;

destructor TClientContext.Destroy;
begin
  FBuffers.Free;
  FBuffers := nil;  
  CloseClientSocket;
  inherited Destroy;
end;

function TClientContext.AppendBuffer(buf:PAnsiChar; len:Cardinal): Cardinal;
begin
  FBuffers.AddBuffer(buf, len);
end;

function TClientContext.readBuffer(buf:PAnsiChar; len:Cardinal): Cardinal;
begin
  Result := FBuffers.readBuffer(buf, len);
end;

end.

//修改后的代码工作线程和listener

unit uD10_IOCP;

interface

uses
  JwaWinsock2, Windows, SysUtils, uMemPool;

const
  DATA_BUFSIZE = 1024;

  IO_TYPE_Accept = 1;
  IO_TYPE_Recv = 2;



type
  //（1）：单IO数据结构
  PWorkData = ^TWorkerData;
  TWorkerData = packed record
    IOCPHandle:THandle;
    WorkerID:Cardinal;
  end;


function D10_IOCPRun(pvData:Pointer): Integer; stdcall;


implementation

uses
  logClientWrapper, uClientContext;


function ServerWorkerThread(pData:Pointer): Integer; stdcall;
var
  CompletionPort:THANDLE;
  lvWorkID:Cardinal;
  BytesTransferred:Cardinal;
  PerIoData:LPPER_IO_OPERATION_DATA;
  Flags:Cardinal;
  RecvBytes:Cardinal;
  lvResultStatus:BOOL;
  lvRet:Integer;
  
  lvClientContext:TClientContext;
begin
  CompletionPort:=PWorkData(pData).IOCPHandle;
  lvWorkID := PWorkData(pData).WorkerID;
   //得到创建线程是传递过来的IOCP
   while(TRUE) do
   begin
        //工作者线程会停止到GetQueuedCompletionStatus函数处，直到接受到数据为止
        lvResultStatus := GetQueuedCompletionStatus(CompletionPort,
          BytesTransferred,
          Cardinal(lvClientContext),
          POverlapped(PerIoData), INFINITE);

        if (lvResultStatus = False) then
        begin
          //当客户端连接断开或者客户端调用closesocket函数的时候,函数GetQueuedCompletionStatus会返回错误。如果我们加入心跳后，在这里就可以来判断套接字是否依然在连接。
          if lvClientContext<>nil then
          begin
            lvClientContext.Free;
            lvClientContext := nil;
          end;
          if PerIoData<>nil then
          begin
            TIODataMemPool.instance.giveBackIOData(PerIoData);
          end;
          continue;
        end;

        if PerIoData = nil then
        begin
          lvClientContext.Free;
          lvClientContext := nil;
          Break;
        end else  if (PerIoData<>nil) then
        begin
          if PerIoData.IO_TYPE = IO_TYPE_Accept then  //连接请求
          begin
            //发送日志显示
            lvRet := TLogClientWrapper.logINfo('工作线程[' + intToStr(lvWorkID) + ']:有新的连接接入');
            TIODataMemPool.instance.giveBackIOData(PerIoData);
          end else if PerIoData.IO_TYPE = IO_TYPE_Recv then
          begin
            //加入到套接字对应的缓存中
            lvClientContext.AppendBuffer(PerIoData.DataBuf.buf, PerIoData.Overlapped.InternalHigh);
            lvRet := TLogClientWrapper.logINfo('工作线程[' + intToStr(lvWorkID) + ']:接收到数据!');
            TIODataMemPool.instance.giveBackIOData(PerIoData);
          end;

          /////分配内存<可以加入内存池>
          PerIoData := TIODataMemPool.instance.borrowIOData;
          PerIoData.IO_TYPE := IO_TYPE_Recv;



          /////异步收取数据
          if (WSARecv(lvClientContext.Socket,
             @PerIoData.DataBuf,
             1,
             PerIoData.WorkBytes,
             PerIOData.WorkFlag,
             @PerIoData^, nil) = SOCKET_ERROR) then
          begin
            lvRet := GetLastError();
            //重叠IO,出现ERROR_IO_PENDING是正常的，
            //表示数据尚未接收完成，如果有数据接收，GetQueuedCompletionStatus会有返回值
            if (lvRet <> ERROR_IO_PENDING) then
            begin
              lvClientContext.Free;
              if PerIoData <> nil then
              begin
                GlobalFree(DWORD(PerIoData));
              end;
              Continue;
            end;
          end;
        end;
   end;
end;

function D10_IOCPRun(pvData:Pointer): Integer;
var
  WSData: TWSAData;
  lvIOPort, lvPerIOPort:THandle;
  hThread, dwThreadId:DWORD;

  sSocket, cSocket:TSocket;
  lvAddr:TSockAddr;
  lvAddrSize:Integer;
  lvMsg:String;
  lvPort:Integer;

  lvSystemInfo: TSystemInfo;
  i:Integer;

  PerIoData:LPPER_IO_OPERATION_DATA;

  lvWorkerData:PWorkData;

  Flags:Cardinal;
  RecvBytes:Cardinal;
  lvCount:Integer;
  lvClientContext:TClientContext;
begin

  lvPort := Integer(pvData);

  //加载SOCKET。使用的是2.2版为了后面方便加入心跳。
  WSAStartup($0202, WSData);

  // 创建一个完成端口（内核对象）
  lvIOPort := CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);


//  GetSystemInfo(lvSystemInfo);
//  lvCount := lvSystemInfo.dwNumberOfProcessors * 2 -1;

  lvCount := 1;

  //ServerWorkerThread 是工作线程
  for I:=1 to lvCount do
  begin
     lvWorkerData := AllocMem(SizeOf(TWorkerData));
     lvWorkerData.IOCPHandle := lvIOPort;
     lvWorkerData.WorkerID := i;
     hThread := CreateThread(nil, 0, @ServerWorkerThread,
       lvWorkerData,0, dwThreadId);
     if (hThread = 0) then
     begin
         Exit;
     end;
     CloseHandle(hThread);
  end;

  //创建一个套接字，将此套接字和一个端口绑定并监听此端口。
  sSocket:=WSASocket(AF_INET,SOCK_STREAM,0,Nil,0,WSA_FLAG_OVERLAPPED);
  if sSocket=SOCKET_ERROR then
  begin
      closesocket(sSocket);
      WSACleanup();
  end;
  lvAddr.sin_family:=AF_INET;
  lvAddr.sin_port:=htons(lvPort);
  lvAddr.sin_addr.s_addr:=htonl(INADDR_ANY);
  if bind(sSocket,@lvAddr,sizeof(lvAddr))=SOCKET_ERROR then
  begin
     closesocket(sSocket);
     exit;
  end;

  listen(sSocket,20);

  //下面循环进行循环获取客户端的请求。
  while (TRUE) do
  begin
     //当客户端有连接请求的时候，WSAAccept函数会新创建一个套接字cSocket。这个套接字就是和客户端通信的时候使用的套接字。
     cSocket:= WSAAccept(sSocket, nil, nil, nil, 0);

     //判断cSocket套接字创建是否成功，如果不成功则退出。
     if (cSocket= SOCKET_ERROR) then
     begin
        closesocket(sSocket);
        exit;
     end;

     //start----将套接字、完成端口绑定在一起。

     //     最开始的时候没有明白为什么还要调用一次createIoCompletionPort
     //
     //     后来经过google,和测试
     //
     //     是将新的套接字(socket)加入到iocp端口<绑定>
     //     这样工作线程才能处理这个套接字(socket)的数据包
     //如果把下面注释掉,WSARecv这个套接字时,GetQueuedCompletionStatus无法处理到收到的数据包
     //    注意第三个参数也需要进行绑定, 否则在工作线程中GetQueuedCompletionStatus时completionKey会取不到cSocket值

     lvClientContext := TClientContext.Create(cSocket);

     //将套接字、完成端口客户端对象绑定在一起。
     //2013年4月20日 13:45:10
     lvPerIOPort := CreateIoCompletionPort(cSocket, lvIOPort, Cardinal(lvClientContext), 0);
     if (lvPerIOPort = 0) then
     begin
        Exit;
     end;
     ////----end

     //初始化数据包
     PerIoData := TIODataMemPool.instance.borrowIOData;

     //数据包中的IO类型:有连接请求
     PerIoData.IO_TYPE := IO_TYPE_Accept;

     //通知工作线程,有新的套接字连接<第三个参数>
     PostQueuedCompletionStatus(
        lvIOPort, 0,
        Cardinal(lvClientContext),
        POverlapped(PerIOData));
  end;


end;

initialization

finalization
   WSACleanup;

end.

//在ClientContext中使用

unit uBuffer;
{
   套接字对应的接收缓存，使用链条模式。
}

interface

uses
  Windows;

type
  PBufRecord = ^_BufRecord;
  _BufRecord = packed record
    len: Cardinal; // the length of the buffer
    buf: PAnsiChar; // the pointer to the buffer

    preBuf:PBufRecord;   //前一个buffer
    nextBuf:PBufRecord;  //后一个buffer
  end;

  TBufferLink = class(TObject)
  private
    FHead:PBufRecord;
    FTail:PBufRecord;

    //当前读到的Buffer
    FRead:PBufRecord;
    //当前读到的Buffer位置
    FReadPosition: Cardinal;

    FMark:PBufRecord;
    FMarkPosition: Cardinal;

    function InnerReadBuf(const pvBufRecord: PBufRecord; pvStartIndex: Cardinal;
        buf: PAnsiChar; len: Cardinal): Cardinal;
  public
    constructor Create;

    procedure markReadIndex;

    procedure AddBuffer(buf:PAnsiChar; len:Cardinal);

    function readBuffer(buf:PAnsiChar; len:Cardinal):Cardinal;
  end;


implementation

constructor TBufferLink.Create;
begin
  inherited Create;
  FReadPosition := 0;
end;

{ TBufferLink }

procedure TBufferLink.AddBuffer(buf: PAnsiChar; len: Cardinal);
var
  lvBuf:PBufRecord;
begin
  New(lvBuf);
  lvBuf.preBuf := nil;
  lvBuf.nextBuf := nil;
  lvBuf.buf := GetMemory(len);
  lvBuf.len := len;
  CopyMemory(lvBuf.buf, Pointer(LongInt(buf)), len);
  if FHead = nil then
  begin
    FHead := lvBuf;
  end;
  
  if FTail = nil then
  begin
    FTail := lvBuf;
  end else
  begin
    FTail.nextBuf := lvBuf;
    lvBuf.preBuf := FTail;
  end;
end;

function TBufferLink.InnerReadBuf(const pvBufRecord: PBufRecord; pvStartIndex:
    Cardinal; buf: PAnsiChar; len: Cardinal): Cardinal;
var
  lvValidCount:Cardinal;
begin
  Result := 0;
  if pvBufRecord <> nil then
  begin
    lvValidCount := pvBufRecord.len-pvStartIndex;
    if lvValidCount <= 0 then
    begin
      Result := 0;
    end else
    begin
      if len <= lvValidCount then
      begin
        CopyMemory(buf, Pointer(Cardinal(pvBufRecord.buf) + pvStartIndex), len);
        Result := len;
      end else
      begin
        CopyMemory(buf, Pointer(Cardinal(pvBufRecord.buf) + pvStartIndex), lvValidCount);
        Result := lvValidCount;
      end;
    end;

  end;
end;

procedure TBufferLink.markReadIndex;
begin
  FMark := FRead;
  FMarkPosition := FReadPosition;
end;

function TBufferLink.readBuffer(buf: PAnsiChar; len: Cardinal): Cardinal;
var
  lvBuf:PBufRecord;
  lvPosition, l, lvReadCount, lvRemain:Cardinal;
begin
  lvReadCount := 0;
  lvBuf := FRead;
  lvPosition := FReadPosition;
  if lvBuf = nil then
  begin
    lvBuf := FHead;
    lvPosition := 0;
  end;

  if lvBuf <> nil then
  begin
    lvRemain := len;
    while lvBuf <> nil do
    begin
      l := InnerReadBuf(lvBuf, lvPosition, Pointer(Cardinal(buf) + lvReadCount), lvRemain);
      if l = lvRemain then
      begin
        //读完
        inc(lvReadCount, l);
        Inc(lvPosition, l);
        FReadPosition := lvPosition;
        FRead := lvBuf;
        lvRemain := 0;
        Break;
      end else if l < lvRemain then  //读取的比需要读的长度小
      begin
        lvRemain := lvRemain - l;
        inc(lvReadCount, l);
        Inc(lvPosition, l);
        FReadPosition := lvPosition;
        FRead := lvBuf;
        lvBuf := lvBuf.nextBuf;
        if lvBuf <> nil then   //读下一个
        begin
          FRead := lvBuf;
          FReadPosition := 0;
          lvPosition := 0;
        end;
      end;
    end;
    Result := lvReadCount;
  end else
  begin
    Result := 0;
  end;

end;

end.

>>>>后面研究Decoder

Delphi-IOCP 学习笔记<六>=====IO内存池和扩展套接字(ClientContext)

热门文章

最新文章

相关电子书