Merge branch 'dapatch'

6 files changed, 102 insertions, 6 deletions

diff --git a/ChangeLog b/ChangeLog
index d8be1f56..0cbe7bba 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,7 @@
+- bugfix: potential abort with DA queue after high watermark is reached
+  There exists a race condition that can lead to a segfault. Thanks
+  go to vbernetr, who performed the analysis and provided patch, which
+  I only tweaked a very little bit.
 ---------------------------------------------------------------------------
 Version 4.1.6  [DEVEL] (rgerhards), 2009-03-??
 - implemented function support in RainerScript. That means the engine
diff --git a/runtime/debug.c b/runtime/debug.c
index 53291fc2..96004e47 100644
--- a/runtime/debug.c
+++ b/runtime/debug.c
@@ -476,6 +476,55 @@ static inline void dbgMutexLockLog(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB,
 		dbgprintf("%s:%d:%s: mutex %p aquired\n", pFuncDB->file, lockLn, pFuncDB->func, (void*)pmut); 
 }
 
+
+/* report trylock on a mutex and add it to the mutex log */
+static inline void dbgMutexPreTryLockLog(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB, int ln)
+{
+   dbgMutLog_t *pHolder;
+   dbgMutLog_t *pLog;
+   char pszBuf[128];
+   char pszHolderThrdName[64];
+   char *pszHolder;
+
+   pthread_mutex_lock(&mutMutLog);
+   pHolder = dbgMutLogFindHolder(pmut);
+   pLog = dbgMutLogAddEntry(pmut, MUTOP_TRYLOCK, pFuncDB, ln);
+
+   if(pHolder == NULL)
+      pszHolder = "[NONE]";
+   else {
+      dbgGetThrdName(pszHolderThrdName, sizeof(pszHolderThrdName), pHolder->thrd, 1);
+      snprintf(pszBuf, sizeof(pszBuf)/sizeof(char), "%s:%d [%s]", pHolder->pFuncDB->file, pHolder->lockLn, pszHolderThrdName);
+      pszHolder = pszBuf;
+   }
+
+   if(bPrintMutexAction)
+      dbgprintf("%s:%d:%s: mutex %p trying to get lock, held by %s\n", pFuncDB->file, ln, pFuncDB->func, (void*)pmut, pszHolder);
+   pthread_mutex_unlock(&mutMutLog);
+}
+
+
+/* report attempted mutex lock */
+static inline void dbgMutexTryLockLog(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB, int lockLn)
+{
+   dbgMutLog_t *pLog;
+
+   pthread_mutex_lock(&mutMutLog);
+
+   /* find and delete "trylock" entry */
+   pLog = dbgMutLogFindSpecific(pmut, MUTOP_TRYLOCK, pFuncDB);
+   assert(pLog != NULL);
+   dbgMutLogDelEntry(pLog);
+
+   /* add "lock" entry */
+   pLog = dbgMutLogAddEntry(pmut, MUTOP_LOCK, pFuncDB, lockLn);
+   dbgFuncDBAddMutexLock(pFuncDB, pmut, lockLn);
+   pthread_mutex_unlock(&mutMutLog);
+   if(bPrintMutexAction)
+      dbgprintf("%s:%d:%s: mutex %p aquired\n", pFuncDB->file, lockLn, pFuncDB->func, (void*)pmut);
+}
+
+
 /* if we unlock, we just remove the lock aquired entry from the log list */
 static inline void dbgMutexUnlockLog(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB, int unlockLn)
 {
@@ -534,6 +583,26 @@ int dbgMutexLock(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB, int ln, int iStack
 }
 
 
+/* wrapper for pthread_mutex_trylock() */
+int dbgMutexTryLock(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB, int ln, int iStackPtr)
+{
+   int ret;
+
+   dbgRecordExecLocation(iStackPtr, ln);
+   dbgMutexPreLockLog(pmut, pFuncDB, ln); // TODO : update this
+   ret = pthread_mutex_trylock(pmut);
+   if(ret == 0 || ret == EBUSY) {
+      // TODO : update this
+      dbgMutexLockLog(pmut, pFuncDB, ln);
+   } else {
+      dbgprintf("%s:%d:%s: ERROR: pthread_mutex_trylock() for mutex %p failed with error %d\n",
+           pFuncDB->file, ln, pFuncDB->func, (void*)pmut, ret);
+   }
+
+   return ret;
+}
+
+
 /* wrapper for pthread_mutex_unlock() */
 int dbgMutexUnlock(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncDB, int ln, int iStackPtr)
 {
diff --git a/runtime/debug.h b/runtime/debug.h
index 59b3e776..1375493d 100644
--- a/runtime/debug.h
+++ b/runtime/debug.h
@@ -89,6 +89,7 @@ void sigsegvHdlr(int signum);
 void dbgoprint(obj_t *pObj, char *fmt, ...) __attribute__((format(printf, 2, 3)));
 void dbgprintf(char *fmt, ...) __attribute__((format(printf, 1, 2)));
 int dbgMutexLock(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncD, int ln, int iStackPtr);
+int dbgMutexTryLock(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncD, int ln, int iStackPtr);
 int dbgMutexUnlock(pthread_mutex_t *pmut, dbgFuncDB_t *pFuncD, int ln, int iStackPtr);
 int dbgCondWait(pthread_cond_t *cond, pthread_mutex_t *pmut, dbgFuncDB_t *pFuncD, int ln, int iStackPtr);
 int dbgCondTimedWait(pthread_cond_t *cond, pthread_mutex_t *pmut, const struct timespec *abstime, dbgFuncDB_t *pFuncD, int ln, int iStackPtr);
@@ -129,18 +130,21 @@ void dbgPrintAllDebugInfo(void);
 #define MUTOP_LOCKWAIT		1
 #define MUTOP_LOCK		2
 #define MUTOP_UNLOCK		3
+#define MUTOP_TRYLOCK		4
 
 
 /* debug aides */
 //#ifdef RTINST
 #if 0 // temporarily removed for helgrind
 #define d_pthread_mutex_lock(x)      dbgMutexLock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT )
+#define d_pthread_mutex_trylock(x)   dbgMutexTryLock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT )
 #define d_pthread_mutex_unlock(x)    dbgMutexUnlock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT )
 #define d_pthread_cond_wait(cond, mut)   dbgCondWait(cond, mut, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT )
 #define d_pthread_cond_timedwait(cond, mut, to)   dbgCondTimedWait(cond, mut, to, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT )
 #define d_free(x)      dbgFree(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT )
 #else
 #define d_pthread_mutex_lock(x)     pthread_mutex_lock(x)
+#define d_pthread_mutex_trylock(x)  pthread_mutex_trylock(x)
 #define d_pthread_mutex_unlock(x)   pthread_mutex_unlock(x)
 #define d_pthread_cond_wait(cond, mut)   pthread_cond_wait(cond, mut)
 #define d_pthread_cond_timedwait(cond, mut, to)   pthread_cond_timedwait(cond, mut, to)
diff --git a/runtime/wti.c b/runtime/wti.c
index 9b3450e6..544bffa7 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -245,10 +245,14 @@ wtiJoinThrd(wti_t *pThis)
 
 	ISOBJ_TYPE_assert(pThis, wti);
 	dbgprintf("waiting for worker %s termination, current state %d\n", wtiGetDbgHdr(pThis), pThis->tCurrCmd);
-	pthread_join(pThis->thrdID, NULL);
-	wtiSetState(pThis, eWRKTHRD_STOPPED, 0, MUTEX_ALREADY_LOCKED); /* back to virgin... */
-	pThis->thrdID = 0; /* invalidate the thread ID so that we do not accidently find reused ones */
+	if (pThis->thrdID == 0) {
+		dbgprintf("worker %s was already stopped\n", wtiGetDbgHdr(pThis));
+	} else {
+		pthread_join(pThis->thrdID, NULL);
+		wtiSetState(pThis, eWRKTHRD_STOPPED, 0, MUTEX_ALREADY_LOCKED); /* back to virgin... */
+		pThis->thrdID = 0; /* invalidate the thread ID so that we do not accidently find reused ones */
 	dbgprintf("worker %s has stopped\n", wtiGetDbgHdr(pThis));
+	}   
 
 	RETiRet;
 }
diff --git a/runtime/wtp.c b/runtime/wtp.c
index 41903576..9f54a9ab 100644
--- a/runtime/wtp.c
+++ b/runtime/wtp.c
@@ -84,6 +84,7 @@ static rsRetVal NotImplementedDummy() { return RS_RET_OK; }
 BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! */
 	pThis->bOptimizeUniProc = glbl.GetOptimizeUniProc();
 	pthread_mutex_init(&pThis->mut, NULL);
+	pthread_mutex_init(&pThis->mutThrdShutdwn, NULL);
 	pthread_cond_init(&pThis->condThrdTrm, NULL);
 	/* set all function pointers to "not implemented" dummy so that we can safely call them */
 	pThis->pfChkStopWrkr = NotImplementedDummy;
@@ -148,6 +149,7 @@ CODESTARTobjDestruct(wtp)
 	/* actual destruction */
 	pthread_cond_destroy(&pThis->condThrdTrm);
 	pthread_mutex_destroy(&pThis->mut);
+	pthread_mutex_destroy(&pThis->mutThrdShutdwn);
 
 	if(pThis->pszDbgHdr != NULL)
 		free(pThis->pszDbgHdr);
@@ -199,11 +201,23 @@ wtpProcessThrdChanges(wtp_t *pThis)
 	if(pThis->bThrdStateChanged == 0)
 		FINALIZE;
 
-	/* go through all threads */
-	for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
-		wtiProcessThrdChanges(pThis->pWrkr[i], LOCK_MUTEX);
+	if(d_pthread_mutex_trylock(&(pThis->mutThrdShutdwn)) != 0) {
+		/* another thread is already in the loop */
+		FINALIZE;
 	}
 
+	do {
+		/* reset the change marker */
+		pThis->bThrdStateChanged = 0;
+		/* go through all threads */
+		for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
+			wtiProcessThrdChanges(pThis->pWrkr[i], LOCK_MUTEX);
+		}
+	/* restart if another change occured while we were processing the changes */
+	} while(pThis->bThrdStateChanged != 0);
+
+	d_pthread_mutex_unlock(&(pThis->mutThrdShutdwn));
+
 finalize_it:
 	RETiRet;
 }
diff --git a/runtime/wtp.h b/runtime/wtp.h
index 88d88afd..b9cb07c5 100644
--- a/runtime/wtp.h
+++ b/runtime/wtp.h
@@ -61,6 +61,7 @@ typedef struct wtp_s {
 	int	bInactivityGuard;/* prevents inactivity due to race condition */
 	rsRetVal (*pConsumer)(void *); /* user-supplied consumer function for dewtpd messages */
 	/* synchronization variables */
+	pthread_mutex_t mutThrdShutdwn; /* mutex to guard thread shutdown processing */
 	pthread_mutex_t mut; /* mutex for the wtp's thread management */
 	pthread_cond_t condThrdTrm;/* signalled when threads terminate */
 	int bThrdStateChanged;	/* at least one thread state has changed if 1 */