CR/CN: Fixed thread-safety issues in HXScheduler

"Petar Basic" <[email protected]>
Newsgroups gmane.comp.multimedia.helix.devel
Message-ID <[email protected]>
Modified by: pbasic at real.com
Reviewed by: milko, gwright
Date: 2008/09/06
Project: strummer

Synopsis:
Fixed thread-safety issues in HXScheduler

Details:
There was a race condition in ClientPQ class.
ClientPQ is derived from PQ which accepts CHXID object as an argument
to the constructor.  ClientPQ is obviously designed as a thread-safe
wrapper around PQ, but each instance of ClientPQ uses its own mutex
object.  However, this is not enough.  Since CHXID object can be
shared among PQ instances, it must be protected by a single mutex.
Instantiations of ClientPQ in HXScheduler are critical to this case
since HXScheduler uses 3 ClientPQ objects which share a single CHXID.
HXOptimizedScheduler currently uses only 1 ClientPQ object.

1.) Additional assert statements which can detect race condition have
been added to CHXID class.
2.) ClientPQ is modified to accept externally created mutex.  In this
way CHXID object can be protected through common mutex by multiple
ClientPQ instances.
3.) Destruction of PQ base in ClientPQ destructor has been made thread-safe.
4.) HXScheduler has been modified to use a single mutex with all of
its ClientPQ instances.
5.) Although it was not necessary at this time, HXOptimizedScheduler
has been modified in the same way as HXScheduler to prepare it for
possible future updates.

Testing:
Performed over 1200 transcoding transfers to USB memory during single
VFMM session without crashes or lockups.  Also performed simultaneous
transfers to 2 USB devices without crashes or lockups.

Files Modified:
common/util/pub/id.h
common/system/pub/pq.h
common/system/pq.cpp
client/common/system/pub/clientpq.h
client/common/system/clientpq.cpp
client/common/system/pub/hxsched.h
client/common/system/hxsched.cpp
client/common/system/pub/hxoptsc.h
client/common/system/hxoptsc.cpp

Platforms and Profiles Build Verified:
system id: win32-i386-vc7
profile: helix-dtdr-all-defines

Platforms and Profiles Functionality Verified:
x86 Windows XP SP2

Branch:
HEAD
hxclient_3_4_0_atlas
hxclient_3_1_0_atlas

Copyright assignment:
I am a RealNetworks employee or contractor.

_______________________________________________
Helix-client-dev mailing list
[email protected]
http://lists.helixcommunity.org/mailman/listinfo/helix-client-dev
common_util.diff (application/octet-stream, 2 KB)
Index: pub/id.h
===================================================================
RCS file: /cvsroot/common/util/pub/id.h,v
retrieving revision 1.9
diff -w -U10 -r1.9 id.h
--- pub/id.h	6 Jul 2007 20:39:23 -0000	1.9
+++ pub/id.h	5 Sep 2008 23:37:51 -0000
@@ -137,24 +138,35 @@
     id--;
     ASSERT(id < table_size);
 
     if (id < table_size)
         table_ptr[id] = ptr;
 }
 
 inline UINT32
 CHXID::create(void* ptr)
 {
+    // we should not be storing NULL pointers
+    ASSERT((ptr != (void*)DEFAULT_VALUE) && "CHXID::create: storing NULL value");
+    ASSERT((slots_used != UINT32(0xFFFFFFFF)) && "CHXID::create: detected possible race condition");
+
     if (slots_used > table_size * 0.7)
     {
         void** tmp_table_ptr = new void* [table_size + increment_factor];
 
+        ASSERT(tmp_table_ptr && "CHXID::create: failed to reserve memory block");
+        if(!tmp_table_ptr)
+        {
+            m_LastError = HXR_OUTOFMEMORY;
+            return 0;
+        }
+
         memcpy (tmp_table_ptr, table_ptr, table_size * sizeof(void*)); /* Flawfinder: ignore */
         memset (tmp_table_ptr + table_size, DEFAULT_VALUE, increment_factor * sizeof(void*));
 
         delete [] table_ptr;
         table_ptr = tmp_table_ptr;
 
         table_size += increment_factor;
         increment_factor = table_size / 2;
     }
 
@@ -172,27 +184,32 @@
 CHXID::destroy(UINT32 id)
 {
     id--;
     ASSERT(id < table_size);
 
     if (id > table_size)
     {
         return 0;
     }
 
+    // we should not be destroying an empty slot
+    ASSERT((table_ptr[id] != (void*)DEFAULT_VALUE) && "CHXID::destroy: destroying empty slot");
+
     void* ptr = table_ptr[id];
     if (ptr == (void*)DEFAULT_VALUE)
         return 0;
 
     table_ptr[id] = (void*)DEFAULT_VALUE;
     slots_used--;
 
+    ASSERT((slots_used != UINT32(0xFFFFFFFF)) && "CHXID::destroy: detected possible race condition");
+
     return ptr;
 }
 
 inline UINT32
 CHXID::get_size()
 {
     return table_size;
 }
 
 #endif
common_system.diff (application/octet-stream, 2.3 KB)
Index: pq.cpp
===================================================================
RCS file: /cvsroot/common/system/pq.cpp,v
retrieving revision 1.8
diff -w -U10 -r1.8 pq.cpp
--- pq.cpp	6 Jul 2007 20:41:55 -0000	1.8
+++ pq.cpp	5 Sep 2008 23:43:04 -0000
@@ -70,22 +70,20 @@
 #ifdef WIN32
 /*
  * XXX This is the 3.0 bug that Glenn found. 
  * We need to fix this correctly!
  */
 #define float double
 #endif
 
 PQ::PQ(CHXID* pIds)
 {
-//    int i;
-
     m_pHead = 0;
     m_pNextZeroInsertion = 0;
     m_lElementCount = 0;
     m_HeadTime.tv_sec = PQ_UNINITIALIZED;
 
     gettimeofday(&m_Bucket0Time, 0);
     m_Bucket0Time.tv_usec = 0;
 
     memset (m_pBuckets, 0, NUM_BUCKETS * sizeof (PQElem*));
 
@@ -96,20 +94,32 @@
     }
     else
     {
 	m_pIds = new CHXID;
 	m_bOwnID = TRUE;
     }
 }
 
 PQ::~PQ()
 {
+    destruct();
+}
+
+void
+PQ::destruct()
+{
+    // check if we're already destructed
+    if (!m_pIds)
+    {
+        return;
+    }
+
     Timeval now;
 
     now.tv_sec = 0x7fffffff;
     now.tv_usec = 0x7fffffff;
     PQElem*	pElem = _remove_head(now);
 
     while (pElem) 
     {
 	PQElem* pElemNext = pElem->m_pNext;
 	m_pIds->destroy(pElem->m_Id);
@@ -135,20 +145,23 @@
 	    if (!pElemNext)
 		break;
 	    pElem = pElemNext; 
 	}
     }
 
     if (m_bOwnID)
     {
 	HX_DELETE(m_pIds);
     }
+
+    // mark as destructed
+    m_pIds = 0;
 }
 
 UINT32
 PQ::enter(Timeval t, IHXCallback* pCallback)
 {
 #ifdef WIN32
     /*
      * Normalize t for NT
      */
 
Index: pub/pq.h
===================================================================
RCS file: /cvsroot/common/system/pub/pq.h,v
retrieving revision 1.8
diff -w -U10 -r1.8 pq.h
--- pub/pq.h	6 Jul 2007 20:41:59 -0000	1.8
+++ pub/pq.h	5 Sep 2008 23:43:51 -0000
@@ -88,20 +88,21 @@
     HXBOOL	    	m_bRemoved;
     UINT32		m_Id;
 };
 
 class PQ
 {
 protected:
     PQElem*		_remove_head(Timeval now);
     PQElem*     dispatch_element(PQElem* pElem);
     void        destroy_element(PQElem* pElem);
+    void        destruct();
 
     PQElem*	m_pBuckets[NUM_BUCKETS];
     PQElem*	m_pHead;
     PQElem*	m_pNextZeroInsertion;
     LONG32	m_lElementCount;
     Timeval	m_Bucket0Time;
     Timeval	m_HeadTime;
     CHXID*	m_pIds;
     HXBOOL	m_bOwnID;
client_common_system.diff (application/octet-stream, 9.9 KB)
Index: clientpq.cpp
===================================================================
RCS file: /cvsroot/client/common/system/clientpq.cpp,v
retrieving revision 1.9.2.1
diff -w -U10 -r1.9.2.1 clientpq.cpp
--- clientpq.cpp	22 Feb 2008 12:05:03 -0000	1.9.2.1
+++ clientpq.cpp	6 Sep 2008 00:15:06 -0000
@@ -57,39 +57,60 @@
 #include "pckunpck.h"
 #include "clientpq.h"
 #include "hxthread.h"
 #include "hxthreadyield.h"
 #include "hxheap.h"
 #ifdef _DEBUG
 #undef HX_THIS_FILE		
 static const char HX_THIS_FILE[] = __FILE__;
 #endif
 
-ClientPQ::ClientPQ(IUnknown* pContext, CHXID* pIds)
+
+ClientPQ::ClientPQ(IUnknown* pContext, CHXID* pIds, IHXMutex* pMutex)
     : PQ(pIds),
       m_pFreeList(NULL),
       m_uNumFreeNodes(0),
       m_uNumNodesToCache(DEFAULT_NODE_CACHE_SIZE),
       m_pMutex(NULL)
 {
+    HX_ASSERT(((pIds && pMutex) || (!pIds && !pMutex)) && "Creating thread-unsafe ClientPQ");
+
+    if(pMutex)
+    {
+        // use externally provided mutex
+        m_pMutex = pMutex;
+        m_pMutex->AddRef();
+    }
+    else
+    {
+        // use private mutex
     CreateInstanceCCF(CLSID_IHXMutex, (void**)&m_pMutex, pContext);  
 }
+}
 
 
 ClientPQ::~ClientPQ()
 {
+    // protect cleanup
+    m_pMutex->Lock();
+
     while(m_pFreeList)
     {
 	PQElem* pElem = m_pFreeList;
 	m_pFreeList = m_pFreeList->m_pNext;
 	delete pElem;
     }
+
+    // destruct base object under thread-safe lock
+    PQ::destruct();
+
+    m_pMutex->Unlock();
     HX_RELEASE(m_pMutex);
 }
 
 
 int ClientPQ::execute(Timeval now)
 {
     int     nCount = 0;
     PQElem* pElem  = NULL;
     
     //Protect just the _remove_head call with the mutex.
Index: hxoptsc.cpp
===================================================================
RCS file: /cvsroot/client/common/system/hxoptsc.cpp,v
retrieving revision 1.16
diff -w -U10 -r1.16 hxoptsc.cpp
--- hxoptsc.cpp	6 Jul 2007 21:58:03 -0000	1.16
+++ hxoptsc.cpp	6 Sep 2008 00:15:11 -0000
@@ -78,51 +78,55 @@
 #define MINIMUM_DIFFERENCE    1
 #define	ALLFS                 0xFFFFFFFF
 
 void* ThreadRoutine(void * pArg);
 
 // HXOptimizedScheduler...
 HXOptimizedScheduler::HXOptimizedScheduler(IUnknown* pContext) :
      m_lRefCount (0)
     ,m_pPQ(0)
     ,m_pID(NULL)
+    ,m_pPQMutex(NULL)
     ,m_pContext(pContext)
     ,m_pScheduler(NULL)
     ,m_ulLastUpdateTime(0)
     ,m_ulLastSyncTime(0)
     ,m_pMutex(NULL)
     ,m_pThread(NULL)
     ,m_pQuitEvent(NULL)
     ,m_pSleepEvent(NULL)
     ,m_ulCurrentGranularity(MINIMUM_GRANULARITY)
     ,m_bIsDone(FALSE)
 {
+    CreateInstanceCCF(CLSID_IHXMutex, (void**)&m_pPQMutex, m_pContext);
+
     m_pID    = new CHXID(50);
-    m_pPQ    = new ClientPQ(pContext, m_pID);
+    m_pPQ    = new ClientPQ(pContext, m_pID, m_pPQMutex);
 
     CreateInstanceCCF(CLSID_IHXMutex, (void**)&m_pMutex, m_pContext);  
 
     gettimeofday((Timeval*)&m_CurrentTimeVal, 0);
     m_ulLastSyncTime = m_ulLastUpdateTime = HX_GET_BETTERTICKCOUNT();
 
     if (m_pContext)
     {
 	m_pContext->AddRef();
     }
 }
 
 HXOptimizedScheduler::~HXOptimizedScheduler()
 {
     StopScheduler();
 
     HX_DELETE(m_pPQ);
     HX_DELETE(m_pID);
+    HX_RELEASE(m_pPQMutex);
     HX_RELEASE(m_pMutex);
     HX_RELEASE(m_pContext);
     HX_RELEASE(m_pScheduler);
 }
 
 /*
  * IUnknown methods
  */
 
 
Index: hxsched.cpp
===================================================================
RCS file: /cvsroot/client/common/system/hxsched.cpp,v
retrieving revision 1.26.2.3
diff -w -U10 -r1.26.2.3 hxsched.cpp
--- hxsched.cpp	29 Apr 2008 18:00:34 -0000	1.26.2.3
+++ hxsched.cpp	6 Sep 2008 00:15:11 -0000
@@ -101,20 +101,21 @@
 #define MINIMUM_DIFFERENCE    5
 
 // HXScheduler...
 HXScheduler::HXScheduler(IUnknown* pContext) :
      m_lRefCount (0)
     ,m_pScheduler(0)
     ,m_bUseDeferredTask(TRUE)
     ,m_pInterruptTimeScheduler(0)
     ,m_pInterruptTimeOnlyScheduler(0)
     ,m_pID(0)
+    ,m_pPQMutex(NULL)
     ,m_pContext(pContext)
     ,m_bLocked(FALSE)
     ,m_ulLastUpdateTime(0)
     ,m_pCoreMutex(NULL)
     ,m_bIsInterruptEnabled(FALSE)
     ,m_headTime(0)
     ,m_interruptHeadTime(0)
     ,m_interruptOnlyHeadTime(0)
     ,m_ulSystemNextDueTime(0)
     ,m_ulInterruptNextDueTime(0)
@@ -125,63 +126,73 @@
     ,m_pAsyncTimer(0)
 #endif 
     ,m_pTimeline(0)
     ,m_ulCurrentGranularity(0)
     ,m_ulMinimumGranularity(MINIMUM_GRANULARITY)
     ,m_pWaitEvent(NULL)
     ,m_bWaitPending(FALSE)
     ,m_bWaitedEventFired(FALSE)
     ,m_ulThreadID(0)
 {
-    m_pID			= new CHXID(100);
-    m_pScheduler		= new ClientPQ(pContext, m_pID);
-    m_pInterruptTimeScheduler	= new ClientPQ(pContext, m_pID);
-    m_pInterruptTimeOnlyScheduler = new ClientPQ(pContext, m_pID);
-
-    (void) gettimeofday(&m_CurrentTimeVal, 0);
-    m_ulLastUpdateTime = HX_GET_TICKCOUNT();
-
-    // Create event that wil be used for scheduler waiting operation
+    // obtain class factory
+    IHXCommonClassFactory* pCCF = NULL;
     if (m_pContext)
     {
-	IHXCommonClassFactory* pCCF = NULL;
-
 	m_pContext->QueryInterface(IID_IHXCommonClassFactory, (void**) &pCCF);
+    }
+
+    // create priority queue mutex
 	if (pCCF)
 	{
-	    pCCF->CreateInstance(IID_IHXEvent, (void**) &m_pWaitEvent);
+        pCCF->CreateInstance(CLSID_IHXMutex, (void**)&m_pPQMutex);
 	}
 
+    // create priority queues
+    m_pID = new CHXID(100);
+    m_pScheduler = new ClientPQ(pContext, m_pID, m_pPQMutex);
+    m_pInterruptTimeScheduler = new ClientPQ(pContext, m_pID, m_pPQMutex);
+    m_pInterruptTimeOnlyScheduler = new ClientPQ(pContext, m_pID, m_pPQMutex);
+
+    // Create event that wil be used for scheduler waiting operation
+    if (pCCF)
+    {
+        pCCF->CreateInstance(IID_IHXEvent, (void**) &m_pWaitEvent);
+
 	if (m_pWaitEvent)
 	{
 	    if (FAILED(m_pWaitEvent->Init("Scheduler_Wait", FALSE)))
 	    {
 		HX_RELEASE(m_pWaitEvent);
 	    }
 	}
 
 	ReadPrefUINT32(m_pContext, "SchedulerMinimumGranularity", m_ulMinimumGranularity);
-	HX_RELEASE(pCCF);
     }
 
+    // release class factory
+    HX_RELEASE(pCCF);
+
+    (void) gettimeofday(&m_CurrentTimeVal, 0);
+    m_ulLastUpdateTime = HX_GET_TICKCOUNT();
     m_ulThreadID = HXGetCurrentThreadID();
 
 }
 
 HXScheduler::~HXScheduler()
 {
     StopScheduler();
 
     HX_DELETE(m_pScheduler);
     HX_DELETE(m_pInterruptTimeScheduler);
     HX_DELETE(m_pInterruptTimeOnlyScheduler);
     HX_DELETE(m_pID);
+    HX_RELEASE(m_pPQMutex);
 
 #if defined(_WIN32) || defined(THREADS_SUPPORTED)
     HX_DELETE(m_pAsyncTimer);
 #endif 
     HX_DELETE(m_pTimeline);
     HX_RELEASE(m_pWaitEvent);
     HX_RELEASE(m_pCoreMutex);
 }
 
 /*
Index: pub/clientpq.h
===================================================================
RCS file: /cvsroot/client/common/system/pub/clientpq.h,v
retrieving revision 1.6.2.1
diff -w -U10 -r1.6.2.1 clientpq.h
--- pub/clientpq.h	22 Feb 2008 12:05:05 -0000	1.6.2.1
+++ pub/clientpq.h	6 Sep 2008 00:15:18 -0000
@@ -60,21 +60,26 @@
 #include "hxassert.h"
 #include "pq.h"
 #include "hxthread.h"
 
 class HXMutex;
 
 #define DEFAULT_NODE_CACHE_SIZE	    50
 class ClientPQ : public PQ
 {
 public:
-    ClientPQ(IUnknown* pContext, CHXID* pIds = NULL);
+    // IMPORTANT: If externally created CHXID object is passed into constructor, make sure
+    // you also provide externally created mutex, otherwise the created ClientPQ object
+    // won't be completely thread-safe. Such mutex must be shared by all ClientPQ instances
+    // which access the same CHXID object.
+    //
+    ClientPQ(IUnknown* pContext, CHXID* pIds = NULL, IHXMutex* pMutex = NULL);
     virtual ~ClientPQ();
 
     virtual int    execute(Timeval now);
     virtual UINT32 enter(Timeval t, IHXCallback* i);
     virtual void   remove(UINT32 handle);
     virtual HXBOOL   removeifexists(UINT32 id);
 
     
     void SetCacheSize(UINT16 uCacheSize) {m_uNumNodesToCache = uCacheSize;};
 
Index: pub/hxoptsc.h
===================================================================
RCS file: /cvsroot/client/common/system/pub/hxoptsc.h,v
retrieving revision 1.9
diff -w -U10 -r1.9 hxoptsc.h
--- pub/hxoptsc.h	6 Jul 2007 21:58:04 -0000	1.9
+++ pub/hxoptsc.h	6 Sep 2008 00:15:19 -0000
@@ -189,20 +189,21 @@
     STDMETHOD(StartScheduler)	(THIS);
     STDMETHOD(StopScheduler)	(THIS);
 
     HX_RESULT	ExecuteCurrentFunctions();
     HXBOOL	GetNextEventDueTime(UINT32& ulNumMs);
 
 protected:
     LONG32	    m_lRefCount;
     ClientPQ*		    m_pPQ;
     CHXID*		    m_pID;
+    IHXMutex*       m_pPQMutex;
     IUnknown*	    m_pContext;
     IHXScheduler*  m_pScheduler;
 
     HXTimeval	    m_CurrentTimeVal;
     UINT32	    m_ulLastUpdateTime;
     UINT32	    m_ulLastSyncTime;
     IHXMutex*	    m_pMutex;
 
     /* Used to schedule events in a separate thread */
     friend void* ThreadRoutine (void * pArg);
Index: pub/hxsched.h
===================================================================
RCS file: /cvsroot/client/common/system/pub/hxsched.h,v
retrieving revision 1.15.2.1
diff -w -U10 -r1.15.2.1 hxsched.h
--- pub/hxsched.h	29 Apr 2008 18:00:36 -0000	1.15.2.1
+++ pub/hxsched.h	6 Sep 2008 00:15:19 -0000
@@ -80,20 +80,21 @@
     HXBOOL      m_bUseDeferredTask;
 
     // Is set to TRUE for most of the time except when we need to remove all
     // callbacks from the scheduler without actually exectuing the functions..
     static HXBOOL	m_sbProcess;
     
     /* special PQ for interrupt safe tasks */
     ClientPQ*   m_pInterruptTimeScheduler;	/* services at either interrupt or non-interrupt time */
     ClientPQ*   m_pInterruptTimeOnlyScheduler;	/* serevics only at interrupt time */
     CHXID*      m_pID;
+    IHXMutex*   m_pPQMutex;
 
     IUnknown*   m_pContext;
 
     /* semaphore to prevent a deferred interrupt task from interrupting
      * a system task
      */
     HXBOOL		m_bLocked; 
 
 public:
 			HXScheduler(IUnknown* pContext);
lmpx.com only provides a reader for public news (NNTP) servers. It is not affiliated with the servers or forums shown here and is not responsible for the content of articles, which is written by their respective authors.