Subj : Re: IOCP problem
To   : comp.programming.threads
From : Richard Hollis
Date : Wed Jan 26 2005 10:39 pm

Hi

> Does the problem seem to only affect you if you're using more than one
> thread, while your stress testing?

Yes.   When there is only one thread it is ok, but any more and the hangs
start and it is as if both threads are waking up for the same pipe.

> Could you post a link to the information you read?

Sure, a bit long though:


http://groups-beta.google.com/group/comp.programming.threads/browse_frm/thread/98abdcce0b3f6b64/41a0b8694517e0e8?q=GetQueuedCompletionStatus+worker+threads&_done=%2Fgroups%3Fsourceid%3Dnavclient%26ie%3DUTF-8%26oe%3DUTF-8%26q%3DGetQueuedCompletionStatus+worker+threads%26&_doneTitle=Back+to+Search&&d#41a0b8694517e0e8

I notice that you appear to be in that thread too :)
> Basically, a skeleton of your worker thread loop would help.

If I just post this for now then and we can go from there.

Thanks
Richard

---------

unsigned __stdcall ThreadPoolProc( void * lpParam)
{
ULONG nSocket;
DWORD nBytesToBeRead, dwLastErr;
OVERLAPPED * lpo = NULL;
BOOL b;
BYTE read_buffer[PIPE_MSG_MAX];


 for(;;)
 {
  BOOL bIORet = GetQueuedCompletionStatus(
   g_hCompletionPort,
   &nBytesToBeRead,
   &nSocket,
   &lpo,
   INFINITE);

  dwLastErr = GetLastError();

  HANDLE hPipe = m_hPipes[nSocket];
  DWORD dwActivePipe = nSocket;
  int & state = m_states[dwActivePipe];

  if(!bIORet && dwLastErr != WAIT_TIMEOUT)
  {
   break;
  }

  if(bIORet && NULL != lpo)
  {
   DWORD dwBytesRead = nBytesToBeRead;
   DWORD dwBytesWritten;

   if( state == 0 )
   {
    b = ReadFile( hPipe,
     &read_buffer,
     PIPE_MSG_MAX,
     &dwBytesRead,
     lpo
    );

    dwLastErr = GetLastError();

    if(!b)
    {
     if( dwLastErr == ERROR_IO_PENDING )
     {
      DWORD event = WaitForSingleObject(lpo->hEvent, INFINITE);

      b = GetOverlappedResult(hPipe,
       lpo,
       &dwBytesRead,
       FALSE);

      if(!b)
      {
       dwLastErr = GetLastError();
      }
     }
     else
     {
      TRACE1("ReadFile dwLastErr = %d\n", dwLastErr);
     }
    }

    if(!b)
    {
     DisconnectNamedPipe( hPipe );
     ConnectNamedPipe( hPipe, lpo );
     state = 0;
     continue;
    }

    // do something with message

    state++;
   }
   else if(state == 1)
   {
    b = WriteFile(hPipe,
     "OK",
     3,
     &dwBytesWritten,
     lpo);

    dwLastErr = GetLastError();

    if(!b)
    {
     if( dwLastErr == ERROR_IO_PENDING )
     {
      DWORD event = WaitForSingleObject(lpo->hEvent, INFINITE );

      if( event == WAIT_OBJECT_0 )
      {
       b = GetOverlappedResult(hPipe,
        lpo,
        &dwBytesWritten,
        TRUE);
      }
     }
     else
     {
      TRACE1("WriteFile dwLastErr = %d\n", dwLastErr);
     }
    }

    if(!b)
    {
     DisconnectNamedPipe( hPipe );
     ConnectNamedPipe( hPipe, lpo );
     state = 0;
     continue;
    }

    state++;
   }
   else if(state == 2)
   {
    DisconnectNamedPipe( hPipe );
    ConnectNamedPipe( hPipe, lpo );
    state = 0;
   }
  }
 }

 _endthreadex(0);

 return 0;
}

BOOL CPipeServer::Start()
{
 // create pipes
 for(i = 0; i < MAX_PIPE_INSTANCES; i++)
 {
  m_hPipes[i] = CreateNamedPipe(TEXT("\\\\.\\pipe\\askcounters"),
         PIPE_ACCESS_DUPLEX |
         FILE_FLAG_OVERLAPPED,
         PIPE_TYPE_MESSAGE |
         PIPE_READMODE_MESSAGE |
         PIPE_WAIT,
         MAX_PIPE_INSTANCES,
         PIPE_MSG_MAX,
         PIPE_MSG_MAX,
         NMPWAIT_USE_DEFAULT_WAIT,
         NULL
         );

  g_hCompletionPort = CreateIoCompletionPort(m_hPipes[i],
             g_hCompletionPort,
             i,
             0);

  m_hPipeEvents[i] = CreateEvent(NULL,
          TRUE,
          FALSE,
          NULL);

  m_oLapArray[i].hEvent = m_hPipeEvents[i];
  BOOL bRet = ConnectNamedPipe( m_hPipes[i], &m_oLapArray[i] );
 }

 for(i = 0; i < nWorkerThreads; i++)
 {
  HANDLE hThread = (HANDLE) _beginthreadex(
   NULL,
   0,
   ThreadPoolProc,
   (void*) this,
   0,
   &threadID
   );
 }
}

.