uciocpserver.pas

楠楠写的DBiocp例子都是源码

  //设置端口复用
  bVal := True;
  nRet := SetSockOpt(m_sListen, SOL_SOCKET, SO_REUSEADDR, PChar(@bVal), SizeOf(bVal));
  if (nRet = SOCKET_ERROR) then begin
    AppendMsgLog(Format('setsockopt(SO_EXCLUSIVEADDRUSE) 创建失败: %d.', [WSAGetLastError()]));
		Winsock2.closesocket(m_sListen);
    Result := FALSE;
    Exit;
  end;

  FillChar(sock_Addr, SizeOf(sock_Addr), 0);
  sock_Addr.sin_family := AF_INET;
  sock_Addr.sin_addr.S_addr := INADDR_ANY;
  sock_Addr.sin_port := htons(m_nPort);
  //绑定套接字
  nRet := Bind(m_sListen, @sock_Addr, sizeof(sock_Addr));
	if (nRet = SOCKET_ERROR) then begin
		AppendMsgLog(Format('bind() 调用失败: %d.', [WSAGetLastError()]));
    Winsock2.closesocket(m_sListen);
    Result := FALSE;
    Exit;
	end;
  //侦听
  nRet := Listen(m_sListen, 5);
	if (nRet = SOCKET_ERROR) then begin
		AppendMsgLog(Format('listen() 调用失败: %d.', [WSAGetLastError()]));
    Winsock2.closesocket(m_sListen);
    Result := FALSE;
    Exit;
	end;

  //将监听套节字关联到完成端口，注意，这里为它传递的CompletionKey为0
	AssociateSocketWithCompletionPort(m_sListen, m_hCompletionPort, DWORD(0));

  //投递AcceptEx I/O
  if not PostWSAAcceptEx then
  begin
    Result :=  FALSE;
    Exit;
  end;

  m_bAcceptConnections := TRUE;	//设置状态，允许客户端连接
	Result := TRUE;
end;

function TCIOCPServer.ChangeSocketModeAccept(m_Socket: TSocket): Boolean;
begin
  if (SOCKET_ERROR = setsockopt( m_Socket,
                                 SOL_SOCKET,
                                 SO_UPDATE_ACCEPT_CONTEXT,
                                 Pansichar(@m_sListen),
                                 SizeOf(m_sListen) )) then
  begin
    AppendMsgLog(Format('setsockopt(SO_UPDATE_ACCEPT_CONTEXT) 调用失败: %d, 关闭客户端连接.', [WSAGetLastError()]));
    WinSock2.closesocket(m_Socket);
    m_Socket := INVALID_SOCKET;
		Result := FALSE;
    Exit;
  end;

	Result := TRUE;
end;

function TCIOCPServer.AllocateFreeContextFromPool: TAbstractContext;
var
  i: Integer;
begin
  Result := nil;

  m_FreeContextQueueLock.Lock;
  if FFreeContextQueue.Count>=1 then
  begin
    Result := TAbstractContext(FFreeContextQueue.Pop);
    Result.InitContext;
  end;
  m_FreeContextQueueLock.UnLock;

  if Result = nil then
  begin
    Result := TAbstractContext.Create;
    Result.InitContext;
  end;
end;

procedure TCIOCPServer.ReleaseClientContextToFreePool(absContext: TAbstractContext);
begin
  m_FreeContextQueueLock.Lock;
  FFreeContextQueue.Push(absContext);
  m_FreeContextQueueLock.UnLock;
end;

procedure TCIOCPServer.AddContextToClientHasTable(absContext: TAbstractContext);
begin
  //加入哈索表
  m_ClientContextTListLock.Lock;
  FClientContext.Put(IntToStr(DWORD(absContext)), absContext);
  m_ClientContextTListLock.UnLock;
end;

function TCIOCPServer.PostWSACloseSocketProcess(absContext: TAbstractContext): Boolean;
begin
  absContext.FRecvBuffer.SetOperation(IOWSACloseSocket);
  absContext.FRecvBuffer.SetupRead;
  PostQueuedCompletionStatus( m_hCompletionPort,
			                        0,
									            DWORD(absContext),
									            @absContext.FRecvBuffer.FPerHandleData );
end;

function TCIOCPServer.PostWSARecvProcess(absContext: TAbstractContext): Boolean;
var
  nRet: integer;
  dwBytes: DWORD;
  dwFlags: DWORD;
begin
  if Assigned(absContext) then begin
     absContext.FContextLock.Lock;
     absContext.FRecvBuffer.SetOperation(IOWSARecv);
     absContext.FRecvBuffer.SetupRead;


     dwBytes := 0;
     dwFlags := 0;
     nRet := WSARecv( absContext.FSocket,
                      @absContext.FRecvBuffer._WSABuf,
                      1,
                      dwBytes,
                      dwFlags,
                      @absContext.FRecvBuffer.FPerHandleData,
                      nil);

     if( ( nRet = SOCKET_ERROR ) and ( WSAGetLastError() <> WSA_IO_PENDING ) ) then begin
			  PostWSACloseSocketProcess(absContext);
			  Result := FALSE;
        Exit;
     end;

     absContext.FContextLock.UnLock;
  end;

  Result := TRUE;
end;

function TCIOCPServer.OnWSARecv(absContext: TAbstractContext; AbsBuffer: TAbstractBuffer; dwIoSize: Cardinal): Boolean;
begin

end;

procedure TCIOCPServer.DelContextToClientHasTable(absContext: TAbstractContext);
begin
  //加入哈索表
  m_ClientContextTListLock.Lock;
  FClientContext.Remove(IntToStr(DWORD(absContext)));
  m_ClientContextTListLock.UnLock;
end;

function TCIOCPServer.ProcessClientWithContext(m_Socket: TSocket): Boolean;
var
  nRet: Integer;
  bVal: LongBool;
  AbsContext: TAbstractContext;
begin
  //系统退出或设置未连接状态，则关闭客户端SOCKET
  if (m_bShutDown) or (not m_bAcceptConnections) then begin
    AppendMsgLog('m_bShutDown OR not m_bAcceptConnections, 关闭客户端连接.');
    Winsock2.closesocket(m_Socket);
    m_Socket := INVALID_SOCKET;
    Result := FALSE;
    Exit;
  end;

 	//是否超过最大连接数：
	if( m_iNumberOfActiveConnections > m_iMaxNumConnections ) then begin
    AppendMsgLog(Format('客户端已经达到最大连接数: %d, 关闭客户端连接.', [m_iMaxNumConnections]));
    Winsock2.closesocket(m_Socket);
    m_Socket := INVALID_SOCKET;
    Result := FALSE;
    Exit;
  end;

  if (m_Socket = INVALID_SOCKET) then
  begin
    Result := FALSE;
    Exit;
  end;

  //从空闲池中分配一个上下文
	AbsContext := AllocateFreeContextFromPool;
  if (AbsContext <> nil) then begin
  
    AbsContext.FSocket := m_Socket;
    //禁止nagle算法
    bVal := TRUE;
    nRet := Winsock2.setsockopt( AbsContext.FSocket,
                                 IPPROTO_TCP,
                                 TCP_NODELAY,
                                 PChar(@bVal),
                                 SizeOf(bVal) );

    if (nRet = SOCKET_ERROR) then begin
       AppendMsgLog(Format('setsockopt(禁止nagle算法) 错误: %d.',[WSAGetLastError()]));
       Winsock2.closesocket(m_Socket);
       m_Socket := INVALID_SOCKET;
       ReleaseClientContextToFreePool(AbsContext);
       Result := FALSE;
       Exit;
    end;

   	//客户上下文插入哈索表中。
		AddContextToClientHasTable(AbsContext);
    if AssociateSocketWithCompletionPort( AbsContext.FSocket,
                                          m_hCompletionPort,
                                          DWORD(Pointer(AbsContext))) then
    begin
      //投递Recv I/O 操作，接收客户端数据。
      PostWSARecvProcess(AbsContext);
      Result := TRUE;
    end
    else
    begin
      AppendMsgLog(Format('ProcessClientWithContext->ProcessClientWithContext失败: %d.',[WSAGetLastError()]));
      Winsock2.closesocket(m_Socket);
      m_Socket := INVALID_SOCKET;

      DelContextToClientHasTable(AbsContext);
      ReleaseClientContextToFreePool(AbsContext);
      Result := FALSE;
      Exit;
    end;

  end;
end;

function TCIOCPServer.OnWSAAcceptEx(AbsBuffer: TAbstractBuffer): Boolean;
begin
  //判断接受到的数据为零，则断开连接。
	if(AbsBuffer.FSocket = INVALID_SOCKET) then begin
    AppendMsgLog('OnWSAAcceptEx接受到dwTrans为零, 客户端关闭Socket连接.');
    WinSock2.closesocket(AbsBuffer.FSocket);
    AbsBuffer.FSocket := INVALID_SOCKET;
    Result := FALSE;
    Exit;
  end;

  //1.设置AcceptEx的Socket为ListenSocket状态。
	//2.生成客户端上下文(投递Recv I/O操作)。
	if ChangeSocketModeAccept(AbsBuffer.FSocket) then
		ProcessClientWithContext(AbsBuffer.FSocket);

  //投递下一个AcceptEx I/O操作
  PostWSAAcceptEx;
end;

function TCIOCPServer.ProcessIOMessage(absContext: TAbstractContext; AbsBuffer: TAbstractBuffer; dwIoSize: DWORD): Boolean;
begin
  case AbsBuffer.GetOperation of
	  IOWSAAcceptEx:
      begin
        OnWSAAcceptEx(AbsBuffer);
      end;
	  IOWSARecv:
      begin
        OnWSARecv(absContext, AbsBuffer, dwIoSize);
      end;
	  IOWSASend:
      begin

      end;
	  IOWSACloseSocket:
      begin

      end;
  end;
end;

function TCIOCPServer.Startup: Boolean;
var
  i: Integer;
  bRet: Boolean;
  workThread: TCIOCPWorkerThread;
begin
	AppendMsgLog(Format('IOCP服务器端启动, 系统版本号: %s.', [IOCP_SERVER_VERSION]));

  //创建停止服务内核事件对象
  m_hShutdownEvent := CreateEvent(Nil, FALSE, FALSE, Nil);
	if (m_hShutdownEvent = WSA_INVALID_EVENT) then begin
    AppendMsgLog(Format('WSACreateEvent()创建失败, 错误信息: %d.', [GetLastError()]));
    WSACloseEvent(m_hShutdownEvent);
		Result := FALSE;
    Exit;
  end;

	m_iNumberOfActiveConnections := 0;	//客户端连接个数置零
  if (m_iWSAInitResult <> NO_ERROR) then begin
    AppendMsgLog('Winsock 加载失败，系统只支持WinXP, Win2k and WinNT.');
		Result := FALSE;
    Exit;
  end
  else
    AppendMsgLog('Winsock 库加载成功.');

  if (not m_bServerStarted) then begin
    AppendMsgLog(Format('服务器允许客户端最大的连接数: %d.', [m_iMaxNumConnections]));

    //创建完成端口
		bRet := CreateCompletionPort();
		if (bRet) then
      AppendMsgLog('CreateCompletionport 创建成功.')
		else
    begin
      AppendMsgLog('系统异常错误，CreateCompletionport 创建失败.');
      Result := FALSE;
      Exit;
    end;

    //服务器端启动侦听
		bRet := ListnerStart();
		if(bRet) then
			AppendMsgLog('服务器端启动套接字侦听.')
    else
    begin
      AppendMsgLog('系统异常错误，服务器端侦听线程启动失败.');
      Result := FALSE;
      Exit;
    end;

    //创建服务器端启动工作线程
    SetLength(m_ThreadParams, m_iMaxIOCPIOWorkers);
    for i := 0 to m_iMaxIOCPIOWorkers - 1 do
    begin
      workThread := TCIOCPWorkerThread.Create(self);
      m_ThreadParams[i].nThread := i+1;
      m_ThreadParams[i].pCIOCPServer := Self;
      m_ThreadParams[i].hEvent := workThread.Handle;
    end;
    AppendMsgLog(Format('服务器启动工作线程成功个数: %d.', [m_iMaxIOCPIOWorkers]));

		m_bServerStarted := TRUE;
  end;

	if (bRet) then begin
    AppendMsgLog(Format('IOCP服务器端侦听地址: %s, 端口号:%d.', [GetHostIPAddr(), m_nPort]));
    AppendMsgLog('IOCP服务器端启动成功.');
  end;

	Result := TRUE;
end;

procedure TCIOCPServer.ShutDown();
begin

end;

end.