1 files changed, 78 insertions, 12 deletions

diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index 785fbeaa..0584e8d6 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -46,6 +46,7 @@
 #include "rule.h"
 #include "errmsg.h"
 #include "parser.h"
+#include "batch.h"
 #include "unicode-helper.h"
 #include "dirty.h" /* for main ruleset queue creation */
 
@@ -59,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.
@@ -134,34 +138,96 @@ finalize_it:
 
 
 
-/* helper to processMsg(), used to call the configured actions. It is
+/* helper to processBatch(), used to call the configured actions. It is
  * executed from within llExecFunc() of the action list.
  * rgerhards, 2007-08-02
  */
-DEFFUNC_llExecFunc(processMsgDoRules)
+DEFFUNC_llExecFunc(processBatchDoRules)
 {
 	rsRetVal iRet;
 	ISOBJ_TYPE_assert(pData, rule);
-	iRet = rule.ProcessMsg((rule_t*) pData, (msg_t*) pParam);
+	dbgprintf("Processing next rule\n");
+	iRet = rule.ProcessBatch((rule_t*) pData, (batch_t*) pParam);
 dbgprintf("ruleset: get iRet %d from rule.ProcessMsg()\n", iRet);
 	return iRet;
 }
 
 
-/* Process (consume) a received message. Calls the actions configured.
+
+/* 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;
+
+	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) {
+				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);
+	}
+	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
+ * basis (what can be optimized to a per-ruleset basis)
  * rgerhards, 2005-10-13
  */
 static rsRetVal
-processMsg(msg_t *pMsg)
+processBatch(batch_t *pBatch)
 {
 	ruleset_t *pThis;
 	DEFiRet;
-	assert(pMsg != NULL);
-
-	pThis = (pMsg->pRuleset == NULL) ? pDfltRuleset : pMsg->pRuleset;
-	ISOBJ_TYPE_assert(pThis, ruleset);
-
-	CHKiRet(llExecFunc(&pThis->llRules, processMsgDoRules, pMsg));
+	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)
+			pThis = pDfltRuleset;
+		ISOBJ_TYPE_assert(pThis, ruleset);
+		CHKiRet(llExecFunc(&pThis->llRules, processBatchDoRules, pBatch));
+	} else {
+		CHKiRet(processBatchMultiRuleset(pBatch));
+	}
 
 finalize_it:
 dbgprintf("ruleset.ProcessMsg() returns %d\n", iRet);
@@ -516,7 +582,7 @@ CODESTARTobjQueryInterface(ruleset)
 	pIf->IterateAllActions = iterateAllActions;
 	pIf->DestructAllActions = destructAllActions;
 	pIf->AddRule = addRule;
-	pIf->ProcessMsg = processMsg;
+	pIf->ProcessBatch = processBatch;
 	pIf->SetName = setName;
 	pIf->DebugPrintAll = debugPrintAll;
 	pIf->GetCurrent = GetCurrent;