diff options
Diffstat (limited to 'tools')
| -rw-r--r-- | tools/omfwd.c | 2 | ||||
| -rw-r--r-- | tools/syslogd.c | 31 |
2 files changed, 23 insertions, 10 deletions
diff --git a/tools/omfwd.c b/tools/omfwd.c index fe65f515..38d9cfa6 100644 --- a/tools/omfwd.c +++ b/tools/omfwd.c @@ -10,7 +10,7 @@ * of the "old" message code without any modifications. However, it * helps to have things at the right place one we go to the meat of it. * - * Copyright 2007 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007, 2009 Rainer Gerhards and Adiscon GmbH. * * This file is part of rsyslog. * diff --git a/tools/syslogd.c b/tools/syslogd.c index 7f995e1c..4c27a662 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -129,12 +129,15 @@ #include "omfile.h" #include "omdiscard.h" #include "threads.h" +#include "wti.h" #include "queue.h" #include "stream.h" #include "conf.h" #include "errmsg.h" #include "datetime.h" #include "parser.h" +//#include "sysvar.h" +#include "batch.h" #include "unicode-helper.h" #include "ruleset.h" #include "rule.h" @@ -293,6 +296,7 @@ static int iMainMsgQtoWrkShutdown = 60000; /* timeout for worker thread shutdo static int iMainMsgQWrkMinMsgs = 100; /* minimum messages per worker needed to start a new one */ static int iMainMsgQDeqSlowdown = 0; /* dequeue slowdown (simple rate limiting) */ static int64 iMainMsgQueMaxDiskSpace = 0; /* max disk space allocated 0 ==> unlimited */ +static int iMainMsgQueDeqBatchSize = 32; /* dequeue batch size */ static int bMainMsgQSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */ static int iMainMsgQueueDeqtWinFromHr = 0; /* hour begin of time frame when queue is to be dequeued */ static int iMainMsgQueueDeqtWinToHr = 25; /* hour begin of time frame when queue is to be dequeued */ @@ -359,6 +363,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a bMainMsgQSaveOnShutdown = 1; MainMsgQueType = QUEUETYPE_FIXED_ARRAY; iMainMsgQueMaxDiskSpace = 0; + iMainMsgQueDeqBatchSize = 32; glbliActionResumeRetryCount = 0; return RS_RET_OK; @@ -918,22 +923,27 @@ finalize_it: /* The consumer of dequeued messages. This function is called by the * queue engine on dequeueing of a message. It runs on a SEPARATE - * THREAD. - * Please note: the message object is destructed by the queue itself! + * THREAD. It receives an array of pointers, which it must iterate + * over. We do not do any further batching, as this is of no benefit + * for the main queue. */ static rsRetVal -msgConsumer(void __attribute__((unused)) *notNeeded, void *pUsr) +msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch) { + int i; + msg_t *pMsg; DEFiRet; - msg_t *pMsg = (msg_t*) pUsr; - assert(pMsg != NULL); + assert(pBatch != NULL); - if((pMsg->msgFlags & NEEDS_PARSING) != 0) { - parseMsg(pMsg); + for(i = 0 ; i < pBatch->nElem ; i++) { + pMsg = (msg_t*) pBatch->pElem[i].pUsrp; + DBGPRINTF("msgConsumer processes msg %d/%d\n", i, pBatch->nElem); + if((pMsg->msgFlags & NEEDS_PARSING) != 0) { + parseMsg(pMsg); + } + ruleset.ProcessMsg(pMsg); } - ruleset.ProcessMsg(pMsg); - msgDestruct(&pMsg); RETiRet; } @@ -1588,6 +1598,7 @@ static void doDie(int sig) # define MSG1 "DoDie called.\n" # define MSG2 "DoDie called 5 times - unconditional exit\n" static int iRetries = 0; /* debug aid */ + dbgprintf(MSG1); if(Debug) write(1, MSG1, sizeof(MSG1) - 1); if(iRetries++ == 4) { @@ -2304,6 +2315,7 @@ init() setQPROP(qqueueSetMaxFileSize, "$MainMsgQueueFileSize", iMainMsgQueMaxFileSize); setQPROP(qqueueSetsizeOnDiskMax, "$MainMsgQueueMaxDiskSpace", iMainMsgQueMaxDiskSpace); + setQPROP(qqueueSetiDeqBatchSize, "$MainMsgQueueDequeueBatchSize", iMainMsgQueDeqBatchSize); setQPROPstr(qqueueSetFilePrefix, "$MainMsgQueueFileName", pszMainMsgQFName); setQPROP(qqueueSetiPersistUpdCnt, "$MainMsgQueueCheckpointInterval", iMainMsgQPersistUpdCnt); setQPROP(qqueueSetbSyncQueueFiles, "$MainMsgQueueSyncQueueFiles", bMainMsgQSyncQeueFiles); @@ -2650,6 +2662,7 @@ static rsRetVal loadBuildInModules(void) CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuedequeueslowdown", 0, eCmdHdlrInt, NULL, &iMainMsgQDeqSlowdown, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueueworkerthreadminimummessages", 0, eCmdHdlrInt, NULL, &iMainMsgQWrkMinMsgs, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuemaxfilesize", 0, eCmdHdlrSize, NULL, &iMainMsgQueMaxFileSize, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuedequeuebatchsize", 0, eCmdHdlrSize, NULL, &iMainMsgQueDeqBatchSize, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuemaxdiskspace", 0, eCmdHdlrSize, NULL, &iMainMsgQueMaxDiskSpace, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuesaveonshutdown", 0, eCmdHdlrBinary, NULL, &bMainMsgQSaveOnShutdown, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuedequeuetimebegin", 0, eCmdHdlrInt, NULL, &iMainMsgQueueDeqtWinFromHr, NULL)); |