diff options
Diffstat (limited to 'tools')
| -rw-r--r-- | tools/omfile.c | 6 | ||||
| -rw-r--r-- | tools/syslogd.c | 35 |
2 files changed, 28 insertions, 13 deletions
diff --git a/tools/omfile.c b/tools/omfile.c index c7283e4d..3e845a73 100644 --- a/tools/omfile.c +++ b/tools/omfile.c @@ -132,7 +132,7 @@ ENDisCompatibleWithFeature BEGINdbgPrintInstInfo CODESTARTdbgPrintInstInfo if(pData->bDynamicName) { - printf("[dynamic]\n\ttemplate='%s'" + dbgprintf("[dynamic]\n\ttemplate='%s'" "\tfile cache size=%d\n" "\tcreate directories: %s\n" "\tfile owner %d, group %d\n" @@ -146,9 +146,9 @@ CODESTARTdbgPrintInstInfo pData->bFailOnChown ? "yes" : "no" ); } else { /* regular file */ - printf("%s", pData->f_fname); + dbgprintf("%s", pData->f_fname); if (pData->fd == -1) - printf(" (unused)"); + dbgprintf(" (unused)"); } ENDdbgPrintInstInfo diff --git a/tools/syslogd.c b/tools/syslogd.c index d925505f..47fa2f58 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -129,6 +129,7 @@ #include "omfile.h" #include "omdiscard.h" #include "threads.h" +#include "wti.h" #include "queue.h" #include "stream.h" #include "conf.h" @@ -137,6 +138,7 @@ #include "datetime.h" #include "parser.h" #include "sysvar.h" +#include "batch.h" #include "unicode-helper.h" /* definitions for objects we access */ @@ -305,6 +307,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 */ @@ -369,6 +372,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; @@ -1170,6 +1174,7 @@ DEFFUNC_llExecFunc(processMsgDoActions) ABORT_FINALIZE(RS_RET_OK); } + /* MULTIQUEUE: look at this below! (I say: batch states!) */ iRetMod = actionCallAction(pAction, pDoActData->pMsg); if(iRetMod == RS_RET_DISCARDMSG) { ABORT_FINALIZE(RS_RET_DISCARDMSG); @@ -1185,7 +1190,9 @@ finalize_it: } -/* Process (consume) a received message. Calls the actions configured. +/* Process (consume) a received message from the main queue. Here, messages are + * filtered and those where the filter evaluates to true are passed to the action + * queue for further processing. * rgerhards, 2005-10-13 */ static void @@ -1222,22 +1229,27 @@ processMsg(msg_t *pMsg) /* 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); + } + processMsg(pMsg); } - processMsg(pMsg); - msgDestruct(&pMsg); RETiRet; } @@ -1943,6 +1955,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) { @@ -2714,6 +2727,7 @@ init(void) setQPROP(qqueueSetMaxFileSize, "$MainMsgQueueFileSize", iMainMsgQueMaxFileSize); setQPROP(qqueueSetsizeOnDiskMax, "$MainMsgQueueMaxDiskSpace", iMainMsgQueMaxDiskSpace); + setQPROP(qqueueSetiDeqBatchSize, "$MainMsgQueueDequeueBatchSize", iMainMsgQueDeqBatchSize); setQPROPstr(qqueueSetFilePrefix, "$MainMsgQueueFileName", pszMainMsgQFName); setQPROP(qqueueSetiPersistUpdCnt, "$MainMsgQueueCheckpointInterval", iMainMsgQPersistUpdCnt); setQPROP(qqueueSettoQShutdown, "$MainMsgQueueTimeoutShutdown", iMainMsgQtoQShutdown ); @@ -3094,6 +3108,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)); |