From 802f6d8a8f39e5ba578e0183e4500bef8e3a198c Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 15 Jun 2010 14:02:34 +0200 Subject: milestone(BUGGY): batch now pushed down to action at least in important cases (not for non-direct action queues and some other minor things). This version is definitely buggy, but may be tried with success on a non-production system. I will continue to work on the correctness, but needed to commit now to get a baseline. --- action.c | 186 +++++++++++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 140 insertions(+), 46 deletions(-) (limited to 'action.c') diff --git a/action.c b/action.c index dc6cdaaf..c8117b78 100644 --- a/action.c +++ b/action.c @@ -585,7 +585,7 @@ finalize_it: * depending on its current state. * rgerhards, 2009-05-07 */ -static rsRetVal actionPrepare(action_t *pThis) +static inline rsRetVal actionPrepare(action_t *pThis) { DEFiRet; @@ -665,7 +665,6 @@ static rsRetVal prepareDoActionParams(action_t *pAction, msg_t *pMsg, uchar **pp for(i = 0 ; i < pAction->iNumTpls ; ++i) { switch(pAction->eParamPassing) { case ACT_STRING_PASSING: - //CHKiRet(tplToString(pAction->ppTpl[i], pMsg, &(((uchar**)pAction->ppMsgs)[i]), &(pAction->lenMsgs[i]))); CHKiRet(tplToString(pAction->ppTpl[i], pMsg, &(ppMsgs[i]), &lenMsgs[i])); break; case ACT_ARRAY_PASSING: @@ -726,29 +725,23 @@ static rsRetVal cleanupDoActionParams(action_t *pAction, uchar **ppMsgs) * rgerhards, 2008-01-28 */ rsRetVal -actionCallDoAction(action_t *pThis, msg_t *pMsg) +actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams) { - uchar *ppMsgs[10]; - size_t lenMsgs[10]; int i; DEFiRet; ASSERT(pThis != NULL); ISOBJ_TYPE_assert(pMsg, msg); - for(i = 0 ; i < 10 ; ++ i) { - ppMsgs[i] = NULL; - lenMsgs[i] = 0; - } - DBGPRINTF("entering actionCalldoAction(), state: %s\n", getActStateName(pThis)); - CHKiRet(prepareDoActionParams(pThis, pMsg, ppMsgs, lenMsgs)); + //CHKiRet(prepareDoActionParams(pThis, pMsg, ppMsgs, lenMsgs)); pThis->bHadAutoCommit = 0; #if 1 //d_pthread_mutex_lock(&pThis->mutActExec); //pthread_cleanup_push(mutexCancelCleanup, &pThis->mutActExec); - iRet = pThis->pMod->mod.om.doAction(ppMsgs, pMsg->msgFlags, pThis->pModData); + // original: iRet = pThis->pMod->mod.om.doAction(ppMsgs, pMsg->msgFlags, pThis->pModData); + iRet = pThis->pMod->mod.om.doAction(actParams, pMsg->msgFlags, pThis->pModData); //pthread_cleanup_pop(1); /* unlock mutex */ //iRet = pThis->pMod->mod.om.doAction(pThis->ppMsgs, pMsg->msgFlags, pThis->pModData); #else @@ -783,6 +776,7 @@ iRet = RS_RET_OK; finalize_it: +#if 0 // THIS NEEDS TO BE DONE TO THE BATCH! switch(pThis->eParamPassing) { case ACT_STRING_PASSING: for(i = 0 ; i < 10 ; ++i) @@ -795,6 +789,7 @@ finalize_it: /* nothing to do in that case */ break; } +#endif RETiRet; @@ -805,8 +800,8 @@ finalize_it: * this readies the action and then calls doAction() * rgerhards, 2008-01-28 */ -rsRetVal -actionProcessMessage(action_t *pThis, msg_t *pMsg) +static inline rsRetVal +actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams) { DEFiRet; @@ -815,7 +810,7 @@ actionProcessMessage(action_t *pThis, msg_t *pMsg) CHKiRet(actionPrepare(pThis)); if(pThis->eState == ACT_STATE_ITX) - CHKiRet(actionCallDoAction(pThis, pMsg)); + CHKiRet(actionCallDoAction(pThis, pMsg, actParams)); iRet = getReturnCode(pThis); finalize_it: @@ -835,8 +830,11 @@ finishBatch(action_t *pThis, batch_t *pBatch) ASSERT(pThis != NULL); - if(pThis->eState == ACT_STATE_RDY) +dbgprintf("ZZZ: finishBatch called, eState %d\n", pThis->eState); + if(pThis->eState == ACT_STATE_RDY) { + /* we just need to flag the batch as commited */ FINALIZE; /* nothing to do */ + } CHKiRet(actionPrepare(pThis)); if(pThis->eState == ACT_STATE_ITX) { @@ -846,7 +844,8 @@ finishBatch(action_t *pThis, batch_t *pBatch) actionCommitted(pThis); /* flag messages as committed */ for(i = 0 ; i < pBatch->nElem ; ++i) { - pBatch->pElem[i].state = BATCH_STATE_COMM; + batchSetElemState(pBatch, i, BATCH_STATE_COMM); +dbgprintf("ZZZ: finishBatch commits element %d\n", i); } break; case RS_RET_SUSPENDED: @@ -881,8 +880,8 @@ finalize_it: /* try to submit a partial batch of elements. * rgerhards, 2009-05-12 */ -static rsRetVal -tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem, int *pbShutdownImmediate) +static inline rsRetVal +tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) { int i; int iElemProcessed; @@ -894,17 +893,22 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem, int *pbShutdownImme assert(pBatch != NULL); assert(pnElem != NULL); +dbgprintf("ZZZ1: tryDoAction, nElem %d, iDoneUpto %d\n", *pnElem, pBatch->iDoneUpTo); i = pBatch->iDoneUpTo; /* all messages below that index are processed */ iElemProcessed = 0; iCommittedUpTo = i; - pAction->pbShutdownImmediate = pbShutdownImmediate; while(iElemProcessed <= *pnElem && i < pBatch->nElem) { - if(*pbShutdownImmediate) + if(*(pBatch->pbShutdownImmediate)) ABORT_FINALIZE(RS_RET_FORCE_TERM); - pMsg = (msg_t*) pBatch->pElem[i].pUsrp; - if(pBatch->pElem[i].state != BATCH_STATE_DISC) { - localRet = actionProcessMessage(pAction, pMsg); +dbgprintf("ZZZ1: tryDoAction loop %d: filter %d, state %d\n", i, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state); + if(pBatch->pElem[i].bFilterOK && pBatch->pElem[i].state != BATCH_STATE_DISC) { +dbgprintf("ZZZ1: trying to execute\n"); + pMsg = (msg_t*) pBatch->pElem[i].pUsrp; + localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams); DBGPRINTF("action call returned %d\n", localRet); + /* Note: we directly modify the batch object state, because we know that + * wo do not overwrite DISC indicators! + */ if(localRet == RS_RET_OK) { /* mark messages as committed */ while(iCommittedUpTo <= i) { @@ -931,9 +935,7 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem, int *pbShutdownImme } finalize_it: - if(pBatch->nElem == 1 && pBatch->pElem[0].state == BATCH_STATE_DISC) { - iRet = RS_RET_DISCARDMSG; - } else if(pBatch->iDoneUpTo != iCommittedUpTo) { + if(pBatch->iDoneUpTo != iCommittedUpTo) { *pnElem += iCommittedUpTo - pBatch->iDoneUpTo; pBatch->iDoneUpTo = iCommittedUpTo; } @@ -944,21 +946,24 @@ finalize_it: /* submit a batch for actual action processing. * The first nElem elements are processed. This function calls itself * recursively if it needs to handle errors. + * Note: we don't need the number of the first message to be processed as a parameter, + * because this is kept track of inside the batch itself (iDoneUpTo). * rgerhards, 2009-05-12 */ static rsRetVal -submitBatch(action_t *pAction, batch_t *pBatch, int nElem, int *pbShutdownImmediate) +submitBatch(action_t *pAction, batch_t *pBatch, int nElem) { int i; int bDone; rsRetVal localRet; DEFiRet; +dbgprintf("ZZZ1: submitBatch, nElem %d\n", nElem); assert(pBatch != NULL); bDone = 0; do { - localRet = tryDoAction(pAction, pBatch, &nElem, pbShutdownImmediate); + localRet = tryDoAction(pAction, pBatch, &nElem); if(localRet == RS_RET_FORCE_TERM) { ABORT_FINALIZE(RS_RET_FORCE_TERM); } @@ -968,38 +973,36 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem, int *pbShutdownImmedi /* try commit transaction, once done, we can simply do so as if * that return state was returned from tryDoAction(). */ - localRet = finishBatch(pAction, pBatch); // TODO: careful, do we need the elem counter? + localRet = finishBatch(pAction, pBatch); } if( localRet == RS_RET_OK || localRet == RS_RET_PREVIOUS_COMMITTED || localRet == RS_RET_DEFER_COMMIT) { bDone = 1; - } else if(localRet == RS_RET_DISCARDMSG) { - iRet = RS_RET_DISCARDMSG; /* TODO: verify this sequence -- rgerhards, 2009-07-30 */ - bDone = 1; } else if(localRet == RS_RET_SUSPENDED) { ; /* do nothing, this will retry the full batch */ } else if(localRet == RS_RET_ACTION_FAILED) { /* in this case, the whole batch can not be processed */ for(i = 0 ; i < nElem ; ++i) { - pBatch->pElem[pBatch->iDoneUpTo++].state = BATCH_STATE_BAD; +dbgprintf("ZZZ2: setting batch state for item %d\n", i); + batchSetElemState(pBatch, i, BATCH_STATE_BAD); } bDone = 1; } else { if(nElem == 1) { - pBatch->pElem[pBatch->iDoneUpTo++].state = BATCH_STATE_BAD; + batchSetElemState(pBatch, i, BATCH_STATE_BAD); bDone = 1; } else { /* retry with half as much. Depth is log_2 batchsize, so recursion is not too deep */ - submitBatch(pAction, pBatch, nElem / 2, pbShutdownImmediate); - submitBatch(pAction, pBatch, nElem - (nElem / 2), pbShutdownImmediate); + submitBatch(pAction, pBatch, nElem / 2); + submitBatch(pAction, pBatch, nElem - (nElem / 2)); bDone = 1; } } - } while(!bDone && !*pbShutdownImmediate); /* do .. while()! */ + } while(!bDone && !*(pBatch->pbShutdownImmediate)); /* do .. while()! */ - if(*pbShutdownImmediate) + if(*(pBatch->pbShutdownImmediate)) ABORT_FINALIZE(RS_RET_FORCE_TERM); finalize_it: @@ -1007,17 +1010,46 @@ finalize_it: } + +/* The following function prepares a batch for processing, that it is + * reinitializes batch states, generates strings and does everything else + * that needs to be done in order to make the batch ready for submission to + * the actual output module. Note that we look at the precomputed + * filter OK condition and process only those messages, that actually matched + * the filter. + * rgerhards, 2010-06-14 + */ +static inline rsRetVal +prepareBatch(action_t *pAction, batch_t *pBatch) +{ + int i; + batch_obj_t *pElem; + DEFiRet; + + pBatch->iDoneUpTo = 0; + for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { + pElem = &(pBatch->pElem[i]); + if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) { + pElem->state = BATCH_STATE_RDY; + prepareDoActionParams(pAction, (msg_t*) pElem->pUsrp, + (uchar**) &(pElem->staticActParams), pElem->staticLenParams); + } + } + RETiRet; +} + + /* receive a batch and process it. This includes retry handling. * rgerhards, 2009-05-12 */ -static rsRetVal -processAction(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) +static inline rsRetVal +processAction(action_t *pAction, batch_t *pBatch) { DEFiRet; assert(pBatch != NULL); - pBatch->iDoneUpTo = 0; - CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem, pbShutdownImmediate)); +dbgprintf("ZZZ1: processAction\n"); + CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem)); iRet = finishBatch(pAction, pBatch); finalize_it: @@ -1033,10 +1065,18 @@ finalize_it: static rsRetVal processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) { + int *pbShutdownImmdtSave; DEFiRet; assert(pBatch != NULL); +dbgprintf("ZZZ1: processBatchMain\n"); + pbShutdownImmdtSave = pBatch->pbShutdownImmediate; + pBatch->pbShutdownImmediate = pbShutdownImmediate; + pAction->pbShutdownImmediate = pBatch->pbShutdownImmediate; + CHKiRet(prepareBatch(pAction, pBatch)); + +dbgprintf("ZZZ1: processBatchMain\n"); /* 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 * if they notify us they are - functionality not yet implemented...). @@ -1045,10 +1085,19 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) d_pthread_mutex_lock(&pAction->mutActExec); pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec); - iRet = processAction(pAction, pBatch, pbShutdownImmediate); + iRet = processAction(pAction, pBatch); + // + // DEBUG + int i; + for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { +dbgprintf("ZZZ: after processBatchMain item %d: filter %d, state %d\n", i, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state); + } + // END DEBUG pthread_cleanup_pop(1); /* unlock mutex */ +finalize_it: + pBatch->pbShutdownImmediate = pbShutdownImmdtSave; RETiRet; } #pragma GCC diagnostic warning "-Wempty-body" @@ -1323,7 +1372,11 @@ doSubmitToActionQNotAllMark(action_t *pAction, msg_t *pMsg) } DBGPRINTF("Called action, logging to %s\n", module.GetStateName(pAction->pMod)); - iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg)); + if(pAction->pQueue->qType == QUEUETYPE_DIRECT) + iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg)); + else + iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg)); + //iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg)); finalize_it: RETiRet; @@ -1341,7 +1394,48 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg) DEFiRet; DBGPRINTF("Called action, logging to %s\n", module.GetStateName(pAction->pMod)); - iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg)); + if(pAction->pQueue->qType == QUEUETYPE_DIRECT) + iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg)); + else + iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg)); + + RETiRet; +} + + +//*** EXPERIMENTAL ***/ +/* This submits the message to the action queue in case we do NOT need to handle repeat + * message processing. That case permits us to gain lots of freedom during processing + * and thus speed. + * rgerhards, 2010-06-08 + */ +rsRetVal +doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) +{ + int i; + DEFiRet; + + DBGPRINTF("Called action(Batch), logging to %s\n", module.GetStateName(pAction->pMod)); + if(pAction->pQueue->qType == QUEUETYPE_DIRECT) + iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch); + else { /* in this case, we do single submits to the queue. + * TODO: optimize this, we may do at least a multi-submit! + */ + for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { + if(pBatch->pElem[i].bFilterOK) { +dbgprintf("ZZZ: submitToActQ item %d:%s\n", i, ((msg_t*)(pBatch->pElem[i].pUsrp))->szRawMsg+15); + pAction->submitToActQ(pAction, (msg_t*)(pBatch->pElem[i].pUsrp)); + } + } + + // DEBUG + for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { + if(pBatch->pElem[i].bFilterOK) { +dbgprintf("ZZZ: batch state after processing item %d: %d\n", i, pBatch->pElem[i].state); + } + } + // END DEBUG + } RETiRet; } -- cgit