For more accurate timing results with the worker queue, make sure that all threads are done processing an image before restarting the compression when testing multiple compressions.

Core/src/WorkerQueue.cpp

@@ -41,16 +41,39 @@ void WorkerThread::operator()() {
     return;
   }
 
-  while(1) {
+  bool quitFlag = false;
+  while(!quitFlag) {
     
-    EAction action = m_Parent->AcceptThreadData(m_ThreadIdx);
-    if(eAction_Quit == action) {
+    switch(m_Parent->AcceptThreadData(m_ThreadIdx)) 
+    {
+
+      case eAction_Quit:
+      {
+	quitFlag = true;
 	break;
       }
 
+      case eAction_Wait:
+      {
+	boost::thread::yield();
+	break;
+      }
+
+      case eAction_DoWork:
+      {
 	const uint8 *src = m_Parent->GetSrcForThread(m_ThreadIdx);
 	uint8 *dst = m_Parent->GetDstForThread(m_ThreadIdx);
 	(*f)(src, dst, 4 * m_Parent->GetNumBlocksForThread(m_ThreadIdx), 4);
+	break;
+      }
+
+      default:
+      {
+	fprintf(stderr, "Unrecognized thread command!\n");
+	quitFlag = true;
+	break;
+      }
+    }
   }
 
   m_Parent->NotifyWorkerFinished();
@@ -70,6 +93,7 @@ WorkerQueue::WorkerQueue(
   : m_NumCompressions(0)
   , m_TotalNumCompressions(max(uint32(1), numCompressions))
   , m_NumThreads(numThreads)
+  , m_WaitingThreads(0)
   , m_ActiveThreads(0)
   , m_JobSize(max(uint32(1), jobSize))
   , m_InBufSz(inBufSz)
@@ -99,6 +123,8 @@ void WorkerQueue::Run() {
   m_StopWatch.Reset();
   m_StopWatch.Start();
 
+  m_WaitingThreads = 0;
+
   // Wait for them to finish...
   while(m_ActiveThreads > 0) {
     m_CV.wait(lock);
@@ -134,9 +160,19 @@ WorkerThread::EAction WorkerQueue::AcceptThreadData(uint32 threadIdx) {
   
   // If we've completed all blocks, then mark the thread for 
   // completion.
-  if(m_NextBlock >= totalBlocks) {
+  if(m_NextBlock == totalBlocks) {
+    if(m_NumCompressions < m_TotalNumCompressions) {
+      if(++m_WaitingThreads == m_ActiveThreads) {
+	m_NextBlock = 0;
+	m_WaitingThreads = 0;
+      } else {
+	return WorkerThread::eAction_Wait;
+      }
+    }
+    else {
       return WorkerThread::eAction_Quit;
     }
+  }
 
   // Otherwise, this thread's offset is the current block...
   m_Offsets[threadIdx] = m_NextBlock;
@@ -149,8 +185,8 @@ WorkerThread::EAction WorkerQueue::AcceptThreadData(uint32 threadIdx) {
   // Make sure the next block is updated.
   m_NextBlock += blocksProcessed;
 
-  if(m_NextBlock == totalBlocks && ++m_NumCompressions < m_TotalNumCompressions) {
-    m_NextBlock = 0;
+  if(m_NextBlock == totalBlocks) {
+    ++m_NumCompressions;
   }
 
   return WorkerThread::eAction_DoWork;

Core/src/WorkerQueue.h

@@ -22,6 +22,7 @@ public:
   void operator ()();
 
   enum EAction {
+    eAction_Wait,
     eAction_DoWork,
     eAction_Quit,
 
@@ -53,10 +54,10 @@ class WorkerQueue {
   const StopWatch &GetStopWatch() const { return m_StopWatch; }
 
  private:
-
   uint32 m_NumCompressions;
   const uint32 m_TotalNumCompressions;
   uint32 m_NumThreads;
+  uint32 m_WaitingThreads;
   uint32 m_ActiveThreads;
   uint32 m_JobSize;
   uint32 m_InBufSz;
@@ -65,6 +66,7 @@ class WorkerQueue {
 
   boost::condition_variable m_CV;
   boost::mutex m_Mutex;
+
   uint32 m_NextBlock;
 
   static const int kMaxNumWorkerThreads = 256;