c++ - ERROR_IO_PENDING 后的多个 WriteFile

Question

在我的应用程序中，当另一个写入操作在命名管道上挂起时，我正在测试 WriteFile 的行为。

管道处于消息模式(不是字节模式)。

为了使写入操作挂起，我将命名管道的缓冲区保持得非常小，并且客户端发送的数据多于缓冲区大小。通过这种方式，我在客户端获得挂起的写操作。

我在服务器端遇到了以下问题:

ReadFile 失败，返回 ERROR_MORE_DATA。它会更改缓冲区的内容，但不会更改接收到的字节数。

服务器代码

//file server.cpp
#include <windows.h> 
#include <stdio.h>
#include <tchar.h>
#include <strsafe.h>

#define CONNECTING_STATE 0 
#define READING_STATE 1 
#define WRITING_STATE 2 
#define INSTANCES 4 
#define PIPE_TIMEOUT 5000
#define BUFSIZE 3
#define BUFSIZE2 1000

typedef struct 
{ 
   OVERLAPPED oOverlap; 
   HANDLE hPipeInst; 
   TCHAR chRequest[BUFSIZE2]; 
   DWORD cbRead;
   DWORD cbReadSoFar;
   TCHAR chReply[BUFSIZE2];
   DWORD cbToWrite; 
   DWORD dwState; 
   BOOL fPendingIO; 
} PIPEINST, *LPPIPEINST; 


VOID DisconnectAndReconnect(DWORD); 
BOOL ConnectToNewClient(HANDLE, LPOVERLAPPED); 
VOID GetAnswerToRequest(LPPIPEINST); 

PIPEINST Pipe[INSTANCES]; 
HANDLE hEvents[INSTANCES]; 

int _tmain(VOID) 
{ 
   DWORD i, dwWait, cbRet, dwErr; 
   BOOL fSuccess; 
   LPTSTR lpszPipename = TEXT("\\\\.\\pipe\\mynamedpipe"); 

   for (i = 0; i < INSTANCES; i++) 
   { 
      hEvents[i] = CreateEvent(NULL, TRUE, TRUE, NULL);

      Pipe[i].oOverlap.hEvent = hEvents[i]; 
      Pipe[i].hPipeInst = CreateNamedPipe(lpszPipename, PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,  
         PIPE_TYPE_MESSAGE|PIPE_READMODE_MESSAGE|PIPE_WAIT, INSTANCES,                
         BUFSIZE*sizeof(TCHAR), BUFSIZE*sizeof(TCHAR), PIPE_TIMEOUT, NULL);                   

      if (Pipe[i].hPipeInst == INVALID_HANDLE_VALUE) 
      {
         printf("CreateNamedPipe failed with %d.\n", GetLastError());
         return 0;
      }

      Pipe[i].fPendingIO = ConnectToNewClient(Pipe[i].hPipeInst, &Pipe[i].oOverlap); 
      Pipe[i].dwState = Pipe[i].fPendingIO ? CONNECTING_STATE : READING_STATE;
   } 

   while (1) 
   {
      dwWait = WaitForMultipleObjects(INSTANCES, hEvents, FALSE, INFINITE);

      i = dwWait - WAIT_OBJECT_0;  // determines which pipe 
      if (i < 0 || i > (INSTANCES - 1)) 
      {
         printf("Index out of range.\n"); 
         return 0;
      }

      if (Pipe[i].fPendingIO) 
      { 
         fSuccess = GetOverlappedResult(Pipe[i].hPipeInst, &Pipe[i].oOverlap, &cbRet, FALSE);

         DWORD lasterror = GetLastError();
         switch (Pipe[i].dwState) 
         { 
            case CONNECTING_STATE: 
               if (! fSuccess) 
               {
                   printf("Error %d.\n", GetLastError()); 
                   return 0;
               }
               Pipe[i].dwState = READING_STATE; 
               Pipe[i].cbReadSoFar = 0;
               break; 

            case READING_STATE: 
                if(ERROR_MORE_DATA == lasterror)
                {
                    Pipe[i].cbReadSoFar += cbRet;
                    break;
                }
                if(ERROR_IO_PENDING == lasterror)
                {
                    continue;
                }
                if (! fSuccess || cbRet == 0) 
                { 
                    DisconnectAndReconnect(i); 
                    continue; 
                }
                Pipe[i].cbRead = cbRet;
                Pipe[i].cbReadSoFar = 0;
                printf("Message received: %s\n", Pipe[i].chRequest);
                break;

            default: 
            {
               printf("Invalid pipe state.\n"); 
               return 0;
            }
         }  
      } 

      switch (Pipe[i].dwState) 
      { 
         case READING_STATE: 
            fSuccess = ReadFile( 
               Pipe[i].hPipeInst, 
               Pipe[i].chRequest + Pipe[i].cbReadSoFar, 
               BUFSIZE*sizeof(TCHAR), 
               &Pipe[i].cbRead, 
               &Pipe[i].oOverlap); 

            dwErr = GetLastError(); 
            if (! fSuccess && (dwErr == ERROR_IO_PENDING)) 
            { 
               Pipe[i].fPendingIO = TRUE; 
               continue; 
            } 

            if(!fSuccess && dwErr == ERROR_MORE_DATA)
            {
                Pipe[i].cbReadSoFar += Pipe[i].cbRead;
                continue;
            }
            if (fSuccess && Pipe[i].cbRead != 0) 
            { 
               Pipe[i].fPendingIO = FALSE;
               Pipe[i].cbReadSoFar = 0;
               //Log the message.
                printf("Message received: %s\n", Pipe[i].chRequest);
               continue; 
            }
         // An error occurred; disconnect from the client. 

            DisconnectAndReconnect(i); 
            break; 

         default: 
         {
            printf("Invalid pipe state.\n"); 
            return 0;
         }
      } 
  } 

  return 0; 
} 

VOID DisconnectAndReconnect(DWORD i) 
{ 
   if (! DisconnectNamedPipe(Pipe[i].hPipeInst) ) 
   {
      printf("DisconnectNamedPipe failed with %d.\n", GetLastError());
   }

   Pipe[i].fPendingIO = ConnectToNewClient( 
      Pipe[i].hPipeInst, 
      &Pipe[i].oOverlap); 

   Pipe[i].dwState = Pipe[i].fPendingIO ? 
      CONNECTING_STATE : // still connecting 
      READING_STATE;     // ready to read 
} 

BOOL ConnectToNewClient(HANDLE hPipe, LPOVERLAPPED lpo) 
{ 
   BOOL fConnected, fPendingIO = FALSE; 
   fConnected = ConnectNamedPipe(hPipe, lpo); 
   if (fConnected) 
   {
      printf("ConnectNamedPipe failed with %d.\n", GetLastError()); 
      return 0;
   }

   switch (GetLastError()) 
   { 
      case ERROR_IO_PENDING: 
         fPendingIO = TRUE; 
         break; 

      case ERROR_PIPE_CONNECTED: 
         if (SetEvent(lpo->hEvent)) 
            break; 

      default: 
      {
         printf("ConnectNamedPipe failed with %d.\n", GetLastError());
         return 0;
      }
   } 

   return fPendingIO; 
}

VOID GetAnswerToRequest(LPPIPEINST pipe)
{
   _tprintf( TEXT("[%d] %s\n"), pipe->hPipeInst, pipe->chRequest);
   StringCchCopy( pipe->chReply, BUFSIZE, TEXT("Default answer from server") );
   pipe->cbToWrite = (lstrlen(pipe->chReply)+1)*sizeof(TCHAR);
}

客户端代码是

//file: client.cpp
#include <windows.h> 
#include <stdio.h>
#include <conio.h>
#include <tchar.h>

#define BUFSIZE 512
#define NUM_MESSAGE 4
int _tmain(int argc, TCHAR *argv[]) 
{ 
   HANDLE hPipe; 
   LPTSTR lpvMessage=TEXT("Default message from the client."); 
   TCHAR  chBuf[NUM_MESSAGE+1][BUFSIZE]; 
   BOOL   fSuccess = FALSE; 
   DWORD  cbRead, cbToWrite, cbWritten, dwMode; 
   LPTSTR lpszPipename = TEXT("\\\\.\\pipe\\mynamedpipe"); 

   while (1) 
   { 
      hPipe = CreateFile(lpszPipename, GENERIC_READ|GENERIC_WRITE, 
         0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);

      if (hPipe != INVALID_HANDLE_VALUE) 
         break; 

      if (GetLastError() != ERROR_PIPE_BUSY) 
      {
         _tprintf( TEXT("Could not open pipe. GLE=%d\n"), GetLastError() ); 
         return -1;
      }

      if ( ! WaitNamedPipe(lpszPipename, 20000)) 
      { 
         printf("Could not open pipe: 20 second wait timed out."); 
         return -1;
      } 
   } 

   dwMode = PIPE_READMODE_MESSAGE; 
   fSuccess = SetNamedPipeHandleState(hPipe, &dwMode, NULL, NULL);
   if (!fSuccess) 
   {
    _tprintf(TEXT("SetNamedPipeHandleState failed. GLE=%d\n"), GetLastError()); 
    return -1;
   }


   cbToWrite = (lstrlen(lpvMessage)+1)*sizeof(TCHAR);
   _tprintf( TEXT("Sending %d byte message: \"%s\"\n"), cbToWrite, lpvMessage); 

   OVERLAPPED woverlapped[NUM_MESSAGE];
   HANDLE   wevent[NUM_MESSAGE];
   memset(woverlapped, 0, sizeof(woverlapped));
   for(int i=0; i<NUM_MESSAGE; ++i)
   {
    wevent[i]   = CreateEvent(NULL, 0, 0, NULL);
    woverlapped[i].hEvent = wevent[i];
   }

   const int retrycount = NUM_MESSAGE;
   int numberofsend = retrycount;   

wretry:   
   cbToWrite = _stprintf(chBuf[retrycount-numberofsend], TEXT("Message %d from Client"), retrycount - numberofsend + 1);
   cbToWrite *= sizeof(TCHAR);

   fSuccess = WriteFile(hPipe, chBuf[retrycount-numberofsend], cbToWrite,
      &cbWritten, &woverlapped[retrycount-numberofsend]);

    --numberofsend;
   if(!fSuccess) 
   {
    _tprintf( TEXT("WriteFile to pipe failed. GLE=%d\n"), GetLastError() ); 
    //return -1;
    if(GetLastError() != ERROR_IO_PENDING)
        return -1;  
   }

    if(numberofsend)
        goto wretry;

    bool wait = true;
    while(wait)
    {
        DWORD retval = WaitForMultipleObjects(NUM_MESSAGE, wevent, FALSE, INFINITE);

        retval -= WAIT_OBJECT_0;
        DWORD cbret;
        DWORD success = ::GetOverlappedResult(hPipe, &woverlapped[retval], &cbret, FALSE);
        DWORD laserr = ::GetLastError();
        ++numberofsend;

        if(numberofsend == retrycount)
            wait = false;
    }
   CloseHandle(hPipe); 

   return 0; 
}

在这种情况下，我在服务器端做错了吗？还是意料之中？

Answer 1

引用MSDN :

If a named pipe is being read in message mode and the next message is longer than the nNumberOfBytesToRead parameter specifies, ReadFile returns FALSE and GetLastError returns ERROR_MORE_DATA. The remainder of the message can be read by a subsequent call to the ReadFile or PeekNamedPipefunction.