summaryrefslogtreecommitdiffstats
path: root/runtime
diff options
context:
space:
mode:
Diffstat (limited to 'runtime')
-rw-r--r--runtime/batch.h40
-rw-r--r--runtime/conf.c6
-rw-r--r--runtime/queue.c87
-rw-r--r--runtime/queue.h9
-rw-r--r--runtime/rsyslog.h1
-rw-r--r--runtime/ruleset.c369
-rw-r--r--runtime/wti.c41
-rw-r--r--runtime/wti.h132
8 files changed, 311 insertions, 374 deletions
diff --git a/runtime/batch.h b/runtime/batch.h
index 2ec07670..b0aa8574 100644
--- a/runtime/batch.h
+++ b/runtime/batch.h
@@ -46,17 +46,6 @@ typedef unsigned char batch_state_t;
*/
struct batch_obj_s {
msg_t *pMsg;
- /* 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 */
- void *staticActParams[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /**< for anything else */
- size_t staticLenStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE];
- /* and the same for the message length (if used) */
- /* end action work variables */
};
/* the batch
@@ -77,11 +66,7 @@ struct batch_s {
int maxElem; /* maximum number of elements that this batch supports */
int nElem; /* actual number of element in this entry */
int nElemDeq; /* actual number of elements dequeued (and thus to be deleted) - see comment above! */
- 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 */
batch_state_t *eltState;/* state (array!) for individual objects.
NOTE: we have moved this out of batch_obj_t because we
@@ -93,24 +78,6 @@ struct batch_s {
};
-/* some inline functions (we may move this off to an object .. or not) */
-static inline void
-batchSetSingleRuleset(batch_t *pBatch, sbool val) {
- pBatch->bSingleRuleset = val;
-}
-
-/* get the batches ruleset (if we have a single ruleset) */
-static inline ruleset_t*
-batchGetRuleset(batch_t *pBatch) {
- return (pBatch->nElem > 0) ? pBatch->pElem[0].pMsg->pRuleset : NULL;
-}
-
-/* get the ruleset of a specifc element of the batch (index not verified!) */
-static inline ruleset_t*
-batchElemGetRuleset(batch_t *pBatch, int i) {
- return pBatch->pElem[i].pMsg->pRuleset;
-}
-
/* get number of msgs for this batch */
static inline int
batchNumMsgs(batch_t *pBatch) {
@@ -134,8 +101,7 @@ batchSetElemState(batch_t *pBatch, int i, batch_state_t newState) {
*/
static inline int
batchIsValidElem(batch_t *pBatch, int i) {
- return( (pBatch->eltState[i] != BATCH_STATE_DISC)
- && (pBatch->active == NULL || pBatch->active[i]));
+ return(pBatch->eltState[i] != BATCH_STATE_DISC);
}
@@ -152,7 +118,7 @@ batchFree(batch_t *pBatch) {
/* staticActParams MUST be freed immediately (if required),
* so we do not need to do that!
*/
- free(pBatch->pElem[i].staticActStrings[j]);
+ //TODO: do this in wti! free(pBatch->pElem[i].staticActStrings[j]);
}
}
free(pBatch->pElem);
@@ -167,11 +133,9 @@ batchFree(batch_t *pBatch) {
static inline rsRetVal
batchInit(batch_t *pBatch, int maxElem) {
DEFiRet;
- pBatch->iDoneUpTo = 0;
pBatch->maxElem = maxElem;
CHKmalloc(pBatch->pElem = calloc((size_t)maxElem, sizeof(batch_obj_t)));
CHKmalloc(pBatch->eltState = calloc((size_t)maxElem, sizeof(batch_state_t)));
- // TODO: replace calloc by inidividual writes?
finalize_it:
RETiRet;
}
diff --git a/runtime/conf.c b/runtime/conf.c
index 2b000c60..83931bca 100644
--- a/runtime/conf.c
+++ b/runtime/conf.c
@@ -518,12 +518,10 @@ rsRetVal cflineDoAction(rsconf_t *conf, uchar **p, action_t **ppAction)
bHadWarning = 1;
iRet = RS_RET_OK;
}
- if(iRet == RS_RET_OK || iRet == RS_RET_SUSPENDED) {
- if((iRet = addAction(&pAction, pMod, pModData, pOMSR, NULL, NULL,
- (iRet == RS_RET_SUSPENDED)? 1 : 0)) == RS_RET_OK) {
+ if(iRet == RS_RET_OK) {
+ if((iRet = addAction(&pAction, pMod, pModData, pOMSR, NULL, NULL)) == RS_RET_OK) {
/* here check if the module is compatible with select features
* (currently, we have no such features!) */
- pAction->eState = ACT_STATE_RDY; /* action is enabled */
conf->actions.nbrActions++; /* one more active action! */
}
break;
diff --git a/runtime/queue.c b/runtime/queue.c
index 8c610b11..8fb734e7 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -81,11 +81,10 @@ static rsRetVal GetDeqBatchSize(qqueue_t *pThis, int *pVal);
static rsRetVal ConsumerDA(qqueue_t *pThis, wti_t *pWti);
static rsRetVal batchProcessed(qqueue_t *pThis, wti_t *pWti);
static rsRetVal qqueueMultiEnqObjNonDirect(qqueue_t *pThis, multi_submit_t *pMultiSub);
-static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub, wti_t *pWti);
-static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti);
+static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub);
+static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg);
static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis);
static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis);
-static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis);
static rsRetVal qDestructDisk(qqueue_t *pThis);
/* some constants for queuePersist () */
@@ -705,9 +704,13 @@ queueSwitchToEmergencyMode(qqueue_t *pThis, rsRetVal initiatingError)
pThis->qType = QUEUETYPE_DIRECT;
pThis->qConstruct = qConstructDirect;
pThis->qDestruct = qDestructDirect;
+ /* these entry points shall not be used in direct mode
+ * To catch program errors, make us abort if that happens!
+ * rgerhards, 2013-11-05
+ */
pThis->qAdd = qAddDirect;
- pThis->qDel = qDelDirect;
pThis->MultiEnq = qqueueMultiEnqObjDirect;
+ pThis->qDel = NULL;
if(pThis->pqParent != NULL) {
DBGOPRINT((obj_t*) pThis, "DA queue is in emergency mode, disabling DA in parent\n");
pThis->pqParent->bIsDA = 0;
@@ -957,13 +960,11 @@ static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis)
return RS_RET_OK;
}
-static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg, wti_t *pWti)
+static rsRetVal qAddDirectWithWti(qqueue_t *pThis, msg_t* pMsg, wti_t *pWti)
{
batch_t singleBatch;
batch_obj_t batchObj;
batch_state_t batchState = BATCH_STATE_RDY;
- sbool active = 1;
- int i;
DEFiRet;
//TODO: init batchObj (states _OK and new fields -- CHECK)
@@ -983,46 +984,29 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg, wti_t *pWti)
singleBatch.nElem = 1; /* there always is only one in direct mode */
singleBatch.pElem = &batchObj;
singleBatch.eltState = &batchState;
- singleBatch.active = &active;
- iRet = pThis->pConsumer(pThis->pAction, &singleBatch, pWti, &pThis->bShutdownImmediate);
- /* delete the batch string params: TODO: create its own "class" for this */
- for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
- free(batchObj.staticActStrings[i]);
- }
+ iRet = pThis->pConsumer(pThis->pAction, &singleBatch, pWti);
msgDestruct(&pMsg);
RETiRet;
}
-/* "enqueue" a batch in direct mode. This is a shortcut which saves all the overhead
- * otherwise incured. -- rgerhards, ~2010-06-23
+/* this is called if we do not have a pWti. This currently only happens
+ * when we are called from a main queue in direct mode. If so, we need
+ * to obtain a dummy pWti.
*/
-rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch, wti_t *pWti)
+static rsRetVal
+qAddDirect(qqueue_t *pThis, msg_t* pMsg)
{
+ wti_t *pWti;
DEFiRet;
- ASSERT(pThis != NULL);
-
- /* calling the consumer is quite different here than it is from a worker thread */
- /* we need to provide the consumer's return value back to the caller because in direct
- * mode the consumer probably has a lot to convey (which get's lost in the other modes
- * because they are asynchronous. But direct mode is deliberately synchronous.
- * rgerhards, 2008-02-12
- * We use our knowledge about the batch_t structure below, but without that, we
- * pay a too-large performance toll... -- rgerhards, 2009-04-22
- */
- iRet = pThis->pConsumer(pThis->pAction, pBatch, pWti, NULL);
-
+ pWti = wtiGetDummy();
+ pWti->pbShutdownImmediate = &pThis->bShutdownImmediate;
+ iRet = qAddDirectWithWti(pThis, pMsg, pWti);
RETiRet;
}
-static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis)
-{
- return RS_RET_OK;
-}
-
-
/* --------------- end type-specific handlers -------------------- */
@@ -1317,7 +1301,7 @@ finalize_it:
* to modify some parameters before the queue is actually started.
*/
rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads,
- int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*, wti_t*, int*))
+ int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*, wti_t*))
{
DEFiRet;
qqueue_t *pThis;
@@ -1876,7 +1860,8 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti)
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave);
- CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, pWti, &pThis->bShutdownImmediate));
+ pWti->pbShutdownImmediate = &pThis->bShutdownImmediate;
+ CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, pWti));
/* we now need to check if we should deliberately delay processing a bit
* and, if so, do that. -- rgerhards, 2008-01-30
@@ -2098,9 +2083,13 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
case QUEUETYPE_DIRECT:
pThis->qConstruct = qConstructDirect;
pThis->qDestruct = qDestructDirect;
+ /* these entry points shall not be used in direct mode
+ * To catch program errors, make us abort if that happens!
+ * rgerhards, 2013-11-05
+ */
pThis->qAdd = qAddDirect;
- pThis->qDel = qDelDirect;
pThis->MultiEnq = qqueueMultiEnqObjDirect;
+ pThis->qDel = NULL;
break;
}
@@ -2679,16 +2668,17 @@ finalize_it:
/* now, the same function, but for direct mode */
static rsRetVal
-qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub, wti_t *pWti)
+qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub)
{
int i;
+ wti_t *pWti;
DEFiRet;
- ISOBJ_TYPE_assert(pThis, qqueue);
- assert(pMultiSub != NULL);
+ pWti = wtiGetDummy();
+ pWti->pbShutdownImmediate = &pThis->bShutdownImmediate;
for(i = 0 ; i < pMultiSub->nElem ; ++i) {
- CHKiRet(qAddDirect(pThis, (void*)pMultiSub->ppMsgs[i], pWti));
+ CHKiRet(qAddDirectWithWti(pThis, (void*)pMultiSub->ppMsgs[i], pWti));
}
finalize_it:
@@ -2697,21 +2687,6 @@ finalize_it:
/* ------------------------------ END multi-enqueue functions ------------------------------ */
-/* enqueue a new user data element in direct mode
- * NOTE/TODO: This is a TESTER/EXPERIEMENTAL, to be changed to better
- * code later on (like multi submit!) 2010-06-10
- * Enqueues the new element and awakes worker thread.
- */
-rsRetVal
-qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti)
-{
- DEFiRet;
- ISOBJ_TYPE_assert(pThis, qqueue);
- iRet = qAddDirect(pThis, pMsg, pWti);
- RETiRet;
-}
-
-
/* enqueue a new user data element
* Enqueues the new element and awakes worker thread.
*/
diff --git a/runtime/queue.h b/runtime/queue.h
index 8e789ebd..86107d24 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -103,11 +103,10 @@ struct queue_s {
* the user really wanted...). -- rgerhards, 2008-04-02
*/
/* end dequeue time window */
- rsRetVal (*pConsumer)(void *,batch_t*, wti_t*,int*); /* user-supplied consumer function for dequeued messages */
+ rsRetVal (*pConsumer)(void *,batch_t*, wti_t*); /* user-supplied consumer function for dequeued messages */
/* calling interface for pConsumer: arg1 is the global user pointer from this structure, arg2 is the
* user pointer array that was dequeued (actual sample: for actions, arg1 is the pAction and arg2
- * is pointer to an array of message message pointers), arg3 is a pointer to an interger which is zero
- * during normal operations and one if the consumer must urgently shut down.
+ * is pointer to an array of message message pointers)
*/
/* type-specific handlers (set during construction) */
rsRetVal (*qConstruct)(struct queue_s *pThis);
@@ -195,14 +194,12 @@ struct queue_s {
/* prototypes */
rsRetVal qqueueDestruct(qqueue_t **ppThis);
-rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti);
rsRetVal qqueueEnqMsg(qqueue_t *pThis, flowControl_t flwCtlType, msg_t *pMsg);
rsRetVal qqueueStart(qqueue_t *pThis);
rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize);
rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix);
rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads,
- int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, wti_t *, int*));
-rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch, wti_t *pWti);
+ int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, wti_t *));
int queueCnfParamsSet(struct nvlst *lst);
rsRetVal qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst);
void qqueueSetDefaultsRulesetQueue(qqueue_t *pThis);
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index edf0c593..ae56f08f 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -49,6 +49,7 @@
#define CONF_PROGNAME_BUFSIZE 16
#define CONF_HOSTNAME_BUFSIZE 32
#define CONF_PROP_BUFSIZE 16 /* should be close to sizeof(ptr) or lighly above it */
+#define CONF_IPARAMS_BUFSIZE 16 /* initial size of iparams array in wti (is automatically extended) */
#define CONF_MIN_SIZE_FOR_COMPRESS 60 /* config param: minimum message size to try compression. The smaller
* the message, the less likely is any compression gain. We check for
* gain before we submit the message. But to do so we still need to
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;
}
diff --git a/runtime/wti.c b/runtime/wti.c
index c2077a51..6b5d82c8 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -51,6 +51,8 @@
DEFobjStaticHelpers
DEFobjCurrIf(glbl)
+pthread_key_t thrd_wti_key;
+
/* forward-definitions */
/* methods */
@@ -208,6 +210,11 @@ wtiConstructFinalize(wti_t *pThis)
/* must use calloc as we need zero-init */
CHKmalloc(pThis->actWrkrInfo = calloc(iActionNbr, sizeof(actWrkrInfo_t)));
+ if(pThis->pWtp == NULL) {
+ dbgprintf("wtiConstructFinalize: pWtp not set, this may be intentional\n");
+ FINALIZE;
+ }
+
/* we now alloc the array for user pointers. We obtain the max from the queue itself. */
CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize));
CHKiRet(batchInit(&pThis->batch, iDeqBatchSize));
@@ -394,6 +401,33 @@ finalize_it:
}
+/* This function returns (and creates if necessary) a dummy wti suitable
+ * for use by the rule engine. It is intended to be used for direct-mode
+ * main queues (folks, don't do that!). Once created, data is stored in
+ * thread-specific storage.
+ * Note: we do NOT do error checking -- if this functions fails, all the
+ * rest will fail as well... (also, it will only fail under OOM, so...).
+ * Memleak: we leak pWti's when run in direct mode. However, this is only
+ * a cosmetic leak, as we need them until all inputs are terminated,
+ * what means essentially until rsyslog itself is terminated. So we
+ * don't care -- it's just not nice in valgrind, but that's it.
+ */
+wti_t *
+wtiGetDummy(void)
+{
+ wti_t *pWti;
+
+ pWti = (wti_t*) pthread_getspecific(thrd_wti_key);
+ if(pWti == NULL) {
+ wtiConstruct(&pWti);
+ wtiConstructFinalize(pWti);
+ if(pthread_setspecific(thrd_wti_key, pWti) != 0) {
+ DBGPRINTF("wtiGetDummy: error setspecific thrd_wti_key\n");
+ }
+ }
+ return pWti;
+}
+
/* dummy */
rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; }
@@ -403,6 +437,7 @@ BEGINObjClassExit(wti, OBJ_IS_CORE_MODULE) /* CHANGE class also in END MACRO! */
CODESTARTObjClassExit(nsdsel_gtls)
/* release objects we no longer need */
objRelease(glbl, CORE_COMPONENT);
+ pthread_key_delete(thrd_wti_key);
ENDObjClassExit(wti)
@@ -411,8 +446,14 @@ ENDObjClassExit(wti)
* rgerhards, 2008-01-09
*/
BEGINObjClassInit(wti, 1, OBJ_IS_CORE_MODULE) /* one is the object version (most important for persisting) */
+ int r;
/* request objects we use */
CHKiRet(objUse(glbl, CORE_COMPONENT));
+ r = pthread_key_create(&thrd_wti_key, NULL);
+ if(r != 0) {
+ dbgprintf("wti.c: pthread_key_create failed\n");
+ iRet = RS_RET_ERR;
+ }
ENDObjClassInit(wti)
/* vi:set ai:
diff --git a/runtime/wti.h b/runtime/wti.h
index 8c43c22e..adc7897c 100644
--- a/runtime/wti.h
+++ b/runtime/wti.h
@@ -1,6 +1,6 @@
/* Definition of the worker thread instance (wti) class.
*
- * Copyright 2008-2012 Adiscon GmbH.
+ * Copyright 2008-2013 Adiscon GmbH.
*
* This file is part of the rsyslog runtime library.
*
@@ -26,11 +26,45 @@
#include "wtp.h"
#include "obj.h"
#include "batch.h"
+#include "action.h"
+#define ACT_STATE_RDY 0 /* action ready, waiting for new transaction */
+#define ACT_STATE_ITX 1 /* transaction active, waiting for new data or commit */
+#define ACT_STATE_COMM 2 /* transaction finished (a transient state) */
+#define ACT_STATE_RTRY 3 /* failure occured, trying to restablish ready state */
+#define ACT_STATE_SUSP 4 /* suspended due to failure (return fail until timeout expired) */
+/* note: 3 bit bit field --> highest value is 7! */
+
+/* The following structure defines immutable parameters which need to
+ * be passed as action parameters. Note that the current implementation
+ * does NOT focus on performance, but on a simple PoC in order to get
+ * things going. TODO: Once it works, revisit this code and think about
+ * an array implementation. We also need to support other passing modes
+ * as well. -- gerhards, 2013-11-04
+ */
+typedef struct actWrkrIParams {
+ int msgFlags;
+ /* 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 */
+ unsigned staticLenStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE];
+ /* and the same for the message length (if used) */
+ void *staticActParams[CONF_OMOD_NUMSTRINGS_MAXSIZE];
+} actWrkrIParams_t;
+
typedef struct actWrkrInfo {
action_t *pAction;
void *actWrkrData;
+ uint16_t uResumeOKinRow;/* number of times in a row that resume said OK with an immediate failure following */
+ int iNbrResRtry; /* number of retries since last suspend */
+ struct {
+ unsigned actState : 3;
+ } flags;
+ actWrkrIParams_t *iparams;
+ int currIParam;
+ int maxIParams; /* current max */
+ void *staticActParams[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /* for non-strings */
} actWrkrInfo_t;
/* the worker thread instance class */
@@ -39,12 +73,20 @@ struct wti_s {
pthread_t thrdID; /* thread ID */
int bIsRunning; /* is this thread currently running? (must be int for atomic op!) */
sbool bAlwaysRunning; /* should this thread always run? */
+ int *pbShutdownImmediate;/* end processing of this batch immediately if set to 1 */
wtp_t *pWtp; /* my worker thread pool (important if only the work thread instance is passed! */
batch_t batch; /* pointer to an object array meaningful for current user pointer (e.g. queue pUsr data elemt) */
uchar *pszDbgHdr; /* header string for debug messages */
actWrkrInfo_t *actWrkrInfo; /* *array* of action wrkr infos for all actions (sized for max nbr of actions in config!) */
pthread_cond_t pcondBusy; /* condition to wake up the worker, protected by pmutUsr in wtp */
DEF_ATOMIC_HELPER_MUT(mutIsRunning);
+ struct {
+ uint8_t bPrevWasSuspended;
+ uint8_t bDoAutoCommit; /* do a commit after each message
+ * this is usually set for batches with 0 element, but may
+ * also be added as a user-selectable option (not implemented yet)
+ */
+ } execState; /* state for the execution engine */
};
@@ -59,8 +101,96 @@ rsRetVal wtiSetAlwaysRunning(wti_t *pThis);
rsRetVal wtiSetState(wti_t *pThis, sbool bNew);
rsRetVal wtiWakeupThrd(wti_t *pThis);
sbool wtiGetState(wti_t *pThis);
+wti_t *wtiGetDummy(void);
PROTOTYPEObjClassInit(wti);
PROTOTYPEpropSetMeth(wti, pszDbgHdr, uchar*);
PROTOTYPEpropSetMeth(wti, pWtp, wtp_t*);
+static inline uint8_t
+getActionStateByNbr(wti_t *pWti, int iActNbr)
+{
+ return((uint8_t) pWti->actWrkrInfo[iActNbr].flags.actState);
+}
+
+static inline uint8_t
+getActionState(wti_t *pWti, action_t *pAction)
+{
+ return((uint8_t) pWti->actWrkrInfo[pAction->iActionNbr].flags.actState);
+}
+
+static inline void
+setActionState(wti_t *pWti, action_t *pAction, uint8_t newState)
+{
+ pWti->actWrkrInfo[pAction->iActionNbr].flags.actState = newState;
+}
+
+static inline uint16_t
+getActionResumeInRow(wti_t *pWti, action_t *pAction)
+{
+ return(pWti->actWrkrInfo[pAction->iActionNbr].uResumeOKinRow);
+}
+
+static inline void
+setActionResumeInRow(wti_t *pWti, action_t *pAction, uint16_t val)
+{
+ pWti->actWrkrInfo[pAction->iActionNbr].uResumeOKinRow = val;
+}
+
+static inline void
+incActionResumeInRow(wti_t *pWti, action_t *pAction)
+{
+ pWti->actWrkrInfo[pAction->iActionNbr].uResumeOKinRow++;
+}
+
+static inline int
+getActionNbrResRtry(wti_t *pWti, action_t *pAction)
+{
+ return(pWti->actWrkrInfo[pAction->iActionNbr].iNbrResRtry);
+}
+
+static inline void
+setActionNbrResRtry(wti_t *pWti, action_t *pAction, uint16_t val)
+{
+ pWti->actWrkrInfo[pAction->iActionNbr].iNbrResRtry = val;
+}
+
+static inline void
+incActionNbrResRtry(wti_t *pWti, action_t *pAction)
+{
+ pWti->actWrkrInfo[pAction->iActionNbr].iNbrResRtry++;
+}
+
+static inline rsRetVal
+wtiNewIParam(wti_t *pWti, action_t *pAction, actWrkrIParams_t **piparams)
+{
+ actWrkrInfo_t *wrkrInfo;
+ actWrkrIParams_t *iparams;
+ int newMax;
+ DEFiRet;
+
+ wrkrInfo = &(pWti->actWrkrInfo[pAction->iActionNbr]);
+ if(wrkrInfo->currIParam == wrkrInfo->maxIParams) {
+ /* we need to extend */
+ dbgprintf("DDDD: extending iparams, max %d\n", wrkrInfo->maxIParams);
+ newMax = (wrkrInfo->maxIParams == 0) ? CONF_IPARAMS_BUFSIZE : 2 * wrkrInfo->maxIParams;
+ CHKmalloc(iparams = realloc(wrkrInfo->iparams, sizeof(actWrkrIParams_t) * newMax));
+ wrkrInfo->iparams = iparams;
+ wrkrInfo->maxIParams = newMax;
+ }
+dbgprintf("DDDD: adding param %d for action %d\n", wrkrInfo->currIParam, pAction->iActionNbr);
+ iparams = wrkrInfo->iparams + wrkrInfo->currIParam;
+ memset(iparams, 0, sizeof(actWrkrIParams_t));
+ *piparams = iparams;
+ ++wrkrInfo->currIParam;
+
+finalize_it:
+ RETiRet;
+}
+
+static inline void
+wtiResetExecState(wti_t *pWti, batch_t *pBatch)
+{
+ pWti->execState.bPrevWasSuspended = 0;
+ pWti->execState.bDoAutoCommit = (batchNumMsgs(pBatch) == 1);
+}
#endif /* #ifndef WTI_H_INCLUDED */
generated by cgit v1.2.3 (git 2.25.1) at 2025-07-08 21:45:23 +0000