unit_connections.cpp

一个完成端口的框架程序

//---------------------------------------------------------------------------


#pragma hdrstop

#include "Unit_Connections.h"

//---------------------------------------------------------------------------

#pragma package(smart_init)
/*For Client(s) */
//accept a new communication
//根据给定Accept来的Socket，以及用户IP和端口创建一个用户连接上下文信息
bool __fastcall TtcpIOCP::AddConnection(const SOCKET sockAccept,
                                        const int RemotePort,
                                        const char *RemoteAddress)
{
   if(sockAccept==INVALID_SOCKET)
      return false;
   if(!FIOCPInvalid)
      return false;
   if(ValidConnections>=FMaxConnections)
      return false;
   LPCONN_CTX pCTX=(LPCONN_CTX) AllocEx(sizeof(CONN_CTX));
   if(!pCTX)
      return false;
   pCTX->sockAccept=sockAccept;
   pCTX->RemotePort=RemotePort;
   pCTX->O_DATA=NULL;
   pCTX->TotalSent=0;
   pCTX->TotalRecv=0;
   pCTX->InDumping=false;
   DWORD Socket_Error;
   int iAddressLen=strlen(RemoteAddress);
   if(iAddressLen>16)iAddressLen=16;
   strncpy(pCTX->RemoteAddr,
           RemoteAddress,
           iAddressLen);

   pCTX->I_OverLappedEx=(LPOVERLAPPEDEX)AllocEx(sizeof(OVERLAPPEDEX));
   if(!pCTX->I_OverLappedEx)
   {
      FreeEx(pCTX);
      pCTX=NULL;
      return false;
   }
   pCTX->O_OverLappedEx=(LPOVERLAPPEDEX)AllocEx(sizeof(OVERLAPPEDEX));
   if(!pCTX->O_OverLappedEx)
   {
      FreeEx(pCTX->I_OverLappedEx);
      pCTX->I_OverLappedEx=NULL;
      FreeEx(pCTX);
      pCTX=NULL;
      return false;
   }
   pCTX->I_OverLappedEx->OverLapped.hEvent=NULL;
   pCTX->I_OverLappedEx->OverLapped.Internal=0;
   pCTX->I_OverLappedEx->OverLapped.InternalHigh=0;
   pCTX->I_OverLappedEx->OverLapped.Offset=0;
   pCTX->I_OverLappedEx->OverLapped.OffsetHigh=0;
   pCTX->I_OverLappedEx->wbuf.buf=(char *)pCTX->I_OverLappedEx->data;
   pCTX->I_OverLappedEx->wbuf.len=MAX_BUF_LEN;
   pCTX->I_OverLappedEx->oper=SVR_IO_READ;
   pCTX->I_OverLappedEx->DataInExtend=false;
   pCTX->I_OverLappedEx->SentBytes=0;
   pCTX->I_OverLappedEx->Length=0;
   pCTX->I_OverLappedEx->LockCount=1;
   #ifdef __DEBUG
   pCTX->I_OverLappedEx->WorkingThreadID=GetCurrentThreadId();
   #endif
   pCTX->I_OverLappedEx->flags=0;

   //ZeroMemory(pCTX->O_OverLappedEx,sizeof(OVERLAPPEDEX));
   pCTX->O_OverLappedEx->OverLapped.hEvent=NULL;
   pCTX->O_OverLappedEx->OverLapped.Internal=0;
   pCTX->O_OverLappedEx->OverLapped.InternalHigh=0;
   pCTX->O_OverLappedEx->OverLapped.Offset=0;
   pCTX->O_OverLappedEx->OverLapped.OffsetHigh=0;
   pCTX->O_OverLappedEx->wbuf.buf=(char *)pCTX->I_OverLappedEx->data;
   pCTX->O_OverLappedEx->wbuf.len=MAX_BUF_LEN;
   pCTX->O_OverLappedEx->oper=SVR_IO_WRITE;
   pCTX->O_OverLappedEx->flags=0;
   pCTX->O_OverLappedEx->LockCount=0;
   pCTX->O_OverLappedEx->DataInExtend=false;
   pCTX->O_OverLappedEx->SentBytes=0;
   pCTX->O_OverLappedEx->Length=0;

   InitializeCriticalSection(&pCTX->ctx_CriticalSection);
   Lock();
   try
   {
      VTCCONN_CTX::iterator pConnection= FConnectionList.insert(FConnectionList.end(),pCTX);
      //将套接口与完成端口绑定
      if(!AssociateWithIoCompletionPort(FHandle,(HANDLE)sockAccept,(DWORD)pCTX))
      {
         FConnectionList.erase(pConnection);
         FreeEx(pCTX->I_OverLappedEx);
         pCTX->I_OverLappedEx=NULL;
         FreeEx(pCTX->O_OverLappedEx);
         pCTX->O_OverLappedEx=NULL;
         FreeEx(pCTX);
         pCTX=NULL;
         return false;
      }
   }
   __finally
   {
      Unlock();
   }
   //投递初始I/O操作
   pCTX->LastDataTime=time(0);
   int nResult=WSARecv(sockAccept,
                       &(pCTX->I_OverLappedEx->wbuf),
                       1,
                       NULL,
                       &(pCTX->I_OverLappedEx->flags),
                       &(pCTX->I_OverLappedEx->OverLapped),
                       NULL);

   if(nResult==SOCKET_ERROR)
   {
      Socket_Error=WSAGetLastError();
      if(Socket_Error!=ERROR_IO_PENDING)
      {
         //pCTX->I_OverLappedEx->LockCount=0;
         RemoveConnection(pCTX,FConnectionList);
         return false;
      }
   }
   try
   {
      InternalOnConnect(pCTX);
   }
   catch(...)
   {
   }
   SetExpectThread(ExpectThreadCount);
   pCTX=NULL;
   return true;
}

//disconnect a communication
//将一个用户连接上下文信息删除
bool __fastcall TtcpIOCP::RemoveConnection(LPCONN_CTX Connection,
                                           VTCCONN_CTX ConnectionList)
{
   bool Result=false;
   if(NULL==Connection)
      return Result;
   Lock();
   try
   {
      VTCCONN_CTX::iterator pConn;
      pConn=find(ConnectionList.begin(),
                 ConnectionList.end(),
                 Connection);
      if(pConn!=ConnectionList.end())
      {
         InternalOnDisconnect(Connection);
         Result=true;
         InternalOnBeforeDisconnect(Connection);
         ForceCloseSocket(Connection->sockAccept,true);
         ConnectionList.erase(pConn);
      }
	  if(((0<Connection->I_OverLappedEx->LockCount)||
              (0<Connection->O_OverLappedEx->LockCount))&&
             (ThreadCount>=1))
	  {
            DumpConnection(Connection);
            return false;
	  }
      InternalFreeDataNode(Connection->O_DATA);
   }
   __finally
   {
      Unlock();
   }
   DeleteCriticalSection(&Connection->ctx_CriticalSection);
   FreeEx(Connection->I_OverLappedEx);
   Connection->I_OverLappedEx=NULL;
   FreeEx(Connection->O_OverLappedEx);
   Connection->O_OverLappedEx=NULL;
   FreeEx(Connection);
   Connection=NULL;
   return Result;
}

//将一个用户连接置入回收站
void __fastcall TtcpIOCP::DumpConnection(LPCONN_CTX Connection)
{
   Lock();
   try
   {
      EnterCriticalSection(&conn_DUMP_CriticalSection);
      try
      {
         VTCCONN_CTX::iterator pConn;
         pConn=find(FConnectionList.begin(),
                    FConnectionList.end(),
                    Connection);
         if(pConn!=FConnectionList.end())
         {
            DWORD Connect_Time;
            int Size=sizeof(DWORD);
            Connection->InDumping=true;
            if((Connection->sockAccept!=INVALID_SOCKET)
               &&(getsockopt(Connection->sockAccept,
                             SOL_SOCKET,
                             0x700C/*SO_CONNECT_TIME*/,
                             (char *)&Connect_Time,
                             &Size)==SOCKET_ERROR))
            {/*取连接时间计数失败*/

               int Result=WSAGetLastError();
               if(Result!=WSAENOTSOCK)
               {
                  InternalOnBeforeDisconnect(Connection);
                  closesocket(Connection->sockAccept);
               }
            }
            //Connection->sockAccept=INVALID_SOCKET;
            InterlockedExchange(&(Connection->LastDataTime),time(0));
            FConnectionList.erase(pConn);            
            if((Connection->O_OverLappedEx->LockCount>0)||
               (Connection->I_OverLappedEx->LockCount>0))
               FrecycleConnectionList.insert(FrecycleConnectionList.end(),Connection);
            else
            {
               InternalOnDisconnect(Connection);
               InternalFreeDataNode(Connection->O_DATA);
               DeleteCriticalSection(&(Connection->ctx_CriticalSection));
               if(!FreeEx(Connection->I_OverLappedEx))
               {
                  FrecycleConnectionList.insert(FrecycleConnectionList.end(),Connection);
                  return;
               }
               Connection->I_OverLappedEx=NULL;
               if(!FreeEx(Connection->O_OverLappedEx))
               {
                  FrecycleConnectionList.insert(FrecycleConnectionList.end(),Connection);
                  return;
               }
               Connection->O_OverLappedEx=NULL;
               if(!FreeEx(Connection))
               {
                  FrecycleConnectionList.insert(FrecycleConnectionList.end(),Connection);
                  return;
               }
               Connection=NULL;
            }
         }
      }
      __finally
      {
         LeaveCriticalSection(&conn_DUMP_CriticalSection);
      }
   }
   __finally
   {
      Unlock();
   }
}

//disconnection all the communications
//清除所有有效用户连接
void __fastcall TtcpIOCP::CleanConnections(void)
{
   Lock();
   try
   {
	  if(FConnectionList.empty())
		 return;
          for(UINT Index=FConnectionList.size();Index>0;Index--)
              DumpConnection(FConnectionList.at(Index-1));
   }
   __finally
   {
      Unlock();
   }
}

//return communication node
//从用户连接列表当中取一用户连接上下文信息
LPCONN_CTX __fastcall TtcpIOCP::GetConnection(const UINT Index)
{
   Lock();
   try
   {
      if(FConnectionList.empty())
	  return NULL;
      if(Index>=FConnectionList.size())
	  return NULL;
      return FConnectionList.at(Index);
   }
   __finally
   {
      Unlock();
   }
   return NULL;
}

//释放数据节点所配置的内存
void __fastcall TtcpIOCP::InternalFreeDataNode(LPIODATANode DataNode)
{
   LPIODATANode InternalDataNode=DataNode;
   while(InternalDataNode)
   {
	  DataNode=InternalDataNode->Next;
	  InternalFreeIOData(InternalDataNode->Data);
	  FreeEx(InternalDataNode);
          InternalDataNode=NULL;
	  InternalDataNode=DataNode;
   }
}

//释放数据元素所配置的内存
void __fastcall TtcpIOCP::InternalFreeIOData(LPIODATA Data)
{
   LPIODATA InternalIOData=Data;
   while(InternalIOData)
   {
	  Data=InternalIOData->Next;
	  FreeEx(InternalIOData);
          InternalIOData=NULL;
	  InternalIOData=Data;
   }
}

//添加一个发送节点
void __fastcall TtcpIOCP::InternalAddDataNode(const LPCONN_CTX Connection,
                                              const LPIODATANode DataNode)
{
   if(NULL==Connection)
	  return;
   if(NULL==DataNode)
	  return;
   if(NULL==Connection->O_DATA)