udiiocpudpserver.pas

楠楠写的DBiocp例子都是源码

unit uDIIocpUdpServer;

interface

uses
  Windows, Sysutils, WinSock2, uCriticalSection, uIOCompletionPort, uDIThread,
  uDIProtocol, uDIBuffer, uDIPoolBuffer, uDIMapBuffer, uDIClientContext, uDISocketServer;
  {$I IOCP.inc}
  
type

  TDIIocpUdpServer = class(TDISocketServer)
  private
    m_dLastTime: DWORD;
    //m_bOneIPPerConnection:          Boolean;                  //标志
    //procedure SetOneIPPerConnection(bValue: Boolean);         //设置是否一个IP只允许一个连接
  protected
    //生成客户端上下文(并投递Recv I/O操作)
    function AssociateIncomingClientWithContext(m_Socket: TSocket): Boolean; overload;

    //以下是消息处理函数
    //投递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 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;

//TDIIocpUdpServer
constructor TDIIocpUdpServer.Create(IOCompletionPort: TIOCompletionPort);
begin
  inherited Create(IOCompletionPort);
end;

destructor TDIIocpUdpServer.Destroy;
begin
  inherited Destroy;
end;

function TDIIocpUdpServer.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 TDIIocpUdpServer.DisconnectClient(FClientContext: TDIClientContext);
var
  lingerStrt: TLinger;
  m_bDisconnect: Boolean;
begin
  if (FClientContext<>nil) then begin
    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('TDIIocpUdpServer.DisconnectClient关闭客户端套接字, Socket: '+IntToStr(FClientContext.m_KeyID));
      {$ENDIF}
    end;
  end;
end;

procedure TDIIocpUdpServer.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;

function TDIIocpUdpServer.ReleaseClientContext(FClientContext: TDIClientContext): Boolean; 
{$IFDEF _ICOP_DEBUG}
var
  key: Integer;
{$ENDIF}
begin
  {$IFDEF _ICOP_DEBUG}
      FClientContext.FContextLock.Lock;
      key := FClientContext.m_KeyID;
      FClientContext.FContextLock.UnLock;
  {$ENDIF}

  //从上下文中移除
  FMapClientContextLock.Lock;
  FMapClientContext.RemoveDIClientContext(FClientContext);
  FMapClientContextLock.UnLock;
  InterlockedDecrement(m_iNumberOfActiveConnections);
  //断开事件
  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}
end;

procedure TDIIocpUdpServer.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);
    FSocketServerLock.UnLock;

    {$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}
  end;

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

    {$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}
  end;

  //异常
  if IOErrorType = IOErrorAbnormalNoramal then begin

    //关闭套接字, 释放内存空间
    FSocketServerLock.Lock;
    DisconnectClient(FClientContext);
    ReleaseClientContext(FClientContext);
    FSocketServerLock.UnLock;

    {$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}
  end;

end;

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

function TDIIocpUdpServer.AssociateIncomingClientWithContext(m_Socket: TSocket): Boolean;
var
  bVal: LongBool;
  FClientContext: TDIClientContext;
begin
  //检查套接字
  if (m_Socket = INVALID_SOCKET) then begin

    {$IFDEF _ICOP_DEBUGERR}
        AppendErrorLogMessage(Format('m_Scket = INVALID_SOCKET, AssociateIncomingClientWithContext退出: %d.', [WSAGetLastError()]));
    {$ENDIF}
    Result := FALSE;
    Exit;
  end;

  //系统退出或设置未连接状态，则关闭客户端SOCKET
  if (m_bShutDown) or (not m_bAcceptConnections) then begin

    {$IFDEF _ICOP_DEBUG}
        AppendLogMessage('m_bShutDown OR not m_bAcceptConnections, 关闭客户端连接.');
    {$ENDIF}
    FWinSocket.CloseWinSocket(m_Socket);
    Result := FALSE;
    Exit;
  end;

 	//是否超过最大连接数：
  FMapClientContextLock.Lock;
	if( m_iNumberOfActiveConnections >= m_iMaxNumConnections ) then begin
  FMapClientContextLock.UnLock;
    {$IFDEF _ICOP_DEBUG}
        AppendLogMessage(Format('客户端已经达到最大连接数: %d, 关闭客户端连接.', [m_iMaxNumConnections]));
    {$ENDIF}
    FWinSocket.CloseWinSocket(m_Socket);
    Result := FALSE;
    Exit;
  end;
  FMapClientContextLock.UnLock;

  //从空闲池中分配一个上下文
  FClientContext := m_PoolClientContext.AllocateFreeClientContextFromPool;
  if not Assigned(FClientContext) then begin

    {$IFDEF _ICOP_DEBUGERR}
        AppendErrorLogMessage(Format('FPoolClientContext.AllocateFreeClientContextFromPool错误: %d.', [WSAGetLastError()]));
    {$ENDIF}
    FWinSocket.CloseWinSocket(m_Socket);
    Result := FALSE;
    Exit;
  end;

  //设置Socket, KeyID, 得到客户端IP地址
  FClientContext.FSocket := m_Socket;
  FClientContext.m_KeyID := FClientContext.FSocket;

  //禁止nagle算法
  bVal := TRUE;
  if not FWinSocket.SetSocketNagle(FClientContext.m_Socket, bVal) then begin
  
    {$IFDEF _ICOP_DEBUGERR}
        AppendErrorLogMessage(Format('setsockopt(禁止nagle算法) 错误: %d.',[WSAGetLastError()]));
    {$ENDIF}
    //回收上下文
    m_PoolClientContext.ReleaseClientContextToPool(FClientContext);
    Result := FALSE;
    Exit;
  end;

  //客户上下文插入上下文管理中。
  FMapClientContextLock.Lock;
  if FMapClientContext.AddDIClientContext( FClientContext ) then begin
  FMapClientContextLock.UnLock;

    //关联完成端口
    if m_IOCompletionPort.AssociateSocketWithCompletionPort( FClientContext.FSocket,
                                                             DWORD(Pointer(FClientContext)) ) then
    begin
      //连接数加1
      InterlockedIncrement(m_iNumberOfActiveConnections);

      //性能分析器 增加完成端口中连接数量
      {$IFDEF _IOCP_MONITOR}
          _DIMonitor.AddActiveConn;
      {$ENDIF}

      //触发客户端连接事件
      if Assigned(OnNewSocketEvent) then OnNewSocketEvent(FClientContext);
      //投递Recv I/O 操作，接收客户端数据。
      PostWSARecv(FClientContext);
      Result :=TRUE;
    end
    else
    begin

      {$IFDEF _ICOP_DEBUGERR}
          AppendErrorLogMessage(Format('m_IOCompletionPort.AssociateSocketWithCompletionPort失败: %d.',[WSAGetLastError()]));
      {$ENDIF}

      //安全关闭客户端套接字
      FWinSocket.CloseWinSocket(m_Socket);
      //移出上下文Map, 释放上下文
      FMapClientContextLock.Lock;
      FMapClientContext.RemoveDIClientContext(FClientContext);
      FMapClientContextLock.UnLock;
      m_PoolClientContext.ReleaseClientContextToPool(FClientContext);
      Result := FALSE;
      Exit;
    end;
  end
  else
  begin
    FMapClientContextLock.UnLock;

    {$IFDEF _ICOP_DEBUGERR}
        AppendErrorLogMessage(Format('m_MapClientContext.AddClientContext失败: %d.',[WSAGetLastError()]));
    {$ENDIF}
    
    //安全关闭客户端套接字, 释放上下文
    FWinSocket.CloseWinSocket(m_Socket);
    m_PoolClientContext.ReleaseClientContextToPool(FClientContext);
    Result := FALSE;
    Exit;
  end;

end;


function TDIIocpUdpServer.PostWSARecv(FClientContext: TDIClientContext): Boolean;
var
  nRet: integer;
  dwIOSize: DWORD;
  dwFlags: DWORD;
begin