summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--action.c141
-rw-r--r--runtime/batch.h5
-rw-r--r--runtime/queue.c4
-rwxr-xr-xtests/diag.sh4
4 files changed, 76 insertions, 78 deletions
diff --git a/action.c b/action.c
index 8f8371f7..bc96ea9d 100644
--- a/action.c
+++ b/action.c
@@ -861,7 +861,7 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch)
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
pElem = &(pBatch->pElem[i]);
- if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) {
+ if(batchIsValidElem(pBatch, i)) {
switch(pAction->eParamPassing) {
case ACT_ARRAY_PASSING:
ppMsgs = (uchar***) pElem->staticActParams;
@@ -1059,8 +1059,7 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem)
/* NOTE: do NOT extend the filter below! Anything else must be done on the
* enq side of the queue (see file header comment)! -- rgerhards, 2011-06-15
*/
- if( pBatch->pElem[i].bFilterOK
- && pBatch->pElem[i].state != BATCH_STATE_DISC) {
+ if(batchIsValidElem(pBatch, i)) {
pMsg = (msg_t*) pBatch->pElem[i].pUsrp;
localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams,
pBatch->pbShutdownImmediate);
@@ -1181,6 +1180,29 @@ finalize_it:
}
+/* copy "active" array of batch, as we need to modify it. The caller
+ * must make sure the new array is freed and the orginal batch
+ * pointer is restored (thus the caller must save it). If active
+ * is currently NULL, this is properly handled.
+ * Note: the batches active pointer is modified, so it must be
+ * saved BEFORE calling this function!
+ * rgerhards, 2012-09-12
+ */
+static rsRetVal
+copyActive(batch_t *pBatch)
+{
+ sbool *active;
+ DEFiRet;
+
+ CHKmalloc(active = malloc(batchNumMsgs(pBatch) * sizeof(sbool)));
+ if(pBatch->active == NULL)
+ memset(active, 1, batchNumMsgs(pBatch));
+ else
+ memcpy(active, pBatch->active, batchNumMsgs(pBatch));
+ pBatch->active = active;
+finalize_it:
+ RETiRet;
+}
/* The following function prepares a batch for processing, that it is
* reinitializes batch states, generates strings and does everything else
@@ -1191,7 +1213,7 @@ finalize_it:
* rgerhards, 2010-06-14
*/
static inline rsRetVal
-prepareBatch(action_t *pAction, batch_t *pBatch)
+prepareBatch(action_t *pAction, batch_t *pBatch, sbool **activeSave, int *bMustRestoreActivePtr)
{
int i;
batch_obj_t *pElem;
@@ -1200,10 +1222,16 @@ prepareBatch(action_t *pAction, batch_t *pBatch)
pBatch->iDoneUpTo = 0;
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
pElem = &(pBatch->pElem[i]);
- if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) {
+ if(batchIsValidElem(pBatch, i)) {
pElem->state = BATCH_STATE_RDY;
- if(prepareDoActionParams(pAction, pElem) != RS_RET_OK)
- pElem->bFilterOK = RSFALSE;
+ if(prepareDoActionParams(pAction, pElem) != RS_RET_OK) {
+ /* make sure we have our copy of "active" array */
+ if(!*bMustRestoreActivePtr) {
+ *activeSave = pBatch->active;
+ copyActive(pBatch);
+ }
+ pBatch->active[i] = RSFALSE;
+ }
}
}
RETiRet;
@@ -1236,6 +1264,8 @@ static rsRetVal
processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
{
int *pbShutdownImmdtSave;
+ sbool *activeSave;
+ int bMustRestoreActivePtr = 0;
rsRetVal localRet;
DEFiRet;
@@ -1243,7 +1273,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
pbShutdownImmdtSave = pBatch->pbShutdownImmediate;
pBatch->pbShutdownImmediate = pbShutdownImmediate;
- CHKiRet(prepareBatch(pAction, pBatch));
+ CHKiRet(prepareBatch(pAction, pBatch, &activeSave, &bMustRestoreActivePtr));
/* We now must guard the output module against execution by multiple threads. The
* plugin interface specifies that output modules must not be thread-safe (except
@@ -1266,6 +1296,11 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
if(iRet == RS_RET_OK)
iRet = localRet;
+
+ if(bMustRestoreActivePtr) {
+ free(pBatch->active);
+ pBatch->active = activeSave;
+ }
finalize_it:
pBatch->pbShutdownImmediate = pbShutdownImmdtSave;
@@ -1591,22 +1626,15 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch)
time_t now = 0;
time_t lastAct;
int i;
- int bModifiedFilter;
- sbool FilterSave[1024];
- sbool *pFilterSave;
+ sbool *activeSave;
DEFiRet;
- if(batchNumMsgs(pBatch) <= (int) (sizeof(FilterSave)/sizeof(sbool))) {
- pFilterSave = FilterSave;
- } else {
- CHKmalloc(pFilterSave = malloc(batchNumMsgs(pBatch) * sizeof(sbool)));
- }
+ activeSave = pBatch->active;
+ copyActive(pBatch);
- bModifiedFilter = 0;
for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
- if(!pBatch->pElem[i].bFilterOK)
+ if(!pBatch->active[i])
continue;
- pFilterSave[i] = pBatch->pElem[i].bFilterOK;
if(now == 0) {
now = datetime.GetTime(NULL); /* good time call - the only one done */
}
@@ -1617,15 +1645,15 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch)
lastAct = pAction->f_time;
if(((msg_t*)(pBatch->pElem[i].pUsrp))->msgFlags & MARK) {
if((now - lastAct) < MarkInterval / 2) {
- pBatch->pElem[i].bFilterOK = 0;
- bModifiedFilter = 1;
- DBGPRINTF("action was recently called, ignoring mark message\n");
+ pBatch->active[i] = 0;
+ DBGPRINTF("batch item %d: action was recently called, ignoring "
+ "mark message\n", i);
break; /* do not update timestamp for non-written mark messages */
}
}
} while(ATOMIC_CAS_time_t(&pAction->f_time, lastAct,
((msg_t*)(pBatch->pElem[i].pUsrp))->ttGenTime, &pAction->mutCAS) == 0);
- if(pBatch->pElem[i].bFilterOK) {
+ if(pBatch->active[i]) {
DBGPRINTF("Called action(NotAllMark), processing batch[%d] via '%s'\n",
i, module.GetStateName(pAction->pMod));
}
@@ -1633,17 +1661,8 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch)
iRet = doSubmitToActionQBatch(pAction, pBatch);
- if(bModifiedFilter) {
- /* in this case, we need to restore previous state */
- for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
- /* note: clang static code analyzer reports a false positive below */
- pBatch->pElem[i].bFilterOK = pFilterSave[i];
- }
- }
-
-finalize_it:
- if(pFilterSave != FilterSave)
- free(pFilterSave);
+ free(pBatch->active);
+ pBatch->active = activeSave;
RETiRet;
}
@@ -1653,8 +1672,7 @@ countStatsBatchEnq(action_t *pAction, batch_t *pBatch)
{
int i;
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
- if( pBatch->pElem[i].bFilterOK
- && pBatch->pElem[i].state != BATCH_STATE_DISC) {
+ if( batchIsValidElem(pBatch, i)) {
STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
}
}
@@ -1668,18 +1686,13 @@ countStatsBatchEnq(action_t *pAction, batch_t *pBatch)
static inline rsRetVal
doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
{
- sbool FilterSave[1024];
- sbool *pFilterSave;
sbool bNeedSubmit;
- sbool bModifiedFilter;
+ sbool *activeSave;
int i;
DEFiRet;
- if(batchNumMsgs(pBatch) <= (int) (sizeof(FilterSave)/sizeof(sbool))) {
- pFilterSave = FilterSave;
- } else {
- CHKmalloc(pFilterSave = malloc(batchNumMsgs(pBatch) * sizeof(sbool)));
- }
+ activeSave = pBatch->active;
+ copyActive(pBatch);
/* note: for direct mode, we need to adjust the filter property. For non-direct
* this is not necessary, because in that case we enqueue only what actually needs
@@ -1687,45 +1700,31 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
*/
if(pAction->bExecWhenPrevSusp) {
bNeedSubmit = 0;
- bModifiedFilter = 0;
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
- pFilterSave[i] = pBatch->pElem[i].bFilterOK;
if(!pBatch->pElem[i].bPrevWasSuspended) {
- DBGPRINTF("action enq stage: change bFilterOK to 0 due to "
+ DBGPRINTF("action enq stage: change active to 0 due to "
"failover case in elem %d\n", i);
- pBatch->pElem[i].bFilterOK = 0;
- bModifiedFilter = 1;
+ pBatch->active[i] = 0;
}
- if(pBatch->pElem[i].bFilterOK && pBatch->pElem[i].state != BATCH_STATE_DISC) {
+ if(batchIsValidElem(pBatch, i)) {
STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
bNeedSubmit = 1;
}
- DBGPRINTF("action %p[%d]: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
- pAction, i, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state,
+ DBGPRINTF("action %p[%d]: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+ pAction, i, batchIsValidElem(pBatch, i), pBatch->pElem[i].state,
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
}
if(bNeedSubmit) {
/* note: stats were already computed above */
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
} else {
- DBGPRINTF("no need to submit batch, all bFilterOK==0 or discarded\n");
- }
- if(bModifiedFilter) {
- for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
- DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
- pAction, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state,
- pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
- /* note: clang static code analyzer reports a false positive below */
- pBatch->pElem[i].bFilterOK = pFilterSave[i];
- }
+ DBGPRINTF("no need to submit batch, all invalid\n");
}
} else {
if(GatherStats)
countStatsBatchEnq(pAction, pBatch);
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
}
-
-finalize_it:
RETiRet;
}
@@ -1748,11 +1747,10 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch)
* TODO: optimize this, we may do at least a multi-submit!
*/
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
- DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
- pAction, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state,
+ DBGPRINTF("action %p: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+ pAction, batchIsValidElem(pBatch, i), pBatch->pElem[i].state,
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
- if( pBatch->pElem[i].bFilterOK
- && pBatch->pElem[i].state != BATCH_STATE_DISC
+ if( batchIsValidElem(pBatch, i)
&& (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) {
doSubmitToActionQ(pAction, (msg_t*)(pBatch->pElem[i].pUsrp));
}
@@ -1777,11 +1775,10 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
DBGPRINTF("Called action %p (complex case), logging to %s\n",
pAction, module.GetStateName(pAction->pMod));
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
- DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
- pAction, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state,
+ DBGPRINTF("action %p: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+ pAction, batchIsValidElem(pBatch, i), pBatch->pElem[i].state,
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
- if( pBatch->pElem[i].bFilterOK
- && pBatch->pElem[i].state != BATCH_STATE_DISC
+ if( batchIsValidElem(pBatch, i)
&& ((pAction->bExecWhenPrevSusp == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
doActionCallAction(pAction, pBatch, i);
}
diff --git a/runtime/batch.h b/runtime/batch.h
index 944889bd..fdacb8e2 100644
--- a/runtime/batch.h
+++ b/runtime/batch.h
@@ -51,7 +51,6 @@ struct batch_obj_s {
/* work variables for action processing; these are reused for each action (or block of
* actions)
*/
- sbool bFilterOK; /* work area for filter processing (per action, reused!) */
sbool bPrevWasSuspended;
/* following are caches to save allocs if not absolutely necessary */
uchar *staticActStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /**< for strings */
@@ -83,6 +82,7 @@ struct batch_s {
int iDoneUpTo; /* all messages below this index have state other than RDY */
qDeqID deqID; /* ID of dequeue operation that generated this batch */
int *pbShutdownImmediate;/* end processing of this batch immediately if set to 1 */
+ sbool *active; /* which messages are active for processing, NULL=all */
sbool bSingleRuleset; /* do all msgs of this batch use a single ruleset? */
batch_obj_t *pElem; /* batch elements */
};
@@ -129,7 +129,8 @@ batchSetElemState(batch_t *pBatch, int i, batch_state_t newState) {
*/
static inline int
batchIsValidElem(batch_t *pBatch, int i) {
- return(pBatch->pElem[i].bFilterOK && pBatch->pElem[i].state != BATCH_STATE_DISC);
+ return( (pBatch->active == NULL || pBatch->active[i])
+ && pBatch->pElem[i].state != BATCH_STATE_DISC);
}
diff --git a/runtime/queue.c b/runtime/queue.c
index bb9ea060..2108e231 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -976,6 +976,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
{
batch_t singleBatch;
batch_obj_t batchObj;
+ sbool active = 1;
int i;
DEFiRet;
@@ -994,9 +995,9 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
memset(&singleBatch, 0, sizeof(batch_t));
batchObj.state = BATCH_STATE_RDY;
batchObj.pUsrp = (obj_t*) pUsr;
- batchObj.bFilterOK = 1;
singleBatch.nElem = 1; /* there always is only one in direct mode */
singleBatch.pElem = &batchObj;
+ singleBatch.active = &active;
iRet = pThis->pConsumer(pThis->pUsr, &singleBatch, &pThis->bShutdownImmediate);
/* delete the batch string params: TODO: create its own "class" for this */
for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
@@ -1596,7 +1597,6 @@ 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;
}
diff --git a/tests/diag.sh b/tests/diag.sh
index b278d2c5..ed0709da 100755
--- a/tests/diag.sh
+++ b/tests/diag.sh
@@ -10,8 +10,8 @@
#valgrind="valgrind --tool=helgrind --log-fd=1"
#valgrind="valgrind --tool=exp-ptrcheck --log-fd=1"
#set -o xtrace
-#export RSYSLOG_DEBUG="debug nologfuncflow noprintmutexaction nostdout"
-#export RSYSLOG_DEBUGLOG="log"
+export RSYSLOG_DEBUG="debug nologfuncflow noprintmutexaction nostdout"
+export RSYSLOG_DEBUGLOG="log"
case $1 in
'init') $srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason
cp $srcdir/testsuites/diag-common.conf diag-common.conf
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-26 23:07:07 +0000