udiiocptcpserver.pas

楠楠写的DBiocp例子都是源码

unit uDIIocpTcpServer;

interface

uses
  Windows, Sysutils, WinSock2, uCriticalSection, uIOCompletionPort, uDIThread,
  uDIProtocol, uDIBuffer, uDIPoolBuffer, uDIMapBuffer, uDIClientContext, uDISocketServer;
  {$I IOCP.inc}
  
type
  TDIIocpTcpServer = class;
  
  TDIListnerThread = class(TDIThread)
  protected
    FSocketServer: TDIIocpTcpServer;
    procedure ThreadRoutine; override;
  public
    constructor Create(SocketServer: TDIIocpTcpServer);
    destructor Destroy; override;
  end;

  TDIIocpTcpServer = class(TDISocketServer)
  private
    m_dLastTime:            DWORD;                  //上一次检查超时时间
    m_iIOCPPostAccepts:     Integer;                //下一次需要投递的AcceptEx数量
    m_nInitPostAcceptEx:    DWORD;                  //初次投递AcceptEx 数量
    
    m_hPostAcceptEvent:     Thandle;                //投递Accept I/O事件内核对象(连接新用户投递维持AcceptEx数量)
    m_hListnerThreadEvent:  Thandle;                //侦听线程事件WSAEventSelectI/O模型处理Accept事件
    m_hShutdownEvent:       Thandle;                //停止服务事件

    m_hListnerThread:       TDIListnerThread;       //侦听线程
    m_hWaitEventList:       array of Thandle;       //Wait线程内核对象列表(AcceptEx线程)

    m_lpfnAcceptEx:                 LPFN_ACCEPTEX;              //AcceptEx函数地址
    m_lpfnGetAcceptExSockaddrs:     LPFN_GETACCEPTEXSOCKADDRS;  //GetAcceptExSockaddrs函数地址

  protected
    m_PoolBuffer:                   TDIPoolBuffer;              //Buffer回收池
    m_MapAcceptExBuffer:            TDIMapBuffer;               //AcceptEx Buffer管理

  published
    property FPoolBuffer: TDIPoolBuffer read m_PoolBuffer;
    property FMapAcceptExBuffer: TDIMapBuffer read m_MapAcceptExBuffer;
  private
    //判断是否是一个安全的连接
    function IsSecurityConnection(FDIBuffer: TDIBuffer): Boolean;
  protected
    //启动侦听服务线程
    function ListnerStart: Boolean;
    procedure ListnerThreadProc;
    //生成客户端上下文(并投递Recv I/O操作)
    function AssociateIncomingClientWithContext(FDIBuffer: TDIBuffer): Boolean; overload;
    function AssociateIncomingClientWithContext(m_Socket: TSocket): Boolean; overload;

    //以下是消息处理函数
    //投递AcceptEx I/O消息
    function PostWSAAcceptEx(AcceptExBuffer: TDIBuffer): Boolean;
    //获取AcceptEx I/O消息
    function OnWSAAcceptEx(FDIBuffer: TDIBuffer; dwIoSize: DWORD): Boolean;
    //投递Recv I/O 消息
    function PostWSARecv(FClientContext: TDIClientContext): Boolean;
    //处理投递的Recv I/O 消息
    function OnWSARecv(FClientContext: TDIClientContext; FRecvBuffer: TDIBuffer; dwIoSize: DWORD): Boolean;
    //投递Send I/O 消息
    function PostWSASend(FClientContext: TDIClientContext): Boolean;
    //处理投递的Send I/O 消息
    function OnWSASend(FClientContext: TDIClientContext; FSendBuffer: TDIBuffer; dwIoSize: DWORD): Boolean;

    //投递Post消息
    function PostWSACloseSocket(FClientContext: TDIClientContext): Boolean;
    //踢出死连接
    procedure KillDeadClientSocket; override;

  protected
    procedure DisconnectAllBufferSocket;
    procedure DisconnectClient(FClientContext: TDIClientContext); override;
    procedure DisconnectAllClient; override;
    function ReleaseClientContext(FClientContext: TDIClientContext): Boolean; override;

    procedure ProcessFreeMemory; override;
    procedure ProcessIOError(FClientContext: TDIClientContext; FIOEventType: IOEventType; IOErrorType: TIOErrorType); override;
    procedure ProcessIOMessage(IOType: IOEventType; FClientContext: TDIClientContext; FBuffer: TDIBuffer; dwIoSize: DWORD); override;
  public
    constructor Create(IOCompletionPort: TIOCompletionPort);
    destructor Destroy; override;
    function Connect(strIPAddr: String; nPort: DWORD): Boolean; override;

    function StartServe( nPort: Integer;
                         iMaxNumConnections: DWORD;
                         iMaxNumberOfFreeContext: DWORD;
                         iMaxNumberOfFreeBuffer: DWORD;
                         StartType: TServerStartType = IOServerStart ): Boolean;  override;

		procedure StopServer; override;
  end;

implementation
  uses uDIMonitor;
  
//TDIListnerThread
constructor TDIListnerThread.Create(SocketServer: TDIIocpTcpServer);
begin
  inherited Create;
  FSocketServer := SocketServer;
end;

destructor TDIListnerThread.Destroy;
begin
  inherited Destroy;
end;

procedure TDIListnerThread.ThreadRoutine;
begin
  FSocketServer.ListnerThreadProc;
end;

//TDIIocpTcpServer
constructor TDIIocpTcpServer.Create(IOCompletionPort: TIOCompletionPort);
begin
  inherited Create(IOCompletionPort);
  m_lpfnAcceptEx := nil;
  m_lpfnGetAcceptExSockaddrs := nil;
  m_hShutdownEvent := INVALID_HANDLE_VALUE;
  m_hListnerThreadEvent := INVALID_HANDLE_VALUE;
  m_hPostAcceptEvent := INVALID_HANDLE_VALUE;
  m_PoolBuffer := TDIPoolBuffer.Create;
  m_MapAcceptExBuffer := TDIMapBuffer.Create;
end;

destructor TDIIocpTcpServer.Destroy;
begin
  FreeAndNil(m_PoolBuffer);
  FreeAndNil(m_MapAcceptExBuffer);
  inherited Destroy;
end;

function TDIIocpTcpServer.Connect(strIPAddr: String; nPort: DWORD): Boolean;
var
  clientSocket: TSocket;
begin
  Result := FALSE;

  if not FWinSocket.CreateWSASocket(clientSocket) then begin

    {$IFDEF _ICOP_DEBUGERR}
        AppendErrorLogMessage(Format('Connect()->socket()创建Socket套按字失败: %d.', [WSAGetLastError()]));
    {$ENDIF}
    Result := FALSE;
    Exit;
  end;

  if not FWinSocket.Connect(clientSocket, strIPAddr, nPort) then begin

    {$IFDEF _ICOP_DEBUGERR}
        AppendErrorLogMessage(Format('Connection 失败: %d.', [WSAGetLastError()]));
    {$ENDIF}
    Result := FALSE;
  end
  else
    Result := AssociateIncomingClientWithContext(clientSocket);
end;

procedure TDIIocpTcpServer.DisconnectClient(FClientContext: TDIClientContext);
var
  lingerStrt: TLinger;
  m_bDisconnect: Boolean;
begin
  if (FClientContext<>nil) then begin
    FClientContext.FContextLock.Lock;
    m_bDisconnect := FClientContext.FSocket <> INVALID_SOCKET;
    //关闭套接字
    if m_bDisconnect then begin
      lingerStrt.l_onoff := 1;
      lingerStrt.l_linger := 0;
      WinSock2.setsockopt( FClientContext.FSocket,
                           SOL_SOCKET,
                           SO_LINGER,
                           @lingerStrt,
                           sizeof(lingerStrt) );
      WinSock2.closesocket(FClientContext.FSocket);
      FClientContext.FSocket := INVALID_SOCKET;
      
      {$IFDEF _ICOP_DEBUG}
          AppendLogMessage('TDIIocpTcpServer.DisconnectClient关闭客户端套接字, Socket: '+IntToStr(FClientContext.m_KeyID));
      {$ENDIF}
    end;
    FClientContext.FContextLock.UnLock;
  end;
end;

procedure TDIIocpTcpServer.DisconnectAllClient;
var
  m_pFreeClientContext: TDIClientContext;
  m_pNextClientContext: TDIClientContext;
begin
  try
    FMapClientContextLock.Lock;

    m_pFreeClientContext := FMapClientContext.FClientContextList;
    m_pNextClientContext := nil;
    while (m_pFreeClientContext<> nil) do begin
      m_pNextClientContext := m_pFreeClientContext.m_pNext;
      DisconnectClient(m_pFreeClientContext);
      m_pFreeClientContext := m_pNextClientContext;
    end;
  finally
    FMapClientContextLock.UnLock;
  end;
end;

procedure TDIIocpTcpServer.DisconnectAllBufferSocket;
var
  lingerStrt: TLinger;
  m_pFreeBuffer: TDIBuffer;
  m_pNextBuffer: TDIBuffer;
begin
  try
    m_MapAcceptExBuffer.FMapBufferLock.Lock;

    m_pFreeBuffer := m_MapAcceptExBuffer.FBufferList;
    m_pNextBuffer := nil;
    while (m_pFreeBuffer<> nil) do begin
      m_pNextBuffer := m_pFreeBuffer.m_pNext;
      if m_pFreeBuffer.m_Socket<>INVALID_SOCKET then begin

        {$IFDEF _ICOP_DEBUG}
            AppendLogMessage('DisconnectAllBufferSocket关闭客户端套接字, Socket: '+IntToStr(m_pFreeBuffer.m_Socket));
        {$ENDIF}
        lingerStrt.l_onoff := 1;
        lingerStrt.l_linger := 0;
        WinSock2.setsockopt( m_pFreeBuffer.m_Socket,
                             SOL_SOCKET,
                             SO_LINGER,
                             @lingerStrt,
                             sizeof(lingerStrt) );
        WinSock2.closesocket(m_pFreeBuffer.m_Socket);
        m_pFreeBuffer.m_Socket := INVALID_SOCKET;
      end;
      m_pFreeBuffer := m_pNextBuffer;
    end;
  finally
    m_MapAcceptExBuffer.FMapBufferLock.UnLock;
  end;
end;

function TDIIocpTcpServer.ReleaseClientContext(FClientContext: TDIClientContext): Boolean; 
{$IFDEF _ICOP_DEBUG}
var
  key: Integer;
{$ENDIF}
begin

  //从上下文中移除
  FMapClientContextLock.Lock;
  if FMapClientContext.RemoveDIClientContext(FClientContext) then
  begin
    FMapClientContextLock.UnLock;
    InterlockedDecrement(m_iNumberOfActiveConnections);
    {$IFDEF _ICOP_DEBUG}
        key := FClientContext.m_KeyID;
    {$ENDIF}

    //断开事件
    if Assigned(OnCloseSocketEvent) then OnCloseSocketEvent(FClientContext);
    //真正释放内存
    FPoolClientContext.ReleaseClientContextToPool(FClientContext);

    {$IFDEF _ICOP_DEBUG}
        AppendLogMessage('释放处理->RemoveDIClientContext成功, Socket: '+ IntToStr(key) );
        AppendLogMessage('释放处理->ReleaseClientContextToPool, Key: '+IntToStr(key));
    {$ENDIF}
    //性能分析器 减少完成端口中连接数量 增加上下文 自定义断开数量
    {$IFDEF _IOCP_MONITOR}
        _DIMonitor.SubActiveConn;
        _DIMonitor.AddExitFreeContextCount;
    {$ENDIF}
    Exit;
  end;
  FMapClientContextLock.UnLock;
end;

procedure TDIIocpTcpServer.ProcessIOError(FClientContext: TDIClientContext; FIOEventType: IOEventType;  IOErrorType: TIOErrorType);
{$IFDEF _ICOP_DEBUG}
var
  key: Integer;
  nError: DWORD;
  nSend, nRecv, nClosed: Integer;
  sMsg: String;
{$ENDIF}
begin
  {$IFDEF _ICOP_DEBUG}
      FClientContext.FContextLock.Lock;
      key := FClientContext.m_KeyID;
      FClientContext.IsOutOfPendIO( nSend, nRecv, nClosed );
      FClientContext.FContextLock.UnLock;
      if FIOEventType = IOWSARecv then sMsg := 'IOWSARecv';
      if FIOEventType = IOWSASend then sMsg := 'IOWSASend';
      if FIOEventType = IOWSACloseSocket then sMsg := 'IOWSACloseSocket';
      if FIOEventType = IOWSAAcceptEx then sMsg := 'IOWSAAcceptEx';
  {$ENDIF}

  nError := GetLastError();
  //正常断开
  if IOErrorType = IOErrorNoramal then begin
    //关闭套接字, 释放内存空间
    //FSocketServerLock.Lock;
    DisconnectClient(FClientContext);
    ReleaseClientContext(FClientContext);

    {$IFDEF _ICOP_DEBUG}
        AppendLogMessage('正常断开信息, Socket : '+ IntToStr(key)+
                         ' I/O投递错误标识：'+sMsg+
                         ' GetLastError 错误：'+ IntToStr(nError)+
                          ' nSend->'+IntToStr(nSend)+'   nRecv->'+IntToStr(nRecv)+' nClosed ->'+IntToStr(nClosed)+
                         ' ClientContextPoolCount: '+IntToStr(FPoolClientContext.GetClientContextCount)+
                         ' FMapClientContext: '+IntToStr(FMapClientContext.FClientContextCount)+
                         ' m_iNumberOfActiveConnections:'+IntToStr(m_iNumberOfActiveConnections)+
                         ' 回收上下文个数：'+IntToStr(_DIMonitor.GetExitFreeContextCount)+
                         ' Post个数：'+ IntToStr( _DIMonitor.GetPostClose));
    {$ENDIF}
    //FSocketServerLock.UnLock;
  end;

  //自定义
  if IOErrorType = IOErrorCustom then begin
    //关闭套接字
    //FSocketServerLock.Lock;
    DisconnectClient(FClientContext);
    ReleaseClientContext(FClientContext);

    {$IFDEF _ICOP_DEBUG}
        AppendLogMessage('自定义断开信息, Socket : '+ IntToStr(key)+
                         ' I/O投递错误标识：'+sMsg+
                         ' GetLastError 错误：'+ IntToStr(nError)+
                         ' nSend->'+IntToStr(nSend)+'   nRecv->'+IntToStr(nRecv)+' nClosed ->'+IntToStr(nClosed)+
                         ' ClientContextPoolCount: '+IntToStr(FPoolClientContext.GetClientContextCount)+
                         ' FMapClientContext: '+IntToStr(FMapClientContext.FClientContextCount)+
                         ' m_iNumberOfActiveConnections:'+IntToStr(m_iNumberOfActiveConnections)+
                         ' 回收上下文个数：'+IntToStr(_DIMonitor.GetExitFreeContextCount)+
                         ' Post个数：'+ IntToStr( _DIMonitor.GetPostClose));
    {$ENDIF}
    //FSocketServerLock.UnLock;
  end;

  //异常
  if IOErrorType = IOErrorAbnormalNoramal then begin
    //关闭套接字, 释放内存空间
    //FSocketServerLock.Lock;
    DisconnectClient(FClientContext);
    ReleaseClientContext(FClientContext);

    {$IFDEF _ICOP_DEBUG}
        AppendLogMessage('异常断开信息, Socket : '+ IntToStr(key)+
                         ' I/O投递错误标识：'+sMsg+
                         ' GetLastError 错误：'+ IntToStr(nError)+
                         ' nSend->'+IntToStr(nSend)+'   nRecv->'+IntToStr(nRecv)+' nClosed ->'+IntToStr(nClosed)+
                         ' ClientContextPoolCount: '+IntToStr(FPoolClientContext.GetClientContextCount)+
                         ' FMapClientContext: '+IntToStr(FMapClientContext.FClientContextCount)+
                         ' m_iNumberOfActiveConnections:'+IntToStr(m_iNumberOfActiveConnections)+
                         ' 回收上下文个数：'+IntToStr(_DIMonitor.GetExitFreeContextCount)+
                         ' Post个数：'+ IntToStr( _DIMonitor.GetPostClose));
    {$ENDIF}
    //FSocketServerLock.UnLock;
  end;

end;

procedure TDIIocpTcpServer.ProcessIOMessage(IOType: IOEventType; FClientContext: TDIClientContext; FBuffer: TDIBuffer; dwIoSize: DWORD);
begin
  case IOType of
	  IOWSAAcceptEx:
      begin
        OnWSAAcceptEx(FBuffer, dwIoSize);
      end;
	  IOWSARecv:
      begin
        OnWSARecv(FClientContext, FBuffer, dwIoSize);
      end;
	  IOWSASend:
      begin
        OnWSASend(FClientContext, FBuffer, dwIoSize);
      end;
  end;