1 files changed, 100 insertions, 269 deletions

diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index e39a1889..2e8d1f0f 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -48,6 +48,7 @@
 #include "rainerscript.h"
 #include "srUtils.h"
 #include "modules.h"
+#include "wti.h"
 #include "dirty.h" /* for main ruleset queue creation */
 
 /* static data */
@@ -68,7 +69,7 @@ static struct cnfparamblk rspblk =
 
 /* forward definitions */
 static rsRetVal processBatch(batch_t *pBatch, wti_t *pWti);
-static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti);
+static void scriptExec(struct cnfstmt *root, msg_t *pMsg, wti_t *pWti);
 
 
 /* ---------- linked-list key handling functions (ruleset) ---------- */
@@ -161,249 +162,89 @@ finalize_it:
 }
 
 
-/* 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.
- * rgerhards, 2010-06-15
- */
-static inline rsRetVal
-processBatchMultiRuleset(batch_t *pBatch, wti_t *pWti)
-{
-	ruleset_t *currRuleset;
-	batch_t snglRuleBatch;
-	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;
-
-	do {
-		bHaveUnprocessed = 0;
-		/* search for first unprocessed element */
-		for(iStart = 0 ; iStart < pBatch->nElem && pBatch->eltState[iStart] == BATCH_STATE_DISC ; ++iStart)
-			/* just search, no action */;
-		if(iStart == pBatch->nElem)
-			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) {
-				/* for performance reasons, we copy only those members that we actually need */
-				snglRuleBatch.pElem[iNew].pMsg = pBatch->pElem[i].pMsg;
-				snglRuleBatch.eltState[iNew] = pBatch->eltState[i];
-				++iNew;
-				/* We indicate the element also as done, so it will not be processed again */
-				pBatch->eltState[i] = 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, pWti);
-		batchFree(&snglRuleBatch);
-	} while(bHaveUnprocessed == 1);
-
-finalize_it:
-	RETiRet;
-}
-
-/* return a new "active" structure for the batch. Free with freeActive(). */
-static inline sbool *newActive(batch_t *pBatch)
+static void
+execAct(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti)
 {
-	return malloc(sizeof(sbool) * batchNumMsgs(pBatch));
-	
-}
-static inline void freeActive(sbool *active) { free(active); }
-
+	if(stmt->d.act->bDisabled) {
+		DBGPRINTF("action %d died, do NOT execute\n", stmt->d.act->iActionNbr);
+		goto done;
+	}
 
-/* for details, see scriptExec() header comment! */
-/* call action for all messages with filter on */
-static rsRetVal
-execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
-{
-	DEFiRet;
-dbgprintf("RRRR: execAct [%s]: batch of %d elements, active %p\n", modGetName(stmt->d.act->pMod), batchNumMsgs(pBatch), active);
-	pBatch->active = active;
-	stmt->d.act->submitToActQ(stmt->d.act, pBatch, pWti);
-	RETiRet;
+	DBGPRINTF("executing action %d\n", stmt->d.act->iActionNbr);
+	stmt->d.act->submitToActQ(stmt->d.act, pWti, pMsg);
+done:	return;
 }
 
-static rsRetVal
-execSet(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+static void
+execSet(struct cnfstmt *stmt, msg_t *pMsg)
 {
-	int i;
 	struct var result;
-	DEFiRet;
-	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
-		if(   pBatch->eltState[i] != BATCH_STATE_DISC
-		   && (active == NULL || active[i])) {
-			cnfexprEval(stmt->d.s_set.expr, &result, pBatch->pElem[i].pMsg);
-			msgSetJSONFromVar(pBatch->pElem[i].pMsg, stmt->d.s_set.varname,
-					  &result);
-			varDelete(&result);
-		}
-	}
-	RETiRet;
+	cnfexprEval(stmt->d.s_set.expr, &result, pMsg);
+	msgSetJSONFromVar(pMsg, stmt->d.s_set.varname, &result);
+	varDelete(&result);
 }
 
-static rsRetVal
-execUnset(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+static void
+execUnset(struct cnfstmt *stmt, msg_t *pMsg)
 {
-	int i;
-	DEFiRet;
-	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
-		if(   pBatch->eltState[i] != BATCH_STATE_DISC
-		   && (active == NULL || active[i])) {
-			msgDelJSON(pBatch->pElem[i].pMsg, stmt->d.s_unset.varname);
-		}
-	}
-	RETiRet;
+	msgDelJSON(pMsg, stmt->d.s_unset.varname);
 }
 
-/* for details, see scriptExec() header comment! */
-/* "stop" simply discards the filtered items - it's just a (hopefully more intuitive
- * shortcut for users.
- */
-static rsRetVal
-execStop(batch_t *pBatch, sbool *active)
-{
-	int i;
-	DEFiRet;
-	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
-		if(   pBatch->eltState[i] != BATCH_STATE_DISC
-		   && (active == NULL || active[i])) {
-			pBatch->eltState[i] = BATCH_STATE_DISC;
-		}
-	}
-	RETiRet;
-}
 static rsRetVal
-execCall(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
+execCall(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti)
 {
-	msg_t *pMsg;
-	int i;
 	DEFiRet;
 	if(stmt->d.s_call.ruleset == NULL) {
-		scriptExec(stmt->d.s_call.stmt, pBatch, active, pWti);
+		scriptExec(stmt->d.s_call.stmt, pMsg, pWti);
 	} else {
-		for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
-			CHKmalloc(pMsg = MsgDup((msg_t*) pBatch->pElem[i].pMsg));
-			DBGPRINTF("CALL: forwarding message %d to async ruleset %p\n",
-				  i, stmt->d.s_call.ruleset->pQueue);
-			MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);
-			MsgSetRuleset(pMsg, stmt->d.s_call.ruleset);
-			/* Note: we intentionally use submitMsg2() here, as we process messages
-			 * that were already run through the rate-limiter.
-			 */
-			submitMsg2(pMsg);
-		}
+		CHKmalloc(pMsg = MsgDup((msg_t*) pMsg));
+		DBGPRINTF("CALL: forwarding message to async ruleset %p\n",
+			  stmt->d.s_call.ruleset->pQueue);
+		MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);
+		MsgSetRuleset(pMsg, stmt->d.s_call.ruleset);
+		/* Note: we intentionally use submitMsg2() here, as we process messages
+		 * that were already run through the rate-limiter.
+		 */
+		submitMsg2(pMsg);
 	}
 finalize_it:
 	RETiRet;
 }
 
-/* for details, see scriptExec() header comment! */
-// save current filter, evaluate new one
-// perform then (if any message)
-// if ELSE given:
-//    set new filter, inverted
-//    perform else (if any messages)
-static rsRetVal
-execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
+static void
+execIf(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti)
 {
-	sbool *newAct;
-	int i;
 	sbool bRet;
-	sbool allInactive = 1;
-	DEFiRet;
-	newAct = newActive(pBatch);
-	for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
-		if(*(pBatch->pbShutdownImmediate))
-			FINALIZE;
-		if(pBatch->eltState[i] == BATCH_STATE_DISC)
-			continue; /* will be ignored in any case */
-		if(active == NULL || active[i]) {
-			bRet = cnfexprEvalBool(stmt->d.s_if.expr, pBatch->pElem[i].pMsg);
-			allInactive = 0;
-		} else 
-			bRet = 0;
-		newAct[i] = bRet;
-		DBGPRINTF("batch: item %d: expr eval: %d\n", i, bRet);
-	}
-
-	if(allInactive) {
-		DBGPRINTF("execIf: all batch elements are inactive, holding execution\n");
-		freeActive(newAct);
-		FINALIZE;
-	}
-
-	if(stmt->d.s_if.t_then != NULL) {
-		scriptExec(stmt->d.s_if.t_then, pBatch, newAct, pWti);
-	}
-	if(stmt->d.s_if.t_else != NULL) {
-		for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
-			if(*(pBatch->pbShutdownImmediate))
-				FINALIZE;
-			if(pBatch->eltState[i] != BATCH_STATE_DISC
-			   && (active == NULL || active[i]))
-				newAct[i] = !newAct[i];
-			}
-		scriptExec(stmt->d.s_if.t_else, pBatch, newAct, pWti);
+	bRet = cnfexprEvalBool(stmt->d.s_if.expr, pMsg);
+	DBGPRINTF("if condition result is %d\n", bRet);
+	if(bRet) {
+		if(stmt->d.s_if.t_then != NULL)
+			scriptExec(stmt->d.s_if.t_then, pMsg, pWti);
+	} else {
+		if(stmt->d.s_if.t_else != NULL)
+			scriptExec(stmt->d.s_if.t_else, pMsg, pWti);
 	}
-	freeActive(newAct);
-finalize_it:
-	RETiRet;
 }
 
-/* for details, see scriptExec() header comment! */
 static void
-execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
+execPRIFILT(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti)
 {
-	sbool *newAct;
-	msg_t *pMsg;
 	int bRet;
-	int i;
-	newAct = newActive(pBatch);
-	for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
-		if(*(pBatch->pbShutdownImmediate))
-			return;
-		if(pBatch->eltState[i] == BATCH_STATE_DISC)
-			continue; /* will be ignored in any case */
-		pMsg = pBatch->pElem[i].pMsg;
-		if(active == NULL || active[i]) {
-			if( (stmt->d.s_prifilt.pmask[pMsg->iFacility] == TABLE_NOPRI) ||
-			   ((stmt->d.s_prifilt.pmask[pMsg->iFacility]
-			            & (1<<pMsg->iSeverity)) == 0) )
-				bRet = 0;
-			else
-				bRet = 1;
-		} else 
-			bRet = 0;
-		newAct[i] = bRet;
-		DBGPRINTF("batch: item %d PRIFILT %d\n", i, newAct[i]);
-	}
-
-	if(stmt->d.s_prifilt.t_then != NULL) {
-		scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct, pWti);
-	}
-	if(stmt->d.s_prifilt.t_else != NULL) {
-		for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
-			if(*(pBatch->pbShutdownImmediate))
-				return;
-			if(pBatch->eltState[i] != BATCH_STATE_DISC
-			   && (active == NULL || active[i]))
-				newAct[i] = !newAct[i];
-			}
-		scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct, pWti);
+	if( (stmt->d.s_prifilt.pmask[pMsg->iFacility] == TABLE_NOPRI) ||
+	   ((stmt->d.s_prifilt.pmask[pMsg->iFacility]
+		    & (1<<pMsg->iSeverity)) == 0) )
+		bRet = 0;
+	else
+		bRet = 1;
+
+	DBGPRINTF("PRIFILT condition result is %d\n", bRet);
+	if(bRet) {
+		if(stmt->d.s_prifilt.t_then != NULL)
+			scriptExec(stmt->d.s_prifilt.t_then, pMsg, pWti);
+	} else {
+		if(stmt->d.s_prifilt.t_else != NULL)
+			scriptExec(stmt->d.s_prifilt.t_else, pMsg, pWti);
 	}
-	freeActive(newAct);
 }
 
 
@@ -498,79 +339,63 @@ done:
 	return bRet;
 }
 
-/* for details, see scriptExec() header comment! */
 static void
-execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
+execPROPFILT(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti)
 {
-	sbool *thenAct;
 	sbool bRet;
-	int i;
-	thenAct = newActive(pBatch);
-	for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
-		if(*(pBatch->pbShutdownImmediate))
-			return;
-		if(pBatch->eltState[i] == BATCH_STATE_DISC)
-			continue; /* will be ignored in any case */
-		if(active == NULL || active[i]) {
-			bRet = evalPROPFILT(stmt, pBatch->pElem[i].pMsg);
-		} else 
-			bRet = 0;
-		thenAct[i] = bRet;
-		DBGPRINTF("batch: item %d PROPFILT %d\n", i, thenAct[i]);
-	}
 
-	scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct, pWti);
-	freeActive(thenAct);
+	bRet = evalPROPFILT(stmt, pMsg);
+	DBGPRINTF("PROPFILT condition result is %d\n", bRet);
+	if(bRet)
+		scriptExec(stmt->d.s_propfilt.t_then, pMsg, pWti);
 }
 
 /* The rainerscript execution engine. It is debatable if that would be better
  * contained in grammer/rainerscript.c, HOWEVER, that file focusses primarily
  * on the parsing and object creation part. So as an actual executor, it is
  * better suited here.
- * param active: if NULL, all messages are active (to be processed), if non-null
- *               this is an array of the same size as the batch. If 1, the message
- *               is to be processed, otherwise not.
- * NOTE: this function must receive batches which contain a single ruleset ONLY!
  * rgerhards, 2012-09-04
  */
-static rsRetVal
-scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti)
+static void
+scriptExec(struct cnfstmt *root, msg_t *pMsg, wti_t *pWti)
 {
-	DEFiRet;
 	struct cnfstmt *stmt;
 
 	for(stmt = root ; stmt != NULL ; stmt = stmt->next) {
+		if(*pWti->pbShutdownImmediate) {
+			DBGPRINTF("scriptExec: ShutdownImmediate set, "
+				  "force terminating\n");	
+			goto done;
+		}
 		if(Debug) {
-			dbgprintf("scriptExec: batch of %d elements, active %p, active[0]:%d\n",
-				  batchNumMsgs(pBatch), active, (active == NULL ? 1 : active[0]));
 			cnfstmtPrintOnly(stmt, 2, 0);
 		}
 		switch(stmt->nodetype) {
 		case S_NOP:
 			break;
 		case S_STOP:
-			execStop(pBatch, active);
+			goto done;
 			break;
 		case S_ACT:
-			execAct(stmt, pBatch, active, pWti);
+			execAct(stmt, pMsg, pWti);
 			break;
 		case S_SET:
-			execSet(stmt, pBatch, active);
+			execSet(stmt, pMsg);
 			break;
 		case S_UNSET:
-			execUnset(stmt, pBatch, active);
+			execUnset(stmt, pMsg);
 			break;
 		case S_CALL:
-			execCall(stmt, pBatch, active, pWti);
+			execCall(stmt, pMsg, pWti);
 			break;
 		case S_IF:
-			execIf(stmt, pBatch, active, pWti);
+			execIf(stmt, pMsg, pWti);
 			break;
 		case S_PRIFILT:
-			execPRIFILT(stmt, pBatch, active, pWti);
+			execPRIFILT(stmt, pMsg, pWti);
 			break;
 		case S_PROPFILT:
-			execPROPFILT(stmt, pBatch, active, pWti);
+			execPROPFILT(stmt, pMsg, pWti);
 			break;
 		default:
 			dbgprintf("error: unknown stmt type %u during exec\n",
@@ -578,36 +403,42 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti)
 			break;
 		}
 	}
-	RETiRet;
+done:	return;
 }
 
 
 /* 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
+ * This is called by MAIN queues.
  */
 static rsRetVal
 processBatch(batch_t *pBatch, wti_t *pWti)
 {
-	ruleset_t *pThis;
+	int i;
+	msg_t *pMsg;
+	ruleset_t *pRuleset;
 	DEFiRet;
-	assert(pBatch != NULL);
-
-	DBGPRINTF("processBatch: batch of %d elements must be processed\n", pBatch->nElem);
-	if(pBatch->bSingleRuleset) {
-		pThis = batchGetRuleset(pBatch);
-		if(pThis == NULL)
-			pThis = ourConf->rulesets.pDflt;
-		ISOBJ_TYPE_assert(pThis, ruleset);
-		CHKiRet(scriptExec(pThis->root, pBatch, NULL, pWti));
-	} else {
-		CHKiRet(processBatchMultiRuleset(pBatch, pWti));
+
+	DBGPRINTF("processBATCH: batch of %d elements must be processed\n", pBatch->nElem);
+
+	wtiResetExecState(pWti, pBatch);
+
+	/* execution phase */
+	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pWti->pbShutdownImmediate) ; ++i) {
+		pMsg = pBatch->pElem[i].pMsg;
+		DBGPRINTF("processBATCH: next msg %d: %.128s\n", i, pMsg->pszRawMsg);
+		pRuleset = (pMsg->pRuleset == NULL) ? ourConf->rulesets.pDflt : pMsg->pRuleset;
+		scriptExec(pRuleset->root, pMsg, pWti);
+		// TODO: think if we need a return state of scriptExec - most probably
+		// the answer is "no", as we need to process the batch in any case!
+		// TODO: we must refactor this!  flag messages as committed
+		batchSetElemState(pBatch, i, BATCH_STATE_COMM);
 	}
 
-finalize_it:
-	DBGPRINTF("ruleset.ProcessMsg() returns %d\n", iRet);
+	/* commit phase */
+	dbgprintf("END batch execution phase, entering to commit phase\n");
+	actionCommitAllDirect(pWti);
+
+	DBGPRINTF("processBATCH: batch of %d elements has been processed\n", pBatch->nElem);
 	RETiRet;
 }