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...

1 files changed, 418 insertions, 1161 deletions

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)
 
 /*