milestone(BUGGY): batch now pushed down to action

at least in important cases (not for non-direct action queues and some
other minor things). This version is definitely buggy, but may be tried
with success on a non-production system. I will continue to work on the
correctness, but needed to commit now to get a baseline.

6 files changed, 190 insertions, 16 deletions

diff --git a/runtime/batch.h b/runtime/batch.h
index 1245df11..80621631 100644
--- a/runtime/batch.h
+++ b/runtime/batch.h
@@ -26,6 +26,7 @@
 #ifndef BATCH_H_INCLUDED
 #define BATCH_H_INCLUDED
 
+#include <string.h>
 #include "msg.h"
 
 /* enum for batch states. Actually, we violate a layer here, in that we assume that a batch is used
@@ -76,6 +77,7 @@ struct batch_obj_s {
  * is completed (else, the whole process does not work correctly).
  */
 struct batch_s {
+	int maxElem;		/* maximum number of elements that this batch supports */
 	int nElem;		/* actual number of element in this entry */
 	int nElemDeq;		/* actual number of elements dequeued (and thus to be deleted) - see comment above! */
 	int iDoneUpTo;		/* all messages below this index have state other than RDY */
@@ -92,15 +94,84 @@ batchSetSingleRuleset(batch_t *pBatch, sbool val) {
 	pBatch->bSingleRuleset = val;
 }
 
-/* get the batches ruleset */
+/* get the batches ruleset (if we have a single ruleset) */
 static inline ruleset_t*
 batchGetRuleset(batch_t *pBatch) {
 	return (pBatch->nElem > 0) ? ((msg_t*) pBatch->pElem[0].pUsrp)->pRuleset : NULL;
 }
 
+/* get the ruleset of a specifc element of the batch (index not verified!) */
+static inline ruleset_t*
+batchElemGetRuleset(batch_t *pBatch, int i) {
+	return ((msg_t*) pBatch->pElem[i].pUsrp)->pRuleset;
+}
+
 /* get number of msgs for this batch */
 static inline int
 batchNumMsgs(batch_t *pBatch) {
 	return pBatch->nElem;
 }
+
+
+/* set the status of the i-th batch element. Note that once the status is
+ * DISC, it will never be reset. So this function can NOT be used to initialize
+ * the state table. -- rgerhards, 2010-06-10
+ */
+static inline void
+batchSetElemState(batch_t *pBatch, int i, batch_state_t newState) {
+	if(pBatch->pElem[i].state != BATCH_STATE_DISC)
+		pBatch->pElem[i].state = newState;
+}
+
+
+/* check if an element is a valid entry. We do NOT verify if the
+ * element index is valid. -- rgerhards, 2010-06-10
+ */
+static inline int
+batchIsValidElem(batch_t *pBatch, int i) {
+	return(pBatch->pElem[i].bFilterOK && pBatch->pElem[i].state != BATCH_STATE_DISC);
+}
+
+
+/* copy one batch element to another.
+ * This creates a complete duplicate in those cases where
+ * it is needed. Use duplication only when absolutely necessary!
+ * rgerhards, 2010-06-10
+ */
+static inline void
+batchCopyElem(batch_obj_t *pDest, batch_obj_t *pSrc) {
+	memcpy(pDest, pSrc, sizeof(batch_obj_t));
+}
+
+
+/* free members of a batch "object". Note that we can not do the usual
+ * destruction as the object typically is allocated on the stack and so the
+ * object itself cannot be freed! -- rgerhards, 2010-06-15
+ */
+static inline void
+batchFree(batch_t *pBatch) {
+	int i;
+	int j;
+	for(i = 0 ; i < pBatch->maxElem ; ++i) {
+		for(j = 0 ; j < CONF_OMOD_NUMSTRINGS_BUFSIZE ; ++j) {
+			free(pBatch->pElem[i].staticActParams[j]);
+		}
+	}
+	free(pBatch->pElem);
+}
+
+
+/* initialiaze a batch "object". The record must already exist,
+ * we "just" initialize it. The max number of elements must be
+ * provided. -- rgerhards, 2010-06-15
+ */
+static inline rsRetVal
+batchInit(batch_t *pBatch, int maxElem) {
+	DEFiRet;
+	pBatch->maxElem = maxElem;
+	CHKmalloc(pBatch->pElem = calloc((size_t)maxElem, sizeof(batch_obj_t)));
+	// TODO: replace calloc by inidividual writes?
+finalize_it:
+	RETiRet;
+}
 #endif /* #ifndef BATCH_H_INCLUDED */
diff --git a/runtime/queue.c b/runtime/queue.c
index d437d590..5e9c67ca 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -841,6 +841,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
 	batch_obj_t batchObj;
 	DEFiRet;
 
+	//TODO: init batchObj (states _OK and new fields -- CHECK)
 	ASSERT(pThis != NULL);
 
 	/* calling the consumer is quite different here than it is from a worker thread */
@@ -861,6 +862,26 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
 	RETiRet;
 }
 
+/*** EXPERIMENTAL ***/
+rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch)
+{
+	DEFiRet;
+
+	ASSERT(pThis != NULL);
+
+	/* calling the consumer is quite different here than it is from a worker thread */
+	/* we need to provide the consumer's return value back to the caller because in direct
+	 * mode the consumer probably has a lot to convey (which get's lost in the other modes
+	 * because they are asynchronous. But direct mode is deliberately synchronous.
+	 * rgerhards, 2008-02-12
+	 * We use our knowledge about the batch_t structure below, but without that, we
+	 * pay a too-large performance toll... -- rgerhards, 2009-04-22
+	 */
+	iRet = pThis->pConsumer(pThis->pUsr, pBatch, &pThis->bShutdownImmediate);
+
+	RETiRet;
+}
+
 
 static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis)
 {
@@ -1364,10 +1385,10 @@ DeleteProcessedBatch(qqueue_t *pThis, batch_t *pBatch)
 	assert(pBatch != NULL);
 
 	for(i = 0 ; i < pBatch->nElem ; ++i) {
-dbgprintf("XXX: enqueueing data element %d of %d\n", i, pBatch->nElem);
 		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;
@@ -1385,7 +1406,7 @@ dbgprintf("XXX: enqueueing data element %d of %d\n", i, pBatch->nElem);
 
 	iRet = DeleteBatchFromQStore(pThis, pBatch);
 
-	pBatch->nElem = pBatch->nElemDeq = 0; /* reset batch */
+	pBatch->nElem = pBatch->nElemDeq = 0; /* reset batch */ // TODO: more fine init, new fields! 2010-06-14
 
 	RETiRet;
 }
@@ -1430,6 +1451,7 @@ 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;
 	}
 
@@ -2273,6 +2295,21 @@ finalize_it:
 /* ------------------------------ END multi-enqueue functions ------------------------------ */
 
 
+/* enqueue a new user data element in direct mode
+ * NOTE/TODO: This is a TESTER/EXPERIEMENTAL, to be changed to better
+ * code later on (like multi submit!) 2010-06-10
+ * Enqueues the new element and awakes worker thread.
+ */
+rsRetVal
+qqueueEnqObjDirect(qqueue_t *pThis, void *pUsr)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, qqueue);
+	iRet = qAddDirect(pThis, pUsr);
+	RETiRet;
+}
+
+
 /* enqueue a new user data element
  * Enqueues the new element and awakes worker thread.
  */
diff --git a/runtime/queue.h b/runtime/queue.h
index 33b21c9a..1c758134 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -177,12 +177,14 @@ struct queue_s {
 
 /* prototypes */
 rsRetVal qqueueDestruct(qqueue_t **ppThis);
+rsRetVal qqueueEnqObjDirect(qqueue_t *pThis, void *pUsr);
 rsRetVal qqueueEnqObj(qqueue_t *pThis, flowControl_t flwCtlType, void *pUsr);
 rsRetVal qqueueStart(qqueue_t *pThis);
 rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize);
 rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix);
 rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads,
 		        int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, int*));
+rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch);
 PROTOTYPEObjClassInit(qqueue);
 PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int);
 PROTOTYPEpropSetMeth(qqueue, bSyncQueueFiles, int);
diff --git a/runtime/rule.c b/runtime/rule.c
index c28e15c9..453e631d 100644
--- a/runtime/rule.c
+++ b/runtime/rule.c
@@ -108,16 +108,19 @@ DEFFUNC_llExecFunc(processBatchDoActions)
 	}
 #endif
 
-	// NEW (potentially): iRetMod = actionSubmit(pAction, (batch_t*) pParam);
+#if 1
+	// NEW (potentially):
+	iRetMod = doSubmitToActionQBatch(pAction, (batch_t*) pParam);
+#else
 	// old code -- milestone check
-dbgprintf("ZZZ: inside processBatchDoActions, begin processing (nElem=%d)\n", batchNumMsgs(pBatch));
 	int i;
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
-dbgprintf("ZZZ: inside processBatchDoActions, processind elem %d/%d\n", i, batchNumMsgs(pBatch));
+dbgprintf("ZZZ: inside processBatchDoActions, processing elem %d/%d\n", i, batchNumMsgs(pBatch));
 		if(pBatch->pElem[i].bFilterOK) {
 			iRetMod = pAction->submitToActQ(pAction, (msg_t*)(pBatch->pElem[i].pUsrp));
 		}
 	}
+#endif
 	//end old code
 #if 0 // TODO: this must be done inside the action as well!
 	if(iRetMod == RS_RET_DISCARDMSG) {
@@ -297,7 +300,7 @@ processBatch(rule_t *pThis, batch_t *pBatch)
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, rule);
-	assert(pMsg != NULL);
+	assert(pBatch != NULL);
 
 	/* first check the filters... */
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index caeb9357..31c2e1a7 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -60,6 +60,9 @@ linkedList_t llRulesets; /* this is NOT a pointer - no typo here ;) */
 ruleset_t *pCurrRuleset = NULL; /* currently "active" ruleset */
 ruleset_t *pDfltRuleset = NULL; /* current default ruleset, e.g. for binding to actions which have no other */
 
+/* forward definitions */
+static rsRetVal processBatch(batch_t *pBatch);
+
 /* ---------- linked-list key handling functions ---------- */
 
 /* destructor for linked list keys.
@@ -149,6 +152,69 @@ dbgprintf("ruleset: get iRet %d from rule.ProcessMsg()\n", iRet);
 }
 
 
+
+/* This function is similar to processBatch(), but works on a batch that
+ * contains rules from multiple rulesets. In this case, we can not push
+ * the whole batch through the ruleset. Instead, we examine it and
+ * partition it into sub-rulesets which we then push through the system.
+ * Note that when we evaluate which message must be processed, we do NOT need
+ * to look at bFilterOK, because this value is only set in a later processing
+ * stage. Doing so caused a bug during development ;)
+ * rgerhards, 2010-06-15
+ */
+static inline rsRetVal
+processBatchMultiRuleset(batch_t *pBatch)
+{
+	ruleset_t *currRuleset;
+	batch_t snglRuleBatch;
+	int i;
+	int iStart;	/* start index of partial batch */
+	int iNew;	/* index for new (temporary) batch */
+	DEFiRet;
+
+	CHKiRet(batchInit(&snglRuleBatch, pBatch->nElem));
+	snglRuleBatch.pbShutdownImmediate = pBatch->pbShutdownImmediate;
+
+dbgprintf("ZZZ: multi-ruleset batch of %d elements must be processed\n", pBatch->nElem);
+	while(1) { /* loop broken inside */
+		/* 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 */
+
+		/* prepare temporary batch */
+		currRuleset = batchElemGetRuleset(pBatch, iStart);
+		iNew = 0;
+		for(i = iStart ; i < pBatch->nElem ; ++i) {
+			if(batchElemGetRuleset(pBatch, i) == currRuleset) {
+dbgprintf("ZZZ: proc elem %d:'%s'\n", i, ((msg_t*)(pBatch->pElem[i].pUsrp))->szRawMsg+15);
+				batchCopyElem(&(snglRuleBatch.pElem[iNew++]), &(pBatch->pElem[i]));
+				/* We indicate the element also as done, so it will not be processed again */
+				pBatch->pElem[i].state = BATCH_STATE_DISC;
+			}
+		}
+		snglRuleBatch.nElem = iNew; /* was left just right by the for loop */
+		batchSetSingleRuleset(&snglRuleBatch, 1);
+		/* process temp batch */
+		processBatch(&snglRuleBatch);
+
+#if 0
+for(i = iStart ; i < pBatch->nElem ; ++i) {
+	dbgprintf("ZZZ: after partial execution item %d state %d\n", i, pBatch->pElem[i].state);
+}
+//dbgprintf("ZZZ: search item %d: state %d, bFilterOK %d, IsValid %d, msg:%s\n",
+//iStart, pBatch->pElem[iStart].state, pBatch->pElem[iStart].bFilterOK, batchIsValidElem(pBatch, iStart),
+//((msg_t*)(pBatch->pElem[iStart].pUsrp))->szRawMsg+40);
+#endif
+	}
+	batchFree(&snglRuleBatch);
+
+finalize_it:
+	RETiRet;
+}
+
 /* Process (consume) a batch of messages. Calls the actions configured.
  * If the whole batch uses a singel ruleset, we can process the batch as 
  * a whole. Otherwise, we need to process it slower, on a message-by-message
@@ -162,6 +228,7 @@ processBatch(batch_t *pBatch)
 	DEFiRet;
 	assert(pBatch != NULL);
 
+dbgprintf("ZZZ: processBatch: batch of %d elements must be processed\n", pBatch->nElem);
 	if(pBatch->bSingleRuleset) {
 		pThis = batchGetRuleset(pBatch);
 		if(pThis == NULL)
@@ -169,13 +236,7 @@ processBatch(batch_t *pBatch)
 		ISOBJ_TYPE_assert(pThis, ruleset);
 		CHKiRet(llExecFunc(&pThis->llRules, processBatchDoRules, pBatch));
 	} else {
-		#warning implementation missing!
-		/* we need to split of the batch according to rulesets used */
-		// TODO: do this at the deque level, much more performant!
-		assert(0); // TODO mandatory to implement!
-		dbgprintf("processbatch missing implementation, terminating!\n");
-		printf("processBatch missing implementation, terminating!\n");
-		exit(0);
+		CHKiRet(processBatchMultiRuleset(pBatch));
 	}
 
 finalize_it:
diff --git a/runtime/wti.c b/runtime/wti.c
index 307f1af1..9343f5c5 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -172,7 +172,7 @@ wtiCancelThrd(wti_t *pThis)
 BEGINobjDestruct(wti) /* be sure to specify the object type also in END and CODESTART macros! */
 CODESTARTobjDestruct(wti)
 	/* actual destruction */
-	free(pThis->batch.pElem);
+	batchFree(&pThis->batch);
 	DESTROY_ATOMIC_HELPER_MUT(pThis->mutIsRunning);
 
 	free(pThis->pszDbgHdr);
@@ -204,7 +204,7 @@ wtiConstructFinalize(wti_t *pThis)
 
 	/* we now alloc the array for user pointers. We obtain the max from the queue itself. */
 	CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize));
-	CHKmalloc(pThis->batch.pElem = calloc((size_t)iDeqBatchSize, sizeof(batch_obj_t)));
+	CHKiRet(batchInit(&pThis->batch, iDeqBatchSize));
 
 finalize_it:
 	RETiRet;