redesigned queue to utilize helper classes for threading support. This is

    finally in a running state for regular (non disk-assisted) queues, with
    a minor nit at shutdown. So I can finally commit the work again to
    CVS...

33 files changed, 1012 insertions, 1689 deletions

diff --git a/Makefile.am b/Makefile.am
index f15fe567..06602861 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -5,6 +5,8 @@ rfc3195d_SOURCES = rfc3195d.c rsyslog.h
 rsyslogd_SOURCES = \
 	syslogd.c \
 	syslogd.h \
+	debug.c \
+	debug.h \
 	glbl.h \
 	pidfile.c \
 	pidfile.h \
@@ -24,6 +26,10 @@ rsyslogd_SOURCES = \
 	threads.h \
 	stream.c \
 	stream.h \
+	wtp.c \
+	wtp.h \
+	wti.c \
+	wti.h \
 	queue.c \
 	queue.h \
 	sync.c \
diff --git a/action.c b/action.c
index 04e72a94..2410c76e 100644
--- a/action.c
+++ b/action.c
@@ -84,7 +84,7 @@ rsRetVal actionConstruct(action_t **ppThis)
 
 finalize_it:
 	*ppThis = pThis;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -97,7 +97,7 @@ static rsRetVal actionResume(action_t *pThis)
 	assert(pThis != NULL);
 	pThis->bSuspended = 0;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -121,7 +121,7 @@ rsRetVal actionSuspend(action_t *pThis)
 	pThis->ttResumeRtry = time(NULL) + pThis->iResumeInterval;
 	pThis->iNbrResRtry = 0; /* tell that we did not yet retry to resume */
 
-	return iRet;
+	RETiRet;
 }
 
 /* try to resume an action -- rgerhards, 2007-08-02
@@ -162,7 +162,7 @@ rsRetVal actionTryResume(action_t *pThis)
 	dbgprintf("actionTryResume: iRet: %d, next retry (if applicable): %u [now %u]\n",
 		iRet, (unsigned) pThis->ttResumeRtry, (unsigned) ttNow);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -187,7 +187,7 @@ rsRetVal actionDbgPrint(action_t *pThis)
 	printf("\tExec only when previous is suspended: %d\n", pThis->bExecWhenPrevSusp);
 	printf("\n");
 
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/cfsysline.c b/cfsysline.c
index 337d362a..7cbd5884 100644
--- a/cfsysline.c
+++ b/cfsysline.c
@@ -74,7 +74,7 @@ static rsRetVal doGetChar(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -92,7 +92,7 @@ static rsRetVal doCustomHdlr(uchar **pp, rsRetVal (*pSetHdlr)(uchar**, void*), v
 	CHKiRet(pSetHdlr(pp, pVal));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -128,7 +128,7 @@ static rsRetVal parseIntVal(uchar **pp, size_t *pVal)
 	*pp = p;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -158,7 +158,7 @@ static rsRetVal doGetInt(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *p
 	*pp = p;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -209,7 +209,7 @@ static rsRetVal doGetSize(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -270,7 +270,7 @@ static rsRetVal doFileCreateMode(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t),
 	*pp = p;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -347,7 +347,7 @@ static rsRetVal doGetGID(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *p
 	skipWhiteSpace(pp); /* skip over any whitespace */
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -389,7 +389,7 @@ static rsRetVal doGetUID(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *p
 	skipWhiteSpace(pp); /* skip over any whitespace */
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -420,7 +420,7 @@ static rsRetVal doBinaryOptionLine(uchar **pp, rsRetVal (*pSetHdlr)(void*, int),
 	skipWhiteSpace(pp); /* skip over any whitespace */
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -481,7 +481,7 @@ finalize_it:
 			rsCStrDestruct(pStrB);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -514,7 +514,7 @@ static rsRetVal cslchConstruct(cslCmdHdlr_t **ppThis)
 
 finalize_it:
 	*ppThis = pThis;
-	return iRet;
+	RETiRet;
 }
 
 /* destructor for linked list keys. As we do not use any dynamic memory,
@@ -606,7 +606,7 @@ static rsRetVal cslchCallHdlr(cslCmdHdlr_t *pThis, uchar **ppConfLine)
 	CHKiRet(pHdlr(ppConfLine, pThis->cslCmdHdlr, pThis->pData));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -655,7 +655,7 @@ static rsRetVal cslcConstruct(cslCmd_t **ppThis, int bChainingPermitted)
 
 finalize_it:
 	*ppThis = pThis;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -678,7 +678,7 @@ finalize_it:
 			cslchDestruct(pCmdHdlr);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -692,7 +692,7 @@ rsRetVal cfsyslineInit(void)
 	CHKiRet(llInit(&llCmdList, cslcDestruct, cslcKeyDestruct, strcasecmp));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -740,7 +740,7 @@ rsRetVal regCfSysLineHdlr(uchar *pCmdName, int bChainingPermitted, ecslCmdHdrlTy
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -774,7 +774,7 @@ DEFFUNC_llExecFunc(unregHdlrsHeadExec)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 /* unregister and destroy cfSysLineHandlers for a specific owner. This method is
  * most importantly used before unloading a loadable module providing some handlers.
@@ -790,7 +790,7 @@ rsRetVal unregCfSysLineHdlrs4Owner(void *pOwnerCookie)
 	 */
 	iRet = llExecFunc(&llCmdList, unregHdlrsHeadExec, pOwnerCookie);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -844,7 +844,7 @@ rsRetVal processCfSysLineCommand(uchar *pCmdName, uchar **p)
 		iRet = iRetLL;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/debug.c b/debug.c
index 1ccb3d60..e473504e 100644
--- a/debug.c
+++ b/debug.c
@@ -289,11 +289,6 @@ dbgMutLog_t *dbgMutLogAddEntry(pthread_mutex_t *pmut, short mutexOp, dbgFuncDB_t
 	pLog->pFuncDB = pFuncDB;
 
 	DLL_Add(MutLog, pLog);
-//RUNLOG_VAR("%p", pLog);
-//RUNLOG_VAR("%p", dbgMutLogListRoot);
-//RUNLOG_VAR("%p", dbgMutLogListLast);
-//RUNLOG_VAR("%p", pLog->pNext);
-//RUNLOG_VAR("%p", pLog->pPrev);
 
 	return pLog;
 }
@@ -520,7 +515,6 @@ int dbgCondTimedWait(pthread_cond_t *cond, pthread_mutex_t *pmut, const struct t
 	dbgprintf("%s:%d:%s: mutex %p waiting on condition %p (with timeout)\n", pFuncDB->file, pFuncDB->line, pFuncDB->func,
 		  (void*)pmut, (void*)cond);
 	ret = pthread_cond_timedwait(cond, pmut, abstime);
-RUNLOG;
 	dbgMutexLockLog(pmut, pFuncDB, ln);
 	return ret;
 }
diff --git a/doc/status.html b/doc/status.html
index 7950cfdb..226a2eb2 100644
--- a/doc/status.html
+++ b/doc/status.html
@@ -13,8 +13,8 @@
 rsyslog v3 compatibility document!</a></font></b><br>
 Documentation for 3.x is currently sparse. If you need assistance, please
 <a href="http://www.rsyslog.com/PNphpBB2.phtml">post in the rsyslog forums</a>!</p>
-<p><b>stable:</b> 2.0.0 - <a href="http://www.rsyslog.com/Article155.phtml">change log</a> -
-<a href="http://www.rsyslog.com/Downloads-index-req-getit-lid-70.phtml">download</a></p>
+<p><b>stable:</b> 2.0.1 - <a href="http://www.rsyslog.com/Article165.phtml">change log</a> -
+<a href="http://www.rsyslog.com/Downloads-index-req-getit-lid-73.phtml">download</a></p>
 <p>&nbsp;(<a href="version_naming.html">How are versions named?</a>)</p>
 <h2>Platforms</h2>
 <p>Thankfully, a number of folks have begin to build packages and help port 
diff --git a/iminternal.c b/iminternal.c
index 86d5097c..fb70d062 100644
--- a/iminternal.c
+++ b/iminternal.c
@@ -53,7 +53,7 @@ static rsRetVal iminternalDestruct(iminternal_t *pThis)
 
 	free(pThis);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -78,7 +78,7 @@ finalize_it:
 
 	*ppThis = pThis;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -111,7 +111,7 @@ finalize_it:
 			iminternalDestruct(pThis);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -142,7 +142,7 @@ rsRetVal iminternalRemoveMsg(int *pPri, msg_t **ppMsg, int *pFlags)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* tell the caller if we have any messages ready for processing.
@@ -166,7 +166,7 @@ rsRetVal modInitIminternal(void)
 
 	iRet = llInit(&llMsgs, iminternalDestruct, NULL, NULL);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -182,7 +182,7 @@ rsRetVal modExitIminternal(void)
 
 	iRet = llDestroy(&llMsgs);
 
-	return iRet;
+	RETiRet;
 }
 
 /*
diff --git a/linkedlist.c b/linkedlist.c
index ff1320e5..9adf40c4 100644
--- a/linkedlist.c
+++ b/linkedlist.c
@@ -79,7 +79,7 @@ static rsRetVal llDestroyElt(linkedList_t *pList, llElt_t *pElt)
 	free(pElt);
 	pList->iNumElts--; /* one less */
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -106,7 +106,7 @@ rsRetVal llDestroy(linkedList_t *pThis)
 	pThis->pRoot = NULL;
 	pThis->pLast = NULL;
 
-	return iRet;
+	RETiRet;
 }
 
 /* llDestroyRootElt - destroy the root element but otherwise
@@ -134,7 +134,7 @@ rsRetVal llDestroyRootElt(linkedList_t *pThis)
 	CHKiRet(llDestroyElt(pThis, pPrev));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -166,7 +166,7 @@ rsRetVal llGetNextElt(linkedList_t *pThis, linkedListCookie_t *ppElt, void **ppU
 
 	*ppElt = pElt;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -204,7 +204,7 @@ static rsRetVal llEltConstruct(llElt_t **ppThis, void *pKey, void *pData)
 
 finalize_it:
 	*ppThis = pThis;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -227,7 +227,7 @@ rsRetVal llAppend(linkedList_t *pThis, void *pKey, void *pData)
 	pThis->pLast = pElt;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -267,7 +267,7 @@ static rsRetVal llUnlinkAndDelteElt(linkedList_t *pThis, llElt_t *pElt, llElt_t
 	CHKiRet(llDestroyElt(pThis, pElt));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* find a user element based on the provided key - this is the
@@ -305,7 +305,7 @@ static rsRetVal llFindElt(linkedList_t *pThis, void *pKey, llElt_t **ppElt, llEl
 	} else
 		iRet = RS_RET_NOT_FOUND;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -323,7 +323,7 @@ rsRetVal llFind(linkedList_t *pThis, void *pKey, void **ppData)
 	*ppData = pElt->pData;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -345,7 +345,7 @@ rsRetVal llFindAndDelete(linkedList_t *pThis, void *pKey)
 	CHKiRet(llUnlinkAndDelteElt(pThis, pElt, pEltPrev));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -360,7 +360,7 @@ rsRetVal llGetNumElts(linkedList_t *pThis, int *piCnt)
 
 	*piCnt = pThis->iNumElts;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -404,7 +404,7 @@ rsRetVal llExecFunc(linkedList_t *pThis, rsRetVal (*pFunc)(void*, void*), void*
 		iRet = iRetLL;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/module-template.h b/module-template.h
index ecd6fe6d..c92ec25a 100644
--- a/module-template.h
+++ b/module-template.h
@@ -100,12 +100,13 @@ static rsRetVal createInstance(instanceData **ppData)\
 #define CODESTARTcreateInstance \
 	if((pData = calloc(1, sizeof(instanceData))) == NULL) {\
 		*ppData = NULL;\
+		ENDfunc \
 		return RS_RET_OUT_OF_MEMORY;\
 	}
 
 #define ENDcreateInstance \
 	*ppData = pData;\
-	return iRet;\
+	RETiRet;\
 }
 
 /* freeInstance()
@@ -129,7 +130,7 @@ static rsRetVal freeInstance(void* pModData)\
 #define ENDfreeInstance \
 	if(pData != NULL)\
 		free(pData); /* we need to free this in any case */\
-	return iRet;\
+	RETiRet;\
 }
 
 /* isCompatibleWithFeature()
@@ -137,12 +138,13 @@ static rsRetVal freeInstance(void* pModData)\
 #define BEGINisCompatibleWithFeature \
 static rsRetVal isCompatibleWithFeature(syslogFeature __attribute__((unused)) eFeat)\
 {\
-	rsRetVal iRet = RS_RET_INCOMPATIBLE;
+	rsRetVal iRet = RS_RET_INCOMPATIBLE; \
+	BEGINfunc
 
 #define CODESTARTisCompatibleWithFeature
 
 #define ENDisCompatibleWithFeature \
-	return iRet;\
+	RETiRet;\
 }
 
 /* doAction()
@@ -156,7 +158,7 @@ static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __at
 	/* ppString may be NULL if the output module requested no strings */
 
 #define ENDdoAction \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -174,7 +176,7 @@ static rsRetVal dbgPrintInstInfo(void *pModData)\
 	pData = (instanceData*) pModData;
 
 #define ENDdbgPrintInstInfo \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -189,13 +191,14 @@ static rsRetVal dbgPrintInstInfo(void *pModData)\
 static rsRetVal needUDPSocket(void *pModData)\
 {\
 	rsRetVal iRet = RS_RET_FALSE;\
-	instanceData *pData = NULL;
+	instanceData *pData = NULL; \
+	BEGINfunc
 
 #define CODESTARTneedUDPSocket \
 	pData = (instanceData*) pModData;
 
 #define ENDneedUDPSocket \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -245,7 +248,7 @@ finalize_it:\
 	}
 
 #define ENDparseSelectorAct \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -267,7 +270,7 @@ static rsRetVal tryResume(instanceData __attribute__((unused)) *pData)\
 	assert(pData != NULL);
 
 #define ENDtryResume \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -281,14 +284,16 @@ static rsRetVal queryEtryPt(uchar *name, rsRetVal (**pEtryPoint)())\
 	DEFiRet;
 
 #define CODESTARTqueryEtryPt \
-	if((name == NULL) || (pEtryPoint == NULL))\
+	if((name == NULL) || (pEtryPoint == NULL)) {\
+		ENDfunc \
 		return RS_RET_PARAM_ERROR;\
+	} \
 	*pEtryPoint = NULL;
 
 #define ENDqueryEtryPt \
 	if(iRet == RS_RET_OK)\
 		iRet = (*pEtryPoint == NULL) ? RS_RET_NOT_FOUND : RS_RET_OK;\
-	return iRet;\
+	RETiRet;\
 }
 
 /* the following definition is the standard block for queryEtryPt for all types
@@ -372,13 +377,15 @@ rsRetVal modInit##uniqName(int iIFVersRequested __attribute__((unused)), int *ip
 
 #define CODESTARTmodInit \
 	assert(pHostQueryEtryPt != NULL);\
-	if((pQueryEtryPt == NULL) || (ipIFVersProvided == NULL))\
-		return RS_RET_PARAM_ERROR;
+	if((pQueryEtryPt == NULL) || (ipIFVersProvided == NULL)) {\
+		ENDfunc \
+		return RS_RET_PARAM_ERROR; \
+	}
 
 #define ENDmodInit \
 finalize_it:\
 	*pQueryEtryPt = queryEtryPt;\
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -408,7 +415,7 @@ static rsRetVal modExit(void)\
 #define CODESTARTmodExit 
 
 #define ENDmodExit \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -423,10 +430,11 @@ static rsRetVal runInput(thrdInfo_t __attribute__((unused)) *pThrd)\
 {\
 	DEFiRet;
 
-#define CODESTARTrunInput 
+#define CODESTARTrunInput \
+	dbgSetThrdName((uchar*)__FILE__); /* we need to provide something better later */
 
 #define ENDrunInput \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -446,7 +454,7 @@ static rsRetVal willRun(void)\
 #define CODESTARTwillRun 
 
 #define ENDwillRun \
-	return iRet;\
+	RETiRet;\
 }
 
 
@@ -465,7 +473,7 @@ static rsRetVal afterRun(void)\
 #define CODESTARTafterRun 
 
 #define ENDafterRun \
-	return iRet;\
+	RETiRet;\
 }
 
 
diff --git a/modules.c b/modules.c
index b8fc54ef..5c252dd3 100644
--- a/modules.c
+++ b/modules.c
@@ -1,6 +1,6 @@
 /* modules.c
  * This is the implementation of syslogd modules object.
- * This object handles plug-ins and buil-in modules of all kind.
+ * This object handles plug-ins and build-in modules of all kind.
  *
  * File begun on 2007-07-22 by RGerhards
  *
@@ -99,7 +99,7 @@ rsRetVal queryHostEtryPt(uchar *name, rsRetVal (**pEtryPoint)())
 
 	if(iRet == RS_RET_OK)
 		iRet = (*pEtryPoint == NULL) ? RS_RET_NOT_FOUND : RS_RET_OK;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -204,7 +204,7 @@ static rsRetVal modPrepareUnload(modInfo_t *pThis)
 	/* END DEVEL */
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -288,7 +288,7 @@ finalize_it:
 			moduleDestruct(pNew);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 /* Print loaded modules. This is more or less a 
@@ -348,7 +348,7 @@ rsRetVal modUnloadAndDestructAll(void)
 		moduleDestruct(pModPrev);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -374,7 +374,7 @@ rsRetVal modUnloadAndDestructDynamic(void)
 		}
 	}
 
-	return iRet;
+	RETiRet;
 }
 /* vi:set ai:
  */
diff --git a/msg.c b/msg.c
index 0961afd4..3bdc1f92 100644
--- a/msg.c
+++ b/msg.c
@@ -225,7 +225,7 @@ rsRetVal MsgConstruct(msg_t **ppThis)
 	*ppThis = pM;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -422,7 +422,7 @@ static rsRetVal MsgSerialize(msg_t *pThis, strm_t *pStrm)
 	CHKiRet(objEndSerialize(pStrm));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -457,7 +457,7 @@ msg_t *MsgAddRef(msg_t *pM)
 static rsRetVal aquirePROCIDFromTAG(msg_t *pM)
 {
 	register int i;
-	int iRet;
+	DEFiRet;
 
 	assert(pM != NULL);
 	if(pM->pCSPROCID != NULL)
@@ -480,8 +480,9 @@ static rsRetVal aquirePROCIDFromTAG(msg_t *pM)
 		return RS_RET_OBJ_CREATION_FAILED; /* best we can do... */
 	rsCStrSetAllocIncrement(pM->pCSPROCID, 16);
 	while((i < pM->iLenTAG) && (pM->pszTAG[i] != ']')) {
-		if((iRet = rsCStrAppendChar(pM->pCSPROCID, pM->pszTAG[i])) != RS_RET_OK)
-			return iRet;
+		if((iRet = rsCStrAppendChar(pM->pCSPROCID, pM->pszTAG[i])) != RS_RET_OK) {
+			RETiRet;
+		}
 		++i;
 	}
 
@@ -497,10 +498,11 @@ static rsRetVal aquirePROCIDFromTAG(msg_t *pM)
 	}
 
 	/* OK, finaally we could obtain a PROCID. So let's use it ;) */
-	if((iRet = rsCStrFinish(pM->pCSPROCID)) != RS_RET_OK)
-		return iRet;
+	if((iRet = rsCStrFinish(pM->pCSPROCID)) != RS_RET_OK) {
+		RETiRet;
+	}
 
-	return RS_RET_OK;
+	RETiRet;
 }
 
 
@@ -523,7 +525,7 @@ static rsRetVal aquirePROCIDFromTAG(msg_t *pM)
 static rsRetVal aquireProgramName(msg_t *pM)
 {
 	register int i;
-	int iRet;
+	DEFiRet;
 
 	assert(pM != NULL);
 	if(pM->pCSProgName == NULL) {
@@ -538,13 +540,15 @@ static rsRetVal aquireProgramName(msg_t *pM)
 		      && (pM->pszTAG[i] != '\0') && (pM->pszTAG[i] != ':')
 		      && (pM->pszTAG[i] != '[')  && (pM->pszTAG[i] != '/')
 		    ; ++i) {
-			if((iRet = rsCStrAppendChar(pM->pCSProgName, pM->pszTAG[i])) != RS_RET_OK)
-				return iRet;
+			if((iRet = rsCStrAppendChar(pM->pCSProgName, pM->pszTAG[i])) != RS_RET_OK) {
+				RETiRet;
+			}
+		}
+		if((iRet = rsCStrFinish(pM->pCSProgName)) != RS_RET_OK) {
+			RETiRet;
 		}
-		if((iRet = rsCStrFinish(pM->pCSProgName)) != RS_RET_OK)
-			return iRet;
 	}
-	return RS_RET_OK;
+	RETiRet;
 }
 
 
@@ -924,7 +928,7 @@ rsRetVal MsgSetAPPNAME(msg_t *pMsg, char* pszAPPNAME)
 	/* if we reach this point, we have the object */
 	iRet = rsCStrSetSzStr(pMsg->pCSAPPNAME, (uchar*) pszAPPNAME);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2128,7 +2132,7 @@ rsRetVal MsgSetProperty(msg_t *pThis, property_t *pProp)
 		memcpy(&pThis->tTIMESTAMP, &pProp->val.vSyslogTime, sizeof(struct syslogTime));
 	}
 
-	return iRet;
+	RETiRet;
 }
 #undef	isProp
 
diff --git a/net.c b/net.c
index d5785e48..e5807ec3 100644
--- a/net.c
+++ b/net.c
@@ -214,7 +214,7 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 			 */
 			logerrorInt("Internal error caused AllowedSender to be ignored, AF = %d",
 				    iAllow->addr.NetAddr->sa_family);
-			return RS_RET_ERR;
+			ABORT_FINALIZE(RS_RET_ERR);
 		}
 		/* OK, entry constructed, now lets add it to the ACL list */
 		iRet = AddAllowedSenderEntry(ppRoot, ppLast, iAllow, iSignificantBits);
@@ -222,7 +222,7 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 		/* we need to process a hostname ACL */
 		if (DisableDNS) {
 			logerror ("Ignoring hostname based ACLs because DNS is disabled.");
-			return RS_RET_OK;
+			ABORT_FINALIZE(RS_RET_OK);
 		}
 		
 		if (!strchr (iAllow->addr.HostWildcard, '*') &&
@@ -246,10 +246,11 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 				
 				if (ACLAddHostnameOnFail) {
 				        logerrorSz("Adding hostname \"%s\" to ACL as a wildcard entry.", iAllow->addr.HostWildcard);
-				        return AddAllowedSenderEntry(ppRoot, ppLast, iAllow, iSignificantBits);
+				        iRet = AddAllowedSenderEntry(ppRoot, ppLast, iAllow, iSignificantBits);
+					FINALIZE;
 				} else {
 				        logerrorSz("Hostname \"%s\" WON\'T be added to ACL.", iAllow->addr.HostWildcard);
-				        return RS_RET_NOENTRY;
+				        ABORT_FINALIZE(RS_RET_NOENTRY);
 				}
 			}
 			
@@ -260,13 +261,13 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 					allowIP.flags = 0;
 					if((allowIP.addr.NetAddr = malloc(res->ai_addrlen)) == NULL) {
 						glblHadMemShortage = 1;
-						return RS_RET_OUT_OF_MEMORY;
+						ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 					}
 					memcpy(allowIP.addr.NetAddr, res->ai_addr, res->ai_addrlen);
 					
 					if((iRet = AddAllowedSenderEntry(ppRoot, ppLast, &allowIP, iSignificantBits))
 						!= RS_RET_OK)
-						return(iRet);
+						FINALIZE;
 					break;
 				case AF_INET6: /* IPv6 - but need to check if it is a v6-mapped IPv4 */
 					if(IN6_IS_ADDR_V4MAPPED (&SIN6(res->ai_addr)->sin6_addr)) {
@@ -277,7 +278,7 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 						if((allowIP.addr.NetAddr = malloc(sizeof(struct sockaddr_in)))
 						    == NULL) {
 							glblHadMemShortage = 1;
-							return RS_RET_OUT_OF_MEMORY;
+							ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 						}
 						SIN(allowIP.addr.NetAddr)->sin_family = AF_INET;
 #ifdef HAVE_STRUCT_SOCKADDR_SA_LEN    
@@ -291,7 +292,7 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 						if((iRet = AddAllowedSenderEntry(ppRoot, ppLast, &allowIP,
 								iSignificantBits))
 							!= RS_RET_OK)
-							return(iRet);
+							FINALIZE;
 					} else {
 						/* finally add IPv6 */
 						
@@ -299,14 +300,14 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 						allowIP.flags = 0;
 						if((allowIP.addr.NetAddr = malloc(res->ai_addrlen)) == NULL) {
 							glblHadMemShortage = 1;
-							return RS_RET_OUT_OF_MEMORY;
+							ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 						}
 						memcpy(allowIP.addr.NetAddr, res->ai_addr, res->ai_addrlen);
 						
 						if((iRet = AddAllowedSenderEntry(ppRoot, ppLast, &allowIP,
 								iSignificantBits))
 							!= RS_RET_OK)
-							return(iRet);
+							FINALIZE;
 					}
 					break;
 				}
@@ -321,7 +322,8 @@ static rsRetVal AddAllowedSender(struct AllowedSenders **ppRoot, struct AllowedS
 		}
 	}
 
-	return iRet;
+finalize_it:
+	RETiRet;
 }
 
 
@@ -700,7 +702,7 @@ rsRetVal gethname(struct sockaddr_storage *f, uchar *pszHostFQDN)
         }
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -833,7 +835,7 @@ rsRetVal cvthname(struct sockaddr_storage *f, uchar *pszHost, uchar *pszHostFQDN
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/obj-types.h b/obj-types.h
index 9d430b72..7eff503d 100644
--- a/obj-types.h
+++ b/obj-types.h
@@ -58,9 +58,11 @@ typedef enum {	/* IDs of known object "types/classes" */
 	OBJNull = 0,	/* no valid object (we do not start at zero so we can detect calloc()) */
 	OBJMsg = 1,
 	OBJstrm = 2,
-	OBJqueue = 3	/* remeber to UPDATE OBJ_NUM_IDS (below) if you add one! */
+	OBJwtp = 3,
+	OBJwti = 4,
+	OBJqueue = 5	/* remeber to UPDATE OBJ_NUM_IDS (below) if you add one! */
 } objID_t;	
-#define OBJ_NUM_IDS 4
+#define OBJ_NUM_IDS 6
 
 typedef enum {	/* IDs of base methods supported by all objects - used for jump table, so
 		 * they must start at zero and be incremented. -- rgerahrds, 2008-01-04
@@ -114,6 +116,31 @@ typedef struct obj {	/* the dummy struct that each derived class can be casted t
 #	define ISOBJ_assert(pObj)
 #endif
 
+#define DEFpropSetMethPTR(obj, prop, dataType)\
+	rsRetVal obj##Set##prop(obj##_t *pThis, dataType *pVal)\
+	{ \
+		/* DEV debug: dbgprintf("%sSet%s()\n", #obj, #prop); */\
+		pThis->prop = pVal; \
+		return RS_RET_OK; \
+	}
+#define PROTOTYPEpropSetMethPTR(obj, prop, dataType)\
+	rsRetVal obj##Set##prop(obj##_t *pThis, dataType*)
+#define DEFpropSetMeth(obj, prop, dataType)\
+	rsRetVal obj##Set##prop(obj##_t *pThis, dataType pVal)\
+	{ \
+		/* DEV debug: dbgprintf("%sSet%s()\n", #obj, #prop); */\
+		pThis->prop = pVal; \
+		return RS_RET_OK; \
+	}
+#define DEFpropSetMethFP(obj, prop, dataType)\
+	rsRetVal obj##Set##prop(obj##_t *pThis, dataType)\
+	{ \
+		/* DEV debug: dbgprintf("%sSet%s()\n", #obj, #prop); */\
+		pThis->prop = pVal; \
+		return RS_RET_OK; \
+	}
+#define PROTOTYPEpropSetMethFP(obj, prop, dataType)\
+	rsRetVal obj##Set##prop(obj##_t *pThis, dataType)
 #define DEFpropSetMeth(obj, prop, dataType)\
 	rsRetVal obj##Set##prop(obj##_t *pThis, dataType pVal)\
 	{ \
@@ -135,7 +162,7 @@ rsRetVal objName##ClassInit(void) \
 #define ENDObjClassInit(objName) \
 	objRegisterObj(OBJ##objName, pObjInfoOBJ); \
 finalize_it: \
-	return iRet; \
+	RETiRet; \
 }
 
 /* this defines both the constructor and initializer
@@ -148,7 +175,7 @@ finalize_it: \
 
 #define ENDobjConstruct(obj) \
 		/* use finalize_it: before calling the macro (if you need it)! */ \
-		return iRet; \
+		RETiRet; \
 	} \
 	rsRetVal obj##Construct(obj##_t **ppThis) \
 	{ \
@@ -166,8 +193,9 @@ finalize_it: \
 	 \
 	finalize_it: \
 		OBJCONSTRUCT_CHECK_SUCCESS_AND_CLEANUP \
-		return iRet; \
+		RETiRet; \
 	} 
 
 
+
 #endif /* #ifndef OBJ_TYPES_H_INCLUDED */
diff --git a/obj.c b/obj.c
index e28c2747..dfd3edd0 100644
--- a/obj.c
+++ b/obj.c
@@ -100,7 +100,7 @@ rsRetVal objInfoConstruct(objInfo_t **ppThis, objID_t objID, uchar *pszName, int
 	*ppThis = pThis;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -152,7 +152,7 @@ static rsRetVal objSerializeHeader(strm_t *pStrm, obj_t *pObj, uchar *pszRecType
 	CHKiRet(strmWriteChar(pStrm, '\n'));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -171,7 +171,7 @@ dbgprintf("objBeginSerialize obj type: %x\n", objGetObjID(pStrm));
 	CHKiRet(objSerializeHeader(pStrm, pObj, (uchar*) "Obj"));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 	
 
@@ -195,7 +195,7 @@ rsRetVal objBeginSerializePropBag(strm_t *pStrm, obj_t *pObj)
 	CHKiRet(objSerializeHeader(pStrm, pObj, (uchar*) "OPB"));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -286,7 +286,7 @@ rsRetVal objSerializeProp(strm_t *pStrm, uchar *pszPropName, propertyType_t prop
 	CHKiRet(strmWriteChar(pStrm, '\n'));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -307,7 +307,7 @@ rsRetVal objEndSerialize(strm_t *pStrm)
 	CHKiRet(strmRecordEnd(pStrm));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -334,7 +334,7 @@ static rsRetVal objDeserializeLong(long *pInt, strm_t *pStrm)
 
 	*pInt = i;
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -368,7 +368,7 @@ finalize_it:
 	if(iRet != RS_RET_OK && pCStr != NULL)
 		rsCStrDestruct(pCStr);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -400,7 +400,7 @@ static rsRetVal objDeserializeSyslogTime(syslogTime_t *pTime, strm_t *pStrm)
 	GETVAL(OffsetMinute);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 #undef GETVAL
 
@@ -443,7 +443,7 @@ static rsRetVal objDeserializeHeader(uchar *pszRecType, objID_t *poID, int* poVe
 	*poVers = oVers;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -515,7 +515,7 @@ static rsRetVal objDeserializeProperty(property_t *pProp, strm_t *pStrm)
 	if(c != '\n') ABORT_FINALIZE(RS_RET_INVALID_PROPFRAME);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -538,7 +538,7 @@ static rsRetVal objDeserializeTrailer(strm_t *pStrm)
 	NEXTC; if(c != '\n') ABORT_FINALIZE(RS_RET_INVALID_TRAILER);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -578,7 +578,7 @@ static rsRetVal objDeserializeTryRecover(strm_t *pStrm)
 
 finalize_it:
 	dbgprintf("deserializer has possibly been able to re-sync and recover, state %d\n", iRet);
-	return iRet;
+	RETiRet;
 }
 
 
@@ -608,7 +608,7 @@ static rsRetVal objDeserializeProperties(obj_t *pObj, objID_t oID, strm_t *pStrm
 
 	CHKiRet(objDeserializeTrailer(pStrm)); /* do trailer checks */
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -668,7 +668,7 @@ finalize_it:
 	if(iRet != RS_RET_OK && pObj != NULL)
 		free(pObj); // TODO: check if we can call destructor 2008-01-13 rger
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -711,7 +711,7 @@ dbgprintf("objDese...AsPropBag 2\n");
 	CHKiRet(objDeserializeProperties(pObj, oID, pStrm));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -756,7 +756,7 @@ rsRetVal objDeserializePropBag(obj_t *pObj, strm_t *pStrm)
 	CHKiRet(objDeserializeProperties(pObj, oID, pStrm));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 #undef NEXTC /* undef helper macro */
@@ -780,7 +780,7 @@ rsRetVal objRegisterObj(objID_t oID, objInfo_t *pInfo)
 	arrObjInfo[oID] = pInfo;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/obj.h b/obj.h
index 1f2aae63..6b965e10 100644
--- a/obj.h
+++ b/obj.h
@@ -85,20 +85,6 @@
 #define OBJSetMethodHandler(methodID, pHdlr) \
 	CHKiRet(objInfoSetMethod(pObjInfoOBJ, methodID, (rsRetVal (*)(void*)) pHdlr))
 
-/* debug aides */
-#if 0
-#define d_pthread_mutex_lock(x)   {dbgprintf("mutex %p   lock %s, %s(), line %d\n", (void*)x, __FILE__, __func__, __LINE__); \
-				   pthread_mutex_lock(x); \
-                                   if(1)dbgprintf("mutex %p   lock aquired %s, %s(), line %d\n",(void*)x, __FILE__, __func__, __LINE__); \
-				  }
-#define d_pthread_mutex_unlock(x) {dbgprintf("mutex %p UNlock %s, %s(), line %d\n", (void*)x ,__FILE__, __func__, __LINE__);\
-				   pthread_mutex_unlock(x); \
-                                   if(1)dbgprintf("mutex %p UNlock done %s, %s(), line %d\n", (void*)x, __FILE__, __func__, __LINE__); \
-				  }
-#else
-#define d_pthread_mutex_lock(x)     pthread_mutex_lock(x)
-#define d_pthread_mutex_unlock(x)   pthread_mutex_unlock(x)
-#endif
 /* prototypes */
 rsRetVal objInfoConstruct(objInfo_t **ppThis, objID_t objID, uchar *pszName, int iObjVers, rsRetVal (*pConstruct)(void *), rsRetVal (*pDestruct)(void *));
 rsRetVal objInfoSetMethod(objInfo_t *pThis, objMethod_t objMethod, rsRetVal (*pHandler)(void*));
diff --git a/objomsr.c b/objomsr.c
index 0158c856..6a617ad1 100644
--- a/objomsr.c
+++ b/objomsr.c
@@ -92,7 +92,7 @@ rsRetVal OMSRconstruct(omodStringRequest_t **ppThis, int iNumEntries)
 	
 abort_it:
 	*ppThis = pThis;
-	return iRet;
+	RETiRet;
 }
 
 /* set a template name and option to the object. Index must be given. The pTplName must be
diff --git a/omfile.c b/omfile.c
index bcfaeabc..36627574 100644
--- a/omfile.c
+++ b/omfile.c
@@ -172,7 +172,7 @@ rsRetVal setDynaFileCacheSize(void __attribute__((unused)) *pVal, int iNewVal)
 	iDynaFileCacheSize = iNewVal;
 	dbgprintf("DynaFileCacheSize changed to %d.\n", iNewVal);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -218,7 +218,7 @@ static rsRetVal cflineParseOutchannel(instanceData *pData, uchar* p, omodStringR
 			 "outchannel '%s' not found - ignoring action line",
 			 szBuf);
 		logerror(errMsg);
-		return RS_RET_NOT_FOUND;
+		ABORT_FINALIZE(RS_RET_NOT_FOUND);
 	}
 
 	/* check if there is a file name in the outchannel... */
@@ -229,7 +229,7 @@ static rsRetVal cflineParseOutchannel(instanceData *pData, uchar* p, omodStringR
 			 "outchannel '%s' has no file name template - ignoring action line",
 			 szBuf);
 		logerror(errMsg);
-		return RS_RET_ERR;
+		ABORT_FINALIZE(RS_RET_ERR);
 	}
 
 	/* OK, we finally got a correct template. So let's use it... */
@@ -242,7 +242,8 @@ static rsRetVal cflineParseOutchannel(instanceData *pData, uchar* p, omodStringR
 
 	iRet = cflineParseTemplateName(&p, pOMSR, iEntry, iTplOpts, (uchar*) " TradFmt");
 
-	return(iRet);
+finalize_it:
+	RETiRet;
 }
 
 
@@ -516,7 +517,7 @@ static rsRetVal writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pDa
 	 */
 	if(pData->bDynamicName) {
 		if(prepareDynFile(pData, ppString[1], iMsgOpts) != 0)
-			return RS_RET_ERR;
+			ABORT_FINALIZE(RS_RET_ERR);
 	}
 
 	/* create the message based on format specified */
@@ -541,7 +542,7 @@ again:
 					 pData->f_fname, (long long) pData->f_sizeLimit, (long long) actualFileSize);
 				errno = 0;
 				logerror(errMsg);
-				return RS_RET_DISABLE_ACTION;
+				ABORT_FINALIZE(RS_RET_DISABLE_ACTION);
 			} else {
 				snprintf(errMsg, sizeof(errMsg),
 					 "file %s had grown beyond configured file size of %lld bytes, actual size was %lld - configured command resolved situation",
@@ -558,14 +559,14 @@ again:
 		/* If a named pipe is full, just ignore it for now
 		   - mrn 24 May 96 */
 		if (pData->fileType == eTypePIPE && e == EAGAIN)
-			return RS_RET_OK;
+			ABORT_FINALIZE(RS_RET_OK);
 
 		/* If the filesystem is filled up, just ignore
 		 * it for now and continue writing when possible
 		 * based on patch for sysklogd by Martin Schulze on 2007-05-24
 		 */
 		if (pData->fileType == eTypeFILE && e == ENOSPC)
-			return RS_RET_OK;
+			ABORT_FINALIZE(RS_RET_OK);
 
 		(void) close(pData->fd);
 		/*
@@ -593,7 +594,9 @@ again:
 		}
 	} else if (pData->bSyncFile)
 		fsync(pData->fd);
-	return(iRet);
+
+finalize_it:
+	RETiRet;
 }
 
 
@@ -638,12 +641,15 @@ CODESTARTparseSelectorAct
 	 * the code further changes.  -- rgerhards, 2007-07-25
 	 */
 	if(*p == '$' || *p == '?' || *p == '|' || *p == '/' || *p == '-') {
-		if((iRet = createInstance(&pData)) != RS_RET_OK)
-			return iRet;
+		if((iRet = createInstance(&pData)) != RS_RET_OK) {
+			ENDfunc
+			return iRet; /* this can not use RET_iRet! */
+		}
 	} else {
 		/* this is not clean, but we need it for the time being
 		 * TODO: remove when cleaning up modularization 
 		 */
+		ENDfunc
 		return RS_RET_CONFLINE_UNPROCESSED;
 	}
 
diff --git a/omfwd.c b/omfwd.c
index a4a546df..66b8a055 100644
--- a/omfwd.c
+++ b/omfwd.c
@@ -201,7 +201,7 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
 		}
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 /* CODE FOR SENDING TCP MESSAGES */
@@ -241,7 +241,7 @@ static rsRetVal TCPSendFrame(void *pvData, char *msg, size_t len)
 		/* TODO: we need to revisit this code -- rgerhards, 2007-12-28 */
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -274,7 +274,7 @@ static rsRetVal TCPSendInit(void *pvData)
 			iRet = RS_RET_TCP_SOCKCREATE_ERR;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -322,7 +322,7 @@ static rsRetVal doTryResume(instanceData *pData)
 		break;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/parse.c b/parse.c
index f8c8e785..3132b570 100644
--- a/parse.c
+++ b/parse.c
@@ -179,7 +179,7 @@ rsRetVal parsInt(rsParsObj *pThis, int* pInt)
 rsRetVal parsSkipAfterChar(rsParsObj *pThis, char c)
 {
 	register unsigned char *pC;
-	rsRetVal iRet;
+	DEFiRet;
 
 	rsCHECKVALIDOBJECT(pThis, OIDrsPars);
 
@@ -203,7 +203,7 @@ rsRetVal parsSkipAfterChar(rsParsObj *pThis, char c)
 		iRet = RS_RET_NOT_FOUND;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 /* Skip whitespace. Often used to trim parsable entries.
@@ -243,12 +243,12 @@ rsRetVal parsDelimCStr(rsParsObj *pThis, rsCStrObj **ppCStr, char cDelim, int bT
 {
 	register unsigned char *pC;
 	rsCStrObj *pCStr;
-	rsRetVal iRet;
+	DEFiRet;
 
 	rsCHECKVALIDOBJECT(pThis, OIDrsPars);
 
 	if((pCStr = rsCStrConstruct()) == NULL)
-		return RS_RET_OUT_OF_MEMORY;
+		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 
 	if(bTrimLeading)
 		parsSkipWhitespace(pThis);
@@ -274,20 +274,21 @@ rsRetVal parsDelimCStr(rsParsObj *pThis, rsCStrObj **ppCStr, char cDelim, int bT
 	 */
 	if((iRet = rsCStrFinish(pCStr)) != RS_RET_OK) {
 		rsCStrDestruct (pCStr);
-		return(iRet);
+		FINALIZE;
 	}
 
 	if(bTrimTrailing) {
 		if((iRet = rsCStrTrimTrailingWhiteSpace(pCStr)) 
 		   != RS_RET_OK) {
 			rsCStrDestruct (pCStr);
-			return iRet;
+			FINALIZE;
 		}
 	}
 
 	/* done! */
 	*ppCStr = pCStr;
-	return RS_RET_OK;
+finalize_it:
+	RETiRet;
 }
 
 /* Parse a quoted string ("-some-data") from the given position.
@@ -309,17 +310,17 @@ rsRetVal parsQuotedCStr(rsParsObj *pThis, rsCStrObj **ppCStr)
 {
 	register unsigned char *pC;
 	rsCStrObj *pCStr;
-	rsRetVal iRet;
+	DEFiRet;
 
 	rsCHECKVALIDOBJECT(pThis, OIDrsPars);
 
 	if((iRet = parsSkipAfterChar(pThis, '"')) != RS_RET_OK)
-		return iRet;
+		FINALIZE;
 	pC = rsCStrGetBufBeg(pThis->pCStr) + pThis->iCurrPos;
 
 	/* OK, we most probably can obtain a value... */
 	if((pCStr = rsCStrConstruct()) == NULL)
-		return RS_RET_OUT_OF_MEMORY;
+		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 
 	while(pThis->iCurrPos < rsCStrLen(pThis->pCStr)) {
 		if(*pC == '"') {
@@ -334,13 +335,13 @@ rsRetVal parsQuotedCStr(rsParsObj *pThis, rsCStrObj **ppCStr)
 				 */
 				if((iRet = rsCStrAppendChar(pCStr, *pC)) != RS_RET_OK) {
 					rsCStrDestruct (pCStr);
-					return(iRet);
+					FINALIZE;
 				}
 			}
 		} else { /* regular character */
 			if((iRet = rsCStrAppendChar(pCStr, *pC)) != RS_RET_OK) {
 				rsCStrDestruct (pCStr);
-				return(iRet);
+				FINALIZE;
 			}
 		}
 		++pThis->iCurrPos;
@@ -352,18 +353,19 @@ rsRetVal parsQuotedCStr(rsParsObj *pThis, rsCStrObj **ppCStr)
 	} else {
 		/* error - improperly quoted string! */
 		rsCStrDestruct (pCStr);
-		return RS_RET_MISSING_TRAIL_QUOTE;
+		ABORT_FINALIZE(RS_RET_MISSING_TRAIL_QUOTE);
 	}
 
 	/* We got the string, let's finish it...  */
 	if((iRet = rsCStrFinish(pCStr)) != RS_RET_OK) {
 		rsCStrDestruct (pCStr);
-		return(iRet);
+		FINALIZE;
 	}
 
 	/* done! */
 	*ppCStr = pCStr;
-	return RS_RET_OK;
+finalize_it:
+	RETiRet;
 }
 
 /* 
@@ -382,7 +384,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 	uchar *pszTmp;
 	struct addrinfo hints, *res = NULL;
 	rsCStrObj *pCStr;
-	rsRetVal iRet;
+	DEFiRet;
 
 	rsCHECKVALIDOBJECT(pThis, OIDrsPars);
 	assert(pIP != NULL);
@@ -401,7 +403,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 	      && *pC != '/' && *pC != ',' && !isspace((int)*pC)) {
 		if((iRet = rsCStrAppendChar(pCStr, *pC)) != RS_RET_OK) {
 			rsCStrDestruct (pCStr);
-			return(iRet);
+			FINALIZE;
 		}
 		++pThis->iCurrPos;
 		++pC;
@@ -410,7 +412,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 	/* We got the string, let's finish it...  */
 	if((iRet = rsCStrFinish(pCStr)) != RS_RET_OK) {
 		rsCStrDestruct (pCStr);
-		return(iRet);
+		FINALIZE;
 	}
 
 	/* now we have the string and must check/convert it to
@@ -424,7 +426,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 		pszTmp = (uchar*)strchr ((char*)pszIP, ']');
 		if (pszTmp == NULL) {
 			free (pszIP);
-			return RS_RET_INVALID_IP;
+			ABORT_FINALIZE(RS_RET_INVALID_IP);
 		}
 		*pszTmp = '\0';
 
@@ -449,7 +451,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 		default:
 			free (pszIP);
 			free (*pIP);
-			return RS_RET_ERR;
+			ABORT_FINALIZE(RS_RET_ERR);
 		}
 		
 		if(*pC == '/') {
@@ -458,7 +460,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 			if((iRet = parsInt(pThis, pBits)) != RS_RET_OK) {
 				free (pszIP);
 				free (*pIP);
-				return(iRet);
+				FINALIZE;
 			}
 			/* we need to refresh pointer (changed by parsInt()) */
 			pC = rsCStrGetBufBeg(pThis->pCStr) + pThis->iCurrPos;
@@ -488,7 +490,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 		default:
 			free (pszIP);
 			free (*pIP);
-			return RS_RET_ERR;
+			ABORT_FINALIZE(RS_RET_ERR);
 		}
 			
 		if(*pC == '/') {
@@ -497,7 +499,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 			if((iRet = parsInt(pThis, pBits)) != RS_RET_OK) {
 				free (pszIP);
 				free (*pIP);
-				return(iRet);
+				FINALIZE;
 			}
 			/* we need to refresh pointer (changed by parsInt()) */
 			pC = rsCStrGetBufBeg(pThis->pCStr) + pThis->iCurrPos;
@@ -518,7 +520,7 @@ rsRetVal parsAddrWithBits(rsParsObj *pThis, struct NetAddr **pIP, int *pBits)
 	iRet = RS_RET_OK;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 #endif  /* #ifdef SYSLOG_INET */
 
diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index 863f0d50..530bb2d4 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -170,12 +170,11 @@ static rsRetVal readSocket(int fd, int bParseHost)
 	} else if (iRcvd < 0 && errno != EINTR) {
 		char errStr[1024];
 		strerror_r(errno, errStr, sizeof(errStr));
-		dbgprintf("UNIX socket error: %d = %s.\n", \
-			errno, errStr);
+		dbgprintf("UNIX socket error: %d = %s.\n", errno, errStr);
 		logerror("recvfrom UNIX");
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -228,7 +227,7 @@ CODESTARTrunInput
 		}
 	}
 
-	return iRet;
+	RETiRet;
 ENDrunInput
 
 
@@ -247,7 +246,7 @@ CODESTARTwillRun
 			dbgprintf("Opened UNIX socket '%s' (fd %d).\n", funixn[i], funix[i]);
 	}
 
-	return RS_RET_OK;
+	RETiRet;
 ENDwillRun
 
 
diff --git a/queue.c b/queue.c
index 89e0168f..c3666db2 100644
--- a/queue.c
+++ b/queue.c
@@ -52,491 +52,23 @@
 #include "stringbuf.h"
 #include "srUtils.h"
 #include "obj.h"
+#include "wtp.h"
+#include "wti.h"
 
 /* static data */
 DEFobjStaticHelpers
 
-/* debug aides */
-#if 0
-#define d_pthread_mutex_lock(x)   {dbgprintf("mutex %p   lock %s, %s(), line %d\n", x, __FILE__, __func__, __LINE__); \
-				   pthread_mutex_lock(x); \
-                                   if(1)dbgprintf("mutex %p   lock aquired %s, %s(), line %d\n", x, __FILE__, __func__, __LINE__); \
-				  }
-#define d_pthread_mutex_unlock(x) {dbgprintf("mutex %p UNlock %s, %s(), line %d\n", x ,__FILE__, __func__, __LINE__);\
-				   pthread_mutex_unlock(x); \
-                                   if(1)dbgprintf("mutex %p UNlock done %s, %s(), line %d\n", x, __FILE__, __func__, __LINE__); \
-				  }
-#else
-#define d_pthread_mutex_lock(x)     pthread_mutex_lock(x)
-#define d_pthread_mutex_unlock(x)   pthread_mutex_unlock(x)
-#endif
-
-
 /* forward-definitions */
 rsRetVal queueChkPersist(queue_t *pThis);
-static void *queueWorker(void *arg);
-static rsRetVal queueChkWrkThrdChanges(queue_t *pThis);
 static rsRetVal queueSetEnqOnly(queue_t *pThis, int bEnqOnly);
+static int queueChkStopWrkrDA(queue_t *pThis);
+static int queueIsIdleDA(queue_t *pThis);
+static rsRetVal queueConsumerDA(queue_t *pThis, wti_t *pWti, int iCancelStateSave);
+static rsRetVal queueConsumerCancelCleanup(void *arg1, void *arg2);
 
 /* methods */
 
 
-/* cancellation cleanup handler - frees provided mutex
- * rgerhards, 2008-01-14
- */
-static void queueMutexCleanup(void *arg)
-{
-	assert(arg != NULL);
-	d_pthread_mutex_unlock((pthread_mutex_t*) arg);
-}
-
-
-/* get the current worker state. For simplicity and speed, we have
- * NOT used our regular calling interface this time. I hope that won't
- * bite in the long term... -- rgerhards, 2008-01-17
- */
-static inline qWrkCmd_t
-qWrkrGetState(qWrkThrd_t *pThis)
-{
-	assert(pThis != NULL);
-	return pThis->tCurrCmd;
-}
-
-
-/* indicate worker thread startup
- * (it would be best if we could do this with an atomic operation)
- * rgerhards, 2008-01-19
- */
-static void
-queueWrkrThrdStartupIndication(queue_t *pThis)
-{
-	int iCancelStateSave;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	pThis->iCurNumWrkThrd++;
-	d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-}
-
-
-/* indicate worker thread shutdown
- * (it would be best if we could do this with an atomic operation)
- * rgerhards, 2008-01-19
- */
-static void
-queueWrkrThrdShutdownIndication(queue_t *pThis)
-{
-	int iCancelStateSave;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	pThis->iCurNumWrkThrd--;
-	pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
-	d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-}
-
-
-/* send a command to a specific thread
- */
-static rsRetVal
-qWrkrSetState(qWrkThrd_t *pThis, qWrkCmd_t tCmd)
-{
-	DEFiRet;
-
-	assert(pThis != NULL);
-
-dbgprintf("Queue 0x%lx: trying to  send command %d to thread %d\n", queueGetID(pThis->pQueue), tCmd, pThis->iThrd);
-	if(pThis->tCurrCmd == eWRKTHRD_SHUTDOWN_IMMEDIATE && tCmd != eWRKTHRD_TERMINATING)
-		FINALIZE;
-
-	dbgprintf("Queue 0x%lx: sending command %d to thread %d\n", queueGetID(pThis->pQueue), tCmd, pThis->iThrd);
-
-	/* change some admin structures */
-	switch(tCmd) {
-		case eWRKTHRD_TERMINATING:
-			pthread_cond_destroy(&pThis->condInitDone);
-			pthread_mutex_destroy(&pThis->mut);
-			dbgprintf("Queue 0x%lx/w%d: thread terminating with %d entries left in queue, %d workers running.\n",
-				  queueGetID(pThis->pQueue), pThis->iThrd, pThis->pQueue->iQueueSize,
-				  pThis->pQueue->iCurNumWrkThrd);
-			break;
-		case eWRKTHRD_RUN_CREATED:
-			pthread_cond_init(&pThis->condInitDone, NULL);
-			pthread_mutex_init(&pThis->mut, NULL);
-			break;
-		case eWRKTHRD_RUN_INIT:
-			break;
-		case eWRKTHRD_RUNNING:
-			pthread_cond_signal(&pThis->condInitDone);
-			break;
-		/* these cases just to satisfy the compiler, we do (yet) not act an them: */
-		case eWRKTHRD_STOPPED:
-		case eWRKTHRD_SHUTDOWN:
-		case eWRKTHRD_SHUTDOWN_IMMEDIATE:
-			/* DO NOTHING */
-			break;
-	}
-
-	pThis->tCurrCmd = tCmd;
-
-finalize_it:
-	return iRet;
-}
-
-/* send a command to a specific active thread. If the thread is not
- * active, the command is not sent.
- */
-static inline rsRetVal
-queueTellActWrkThrd(queue_t *pThis, int iIdx, qWrkCmd_t tCmd)
-{
-	DEFiRet;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	assert(iIdx >= 0 && iIdx <= pThis->iNumWorkerThreads);
-
-	if(pThis->pWrkThrds[iIdx].tCurrCmd >= eWRKTHRD_RUN_CREATED) {
-		qWrkrSetState(&pThis->pWrkThrds[iIdx], tCmd);
-	} else {
-		dbgprintf("Queue 0x%lx: command %d NOT sent to inactive thread %d\n", queueGetID(pThis), tCmd, iIdx);
-	}
-
-	return iRet;
-}
-
-
-/* Finalize construction of a wWrkrThrd_t "object"
- * rgerhards, 2008-01-17
- */
-static inline rsRetVal
-qWrkrConstructFinalize(qWrkThrd_t *pThis, queue_t *pQueue, int i)
-{
-	assert(pThis != NULL);
-	ISOBJ_TYPE_assert(pQueue, queue);
-
-	dbgprintf("Queue 0x%lx: finalizing construction of worker %d instance data\n", queueGetID(pQueue), i);
-
-	/* initialize our thread instance descriptor */
-	pThis = pQueue->pWrkThrds + i;
-	pThis->pQueue = pQueue;
-	pThis->iThrd = i;
-	pThis->pUsr = NULL;
-
-	qWrkrSetState(pThis, eWRKTHRD_STOPPED);
-
-	return RS_RET_OK;
-}
-
-
-/* Waits until the specified worker thread 
- * changed to full running state (aka have started up). This function
- * MUST NOT be called while the queue mutex is locked as it does
- * this itself. The wait is without timeout.
- * rgerhards, 2008-01-17
- */
-static inline rsRetVal
-qWrkrWaitStartup(qWrkThrd_t *pThis)
-{
-	int iCancelStateSave;
-
-	assert(pThis != NULL);
-
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(pThis->pQueue->mut);
-	if((pThis->tCurrCmd == eWRKTHRD_RUN_CREATED) || (pThis->tCurrCmd == eWRKTHRD_RUN_CREATED)) {
-		dbgprintf("Queue 0x%lx: waiting on worker thread %d startup\n", queueGetID(pThis->pQueue),
-			  pThis->iThrd);
-		pthread_cond_wait(&pThis->condInitDone, pThis->pQueue->mut);
-dbgprintf("worker startup done!\n");
-	}
-	d_pthread_mutex_unlock(pThis->pQueue->mut);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-
-	return RS_RET_OK;
-}
-
-
-/* waits until all worker threads that a currently initializing are fully started up
- * rgerhards, 2008-01-18
- */
-static rsRetVal
-qWrkrWaitAllWrkrStartup(queue_t *pThis)
-{
-	DEFiRet;
-	int i;
-
-	for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
-		qWrkrWaitStartup(pThis->pWrkThrds + i);
-	}
-
-	return iRet;
-}
-
-
-/* initialize the qWrkThrd_t structure - this MUST be called right after
- * startup of a worker thread. -- rgerhards, 2008-01-17
- */
-static inline rsRetVal
-qWrkrInit(qWrkThrd_t **ppThis, queue_t *pQueue)
-{
-	qWrkThrd_t *pThis;
-	int i;
-
-	assert(ppThis != NULL);
-	ISOBJ_TYPE_assert(pQueue, queue);
-
-	/* find myself in the queue's thread table */
-	for(i = 0 ; i <= pQueue->iNumWorkerThreads ; ++i)
-		if(pQueue->pWrkThrds[i].thrdID == pthread_self())
-			break;
-dbgprintf("Queue %p, worker start, thrd id found %x, idx %d, self %x\n", pQueue,
-	(unsigned) pQueue->pWrkThrds[i].thrdID, i, (unsigned) pthread_self());
-	assert(pQueue->pWrkThrds[i].thrdID == pthread_self());
-
-	/* initialize our thread instance descriptor */
-	pThis = pQueue->pWrkThrds + i;
-	pThis->pQueue = pQueue;
-	pThis->iThrd = i;
-	pThis->pUsr = NULL;
-
-	*ppThis = pThis;
-	qWrkrSetState(pThis, eWRKTHRD_RUN_INIT);
-
-	return RS_RET_OK;
-}
-
-
-/* join a specific worker thread
- */
-static inline rsRetVal
-queueJoinWrkThrd(queue_t *pThis, int iIdx)
-{
-	DEFiRet;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	assert(iIdx >= 0 && iIdx <= pThis->iNumWorkerThreads);
-	assert(pThis->pWrkThrds[iIdx].tCurrCmd != eWRKTHRD_STOPPED);
-
-	dbgprintf("Queue 0x%lx: thread %d state %d, waiting for exit\n", queueGetID(pThis), iIdx,
-		  pThis->pWrkThrds[iIdx].tCurrCmd);
-	pthread_join(pThis->pWrkThrds[iIdx].thrdID, NULL);
-	qWrkrSetState(&pThis->pWrkThrds[iIdx], eWRKTHRD_STOPPED); /* back to virgin... */
-	pThis->pWrkThrds[iIdx].thrdID = 0; /* invalidate the thread ID so that we do not accidently find reused ones */
-	dbgprintf("Queue 0x%lx: thread %d state %d, has stopped\n", queueGetID(pThis), iIdx,
-		  pThis->pWrkThrds[iIdx].tCurrCmd);
-
-	return iRet;
-}
-
-
-/* Starts a worker thread (on a specific index [i]!)
- */
-static inline rsRetVal
-queueStrtWrkThrd(queue_t *pThis, int i)
-{
-	DEFiRet;
-	int iState;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	assert(i >= 0 && i <= pThis->iNumWorkerThreads);
-	assert(pThis->pWrkThrds[i].tCurrCmd < eWRKTHRD_RUN_CREATED);
-
-	qWrkrSetState(&pThis->pWrkThrds[i], eWRKTHRD_RUN_CREATED);
-	iState = pthread_create(&(pThis->pWrkThrds[i].thrdID), NULL, queueWorker, (void*) pThis);
-	dbgprintf("Queue 0x%lx: starting Worker thread %x, index %d with state %d.\n",
-		  (unsigned long) pThis, (unsigned) pThis->pWrkThrds[i].thrdID, i, iState);
-
-	return iRet;
-}
-
-
-/* start the DA worker thread (if not already running)
- */
-static inline rsRetVal
-queueStrtDAWrkr(queue_t *pThis)
-{
-	DEFiRet;
-	int iCancelStateSave;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->pWrkThrds[0].mut);
-	if(pThis->pWrkThrds[0].tCurrCmd == eWRKTHRD_STOPPED) {
-		iRet = queueStrtWrkThrd(pThis, 0);
-	}
-	d_pthread_mutex_unlock(&pThis->pWrkThrds[0].mut);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-
-	return iRet;
-}
-
-/* Starts a *new* worker thread. Function searches itself for a free index spot. It must only
- * be called when we have less than max workers active. Pending wrkr thread requests MUST have
- * been processed before calling this function. -- rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueStrtNewWrkThrd(queue_t *pThis)
-{
-	DEFiRet;
-	int i;
-	int iStartingUp;
-	int iState;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-
-	/* find free spot in thread table. If we find at least one worker that is in initializiation,
-	 * we do NOT start a new one. Let's give the other one a chance, first.
-	 */
-	iStartingUp = -1;
-	for(i = 1 ; i <= pThis->iNumWorkerThreads ; ++i) {
-dbgprintf("Queue %p: search thrd tbl slot: i %d, CuccCmd %d\n", pThis, i, pThis->pWrkThrds[i].tCurrCmd);
-		if(pThis->pWrkThrds[i].tCurrCmd == eWRKTHRD_STOPPED) {
-			break;
-		} else if(pThis->pWrkThrds[i].tCurrCmd == eWRKTHRD_RUN_CREATED) {
-			iStartingUp = i;
-			break;
-		}
-	}
-
-dbgprintf("Queue %p: after thrd search: i %d, iStartingUp %d\n", pThis, i, iStartingUp);
-	if(iStartingUp > -1)
-		ABORT_FINALIZE(RS_RET_ALREADY_STARTING);
-
-	assert(i <= pThis->iNumWorkerThreads); /* now there must be a free spot, else something is really wrong! */
-
-	qWrkrSetState(&pThis->pWrkThrds[i], eWRKTHRD_RUN_CREATED);
-	iState = pthread_create(&(pThis->pWrkThrds[i].thrdID), NULL, queueWorker, (void*) pThis);
-	dbgprintf("Queue 0x%lx: Worker thread %x, index %d started with state %d.\n",
-		  (unsigned long) pThis, (unsigned) pThis->pWrkThrds[i].thrdID, i, iState);
-	/* we try to give the starting worker a little boost. It won't help much as we still
- 	 * hold the queue's mutex, but at least it has a chance to start on a single-CPU system.
- 	 */
-	pthread_yield();
-
-finalize_it:
-	return iRet;
-}
-
-
-/* send a command to all active worker threads. A start index can be
- * given. Usually, this is 0 or 1. Thread 0 is reserved to disk-assisted
- * mode and this start index take care of the special handling it needs to
- * receive. -- rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueTellActWrkThrds(queue_t *pThis, int iStartIdx, qWrkCmd_t tCmd)
-{
-	DEFiRet;
-	int i;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	assert(iStartIdx == 0 || iStartIdx == 1);
-
-	/* tell the workers our request */
-	for(i = iStartIdx ; i <= pThis->iNumWorkerThreads ; ++i)
-		if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRD_TERMINATING)
-			queueTellActWrkThrd(pThis, i, tCmd);
-
-	return iRet;
-}
-
-
-/* compute an absolute time timeout suitable for calls to pthread_cond_timedwait()
- * rgerhards, 2008-01-14
- */
-static rsRetVal
-queueTimeoutComp(struct timespec *pt, int iTimeout)
-{
-	assert(pt != NULL);
-	/* compute timeout */
-	clock_gettime(CLOCK_REALTIME, pt);
-	pt->tv_nsec += (iTimeout % 1000) * 1000000; /* think INTEGER arithmetic! */
-	if(pt->tv_nsec > 999999999) { /* overrun? */
-		pt->tv_nsec -= 1000000000;
-		++pt->tv_sec;
-	}
-	pt->tv_sec += iTimeout / 1000;
-	return RS_RET_OK; /* so far, this is static... */
-}
-
-
-/* wake up all worker threads. Param bWithDAWrk tells if the DA worker
- * is to be awaken, too. It needs special handling because it waits on
- * two different conditions depending on processing state.
- * rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueWakeupWrkThrds(queue_t *pThis, int bWithDAWrk)
-{
-	DEFiRet;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-
-	pthread_cond_broadcast(&pThis->notEmpty);
-	if(bWithDAWrk && pThis->qRunsDA != QRUNS_REGULAR) {
-		/* if running disk-assisted, workers may wait on that condition, too */
-		pthread_cond_broadcast(&pThis->condDA);
-	}
-
-	return iRet;
-}
-
-
-/* This function checks if (another) worker threads needs to be started. It
- * must be called while the caller holds a lock on the queue mutex. So it must not
- * do anything that either reaquires the mutex or forces somebody else to aquire
- * it (that would lead to a deadlock).
- * rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueChkAndStrtWrk(queue_t *pThis)
-{
-	DEFiRet;
-	int iCancelStateSave;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-
-	/* process any pending thread requests */
-	queueChkWrkThrdChanges(pThis);
-
-	if(pThis->bEnqOnly == 1)
-		FINALIZE;	/* in enqueue-only mode we have no workers */
-
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	pthread_cleanup_push(queueMutexCleanup, &pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-	/* check if we need to start up another worker */
-	if(pThis->qRunsDA == QRUNS_REGULAR) {
-		if(pThis->iCurNumWrkThrd < pThis->iNumWorkerThreads) {
-dbgprintf("Queue %p: less than max workers are running, qsize %d, workers %d, qRunsDA: %d\n",
-	   pThis, pThis->iQueueSize, pThis->iCurNumWrkThrd, pThis->qRunsDA);
-			/* check if we satisfy the min nbr of messages per worker to start a new one */
-			if(pThis->iCurNumWrkThrd == 0 ||
-			   pThis->iQueueSize / pThis->iCurNumWrkThrd > pThis->iMinMsgsPerWrkr) {
-				dbgprintf("Queue 0x%lx: high activity - starting additional worker thread.\n",
-					  queueGetID(pThis));
-				queueStrtNewWrkThrd(pThis);
-			}
-		}
-	} else {
-		if(pThis->iCurNumWrkThrd == 0 && pThis->bEnqOnly == 0) {
-dbgprintf("Queue %p: DA worker is no longer running, restarting, qsize %d, workers %d, qRunsDA: %d\n",
-	   pThis, pThis->iQueueSize, pThis->iCurNumWrkThrd, pThis->qRunsDA);
-			/* DA worker has timed out and needs to be restarted */
-			iRet = queueStrtDAWrkr(pThis);
-		}
-	}
-	pthread_cleanup_pop(1);
-
-finalize_it:
-	return iRet;
-}
-
 
 /* --------------- code for disk-assisted (DA) queue modes -------------------- */
 
@@ -555,7 +87,7 @@ queueTurnOffDAMode(queue_t *pThis)
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, queue);
-	assert(pThis->qRunsDA != QRUNS_REGULAR);
+	assert(pThis->bRunsDA);
 
 	/* if we need to pull any data that we still need from the (child) disk queue,
 	 * now would be the time to do so. At present, we do not need this, but I'd like to
@@ -572,62 +104,18 @@ queueTurnOffDAMode(queue_t *pThis)
 dbgprintf("Queue 0x%lx: disk-assistance being been turned off, bEnqOnly %d, bQueInDestr %d, NumWrkd %d\n",
 	  queueGetID(pThis), 
 	pThis->bEnqOnly,pThis->bQueueInDestruction,pThis->iCurNumWrkThrd);
-	// TODO: think about this code - there is a race
-	if(pThis->bEnqOnly == 0 && pThis->bQueueInDestruction == 0 && pThis->iCurNumWrkThrd < 2)
-		queueStrtNewWrkThrd(pThis);
-	pThis->qRunsDA = QRUNS_REGULAR; /* tell the world we are back in non-DA mode */
+	// TODO: mutex?
+	pThis->bRunsDA = 0; /* tell the world we are back in non-DA mode */
 
 	/* we destruct the queue object, which will also shutdown the queue worker. As the queue is empty,
 	 * this will be quick.
 	 */
 	queueDestruct(&pThis->pqDA); /* and now we are ready to destruct the DA queue */
 
-	/* now free the remaining resources */
-	pthread_mutex_destroy(&pThis->mutDA);
-	pthread_cond_destroy(&pThis->condDA);
-
-	queueTellActWrkThrd(pThis, 0, eWRKTHRD_SHUTDOWN_IMMEDIATE);/* finally, tell ourselves to shutdown */
 	dbgprintf("Queue 0x%lx: disk-assistance has been turned off, disk queue was empty (iRet %d)\n",
 		  queueGetID(pThis), iRet);
 
-	return iRet;
-}
-
-/* check if we had any worker thread changes and, if so, act
- * on them. At a minimum, terminated threads are harvested (joined).
- * This function MUST NEVER block on the queue mutex!
- */
-static rsRetVal
-queueChkWrkThrdChanges(queue_t *pThis)
-{
-	DEFiRet;
-	int i;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-
-	if(pThis->bThrdStateChanged == 0)
-		FINALIZE;
-
-	/* go through all threads (including DA thread) */
-	for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
-		switch(pThis->pWrkThrds[i].tCurrCmd) {
-			case eWRKTHRD_TERMINATING:
-				queueJoinWrkThrd(pThis, i);
-				break;
-			/* these cases just to satisfy the compiler, we do not act an them: */
-			case eWRKTHRD_STOPPED:
-			case eWRKTHRD_RUN_CREATED:
-			case eWRKTHRD_RUN_INIT:
-			case eWRKTHRD_RUNNING:
-			case eWRKTHRD_SHUTDOWN:
-			case eWRKTHRD_SHUTDOWN_IMMEDIATE:
-				/* DO NOTHING */
-				break;
-		}
-	}
-
-finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -644,6 +132,7 @@ queueChkIsDA(queue_t *pThis)
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, queue);
+RUNLOG_VAR("%s", pThis->pszFilePrefix);
 	if(pThis->pszFilePrefix != NULL) {
 		pThis->bIsDA = 1;
 		dbgprintf("Queue 0x%lx: is disk-assisted, disk will be used on demand\n", queueGetID(pThis));
@@ -651,7 +140,7 @@ queueChkIsDA(queue_t *pThis)
 		dbgprintf("Queue 0x%lx: is NOT disk-assisted\n", queueGetID(pThis));
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -660,11 +149,11 @@ queueChkIsDA(queue_t *pThis)
  * chore of this function is to create the DA queue object. If that function fails,
  * the DA worker should return with an appropriate state, which in turn should lead to
  * a re-set to non-DA mode in the Enq process. The queue mutex must be locked when this
- * function is called, else a race on pThis->qRunsDA may happen.
+ * function is called, else a race on pThis->bRunsDA may happen.
  * rgerhards, 2008-01-15
  */
 static rsRetVal
-queueStrtDA(queue_t *pThis)
+queueStartDA(queue_t *pThis)
 {
 	DEFiRet;
 
@@ -682,30 +171,19 @@ dbgprintf("Queue %p: queueSTrtDA after child queue construct, q %p\n", pThis, pT
 	/* as the created queue is the same object class, we take the
 	 * liberty to access its properties directly.
 	 */
-	pThis->pqDA->condSignalOnEmpty = &pThis->condDA;
-	pThis->pqDA->mutSignalOnEmpty = &pThis->mutDA;
-	pThis->pqDA->condSignalOnEmpty2 = &pThis->notEmpty;
-	pThis->pqDA->bSignalOnEmpty = 2;
 	pThis->pqDA->pqParent = pThis;
-dbgprintf("Queue %p: queueSTrtDA after assign\n", pThis);
 
 	CHKiRet(queueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize));
 	CHKiRet(queueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix));
 	CHKiRet(queueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt));
 	CHKiRet(queueSettoActShutdown(pThis->pqDA, pThis->toActShutdown));
 	CHKiRet(queueSettoEnq(pThis->pqDA, pThis->toEnq));
-dbgprintf("Queue %p: queueSTrtDA 10\n", pThis);
 	CHKiRet(queueSetEnqOnly(pThis->pqDA, pThis->bDAEnqOnly));
-dbgprintf("Queue %p: queueSTrtDA 15\n", pThis);
 	CHKiRet(queueSetiHighWtrMrk(pThis->pqDA, 0));
-dbgprintf("Queue %p: queueSTrtDA 20\n", pThis);
 	CHKiRet(queueSetiDiscardMrk(pThis->pqDA, 0));
-dbgprintf("Queue %p: queueSTrtDA 25\n", pThis);
 	if(pThis->toQShutdown == 0) {
-dbgprintf("Queue %p: queueSTrtDA 30a\n", pThis);
 		CHKiRet(queueSettoQShutdown(pThis->pqDA, 0)); /* if the user really wants... */
 	} else {
-dbgprintf("Queue %p: queueSTrtDA 30b\n", pThis);
 		/* we use the shortest possible shutdown (0 is endless!) because when we run on disk AND
 		 * have an obviously large backlog, we can't finish it in any case. So there is no point
 		 * in holding shutdown longer than necessary. -- rgerhards, 2008-01-15
@@ -713,30 +191,21 @@ dbgprintf("Queue %p: queueSTrtDA 30b\n", pThis);
 		CHKiRet(queueSettoQShutdown(pThis->pqDA, 1));
 	}
 
-dbgprintf("Queue %p: queueSTrtDA pre start\n", pThis);
+dbgprintf("Queue %p: queueStartDA pre start\n", pThis);
 	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)
 		FINALIZE; /* something is wrong */
 
-	/* tell our fellow workers to shut down
-	 * NOTE: we do NOT join them by intension! If we did, we would hold draining
-	 * the queue until some potentially long-running actions are finished. Having
-	 * the ability to immediatly drain the queue was the primary intension of
-	 * reserving worker thread 0 for DA queues. So if we would join the other
-	 * workers, we would screw up and do against our design goal.
-	 */
-	CHKiRet(queueTellActWrkThrds(pThis, 1, eWRKTHRD_SHUTDOWN_IMMEDIATE));
-
 	/* as we are right now starting DA mode because we are so busy, it is
-	 * extremely unlikely that any worker is sleeping on empty queue. HOWEVER,
+	 * extremely unlikely that any regular worker is sleeping on empty queue. HOWEVER,
 	 * we want to be on the safe side, and so we awake anyone that is waiting
 	 * on one. So even if the scheduler plays badly with us, things should be
 	 * quite well. -- rgerhards, 2008-01-15
 	 */
-	queueWakeupWrkThrds(pThis, 0); /* awake all workers, but not ourselves ;) */
+	wtpWakeupWrkr(pThis->pWtpReg); /* awake all workers, but not ourselves ;) */
 
-	pThis->qRunsDA = QRUNS_DA;	/* we are now in DA mode! */
+	pThis->bRunsDA = 1;	/* we are now in DA mode! */
 
 	dbgprintf("Queue 0x%lx: is now running in disk assisted mode, disk queue 0x%lx\n",
 		  queueGetID(pThis), queueGetID(pThis->pqDA));
@@ -751,31 +220,63 @@ finalize_it:
 		pThis->bIsDA = 0;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
 /* initiate DA mode
  * param bEnqOnly tells if the disk queue is to be run in enqueue-only mode. This may
  * be needed during shutdown of memory queues which need to be persisted to disk.
+ * If this function fails (should not happen), DA mode is not turned on.
  * rgerhards, 2008-01-16
  */
 static inline rsRetVal
-queueInitDA(queue_t *pThis, int bEnqOnly)
+queueInitDA(queue_t *pThis, int bEnqOnly, int bLockMutex)
 {
 	DEFiRet;
+	DEFVARS_mutexProtection;
+	uchar pszBuf[64];
+	size_t lenBuf;
+
+	BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, bLockMutex);
+	/* check if we already have a DA worker pool. If not, initiate one. Please note that the
+	 * pool is created on first need but never again destructed (until the queue is). This
+	 * is intentional. We assume that when we need it once, we may also need it on another
+	 * occasion. Ressources used are quite minimal when no worker is running.
+	 * rgerhards, 2008-01-24
+	 */
+	if(pThis->pWtpDA == NULL) {
+		lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "Queue 0x%lx/DA", (unsigned long) pThis);
+		CHKiRet(wtpConstruct		(&pThis->pWtpDA));
+		CHKiRet(wtpSetDbgHdr		(pThis->pWtpDA, pszBuf, lenBuf));
+		CHKiRet(wtpSetpfChkStopWrkr	(pThis->pWtpDA, queueChkStopWrkrDA));
+		CHKiRet(wtpSetpfIsIdle		(pThis->pWtpDA, queueIsIdleDA));
+		CHKiRet(wtpSetpfDoWork		(pThis->pWtpDA, queueConsumerDA));
+		CHKiRet(wtpSetpfOnWorkerCancel	(pThis->pWtpDA, queueConsumerCancelCleanup));
+		CHKiRet(wtpSetpfOnWorkerStartup	(pThis->pWtpDA, queueStartDA));
+		CHKiRet(wtpSetpfOnWorkerShutdown(pThis->pWtpDA, queueTurnOffDAMode));
+		CHKiRet(wtpSetpmutUsr		(pThis->pWtpDA, pThis->mut));
+		CHKiRet(wtpSetpcondBusy		(pThis->pWtpDA, &pThis->notEmpty));
+		CHKiRet(wtpSetiNumWorkerThreads	(pThis->pWtpDA, 1));
+		CHKiRet(wtpSetpUsr		(pThis->pWtpDA, pThis));
+		CHKiRet(wtpConstructFinalize	(pThis->pWtpDA));
+	}
+	/* if we reach this point, we have a "good" DA worker pool */
 
 	/* indicate we now run in DA mode - this is reset by the DA worker if it fails */
-	pThis->qRunsDA = QRUNS_DA_INIT;
+	pThis->bRunsDA = 1;
 	pThis->bDAEnqOnly = bEnqOnly;
 
-	/* now we must start our DA worker thread - it does the rest of the initialization
-	 * In enqueue-only mode, we do not start any workers.
+	/* now we must now adivse the wtp that we need one worker. If none is yet active,
+	 * that will also start one up. If we forgot that step, everything would be stalled
+	 * until the next enqueue request.
 	 */
 	if(pThis->bEnqOnly == 0)
-		iRet = queueStrtDAWrkr(pThis);
+		wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* one worker only for disk queues! */
 
-	return iRet;
+finalize_it:
+	END_MTX_PROTECTED_OPERATIONS(pThis->mut);
+	RETiRet;
 }
 
 
@@ -783,7 +284,7 @@ queueInitDA(queue_t *pThis, int bEnqOnly)
  * keep it running if we are already in it.
  * rgerhards, 2008-01-14
  */
-static rsRetVal
+static inline rsRetVal
 queueChkStrtDA(queue_t *pThis)
 {
 	DEFiRet;
@@ -794,35 +295,31 @@ queueChkStrtDA(queue_t *pThis)
 	if(pThis->iQueueSize != pThis->iHighWtrMrk)
 		ABORT_FINALIZE(RS_RET_OK);
 
-	if(pThis->qRunsDA != QRUNS_REGULAR) {
+dbgprintf("Queue %p: chkStartDA\n", pThis);
+	if(pThis->bRunsDA) {
 		/* then we need to signal that we are at the high water mark again. If that happens
 		 * on our way down the queue, that doesn't matter, because then nobody is waiting
 		 * on the condition variable.
+		 * (Remember that a DA queue stops draining the queue once it has reached the low
+		 * water mark and restarts it when the high water mark is reached again - this is
+		 * what this code here is responsible for. Please note that all workers may have been
+		 * terminated due to the inactivity timeout, thus we need to advise the pool that
+		 * we need at least one).
 		 */
 		dbgprintf("Queue 0x%lx: %d entries - passed high water mark in DA mode, send notify\n",
 			  queueGetID(pThis), pThis->iQueueSize);
-		//pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-		// TODO: mutex call order check! this must aquire the queue mutex
-		//d_pthread_mutex_lock(&pThis->mutDA);
-		pthread_cond_signal(&pThis->condDA);
-		//d_pthread_mutex_unlock(&pThis->mutDA);
-		//pthread_setcancelstate(iCancelStateSave, NULL);
-		queueChkWrkThrdChanges(pThis); /* the queue mode may have changed while we waited, so check! */
-	}
-
-	/* we need to re-check if we run disk-assisted, because that status may have changed
-	 * in our high water mark processing.
-	 */
-	if(pThis->qRunsDA == QRUNS_REGULAR) {
-		/* if we reach this point, we are NOT currently running in DA mode. */
+		wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* run again [see comment above] ;) */
+	} else {
+		/* this is the case when we are currently not running in DA mode. So it is time
+		 * to turn it back on.
+		 */
 		dbgprintf("Queue 0x%lx: %d entries - passed high water mark for disk-assisted mode, initiating...\n",
 			  queueGetID(pThis), pThis->iQueueSize);
-
-		queueInitDA(pThis, QUEUE_MODE_ENQDEQ); /* initiate DA mode */
+		queueInitDA(pThis, QUEUE_MODE_ENQDEQ, MUTEX_ALREADY_LOCKED); /* initiate DA mode */
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -855,7 +352,7 @@ static rsRetVal qConstructFixedArray(queue_t *pThis)
 	queueChkIsDA(pThis);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -868,7 +365,7 @@ static rsRetVal qDestructFixedArray(queue_t *pThis)
 	if(pThis->tVars.farray.pBuf != NULL)
 		free(pThis->tVars.farray.pBuf);
 
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qAddFixedArray(queue_t *pThis, void* in)
@@ -881,7 +378,7 @@ static rsRetVal qAddFixedArray(queue_t *pThis, void* in)
 	if (pThis->tVars.farray.tail == pThis->iMaxQueueSize)
 		pThis->tVars.farray.tail = 0;
 
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qDelFixedArray(queue_t *pThis, void **out)
@@ -895,7 +392,7 @@ static rsRetVal qDelFixedArray(queue_t *pThis, void **out)
 	if (pThis->tVars.farray.head == pThis->iMaxQueueSize)
 		pThis->tVars.farray.head = 0;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -911,7 +408,7 @@ static rsRetVal qConstructLinkedList(queue_t *pThis)
 
 	queueChkIsDA(pThis);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -925,7 +422,7 @@ static rsRetVal qDestructLinkedList(queue_t __attribute__((unused)) *pThis)
 	 * dynamic left with the linked list.
 	 */
 
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qAddLinkedList(queue_t *pThis, void* pUsr)
@@ -949,7 +446,7 @@ static rsRetVal qAddLinkedList(queue_t *pThis, void* pUsr)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qDelLinkedList(queue_t *pThis, void **ppUsr)
@@ -971,7 +468,7 @@ static rsRetVal qDelLinkedList(queue_t *pThis, void **ppUsr)
 	}
 	free(pEntry);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -986,7 +483,7 @@ queueLoadPersStrmInfoFixup(strm_t *pStrm, queue_t *pThis)
 	ISOBJ_TYPE_assert(pThis, queue);
 	CHKiRet(strmSetDir(pStrm, glblGetWorkDir(), strlen((char*)glblGetWorkDir())));
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1024,7 +521,7 @@ queueHaveQIF(queue_t *pThis)
 	/* If we reach this point, we have a .qi file */
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1092,7 +589,7 @@ finalize_it:
 			  queueGetID(pThis), iRet);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1150,7 +647,7 @@ CHKiRet(strmSetiMaxFileSize(pThis->tVars.disk.pWrite, pThis->iMaxFileSize));
 CHKiRet(strmSetiMaxFileSize(pThis->tVars.disk.pRead, pThis->iMaxFileSize));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1166,7 +663,7 @@ static rsRetVal qDestructDisk(queue_t *pThis)
 	if(pThis->pszSpoolDir != NULL)
 		free(pThis->pszSpoolDir);
 
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qAddDisk(queue_t *pThis, void* pUsr)
@@ -1179,7 +676,7 @@ static rsRetVal qAddDisk(queue_t *pThis, void* pUsr)
 	CHKiRet(strmFlush(pThis->tVars.disk.pWrite));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qDelDisk(queue_t *pThis, void **ppUsr)
@@ -1216,7 +713,7 @@ static rsRetVal qAddDirect(queue_t *pThis, void* pUsr)
 			      * queue anything ;)
 			      */
 
-	return iRet;
+	RETiRet;
 }
 
 static rsRetVal qDelDirect(queue_t __attribute__((unused)) *pThis, __attribute__((unused)) void **out)
@@ -1228,6 +725,7 @@ static rsRetVal qDelDirect(queue_t __attribute__((unused)) *pThis, __attribute__
 /* --------------- end type-specific handlers -------------------- */
 
 
+
 /* generic code to add a queue entry */
 static rsRetVal
 queueAdd(queue_t *pThis, void *pUsr)
@@ -1242,7 +740,7 @@ queueAdd(queue_t *pThis, void *pUsr)
 	dbgprintf("Queue 0x%lx: entry added, size now %d entries\n", queueGetID(pThis), pThis->iQueueSize);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1265,121 +763,31 @@ queueDel(queue_t *pThis, void *pUsr)
 	dbgprintf("Queue 0x%lx: entry deleted, state %d, size now %d entries\n",
 		  queueGetID(pThis), iRet, pThis->iQueueSize);
 
-	return iRet;
-}
-
-
-/* Send a shutdown command to all workers and awake them. This function
- * does NOT wait for them to terminate. Set bIncludeDAWRk to send the
- * termination command to the DA worker, too (else this does not happen).
- * rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueWrkThrdReqTrm(queue_t *pThis, qWrkCmd_t tShutdownCmd, int bIncludeDAWrk)
-{
-	DEFiRet;
-
-	if(bIncludeDAWrk) {
-		queueTellActWrkThrds(pThis, 0, tShutdownCmd);/* first tell the workers our request */
-		queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
-	} else {
-		queueTellActWrkThrds(pThis, 1, tShutdownCmd);/* first tell the workers our request */
-		queueWakeupWrkThrds(pThis, 0); /* awake all workers but not DA-worker */
-	}
-
-	return iRet;
-}
-
-
-/* Send a shutdown command to all workers and see if they terminate.
- * A timeout may be specified.
- * rgerhards, 2008-01-14
- */
-static rsRetVal
-queueWrkThrdTrm(queue_t *pThis, qWrkCmd_t tShutdownCmd, long iTimeout)
-{
-	DEFiRet;
-	int bTimedOut;
-	struct timespec t;
-	int iCancelStateSave;
-
-	queueTellActWrkThrds(pThis, 0, tShutdownCmd);/* first tell the workers our request */
-	queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
-	/* race: must make sure all are running! */
-	queueTimeoutComp(&t, iTimeout);/* get timeout */
-		
-	/* and wait for their termination */
-dbgprintf("Queue %p: waiting for mutex %p\n", pThis, &pThis->mutThrdMgmt);
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	pthread_cleanup_push(queueMutexCleanup, &pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-	bTimedOut = 0;
-	while(pThis->iCurNumWrkThrd > 0 && !bTimedOut) {
-		dbgprintf("Queue 0x%lx: waiting %ldms on worker thread termination, %d still running\n",
-			   queueGetID(pThis), iTimeout, pThis->iCurNumWrkThrd);
-
-		if(pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mutThrdMgmt, &t) != 0) {
-			dbgprintf("Queue 0x%lx: timeout waiting on worker thread termination\n", queueGetID(pThis));
-			bTimedOut = 1;	/* we exit the loop on timeout */
-		}
-	}
-	pthread_cleanup_pop(1);
-
-	if(bTimedOut)
-		iRet = RS_RET_TIMED_OUT;
-
-	return iRet;
+	RETiRet;
 }
 
 
-/* Unconditionally cancel all running worker threads.
- * rgerhards, 2008-01-14
- */
-static rsRetVal
-queueWrkThrdCancel(queue_t *pThis)
-{
-	DEFiRet;
-	int i;
-	// TODO: we need to implement peek(), without it (today!) we lose one message upon
-	// worker cancellation! -- rgerhards, 2008-01-14
-
-	/* process any pending thread requests so that we know who actually is still running */
-	queueChkWrkThrdChanges(pThis);
-
-	/* awake the workers one more time, just to be sure */
-	queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
-
-	/* first tell the workers our request */
-	for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
-		if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRD_TERMINATING) {
-			dbgprintf("Queue 0x%lx: canceling worker thread %d\n", queueGetID(pThis), i);
-			pthread_cancel(pThis->pWrkThrds[i].thrdID);
-		}
-	}
-
-	return iRet;
-}
-
-
-/* Worker thread management function carried out when the main
- * worker is about to terminate.
+/* This function shuts down all worker threads and waits until they
+ * have terminated. If they timeout, they are cancelled. Parameters have been set
+ * before this function is called so that DA queues will be fully persisted to
+ * disk (if configured to do so).
+ * rgerhards, 2008-01-24
  */
 static rsRetVal queueShutdownWorkers(queue_t *pThis)
 {
 	DEFiRet;
 	int i;
 
-	assert(pThis != NULL);
+	ISOBJ_TYPE_assert(pThis, queue);
 
-	dbgprintf("Queue 0x%lx: initiating worker thread shutdown sequence\n", (unsigned long) pThis);
+	dbgprintf("Queue 0x%lx: initiating worker thread shutdown sequence\n", queueGetID(pThis));
 
 	/* even if the timeout count is set to 0 (run endless), we still call the queueWrkThrdTrm(). This
 	 * is necessary so that all threads get sent the termination command. With a timeout of 0, however,
 	 * the function returns immediate with RS_RET_TIMED_OUT. We catch that state and accept it as
 	 * good.
 	 */
-	iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN, pThis->toQShutdown);
+	wtpShutdownAll(pThis->pWtpReg, wtpState_SHUTDOWN, pThis->toQShutdown);
 	if(iRet == RS_RET_TIMED_OUT) {
 		if(pThis->toQShutdown == 0) {
 			iRet = RS_RET_OK;
@@ -1387,418 +795,69 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis)
 			/* OK, we now need to try force the shutdown */
 			dbgprintf("Queue 0x%lx: regular worker shutdown timed out, now trying immediate\n",
 				  queueGetID(pThis));
-			iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
+			iRet = wtpShutdownAll(pThis->pWtpReg, wtpState_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
 		}
 	}
 
 	if(iRet != RS_RET_OK) { /* this is true on actual error on first try or timeout and error on second */
 		/* still didn't work out - so we now need to cancel the workers */
 		dbgprintf("Queue 0x%lx: worker threads could not be shutdown, now canceling them\n", (unsigned long) pThis);
-		iRet = queueWrkThrdCancel(pThis);
+		iRet = wtpCancelAll(pThis->pWtpReg);
+	}
+
+	// TODO: do it just once but right ;)
+	if(pThis->pWtpDA != NULL) {
+		wtpShutdownAll(pThis->pWtpDA, pThis->toQShutdown, pThis->toQShutdown);
+		if(iRet == RS_RET_TIMED_OUT) {
+			if(pThis->toQShutdown == 0) {
+				iRet = RS_RET_OK;
+			} else {
+				/* OK, we now need to try force the shutdown */
+				dbgprintf("Queue 0x%lx: regular worker shutdown timed out, now trying immediate\n",
+					  queueGetID(pThis));
+				iRet = wtpShutdownAll(pThis->pWtpDA, wtpState_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
+			}
+		}
+
+		if(iRet != RS_RET_OK) { /* this is true on actual error on first try or timeout and error on second */
+			/* still didn't work out - so we now need to cancel the workers */
+			dbgprintf("Queue 0x%lx: worker threads could not be shutdown, now canceling them\n", (unsigned long) pThis);
+			iRet = wtpCancelAll(pThis->pWtpDA);
+		}
 	}
 	
+	
 	/* finally join the threads
 	 * In case of a cancellation, this may actually take some time. This is also
 	 * needed to clean up the thread descriptors, even with a regular termination.
 	 * And, most importantly, this is needed if we have an indifitite termination
 	 * time set (timeout == 0)! -- rgerhards, 2008-01-14
 	 */
+#if 0 // totally wrong, we must implement something along these lines in wtp!
+RUNLOG;
 	for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
-		if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRD_STOPPED) {
-			queueJoinWrkThrd(pThis, i);
+		if(pThis->pWtpReg->pWrkr[i]->tCurrCmd != eWRKTHRD_STOPPED) {
+			wtiJoinThrd(pThis->pWtpReg->pWrkr[i]);
 		}
 	}
 
-	dbgprintf("Queue 0x%lx: worker threads terminated, remaining queue size %d.\n",
-		  queueGetID(pThis), pThis->iQueueSize);
-
-	return iRet;
-}
-
-/* This is a special consumer to feed the disk-queue in disk-assited mode.
- * When active, our own queue more or less acts as a memory buffer to the disk.
- * So this consumer just needs to drain the memory queue and submit entries
- * to the disk queue. The disk queue will then call the actual consumer from
- * the app point of view (we chain two queues here).
- * rgerhards, 2008-01-14
- */
-static inline rsRetVal
-queueDAConsumer(queue_t *pThis, qWrkThrd_t *pWrkrInst)
-{
-	DEFiRet;
-	int iCancelStateSave;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	assert(pThis->qRunsDA != QRUNS_REGULAR);
-	ISOBJ_assert(pWrkrInst->pUsr);
-
-dbgprintf("Queue %p/w%d: queueDAConsumer, queue size %d\n", pThis, pWrkrInst->iThrd, pThis->iQueueSize);/* dirty iQueueSize! */
-	CHKiRet(queueEnqObj(pThis->pqDA, pWrkrInst->pUsr));
-
-	/* We check if we reached the low water mark (but only if we are not in shutdown mode)
-	 * Note that the child queue now in almost all cases is non-empty, because we just enqueued
-	 * a message. Note that we need a quick check below to see if we are still in running state.
-	 * If not, we do not go into the wait, because that's not a good thing to do. We do not
-	 * do a full termination check, as this is done when we go back to the main worker loop.
-	 * We need to re-aquire the queue mutex here, because we need to have a consistent
-	 * access to the queue's admin data.
-	 */
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-dbgprintf("pre mutex lock (think about CLEANUP!)\n");
-	d_pthread_mutex_lock(pThis->mut);
-	pthread_cleanup_push(queueMutexCleanup, pThis->mut);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-dbgprintf("mutex locked (think about CLEANUP!)\n");
-	if(pThis->iQueueSize <= pThis->iLowWtrMrk && pWrkrInst->tCurrCmd == eWRKTHRD_RUNNING) {
-		dbgprintf("Queue 0x%lx/w%d: %d entries - passed low water mark in DA mode, sleeping\n",
-			  queueGetID(pThis), pWrkrInst->iThrd, pThis->iQueueSize);
-		/* wait for either passing the high water mark or the child disk queue drain */
-		pthread_cond_wait(&pThis->condDA, pThis->mut);
-	}
-	pthread_cleanup_pop(1); /* release mutex in an atomic way via cleanup handler */
-
-finalize_it:
-dbgprintf("DAConsumer returns with iRet %d\n", iRet);
-	return iRet;
-}
-
-
-/* This is a helper for queueWorker () it either calls the configured
- * consumer or the DA-consumer (if in disk-assisted mode). It is 
- * protected by the queue mutex, but MUST release it as soon as possible.
- * Most importantly, it must release it before the consumer is called.
- * rgerhards, 2008-01-14
- */
-static inline rsRetVal
-queueWorkerChkAndCallConsumer(queue_t *pThis, qWrkThrd_t *pWrkrInst, int iCancelStateSave)
-{
-	DEFiRet;
-	rsRetVal iRetLocal;
-	int iSeverity;
-	int iQueueSize;
-	void *pUsr;
-	int qRunsDA;
-	int iMyThrdIndx;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-	assert(pWrkrInst != NULL);
-
-	iMyThrdIndx = pWrkrInst->iThrd;
-
-	/* dequeue element (still protected from mutex) */
-	iRet = queueDel(pThis, &pUsr);
-	queueChkPersist(pThis); // when we support peek(), we must do this down after the del!
-	iQueueSize = pThis->iQueueSize; /* cache this for after mutex release */
-	pWrkrInst->pUsr = pUsr; /* save it for the cancel cleanup handler */
-	qRunsDA = pThis->qRunsDA;
-	d_pthread_mutex_unlock(pThis->mut);
-	pthread_cond_signal(&pThis->notFull);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-	/* WE ARE NO LONGER PROTECTED FROM THE MUTEX */
-
-	/* do actual processing (the lengthy part, runs in parallel)
-	 * If we had a problem while dequeing, we do not call the consumer,
-	 * but we otherwise ignore it. This is in the hopes that it will be
-	 * self-healing. However, this is really not a good thing.
-	 * rgerhards, 2008-01-03
-	 */
-	if(iRet != RS_RET_OK)
-		FINALIZE;
-
-	/* call consumer depending on queue mode (in DA mode, we have just one thread, so it can not change) */
-	if(qRunsDA == QRUNS_DA) {
-		queueDAConsumer(pThis, pWrkrInst);
-	} else {
-		/* we are running in normal, non-disk-assisted mode 
-		 * do a quick check if we need to drain the queue. It is OK to use the cached
-		 * iQueueSize here, because it does not hurt if it is slightly wrong.
-		 */
-		if(pThis->iDiscardMrk > 0 && iQueueSize >= pThis->iDiscardMrk) {
-			iRetLocal = objGetSeverity(pUsr, &iSeverity);
-			if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
-				dbgprintf("Queue 0x%lx/w%d: dequeue/queue nearly full (%d entries), "
-					  "discarded severity %d message\n",
-					  queueGetID(pThis), iMyThrdIndx, iQueueSize, iSeverity);
-				objDestruct(pUsr);
-			}
-		} else {
-			dbgprintf("Queue 0x%lx/w%d: worker executes consumer...\n",
-				   queueGetID(pThis), iMyThrdIndx);
-			iRetLocal = pThis->pConsumer(pUsr);
-			if(iRetLocal != RS_RET_OK) {
-				dbgprintf("Queue 0x%lx/w%d: Consumer returned iRet %d\n",
-					  queueGetID(pThis), iMyThrdIndx, iRetLocal);
+RUNLOG;
+	if(pThis->pWtpDA != NULL) {
+		for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
+			if(pThis->pWtpDA->pWrkr[i]->tCurrCmd != eWRKTHRD_STOPPED) {
+				wtiJoinThrd(pThis->pWtpDA->pWrkr[i]);
 			}
 		}
 	}
+#endif
 
-finalize_it:
-	if(iRet != RS_RET_OK) {
-		dbgprintf("Queue 0x%lx/w%d: error %d dequeueing element - ignoring, but strange things "
-			  "may happen\n", queueGetID(pThis), iMyThrdIndx, iRet);
-	}
-dbgprintf("CallConsumer returns %d\n", iRet);
-	return iRet;
-}
-
-
-
-/* cancellation cleanup handler for queueWorker ()
- * Updates admin structure and frees ressources.
- * rgerhards, 2008-01-16
- */
-static void queueWorkerCancelCleanup(void *arg)
-{
-	qWrkThrd_t *pWrkrInst = (qWrkThrd_t*) arg;
-	queue_t *pThis;
-	int iCancelStateSave;
-
-	assert(pWrkrInst != NULL);
-	ISOBJ_TYPE_assert(pWrkrInst->pQueue, queue);
-	pThis = pWrkrInst->pQueue;
-
-	dbgprintf("Queue 0x%lx/w%d: cancelation cleanup handler called (NOT FULLY IMPLEMENTED, one msg lost!)\n",
-		  queueGetID(pThis), pWrkrInst->iThrd);
-	
-	/* TODO: re-enqueue the data element! */
-
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	qWrkrSetState(&pThis->pWrkThrds[pWrkrInst->iThrd], eWRKTHRD_TERMINATING);
-	pThis->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
-
-	dbgprintf("Queue 0x%lx/w%d: thread CANCELED with %d entries left in queue, %d workers running.\n",
-		  queueGetID(pThis), pWrkrInst->iThrd, pThis->iQueueSize, pThis->iCurNumWrkThrd - 1);
-
-	pThis->iCurNumWrkThrd--;
-	pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
-	d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-}
-
-
-/* This function is created to keep the code in queueWorker () short. Thus it
- * also does not abide to the usual calling conventions used in rsyslog. It is more
- * like a macro. Its sole purpose is to have a handy shortcut for the queue
- * termination condition. For the same reason, the calling parameters are a bit
- * more verbose than the need to be in theory. The reasoning is the Worker has
- * everything handy and so we do not need to access it from memory (OK, the
- * optimized would probably have created the same code, but why not do it
- * optimal right away...). The function returns 0 if the worker should terminate
- * and something else if it should continue to run.
- * rgerhards, 2008-01-18
- */
-static inline int
-queueWorkerRemainActive(queue_t *pThis, qWrkThrd_t *pWrkrInst)
-{
-	register int b; /* this is a boolean! */
-
-	/* first check the usual termination condition that applies to all workers */
-	b = (   (qWrkrGetState(pWrkrInst) == eWRKTHRD_RUNNING)
-	     || ((qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN) && (pThis->iQueueSize > 0)));
-dbgprintf("Queue %p/w%d: chk 1 pre empty queue, qsize %d (high wtr %d), cont run: %d, cmd %d, DA qsize %d\n", pThis,
-	  pWrkrInst->iThrd,
-	  pThis->iQueueSize, pThis->iHighWtrMrk, b, qWrkrGetState(pWrkrInst),
-	  (pThis->pqDA == NULL) ? -1 : pThis->pqDA->iQueueSize);
-	if(b && pWrkrInst->iThrd == 0 && pThis->qRunsDA == QRUNS_DA) {
-		b = pThis->iQueueSize >= pThis->iHighWtrMrk || pThis->pqDA->iQueueSize != 0;
-	}
+	dbgprintf("Queue 0x%lx: worker threads terminated, remaining queue size %d.\n",
+		  queueGetID(pThis), pThis->iQueueSize);
 
-dbgprintf("Queue %p/w%d: pre empty queue, qsize %d, cont run: %d\n", pThis, pWrkrInst->iThrd, pThis->iQueueSize, b);
-	return b;
+	RETiRet;
 }
 
 
-/* Each queue has at least one associated worker (consumer) thread. It will pull
- * the message from the queue and pass it to a user-defined function.
- * This function was provided on construction. It MUST be thread-safe.
- * Worker thread 0 is always reserved for disk-assisted mode (if the queue
- * is not DA, this worker will be dormant). All other workers are for
- * regular operations mode. Workers are started and stopped as need arises.
- * rgerhards, 2008-01-15
- */
-static void *
-queueWorker(void *arg)
-{
-	queue_t *pThis = (queue_t*) arg;
-	sigset_t sigSet;
-	struct timespec t;
-	int iMyThrdIndx;	/* index for this thread in queue thread table */
-	int iCancelStateSave;
-	qWrkThrd_t *pWrkrInst; /* for cleanup handler */
-	int bContinueRun;
-
-	ISOBJ_TYPE_assert(pThis, queue);
-
-	sigfillset(&sigSet);
-	pthread_sigmask(SIG_BLOCK, &sigSet, NULL);
-
-	/* do some one-time thread initialization */
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-
-	/* initialize our thread instance descriptor */
-	qWrkrInit(&pWrkrInst, pThis);
-
-	iMyThrdIndx = pWrkrInst->iThrd;
-
-	dbgprintf("Queue 0x%lx/w%d: worker thread startup.\n", queueGetID(pThis), iMyThrdIndx);
-
-dbgprintf("qRunsDA %d, check against %d\n", pThis->qRunsDA, QRUNS_DA);
-	if((iMyThrdIndx == 0) && (pThis->qRunsDA != QRUNS_DA)) { /* are we the DA worker? */
-		if(queueStrtDA(pThis) != RS_RET_OK) { /* then fully initialize the DA queue! */
-			/* if we could not init the DA queue, we have nothing to do, so shut down. */
-			queueTellActWrkThrd(pThis, 0, eWRKTHRD_SHUTDOWN_IMMEDIATE);
-		}
-	}
-			
-	/* finally change to RUNNING state. We need to check if we actually should still run,
-	 * because someone may have requested us to shut down even before we got a chance to do
-	 * our init. That would be a bad race... -- rgerhards, 2008-01-16
-	 */
-	pThis->iCurNumWrkThrd++;
-	if(qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_INIT)
-		qWrkrSetState(pWrkrInst, eWRKTHRD_RUNNING); /* we are running now! */
-
-	pthread_cleanup_push(queueWorkerCancelCleanup, pWrkrInst);
-
-	d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-	/* end one-time stuff */
-
-	/* now we have our identity, on to real processing */
-	bContinueRun = 1; /* we need this variable, because we need to check the actual termination condition 
-			   * while protected by mutex */
-	while(bContinueRun) {
-dbgprintf("Queue 0x%lx/w%d: start worker run, queue cmd currently %d\n", 
-	  queueGetID(pThis), iMyThrdIndx, pThis->pWrkThrds[iMyThrdIndx].tCurrCmd);
-		/* process any pending thread requests */
-		queueChkWrkThrdChanges(pThis);
-
-		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-		d_pthread_mutex_lock(pThis->mut);
-dbgprintf("pthis 2: %p\n", pThis);
-
-		if((bContinueRun = queueWorkerRemainActive(pThis, pWrkrInst)) == 0) {
-dbgprintf("pthis 2a: %p\n", pThis);
-			d_pthread_mutex_unlock(pThis->mut);
-			pthread_setcancelstate(iCancelStateSave, NULL);
-dbgprintf("pthis 2b: %p\n", pThis);
-			continue; /* and break loop */
-		}
-
-		/* if we reach this point, we are still protected by the mutex */
-dbgprintf("pthis 3: %p\n", pThis);
-
-		if(pThis->iQueueSize == 0) {
-			dbgprintf("Queue 0x%lx/w%d: queue EMPTY, waiting for next message.\n",
-				  queueGetID(pThis), iMyThrdIndx);
-			/* check if the parent DA worker is running and, if not, initiate it. Thanks
-			 * to queueStrtDAWrkr (), we do not actually need to check (that routines does
-			 * that for us, but we need to aquire the parent queue's mutex to call it.
-			 */
-			if(pThis->pqParent != NULL) {
-			dbgprintf("Queue %p: pre start parent %p worker\n", pThis, pThis->pqParent);
-				queueStrtDAWrkr(pThis->pqParent);
-			}
-
-			if(pThis->bSignalOnEmpty > 0) {
-				/* we need to signal our parent queue that we are empty */
-	dbgprintf("Queue %p/w%d: signal parent we are empty\n", pThis, iMyThrdIndx);
-				pthread_cond_signal(pThis->condSignalOnEmpty);
-	dbgprintf("Queue %p/w%d: signaling parent empty done\n", pThis, iMyThrdIndx);
-			}
-			if(pThis->bSignalOnEmpty > 1) {
-				/* no mutex associated with this condition, it's just a try (but needed
-				 * to wakeup a parent worker if e.g. the queue was restarted from disk) */
-				pthread_cond_signal(pThis->condSignalOnEmpty2);
-			}
-			/* If we arrive here, we have the regular case, where we can safely assume that
-			 * iQueueSize and tCmd have not changed since the while().
-			 */
-	dbgprintf("Queue %p/w%d: pre condwait ->notEmpty, worker shutdown %d\n", pThis, iMyThrdIndx, pThis->toWrkShutdown);
-			/* DA worker and first worker never have an inactivity timeout */
-			if(pWrkrInst->iThrd < 2 || pThis->toWrkShutdown == -1) {
-		//xxx	if(pThis->toWrkShutdown == -1) {
-	dbgprintf("worker never times out!\n");
-				/* never shut down any started worker */
-				pthread_cond_wait(&pThis->notEmpty, pThis->mut);
-			} else {
-				queueTimeoutComp(&t, pThis->toWrkShutdown);/* get absolute timeout */
-				if(pthread_cond_timedwait(&pThis->notEmpty, pThis->mut, &t) != 0) {
-					dbgprintf("Queue 0x%lx/w%d: inactivity timeout, worker terminating...\n",
-						  queueGetID(pThis), iMyThrdIndx);
-					/* we use SHUTDOWN (and not SHUTDOWN_IMMEDIATE) so that the worker
-					 * does not terminate if in the mean time a new message arrived.
-					 */
-					qWrkrSetState(pWrkrInst, eWRKTHRD_SHUTDOWN);
-				}
-			}
-	dbgprintf("Queue %p/w%d: post condwait ->notEmpty\n", pThis, iMyThrdIndx);
-			d_pthread_mutex_unlock(pThis->mut);
-			pthread_setcancelstate(iCancelStateSave, NULL);
-			pthread_testcancel(); /* see big comment below */
-			pthread_yield(); /* see big comment below */
-			continue; /* request next iteration */
-		}
-
-		/* if we reach this point, we have a non-empty queue (and are still protected by mutex) */
-		queueWorkerChkAndCallConsumer(pThis, pWrkrInst, iCancelStateSave);
-
-		/* Now make sure we can get canceled - it is not specified if pthread_setcancelstate() is
-		 * a cancellation point in itself. As we run most of the time without cancel enabled, I fear
-		 * we may never get cancelled if we do not create a cancellation point ourselfs.
-		 */
-		pthread_testcancel();
-		/* We now yield to give the other threads a chance to obtain the mutex. If we do not
-		 * do that, this thread may very well aquire the mutex again before another thread
-		 * has even a chance to run. The reason is that mutex operations are free to be
-		 * implemented in the quickest possible way (and they typically are!). That is, the
-		 * mutex lock/unlock most probably just does an atomic memory swap and does not necessarily
-		 * schedule other threads waiting on the same mutex. That can lead to the same thread
-		 * aquiring the mutex ever and ever again while all others are starving for it. We
-		 * have exactly seen this behaviour when we deliberately introduced a long-running
-		 * test action which basically did a sleep. I understand that with real actions the
-		 * likelihood of this starvation condition is very low - but it could still happen
-		 * and would be very hard to debug. The yield() is a sure fix, its performance overhead
-		 * should be well accepted given the above facts. -- rgerhards, 2008-01-10
-		 */
-		pthread_yield();
-		if(Debug && (qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN) && pThis->iQueueSize > 0)
-			dbgprintf("Queue 0x%lx/w%d: worker does not yet terminate because it still has "
-				  " %d messages to process.\n", queueGetID(pThis), iMyThrdIndx, pThis->iQueueSize);
-	}
-
-	/* indicate termination */
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-dbgprintf("Queue %p: worker waiting for mutex\n", pThis);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	/* check if we are the DA worker and, if so, switch back to regular mode */
-	if(pWrkrInst->iThrd == 0) {
-		queueTurnOffDAMode(pThis);
-	}
-	pthread_cleanup_pop(0); /* remove cleanup handler */
-
-	pThis->iCurNumWrkThrd--; /* one less ;) */
-	/* if we ever need finalize_it, here would be the place for it! */
-	if(qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN ||
-	   qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN_IMMEDIATE ||
-	   qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_INIT ||
-	   qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_CREATED) {
-	   	/* in shutdown case, we need to flag termination. All other commands
-		 * have a meaning to the thread harvester, so we can not overwrite them
-		 */
-dbgprintf("Queue 0x%lx/w%d: setting termination state\n", queueGetID(pThis), iMyThrdIndx);
-		qWrkrSetState(pWrkrInst, eWRKTHRD_TERMINATING);
-	}
-	pThis->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
-	pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
-	d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
-	pthread_setcancelstate(iCancelStateSave, NULL);
-
-	pthread_exit(0);
-}
-
 
 /* Constructor for the queue object
  * This constructs the data structure, but does not yet start the queue. That
@@ -1811,6 +870,8 @@ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads,
 	DEFiRet;
 	queue_t *pThis;
 
+int *pBoom = NULL;
+//*pBoom = 'A';
 	assert(ppThis != NULL);
 	assert(pConsumer != NULL);
 	assert(iWorkerThreads >= 0);
@@ -1866,7 +927,205 @@ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads,
 
 finalize_it:
 	OBJCONSTRUCT_CHECK_SUCCESS_AND_CLEANUP
-	return iRet;
+	RETiRet;
+}
+
+
+/* cancellation cleanup handler for queueWorker ()
+ * Updates admin structure and frees ressources.
+ * rgerhards, 2008-01-16
+ */
+static rsRetVal
+queueConsumerCancelCleanup(void *arg1, void *arg2)
+{
+	queue_t *pThis = (queue_t*) arg1;
+	wti_t *pWti = (wti_t*) arg2;
+
+	ISOBJ_TYPE_assert(pThis, queue);
+	ISOBJ_TYPE_assert(pWti, wti);
+
+	dbgprintf("Queue 0x%lx: cancelation cleanup handler consumer called (NOT FULLY IMPLEMENTED, one msg lost!)\n",
+		  queueGetID(pThis));
+	
+	/* TODO: re-enqueue the data element! */
+
+	return RS_RET_OK;
+}
+
+
+
+/* This function checks if the provided message shall be discarded and does so, if needed.
+ * In DA mode, we do not discard any messages as we assume the disk subsystem is fast enough to
+ * provide real-time creation of spool files.
+ * Note: cached copies of iQueueSize and bRunsDA are provided so that no mutex locks are required.
+ * The caller must have obtained them while the mutex was locked. Of course, these values may no
+ * longer be current, but that is OK for the discard check. At worst, the message is either processed
+ * or discarded when it should not have been. As discarding is in itself somewhat racy and erratic,
+ * that is no problems for us. This function MUST NOT lock the queue mutex, it could result in
+ * deadlocks!
+ * If the message is discarded, it can no longer be processed by the caller. So be sure to check
+ * the return state!
+ * rgerhards, 2008-01-24
+ */
+static int queueChkDiscardMsg(queue_t *pThis, int iQueueSize, int bRunsDA, void *pUsr)
+{
+	DEFiRet;
+	rsRetVal iRetLocal;
+	int iSeverity;
+
+	ISOBJ_TYPE_assert(pThis, queue);
+	ISOBJ_assert(pUsr);
+
+	if(pThis->iDiscardMrk > 0 && iQueueSize >= pThis->iDiscardMrk && bRunsDA == 0) {
+		iRetLocal = objGetSeverity(pUsr, &iSeverity);
+		if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
+			dbgprintf("Queue 0x%lx: queue nearly full (%d entries), discarded severity %d message\n",
+				  queueGetID(pThis), iQueueSize, iSeverity);
+			objDestruct(pUsr);
+			ABORT_FINALIZE(RS_RET_QUEUE_FULL);
+		} else {
+			dbgprintf("Queue 0x%lx: queue nearly full (%d entries), but could not drop msg "
+				  "(iRet: %d, severity %d)\n", queueGetID(pThis), iQueueSize,
+				  iRetLocal, iSeverity);
+		}
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* dequeue the queued object for the queue consumers.
+ * rgerhards, 2008-10-21
+ */
+static rsRetVal
+queueDequeueConsumable(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
+{
+	DEFiRet;
+	void *pUsr;
+	int iQueueSize;
+	int bRunsDA;	 /* cache for early mutex release */
+
+	/* dequeue element (still protected from mutex) */
+	iRet = queueDel(pThis, &pUsr);
+	queueChkPersist(pThis); // when we support peek(), we must do this down after the del!
+	iQueueSize = pThis->iQueueSize; /* 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 */
+	d_pthread_mutex_unlock(pThis->mut);
+	pthread_cond_signal(&pThis->notFull);
+	pthread_setcancelstate(iCancelStateSave, NULL);
+	/* WE ARE NO LONGER PROTECTED BY THE MUTEX */
+
+	/* do actual processing (the lengthy part, runs in parallel)
+	 * If we had a problem while dequeing, we do not call the consumer,
+	 * but we otherwise ignore it. This is in the hopes that it will be
+	 * self-healing. However, this is really not a good thing.
+	 * rgerhards, 2008-01-03
+	 */
+	if(iRet != RS_RET_OK)
+		FINALIZE;
+
+	/* we are running in normal, non-disk-assisted mode do a quick check if we need to drain the queue.
+	 * In DA mode, we do not discard any messages as we assume the disk subsystem is fast enough to
+	 * provide real-time creation of spool files.
+	 * Note: It is OK to use the cached iQueueSize here, because it does not hurt if it is slightly wrong.
+	 */
+	CHKiRet(queueChkDiscardMsg(pThis, iQueueSize, bRunsDA, pUsr));
+
+finalize_it:
+	if(iRet != RS_RET_OK && iRet != RS_RET_DISCARDMSG) {
+		dbgprintf("Queue 0x%lx/w?: error %d dequeueing element - ignoring, but strange things "
+			  "may happen\n", queueGetID(pThis), iRet);
+	}
+	RETiRet;
+}
+
+
+/* This is the queue consumer in the regular (non-DA) case. It is 
+ * protected by the queue mutex, but MUST release it as soon as possible.
+ * rgerhards, 2008-01-21
+ */
+static rsRetVal
+queueConsumerReg(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, queue);
+	ISOBJ_TYPE_assert(pWti, wti);
+
+	CHKiRet(queueDequeueConsumable(pThis, pWti, iCancelStateSave));
+	CHKiRet(pThis->pConsumer(pWti->pUsrp));
+
+finalize_it:
+dbgprintf("Queue %p: regular consumer returns %d\n", pThis, iRet);
+	RETiRet;
+}
+
+
+/* This is a special consumer to feed the disk-queue in disk-assited mode.
+ * When active, our own queue more or less acts as a memory buffer to the disk.
+ * So this consumer just needs to drain the memory queue and submit entries
+ * to the disk queue. The disk queue will then call the actual consumer from
+ * the app point of view (we chain two queues here).
+ * When this method is entered, the mutex is always locked and needs to be unlocked
+ * as part of the processing.
+ * rgerhards, 2008-01-14
+ */
+static rsRetVal
+queueConsumerDA(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, queue);
+	ISOBJ_TYPE_assert(pWti, wti);
+
+dbgprintf("Queue %p/w?: queueDAConsumer, queue size %d\n", pThis, pThis->iQueueSize);/* dirty iQueueSize! */
+	CHKiRet(queueDequeueConsumable(pThis, pWti, iCancelStateSave));
+	CHKiRet(queueEnqObj(pThis->pqDA, pWti->pUsrp));
+
+finalize_it:
+dbgprintf("DAConsumer returns with iRet %d\n", iRet);
+	RETiRet;
+}
+
+
+/* must only be called when the queue mutex is locked, else results
+ * are not stable!
+ * Version when running in DA mode.
+ */
+static int
+queueChkStopWrkrDA(queue_t *pThis)
+{
+	return pThis->bEnqOnly || !pThis->bRunsDA;
+}
+
+/* must only be called when the queue mutex is locked, else results
+ * are not stable!
+ * Version when running in non-DA mode.
+ */
+static int
+queueChkStopWrkrReg(queue_t *pThis)
+{
+	return pThis->bEnqOnly || pThis->bRunsDA;
+}
+
+
+/* must only be called when the queue mutex is locked, else results
+ * are not stable! DA version
+ */
+static int
+queueIsIdleDA(queue_t *pThis)
+{
+	return (pThis->iQueueSize == 0 || (pThis->bRunsDA && pThis->iQueueSize <= pThis->iLowWtrMrk));
+}
+/* must only be called when the queue mutex is locked, else results
+ * are not stable! Regular version
+ */
+static int
+queueIsIdleReg(queue_t *pThis)
+{
+	return (pThis->iQueueSize == 0);
 }
 
 
@@ -1878,7 +1137,8 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */
 	DEFiRet;
 	rsRetVal iRetLocal;
 	int bInitialized = 0; /* is queue already initialized? */
-	int i;
+	uchar pszBuf[64];
+	size_t lenBuf;
 
 	assert(pThis != NULL);
 
@@ -1888,7 +1148,7 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */
 	if(pThis->pqParent == NULL) {
 dbgprintf("Queue %p: no parent, alloc mutex\n", pThis);
 		pThis->mut = (pthread_mutex_t *) malloc (sizeof (pthread_mutex_t));
-		pthread_mutex_init (pThis->mut, NULL);
+		pthread_mutex_init(pThis->mut, NULL);
 	} else {
 		/* child queue, we need to use parent's mutex */
 		pThis->mut = pThis->pqParent->mut;
@@ -1909,15 +1169,23 @@ dbgprintf("Queue %p: post mutexes, mut %p\n", pThis, pThis->mut);
 	if(pThis->qType == QUEUETYPE_DIRECT)
 		FINALIZE;	/* with direct queues, we are already finished... */
 
-	/* initialize worker thread instances 
-	 * TODO: move to separate function
+	/* create worker thread pools for regular operation. The DA pool is created on an as-needed
+	 * basis, which potentially means never under most circumstances.
 	 */
-	if((pThis->pWrkThrds = calloc(pThis->iNumWorkerThreads + 1, sizeof(qWrkThrd_t))) == NULL)
-		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
-	for(i = 0 ; i < pThis->iNumWorkerThreads + 1 ; ++i) {
-		qWrkrConstructFinalize(&pThis->pWrkThrds[i], pThis, i);
-	}
-
+	lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "Queue 0x%lx/Reg", (unsigned long) pThis);
+	CHKiRet(wtpConstruct		(&pThis->pWtpReg));
+	CHKiRet(wtpSetDbgHdr		(pThis->pWtpReg, pszBuf, lenBuf));
+	CHKiRet(wtpSetpfChkStopWrkr	(pThis->pWtpReg, queueChkStopWrkrReg));
+	CHKiRet(wtpSetpfIsIdle		(pThis->pWtpReg, queueIsIdleReg));
+	CHKiRet(wtpSetpfDoWork		(pThis->pWtpReg, queueConsumerReg));
+	CHKiRet(wtpSetpfOnWorkerCancel	(pThis->pWtpReg, queueConsumerCancelCleanup));
+	CHKiRet(wtpSetpmutUsr		(pThis->pWtpReg, pThis->mut));
+	CHKiRet(wtpSetpcondBusy		(pThis->pWtpReg, &pThis->notEmpty));
+	CHKiRet(wtpSetiNumWorkerThreads	(pThis->pWtpReg, pThis->iNumWorkerThreads));
+	CHKiRet(wtpSetpUsr		(pThis->pWtpReg, pThis));
+	CHKiRet(wtpConstructFinalize	(pThis->pWtpReg));
+
+	/* initialize worker thread instances */
 	if(pThis->bIsDA) {
 		/* If we are disk-assisted, we need to check if there is a QIF file
 		 * which we need to load. -- rgerhards, 2008-01-15
@@ -1927,7 +1195,11 @@ dbgprintf("Queue %p: post mutexes, mut %p\n", pThis, pThis->mut);
 			dbgprintf("Queue 0x%lx: on-disk queue present, needs to be reloaded\n",
 				  queueGetID(pThis));
 
-			queueInitDA(pThis, QUEUE_MODE_ENQDEQ); /* initiate DA mode */
+			queueInitDA(pThis, QUEUE_MODE_ENQDEQ, LOCK_MUTEX); /* initiate DA mode */
+			/* we need to start the DA worker thread so that messages will be processed. So 
+			 * we advise the worker pool there is at least one needed. The wtp does the rest...
+			 */
+			wtpAdviseMaxWorkers(pThis->pWtpDA, 1);
 			bInitialized = 1; /* we are done */
 		} else {
 			// TODO: use logerror? -- rgerhards, 2008-01-16
@@ -1938,16 +1210,14 @@ dbgprintf("Queue %p: post mutexes, mut %p\n", pThis, pThis->mut);
 
 	if(!bInitialized) {
 		dbgprintf("Queue 0x%lx: queue starts up without (loading) any disk state\n", queueGetID(pThis));
-		/* fire up the worker threads */
-		if(pThis->bEnqOnly == 0)
-			queueStrtNewWrkThrd(pThis);
-		// TODO: preforked workers! queueStrtAllWrkThrds(pThis);
+		/* we do not fire up any worker threads here, this happens automatically when they are needed */
+		// TODO: preforked workers? queueStrtAllWrkThrds(pThis);
 	}
 	pThis->bQueueStarted = 1;
 
 finalize_it:
 dbgprintf("queueStart() exit, iret %d\n", iRet);
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2023,7 +1293,7 @@ finalize_it:
 	if(psQIF != NULL)
 		strmDestruct(&psQIF);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2043,15 +1313,16 @@ rsRetVal queueChkPersist(queue_t *pThis)
 		pThis->iUpdsSincePersist = 0;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
 /* destructor for the queue object */
 rsRetVal queueDestruct(queue_t **ppThis)
 {
-	queue_t *pThis;
 	DEFiRet;
+	queue_t *pThis;
+	DEFVARS_mutexProtection;
 
 	assert(ppThis != NULL);
 	pThis = *ppThis;
@@ -2064,52 +1335,38 @@ pThis->bSaveOnShutdown = 1; // TODO: Test remove
 	/* we do not need to take care of any messages left in queue if we are in enqueue only mode */
 	if(!pThis->bEnqOnly) {
 		/* in regular mode, need look at termination */
+		BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, LOCK_MUTEX);	/* some workers may be running in parallel! */
 		/* optimize parameters for shutdown of DA-enabled queues */
 		if(pThis->bIsDA && pThis->iQueueSize > 0) { // TODO: atomic iQueueSize!
 	dbgprintf("IsDA queue, modifying params for draining\n");
 			pThis->iHighWtrMrk = 1; /* make sure we drain */
 			pThis->iLowWtrMrk = 0; /* disable low water mark algo */
-			if(pThis->qRunsDA == QRUNS_REGULAR) {
+			if(pThis->bRunsDA == 0) {
 				if(pThis->iQueueSize > 0) {
-					queueInitDA(pThis, QUEUE_MODE_ENQONLY); /* initiate DA mode */
+					queueInitDA(pThis, QUEUE_MODE_ENQONLY, MUTEX_ALREADY_LOCKED); /* initiate DA mode */
 				}
 			} else {
 				queueSetEnqOnly(pThis->pqDA, QUEUE_MODE_ENQONLY); /* turn on enqueue-only mode */
+				/* worker may have been waited on low water mark, reactivate */
+				wtpAdviseMaxWorkers(pThis->pWtpDA, 1);
 			}
 			if(pThis->bSaveOnShutdown) {
 	dbgprintf("bSaveOnShutdown set, eternal timeout set\n");
 				pThis->toQShutdown = QUEUE_TIMEOUT_ETERNAL;
 			}
-			/* now we need to activate workers (read doc/dev_queue.html) */
+			END_MTX_PROTECTED_OPERATIONS(pThis->mut);
 		}
-
-		/* wait until all pending workers are started up */
-		qWrkrWaitAllWrkrStartup(pThis);
-
-		// We need to startup a worker if we are in non-DA mode and the queue is not empty and not in enque-only mode */
-	dbgprintf("Queue %p: queueDestruct probing if any regular workers need to be started, CurWrkr %d, qsize %d, qRunsDA %d\n",
-		  pThis, pThis->iCurNumWrkThrd, pThis->iQueueSize, pThis->qRunsDA);
-		d_pthread_mutex_lock(pThis->mut);
-	dbgprintf("queueDestruct mutex locked\n");
-		if(pThis->iCurNumWrkThrd == 0 && pThis->iQueueSize > 0 && !pThis->bEnqOnly) {
-	dbgprintf("Queue %p: queueDestruct must start regular workers!\n", pThis);
-	// TODO check mutex call order - doies function aquire mutex?
-			queueStrtNewWrkThrd(pThis);
-		}
-		d_pthread_mutex_unlock(pThis->mut);
-	dbgprintf("queueDestruct mutex unlocked\n");
-
-		/* wait again in case a new worker was started */
-		qWrkrWaitAllWrkrStartup(pThis);
 	}
 
-	/* terminate our own worker threads */
-	if(pThis->pWrkThrds != NULL) {
+	/* at this point, the queue is either empty with all workers being idle (or deact) or the queue
+	 * is full and all workers are running. We now need to wait for everyone to become idle.
+	 */
+	if(pThis->qType != QUEUETYPE_DIRECT) {
 		queueShutdownWorkers(pThis);
 	}
 
-	/* if still running DA, terminate disk queue */
-	if(pThis->qRunsDA != QRUNS_REGULAR)
+	/* if still running DA, terminate disk queue (note that the DA queue is NULL if it was never used) */
+	if(pThis->bRunsDA && pThis->pqDA != NULL)
 		queueDestruct(&pThis->pqDA);
 
 	/* persist the queue (we always do that - queuePersits() does cleanup if the queue is empty) */
@@ -2118,9 +1375,10 @@ pThis->bSaveOnShutdown = 1; // TODO: Test remove
 	}
 
 	/* ... then free resources */
-	if(pThis->pWrkThrds != NULL) {
-		free(pThis->pWrkThrds);
-		pThis->pWrkThrds = NULL;
+	if(pThis->qType != QUEUETYPE_DIRECT) {
+		wtpDestruct(&pThis->pWtpReg);
+		if(pThis->pWtpDA != NULL)	
+			wtpDestruct(&pThis->pWtpDA);
 	}
 
 	if(pThis->pqParent == NULL) {
@@ -2141,7 +1399,7 @@ pThis->bSaveOnShutdown = 1; // TODO: Test remove
 	free(pThis);
 	*ppThis = NULL;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2167,7 +1425,7 @@ queueSetFilePrefix(queue_t *pThis, uchar *pszPrefix, size_t iLenPrefix)
 	pThis->lenFilePrefix = iLenPrefix;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* set the queue's maximum file size
@@ -2187,7 +1445,7 @@ queueSetMaxFileSize(queue_t *pThis, size_t iMaxFileSize)
 	pThis->iMaxFileSize = iMaxFileSize;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2203,65 +1461,64 @@ queueEnqObj(queue_t *pThis, void *pUsr)
 {
 	DEFiRet;
 	int iCancelStateSave;
+	int iMaxWorkers;
 	int i;
 	struct timespec t;
-	int iSeverity = 8;
-	rsRetVal iRetLocal;
 
 	ISOBJ_TYPE_assert(pThis, queue);
 
-	/* process any pending thread requests */
-	queueChkWrkThrdChanges(pThis);
-
+dbgprintf("Queue %p: EnqObj() 1\n", pThis);
 	/* Please note that this function is not cancel-safe and consequently
 	 * sets the calling thread's cancelibility state to PTHREAD_CANCEL_DISABLE
 	 * during its execution. If that is not done, race conditions occur if the
 	 * thread is canceled (most important use case is input module termination).
 	 * rgerhards, 2008-01-08
 	 */
-	if(pThis->pWrkThrds != NULL) {
+	if(pThis->qType != QUEUETYPE_DIRECT) {
 		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
 		d_pthread_mutex_lock(pThis->mut);
 	}
 
-	/* first check if we can discard anything */
-	if(pThis->iDiscardMrk > 0 && pThis->iQueueSize >= pThis->iDiscardMrk) {
-		iRetLocal = objGetSeverity(pUsr, &iSeverity);
-		if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
-			dbgprintf("Queue 0x%lx: queue nearly full (%d entries), discarded severity %d message\n",
-				  queueGetID(pThis), pThis->iQueueSize, iSeverity);
-			objDestruct(pUsr);
-			ABORT_FINALIZE(RS_RET_QUEUE_FULL);
-		} else {
-			dbgprintf("Queue 0x%lx: queue nearly full (%d entries), but could not drop msg "
-				  "(iRet: %d, severity %d)\n", queueGetID(pThis), pThis->iQueueSize,
-				  iRetLocal, iSeverity);
-		}
-	}
+	/* first check if we need to discard this message (which will cause CHKiRet() to exit) */
+	CHKiRet(queueChkDiscardMsg(pThis, pThis->iQueueSize, pThis->bRunsDA, pUsr));
 
+dbgprintf("Queue %p: EnqObj() 10\n", pThis);
 	/* then check if we need to add an assistance disk queue */
 	if(pThis->bIsDA)
 		CHKiRet(queueChkStrtDA(pThis));
 	
-	/* re-process any new pending thread requests and see if we need to start workers */
-	queueChkAndStrtWrk(pThis);
+RUNLOG_VAR("%d", pThis->bIsDA);
+	/* make sure at least one worker is running. */
+	if(pThis->bRunsDA) {
+RUNLOG;
+		wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* disk queues have always one worker */
+	} else {
+		if(pThis->qType == QUEUETYPE_DISK || pThis->iMinMsgsPerWrkr == 0) {
+			iMaxWorkers = 1;
+		} else {
+			iMaxWorkers = pThis->iQueueSize / pThis->iMinMsgsPerWrkr + 1;
+		}
+RUNLOG;
+		wtpAdviseMaxWorkers(pThis->pWtpReg, iMaxWorkers);
+	}
 
-	/* and finally (try to) enqueue what is left over */
+	/* wait for the queue to be ready... */
 	while(pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize) {
 		dbgprintf("Queue 0x%lx: enqueueMsg: queue FULL - waiting to drain.\n", queueGetID(pThis));
-		queueTimeoutComp(&t, pThis->toEnq);
+		timeoutComp(&t, pThis->toEnq);
 		if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
 			dbgprintf("Queue 0x%lx: enqueueMsg: cond timeout, dropping message!\n", queueGetID(pThis));
 			objDestruct(pUsr);
 			ABORT_FINALIZE(RS_RET_QUEUE_FULL);
 		}
 	}
+
+	/* and finally enqueue the message */
 	CHKiRet(queueAdd(pThis, pUsr));
 	queueChkPersist(pThis);
 
 finalize_it:
-	/* now awake sleeping worker threads */
-	if(pThis->pWrkThrds != NULL) {
+	if(pThis->qType != QUEUETYPE_DIRECT) {
 		d_pthread_mutex_unlock(pThis->mut);
 		i = pthread_cond_signal(&pThis->notEmpty);
 		dbgprintf("Queue 0x%lx: EnqueueMsg signaled condition (%d)\n", (unsigned long) pThis, i);
@@ -2269,7 +1526,7 @@ finalize_it:
 	}
 
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2307,7 +1564,8 @@ queueSetEnqOnly(queue_t *pThis, int bEnqOnly)
 			/* this means we need to terminate all workers - that's it... */
 			dbgprintf("Queue 0x%lx: switching to enqueue-only mode, terminating all worker threads\n",
 				   queueGetID(pThis));
-			queueWrkThrdReqTrm(pThis, eWRKTHRD_SHUTDOWN_IMMEDIATE, 0);
+			wtpWakeupAllWrkr(pThis->pWtpDA);
+			wtpWakeupAllWrkr(pThis->pWtpReg);
 		} else {
 			/* switch back to regular mode */
 			ABORT_FINALIZE(RS_RET_NOT_IMPLEMENTED); /* we don't need this so far... */
@@ -2321,7 +1579,7 @@ finalize_it:
 		d_pthread_mutex_unlock(pThis->mut);
 		pthread_setcancelstate(iCancelStateSave, NULL);
 	}
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2360,19 +1618,18 @@ static rsRetVal queueSetProperty(queue_t *pThis, property_t *pProp)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 #undef	isProp
-
-
 /* Initialize the stream class. Must be called as the very first method
  * before anything else is called inside this class.
  * rgerhards, 2008-01-09
  */
 BEGINObjClassInit(queue, 1)
-	//OBJSetMethodHandler(objMethod_SERIALIZE, strmSerialize);
 	OBJSetMethodHandler(objMethod_SETPROPERTY, queueSetProperty);
 	//OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strmConstructFinalize);
+//fprintf(stdout, "queueChkStopWrkrReg: %p\n", queueChkStopWrkrReg);
+//fprintf(stdout, "queueChkStopWrkrDA:  %p\n", queueChkStopWrkrDA);
 ENDObjClassInit(queue)
 
 /*
diff --git a/queue.h b/queue.h
index dca80ffd..ee2637da 100644
--- a/queue.h
+++ b/queue.h
@@ -25,27 +25,9 @@
 
 #include <pthread.h>
 #include "obj.h"
+#include "wtp.h"
 #include "stream.h"
 
-/* some information about disk files used by the queue. In the long term, we may
- * export this settings to a separate file module - or not (if they are too
- * queue-specific. I just thought I mention it here so that everyone is aware
- * of this possibility. -- rgerhards, 2008-01-07
- */
-typedef struct {
-	int fd;		/* the file descriptor, -1 if closed */
-	uchar *pszFileName; /* name of current file (if open) */
-	int iCurrFileNum;/* current file number (NOT descriptor, but the number in the file name!) */
-	size_t iCurrOffs;/* current offset */
-	uchar *pIOBuf;	/* io Buffer */
-	int iBufPtrMax;	/* current max Ptr in Buffer (if partial read!) */
-	int iBufPtr;	/* pointer into current buffer */
-	int iUngetC;	/* char set via UngetChar() call or -1 if none set */
-	int bDeleteOnClose; /* set to 1 to auto-delete on close -- be careful with that setting! */
-} queueFileDescription_t;
-#define qFILE_IOBUF_SIZE 4096 /* size of the IO buffer */
-
-
 /* queue types */
 typedef enum {
 	QUEUETYPE_FIXED_ARRAY = 0,/* a simple queue made out of a fixed (initially malloced) array fast but memoryhog */
@@ -60,17 +42,6 @@ typedef struct qLinkedList_S {
 	void *pUsr;
 } qLinkedList_t;
 
-/* commands and states for worker threads. */
-typedef enum {
-	eWRKTHRD_STOPPED = 0,	/* worker thread is not running (either actually never ran or was shut down) */
-	eWRKTHRD_TERMINATING = 1,/* worker thread has shut down, but some finalzing is still needed */
-	/* ALL active states MUST be numerically higher than eWRKTHRD_TERMINATED and NONE must be lower! */
-	eWRKTHRD_RUN_CREATED = 2,/* worker thread has been created, but not yet begun initialization (prob. not yet scheduled) */
-	eWRKTHRD_RUN_INIT = 3,	/* worker thread is initializing, but not yet fully running */
-	eWRKTHRD_RUNNING = 4,	/* worker thread is up and running and shall continue to do so */
-	eWRKTHRD_SHUTDOWN = 5,	/* worker thread is running but shall terminate when queue is empty */
-	eWRKTHRD_SHUTDOWN_IMMEDIATE = 6/* worker thread is running but shall terminate even if queue is full */
-} qWrkCmd_t;
 
 typedef struct qWrkThrd_s {
 	pthread_t thrdID;  /* thread ID */
@@ -95,7 +66,8 @@ typedef struct queue_s {
 	int 	iNumWorkerThreads;/* number of worker threads to use */
 	int 	iCurNumWrkThrd;/* current number of active worker threads */
 	int	iMinMsgsPerWrkr;/* minimum nbr of msgs per worker thread, if more, a new worker is started until max wrkrs */
-	qWrkThrd_t *pWrkThrds;/* array with control structure for the worker thread(s) associated with this queue */
+	wtp_t	*pWtpDA;
+	wtp_t	*pWtpReg;
 	int	iUpdsSincePersist;/* nbr of queue updates since the last persist call */
 	int	iPersistUpdCnt;	/* persits queue info after this nbr of updates - 0 -> persist only on shutdown */
 	int	iHighWtrMrk;	/* high water mark for disk-assisted memory queues */
@@ -138,11 +110,7 @@ typedef struct queue_s {
 	int iNumberFiles;	/* how many files make up the queue? */
 	size_t iMaxFileSize;	/* max size for a single queue file */
 	int bIsDA;		/* is this queue disk assisted? */
-	enum {
-		QRUNS_REGULAR,
-		QRUNS_DA_INIT,
-		QRUNS_DA
-	} qRunsDA;		/* is this queue actually *running* disk assisted? if so, which mode? */
+	int bRunsDA;		/* is this queue actually *running* disk assisted? */
 	pthread_mutex_t mutDA;	/* mutex for low water mark algo */
 	pthread_cond_t condDA;	/* and its matching condition */
 	struct queue_s *pqDA;	/* queue for disk-assisted modes */
diff --git a/rsyslog.h b/rsyslog.h
index e37263e3..dd7b1c1e 100644
--- a/rsyslog.h
+++ b/rsyslog.h
@@ -112,6 +112,7 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_TIMED_OUT = -2041, /**< timeout occured (not necessarily an error) */
 	RS_RET_QSIZE_ZERO = -2042, /**< queue size is zero where this is not supported */
 	RS_RET_ALREADY_STARTING = -2043, /**< something (a thread?) is already starting - not necessarily an error */
+	RS_RET_NO_MORE_THREADS = -2044, /**< no more threads available, not necessarily an error */
 	RS_RET_OK_DELETE_LISTENTRY = 1,	/**< operation successful, but callee requested the deletion of an entry (special state) */
 	RS_RET_TERMINATE_NOW = 2,	/**< operation successful, function is requested to terminate (mostly used with threads) */
 	RS_RET_NO_RUN = 3,		/**< operation successful, but function does not like to be executed */
@@ -128,11 +129,9 @@ typedef enum rsRetVal_ rsRetVal; /**< friendly type for global return value */
 #define CHKiRet_Hdlr(code) if((iRet = code) != RS_RET_OK)
 /* macro below is used in conjunction with CHKiRet_Hdlr, else use ABORT_FINALIZE */
 #define FINALIZE goto finalize_it;
-#if 0 /* DEV debug: set to 1 to get a rough call trace -- rgerhards, 2008-01-13 */
-#	define DEFiRet dbgprintf("Entering %s, line %d\n", __FILE__, __LINE__); rsRetVal iRet = RS_RET_OK
-#else
-#	define DEFiRet rsRetVal iRet = RS_RET_OK
-#endif
+#define DEFiRet BEGINfunc rsRetVal iRet = RS_RET_OK
+#define RETiRet do{ ENDfunc return iRet; }while(0)
+
 #define ABORT_FINALIZE(errCode)			\
 	do {					\
 		iRet = errCode;			\
@@ -197,6 +196,8 @@ typedef unsigned char uchar;
 /* The following prototype is convenient, even though it may not be the 100% correct place.. -- rgerhards 2008-01-07 */
 void dbgprintf(char *, ...) __attribute__((format(printf, 1, 2)));
 
+#include "debug.h"
+
 #endif /* multi-include protection */
 /*
  * vi:set ai:
diff --git a/srUtils.c b/srUtils.c
index 35b437bd..590e7eff 100755
--- a/srUtils.c
+++ b/srUtils.c
@@ -291,7 +291,7 @@ rsRetVal genFileName(uchar **ppName, uchar *pDirName, size_t lenDirName, uchar *
 	*ppName = pName;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* get the number of digits required to represent a given number. We use an
@@ -337,8 +337,10 @@ timeoutComp(struct timespec *pt, int iTimeout)
 void
 mutexCancelCleanup(void *arg)
 {
+	BEGINfunc
 	assert(arg != NULL);
 	d_pthread_mutex_unlock((pthread_mutex_t*) arg);
+	ENDfunc
 }
 
 
diff --git a/stream.c b/stream.c
index 15d9dcf6..cb6d5e29 100644
--- a/stream.c
+++ b/stream.c
@@ -107,7 +107,7 @@ static rsRetVal strmOpenFile(strm_t *pThis)
 		  iFlags, pThis->fd);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -138,7 +138,7 @@ static rsRetVal strmCloseFile(strm_t *pThis)
 		pThis->pszCurrFName = NULL;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -165,7 +165,7 @@ strmNextFile(strm_t *pThis)
 	pThis->iCurrFNum = (pThis->iCurrFNum + 1) % pThis->iMaxFiles;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -228,7 +228,7 @@ rsRetVal strmReadChar(strm_t *pThis, uchar *pC)
 //dbgprintf("ReadChar: read %c, offset %d\n", *pC, pThis->iCurrOffs);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -285,7 +285,7 @@ finalize_it:
 	if(iRet != RS_RET_OK && pCStr != NULL)
 		rsCStrDestruct(pCStr);
 
-	return iRet;
+	RETiRet;
 }
 
 #endif /* #if 0 - saved code */
@@ -293,13 +293,13 @@ finalize_it:
 
 /* Standard-Constructor for the strm object
  */
-BEGINobjConstruct(strm)
+BEGINobjConstruct(strm) /* be sure to specify the object type also in END macro! */
 	pThis->iCurrFNum = 1;
 	pThis->fd = -1;
 	pThis->iUngetC = -1;
 	pThis->sType = STREAMTYPE_FILE_SINGLE;
 	pThis->sIOBufSize = glblGetIOBufSize();
-	pThis->tOpenMode = 0600;
+	pThis->tOpenMode = 0600; /* TODO: make configurable */
 ENDobjConstruct(strm)
 
 
@@ -319,7 +319,7 @@ rsRetVal strmConstructFinalize(strm_t *pThis)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -347,7 +347,7 @@ rsRetVal strmDestruct(strm_t **ppThis)
 	free(pThis);
 	*ppThis = NULL;
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -370,7 +370,7 @@ static rsRetVal strmCheckNextOutputFile(strm_t *pThis)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* write memory buffer to a stream object.
@@ -413,7 +413,7 @@ static rsRetVal strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 finalize_it:
 	pThis->iBufPtr = 0; /* see comment above */
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -432,7 +432,7 @@ rsRetVal strmFlush(strm_t *pThis)
 		iRet = strmWriteInternal(pThis, pThis->pIOBuf, pThis->iBufPtr);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -457,7 +457,7 @@ dbgprintf("seek(%d, %ld): %d\n", pThis->fd, offs, i);
 	pThis->iCurrOffs = offs; /* we are now at *this* offset */
 	pThis->iBufPtr = 0; /* buffer invalidated */
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -471,7 +471,7 @@ rsRetVal strmSeekCurrOffs(strm_t *pThis)
 	ISOBJ_TYPE_assert(pThis, strm);
 
 	iRet = strmSeek(pThis, pThis->iCurrOffs);
-	return iRet;
+	RETiRet;
 }
 
 
@@ -492,7 +492,7 @@ rsRetVal strmWriteChar(strm_t *pThis, uchar c)
 	pThis->iBufPtr++;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -508,7 +508,7 @@ rsRetVal strmWriteLong(strm_t *pThis, long i)
 	CHKiRet(strmWrite(pThis, szBuf, strlen((char*)szBuf)));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -551,7 +551,7 @@ rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -581,7 +581,7 @@ rsRetVal strmSetiAddtlOpenFlags(strm_t *pThis, int iNewVal)
 	pThis->iAddtlOpenFlags = iNewVal;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -608,7 +608,7 @@ strmSetFName(strm_t *pThis, uchar *pszName, size_t iLenName)
 	pThis->lenFName = iLenName;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -635,7 +635,7 @@ strmSetDir(strm_t *pThis, uchar *pszDir, size_t iLenDir)
 	pThis->lenDir = iLenDir;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -678,7 +678,7 @@ rsRetVal strmRecordEnd(strm_t *pThis)
 	pThis->bInRecord = 0;
 	iRet = strmCheckNextOutputFile(pThis); /* check if we need to switch files */
 
-	return iRet;
+	RETiRet;
 }
 /* end stream record support functions */
 
@@ -723,7 +723,7 @@ rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm)
 	CHKiRet(objEndSerialize(pStrm));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -763,7 +763,7 @@ rsRetVal strmSetProperty(strm_t *pThis, property_t *pProp)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 #undef	isProp
 
diff --git a/stringbuf.c b/stringbuf.c
index b777c40d..e89ce768 100755
--- a/stringbuf.c
+++ b/stringbuf.c
@@ -180,7 +180,7 @@ static rsRetVal rsCStrExtendBuf(rsCStrObj *pThis, size_t iMinNeeded)
 	pThis->pBuf = pNewBuf;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -191,7 +191,7 @@ finalize_it:
  */
 rsRetVal rsCStrAppendStrWithLen(rsCStrObj *pThis, uchar* psz, size_t iStrLen)
 {
-	rsRetVal iRet;
+	DEFiRet;
 
 	rsCHECKVALIDOBJECT(pThis, OIDrsCStr);
 	assert(psz != NULL);
@@ -206,7 +206,7 @@ rsRetVal rsCStrAppendStrWithLen(rsCStrObj *pThis, uchar* psz, size_t iStrLen)
 	pThis->iStrLen += iStrLen;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -223,15 +223,16 @@ rsRetVal rsCStrAppendStr(rsCStrObj *pThis, uchar* psz)
 
 rsRetVal rsCStrAppendInt(rsCStrObj *pThis, long i)
 {
-	rsRetVal iRet;
+	DEFiRet;
 	uchar szBuf[32];
 
 	rsCHECKVALIDOBJECT(pThis, OIDrsCStr);
 
-	if((iRet = srUtilItoA((char*) szBuf, sizeof(szBuf), i)) != RS_RET_OK)
-		return iRet;
+	CHKiRet(srUtilItoA((char*) szBuf, sizeof(szBuf), i));
 
-	return rsCStrAppendStr(pThis, szBuf);
+	iRet = rsCStrAppendStr(pThis, szBuf);
+finalize_it:
+	RETiRet;
 }
 
 
@@ -255,7 +256,7 @@ rsRetVal rsCStrAppendChar(rsCStrObj *pThis, uchar c)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -420,7 +421,7 @@ finalize_it:
 		free(pThis->pBuf);
 	RSFREEOBJ(pThis);
 	
-	return(iRet);
+	RETiRet;
 }
 
 
@@ -625,10 +626,12 @@ int rsCStrStartsWithSzStr(rsCStrObj *pCS1, uchar *psz, size_t iLenSz)
 int rsCStrSzStrMatchRegex(rsCStrObj *pCS1, uchar *psz)
 {
     regex_t preq;
+    BEGINfunc
     regcomp(&preq, (char*) rsCStrGetSzStr(pCS1), 0);
-    int iRet = regexec(&preq, (char*) psz, 0, NULL, 0);
+    int ret = regexec(&preq, (char*) psz, 0, NULL, 0);
     regfree(&preq);
-    return iRet;
+    ENDfunc
+    return ret;
 }
 
 /* compare a rsCStr object with a classical sz string.  This function
@@ -654,6 +657,7 @@ int rsCStrSzStrMatchRegex(rsCStrObj *pCS1, uchar *psz)
  */
 int rsCStrOffsetSzStrCmp(rsCStrObj *pCS1, size_t iOffset, uchar *psz, size_t iLenSz)
 {
+	BEGINfunc
 	rsCHECKVALIDOBJECT(pCS1, OIDrsCStr);
 	assert(iOffset < pCS1->iStrLen);
 	assert(psz != NULL);
@@ -662,9 +666,10 @@ int rsCStrOffsetSzStrCmp(rsCStrObj *pCS1, size_t iOffset, uchar *psz, size_t iLe
 		/* we are using iLenSz below, because the lengths
 		 * are equal and iLenSz is faster to access
 		 */
-		if(iLenSz == 0)
+		if(iLenSz == 0) {
 			return 0; /* zero-sized strings are equal ;) */
-		else {  /* we now have two non-empty strings of equal
+			ENDfunc
+		} else {  /* we now have two non-empty strings of equal
 			 * length, so we need to actually check if they
 			 * are equal.
 			 */
@@ -675,10 +680,13 @@ int rsCStrOffsetSzStrCmp(rsCStrObj *pCS1, size_t iOffset, uchar *psz, size_t iLe
 			}
 			/* if we arrive here, the strings are equal */
 			return 0;
+			ENDfunc
 		}
 	}
-	else
+	else {
 		return pCS1->iStrLen - iOffset - iLenSz;
+		ENDfunc
+	}
 }
 
 
diff --git a/syslogd.c b/syslogd.c
index 7801d41f..09a246f9 100644
--- a/syslogd.c
+++ b/syslogd.c
@@ -178,6 +178,8 @@
 #include "threads.h"
 #include "queue.h"
 #include "stream.h"
+#include "wti.h"
+#include "wtp.h"
 
 /* We define our own set of syslog defintions so that we
  * do not need to rely on (possibly different) implementations.
@@ -259,7 +261,6 @@ char	ctty[] = _PATH_CONSOLE;	/* this is read-only; used by omfile -- TODO: remov
 
 static pid_t myPid;	/* our pid for use in self-generated messages, e.g. on startup */
 /* mypid is read-only after the initial fork() */
-static int debugging_on = 0; /* read-only, except on sig USR1 */
 static int restart = 0; /* do restart (config read) - multithread safe */
 
 int glblHadMemShortage = 0; /* indicates if we had memory shortage some time during the run */
@@ -368,7 +369,6 @@ static int	bDropTrailingLF = 1; /* drop trailing LF's on reception? */
 int	iCompatibilityMode = 0;		/* version we should be compatible with; 0 means sysklogd. It is
 					   the default, so if no -c<n> option is given, we make ourselvs
 					   as compatible to sysklogd as possible. */
-int	Debug;		/* debug flag  - read-only after startup */
 static int	bDebugPrintTemplateList = 1;/* output template list in debug mode? */
 static int	bDebugPrintCfSysLineHandlerList = 1;/* output cfsyslinehandler list in debug mode? */
 static int	bDebugPrintModuleList = 1;/* output module list in debug mode? */
@@ -558,7 +558,6 @@ static void debug_switch();
 static rsRetVal cfline(uchar *line, selector_t **pfCurr);
 static int decode(uchar *name, struct code *codetab);
 static void sighup_handler();
-//static void die(int sig);
 static void freeSelectors(void);
 static rsRetVal processConfFile(uchar *pConfFile);
 static rsRetVal selectorAddList(selector_t *f);
@@ -1157,7 +1156,7 @@ finalize_it:
 		}
 	}
 	*ppThis = pThis;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1315,7 +1314,7 @@ rsRetVal printline(char *hname, char *msg, int bParseHost)
 	logmsg(pri, pMsg, SYNC_FILE);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1531,7 +1530,7 @@ logmsgInternal(int pri, char *msg, int flags)
 		logmsg(pri, pMsg, flags);
 	}
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* This functions looks at the given message and checks if it matches the
@@ -1761,7 +1760,7 @@ finalize_it:
 	UnlockObj(pAction);
 	pthread_cleanup_pop(0); /* remove mutex cleanup handler */
 	pthread_setcancelstate(iCancelStateSave, NULL);
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1798,7 +1797,7 @@ DEFFUNC_llExecFunc(processMsgDoActions)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1812,6 +1811,7 @@ processMsg(msg_t *pMsg)
 	int bContinue;
 	processMsgDoActions_t DoActData;
 
+	BEGINfunc
 	assert(pMsg != NULL);
 
 	/* log the message to the particular outputs */
@@ -1829,6 +1829,7 @@ processMsg(msg_t *pMsg)
 		if(llExecFunc(&f->llActList, processMsgDoActions, (void*)&DoActData) == RS_RET_DISCARDMSG)
 			bContinue = 0;
 	}
+	ENDfunc
 }
 
 
@@ -1843,6 +1844,7 @@ processMsg(msg_t *pMsg)
 static rsRetVal
 msgConsumer(void *pUsr)
 {
+	DEFiRet;
 	msg_t *pMsg = (msg_t*) pUsr;
 
 	assert(pMsg != NULL);
@@ -1850,7 +1852,7 @@ msgConsumer(void *pUsr)
 	processMsg(pMsg);
 	MsgDestruct(&pMsg);
 
-	return RS_RET_OK;
+	RETiRet;
 }
 
 
@@ -2251,6 +2253,7 @@ logmsg(int pri, msg_t *pMsg, int flags)
 	char *msg;
 	char PRItext[20];
 
+	BEGINfunc
 	assert(pMsg != NULL);
 	assert(pMsg->pszUxTradMsg != NULL);
 	msg = (char*) pMsg->pszUxTradMsg;
@@ -2320,6 +2323,7 @@ logmsg(int pri, msg_t *pMsg, int flags)
 	pMsg->msgFlags = flags;
 	MsgPrepareEnqueue(pMsg);
 	queueEnqObj(pMsgQueue, (void*) pMsg);
+	ENDfunc
 }
 
 
@@ -2432,7 +2436,7 @@ finalize_it:
 		pAction->f_pMsg = pMsgSave;	/* restore it */
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2545,6 +2549,7 @@ void logerror(char *type)
 	char buf[1024];
 	char errStr[1024];
 
+	BEGINfunc
 	dbgprintf("Called logerr, msg: %s\n", type);
 
 	if (errno == 0)
@@ -2556,6 +2561,7 @@ void logerror(char *type)
 	buf[sizeof(buf)/sizeof(char) - 1] = '\0'; /* just to be on the safe side... */
 	errno = 0;
 	logmsgInternal(LOG_SYSLOG|LOG_ERR, buf, ADDDATE);
+	ENDfunc
 	return;
 }
 
@@ -2569,7 +2575,12 @@ void logerror(char *type)
  */
 static void doDie(int sig)
 {
+	static int iRetries = 0; /* debug aid */
 	dbgprintf("DoDie called.\n");
+	if(iRetries++ == 4) {
+		dbgprintf("DoDie called 5 times - unconditional exit\n");
+		exit(1);
+	}
 	bFinished = sig;
 }
 
@@ -2663,6 +2674,9 @@ die(int sig)
 	if(pModDir != NULL)
 		free(pModDir);
 
+	/* exit classes... */
+	dbgClassExit();
+
 	dbgprintf("Clean shutdown completed, bye.\n");
 	exit(0); /* "good" exit, this is the terminator function for rsyslog [die()] */
 }
@@ -2747,7 +2761,7 @@ finalize_it:
 	if(pDir != NULL)
 		closedir(pDir);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2797,7 +2811,7 @@ static rsRetVal doIncludeLine(uchar **pp, __attribute__((unused)) void* pVal)
 	globfree(&cfgFiles);
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2873,7 +2887,7 @@ static rsRetVal doModLoad(uchar **pp, __attribute__((unused)) void* pVal)
 	skipWhiteSpace(pp); /* skip over any whitespace */
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 /* parse and interpret a $-config line that starts with
@@ -2933,7 +2947,7 @@ static rsRetVal doNameLine(uchar **pp, void* pVal)
 	*pp = p;
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -2995,7 +3009,7 @@ rsRetVal cfsysline(uchar *p)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -3059,7 +3073,7 @@ DEFFUNC_llExecFunc(dbgPrintInitInfoAction)
 	iRet = actionDbgPrint((action_t*) pData);
 	printf("\n");
 
-	return iRet;
+	RETiRet;
 }
 
 /* print debug information as part of init(). This pretty much
@@ -3235,7 +3249,7 @@ finalize_it:
 		dbgprintf("error %d processing config file '%s'; os error (if any): %s\n",
 			iRet, pConfFile, errStr);
 	}
-	return iRet;
+	RETiRet;
 }
 
 
@@ -3263,6 +3277,7 @@ startInputModules(void)
 	pMod = modGetNxtType(pMod, eMOD_IN);
 	}
 
+	ENDfunc
 	return RS_RET_OK; /* intentional: we do not care about module errors */
 }
 
@@ -3448,6 +3463,7 @@ init(void)
 	sigaction(SIGHUP, &sigAct, NULL);
 
 	dbgprintf(" (re)started.\n");
+	ENDfunc
 }
 
 
@@ -3510,7 +3526,7 @@ rsRetVal cflineParseTemplateName(uchar** pp, omodStringRequest_t *pOMSR, int iEn
 finalize_it:
 	*pp = p;
 
-	return iRet;
+	RETiRet;
 }
 
 /* Helper to cfline(). Parses a file name up until the first
@@ -3540,7 +3556,7 @@ rsRetVal cflineParseFileName(uchar* p, uchar *pFileName, omodStringRequest_t *pO
 
 	iRet = cflineParseTemplateName(&p, pOMSR, iEntry, iTplOpts, (uchar*) " TradFmt");
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -3999,7 +4015,7 @@ finalize_it:
 			actionDestruct(pAction);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4037,7 +4053,7 @@ static rsRetVal cflineDoFilter(uchar **pp, selector_t *f)
 			return(iRet);
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4087,7 +4103,7 @@ static rsRetVal cflineDoAction(uchar **p, action_t **ppAction)
 	}
 
 	*ppAction = pAction;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4105,7 +4121,7 @@ DEFFUNC_llExecFunc(selectorAddListCheckActionsChecker)
 		Forwarding++;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 /* loop through a list of actions and perform necessary checks and
@@ -4123,7 +4139,7 @@ static rsRetVal selectorAddListCheckActions(selector_t *f)
 	CHKiRet(llExecFunc(&f->llActList, selectorAddListCheckActionsChecker, NULL));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4174,7 +4190,7 @@ static rsRetVal selectorAddList(selector_t *f)
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4215,7 +4231,7 @@ static rsRetVal cflineClassic(uchar *p, selector_t **pfCurr)
 
 finalize_it:
 	*pfCurr = fCurr;
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4249,7 +4265,7 @@ static rsRetVal cfline(uchar *line, selector_t **pfCurr)
 			break;
 	}
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4278,7 +4294,7 @@ static rsRetVal setMainMsgQueType(void __attribute__((unused)) *pVal, uchar *psz
 	}
 	free(pszType); /* no longer needed */
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4312,49 +4328,6 @@ int decode(uchar *name, struct code *codetab)
 	return (-1);
 }
 
-extern void dbgprintf(char *fmt, ...) __attribute__((format(printf,1, 2)));
-void
-dbgprintf(char *fmt, ...)
-{
-	static pthread_t ptLastThrdID = 0;
-	static int bWasNL = FALSE;
-	va_list ap;
-
-	if ( !(Debug && debugging_on) )
-		return;
-	
-	/* The bWasNL handler does not really work. It works if no thread
-	 * switching occurs during non-NL messages. Else, things are messed
-	 * up. Anyhow, it works well enough to provide useful help during
-	 * getting this up and running. It is questionable if the extra effort
-	 * is worth fixing it, giving the limited appliability.
-	 * rgerhards, 2005-10-25
-	 * I have decided that it is not worth fixing it - especially as it works
-	 * pretty well.
-	 * rgerhards, 2007-06-15
-	 */
-	if(ptLastThrdID != pthread_self()) {
-		if(!bWasNL) {
-			fprintf(stdout, "\n");
-			bWasNL = 1;
-		}
-		ptLastThrdID = pthread_self();
-	}
-
-	if(bWasNL) {
-		fprintf(stdout, "%8.8x: ", (unsigned int) pthread_self());
-		//fprintf(stderr, "%8.8x: ", (unsigned int) pthread_self());
-	}
-	bWasNL = (*(fmt + strlen(fmt) - 1) == '\n') ? TRUE : FALSE;
-	va_start(ap, fmt);
-	vfprintf(stdout, fmt, ap);
-	//vfprintf(stderr, fmt, ap);
-	va_end(ap);
-
-	//fflush(stderr);
-	fflush(stdout);
-	return;
-}
 
 
 /*
@@ -4448,6 +4421,7 @@ mainloop(void)
 {
 	struct timeval tvSelectTimeout;
 
+	BEGINfunc
 	while(!bFinished){
 		/* first check if we have any internal messages queued and spit them out */
 		/* TODO: do we need this any longer? I doubt it, but let's care about it
@@ -4493,6 +4467,7 @@ mainloop(void)
 			continue;
 		}
 	}
+	ENDfunc
 }
 
 /* If user is not root, prints warnings or even exits 
@@ -4542,16 +4517,20 @@ static rsRetVal loadBuildInModules(void)
 {
 	DEFiRet;
 
-	if((iRet = doModInit(modInitFile, (uchar*) "builtin-file", NULL)) != RS_RET_OK)
-		return iRet;
+	if((iRet = doModInit(modInitFile, (uchar*) "builtin-file", NULL)) != RS_RET_OK) {
+		RETiRet;
+	}
 #ifdef SYSLOG_INET
-	if((iRet = doModInit(modInitFwd, (uchar*) "builtin-fwd", NULL)) != RS_RET_OK)
-		return iRet;
+	if((iRet = doModInit(modInitFwd, (uchar*) "builtin-fwd", NULL)) != RS_RET_OK) {
+		RETiRet;
+	}
 #endif
-	if((iRet = doModInit(modInitShell, (uchar*) "builtin-shell", NULL)) != RS_RET_OK)
-		return iRet;
-	if((iRet = doModInit(modInitDiscard, (uchar*) "builtin-discard", NULL)) != RS_RET_OK)
-		return iRet;
+	if((iRet = doModInit(modInitShell, (uchar*) "builtin-shell", NULL)) != RS_RET_OK) {
+		RETiRet;
+	}
+	if((iRet = doModInit(modInitDiscard, (uchar*) "builtin-discard", NULL)) != RS_RET_OK) {
+		RETiRet;
+	}
 
 	/* dirty, but this must be for the time being: the usrmsg module must always be
 	 * loaded as last module. This is because it processes any time of action selector.
@@ -4563,7 +4542,7 @@ static rsRetVal loadBuildInModules(void)
 	 *   [a-zA-Z0-9_.]
 	 */
 	if((iRet = doModInit(modInitUsrMsg, (uchar*) "builtin-usrmsg", NULL)) != RS_RET_OK)
-		return iRet;
+		RETiRet;
 
 	/* ok, initialization of the command handler probably does not 100% belong right in
 	 * this space here. However, with the current design, this is actually quite a good
@@ -4607,7 +4586,7 @@ static rsRetVal loadBuildInModules(void)
 	CHKiRet(regCfSysLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, NULL));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -4708,6 +4687,7 @@ static void mainThread()
 	 */
 
 	mainloop();
+	ENDfunc
 }
 
 
@@ -4719,20 +4699,26 @@ static rsRetVal InitGlobalClasses(void)
 	DEFiRet;
 
 	CHKiRet(objClassInit()); /* *THIS* *MUST* always be the first class initilizere called! */
+fprintf(stdout, " calling dbgClassInit\n");
+	//CHKiRet(dbgClassInit());
 	CHKiRet(MsgClassInit());
 	CHKiRet(strmClassInit());
+	CHKiRet(wtiClassInit());
+	CHKiRet(wtpClassInit());
 	CHKiRet(queueClassInit());
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
+
 /* This is the main entry point into rsyslogd. Over time, we should try to
  * modularize it a bit more...
  */
 int main(int argc, char **argv)
 {	
+dbgClassInit();
 	DEFiRet;
 
 	register int i;
@@ -4987,6 +4973,8 @@ int main(int argc, char **argv)
 	memset(&sigAct, 0, sizeof (sigAct));
 	sigemptyset(&sigAct.sa_mask);
 
+	sigAct.sa_handler = sigsegvHdlr;
+	sigaction(SIGSEGV, &sigAct, NULL);
 	sigAct.sa_handler = doDie;
 	sigaction(SIGTERM, &sigAct, NULL);
 	sigAct.sa_handler = Debug ? doDie : SIG_IGN;
@@ -4995,7 +4983,8 @@ int main(int argc, char **argv)
 	sigAct.sa_handler = reapchild;
 	sigaction(SIGCHLD, &sigAct, NULL);
 	sigAct.sa_handler = Debug ? debug_switch : SIG_IGN;
-	sigaction(SIGUSR1, &sigAct, NULL);
+// TODO: use signal 2
+	//sigaction(SIGUSR1, &sigAct, NULL);
 	sigAct.sa_handler = SIG_IGN;
 	sigaction(SIGPIPE, &sigAct, NULL);
 	sigaction(SIGXFSZ, &sigAct, NULL); /* do not abort if 2gig file limit is hit */
@@ -5012,6 +5001,7 @@ finalize_it:
 	if(iRet != RS_RET_OK)
 		fprintf(stderr, "rsyslogd run failed with error %d.\n", iRet);
 
+	ENDfunc
 	return 0;
 }
 
diff --git a/tcpsyslog.c b/tcpsyslog.c
index 01655620..80f0d82a 100644
--- a/tcpsyslog.c
+++ b/tcpsyslog.c
@@ -1220,7 +1220,7 @@ static rsRetVal TCPSendBldFrame(TCPFRAMINGMODE rqdFraming, char **pmsg, size_t *
 	}
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -1282,7 +1282,7 @@ int TCPSend(void *pData, char *msg, size_t len, TCPFRAMINGMODE rqdFraming,
 finalize_it:
 	if(bMsgMustBeFreed)
 		free(msg);
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/template.c b/template.c
index ac436081..12096c91 100644
--- a/template.c
+++ b/template.c
@@ -132,7 +132,7 @@ rsRetVal tplToString(struct template *pTpl, msg_t *pMsg, uchar** ppSz)
 finalize_it:
 	*ppSz = (iRet == RS_RET_OK) ? pVal : NULL;
 
-	return iRet;
+	RETiRet;
 }
 
 /* Helper to doSQLEscape. This is called if doSQLEscape
diff --git a/threads.c b/threads.c
index 8ce5515d..c7b5f4e4 100644
--- a/threads.c
+++ b/threads.c
@@ -143,6 +143,7 @@ static void* thrdStarter(void *arg)
 	iRet = pThis->pUsrThrdMain(pThis);
 
 	dbgprintf("thrdStarter: usrThrdMain 0x%lx returned with iRet %d, exiting now.\n", (unsigned long) pThis->thrdID, iRet);
+	ENDfunc
 	pthread_exit(0);
 }
 
@@ -166,7 +167,7 @@ rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdI
 	CHKiRet(llAppend(&llThrds, NULL, pThis));
 
 finalize_it:
-	return iRet;
+	RETiRet;
 }
 
 
@@ -195,7 +196,7 @@ rsRetVal thrdInit(void)
 	sigAct.sa_handler = sigusr2Dummy;
 	sigaction(SIGUSR2, &sigAct, NULL);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -208,7 +209,7 @@ rsRetVal thrdExit(void)
 
 	iRet = llDestroy(&llThrds);
 
-	return iRet;
+	RETiRet;
 }
 
 
@@ -232,7 +233,7 @@ thrdSleep(thrdInfo_t *pThis, int iSeconds, int iuSeconds)
 	select(1, NULL, NULL, NULL, &tvSelectTimeout);
 	if(pThis->bShallStop)
 		iRet = RS_RET_TERMINATE_NOW;
-	return iRet;
+	RETiRet;
 }
 
 
diff --git a/wti.c b/wti.c
index 0e33b60f..045330c6 100644
--- a/wti.c
+++ b/wti.c
@@ -73,18 +73,20 @@ wtiGetDbgHdr(wti_t *pThis)
  * bite in the long term... -- rgerhards, 2008-01-17
  * TODO: may be performance optimized by atomic operations
  */
-static inline qWrkCmd_t
+qWrkCmd_t
 wtiGetState(wti_t *pThis, int bLockMutex)
 {
 	DEFVARS_mutexProtection;
 	qWrkCmd_t tCmd;
 
+	BEGINfunc
 	ISOBJ_TYPE_assert(pThis, wti);
 
 	BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex);
 	tCmd = pThis->tCurrCmd;
 	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 
+	ENDfunc
 	return tCmd;
 }
 
@@ -95,19 +97,20 @@ wtiGetState(wti_t *pThis, int bLockMutex)
  * in an active state. -- rgerhards, 2008-01-20
  */
 rsRetVal
-wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly)
+wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex)
 {
 	DEFiRet;
-	DEFVARS_mutex_cancelsafeLock;
+	DEFVARS_mutexProtection;
 
 	ISOBJ_TYPE_assert(pThis, wti);
 	assert(tCmd <= eWRKTHRD_SHUTDOWN_IMMEDIATE);
 
-	mutex_cancelsafe_lock(&pThis->mut);
+	BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex);
 
+RUNLOG_VAR("%d", bActiveOnly);
 	/* all worker states must be followed sequentially, only termination can be set in any state */
 	if(   (bActiveOnly && (pThis->tCurrCmd < eWRKTHRD_RUN_CREATED))
-	   || (pThis->tCurrCmd > tCmd && tCmd != eWRKTHRD_TERMINATING)) {
+	   || (pThis->tCurrCmd > tCmd && !(tCmd == eWRKTHRD_TERMINATING || tCmd == eWRKTHRD_STOPPED))) {
 		dbgprintf("%s: command %d can not be accepted in current %d processing state - ignored\n",
 			  wtiGetDbgHdr(pThis), tCmd, pThis->tCurrCmd);
 	} else {
@@ -137,7 +140,7 @@ wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly)
 		pThis->tCurrCmd = tCmd; /* apply the new state */
 	}
 
-	mutex_cancelsafe_unlock(&pThis->mut);
+	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 	RETiRet;
 }
 
@@ -188,15 +191,17 @@ ENDobjConstruct(wti)
 rsRetVal
 wtiConstructFinalize(wti_t *pThis)
 {
+	DEFiRet;
+
 	ISOBJ_TYPE_assert(pThis, wti);
 
 	dbgprintf("%s: finalizing construction of worker instance data\n", wtiGetDbgHdr(pThis));
 
 	/* initialize our thread instance descriptor */
-	pThis->pUsr = NULL;
+	pThis->pUsrp = NULL;
 	pThis->tCurrCmd = eWRKTHRD_STOPPED;
 
-	return RS_RET_OK;
+	RETiRet;
 }
 
 
@@ -211,7 +216,9 @@ wtiJoinThrd(wti_t *pThis)
 	ISOBJ_TYPE_assert(pThis, wti);
 	dbgprintf("wti: waiting for worker %s termination, current state %d\n", wtiGetDbgHdr(pThis), pThis->tCurrCmd);
 	pthread_join(pThis->thrdID, NULL);
-	wtiSetState(pThis, eWRKTHRD_STOPPED, 0); /* back to virgin... */
+RUNLOG;
+	wtiSetState(pThis, eWRKTHRD_STOPPED, 0, MUTEX_ALREADY_LOCKED); /* back to virgin... */
+RUNLOG;
 	pThis->thrdID = 0; /* invalidate the thread ID so that we do not accidently find reused ones */
 	dbgprintf("wti: worker %s has stopped\n", wtiGetDbgHdr(pThis));
 
@@ -261,6 +268,7 @@ wtiWorkerCancelCleanup(void *arg)
 	wtp_t *pWtp;
 	int iCancelStateSave;
 
+	BEGINfunc
 	ISOBJ_TYPE_assert(pThis, wti);
 	pWtp = pThis->pWtp;
 	ISOBJ_TYPE_assert(pWtp, wtp);
@@ -268,17 +276,17 @@ wtiWorkerCancelCleanup(void *arg)
 	dbgprintf("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis));
 	
 	/* call user supplied handler (that one e.g. requeues the element) */
-	pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr);
+	pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr, pThis->pUsrp);
 
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
 	d_pthread_mutex_lock(&pWtp->mut);
-	wtiSetState(pThis, eWRKTHRD_TERMINATING, 0);
+	wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED);
 	// TODO: sync access!
 	pWtp->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
 
-	pthread_cond_signal(&pWtp->condThrdTrm); /* activate anyone waiting on thread shutdown */
 	d_pthread_mutex_unlock(&pWtp->mut);
 	pthread_setcancelstate(iCancelStateSave, NULL);
+	ENDfunc
 }
 
 
@@ -318,8 +326,11 @@ wtiWorker(wti_t *pThis)
 	pWtp = pThis->pWtp; /* shortcut */
 	ISOBJ_TYPE_assert(pWtp, wtp);
 
+	dbgSetThrdName(pThis->pszDbgHdr);
 	pthread_cleanup_push(wtiWorkerCancelCleanup, pThis);
 
+	pWtp->pfOnWorkerStartup(pWtp->pUsr);
+
 	/* now we have our identity, on to real processing */
 	while(1) { /* loop will be broken below - need to do mutex locks */
 dbgprintf("%s: start worker run, queue cmd currently %d\n", wtiGetDbgHdr(pThis), pThis->tCurrCmd);
@@ -349,21 +360,22 @@ dbgprintf("%s: start worker run, queue cmd currently %d\n", wtiGetDbgHdr(pThis),
 			if(pWtp->toWrkShutdown == -1) {
 				dbgprintf("worker never times out!\n"); // DEL
 				/* never shut down any started worker */
-				pthread_cond_wait(pWtp->pcondBusy, pWtp->pmutUsr);
+				d_pthread_cond_wait(pWtp->pcondBusy, pWtp->pmutUsr);
 			} else {
 				timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */
-				if(pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) {
+				if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) {
 					dbgprintf("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis));
 					bInactivityTOOccured = 1; /* indicate we had a timeout */
 				}
 			}
-			dbgprintf("%s: post condwait ->notEmpty\n", wtiGetDbgHdr(pThis)); // DEL
+			dbgprintf("%s: post condwait ->Busy or timeout\n", wtiGetDbgHdr(pThis)); // DEL
 			END_MTX_PROTECTED_OPERATIONS(pWtp->pmutUsr);
 			continue; /* request next iteration */
 		}
 
 		/* if we reach this point, we have a non-empty queue (and are still protected by mutex) */
-		pWtp->pfDoWork(pThis, iCancelStateSave);
+		dbgprintf("%s: calling consumer\n", wtiGetDbgHdr(pThis));
+		pWtp->pfDoWork(pWtp->pUsr, pThis, iCancelStateSave);
 
 		/* TODO: move this above into one of the chck Term functions */
 		//if(Debug && (qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN) && pThis->iQueueSize > 0)
@@ -371,6 +383,8 @@ dbgprintf("%s: start worker run, queue cmd currently %d\n", wtiGetDbgHdr(pThis),
 		//		  " %d messages to process.\n", wtiGetDbgHdr(pThis), pThis->iQueueSize);
 	}
 
+	pWtp->pfOnWorkerShutdown(pWtp->pUsr);
+
 	/* indicate termination */
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
 dbgprintf("%s: worker waiting for mutex\n", wtiGetDbgHdr(pThis));
@@ -391,11 +405,9 @@ dbgprintf("%s: setting termination state\n", wtiGetDbgHdr(pThis));
 		wtiSetState(pWrkrInst, eWRKTHRD_TERMINATING, 0);
 	}
 #else
-	wtiSetState(pThis, eWRKTHRD_TERMINATING, 0);
+	wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED);
 #endif
-	// TODO: call, mutex:
 	pWtp->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
-	pthread_cond_signal(&pWtp->condThrdTrm); /* activate anyone waiting on thread shutdown */
 	d_pthread_mutex_unlock(&pThis->mut);
 	pthread_setcancelstate(iCancelStateSave, NULL);
 
@@ -404,6 +416,7 @@ dbgprintf("%s: setting termination state\n", wtiGetDbgHdr(pThis));
 
 
 /* some simple object access methods */
+DEFpropSetMeth(wti, pWtp, wtp_t*);
 
 /* set the debug header message
  * The passed-in string is duplicated. So if the caller does not need
@@ -441,7 +454,7 @@ finalize_it:
  * rgerhards, 2008-01-09
  */
 BEGINObjClassInit(wti, 1) /* one is the object version (most important for persisting) */
-ENDObjClassInit(queue)
+ENDObjClassInit(wti)
 
 /*
  * vi:set ai:
diff --git a/wti.h b/wti.h
index 4b028f73..db782d2b 100644
--- a/wti.h
+++ b/wti.h
@@ -32,7 +32,7 @@ typedef struct wti_s {
 	BEGINobjInstance;
 	pthread_t thrdID;  /* thread ID */
 	qWrkCmd_t tCurrCmd; /* current command to be carried out by worker */
-	obj_t *pUsr;		/* current user object being processed (or NULL if none) */
+	obj_t *pUsrp;		/* pointer to an object meaningful for current user pointer (e.g. queue pUsr data elemt) */
 	wtp_t *pWtp; /* my worker thread pool (important if only the work thread instance is passed! */
 	pthread_cond_t condInitDone; /* signaled when the thread startup is done (once per thread existance) */
 	pthread_mutex_t mut;
@@ -47,8 +47,13 @@ rsRetVal wtiConstruct(wti_t **ppThis);
 rsRetVal wtiConstructFinalize(wti_t *pThis);
 rsRetVal wtiDestruct(wti_t **ppThis);
 rsRetVal wtiWorker(wti_t *pThis);
+rsRetVal wtiProcessThrdChanges(wti_t *pThis, int bLockMutex);
+rsRetVal wtiSetDbgHdr(wti_t *pThis, uchar *pszMsg, size_t lenMsg);
+rsRetVal wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex);
+rsRetVal wtiJoinThrd(wti_t *pThis);
+qWrkCmd_t wtiGetState(wti_t *pThis, int bLockMutex);
 PROTOTYPEObjClassInit(wti);
 PROTOTYPEpropSetMeth(wti, pszDbgHdr, uchar*);
-#define wtiGetID(pThis) ((unsigned long) pThis)
+PROTOTYPEpropSetMeth(wti, pWtp, wtp_t*);
 
 #endif /* #ifndef WTI_H_INCLUDED */
diff --git a/wtp.c b/wtp.c
index 4bc0cd4f..4c3ea921 100644
--- a/wtp.c
+++ b/wtp.c
@@ -71,52 +71,59 @@ wtpGetDbgHdr(wtp_t *pThis)
 
 
 /* Not implemented dummy function for constructor */
-static rsRetVal NotImplementedDummy() { return RS_RET_NOT_IMPLEMENTED; }
+static rsRetVal NotImplementedDummy() { return RS_RET_OK; }
 /* Standard-Constructor for the wtp object
  */
 BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! */
-	int i;
-	uchar pszBuf[64];
-	size_t lenBuf;
-	wti_t *pWti;
-
 	pthread_mutex_init(&pThis->mut, 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;
 	pThis->pfIsIdle = NotImplementedDummy;
 	pThis->pfDoWork = NotImplementedDummy;
-	pThis->pfOnShutdownAdvise = NotImplementedDummy;
 	pThis->pfOnIdle = NotImplementedDummy;
 	pThis->pfOnWorkerCancel = NotImplementedDummy;
+	pThis->pfOnWorkerStartup = NotImplementedDummy;
+	pThis->pfOnWorkerShutdown = NotImplementedDummy;
+ENDobjConstruct(wtp)
+
+
+/* Construction finalizer
+ * rgerhards, 2008-01-17
+ */
+rsRetVal
+wtpConstructFinalize(wtp_t *pThis)
+{
+	DEFiRet;
+	int i;
+	uchar pszBuf[64];
+	size_t lenBuf;
+	wti_t *pWti;
 
-	/* alloc and construct workers */
+	ISOBJ_TYPE_assert(pThis, wtp);
+
+	dbgprintf("%s: finalizing construction of worker thread pool\n", wtpGetDbgHdr(pThis));
+	/* alloc and construct workers - this can only be done in finalizer as we previously do
+	 * not know the max number of workers
+	 */
+RUNLOG;
 	if((pThis->pWrkr = malloc(sizeof(wti_t*) * pThis->iNumWorkerThreads)) == NULL)
 		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 
 	for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
+RUNLOG_VAR("%d", i);
+RUNLOG_VAR("%p", pThis->pWrkr[i]);
 		CHKiRet(wtiConstruct(&pThis->pWrkr[i]));
 		pWti = pThis->pWrkr[i];
 		lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "%s/w%d", wtpGetDbgHdr(pThis), i);
 		CHKiRet(wtiSetDbgHdr(pWti, pszBuf, lenBuf));
+		CHKiRet(wtiSetpWtp(pWti, pThis));
 		CHKiRet(wtiConstructFinalize(pWti));
 	}
 		
-finalize_it:
-ENDobjConstruct(wtp)
-
 
-/* Construction finalizer
- * rgerhards, 2008-01-17
- */
-rsRetVal
-wtpConstructFinalize(wtp_t __attribute__((unused)) *pThis)
-{
-	ISOBJ_TYPE_assert(pThis, wtp);
-
-	dbgprintf("%s: finalizing construction of worker thread pool\n", wtpGetDbgHdr(pThis));
-
-	return RS_RET_OK;
+finalize_it:
+	RETiRet;
 }
 
 
@@ -129,6 +136,8 @@ wtpDestruct(wtp_t **ppThis)
 	int iCancelStateSave;
 	int i;
 
+dbgPrintAllDebugInfo();
+RUNLOG;
 	assert(ppThis != NULL);
 	pThis = *ppThis;
 	ISOBJ_TYPE_assert(pThis, wtp);
@@ -161,17 +170,31 @@ wtpDestruct(wtp_t **ppThis)
 }
 
 
-/* wake up all worker threads. Param bWithDAWrk tells if the DA worker
- * is to be awaken, too. It needs special handling because it waits on
- * two different conditions depending on processing state.
+/* wake up at least one worker thread.
+ * rgerhards, 2008-01-20
+ */
+rsRetVal
+wtpWakeupWrkr(wtp_t *pThis)
+{
+	DEFiRet;
+
+	// TODO; mutex?
+	ISOBJ_TYPE_assert(pThis, wtp);
+dbgprintf("wtpWakeupWrkr 1, cond %p\n", pThis->pcondBusy);
+	pthread_cond_signal(pThis->pcondBusy);
+dbgprintf("wtpWakeupWrkr 2\n");
+	RETiRet;
+}
+/* wake up all worker threads.
  * rgerhards, 2008-01-16
  */
-static inline rsRetVal
-wtpWakeupWrks(wtp_t *pThis)
+rsRetVal
+wtpWakeupAllWrkr(wtp_t *pThis)
 {
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, wtp);
+	// TODO; mutex?
 	pthread_cond_broadcast(pThis->pcondBusy);
 	RETiRet;
 }
@@ -189,8 +212,10 @@ wtpProcessThrdChanges(wtp_t *pThis)
 
 	ISOBJ_TYPE_assert(pThis, wtp);
 
+	RUNLOG;
 	if(pThis->bThrdStateChanged == 0)
 		FINALIZE;
+	RUNLOG;
 
 	/* go through all threads */
 	for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
@@ -198,6 +223,7 @@ wtpProcessThrdChanges(wtp_t *pThis)
 	}
 
 finalize_it:
+	RUNLOG;
 	RETiRet;
 }
 
@@ -218,39 +244,68 @@ wtpSetState(wtp_t *pThis, wtpState_t iNewState)
 }
 
 
-#if 0
+/* check if the worker shall shutdown (1 = yes, 0 = no)
+ * TODO: check if we can use atomic operations to enhance performance
+ * Note: there may be two mutexes locked, the bLockUsrMutex is the one in our "user"
+ * (e.g. the queue clas)
+ * rgerhards, 2008-01-21
+ */
+rsRetVal
+wtpChkStopWrkr(wtp_t *pThis, int bLockMutex, int bLockUsrMutex)
+{
+	DEFiRet;
+	DEFVARS_mutexProtection;
+
+	BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex);
+	if(   (pThis->wtpState == wtpState_SHUTDOWN_IMMEDIATE)
+	   || ((pThis->wtpState == wtpState_SHUTDOWN) && pThis->pfIsIdle(pThis->pUsr, bLockUsrMutex)))
+		iRet = RS_RET_TERMINATE_NOW;
+	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
+
+	/* try customer handler if one was set and we do not yet have a definite result */
+	if(iRet == RS_RET_OK && pThis->pfChkStopWrkr != NULL) {
+		iRet = pThis->pfChkStopWrkr(pThis->pUsr, bLockUsrMutex);
+	}
+
+	RETiRet;
+}
+
+
 /* Send a shutdown command to all workers and see if they terminate.
  * A timeout may be specified.
  * rgerhards, 2008-01-14
  */
-static rsRetVal
-wtpWrkrShutdown(wtp_t *pThis)
+rsRetVal
+wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, long iTimeout)
 {
 	DEFiRet;
 	int bTimedOut;
 	struct timespec t;
 	int iCancelStateSave;
 
-	// TODO: implement
-	ISOBJ_TYPE_assert(pThis);
-	queueTellActWrkThrds(pThis, 0, tShutdownCmd);/* first tell the workers our request */
-	queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
-	/* race: must make sure all are running! */
-	queueTimeoutComp(&t, iTimeout);/* get timeout */
+dbgPrintAllDebugInfo();
+RUNLOG_VAR("%p", pThis);
+RUNLOG_VAR("%ld", iTimeout);
+RUNLOG_VAR("%d", tShutdownCmd);
+	ISOBJ_TYPE_assert(pThis, wtp);
+
+	wtpSetState(pThis, tShutdownCmd);
+	wtpWakeupAllWrkr(pThis);
+	timeoutComp(&t, iTimeout);/* get timeout */
 		
 	/* and wait for their termination */
-dbgprintf("Queue %p: waiting for mutex %p\n", pThis, &pThis->mutThrdMgmt);
+dbgprintf("%s: waiting for mutex %p\n", wtpGetDbgHdr(pThis), &pThis->mut);
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-	d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-	pthread_cleanup_push(queueMutexCleanup, &pThis->mutThrdMgmt);
+	d_pthread_mutex_lock(&pThis->mut);
+	pthread_cleanup_push(mutexCancelCleanup, &pThis->mut);
 	pthread_setcancelstate(iCancelStateSave, NULL);
 	bTimedOut = 0;
 	while(pThis->iCurNumWrkThrd > 0 && !bTimedOut) {
-		dbgprintf("Queue 0x%lx: waiting %ldms on worker thread termination, %d still running\n",
-			   queueGetID(pThis), iTimeout, pThis->iCurNumWrkThrd);
+		dbgprintf("%s: waiting %ldms on worker thread termination, %d still running\n",
+			   wtpGetDbgHdr(pThis), iTimeout, pThis->iCurNumWrkThrd);
 
-		if(pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mutThrdMgmt, &t) != 0) {
-			dbgprintf("Queue 0x%lx: timeout waiting on worker thread termination\n", queueGetID(pThis));
+		if(d_pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mut, &t) != 0) {
+			dbgprintf("%s: timeout waiting on worker thread termination\n", wtpGetDbgHdr(pThis));
 			bTimedOut = 1;	/* we exit the loop on timeout */
 		}
 	}
@@ -259,122 +314,61 @@ dbgprintf("Queue %p: waiting for mutex %p\n", pThis, &pThis->mutThrdMgmt);
 	if(bTimedOut)
 		iRet = RS_RET_TIMED_OUT;
 
+dbgprintf("wtpShutdownAll exit");
 	RETiRet;
 }
 
 
-/* Unconditionally cancel all running worker threads.
- * rgerhards, 2008-01-14
+/* indicate that a thread has terminated and awake anyone waiting on it
+ * rgerhards, 2008-01-23
  */
-static rsRetVal
-wtpWrkrCancel(wtp_t *pThis)
+rsRetVal wtpSignalWrkrTermination(wtp_t *pThis)
 {
 	DEFiRet;
-	int i;
-	// TODO: we need to implement peek(), without it (today!) we lose one message upon
-	// worker cancellation! -- rgerhards, 2008-01-14
-
-	ISOB_TYPE_assert(pThis);
-	/* process any pending thread requests so that we know who actually is still running */
-	queueChkWrkThrdChanges(pThis);
-
-	/* awake the workers one more time, just to be sure */
-	queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
+	//TODO: mutex or not mutex, that's the question ;)DEFVARS_mutexProtection;
 
-	/* first tell the workers our request */
-	for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
-		if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRD_TERMINATING) {
-			dbgprintf("Queue 0x%lx: canceling worker thread %d\n", queueGetID(pThis), i);
-			pthread_cancel(pThis->pWrkThrds[i].thrdID);
-		}
-	}
+	ISOBJ_TYPE_assert(pThis, wtp);
 
+	//BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, LOCK_MUTEX);
+dbgprintf("signaling thread termination, cond %p\n", &pThis->condThrdTrm);
+	pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
+	//END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 	RETiRet;
 }
 
 
-/* Worker thread management function carried out when the main
- * worker is about to terminate.
+/* Unconditionally cancel all running worker threads.
+ * rgerhards, 2008-01-14
  */
-static rsRetVal
-wtpShutdownWorkers(wtp_t *pThis)
+rsRetVal
+wtpCancelAll(wtp_t *pThis)
 {
 	DEFiRet;
 	int i;
+	// TODO: we need to implement peek(), without it (today!) we lose one message upon
+	// worker cancellation! -- rgerhards, 2008-01-14
 
-	assert(pThis != NULL);
+	ISOBJ_TYPE_assert(pThis, wtp);
 
-	dbgprintf("Queue 0x%lx: initiating worker thread shutdown sequence\n", (unsigned long) pThis);
+	/* process any pending thread requests so that we know who actually is still running */
+	wtpProcessThrdChanges(pThis);
 
-	ISOB_TYPE_assert(pThis);
-	/* even if the timeout count is set to 0 (run endless), we still call the queueWrkThrdTrm(). This
-	 * is necessary so that all threads get sent the termination command. With a timeout of 0, however,
-	 * the function returns immediate with RS_RET_TIMED_OUT. We catch that state and accept it as
-	 * good.
-	 */
-	iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN, pThis->toQShutdown);
-	if(iRet == RS_RET_TIMED_OUT) {
-		if(pThis->toQShutdown == 0) {
-			iRet = RS_RET_OK;
-		} else {
-			/* OK, we now need to try force the shutdown */
-			dbgprintf("Queue 0x%lx: regular worker shutdown timed out, now trying immediate\n",
-				  queueGetID(pThis));
-			iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
-		}
-	}
+	/* awake the workers one more time, just to be sure */
+	wtpWakeupAllWrkr(pThis);
 
-	if(iRet != RS_RET_OK) { /* this is true on actual error on first try or timeout and error on second */
-		/* still didn't work out - so we now need to cancel the workers */
-		dbgprintf("Queue 0x%lx: worker threads could not be shutdown, now canceling them\n", (unsigned long) pThis);
-		iRet = queueWrkThrdCancel(pThis);
-	}
-	
-	/* finally join the threads
-	 * In case of a cancellation, this may actually take some time. This is also
-	 * needed to clean up the thread descriptors, even with a regular termination.
-	 * And, most importantly, this is needed if we have an indifitite termination
-	 * time set (timeout == 0)! -- rgerhards, 2008-01-14
-	 */
+	/* first tell the workers our request */
 	for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
-		if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRD_STOPPED) {
-			queueJoinWrkThrd(pThis, i);
+		// TODO: mutex lock!
+		if(pThis->pWrkr[i]->tCurrCmd >= eWRKTHRD_TERMINATING) {
+			dbgprintf("%s: canceling worker thread %d\n", wtpGetDbgHdr(pThis), i);
+			pthread_cancel(pThis->pWrkr[i]->thrdID);
 		}
 	}
 
-	dbgprintf("Queue 0x%lx: worker threads terminated, remaining queue size %d.\n",
-		  queueGetID(pThis), pThis->iQueueSize);
-
 	RETiRet;
 }
-#endif
-
-
-
-/* check if the worker shall shutdown
- * TODO: check if we can use atomic operations to enhance performance
- * Note: there may be two mutexes locked, the bLockUsrMutex is the one in our "user"
- * (e.g. the queue clas)
- * rgerhards, 2008-01-21
- */
-rsRetVal
-wtpChkStopWrkr(wtp_t *pThis, int bLockMutex, int bLockUsrMutex)
-{
-	DEFiRet;
-	DEFVARS_mutexProtection;
 
-	BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex);
-	if(   (pThis->wtpState == wtpState_SHUTDOWN_IMMEDIATE)
-	   || ((pThis->wtpState == wtpState_SHUTDOWN) && pThis->pfIsIdle(pThis->pUsr, bLockUsrMutex)))
-		iRet = RS_RET_TERMINATE_NOW;
-	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 
-	/* try customer handler if one was set and we do not yet have a definite result */
-	if(iRet == RS_RET_OK && pThis->pfChkStopWrkr != NULL)
-		iRet = pThis->pfChkStopWrkr(pThis->pUsr, bLockUsrMutex);
-
-	RETiRet;
-}
 
 /* Set the Inactivity Guard
  * rgerhards, 2008-01-21
@@ -385,9 +379,13 @@ wtpSetInactivityGuard(wtp_t *pThis, int bNewState, int bLockMutex)
 	DEFiRet;
 	DEFVARS_mutexProtection;
 
+RUNLOG;
 	BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex);
+RUNLOG;
 	pThis->bInactivityGuard = bNewState;
+RUNLOG;
 	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
+RUNLOG;
 
 	RETiRet;
 }
@@ -404,6 +402,7 @@ wtpWrkrExecCancelCleanup(void *arg)
 
 	ISOBJ_TYPE_assert(pThis, wtp);
 	pThis->iCurNumWrkThrd--;
+	wtpSignalWrkrTermination(pThis);
 
 	dbgprintf("%s: thread CANCELED with %d workers running.\n", wtpGetDbgHdr(pThis), pThis->iCurNumWrkThrd);
 
@@ -415,7 +414,7 @@ wtpWrkrExecCancelCleanup(void *arg)
  * rgerhards, 2008-01-21
  */
 static void *
-wtpWorker(void *arg)
+wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in wtp! */
 {
 	DEFiRet;
 	DEFVARS_mutexProtection;
@@ -442,7 +441,7 @@ wtpWorker(void *arg)
 	 * our init. That would be a bad race... -- rgerhards, 2008-01-16
 	 */
 	//if(qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_INIT)
-		wtiSetState(pWti, eWRKTHRD_RUNNING, MUTEX_ALREADY_LOCKED); /* we are running now! */
+		wtiSetState(pWti, eWRKTHRD_RUNNING, 0, MUTEX_ALREADY_LOCKED); /* we are running now! */
 
 	do {
 		END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
@@ -459,12 +458,14 @@ wtpWorker(void *arg)
 
 	pthread_cleanup_pop(0);
 	pThis->iCurNumWrkThrd--;
+	wtpSignalWrkrTermination(pThis);
 
 	dbgprintf("%s: Worker thread %lx, terminated, num workers now %d\n",
 		  wtpGetDbgHdr(pThis), (unsigned long) pWti, pThis->iCurNumWrkThrd);
 
 	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 
+	ENDfunc
 	pthread_exit(0);
 }
 
@@ -492,7 +493,9 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex)
 	 */
 	for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
 		// TODO: sync!
-		if(pThis->pWrkr[i]->tCurrCmd == eWRKTHRD_STOPPED) {
+RUNLOG;
+dbgprintf("%s: i %d, wti_T* %p\n", wtpGetDbgHdr(pThis), i, pThis->pWrkr[i]);
+		if(wtiGetState(pThis->pWrkr[i], LOCK_MUTEX) == eWRKTHRD_STOPPED) {
 			break;
 		}
 	}
@@ -502,11 +505,12 @@ dbgprintf("%s: after thrd search: i %d, max %d\n", wtpGetDbgHdr(pThis), i, pThis
 		ABORT_FINALIZE(RS_RET_NO_MORE_THREADS);
 
 	pWti = pThis->pWrkr[i];
-	wtiSetState(pWti, eWRKTHRD_RUN_CREATED, LOCK_MUTEX); // TODO: thuink about mutex lock
+	wtiSetState(pWti, eWRKTHRD_RUN_CREATED, 0, LOCK_MUTEX);
 	iState = pthread_create(&(pWti->thrdID), NULL, wtpWorker, (void*) pWti);
 	dbgprintf("%s: started with state %d, num workers now %d\n",
 		  wtpGetDbgHdr(pThis), iState, pThis->iCurNumWrkThrd);
 
+RUNLOG;
 	/* we try to give the starting worker a little boost. It won't help much as we still
  	 * hold the queue's mutex, but at least it has a chance to start on a single-CPU system.
  	 */
@@ -517,6 +521,7 @@ dbgprintf("%s: after thrd search: i %d, max %d\n", wtpGetDbgHdr(pThis), i, pThis
 
 finalize_it:
 	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
+RUNLOG;
 	RETiRet;
 }
 
@@ -524,7 +529,9 @@ finalize_it:
 /* set the number of worker threads that should be running. If less than currently running,
  * a new worker may be started. Please note that there is no guarantee the number of workers
  * said will be running after we exit this function. It is just a hint. If the number is
- * higher than one, the "busy" condition is also signaled to awake a worker.
+ * higher than one, and no worker is started, the "busy" condition is signaled to awake a worker.
+ * So the caller can assume that there is at least one worker re-checking if there is "work to do"
+ * after this function call.
  * rgerhards, 2008-01-21
  */
 rsRetVal
@@ -535,8 +542,10 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
 	int nMissing; /* number workers missing to run */
 	int i;
 
+	if(pThis == NULL) dbgPrintAllDebugInfo();
 	ISOBJ_TYPE_assert(pThis, wtp);
 
+dbgprintf("%s: wtpAdviseMaxWorker with %d called, currNum %d, max %d\n", wtpGetDbgHdr(pThis), nMaxWrkr, pThis->iCurNumWrkThrd, pThis->iNumWorkerThreads);
 	if(nMaxWrkr == 0)
 		FINALIZE;
 
@@ -548,15 +557,20 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
 	else if(nMissing < 0)
 		nMissing = 0;
 
-	dbgprintf("%s: high activity - starting %d additional worker thread(s).\n", wtpGetDbgHdr(pThis), nMissing);
-	/* start the rqtd nbr of workers */
-	for(i = 0 ; i < nMissing ; ++i) {
-		CHKiRet(wtpStartWrkr(pThis, MUTEX_ALREADY_LOCKED));
+	if(nMissing > 0) {
+		dbgprintf("%s: high activity - starting %d additional worker thread(s).\n", wtpGetDbgHdr(pThis), nMissing);
+		/* start the rqtd nbr of workers */
+		for(i = 0 ; i < nMissing ; ++i) {
+			CHKiRet(wtpStartWrkr(pThis, MUTEX_ALREADY_LOCKED));
+		}
+	} else {
+dbgprintf("wtpAdviseMaxWorkers signals busy\n");
+		wtpWakeupWrkr(pThis);
 	}
-	
-	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 
+	
 finalize_it:
+	END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
 	RETiRet;
 }
 
@@ -564,6 +578,17 @@ finalize_it:
 /* some simple object access methods */
 DEFpropSetMeth(wtp, toWrkShutdown, long);
 DEFpropSetMeth(wtp, wtpState, wtpState_t);
+DEFpropSetMeth(wtp, iNumWorkerThreads, int);
+DEFpropSetMeth(wtp, pUsr, void*);
+DEFpropSetMethPTR(wtp, pmutUsr, pthread_mutex_t);
+DEFpropSetMethPTR(wtp, pcondBusy, pthread_cond_t);
+DEFpropSetMethFP(wtp, pfChkStopWrkr, rsRetVal(*pVal)(void*, int));
+DEFpropSetMethFP(wtp, pfIsIdle, rsRetVal(*pVal)(void*, int));
+DEFpropSetMethFP(wtp, pfDoWork, rsRetVal(*pVal)(void*, void*, int));
+DEFpropSetMethFP(wtp, pfOnIdle, rsRetVal(*pVal)(void*, int));
+DEFpropSetMethFP(wtp, pfOnWorkerCancel, rsRetVal(*pVal)(void*, void*));
+DEFpropSetMethFP(wtp, pfOnWorkerStartup, rsRetVal(*pVal)(void*));
+DEFpropSetMethFP(wtp, pfOnWorkerShutdown, rsRetVal(*pVal)(void*));
 
 
 /* set the debug header message
@@ -572,11 +597,12 @@ DEFpropSetMeth(wtp, wtpState, wtpState_t);
  * rgerhards, 2008-01-09
  */
 rsRetVal
-wtpSetDbgHdr(wti_t *pThis, uchar *pszMsg, size_t lenMsg)
+wtpSetDbgHdr(wtp_t *pThis, uchar *pszMsg, size_t lenMsg)
 {
 	DEFiRet;
 
-	ISOBJ_TYPE_assert(pThis, wti);
+dbgprintf("objID: %d\n", pThis->pObjInfo->objID);
+	ISOBJ_TYPE_assert(pThis, wtp);
 	assert(pszMsg != NULL);
 	
 	if(lenMsg < 1)
diff --git a/wtp.h b/wtp.h
index 45fa2b8d..58fc8a5f 100644
--- a/wtp.h
+++ b/wtp.h
@@ -67,12 +67,13 @@ typedef struct wtp_s {
 	void *pUsr;		/* pointer to user object */
 	pthread_mutex_t *pmutUsr;
 	pthread_cond_t *pcondBusy; /* condition the user will signal "busy again, keep runing" on (awakes worker) */
-	rsRetVal (*pfChkStopWrkr)(void *, int);
-	rsRetVal (*pfIsIdle)(void *, int);
-	rsRetVal (*pfDoWork)(void *, int);
-	rsRetVal (*pfOnShutdownAdvise)(void *, int);
-	rsRetVal (*pfOnIdle)(void *, int);
-	rsRetVal (*pfOnWorkerCancel)(void *);
+	rsRetVal (*pfChkStopWrkr)(void *pUsr, int);
+	rsRetVal (*pfIsIdle)(void *pUsr, int);
+	rsRetVal (*pfDoWork)(void *pUsr, void *pWti, int);
+	rsRetVal (*pfOnIdle)(void *pUsr, int);
+	rsRetVal (*pfOnWorkerCancel)(void *pUsr, void*pWti);
+	rsRetVal (*pfOnWorkerStartup)(void *pUsr);
+	rsRetVal (*pfOnWorkerShutdown)(void *pUsr);
 	/* end user objects */
 	uchar *pszDbgHdr;	/* header string for debug messages */
 } wtp_t;
@@ -89,9 +90,25 @@ rsRetVal wtpProcessThrdChanges(wtp_t *pThis);
 rsRetVal wtpSetInactivityGuard(wtp_t *pThis, int bNewState, int bLockMutex);
 rsRetVal wtpChkStopWrkr(wtp_t *pThis, int bLockMutex, int bLockUsrMutex);
 rsRetVal wtpSetState(wtp_t *pThis, wtpState_t iNewState);
+rsRetVal wtpWakeupWrkr(wtp_t *pThis);
+rsRetVal wtpWakeupAllWrkr(wtp_t *pThis);
+rsRetVal wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, long iTimeout);
+rsRetVal wtpCancelAll(wtp_t *pThis);
+rsRetVal wtpSetDbgHdr(wtp_t *pThis, uchar *pszMsg, size_t lenMsg);
+rsRetVal wtpSignalWrkrTermination(wtp_t *pWtp);
 PROTOTYPEObjClassInit(wtp);
+PROTOTYPEpropSetMethFP(wtp, pfChkStopWrkr, rsRetVal(*pVal)(void*, int));
+PROTOTYPEpropSetMethFP(wtp, pfIsIdle, rsRetVal(*pVal)(void*, int));
+PROTOTYPEpropSetMethFP(wtp, pfDoWork, rsRetVal(*pVal)(void*, void*, int));
+PROTOTYPEpropSetMethFP(wtp, pfOnIdle, rsRetVal(*pVal)(void*, int));
+PROTOTYPEpropSetMethFP(wtp, pfOnWorkerCancel, rsRetVal(*pVal)(void*,void*));
+PROTOTYPEpropSetMethFP(wtp, pfOnWorkerStartup, rsRetVal(*pVal)(void*));
+PROTOTYPEpropSetMethFP(wtp, pfOnWorkerShutdown, rsRetVal(*pVal)(void*));
 PROTOTYPEpropSetMeth(wtp, toWrkShutdown, long);
 PROTOTYPEpropSetMeth(wtp, wtpState, wtpState_t);
-#define wtpGetID(pThis) ((unsigned long) pThis)
+PROTOTYPEpropSetMeth(wtp, iMaxWorkerThreads, int);
+PROTOTYPEpropSetMeth(wtp, pUsr, void*);
+PROTOTYPEpropSetMethPTR(wtp, pmutUsr, pthread_mutex_t);
+PROTOTYPEpropSetMethPTR(wtp, pcondBusy, pthread_cond_t);
 
 #endif /* #ifndef WTP_H_INCLUDED */