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);