io completion ports with named pipes

See more:

C++

C

Windows

The following code results in the "client" call to TransactNamedPipe() returning immediately with GetLastError() returning ERROR_IO_PENDING; all of which, I believe, is correct from the client's point of view. The client then proceeds to wait on the overlapped event.

However, the "server" listener thread never returns from its call to GetQueuedCompletionStatus(), and hence, never processes the client request.

Why isn't my client request making it thru to the listener?

Thanks in advance.

#include "stdafx.h"
#include <windows.h>
#define PIPE_NAME       _T( "\\\\.\\pipe\\pipe1" )
#define PIPE_TIMEOUT    5000
#define BUF_SIZE        4096
static DWORD listenerThreadProc( void *pData );
static void closeHandle( HANDLE &hObj );
HANDLE g_hPipe              = NULL;
HANDLE g_hIoCompletionPort  = NULL;
HANDLE g_hListenerThread    = NULL;
enum TIoState { iosRead = 0, iosWrite };

int _tmain( int argc, _TCHAR* argv[] )
{
  UNREFERENCED_PARAMETER( argc );
  UNREFERENCED_PARAMETER( argv );
  int         result      = 1; // 0 == success.
  DWORD       lastError   = 0;
  OVERLAPPED  overlapped  = {0};
  HANDLE      hPipe       =  NULL;
  HANDLE      hClientRequestEvent = NULL;

  ///////////////////////////////
  // Get the "server" running.
  ///////////////////////////////
  // Create the pipe.
  g_hPipe = CreateNamedPipe(
    PIPE_NAME,                  // pipe name
    PIPE_ACCESS_DUPLEX |        // read/write access
    FILE_FLAG_OVERLAPPED,       // overlapped mode
    PIPE_TYPE_MESSAGE |         // message-type pipe
    PIPE_READMODE_MESSAGE |     // message read mode
    PIPE_WAIT,                  // blocking mode
    PIPE_UNLIMITED_INSTANCES,   // unlimited instances
    BUF_SIZE,                   // output buffer size
    BUF_SIZE,                   // input buffer size
    PIPE_TIMEOUT,               // client time-out
    NULL );                     // default security attributes
  if ( INVALID_HANDLE_VALUE == g_hPipe )
  {
    _tprintf(_T( "Failed to create \"%s\" pipe. Error = %d." ),
             PIPE_NAME,
             GetLastError() );
    goto DONE;
  }
  // Create the io completion port and associate it to the pipe.
  g_hIoCompletionPort = CreateIoCompletionPort(
    g_hPipe,
    NULL,
    iosRead,
    0 );
  if ( NULL == g_hIoCompletionPort )
  {
    _tprintf(_T( "Failed to create the IO completion port. Error = %d." ), 
             GetLastError() );
    goto DONE;
  }
  // Start the listener thread.
  g_hListenerThread =
    CreateThread(
       NULL,                   // Default security attributes.
       0,                      // Default stack size.
       (LPTHREAD_START_ROUTINE)listenerThreadProc,
       (LPVOID)NULL,           // Thread data.
       0,                      // run immediately.
       NULL                    // Don't care about the thread id.
  );
  if ( NULL == g_hListenerThread )
  {
     _tprintf( _T( "Failed to create listener thread. Error = %d." ), GetLastError() );
     goto DONE;
  }
  
  // Give the listener thread a chance to get started.
  Sleep( 1000 );

  ////////////////////////////////
  // Get the "client" running.
  ////////////////////////////////
  // Open the pipe to the server.
  hPipe = CreateFile(
     PIPE_NAME,                      // pipe name
     GENERIC_READ | GENERIC_WRITE,   // read and write access
     0,                              // no sharing
     NULL,                           // default security attributes
     OPEN_EXISTING,                  // opens existing pipe
     FILE_ATTRIBUTE_NORMAL |
     FILE_FLAG_OVERLAPPED,           // asynchronous
     NULL );                         // no template file

  if ( INVALID_HANDLE_VALUE == hPipe )
  {
    _tprintf(_T( "Failed to open server pipe. Error = %d." ), 
             GetLastError() );
    goto DONE;
  }
  
  // The pipe connected; change to message-read mode.
  DWORD dwMode = PIPE_READMODE_MESSAGE;
  BOOL retVal = SetNamedPipeHandleState(
     hPipe,      // pipe handle
     &dwMode,    // new pipe mode
     NULL,       // don't set maximum bytes
     NULL );     // don't set maximum time

  if ( !retVal )
  {
    _tprintf(_T( "Failed to set server pipe mode. Error = %d." ),
             GetLastError() );
    goto DONE;
  }
  
  // Create the event to signal that the asynch io has completed.
  hClientRequestEvent = CreateEvent( NULL, TRUE, FALSE, NULL );
  if ( NULL == hClientRequestEvent )
  {
    _tprintf(_T( "Failed to create client event. Error = %d." ),
             GetLastError() );
     goto DONE;
  }
  
  // Send an asynchronous transaction to the pipe. This should return immediately.
  overlapped.hEvent = hClientRequestEvent;
  BYTE  response[35]  = {0};
  DWORD cbRead        = 0;
  retVal = TransactNamedPipe(
    hPipe,                  // pipe handle
    (LPVOID) _T( "aaa" ),   // message to server
    4 * sizeof( _TCHAR ),   // message length
    response,               // buffer to receive reply
    35,                     // size of read buffer
    &cbRead,                // bytes read
    &overlapped );          // overlapped
    
  // ...For this test client, I'm assuming that sizeDataOutBytes 
  // is sufficient for the expected response. Therefore, 
  // I'm not handling the ERROR_MORE_DATA failure case here.
  lastError = GetLastError();
  if ( !retVal && ERROR_IO_PENDING != lastError )
  {
    _tprintf(_T( "TransactNamedPipe() failed. Error = %d." ),
             lastError );
     goto DONE;
  }
  
  // Wait for the transaction to finish.
  bool clientSuccess = false;
  DWORD dwResult = WaitForSingleObject( hClientRequestEvent, 3000 );
  switch ( dwResult )
  {
    case WAIT_OBJECT_0 : clientSuccess = true; break;
    case WAIT_TIMEOUT :
       _tprintf( _T( "Client timed out waiting for response.\n" ) );
       goto DONE;
    case WAIT_ABANDONED :
    case WAIT_FAILED :
    default :
      _tprintf( _T( "Client failed waiting for response.\n" ) );
      goto DONE;
  }

  result = ( clientSuccess ? 0 : 1 );
  // **TODO**: DSB, 07/30/2010 - Not implemented yet. 
  // I should queue a "well-known" completion packet 
  // to the port to signal the listener thread to exit cleanly.
DONE:
  closeHandle( hClientRequestEvent );
  closeHandle( hPipe );
  closeHandle( g_hListenerThread );
  closeHandle( g_hPipe );
  closeHandle( g_hIoCompletionPort );
  return result;
}

void closeHandle( HANDLE &hObj )
{
  if ( !( NULL == hObj || INVALID_HANDLE_VALUE == hObj ) )
  {
    CloseHandle( hObj );
    hObj = NULL;
  }
}

DWORD listenerThreadProc( void *pData )
{
  HANDLE hListenerThread = GetCurrentThread();
  _tprintf( _T( "Listener thread %x started." ), hListenerThread );
  DWORD result = 1; // Failure. Return zero for success.
  BOOL            bSuccess            = FALSE;
  DWORD           bytesCopied         = 0;
  DWORD           completionKey       = 0;
  DWORD           lastError           = 0;
  OVERLAPPED *    pOverlap             = NULL;
  while ( true )
  {
    bSuccess = GetQueuedCompletionStatus(
      g_hIoCompletionPort,
      &bytesCopied,
      &completionKey,
      (LPOVERLAPPED*)&pOverlap,
      INFINITE );
    if( !bSuccess )
    {
      lastError = GetLastError();
      _tprintf(_T( "Listener thread %x - GetQueuedCompletionStatus() failed")
               _T(". Error = %d." ),
                hListenerThread,
                lastError );
      
      if ( NULL == pOverlap && ERROR_ABANDONED_WAIT_0 == lastError )
      {
        // The port handle was closed while the thread was waiting.
        _tprintf(_T( "Listener thread %x - The IO completion port was")
                 _T(" closed prematurely." ),
                 hListenerThread );
      }
      break;
    }
    else
    {
      _tprintf(_T( "Listener thread %x processing request..." ), 
               hListenerThread );
      if ( NULL != pOverlap )
      {
        // **TODO**: DSB, 07/29/2010 - Not implemented yet. Not finished.
        //
        // Process completion packet
        DWORD numBytesWritten = 0;
        if ( !WriteFile(g_hPipe,
                        (LPCVOID)_T( "xxx" ),
                        4 * sizeof( _TCHAR ),
                        &numBytesWritten,
                        (LPOVERLAPPED)pOverlap ) )
        {
          _tprintf(_T( "Listener thread %x - WriteFile() failed. ")
                   _T("Error = %d." ),
                   hListenerThread,
                   GetLastError() );
        }
      }
      else
        _tprintf(_T( "Listener thread %x - GetQueuedCompletionStatus() ")
                 _T("returned a null LPPIPE_INST instance." ),
                 hListenerThread );
    _tprintf(_T( "Listener thread %x done processing request." ), 
             hListenerThread );
    }
  }
  result = 0; // Success.
  _tprintf( _T( "Listener thread %x stopped." ), hListenerThread );
  return result;
 }