用重叠IO的完成例程模型,在传送大文件时,服务器阻塞

Question

我自已写了一个完成例程的服务器,客户端在一般情况下,都是用短连接来连接服务器, 当有上传日志任务时,就用长连接来传送文件, 而且客户端在收到服务器应答后才发下一个报文.但不知道什么原因,我在传送大于50K
的文件时,服务器处理的报文为30K左右(这里的报文处理主要是把收到的报文写到文件中),服务器在而这时候我发现客户端
已发送的文件已有60多K.我个人认为不太可能是服务器来不及处理,因为客户端是在收到服务器的报文后才开始发一下报文的。
我又试了一个20K和40K的文件传输，发现都没有问题 这个问题已困扰我几天了,各位高手帮帮我呀,我把主要代码贴出来给大家
看一下.在主线程里启动监听线程和处理线程
AfxBeginThread(_ServerListenThread,this);    // 开始监听线程
AfxBeginThread(_OverlappedThread,this);  //启动处理线程//定义的结构
typedef struct
{
  WSAOVERLAPPED overlap;
  WSABUF        Buffer;
  char          szMessage[MSGSIZE];
  DWORD         NumberOfBytesRecvd;
  DWORD         Flags;
  SOCKET        sClient;
}PER_IO_OPERATION_DATA, *LPPER_IO_OPERATION_DATA;// 监听线程
UINT _ServerListenThread(LPVOID lParam)
{  
    //CServerSocket* pServer = (CServerSocket*)lParam;
    SOCKADDR_IN client;
    int  iaddrSize = sizeof(SOCKADDR_IN);
    while (TRUE)
   {
        // Accept a connection
        //sockListen为监听端口
        g_sNewClientConnection = accept(sockListen, (struct sockaddr *)&client, &iaddrSize);
        g_bNewConnectionArrived = TRUE;
   }
   return 0;
}// 重叠I/O处理线程
UINT _OverlappedThread(LPVOID lParam)
{     
    LPPER_IO_OPERATION_DATA lpPerIOData = NULL;
    while (TRUE)
    {
       if (g_bNewConnectionArrived)
       {
  // Launch an asynchronous operation for new arrived connection
          lpPerIOData = (LPPER_IO_OPERATION_DATA)HeapAlloc(
                                               GetProcessHeap(),
                                               HEAP_ZERO_MEMORY,
                                               sizeof(PER_IO_OPERATION_DATA));
            lpPerIOData->Buffer.len = MSGSIZE;
            lpPerIOData->Buffer.buf = lpPerIOData->szMessage;
            lpPerIOData->sClient = g_sNewClientConnection;
      
            WSARecv(lpPerIOData->sClient,
                     &lpPerIOData->Buffer,
                     1,
                     &lpPerIOData->NumberOfBytesRecvd,
                     &lpPerIOData->Flags,
                     &lpPerIOData->overlap,
                     CompletionROUTINE);
            g_bNewConnectionArrived = FALSE;
}
        SleepEx(1000, TRUE);
  }
  return 0;
}//完成例程
void CALLBACK CompletionROUTINE(DWORD dwError,
                                DWORD cbTransferred,
                                LPWSAOVERLAPPED lpOverlapped,
                                DWORD dwFlags)
{
    LPPER_IO_OPERATION_DATA lpPerIOData = (LPPER_IO_OPERATION_DATA)lpOverlapped;  
    if (dwError != 0 || cbTransferred == 0)
    {
// Connection was closed by client
        closesocket(lpPerIOData->sClient);
        HeapFree(GetProcessHeap(), 0, lpPerIOData);
    }
    else
   {
//进行业务处理把收到的报文写入文件中,并且组织应答报文-------------------------------------
workProcessReceiveBuf(lpPerIOData->szMessage,pServer);
        //发送应答报文
        send(lpPerIOData->sClient, lpPerIOData->szMessage, cbTransferred, 0)

        // Launch another asynchronous operation
        memset(&lpPerIOData->overlap, 0, sizeof(WSAOVERLAPPED));
        lpPerIOData->Buffer.len = MSGSIZE;
        lpPerIOData->Buffer.buf = lpPerIOData->szMessage;           WSARecv(lpPerIOData->sClient,
                &lpPerIOData->Buffer,
                1,
                &lpPerIOData->NumberOfBytesRecvd,
                &lpPerIOData->Flags,
                &lpPerIOData->overlap,
                CompletionROUTINE);
   }
}运行结果:
服务器收到最后一个报文的文件偏移量为30720字节，客户端读出最后一个报文的文件偏移量61952字节，但客户端最后一个
报文没有发送，好像死在那里了一样，而我用其他客户端能够发送报文给服务器，但服务器都没有对客户端进行应答。而这时
我关闭服务器，服务器出现异常， 而服务器退出时，客户端马上显示发送失败。请高手帮忙分析一下呀。PS： 我分数比较低，只能给10分，不好意思呀。

Answer 1

服务器端的WSASend提交得太多，导致核心运转缓慢无比，须对提交和完成的WSASend进行计数，并且根据输出完成状况进行限量。

Answer 2

把你的代码发给我，我帮你调试，[email protected]

Answer 3

是不是因为服务器来不及处理啊，我看你的代码很糟糕啊1，_OverlappedThread干吗用啊，而且是死循环，没有跳出循环的条件
2，_OverlappedThread里面lpPerIOData的sClient是怎么来的啊？
3，if(g_bNewConnectionArrived) 的BOOL 变量能保证同步关系吗？可能_ServerListenThread一下子收到好几个连接，会不会对中间连接进来的socket漏掉对其WSARecv的调用
4，半中半洋，接收用完成端口模型，发送却直接用阻塞的send了，
5，workProcessReceiveBuf是怎么处理的？会不会浪费太多时间

Answer 4

好的,zhoujinjun0858,我已给你发了邮件了.3，if(g_bNewConnectionArrived)   的BOOL   变量能保证同步关系吗？可能_ServerListenThread一下子收到好几个连接，会不会对中间连接进来的socket漏掉对其WSARecv的调用 我觉得楼上说得没错,这个存在问题. 我想用事件可能会好一点,现在正在调试事件.

Answer 5

我把最新修改的代码贴出来,希望高手帮忙分析一下.问题还是以前那样,现在是传到600K时就出现阻塞现象了,比之前的好一点了,但我觉得还是没有从根本上解决这个问题.
服务器刚启动,有客户端连接上来的图片:
客户端和服务器端传送文件阻塞时的图片:
客户端传送文件阻塞时的图片:
运行结果有个奇怪现象:
客户端连接上服务器后,服务器显示收到的内容,并给客户端应答,而客户端收到应答后,就马上断开连接,这时服务器照理说应该不会再去接收,直到客户端有新的连接触发时,才会再去接收,各位高手帮忙分析一下我的代码变量定义:
#define DATA_BUFSIZE 1024
#define DEFAULT_PORT 6069CWinThread* pServerListenThread = NULL;
CWinThread* pOverlappedThread = NULL;typedef struct _SOCKET_INFORMATION {
   OVERLAPPED Overlapped;
   SOCKET Socket;
   CHAR Buffer[DATA_BUFSIZE];
   WSABUF DataBuf;
   DWORD BytesSEND;  //发送标记
   DWORD BytesRECV;  //接收标记
} SOCKET_INFORMATION, * LPSOCKET_INFORMATION; void CALLBACK WorkerRoutine(DWORD Error, DWORD BytesTransferred,
   LPWSAOVERLAPPED Overlapped, DWORD InFlags);SOCKET sockListen,AcceptSocket;
CServerSocket* pServer;
WSAEVENT AcceptEvent;
主线程启动监听的程序:// 启动监听程序
int CServerSocket::StartListening()
{
m_nPort = DEFAULT_PORT;
WSADATA wsaData;
int nRet;
nRet=WSAStartup(MAKEWORD(2,2),&wsaData);                      //开启winsock.dll
if(nRet!=0)
{
AfxMessageBox("Load winsock2 failed");
WSACleanup();
return -1;
} sockListen = socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);         //创建服务套接字(流式) SOCKADDR_IN ServerAddr;                                       //分配端口及协议族并绑定 ServerAddr.sin_family=AF_INET;                                
ServerAddr.sin_addr.S_un.S_addr=inet_addr("10.8.205.239");
ServerAddr.sin_port=htons(DEFAULT_PORT); nRet=bind(sockListen,(LPSOCKADDR)&ServerAddr,sizeof(ServerAddr)); // 绑定套接字 if(nRet==SOCKET_ERROR)
{
AfxMessageBox("Bind Socket Fail!");
closesocket(sockListen);
return -1;
} nRet=listen(sockListen,5);    //开始监听，并设置监听客户端数量
if(nRet==SOCKET_ERROR)     
{
AfxMessageBox("Listening Fail!");
return -1;
}
CString strSock;
strSock.Format("服务器 %d 端口开始监听...",DEFAULT_PORT );
::SendMessage(this->m_hNotifyWnd,WM_MSG_NEW_SOCKET,
(LPARAM)(LPCTSTR)strSock,(LPARAM)(LPCTSTR)" "); AcceptEvent = WSACreateEvent();
pServer = this;
pServerListenThread = AfxBeginThread(_ServerListenThread,this);    // 开始监听线程
pOverlappedThread = AfxBeginThread(_OverlappedThread,(LPVOID)AcceptEvent); //工作线程 return 0;
}// 监听线程
UINT _ServerListenThread(LPVOID lParam)
{
while(TRUE)
{
 AcceptSocket = accept(sockListen, NULL, NULL);
         if (WSASetEvent(AcceptEvent) == FALSE)
 {
 printf("WSASetEvent failed with error %d\n", WSAGetLastError());
             return 0;
 }
}
return 0;
}// 重叠I/O处理线程
UINT _OverlappedThread(LPVOID lParam)
{
   DWORD Flags;
   LPSOCKET_INFORMATION SocketInfo;
   WSAEVENT EventArray[1];
   DWORD Index;
   DWORD RecvBytes;   // Save the accept event in the event array.
   EventArray[0] = (WSAEVENT) lParam;   while(TRUE)
   {
      // Wait for accept() to signal an event and also process WorkerRoutine() returns.
      while(TRUE)
      {
         Index = WSAWaitForMultipleEvents(1, EventArray, FALSE, WSA_INFINITE, TRUE);         if (Index == WSA_WAIT_FAILED)
         {
            printf("WSAWaitForMultipleEvents failed with error %d\n", WSAGetLastError());
            return FALSE;
         }         if (Index != WAIT_IO_COMPLETION)
         {
            // An accept() call event is ready - break the wait loop
            break;
         } 
      }      WSAResetEvent(EventArray[Index - WSA_WAIT_EVENT_0]);
   
      // Create a socket information structure to associate with the accepted socket.      if ((SocketInfo = (LPSOCKET_INFORMATION) GlobalAlloc(GPTR,
         sizeof(SOCKET_INFORMATION))) == NULL)
      {
         printf("GlobalAlloc() failed with error %d\n", GetLastError());
         return FALSE;
      }       // Fill in the details of our accepted socket.      SocketInfo->Socket = AcceptSocket;
      ZeroMemory(&(SocketInfo->Overlapped), sizeof(WSAOVERLAPPED));  
      SocketInfo->BytesSEND = 0;
      SocketInfo->BytesRECV = 0;
      SocketInfo->DataBuf.len = DATA_BUFSIZE;
      SocketInfo->DataBuf.buf = SocketInfo->Buffer;      Flags = 0;
      if (WSARecv(SocketInfo->Socket, &(SocketInfo->DataBuf), 1, &RecvBytes, &Flags,
         &(SocketInfo->Overlapped), WorkerRoutine) == SOCKET_ERROR)
      {
         if (WSAGetLastError() != WSA_IO_PENDING)
         {
            printf("WSARecv() failed with error %d\n", WSAGetLastError());
            return FALSE;
         }
      }
    else
    {
          ::SendMessage(pServer->m_hNotifyWnd,WM_MSG_NEW_MSG,
    (LPARAM)(LPCTSTR)"服务器收到的报文: ",(LPARAM)(LPCTSTR)SocketInfo->DataBuf.buf);
 
    }
   }   return TRUE;
 
     
}void CALLBACK WorkerRoutine(DWORD Error, DWORD BytesTransferred,
   LPWSAOVERLAPPED Overlapped, DWORD InFlags)
{
   DWORD SendBytes, RecvBytes;
   DWORD Flags;
   // Reference the WSAOVERLAPPED structure as a SOCKET_INFORMATION structure
   LPSOCKET_INFORMATION SI = (LPSOCKET_INFORMATION) Overlapped;       if (Error != 0)
   {
       ::SendMessage(pServer->m_hNotifyWnd,WM_MSG_NEW_MSG,
            (LPARAM)(LPCTSTR)"IO操作失败: ",(LPARAM)(LPCTSTR)"ErrorCode!=0");
   closesocket(SI->Socket);
       GlobalFree(SI);
       return;
   }   if (BytesTransferred == 0)
   {
      ::SendMessage(pServer->m_hNotifyWnd,WM_MSG_NEW_MSG,
            (LPARAM)(LPCTSTR)"客户端断开连接 ",(LPARAM)(LPCTSTR)" ");
  closesocket(SI->Socket);
      GlobalFree(SI);
      return;
   }
   // Check to see if the BytesRECV field equals zero. If this is so, then
   // this means a WSARecv call just completed so update the BytesRECV field
   // with the BytesTransferred value from the completed WSARecv() call.   if(SI->BytesRECV==0)
   {
   SI->BytesRECV=1;
   SI->BytesSEND=0;
   }
   else
   {
   SI->BytesSEND=1;
   SI->BytesRECV=0;
   }
      if (SI->BytesRECV > SI->BytesSEND)
   {      // Post another WSASend() request.
      // Since WSASend() is not gauranteed to send all of the bytes requested,
      // continue posting WSASend() calls until all received bytes are sent.     //处理接收到的报文,并把要发送的报文的值传回给输入输出参数SI->DataBuf.buf
      pServer->workProcessReceiveBuf(SI->DataBuf.buf,pServer);      ZeroMemory(&(SI->Overlapped), sizeof(WSAOVERLAPPED));
      SI->DataBuf.len = strlen(SI->DataBuf.buf);   //在对话框中显示
  ::SendMessage(pServer->m_hNotifyWnd,WM_MSG_NEW_MSG,
   (LPARAM)(LPCTSTR)"服务器发送的报文: ",(LPARAM)(LPCTSTR)SI->DataBuf.buf);
 
      if (WSASend(SI->Socket, &(SI->DataBuf), 1, &SendBytes, 0,
         &(SI->Overlapped), WorkerRoutine) == SOCKET_ERROR)
      {
         if (WSAGetLastError() != WSA_IO_PENDING)
         {
            printf("WSASend() failed with error %d\n", WSAGetLastError());
            return;
         }
      }

  
   }
   else
   {
  
      //SI->BytesRECV = 0;
      // Now that there are no more bytes to send post another WSARecv() request.
      Flags = 0;
      ZeroMemory(&(SI->Overlapped), sizeof(WSAOVERLAPPED));
      SI->DataBuf.len = DATA_BUFSIZE;
      SI->DataBuf.buf = SI->Buffer;      //在对话框中显示
  ::SendMessage(pServer->m_hNotifyWnd,WM_MSG_NEW_MSG,
              (LPARAM)(LPCTSTR)"服务器收到的报文2: ",(LPARAM)(LPCTSTR)SI->DataBuf.buf);
        if (WSARecv(SI->Socket, &(SI->DataBuf), 1, &RecvBytes, &Flags,
         &(SI->Overlapped), WorkerRoutine) == SOCKET_ERROR)
      {
         if (WSAGetLastError() != WSA_IO_PENDING )
         {
            printf("WSARecv() failed with error %d\n", WSAGetLastError());
            return;
         }
      }   
   }
}

Answer 6

我现在出差外面，人家电脑太烂,不能安装vc++ 2005,我郁闷,你里面有个什么:
  #import  <msxml4.dll> 
  using namespace MSXML2;//此命名空间在VC.net下面会有冲突，需要重命名
在vc++ 6.0下编译不通过~~不过我看是不是你重新投递IO的时候没有重新设置事件对象啊?
要用ResetEvent(&lpPerIOData-> overlap.hEvent) 吧?不然你下次投递IO的时候马上就返回了然后有可能会漏掉数据哦

Answer 7

不会吧,我的是就是在VC++ 6.0下编译的呀. 
哦,对了,你要安装一个MSXML 4.0 SP2.msi才行. 

Answer 8

我可不是逃避责任啊,真的，我现在外面，等我回去了一定给你个答复,
不过你这么着急,我也建议你可以在投递IO的时候加上判断啊
ret = WSARecv(lpPerIOData-> sClient, 
                  &lpPerIOData-> Buffer, 
                  1, 
                  &lpPerIOData-> NumberOfBytesRecvd, 
                  &lpPerIOData-> Flags, 
                  &lpPerIOData-> overlap, 
                  CompletionROUTINE); if(ret == WSA_WAIT_FAILED)
{
//处理错误
}
if(ret == WSA_WAIT_TIMEOUT)
{
//处理超时
}
if(ret == WSA_IO_PENDING)
{
 //表示IO正在进行中
}

Answer 9

好的，我马上就去下载 MSXML   4.0   SP2.msi

Answer 10

运行结果有个奇怪现象: 
客户端连接上服务器后,服务器显示收到的内容,并给客户端应答,而客户端收到应答后,就马上断开连接,这时服务器照理说应该不会再去接收,直到客户端有新的连接触发时,才会再去接收,各位高手帮忙分析一下我的代码这个原因是不是断开的时候在 _OverlappedThread(LPVOID   lParam) 里面出发了事件 WSAWaitForMultipleEvents(1,   EventArray,   FALSE,   WSA_INFINITE,   TRUE); 而你又当成是ACCEPT进来一个socket处理
给楼主几点意见：
1，_OverlappedThread里面，while(1)里面再一个while(1)不可取
   连接完全也可以用重叠模式(用AcceptEx)处理，而且可控制性更强，比如你Accept进一个连接后，可以再投递一个Accept，保持在等待Accept的socket有好几个（根据实际需求定），在并发量高的时候很好用，完成端口就高效的引进这在种方式
2，象上面说的，你确认Accept的方式，肯定是个bug，很大的bug
3，你完全可以把Accept,send,recv封装成一个接口PostAccept,PostSend,PostRecv,不仅仅是养成代码风格好习惯，对自己调式，查问题的时候很有帮助
4，      if(SI-> BytesRECV==0) 
      { 
      SI-> BytesRECV=1; 
      SI-> BytesSEND=0; 
      } 
      else 
      { 
      SI-> BytesSEND=1; 
      SI-> BytesRECV=0; 
完全可以不用这种方式，
typedef   struct   _SOCKET_INFORMATION   { 
      OVERLAPPED   Overlapped; 
      SOCKET   Socket; 
      CHAR   Buffer[DATA_BUFSIZE]; 
      WSABUF   DataBuf; 
      int      OP_TYPE
#define OP_ACCEPT 1
#define OP_READ 2
#define OP_WRITE 3
}   SOCKET_INFORMATION,   *   LPSOCKET_INFORMATION;在PostAccept里把OP_TYPE置成OP_ACCEPT，在PostSend,PostRecv里面设置，你应该明白意思了再在WorkRotine里面，根据OP_TYPE判断是Accept,send,recv完成了，进行相应处理就行了

Answer 11

 我要纠正我刚才说的话，完成例程是不用设置Event事件的,主要是好久都没有搞这个了,有点忘记了,刚才查了一下资料，发现重叠IO才需要设置OVERLAPPED的hEvent对象,原谅~~~ lz,你的程序我已经帮你调试了,我觉得你的发送和接受数据的思路是没有问题的，而且单就你的完成例程的代码也是正确的，但是你的协议在实现方面有问题，不协调，你好好看看,而且好象你的服务器在接收完文件以后没有关闭该SOCKET,导致服务器出现发送端口错误~~,因为客户端已经关闭该SOCKET了!
   你是用收到0byte来表示对方已经关闭,这个不行,书上是这样举例的，但是实际当中不可靠,要象上面那位说的那样来自己定协议来进行关闭，而且为了防止客户端莫名其妙的断开你还必须用心跳包的方法,专门启动一个线程定时发送信息给客户端,客户端也有一个线程专门应答,如果客户端不答应就表示对方关闭了~~   你的客户端在读取并发送文件方面好象也有问题，我只不过放了一个609字节的.zip文件在里面，居然传半天都传不完，而且客户端一调试就死掉，你好好找找原因~~  接收客户端连接方面我觉得可以这样搞
// 监听线程
UINT _ServerListenThread(LPVOID lParam)
{
     CString strSock="";
     CServerSocket* pServer = (CServerSocket*)lParam;
     SOCKADDR_IN client;
     int         iaddrSize = sizeof(SOCKADDR_IN); while (!bStopAccept) //用一个bStopAccept变量来控制是否需要继续监听
{
// 本来用AcceptEx()来搞最好，但是你如果客户量不是很大就不用了
        g_sNewClientConnection = accept(sockListen, (struct sockaddr *)&client, &iaddrSize);
        InitNewConnection(g_sNewClient); //直接用一个函数来处理新的连接,当同时连接的客户非常多的时候会 
                                //出现处理不过来,但是一般情况下完全可以，更不会象你那样造成完全可以连接的
                                //也连接不上  
}
return 0;
}//专门处理新连接的函数
void InitNewConnection(SOCKET new_sock)
{
      lpPerIOData = (LPPER_IO_OPERATION_DATA)HeapAlloc(
                                               GetProcessHeap(),
                                               HEAP_ZERO_MEMORY,
                                               sizeof(PER_IO_OPERATION_DATA));
       lpPerIOData->Flags=0;//**************
       lpPerIOData->Buffer.len = MSGSIZE;
       lpPerIOData->Buffer.buf = lpPerIOData->szMessage;
       lpPerIOData->sClient = g_sNewClientConnection;
           
       if(WSARecv(lpPerIOData->sClient,
                     &lpPerIOData->Buffer,
                     1,
                     &lpPerIOData->NumberOfBytesRecvd,
                     &lpPerIOData->Flags,
                     &lpPerIOData->overlap,
                     CompletionROUTINE)==SOCKET_ERROR)
       {
  if (WSAGetLastError() != WSA_IO_PENDING)
  {
             return 0;
  }
}
else
{
  ::SendMessage(pServer->m_hNotifyWnd,WM_MSG_NEW_MSG,
              (LPARAM)(LPCTSTR)"服务器收到的报文2: ",
                    (LPARAM)(LPCTSTR)PerIOData_>Buffer.buf);
        }    
}
  而且说实在的，你的代码挺混乱的，我觉得你的C++基础实在有问题，你还没有理解面向对象的概念，你照搬书上的例子,不可取,那个例子太简单化了,实际的商业程序哪里能这样搞啊~~

Answer 12

调试你的程序也很痛苦，特别是你的客户端,我一调试就死掉，我觉得你的问题在客户端的多一点，好好检查你的客户端吧~~
你可以先把的文件弄简单一点，不要搞那么多格式，就是一般的文本传输你看看行不行,等文本传输没有问题了就表示你的框架没有问题了,然后再把那些烂七八糟的格式放进去,而且你要多用WSAGetLastErr()函数来查看错误代码

Answer 13

这么难吗？代码发我一份看看,在上面说了这么久，我想如果这些时间我给你做一个的话，可能都已经做好了
[email protected]

Answer 14

好的.已经搞定了. 原来是客户端存在问题, 客户端接收的缓冲区太小了. 把它改成大一点后,就Ok了.
在此非常感谢jourbin兄和zhoujinjun0858兄给的帮助.