rsyslog/rsyslog
1
0
Fork 0
mirror of https://github.com/rsyslog/rsyslog.git synced 2025-12-16 13:20:41 +01:00
Code Issues Packages Projects Releases Wiki Activity

temporarily remove bExecWhenPrevWasSuspended handling

Browse Source 
also add comments on howto re-implement it inside the script engine
(but we do not do this right now as we would like to do this together
when we touch the script engine -- for now focussing on action
handling).
This commit is contained in:
Rainer Gerhards 2013-11-02 17:43:24 +01:00
parent 8878a0b1cf
commit d200ef2d8b
3 changed files with 23 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
action.c
View File

@ -36,9 +36,6 @@
*
* After dequeue, processing is as follows:
* - processBatchMain
* - processAction
* - submitBatch
* - tryDoAction
* - ...
*
* MORE ON PROCESSING, QUEUES and FILTERING
@ -1275,13 +1272,6 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch, wti_t *pWti)
iRet = actionWriteToAction(pAction, pMsg, pWti);
finalize_it:
/* we need to update the batch to handle failover processing correctly */
if(iRet == RS_RET_OK) {
pBatch->pElem[idxBtch].bPrevWasSuspended = 0;
} else if(iRet == RS_RET_ACTION_FAILED) {
pBatch->pElem[idxBtch].bPrevWasSuspended = 1;
}
RETiRet;
}
@ -1398,48 +1388,10 @@ countStatsBatchEnq(action_t *pAction, batch_t *pBatch)
static inline rsRetVal
doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
{
sbool bNeedSubmit;
sbool *activeSave;
int i;
DEFiRet;
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
* to be processed.
*/
if(pAction->bExecWhenPrevSusp) {
bNeedSubmit = 0;
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
if(!pBatch->pElem[i].bPrevWasSuspended) {
DBGPRINTF("action enq stage: change active to 0 due to "
"failover case in elem %d\n", i);
pBatch->active[i] = 0;
}
if(batchIsValidElem(pBatch, i)) {
STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
bNeedSubmit = 1;
}
DBGPRINTF("action %p[%d]: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
pAction, i, batchIsValidElem(pBatch, i), pBatch->eltState[i],
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
}
if(bNeedSubmit) {
/* note: stats were already computed above */
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti);
} else {
DBGPRINTF("no need to submit batch, all invalid\n");
}
} else {
if(GatherStats)
countStatsBatchEnq(pAction, pBatch);
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti);
}
free(pBatch->active);
pBatch->active = activeSave;
if(GatherStats)
countStatsBatchEnq(pAction, pBatch);
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti);
RETiRet;
}
@ -1462,11 +1414,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
* TODO: optimize this, we may do at least a multi-submit!
*/
for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
DBGPRINTF("action %p: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
pAction, batchIsValidElem(pBatch, i), pBatch->eltState[i],
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
if( batchIsValidElem(pBatch, i)
&& (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) {
if(batchIsValidElem(pBatch, i)) {
doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg, pWti);
}
}
@ -1490,11 +1438,10 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWt
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: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
DBGPRINTF("action %p: valid:%d state:%d execWhenPrev:%d\n",
pAction, batchIsValidElem(pBatch, i), pBatch->eltState[i],
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
if( batchIsValidElem(pBatch, i)
&& ((pAction->bExecWhenPrevSusp == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
pAction->bExecWhenPrevSusp);
if(batchIsValidElem(pBatch, i)) {
doActionCallAction(pAction, pBatch, i, pWti);
}
}

1
runtime/batch.h
View File

@ -49,7 +49,6 @@ struct batch_obj_s {
/* work variables for action processing; these are reused for each action (or block of
* actions)
*/
sbool bPrevWasSuspended;
/* following are caches to save allocs if not absolutely necessary */
uchar *staticActStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /**< for strings */
/* a cache to save malloc(), if not absolutely necessary */

16
runtime/ruleset.c
View File

@ -231,7 +231,23 @@ 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;
// TODO: check here if bPrevWasSuspsended was required and, if so
// if we actually are permitted to execute this action.
//if(pAction->bExecWhenPrevSusp) {
stmt->d.act->submitToActQ(stmt->d.act, pBatch, pWti);
#warning implement action return code checking
// we should store the return code and make it available
// to users via a special function (or maybe variable)
// internally, we can use this for bPrevWasSuspended checking
// to implement this system, we need to keep a kind of
// "execution state" when running the rule engine. This most
// probably is best done inside the wti object.
// I think in v7 there was a bug, so that bPrevWasSuspended did
// not properly make it onto the next batch (because it was
// stored within the batch state) -- but even if so, the
// exposure window was minimal, as the action would probably
// fail the next time again. [TODO: check if batch object survived
// end of batch, in which case it was probably correctly handled]
RETiRet;
}