Imported Upstream version 7.1.9upstream/7.1.9

1 files changed, 417 insertions, 135 deletions

diff --git a/runtime/queue.c b/runtime/queue.c
index e968806..0cd3370 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -12,7 +12,7 @@
  * function names - this makes it really hard to read and does not provide much
  * benefit, at least I (now) think so...
  *
- * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2008-2011 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of the rsyslog runtime library.
  *
@@ -83,17 +83,55 @@ static rsRetVal ConsumerDA(qqueue_t *pThis, wti_t *pWti);
 static rsRetVal batchProcessed(qqueue_t *pThis, wti_t *pWti);
 static rsRetVal qqueueMultiEnqObjNonDirect(qqueue_t *pThis, multi_submit_t *pMultiSub);
 static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub);
+static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr);
+static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis);
+static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis);
+static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis);
+static rsRetVal qDestructDisk(qqueue_t *pThis);
 
 /* some constants for queuePersist () */
 #define QUEUE_CHECKPOINT	1
 #define QUEUE_NO_CHECKPOINT	0
 
+/* tables for interfacing with the v6 config system */
+static struct cnfparamdescr cnfpdescr[] = {
+	{ "queue.filename", eCmdHdlrGetWord, 0 },
+	{ "queue.size", eCmdHdlrSize, 0 },
+	{ "queue.dequeuebatchsize", eCmdHdlrInt, 0 },
+	{ "queue.maxdiskspace", eCmdHdlrSize, 0 },
+	{ "queue.highwatermark", eCmdHdlrInt, 0 },
+	{ "queue.lowwatermark", eCmdHdlrInt, 0 },
+	{ "queue.fulldelaymark", eCmdHdlrInt, 0 },
+	{ "queue.lightdelaymark", eCmdHdlrInt, 0 },
+	{ "queue.discardmark", eCmdHdlrInt, 0 },
+	{ "queue.discardseverity", eCmdHdlrFacility, 0 },
+	{ "queue.checkpointinterval", eCmdHdlrInt, 0 },
+	{ "queue.syncqueuefiles", eCmdHdlrBinary, 0 },
+	{ "queue.type", eCmdHdlrQueueType, 0 },
+	{ "queue.workerthreads", eCmdHdlrInt, 0 },
+	{ "queue.timeoutshutdown", eCmdHdlrInt, 0 },
+	{ "queue.timeoutactioncompletion", eCmdHdlrInt, 0 },
+	{ "queue.timeoutenqueue", eCmdHdlrInt, 0 },
+	{ "queue.timeoutworkerthreadshutdown", eCmdHdlrInt, 0 },
+	{ "queue.workerthreadminimummessages", eCmdHdlrInt, 0 },
+	{ "queue.maxfilesize", eCmdHdlrSize, 0 },
+	{ "queue.saveonshutdown", eCmdHdlrBinary, 0 },
+	{ "queue.dequeueslowdown", eCmdHdlrInt, 0 },
+	{ "queue.dequeuetimebegin", eCmdHdlrInt, 0 },
+	{ "queue.dequeuetimeend", eCmdHdlrInt, 0 },
+};
+static struct cnfparamblk pblk =
+	{ CNFPARAMBLK_VERSION,
+	  sizeof(cnfpdescr)/sizeof(struct cnfparamdescr),
+	  cnfpdescr
+	};
+
 /* debug aid */
 static void displayBatchState(batch_t *pBatch)
 {
 	int i;
 	for(i = 0 ; i < pBatch->nElem ; ++i) {
-		dbgprintf("XXXXX: displayBatchState %p[%d]: %d\n", pBatch, i, pBatch->pElem[i].state);
+		DBGPRINTF("displayBatchState %p[%d]: %d\n", pBatch, i, pBatch->pElem[i].state);
 	}
 }
 
@@ -186,6 +224,59 @@ finalize_it:
 
 /* methods */
 
+static inline char *
+getQueueTypeName(queueType_t t)
+{
+	char *r;
+
+	switch(t) {
+	case QUEUETYPE_FIXED_ARRAY: 
+		r = "FixedArray";
+		break;
+	case QUEUETYPE_LINKEDLIST: 
+		r = "LinkedList";
+		break;
+	case QUEUETYPE_DISK: 
+		r = "Disk";
+		break;
+	case QUEUETYPE_DIRECT: 
+		r = "Direct";
+		break;
+	}
+	return r;
+}
+
+void
+qqueueDbgPrint(qqueue_t *pThis)
+{
+	dbgoprint((obj_t*) pThis, "parameter dump:\n");
+	dbgoprint((obj_t*) pThis, "queue.filename '%s'\n",
+		(pThis->pszFilePrefix == NULL) ? "[NONE]" : (char*)pThis->pszFilePrefix);
+	dbgoprint((obj_t*) pThis, "queue.size: %d\n", pThis->iMaxQueueSize);
+	dbgoprint((obj_t*) pThis, "queue.dequeuebatchsize: %d\n", pThis->iDeqBatchSize);
+	dbgoprint((obj_t*) pThis, "queue.maxdiskspace: %lld\n", pThis->iMaxFileSize);
+	dbgoprint((obj_t*) pThis, "queue.highwatermark: %d\n", pThis->iHighWtrMrk);
+	dbgoprint((obj_t*) pThis, "queue.lowwatermark: %d\n", pThis->iLowWtrMrk);
+	dbgoprint((obj_t*) pThis, "queue.fulldelaymark: %d\n", pThis->iFullDlyMrk);
+	dbgoprint((obj_t*) pThis, "queue.lightdelaymark: %d\n", pThis->iLightDlyMrk);
+	dbgoprint((obj_t*) pThis, "queue.discardmark: %d\n", pThis->iDiscardMrk);
+	dbgoprint((obj_t*) pThis, "queue.discardseverity: %d\n", pThis->iDiscardSeverity);
+	dbgoprint((obj_t*) pThis, "queue.checkpointinterval: %d\n", pThis->iPersistUpdCnt);
+	dbgoprint((obj_t*) pThis, "queue.syncqueuefiles: %d\n", pThis->bSyncQueueFiles);
+	dbgoprint((obj_t*) pThis, "queue.type: %d [%s]\n", pThis->qType, getQueueTypeName(pThis->qType));
+	dbgoprint((obj_t*) pThis, "queue.workerthreads: %d\n", pThis->iNumWorkerThreads);
+	dbgoprint((obj_t*) pThis, "queue.timeoutshutdown: %d\n", pThis->toQShutdown);
+	dbgoprint((obj_t*) pThis, "queue.timeoutactioncompletion: %d\n", pThis->toActShutdown);
+	dbgoprint((obj_t*) pThis, "queue.timeoutenqueue: %d\n", pThis->toEnq);
+	dbgoprint((obj_t*) pThis, "queue.timeoutworkerthreadshutdown: %d\n", pThis->toWrkShutdown);
+	dbgoprint((obj_t*) pThis, "queue.workerthreadminimummessages: %d\n", pThis->iMinMsgsPerWrkr);
+	dbgoprint((obj_t*) pThis, "queue.maxfilesize: %lld\n", pThis->iMaxFileSize);
+	dbgoprint((obj_t*) pThis, "queue.saveonshutdown: %d\n", pThis->bSaveOnShutdown);
+	dbgoprint((obj_t*) pThis, "queue.dequeueslowdown: %d\n", pThis->iDeqSlowdown);
+	dbgoprint((obj_t*) pThis, "queue.dequeuetimebegin: %d\n", pThis->iDeqtWinFromHr);
+	dbgoprint((obj_t*) pThis, "queuedequeuetimend.: %d\n", pThis->iDeqtWinToHr);
+}
+
 
 /* get the physical queue size. Must only be called
  * while mutex is locked!
@@ -592,6 +683,47 @@ static rsRetVal qDelLinkedList(qqueue_t *pThis)
 /* -------------------- disk  -------------------- */
 
 
+/* The following function is used to "save" ourself from being killed by
+ * a fatally failed disk queue. A fatal failure is, for example, if no 
+ * data can be read or written. In that case, the disk support is disabled,
+ * with all on-disk structures kept as-is as much as possible. Instead, the
+ * queue is switched to direct mode, so that at least 
+ * some processing can happen. Of course, this may still have lots of
+ * undesired side-effects, but is probably better than aborting the
+ * syslogd. Note that this function *must* succeed in one way or another, as
+ * we can not recover from failure here. But it may emit different return
+ * states, which can trigger different processing in the higher layers.
+ * rgerhards, 2011-05-03
+ */
+static inline rsRetVal
+queueSwitchToEmergencyMode(qqueue_t *pThis, rsRetVal initiatingError)
+{
+	pThis->iQueueSize = 0;
+	pThis->nLogDeq = 0;
+	qDestructDisk(pThis); /* free disk structures */
+
+	pThis->qType = QUEUETYPE_DIRECT;
+	pThis->qConstruct = qConstructDirect;
+	pThis->qDestruct = qDestructDirect;
+	pThis->qAdd = qAddDirect;
+	pThis->qDel = qDelDirect;
+	pThis->MultiEnq = qqueueMultiEnqObjDirect;
+	if(pThis->pqParent != NULL) {
+		DBGOPRINT((obj_t*) pThis, "DA queue is in emergency mode, disabling DA in parent\n");
+		pThis->pqParent->bIsDA = 0;
+		pThis->pqParent->pqDA = NULL;
+		/* This may have undesired side effects, not sure if I really evaluated
+		 * all. So you know where to look at if you come to this point during
+		 * troubleshooting ;) -- rgerhards, 2011-05-03
+		 */
+	}
+
+	errmsg.LogError(0, initiatingError, "fatal error on disk queue '%s', emergency switch to direct mode",
+			obj.GetName((obj_t*) pThis));
+	return RS_RET_ERR_QUEUE_EMERGENCY;
+}
+
+
 static rsRetVal
 qqueueLoadPersStrmInfoFixup(strm_t *pStrm, qqueue_t __attribute__((unused)) *pThis)
 {
@@ -794,10 +926,7 @@ finalize_it:
 static rsRetVal qDeqDisk(qqueue_t *pThis, void **ppUsr)
 {
 	DEFiRet;
-
-	CHKiRet(obj.Deserialize(ppUsr, (uchar*) "msg", pThis->tVars.disk.pReadDeq, NULL, NULL));
-
-finalize_it:
+	iRet = obj.Deserialize(ppUsr, (uchar*) "msg", pThis->tVars.disk.pReadDeq, NULL, NULL);
 	RETiRet;
 }
 
@@ -851,6 +980,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
 {
 	batch_t singleBatch;
 	batch_obj_t batchObj;
+	sbool active = 1;
 	int i;
 	DEFiRet;
 
@@ -869,9 +999,9 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
 	memset(&singleBatch, 0, sizeof(batch_t));
 	batchObj.state = BATCH_STATE_RDY;
 	batchObj.pUsrp = (obj_t*) pUsr;
-	batchObj.bFilterOK = 1;
 	singleBatch.nElem = 1; /* there always is only one in direct mode */
 	singleBatch.pElem = &batchObj;
+	singleBatch.active = &active;
 	iRet = pThis->pConsumer(pThis->pUsr, &singleBatch, &pThis->bShutdownImmediate);
 	/* delete the batch string params: TODO: create its own "class" for this */
 	for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
@@ -1224,8 +1354,8 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread
 		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 
 	/* set some water marks so that we have useful defaults if none are set specifically */
-	pThis->iFullDlyMrk  = iMaxQueueSize - (iMaxQueueSize / 100) *  3; /* default 97% */
-	pThis->iLightDlyMrk = iMaxQueueSize - (iMaxQueueSize / 100) * 30; /* default 70% */
+	pThis->iFullDlyMrk  = -1;
+	pThis->iLightDlyMrk = -1;
 	pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir);
 	pThis->iMaxFileSize = 1024 * 1024; /* default is 1 MiB */
 	pThis->iQueueSize = 0;
@@ -1239,42 +1369,6 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread
 	pThis->pszFilePrefix = NULL;
 	pThis->qType = qType;
 
-	/* set type-specific handlers and other very type-specific things (we can not totally hide it...) */
-	switch(qType) {
-		case QUEUETYPE_FIXED_ARRAY:
-			pThis->qConstruct = qConstructFixedArray;
-			pThis->qDestruct = qDestructFixedArray;
-			pThis->qAdd = qAddFixedArray;
-			pThis->qDeq = qDeqFixedArray;
-			pThis->qDel = qDelFixedArray;
-			pThis->MultiEnq = qqueueMultiEnqObjNonDirect;
-			break;
-		case QUEUETYPE_LINKEDLIST:
-			pThis->qConstruct = qConstructLinkedList;
-			pThis->qDestruct = qDestructLinkedList;
-			pThis->qAdd = qAddLinkedList;
-			pThis->qDeq = (rsRetVal (*)(qqueue_t*,void**)) qDeqLinkedList;
-			pThis->qDel = (rsRetVal (*)(qqueue_t*)) qDelLinkedList;
-			pThis->MultiEnq = qqueueMultiEnqObjNonDirect;
-			break;
-		case QUEUETYPE_DISK:
-			pThis->qConstruct = qConstructDisk;
-			pThis->qDestruct = qDestructDisk;
-			pThis->qAdd = qAddDisk;
-			pThis->qDeq = qDeqDisk;
-			pThis->qDel = qDelDisk;
-			pThis->MultiEnq = qqueueMultiEnqObjNonDirect;
-			/* special handling */
-			pThis->iNumWorkerThreads = 1; /* we need exactly one worker */
-			break;
-		case QUEUETYPE_DIRECT:
-			pThis->qConstruct = qConstructDirect;
-			pThis->qDestruct = qDestructDirect;
-			pThis->qAdd = qAddDirect;
-			pThis->qDel = qDelDirect;
-			pThis->MultiEnq = qqueueMultiEnqObjDirect;
-			break;
-	}
 
 	INIT_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
 	INIT_ATOMIC_HELPER_MUT(pThis->mutLogDeq);
@@ -1285,6 +1379,40 @@ finalize_it:
 }
 
 
+/* set default inisde queue object suitable for action queues.
+ * This shall be called directly after queue construction. This functions has
+ * been added in support of the new v6 config system. It expect properly pre-initialized
+ * objects, but we need to differentiate between ruleset main and action queues.
+ * In order to avoid unnecessary complexity, we provide the necessary defaults
+ * via specific function calls.
+ */
+void
+qqueueSetDefaultsActionQueue(qqueue_t *pThis)
+{
+	pThis->qType = QUEUETYPE_DIRECT;	/* type of the main message queue above */
+	pThis->iMaxQueueSize = 1000;		/* size of the main message queue above */
+	pThis->iDeqBatchSize = 128; 		/* default batch size */
+	pThis->iHighWtrMrk = 800;		/* high water mark for disk-assisted queues */
+	pThis->iLowWtrMrk = 200;		/* low water mark for disk-assisted queues */
+	pThis->iDiscardMrk = 9800;		/* begin to discard messages */
+	pThis->iDiscardSeverity = 8;		/* turn off */
+	pThis->iNumWorkerThreads = 1;		/* number of worker threads for the mm queue above */
+	pThis->iMaxFileSize = 1024*1024;
+	pThis->iPersistUpdCnt = 0;		/* persist queue info every n updates */
+	pThis->bSyncQueueFiles = 0;
+	pThis->toQShutdown = 0;			/* queue shutdown */ 
+	pThis->toActShutdown = 1000;		/* action shutdown (in phase 2) */ 
+	pThis->toEnq = 2000;			/* timeout for queue enque */ 
+	pThis->toWrkShutdown = 60000;		/* timeout for worker thread shutdown */
+	pThis->iMinMsgsPerWrkr = 100;		/* minimum messages per worker needed to start a new one */
+	pThis->bSaveOnShutdown = 1;		/* save queue on shutdown (when DA enabled)? */
+	pThis->sizeOnDiskMax = 0;		/* unlimited */
+	pThis->iDeqSlowdown = 0;
+	pThis->iDeqtWinFromHr = 0;
+	pThis->iDeqtWinToHr = 25;		 /* disable time-windowed dequeuing by default */
+}
+
+
 /* 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.
@@ -1312,6 +1440,7 @@ static int qqueueChkDiscardMsg(qqueue_t *pThis, int iQueueSize, void *pUsr)
 		if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
 			DBGOPRINT((obj_t*) pThis, "queue nearly full (%d entries), discarded severity %d message\n",
 				  iQueueSize, iSeverity);
+			STATSCOUNTER_INC(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd);
 			objDestruct(pUsr);
 			ABORT_FINALIZE(RS_RET_QUEUE_FULL);
 		} else {
@@ -1343,7 +1472,8 @@ DoDeleteBatchFromQStore(qqueue_t *pThis, int nElem)
 	/* iQueueSize is not decremented by qDel(), so we need to do it ourselves */
 	ATOMIC_SUB(&pThis->iQueueSize, nElem, &pThis->mutQueueSize);
 	ATOMIC_SUB(&pThis->nLogDeq, nElem, &pThis->mutLogDeq);
-dbgprintf("delete batch from store, new sizes: log %d, phys %d\n", getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis));
+	DBGPRINTF("delete batch from store, new sizes: log %d, phys %d\n",
+		  getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis));
 	++pThis->deqIDDel; /* one more batch dequeued */
 
 	RETiRet;
@@ -1379,7 +1509,7 @@ DeleteBatchFromQStore(qqueue_t *pThis, batch_t *pBatch)
 		DoDeleteBatchFromQStore(pThis, pBatch->nElem);
 	} else {
 		/* can not delete, insert into to-delete list */
-		dbgprintf("not at head of to-delete list, enqueue %d\n", (int) pBatch->deqID);
+		DBGPRINTF("not at head of to-delete list, enqueue %d\n", (int) pBatch->deqID);
 		CHKiRet(tdlAdd(pThis, pBatch->deqID, pBatch->nElem));
 	}
 
@@ -1409,7 +1539,6 @@ DeleteProcessedBatch(qqueue_t *pThis, batch_t *pBatch)
 		pUsr = pBatch->pElem[i].pUsrp;
 		if(   pBatch->pElem[i].state == BATCH_STATE_RDY
 		   || pBatch->pElem[i].state == BATCH_STATE_SUB) {
-dbgprintf("XXX: DeleteProcessedBatch re-enqueue %d of %d, state %d\n", i, pBatch->nElem, pBatch->pElem[i].state);
 			localRet = doEnqSingleObj(pThis, eFLOWCTL_NO_DELAY,
 				       (obj_t*)MsgAddRef((msg_t*) pUsr));
 			++nEnqueued;
@@ -1420,7 +1549,7 @@ dbgprintf("XXX: DeleteProcessedBatch re-enqueue %d of %d, state %d\n", i, pBatch
 		objDestruct(pUsr);
 	}
 
-	dbgprintf("we deleted %d objects and enqueued %d objects\n", i-nEnqueued, nEnqueued);
+	DBGPRINTF("we deleted %d objects and enqueued %d objects\n", i-nEnqueued, nEnqueued);
 
 	if(nEnqueued > 0)
 		qqueueChkPersist(pThis, nEnqueued);
@@ -1472,7 +1601,6 @@ DequeueConsumableElements(qqueue_t *pThis, wti_t *pWti, int *piRemainingQueueSiz
 		/* all well, use this element */
 		pWti->batch.pElem[nDequeued].pUsrp = pUsr;
 		pWti->batch.pElem[nDequeued].state = BATCH_STATE_RDY;
-		pWti->batch.pElem[nDequeued].bFilterOK = 1; // TODO: think again if we can handle that with more performance
 		++nDequeued;
 	}
 
@@ -1693,7 +1821,18 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti)
 	ISOBJ_TYPE_assert(pThis, qqueue);
 	ISOBJ_TYPE_assert(pWti, wti);
 
-	CHKiRet(DequeueForConsumer(pThis, pWti));
+	iRet = DequeueForConsumer(pThis, pWti);
+	if(iRet == RS_RET_FILE_NOT_FOUND) {
+		/* This is a fatal condition and means the queue is almost unusable */
+		d_pthread_mutex_unlock(pThis->mut);
+		DBGOPRINT((obj_t*) pThis, "got 'file not found' error %d, queue defunct\n", iRet);
+		iRet = queueSwitchToEmergencyMode(pThis, iRet);
+		// TODO: think about what to return as iRet -- keep RS_RET_FILE_NOT_FOUND?
+		d_pthread_mutex_lock(pThis->mut);
+	}
+	if (iRet != RS_RET_OK) {
+		FINALIZE;
+	}
 
 	/* we now have a non-idle batch of work, so we can release the queue mutex and process it */
 	d_pthread_mutex_unlock(pThis->mut);
@@ -1743,6 +1882,7 @@ ConsumerDA(qqueue_t *pThis, wti_t *pWti)
 {
 	int i;
 	int iCancelStateSave;
+	int bNeedReLock = 0;	/**< do we need to lock the mutex again? */
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, qqueue);
@@ -1752,6 +1892,7 @@ ConsumerDA(qqueue_t *pThis, wti_t *pWti)
 
 	/* we now have a non-idle batch of work, so we can release the queue mutex and process it */
 	d_pthread_mutex_unlock(pThis->mut);
+	bNeedReLock = 1;
 
 	/* at this spot, we may be cancelled */
 	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave);
@@ -1770,10 +1911,10 @@ ConsumerDA(qqueue_t *pThis, wti_t *pWti)
 	/* but now cancellation is no longer permitted */
 	pthread_setcancelstate(iCancelStateSave, NULL);
 
-	/* now we are done, but need to re-aquire the mutex */
-	d_pthread_mutex_lock(pThis->mut);
-
 finalize_it:
+	/* now we are done, but potentially need to re-aquire the mutex */
+	if(bNeedReLock)
+		d_pthread_mutex_lock(pThis->mut);
 	DBGOPRINT((obj_t*) pThis, "DAConsumer returns with iRet %d\n", iRet);
 	RETiRet;
 }
@@ -1850,6 +1991,52 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 
 	ASSERT(pThis != NULL);
 
+	/* set type-specific handlers and other very type-specific things
+	 * (we can not totally hide it...)
+	 */
+	switch(pThis->qType) {
+		case QUEUETYPE_FIXED_ARRAY:
+			pThis->qConstruct = qConstructFixedArray;
+			pThis->qDestruct = qDestructFixedArray;
+			pThis->qAdd = qAddFixedArray;
+			pThis->qDeq = qDeqFixedArray;
+			pThis->qDel = qDelFixedArray;
+			pThis->MultiEnq = qqueueMultiEnqObjNonDirect;
+			break;
+		case QUEUETYPE_LINKEDLIST:
+			pThis->qConstruct = qConstructLinkedList;
+			pThis->qDestruct = qDestructLinkedList;
+			pThis->qAdd = qAddLinkedList;
+			pThis->qDeq = (rsRetVal (*)(qqueue_t*,void**)) qDeqLinkedList;
+			pThis->qDel = (rsRetVal (*)(qqueue_t*)) qDelLinkedList;
+			pThis->MultiEnq = qqueueMultiEnqObjNonDirect;
+			break;
+		case QUEUETYPE_DISK:
+			pThis->qConstruct = qConstructDisk;
+			pThis->qDestruct = qDestructDisk;
+			pThis->qAdd = qAddDisk;
+			pThis->qDeq = qDeqDisk;
+			pThis->qDel = qDelDisk;
+			pThis->MultiEnq = qqueueMultiEnqObjNonDirect;
+			/* special handling */
+			pThis->iNumWorkerThreads = 1; /* we need exactly one worker */
+			break;
+		case QUEUETYPE_DIRECT:
+			pThis->qConstruct = qConstructDirect;
+			pThis->qDestruct = qDestructDirect;
+			pThis->qAdd = qAddDirect;
+			pThis->qDel = qDelDirect;
+			pThis->MultiEnq = qqueueMultiEnqObjDirect;
+			break;
+	}
+
+	if(pThis->iFullDlyMrk == -1)
+		pThis->iFullDlyMrk  = pThis->iMaxQueueSize
+			- (pThis->iMaxQueueSize / 100) *  3; /* default 97% */
+	if(pThis->iLightDlyMrk == -1)
+		pThis->iLightDlyMrk = pThis->iMaxQueueSize
+			- (pThis->iMaxQueueSize / 100) * 30; /* default 70% */
+
 	/* we need to do a quick check if our water marks are set plausible. If not,
 	 * we correct the most important shortcomings. TODO: do that!!!! -- rgerhards, 2008-03-14
 	 */
@@ -1892,6 +2079,7 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 		  pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk,
 		  pThis->iDeqBatchSize);
 
+	pThis->bQueueStarted = 1;
 	if(pThis->qType == QUEUETYPE_DIRECT)
 		FINALIZE;	/* with direct queues, we are already finished... */
 
@@ -1922,7 +2110,6 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 	 * the case when a disk queue has been loaded. If we did not start it here, it would never start.
 	 */
 	qqueueAdviseMaxWorkers(pThis);
-	pThis->bQueueStarted = 1;
 
 	/* support statistics gathering */
 	qName = obj.GetName((obj_t*)pThis);
@@ -1941,6 +2128,13 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 	CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("full"),
 		ctrType_IntCtr, &pThis->ctrFull));
 
+	STATSCOUNTER_INIT(pThis->ctrFDscrd, pThis->mutCtrFDscrd);
+	CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.full"),
+		ctrType_IntCtr, &pThis->ctrFDscrd));
+	STATSCOUNTER_INIT(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd);
+	CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.nf"),
+		ctrType_IntCtr, &pThis->ctrNFDscrd));
+
 	pThis->ctrMaxqsize = 0; /* no mutex needed, thus no init call */
 	CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("maxqsize"),
 		ctrType_Int, &pThis->ctrMaxqsize));
@@ -2113,73 +2307,75 @@ DoSaveOnShutdown(qqueue_t *pThis)
 /* destructor for the queue object */
 BEGINobjDestruct(qqueue) /* be sure to specify the object type also in END and CODESTART macros! */
 CODESTARTobjDestruct(qqueue)
-	/* shut down all workers
-	 * We do not need to shutdown workers when we are in enqueue-only mode or we are a
-	 * direct queue - because in both cases we have none... ;)
-	 * with a child! -- rgerhards, 2008-01-28
-	 */
-	if(pThis->qType != QUEUETYPE_DIRECT && !pThis->bEnqOnly && pThis->pqParent == NULL
-	   && pThis->pWtpReg != NULL)
-		ShutdownWorkers(pThis);
+	if(pThis->bQueueStarted) {
+		/* shut down all workers
+		 * We do not need to shutdown workers when we are in enqueue-only mode or we are a
+		 * direct queue - because in both cases we have none... ;)
+		 * with a child! -- rgerhards, 2008-01-28
+		 */
+		if(pThis->qType != QUEUETYPE_DIRECT && !pThis->bEnqOnly && pThis->pqParent == NULL
+		   && pThis->pWtpReg != NULL)
+			ShutdownWorkers(pThis);
 
-	if(pThis->bIsDA && getPhysicalQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) {
-		CHKiRet(DoSaveOnShutdown(pThis));
-	}
+		if(pThis->bIsDA && getPhysicalQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) {
+			CHKiRet(DoSaveOnShutdown(pThis));
+		}
 
-	/* finally destruct our (regular) worker thread pool
-	 * Note: currently pWtpReg is never NULL, but if we optimize our logic, this may happen,
-	 * e.g. when they are not created in enqueue-only mode. We already check the condition
-	 * as this may otherwise be very hard to find once we optimize (and have long forgotten
-	 * about this condition here ;)
-	 * rgerhards, 2008-01-25
-	 */
-	if(pThis->qType != QUEUETYPE_DIRECT && pThis->pWtpReg != NULL) {
-		wtpDestruct(&pThis->pWtpReg);
-	}
+		/* finally destruct our (regular) worker thread pool
+		 * Note: currently pWtpReg is never NULL, but if we optimize our logic, this may happen,
+		 * e.g. when they are not created in enqueue-only mode. We already check the condition
+		 * as this may otherwise be very hard to find once we optimize (and have long forgotten
+		 * about this condition here ;)
+		 * rgerhards, 2008-01-25
+		 */
+		if(pThis->qType != QUEUETYPE_DIRECT && pThis->pWtpReg != NULL) {
+			wtpDestruct(&pThis->pWtpReg);
+		}
 
-	/* Now check if we actually have a DA queue and, if so, destruct it.
-	 * Note that the wtp must be destructed first, it may be in cancel cleanup handler
-	 * *right now* and actually *need* to access the queue object to persist some final
-	 * data (re-queueing case). So we need to destruct the wtp first, which will make 
-	 * sure all workers have terminated. Please note that this also generates a situation
-	 * where it is possible that the DA queue has a parent pointer but the parent has
-	 * no WtpDA associated with it - which is perfectly legal thanks to this code here.
-	 */
-	if(pThis->pWtpDA != NULL) {
-		wtpDestruct(&pThis->pWtpDA);
-	}
-	if(pThis->pqDA != NULL) {
-		qqueueDestruct(&pThis->pqDA);
-	}
+		/* Now check if we actually have a DA queue and, if so, destruct it.
+		 * Note that the wtp must be destructed first, it may be in cancel cleanup handler
+		 * *right now* and actually *need* to access the queue object to persist some final
+		 * data (re-queueing case). So we need to destruct the wtp first, which will make 
+		 * sure all workers have terminated. Please note that this also generates a situation
+		 * where it is possible that the DA queue has a parent pointer but the parent has
+		 * no WtpDA associated with it - which is perfectly legal thanks to this code here.
+		 */
+		if(pThis->pWtpDA != NULL) {
+			wtpDestruct(&pThis->pWtpDA);
+		}
+		if(pThis->pqDA != NULL) {
+			qqueueDestruct(&pThis->pqDA);
+		}
 
-	/* persist the queue (we always do that - queuePersits() does cleanup if the queue is empty)
-	 * This handler is most important for disk queues, it will finally persist the necessary
-	 * on-disk structures. In theory, other queueing modes may implement their other (non-DA)
-	 * methods of persisting a queue between runs, but in practice all of this is done via
-	 * disk queues and DA mode. Anyhow, it doesn't hurt to know that we could extend it here
-	 * if need arises (what I doubt...) -- rgerhards, 2008-01-25
-	 */
-	CHKiRet_Hdlr(qqueuePersist(pThis, QUEUE_NO_CHECKPOINT)) {
-		DBGOPRINT((obj_t*) pThis, "error %d persisting queue - data lost!\n", iRet);
-	}
+		/* persist the queue (we always do that - queuePersits() does cleanup if the queue is empty)
+		 * This handler is most important for disk queues, it will finally persist the necessary
+		 * on-disk structures. In theory, other queueing modes may implement their other (non-DA)
+		 * methods of persisting a queue between runs, but in practice all of this is done via
+		 * disk queues and DA mode. Anyhow, it doesn't hurt to know that we could extend it here
+		 * if need arises (what I doubt...) -- rgerhards, 2008-01-25
+		 */
+		CHKiRet_Hdlr(qqueuePersist(pThis, QUEUE_NO_CHECKPOINT)) {
+			DBGOPRINT((obj_t*) pThis, "error %d persisting queue - data lost!\n", iRet);
+		}
 
-	/* finally, clean up some simple things... */
-	if(pThis->pqParent == NULL) {
-		/* if we are not a child, we allocated our own mutex, which we now need to destroy */
-		pthread_mutex_destroy(pThis->mut);
-		free(pThis->mut);
-	}
-	pthread_mutex_destroy(&pThis->mutThrdMgmt);
-	pthread_cond_destroy(&pThis->notFull);
-	pthread_cond_destroy(&pThis->notEmpty);
-	pthread_cond_destroy(&pThis->belowFullDlyWtrMrk);
-	pthread_cond_destroy(&pThis->belowLightDlyWtrMrk);
+		/* finally, clean up some simple things... */
+		if(pThis->pqParent == NULL) {
+			/* if we are not a child, we allocated our own mutex, which we now need to destroy */
+			pthread_mutex_destroy(pThis->mut);
+			free(pThis->mut);
+		}
+		pthread_mutex_destroy(&pThis->mutThrdMgmt);
+		pthread_cond_destroy(&pThis->notFull);
+		pthread_cond_destroy(&pThis->notEmpty);
+		pthread_cond_destroy(&pThis->belowFullDlyWtrMrk);
+		pthread_cond_destroy(&pThis->belowLightDlyWtrMrk);
 
-	DESTROY_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
-	DESTROY_ATOMIC_HELPER_MUT(pThis->mutLogDeq);
+		DESTROY_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
+		DESTROY_ATOMIC_HELPER_MUT(pThis->mutLogDeq);
 
-	/* type-specific destructor */
-	iRet = pThis->qDestruct(pThis);
+		/* type-specific destructor */
+		iRet = pThis->qDestruct(pThis);
+	}
 
 	free(pThis->pszFilePrefix);
 	free(pThis->pszSpoolDir);
@@ -2247,10 +2443,6 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 	int err;
 	struct timespec t;
 
-	if(glbl.GetGlobalInputTermState()) {
-		ABORT_FINALIZE(RS_RET_FORCE_TERM);
-	}
-
 	STATSCOUNTER_INC(pThis->ctrEnqueued, pThis->mutCtrEnqueued);
 	/* first check if we need to discard this message (which will cause CHKiRet() to exit)
 	 */
@@ -2277,9 +2469,7 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 	 * It's a side effect, but a good one ;) -- rgerhards, 2008-03-14
 	 */
 	if(flowCtlType == eFLOWCTL_FULL_DELAY) {
-		DBGOPRINT((obj_t*) pThis, "enqueueMsg: FullDelay mark reached for full delayable message "
-		           "- blocking.\n");
-		while(pThis->iQueueSize >= pThis->iFullDlyMrk) {
+		while(pThis->iQueueSize >= pThis->iFullDlyMrk&& ! glbl.GetGlobalInputTermState()) {
 			/* We have a problem during shutdown if we block eternally. In that
 			 * case, the the input thread cannot be terminated. So we wake up
 			 * from time to time to check for termination.
@@ -2291,6 +2481,8 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 			 * In any case, this was the old code (if we do the TODO):
 			 * pthread_cond_wait(&pThis->belowFullDlyWtrMrk, pThis->mut);
 			 */
+			DBGOPRINT((obj_t*) pThis, "enqueueMsg: FullDelay mark reached for full delayable message "
+				   "- blocking, queue size is %d.\n", pThis->iQueueSize);
 			timeoutComp(&t, 1000);
 			err = pthread_cond_timedwait(&pThis->belowLightDlyWtrMrk, pThis->mut, &t);
 			if(err != 0 && err != ETIMEDOUT) {
@@ -2303,11 +2495,8 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 				
 			}
 			DBGPRINTF("wti worker in full delay timed out, checking termination...\n");
-			if(glbl.GetGlobalInputTermState()) {
-				ABORT_FINALIZE(RS_RET_FORCE_TERM);
-			}
 		}
-	} else if(flowCtlType == eFLOWCTL_LIGHT_DELAY) {
+	} else if(flowCtlType == eFLOWCTL_LIGHT_DELAY && !glbl.GetGlobalInputTermState()) {
 		if(pThis->iQueueSize >= pThis->iLightDlyMrk) {
 			DBGOPRINT((obj_t*) pThis, "enqueueMsg: LightDelay mark reached for light "
 			          "delayable message - blocking a bit.\n");
@@ -2330,19 +2519,27 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 	while(   (pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize)
 	      || (pThis->qType == QUEUETYPE_DISK && pThis->sizeOnDiskMax != 0
 	      	  && pThis->tVars.disk.sizeOnDisk > pThis->sizeOnDiskMax)) {
-		DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - waiting to drain.\n");
-		if(glbl.GetGlobalInputTermState()) {
-			ABORT_FINALIZE(RS_RET_FORCE_TERM);
-		}
-		timeoutComp(&t, pThis->toEnq);
 		STATSCOUNTER_INC(pThis->ctrFull, pThis->mutCtrFull);
-// TODO : handle enqOnly => discard!
-		if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
-			DBGOPRINT((obj_t*) pThis, "enqueueMsg: cond timeout, dropping message!\n");
+		if(pThis->toEnq == 0 || pThis->bEnqOnly) {
+			DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - configured for immediate discarding.\n");
+			STATSCOUNTER_INC(pThis->ctrFDscrd, pThis->mutCtrFDscrd);
 			objDestruct(pUsr);
 			ABORT_FINALIZE(RS_RET_QUEUE_FULL);
-		}
+		} else {
+			DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - waiting %dms to drain.\n", pThis->toEnq);
+			if(glbl.GetGlobalInputTermState()) {
+				DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL, discard due to FORCE_TERM.\n");
+				ABORT_FINALIZE(RS_RET_FORCE_TERM);
+			}
+			timeoutComp(&t, pThis->toEnq);
+			if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
+				DBGOPRINT((obj_t*) pThis, "enqueueMsg: cond timeout, dropping message!\n");
+				STATSCOUNTER_INC(pThis->ctrFDscrd, pThis->mutCtrFDscrd);
+				objDestruct(pUsr);
+				ABORT_FINALIZE(RS_RET_QUEUE_FULL);
+			}
 		dbgoprint((obj_t*) pThis, "enqueueMsg: wait solved queue full condition, enqueing\n");
+		}
 	}
 
 	/* and finally enqueue the message */
@@ -2465,6 +2662,90 @@ finalize_it:
 }
 
 
+/* take v6 config list and extract the queue params out of it. Hand the
+ * param values back to the caller. Caller is responsible for destructing
+ * them when no longer needed. Caller can use this param block to configure
+ * all parameters for a newly created queue with one call to qqueueSetParams().
+ * rgerhards, 2011-07-22
+ */
+rsRetVal
+qqueueDoCnfParams(struct nvlst *lst, struct cnfparamvals **ppvals)
+{
+	*ppvals = nvlstGetParams(lst, &pblk, NULL);
+	return RS_RET_OK;
+}
+
+/* apply all params from param block to queue. Must be called before
+ * finalizing. This supports the v6 config system. Defaults were already
+ * set during queue creation. The pvals object is destructed by this
+ * function.
+ */
+rsRetVal
+qqueueApplyCnfParam(qqueue_t *pThis, struct cnfparamvals *pvals)
+{
+	int i;
+	for(i = 0 ; i < pblk.nParams ; ++i) {
+		if(!pvals[i].bUsed)
+			continue;
+		if(!strcmp(pblk.descr[i].name, "queue.filename")) {
+			pThis->pszFilePrefix = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL);
+			pThis->lenFilePrefix = es_strlen(pvals[i].val.d.estr);
+		} else if(!strcmp(pblk.descr[i].name, "queue.size")) {
+			pThis->iMaxQueueSize = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.dequeuebatchsize")) {
+			pThis->iDeqBatchSize = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.maxdiskspace")) {
+			pThis->iMaxFileSize = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.highwatermark")) {
+			pThis->iHighWtrMrk = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.lowwatermark")) {
+			pThis->iLowWtrMrk = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.fulldelaymark")) {
+			pThis->iFullDlyMrk = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.lightdelaymark")) {
+			pThis->iLightDlyMrk = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.discardmark")) {
+			pThis->iDiscardMrk = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.discardseverity")) {
+			pThis->iDiscardSeverity = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.checkpointinterval")) {
+			pThis->iPersistUpdCnt = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.syncqueuefiles")) {
+			pThis->bSyncQueueFiles = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.type")) {
+			pThis->qType = (queueType_t) pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.workerthreads")) {
+			pThis->iNumWorkerThreads = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.timeoutshutdown")) {
+			pThis->toQShutdown = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.timeoutactioncompletion")) {
+			pThis->toActShutdown = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.timeoutenqueue")) {
+			pThis->toEnq = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.timeoutworkerthreadshutdown")) {
+			pThis->toWrkShutdown = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.workerthreadminimummessages")) {
+			pThis->iMinMsgsPerWrkr = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.maxfilesize")) {
+			pThis->iMaxFileSize = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.saveonshutdown")) {
+			pThis->bSaveOnShutdown = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.dequeueslowdown")) {
+			pThis->iDeqSlowdown = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queue.dequeuetimebegin")) {
+			pThis->iDeqtWinFromHr = pvals[i].val.d.n;
+		} else if(!strcmp(pblk.descr[i].name, "queuedequeuetimend.")) {
+			pThis->iDeqtWinToHr = pvals[i].val.d.n;
+		} else {
+			DBGPRINTF("queue: program error, non-handled "
+			  "param '%s'\n", pblk.descr[i].name);
+		}
+	}
+	cnfparamvalsDestruct(pvals, &pblk);
+	return RS_RET_OK;
+}
+
+
 /* some simple object access methods */
 DEFpropSetMeth(qqueue, bSyncQueueFiles, int)
 DEFpropSetMeth(qqueue, iPersistUpdCnt, int)
@@ -2479,6 +2760,7 @@ DEFpropSetMeth(qqueue, iLowWtrMrk, int)
 DEFpropSetMeth(qqueue, iDiscardMrk, int)
 DEFpropSetMeth(qqueue, iFullDlyMrk, int)
 DEFpropSetMeth(qqueue, iDiscardSeverity, int)
+DEFpropSetMeth(qqueue, iLightDlyMrk, int)
 DEFpropSetMeth(qqueue, bIsDA, int)
 DEFpropSetMeth(qqueue, iMinMsgsPerWrkr, int)
 DEFpropSetMeth(qqueue, bSaveOnShutdown, int)