Another patch to review

Here is another change I'm planning to make.
Feel free to comment.

Since QueueRunnable and ThreadState have the same lifetimes,
it seems simpler to me to eliminate QueueRunnable and make
ThreadState inherit from Runnable. This allows the IncRef() / DecRef()
to be eliminated in ThreadState::Run() since ThreadState::threadObject
holds a reference to Thread which holds a reference to
ThreadState::Runnable.
The ~ThreadedJobQueue() breaks the reference count cycle when stopping
the threads and destroying the ThreadState(s).

Index: include/csutil/threadjobqueue.h
===================================================================
--- include/csutil/threadjobqueue.h	(revision 39570)
+++ include/csutil/threadjobqueue.h	(working copy)
@@ -65,40 +65,19 @@
 
   /// Get name of this queue
   const char* GetName () const { return name; }
+
 private:
-
   bool PullFromQueues (iJob* job);
   JobStatus CheckCompletion (iJob* job, bool waitForCompletion);
 
-  // Runnable
-  struct ThreadState;  
-
-  class QueueRunnable : public Runnable
+  // Per thread state
+  struct ThreadState : public Runnable
   {
-  public:
-    QueueRunnable (ThreadedJobQueue* queue, ThreadState* ts, unsigned
int id);
+    ThreadState (ThreadedJobQueue* queue, unsigned int id,
ThreadPriority pri);
 
     virtual void Run ();
     virtual const char* GetName () const;
-  private:
-    friend class ThreadedJobQueue;
-    
-    ThreadedJobQueue* ownerQueue;
-    int32 shutdownQueue;
-    csRef<ThreadState> threadState;
-    csString name;
-  };
 
-  // Per thread state
-  struct ThreadState : public CS::Utility::AtomicRefCount
-  {
-    ThreadState (ThreadedJobQueue* queue, unsigned int id)
-    {
-      runnable.AttachNew (new QueueRunnable (queue, this, id));
-      threadObject.AttachNew (new Thread (runnable, false));
-    }
-
-    csRef<QueueRunnable> runnable;
     csRef<Thread> threadObject;
     csRef<iJob> currentJob;
     
@@ -108,6 +87,10 @@
     Condition tsJobFinished;
 
     csFIFO<csRef<iJob> > jobQueue;
+
+    ThreadedJobQueue* ownerQueue;
+    int32 shutdownQueue;
+    csString tsname;
   };
 
   csRef<ThreadState>* allThreadState;
Index: libs/csutil/threadjobqueue.cpp
===================================================================
--- libs/csutil/threadjobqueue.cpp	(revision 39570)
+++ libs/csutil/threadjobqueue.cpp	(working copy)
@@ -48,11 +48,9 @@
     allThreadState = new csRef<ThreadState>[numWorkerThreads];
 
     // Start up the threads
-    for (unsigned int i = 0; i < numWorkerThreads; ++i)
+    for (size_t i = 0; i < numWorkerThreads; ++i)
     {
-      allThreadState[i].AttachNew (new ThreadState (this, i)); 
-      allThreadState[i]->threadObject->SetPriority(priority);
-
+      allThreadState[i].AttachNew (new ThreadState (this, i,
priority)); 
       allThreads.Add (allThreadState[i]->threadObject);
     }
     
@@ -62,10 +60,10 @@
   ThreadedJobQueue::~ThreadedJobQueue ()
   {
     // Kill all threads, friendly
-    for(size_t i = 0; i < numWorkerThreads; ++i)
+    for (size_t i = 0; i < numWorkerThreads; ++i)
     {
       CS::Threading::AtomicOperations::Set (
-	&(allThreadState[i]->runnable->shutdownQueue), 0xff);    
+	&(allThreadState[i]->shutdownQueue), 0xff);    
       allThreadState[i]->tsNewJob.NotifyAll ();
     }
 
@@ -74,7 +72,10 @@
     // Deallocate
     for (size_t i = 0; i < numWorkerThreads; ++i)
     {
-      allThreadState[i]->runnable.Invalidate();
+      // The Thread object holds a reference to this so there is a
+      // reference count loop that we have to break.
+      // It also means we don't have to IncRef() / DecRef() inside
Run().
+      allThreadState[i]->threadObject.Invalidate();
     }
     delete[] allThreadState;
   }
@@ -256,29 +257,25 @@
   }
 
 
-  ThreadedJobQueue::QueueRunnable::QueueRunnable (ThreadedJobQueue*
queue, 
-    ThreadState* ts, unsigned int id)
-    : ownerQueue (queue), shutdownQueue (0), threadState (ts)
+  ThreadedJobQueue::ThreadState::ThreadState (ThreadedJobQueue* queue, 
+    unsigned int id, ThreadPriority pri)
+    : ownerQueue (queue), shutdownQueue (0)
   {
-    name.Format ("#%u %s", id, queue->GetName());
+    tsname.Format ("#%u %s", id, queue->GetName());
+    threadObject.AttachNew (new Thread (this, false, pri));
   }
 
-  void ThreadedJobQueue::QueueRunnable::Run ()
-  {    
-    // Forcibly keep QueueRunnable object alive until we got a shutdown
-    this->IncRef();
-    while
(CS::Threading::AtomicOperations::Read(&(/*ownerQueue->*/shutdownQueue))
== 0x0)
+  void ThreadedJobQueue::ThreadState::Run ()
+  {
+    while (CS::Threading::AtomicOperations::Read(&shutdownQueue) ==
0x0)
     {
-      // Get a job
-      csRef<iJob> currentJob;
-
       // Try our own list first
       // We need to hold this until currentJob is set, otherwise
something might slip through in "wait"
-      threadState->tsMutex.Lock ();
+      tsMutex.Lock ();
 
-      if (threadState->jobQueue.GetSize () > 0)
+      if (jobQueue.GetSize () > 0)
       {
-        currentJob = threadState->jobQueue.PopTop ();
+        currentJob = jobQueue.PopTop ();
       }
       
       if (!currentJob)
@@ -297,7 +294,7 @@
              )
         {
           ThreadState* foreignTS = ownerQueue->allThreadState[index];
-          if (foreignTS == threadState)
+          if (foreignTS == this)
             continue;
 
           // Try to lock it, but never wait for a lock
@@ -319,8 +316,7 @@
       if (currentJob)
       {
         // Got one, execute
-        threadState->currentJob = currentJob;        
-        threadState->tsMutex.Unlock (); // Unlock our own TS after
getting a job
+        tsMutex.Unlock (); // Unlock our own TS after getting a job
 
         currentJob->Run ();
 
@@ -332,32 +328,26 @@
 	   TM for the last time. */
 	csRef<iJob> justKeepCurrentJobRefALittleLonger (currentJob);
         {
-          MutexScopedLock l (threadState->tsMutex);
-          threadState->currentJob = 0;
+          MutexScopedLock l (tsMutex);
           currentJob = 0;
         }
        

CS::Threading::AtomicOperations::Decrement(&(ownerQueue->outstandingJobs));
 	
-        threadState->tsJobFinished.NotifyAll ();
+        tsJobFinished.NotifyAll ();
       }
       else
       {
         // Couldn't get one, wait for a newly added job        
-        threadState->tsNewJob.Wait (threadState->tsMutex);
-        threadState->tsMutex.Unlock ();
+        tsNewJob.Wait (tsMutex);
+        tsMutex.Unlock ();
       }
     }
-    
-    // There is a circular ref between ThreadState and QueueRunnable,
break it up
-    threadState.Invalidate();
-    
-    this->DecRef();
   }
 
-  const char* ThreadedJobQueue::QueueRunnable::GetName () const
+  const char* ThreadedJobQueue::ThreadState::GetName () const
   {
-    return name.GetDataSafe ();
+    return tsname.GetDataSafe ();
   }
 
 



------------------------------------------------------------------------------
Get your SQL database under version control now!
Version control is standard for application code, but databases havent 
caught up. So what steps can you take to put your SQL databases under 
version control? Why should you start doing it? Read more to find out.
http://pubads.g.doubleclick.net/gampad/clk?id=48897031&iu=/4140/ostg.clktrk