diff options
| -rw-r--r-- | action.c | 67 | ||||
| -rw-r--r-- | runtime/batch.h | 1 | ||||
| -rw-r--r-- | runtime/ruleset.c | 16 |
3 files changed, 23 insertions, 61 deletions
@@ -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); } } diff --git a/runtime/batch.h b/runtime/batch.h index 4120cc51..75eecdb1 100644 --- a/runtime/batch.h +++ b/runtime/batch.h @@ -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 */ diff --git a/runtime/ruleset.c b/runtime/ruleset.c index db253d28..6cc98105 100644 --- a/runtime/ruleset.c +++ b/runtime/ruleset.c @@ -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; } |