bugfix: disk queue was not persisted on shutdown, regression of fix to

http://bugzilla.adiscon.com/show_bug.cgi?id=299
The new code also handles the case of shutdown of blocking light and
full delayable sources somewhat smarter and permits, assuming sufficient
timouts, to persist message up to the max queue capacity. Also some nits
debug instrumentation have been fixed.

2 files changed, 13 insertions, 11 deletions

diff --git a/ChangeLog b/ChangeLog
index 07ac7788..4f84804f 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,4 +1,12 @@
 ---------------------------------------------------------------------------
+Version 5.9.8  [V5-BETA], 2012-05-??
+- bugfix: disk queue was not persisted on shutdown, regression of fix to
+  http://bugzilla.adiscon.com/show_bug.cgi?id=299
+  The new code also handles the case of shutdown of blocking light and 
+  full delayable sources somewhat smarter and permits, assuming sufficient
+  timouts, to persist message up to the max queue capacity. Also some nits
+  in debug instrumentation have been fixed.
+---------------------------------------------------------------------------
 Version 5.9.7  [V5-BETA], 2012-05-10
 - added capability to specify substrings for field extraction mode
 - bugfix: ommysql did not properly init/exit the mysql runtime library
diff --git a/runtime/queue.c b/runtime/queue.c
index 7893a1db..50c65c5e 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -2309,10 +2309,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)
 	 */
@@ -2339,9 +2335,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.
@@ -2353,6 +2347,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) {
@@ -2365,11 +2361,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");
@@ -2394,6 +2387,7 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 	      	  && pThis->tVars.disk.sizeOnDisk > pThis->sizeOnDiskMax)) {
 		DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - waiting to drain.\n");
 		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);