bugfix: fixed a memory leak and potential abort condition

this could happen if multiple rulesets were used and some output batches
contained messages belonging to more than one ruleset.
fixes: http://bugzilla.adiscon.com/show_bug.cgi?id=226
fixes: http://bugzilla.adiscon.com/show_bug.cgi?id=218

ChangeLog

@@ -1,5 +1,10 @@
 ---------------------------------------------------------------------------
 Version 5.7.6  [V5-BETA] (rgerhards), 2011-02-??
+- bugfix: fixed a memory leak and potential abort condition
+  this could happen if multiple rulesets were used and some output batches
+  contained messages belonging to more than one ruleset.
+  fixes: http://bugzilla.adiscon.com/show_bug.cgi?id=226
+  fixes: http://bugzilla.adiscon.com/show_bug.cgi?id=218
 - bugfix: memory leak when $RepeatedMsgReduction on was used
   bug tracker: http://bugzilla.adiscon.com/show_bug.cgi?id=225
 ---------------------------------------------------------------------------

runtime/batch.h

@@ -136,11 +136,16 @@ batchIsValidElem(batch_t *pBatch, int i) {
 /* copy one batch element to another.
  * This creates a complete duplicate in those cases where
  * it is needed. Use duplication only when absolutely necessary!
+ * Note that all working fields are reset to zeros. If that were 
+ * not done, we would have potential problems with invalid
+ * or double pointer frees.
  * rgerhards, 2010-06-10
  */
 static inline void
 batchCopyElem(batch_obj_t *pDest, batch_obj_t *pSrc) {
-	memcpy(pDest, pSrc, sizeof(batch_obj_t));
+	memset(pDest, 0, sizeof(batch_obj_t));
+	pDest->pUsrp = pSrc->pUsrp;
+	pDest->state = pSrc->state;
 }
 
 
@@ -171,6 +176,7 @@ batchFree(batch_t *pBatch) {
 static inline rsRetVal
 batchInit(batch_t *pBatch, int maxElem) {
 	DEFiRet;
+	pBatch->iDoneUpTo = 0;
 	pBatch->maxElem = maxElem;
 	CHKmalloc(pBatch->pElem = calloc((size_t)maxElem, sizeof(batch_obj_t)));
 	// TODO: replace calloc by inidividual writes?

runtime/ruleset.c

@@ -171,35 +171,40 @@ processBatchMultiRuleset(batch_t *pBatch)
 	int i;
 	int iStart;	/* start index of partial batch */
 	int iNew;	/* index for new (temporary) batch */
+	int bHaveUnprocessed;	/* do we (still) have unprocessed entries? (loop term predicate) */
 	DEFiRet;
 
-	CHKiRet(batchInit(&snglRuleBatch, pBatch->nElem));
-	snglRuleBatch.pbShutdownImmediate = pBatch->pbShutdownImmediate;
-
-	while(1) { /* loop broken inside */
+	do {
+		bHaveUnprocessed = 0;
 		/* search for first unprocessed element */
 		for(iStart = 0 ; iStart < pBatch->nElem && pBatch->pElem[iStart].state == BATCH_STATE_DISC ; ++iStart)
 			/* just search, no action */;
-
 		if(iStart == pBatch->nElem)
-			FINALIZE; /* everything processed */
+			break; /* everything processed */
 
 		/* prepare temporary batch */
+		CHKiRet(batchInit(&snglRuleBatch, pBatch->nElem));
+		snglRuleBatch.pbShutdownImmediate = pBatch->pbShutdownImmediate;
 		currRuleset = batchElemGetRuleset(pBatch, iStart);
 		iNew = 0;
 		for(i = iStart ; i < pBatch->nElem ; ++i) {
 			if(batchElemGetRuleset(pBatch, i) == currRuleset) {
-				batchCopyElem(&(snglRuleBatch.pElem[iNew++]), &(pBatch->pElem[i]));
+				/* for performance reasons, we copy only those members that we actually need */
+				snglRuleBatch.pElem[iNew].pUsrp = pBatch->pElem[i].pUsrp;
+				snglRuleBatch.pElem[iNew].state = pBatch->pElem[i].state;
+				++iNew;
 				/* We indicate the element also as done, so it will not be processed again */
 				pBatch->pElem[i].state = BATCH_STATE_DISC;
+			} else {
+				bHaveUnprocessed = 1;
 			}
 		}
 		snglRuleBatch.nElem = iNew; /* was left just right by the for loop */
 		batchSetSingleRuleset(&snglRuleBatch, 1);
 		/* process temp batch */
 		processBatch(&snglRuleBatch);
-	}
-	batchFree(&snglRuleBatch);
+		batchFree(&snglRuleBatch);
+	} while(bHaveUnprocessed == 1);
 
 finalize_it:
 	RETiRet;