unit_connections.cpp

一个完成端口的框架程序

   {
	  Connection->O_DATA=DataNode;
	  return;
   }
   LPIODATANode InternalDataNode=Connection->O_DATA;
   while(InternalDataNode->Next)
   {
	  InternalDataNode=InternalDataNode->Next;
   }
   InternalDataNode->Next=DataNode;
   return;
}

/*For Client(s) */
//断开一个用户连接，并将相关上下文信息置回收站，以便于回收相关的资源
bool __fastcall TtcpIOCP::DropConnection(LPCONN_CTX Connection)
{
   if(NULL==Connection)
      return false;
   Lock();
   try
   {
      if(Connection->sockAccept==INVALID_SOCKET)
         return false;
      DumpConnection(Connection);
      return true;
   }
   __finally
   {
      Unlock();
   }
   return false;
}

//send text message to a communication node
//发送一个字符串
bool __fastcall TtcpIOCP::SendText(const LPCONN_CTX Connection,
                                   const char Message[])
{
   int Length=strlen(Message);
   if(0>=Length)
      return false;
   return SendBuffer(Connection,
                     Message,
                     Length);
}

//send binary message(s) to a communication node
//发送一段缓冲区
bool __fastcall TtcpIOCP::SendBuffer(const LPCONN_CTX Connection,
                                     const char *Buffer,
                                     const UINT Length)
{
   if(0==Length)
      return false;
   if(NULL==Connection)
      return false;
   int nResult;
   bool InternalSending=false;
   LPOVERLAPPEDEX  lpPerIOData;
   lpPerIOData=Connection->O_OverLappedEx;
   if((Length>MAX_BUF_LEN)||(true==lpPerIOData->DataInExtend)||(lpPerIOData->LockCount>0))
   {
      LPIODATANode DataNode;
      LPIODATA PreData;
      DataNode=(LPIODATANode)AllocEx(sizeof(IODATANode));
      if(NULL==DataNode)
         return false;
      DataNode->Next=NULL;
      LPIODATA Data;
      Data=(LPIODATA)AllocEx(sizeof(IODATA));
      if(NULL==Data)
      {
         InternalFreeDataNode(DataNode);
         return false;
      }
      DataNode->Data=Data;
      int InternalLength=Length;
      int InternalBufferSize=(InternalLength>MAX_BUF_LEN
                              ?MAX_BUF_LEN
                              :InternalLength);
      int InternalPointer=0;
      CopyMemory(Data->Buffer,
                 Buffer,
                 InternalBufferSize);
      Data->Length=InternalBufferSize;
      InternalPointer+=InternalBufferSize;
      InternalLength-=InternalBufferSize;
      PreData=Data;
      while(InternalLength>0)
      {
         LPIODATA InternalData;
         InternalData=(LPIODATA)AllocEx(sizeof(IODATA));
         if(NULL==InternalData)
         {
            InternalFreeDataNode(DataNode);
            return false;
         }
         ZeroMemory(InternalData,sizeof(IODATA));
         PreData->Next=InternalData;
         InternalBufferSize=(InternalLength>MAX_BUF_LEN
                             ?MAX_BUF_LEN
                             :InternalLength);
         CopyMemory(InternalData->Buffer,
                    Buffer+InternalPointer,
                    InternalBufferSize);
         InternalData->Length=InternalBufferSize;
         InternalPointer+=InternalBufferSize;
         InternalLength-=InternalBufferSize;
         PreData=InternalData;
      }
      return SendData(Connection,DataNode);
   }
   
   /*为保证所操作的连接有效性              */
   /*先进入连接列表临界区，查看连接有效性，*/
   /*无效则不再操作，有效则增加引用计数，  */
   /*防止作业过程当中被删除                */
   Lock();
   try
   {
      if(FConnectionList.end()==find(FConnectionList.begin(),FConnectionList.end(),Connection))
      {
         //该连接已失效，返回操作失败
         return false;
      }
      /*增加引用计数*/
      InterlockedIncrement(&(Connection->O_OverLappedEx->LockCount));
   }
   __finally
   {
      Unlock();
   }
   EnterCriticalSection(&(Connection->ctx_CriticalSection));
   try
   {
      lpPerIOData->OverLapped.hEvent=NULL;
      lpPerIOData->OverLapped.Internal=0;
      lpPerIOData->OverLapped.InternalHigh=0;
      lpPerIOData->OverLapped.Offset=0;
      lpPerIOData->OverLapped.OffsetHigh=0;
      lpPerIOData->wbuf.buf=(char*)&(lpPerIOData->data);
      lpPerIOData->wbuf.len=Length;
      CopyMemory(lpPerIOData->data,Buffer,Length);
      lpPerIOData->oper=SVR_IO_WRITE;
      lpPerIOData->DataInExtend=true;
      lpPerIOData->Length=Length;
      lpPerIOData->SentBytes=0;
      lpPerIOData->flags=0;
      #ifdef __DEBUG
      lpPerIOData->WorkingThreadID=GetCurrentThreadId();
      #endif
      InternalSending=true;
      InterlockedExchange(&(Connection->LastDataTime),time(0));
      nResult=WSASend(Connection->sockAccept,
                      &(lpPerIOData->wbuf),
                      1,
                      NULL,
                      lpPerIOData->flags,
                      &(lpPerIOData->OverLapped),
                      NULL);
   }
   __finally
   {
      LeaveCriticalSection(&(Connection->ctx_CriticalSection));
      if(InternalSending)
         if(nResult==SOCKET_ERROR)
         {
            int Socket_Error=WSAGetLastError();
            if(Socket_Error!=ERROR_IO_PENDING&&Socket_Error!=WSAENOTSOCK)
            {
                                 #ifdef __ERRORLOG
                                 {
                                 char *Msg;
                                 Msg=(char *) AllocEx(256);
                                 strcpy(Msg,"ThrdID:");
                                 ltoa(GetCurrentThreadId(),(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|SocketError:");
                                 ltoa(Socket_Error,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|ThreadError:");
                                 ltoa(GetLastError(),(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|SOCKET:");
                                 itoa(Connection->sockAccept,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|Oper:SendBuffer");
                                 strcat(Msg,"|LockCount:");
                                 ltoa(lpPerIOData->LockCount,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|PerIOData:");
                                 ltoa((DWORD)lpPerIOData,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|ConnCtx:");
                                 ltoa((DWORD)Connection,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|O_OverLappedEx:");
                                 ltoa((DWORD)Connection->O_OverLappedEx,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|InternalSendBuffer:Failed");
                                 SendMessage(HWND_BROADCAST,RegisterWindowMessage("DumpMessage"),(WPARAM)Msg,(LPARAM)strlen(Msg));
                                 FreeEx(Msg);
                                 Msg=NULL;
                                 }
                                 #endif
               InterlockedDecrement(&(lpPerIOData->LockCount));
               lpPerIOData=NULL;
               DumpConnection(Connection);
               return false;
            }
         }
   }
   return true;
}

//send Structured message(s) to a communication node
//发送一个节点数据
bool __fastcall TtcpIOCP::SendData(const LPCONN_CTX Connection,
                                   const LPIODATANode DataNode)
{
   if(NULL==DataNode)
      return false;
   if(NULL==Connection)
   {
      InternalFreeDataNode(DataNode);
      return false;
   }
   int nResult;
   bool InternalSending=false;
   LPOVERLAPPEDEX  lpPerIOData;

   /*为保证所操作的连接有效性              */
   /*先进入连接列表临界区，查看连接有效性，*/
   /*无效则不再操作，有效则增加引用计数，  */
   /*防止作业过程当中被删除                */
   Lock();
   try
   {
      if(FConnectionList.end()==find(FConnectionList.begin(),FConnectionList.end(),Connection))
      {
         //该连接已失效，将数据节点资源释放掉并返回
         InternalFreeDataNode(DataNode);
         return false;
      }
      /*增加引用计数*/
      InterlockedIncrement(&(Connection->O_OverLappedEx->LockCount));
   }
   __finally
   {
      Unlock();
   }
   EnterCriticalSection(&(Connection->ctx_CriticalSection));
   try
   {
      //增加一个待发送节点
      InternalAddDataNode(Connection,DataNode);
      if(Connection->O_OverLappedEx->LockCount>1)
      {
         //在此之前仍有未决的WSASend，不再发出第二个WSASend
         InterlockedDecrement(&(Connection->O_OverLappedEx->LockCount));
         return true;
      }

      /*提交发送请求...*/
      lpPerIOData=Connection->O_OverLappedEx;
      //if(false==lpPerIOData->DataInExtend)
      //   ZeroMemory(lpPerIOData,sizeof(OVERLAPPEDEX));
      lpPerIOData->OverLapped.hEvent=NULL;
      lpPerIOData->OverLapped.Internal=0;
      lpPerIOData->OverLapped.InternalHigh=0;
      lpPerIOData->OverLapped.Offset=0;
      lpPerIOData->OverLapped.OffsetHigh=0;
      if(false==lpPerIOData->DataInExtend)
      {
         LPIODATANode InternalDataNode=Connection->O_DATA;
         LPIODATA InternalData=InternalDataNode->Data;
         lpPerIOData->wbuf.buf=(char *)(InternalData->Buffer+InternalData->SentBytes);
         lpPerIOData->wbuf.len=InternalData->Length-InternalData->SentBytes;
      }
      else
      {
         lpPerIOData->wbuf.buf=(char*)(lpPerIOData->data+lpPerIOData->SentBytes);
         lpPerIOData->wbuf.len=lpPerIOData->Length-lpPerIOData->SentBytes;
      }

      lpPerIOData->oper=SVR_IO_WRITE;
      lpPerIOData->flags=0;
      #ifdef __DEBUG
      lpPerIOData->WorkingThreadID=GetCurrentThreadId();
      #endif
      InternalSending=true;
      InterlockedExchange(&(Connection->LastDataTime),time(0));
      nResult=WSASend(Connection->sockAccept,
                      &(lpPerIOData->wbuf),
                      1,
                      NULL,
                      lpPerIOData->flags,
                      &(lpPerIOData->OverLapped),
                      NULL);
   }
   __finally
   {
      LeaveCriticalSection(&(Connection->ctx_CriticalSection));
      if(InternalSending)
         if(nResult==SOCKET_ERROR)
         {
            int Socket_Error=WSAGetLastError();
            if(/*Socket_Error!=WSAENOTSOCK&&*/Socket_Error!=ERROR_IO_PENDING)
            {
                                 #ifdef __ERRORLOG
                                 {
                                 char *Msg;
                                 Msg=(char *) AllocEx(256);
                                 strcpy(Msg,"ThrdID:");
                                 ltoa(GetCurrentThreadId(),(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|SocketError:");
                                 ltoa(Socket_Error,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|ThreadError:");
                                 ltoa(GetLastError(),(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|SOCKET:");
                                 itoa(Connection->sockAccept,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|Oper:SendData");
                                 strcat(Msg,"|LockCount:");
                                 ltoa(lpPerIOData->LockCount,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|PerIOData:");
                                 ltoa((DWORD)lpPerIOData,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|ConnCtx:");
                                 ltoa((DWORD)Connection,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|O_OverLappedEx:");
                                 ltoa((DWORD)Connection->O_OverLappedEx,(char *)(&Msg[0]+strlen(Msg)),10);
                                 strcat(Msg,"|InternalSendData:Failed");
                                 SendMessage(HWND_BROADCAST,RegisterWindowMessage("DumpMessage"),(WPARAM)Msg,(LPARAM)strlen(Msg));
                                 FreeEx(Msg);
                                 Msg=NULL;
                                 }
                                 #endif
               InterlockedDecrement(&(lpPerIOData->LockCount));
               DumpConnection(Connection);
               return false;
            }
         }
   }
   return true;
}