bugfix: queue cancel cleanup handler could be called with invalid pointer

    if dequeue failed

5 files changed, 25 insertions, 15 deletions

diff --git a/ChangeLog b/ChangeLog
index 021189ac..52382c8e 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -5,8 +5,11 @@ Version 3.11.6 (rgerhards), 2008-02-??
 - enabled imgssapi to be loaded side-by-side with imtcp
 - added InputGSSServerPermitPlainTCP config directive
 - split imgssapi source code somewhat from imtcp
+- bugfix: queue cancel cleanup handler could be called with
+  invalid pointer if dequeue failed
 - bugfix: rsyslogd segfaulted on second SIGHUP
   tracker: http://bugzilla.adiscon.com/show_bug.cgi?id=38
+- improved stability of queuee engine
 - applied patch from varmojfekoj to fix an issue with compatibility 
   mode and default module directories (many thanks!):
   I've also noticed a bug in the compatibility code; the problem is that 
diff --git a/debug.c b/debug.c
index b3a6aa8a..5814509e 100644
--- a/debug.c
+++ b/debug.c
@@ -57,10 +57,10 @@ static dbgThrdInfo_t *dbgGetThrdInfo(void);
 /* static data (some time to be replaced) */
 int	Debug;		/* debug flag  - read-only after startup */
 int debugging_on = 0;	 /* read-only, except on sig USR1 */
-static int bLogFuncFlow = 0; /* shall the function entry and exit be logged to the debug log? */
+static int bLogFuncFlow = 1; /* shall the function entry and exit be logged to the debug log? */
 static int bLogAllocFree = 0; /* shall calls to (m/c)alloc and free be logged to the debug log? */
 static int bPrintFuncDBOnExit = 0; /* shall the function entry and exit be logged to the debug log? */
-static int bPrintMutexAction = 1; /* shall mutex calls be printed to the debug log? */
+static int bPrintMutexAction = 0; /* shall mutex calls be printed to the debug log? */
 static int bPrintTime = 1;	/* print a timestamp together with debug message */
 static char *pszAltDbgFileName = NULL; /* if set, debug output is *also* sent to here */
 static FILE *altdbg = NULL;	/* and the handle for alternate debug output */
@@ -930,8 +930,11 @@ int dbgEntrFunc(dbgFuncDB_t *pFuncDB, int line)
 	/* when we reach this point, we have a fully-initialized FuncDB! */
 
 	//if(bLogFuncFlow) /* quick debug hack...  select the best for you! */
-	if(bLogFuncFlow && !strcmp((char*)pFuncDB->file, "vm.c")) /* quick debug hack... select the best for you! */
+	//if(bLogFuncFlow && !strcmp((char*)pFuncDB->file, "vm.c")) /* quick debug hack... select the best for you! */
 	//if(bLogFuncFlow && !strcmp((char*)pFuncDB->file, "expr.c")) /* quick debug hack... select the best for you! */
+	if(bLogFuncFlow && (!strcmp((char*)pFuncDB->file, "wti.c")
+	                    ||!strcmp((char*)pFuncDB->file, "wtp.c")
+	                    ||!strcmp((char*)pFuncDB->file, "queue.c"))) /* quick debug hack... select the best for you! */
 		dbgprintf("%s:%d: %s: enter\n", pFuncDB->file, pFuncDB->line, pFuncDB->func);
 	if(pThrd->stackPtr >= (int) (sizeof(pThrd->callStack) / sizeof(dbgFuncDB_t*))) {
 		dbgprintf("%s:%d: %s: debug module: call stack for this thread full, suspending call tracking\n",
@@ -961,8 +964,11 @@ void dbgExitFunc(dbgFuncDB_t *pFuncDB, int iStackPtrRestore)
 
 	dbgFuncDBPrintActiveMutexes(pFuncDB, "WARNING: mutex still owned by us as we exit function, mutex: ", pthread_self());
 	//if(bLogFuncFlow) /* quick debug hack... select the best for you! */
-	if(bLogFuncFlow && !strcmp((char*)pFuncDB->file, "vm.c")) /* quick debug hack... select the best for you! */
+	//if(bLogFuncFlow && !strcmp((char*)pFuncDB->file, "vm.c")) /* quick debug hack... select the best for you! */
 	//if(bLogFuncFlow && !strcmp((char*)pFuncDB->file, "expr.c")) /* quick debug hack... select the best for you! */
+	if(bLogFuncFlow && (!strcmp((char*)pFuncDB->file, "wti.c")
+	                    ||!strcmp((char*)pFuncDB->file, "wtp.c")
+	                    ||!strcmp((char*)pFuncDB->file, "queue.c"))) /* quick debug hack... select the best for you! */
 		dbgprintf("%s:%d: %s: exit\n", pFuncDB->file, pFuncDB->line, pFuncDB->func);
 	pThrd->stackPtr = iStackPtrRestore;
 	if(pThrd->stackPtr < 0) {
diff --git a/queue.c b/queue.c
index 70ee6f1f..c54d3991 100644
--- a/queue.c
+++ b/queue.c
@@ -263,7 +263,6 @@ queueStartDA(queue_t *pThis)
 		CHKiRet(queueSettoQShutdown(pThis->pqDA, 1));
 	}
 
-dbgoprint((obj_t*) pThis, "queueStartDA pre start\n");
 	iRet = queueStart(pThis->pqDA);
 	/* file not found is expected, that means it is no previous QIF available */
 	if(iRet != RS_RET_OK && iRet != RS_RET_FILE_NOT_FOUND)
@@ -1385,7 +1384,18 @@ queueDequeueConsumable(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
 	queueChkPersist(pThis);
 	iQueueSize = queueGetOverallQueueSize(pThis); /* cache this for after mutex release */
 	bRunsDA = pThis->bRunsDA; /* cache this for after mutex release */
-	pWti->pUsrp = pUsr; /* save it for the cancel cleanup handler */
+
+	/* We now need to save the user pointer for the cancel cleanup handler, BUT ONLY
+	 * if we could successfully obtain a user pointer. Otherwise, we would bring the
+	 * cancel cleanup handler into big troubles (and we did ;)). Note that we can
+	 * NOT set the variable further below, as this may lead to an object leak. We 
+	 * may get cancelled before we reach that part of the code, so the only 
+	 * solution is to do it here. -- rgerhards, 2008-02-27
+	 */
+	if(iRet == RS_RET_OK) {
+		pWti->pUsrp = pUsr;
+	}
+
 	d_pthread_mutex_unlock(pThis->mut);
 	pthread_cond_signal(&pThis->notFull);
 	pthread_setcancelstate(iCancelStateSave, NULL);
@@ -1561,9 +1571,6 @@ queueRegOnWrkrShutdown(queue_t *pThis)
 	ISOBJ_TYPE_assert(pThis, queue);
 
 	if(pThis->pqParent != NULL) {
-RUNLOG_VAR("%p", pThis->pqParent->pWtpDA);
-if(pThis->pqParent->pWtpDA == NULL)
-	FINALIZE;
 		ASSERT(pThis->pqParent->pWtpDA != NULL);
 		pThis->pqParent->bChildIsDone = 1; /* indicate we are done */
 		wtpAdviseMaxWorkers(pThis->pqParent->pWtpDA, 1); /* reactivate DA worker (always 1) */
diff --git a/wti.c b/wti.c
index 402b6bda..e7120f8a 100644
--- a/wti.c
+++ b/wti.c
@@ -405,11 +405,9 @@ wtiWorker(wti_t *pThis)
 
 	/* indicate termination */
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-// TODO: this deadlocks with wtiProcessThrdChanges(), where it waits on the thread to join (on US TO JOIN!!!)
 	d_pthread_mutex_lock(&pThis->mut);
 	pthread_cleanup_pop(0); /* remove cleanup handler */
 
-RUNLOG_VAR("%p", pWtp->pUsr);
 	pWtp->pfOnWorkerShutdown(pWtp->pUsr);
 
 	wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED);
diff --git a/wtp.c b/wtp.c
index b1745865..599e63c3 100644
--- a/wtp.c
+++ b/wtp.c
@@ -323,8 +323,6 @@ wtpCancelAll(wtp_t *pThis)
 {
 	DEFiRet;
 	int i;
-	int numCancelled = 0;
-	/* TODO: mutex?? TODO: cancellation in wti (but OK as is [though ugly form an isolation POV]!) */
 
 	ISOBJ_TYPE_assert(pThis, wtp);
 
@@ -335,10 +333,8 @@ wtpCancelAll(wtp_t *pThis)
 	for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
 		dbgprintf("%s: try canceling worker thread %d\n", wtpGetDbgHdr(pThis), i);
 		wtiCancelThrd(pThis->pWrkr[i]);
-RUNLOG;
 	}
 
-	dbgprintf("%s: cancelled %d worker threads\n", wtpGetDbgHdr(pThis), numCancelled);
 	RETiRet;
 }