具有Reset功能的多线程同步队列 - 4

BOOL CTreadSafeMsgQueue::PostMsg(const MsgItem Msg, int WaitTime)

{

         if (!m_bInitedOK) return FALSE;

         // 防止在reset期间导致Semaphore计数不正确

         while (m_bStop)

         {

                  TRACE("Thread %d Sleep\n", GetCurrentThreadId());

                  Sleep(SLEEP_TIME);

         }

         // 进入同步操作状态

         m_WritingThreadNum++;

         if (WaitForSingleObject(m_S_Producer, WaitTime) == WAIT_OBJECT_0)

         {

                   if (WaitForSingleObject(m_E_Queue, WaitTime) == WAIT_OBJECT_0)

                  {

                            // OK now, post message then

                            m_Queue[m_HeaderToWrite++] = Msg;

                            if (m_HeaderToWrite >= MAX_QUE_SIZE) m_HeaderToWrite = 0;

                            TRACE("Post message *** %d\n", Msg.MsgID);

                            ReleaseSemaphore(m_S_Consumer, 1, NULL);

                            SetEvent(m_E_Queue);

                            m_WritingThreadNum--;

                            return TRUE;

                   }

                   else // wait event time out

                   {

                            // not post message so release producer

                            ReleaseSemaphore(m_S_Producer, 1, NULL);

                            SetEvent(m_E_Queue);

                            m_WritingThreadNum--;

                            return FALSE;

                   }

         }

         else // wait semaphore time out

         {

                  m_WritingThreadNum--;

                   return FALSE;

         }

}

BOOL CTreadSafeMsgQueue::Reset()

{

         if (!m_bInitedOK) return FALSE;

         // 防止重入Reset

         while (m_bStop) Sleep(SLEEP_TIME);

         // 防止新的线程进入读写操作状态

         m_bStop = TRUE;

         TRACE("--------------Begin to Reset-------------\n");

         // 等待至少一类线程（读线程或写线程）退出同步状态

    while ( (m_WritingThreadNum != 0) && (m_ReadingThreadNum != 0)) Sleep(SLEEP_TIME);

         // 此时，可能（最坏可能）还有一类线程处于同步状态，并被阻塞。

         // 它们必定都被阻塞于信号量（Semaphore）状态, 注意此时Event事件处于信号态!!

         // 如果被阻塞的是写线程，它们将被释放，并写入数据，但它们写入的数据将被抛弃。

         // 释放写线程将改写m_S_Consumer，所以必须先释放写线程;

         // ReleaseSemaphore(m_S_Producer)令m_S_Producer=MAX_QUE_SIZE;

         while (m_WritingThreadNum > 0)

         {

                   // 必须Sleep以出让CPU控制权

                  ReleaseSemaphore(m_S_Producer, 1, NULL);

                  Sleep(5);

         }

         // 继续阻塞读线程，令m_S_Consumer=0;

         while (WaitForSingleObject(m_S_Consumer, 10) == WAIT_OBJECT_0);

         // 由于写操作会改变m_HeaderToWrite，必须在写操作之后对其重置

         m_HeaderToWrite = m_TailToRead = 0;

         // 注意

         TRACE("--------------Finished Reseting-------------\n");

         m_bStop =  FALSE;

         return TRUE;

}