From e4b3f6d287d74b34d27b4e296c33cb3f1294a58c Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 22 Apr 2009 16:39:58 +0200 Subject: now batches are handed down to the actual consumer ... but the action consumer does not do anything really intelligent with them. But the DA consumer is already done, as is the main message queue consumer. --- tools/syslogd.c | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index 8c86c12e..f48fa759 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" @@ -1202,22 +1203,29 @@ 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, aUsrp_t *paUsrp) { + int i; + msg_t *pMsg; DEFiRet; - msg_t *pMsg = (msg_t*) pUsr; - assert(pMsg != NULL); + assert(paUsrp != NULL); - if((pMsg->msgFlags & NEEDS_PARSING) != 0) { - parseMsg(pMsg); + for(i = 0 ; i < paUsrp->nElem ; i++) { + pMsg = (msg_t*) paUsrp->pUsrp[i]; +dbgprintf("msgConsumer..MULTIQUEUE: i: %d, pMsg: %p\n", i, pMsg); + if((pMsg->msgFlags & NEEDS_PARSING) != 0) { + parseMsg(pMsg); + } + processMsg(pMsg); + msgDestruct(&pMsg); } - processMsg(pMsg); - msgDestruct(&pMsg); +dbgprintf("DONE msgConsumer..MULTIQUEUE:\n"); RETiRet; } -- cgit From 5c0aeae8ab1f344a022d586dc26c5d78203f7e0b Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 23 Apr 2009 12:50:07 +0200 Subject: added $MainMsgQueueDequeueBatchSize and $ActionQueueDequeueBatchSize configuration directives --- tools/syslogd.c | 4 ++++ 1 file changed, 4 insertions(+) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index f48fa759..5b795755 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -304,6 +304,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 */ @@ -370,6 +371,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; @@ -2508,6 +2510,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 ); @@ -2888,6 +2891,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)); -- cgit From bb79e96dc300fa5a2182e7c047afb3b15c5dc870 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 12 May 2009 15:27:40 +0200 Subject: moving to a cleaner implementation of batches ... now that we know what we need from a theoretical POV. --- tools/syslogd.c | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index 866c0173..7ee5dbd7 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -138,6 +138,7 @@ #include "datetime.h" #include "parser.h" #include "sysvar.h" +#include "batch.h" /* definitions for objects we access */ DEFobjCurrIf(obj) @@ -1211,16 +1212,16 @@ processMsg(msg_t *pMsg) * for the main queue. */ static rsRetVal -msgConsumer(void __attribute__((unused)) *notNeeded, aUsrp_t *paUsrp) +msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch) { int i; msg_t *pMsg; DEFiRet; - assert(paUsrp != NULL); + assert(pBatch != NULL); - for(i = 0 ; i < paUsrp->nElem ; i++) { - pMsg = (msg_t*) paUsrp->pUsrp[i]; + for(i = 0 ; i < pBatch->nElem ; i++) { + pMsg = (msg_t*) pBatch->pElem[i].pUsrp; dbgprintf("msgConsumer..MULTIQUEUE: i: %d, pMsg: %p\n", i, pMsg); if((pMsg->msgFlags & NEEDS_PARSING) != 0) { parseMsg(pMsg); -- cgit From 73b16a5d7703078a46d960bd8922d2ae3a662769 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 13 May 2009 13:22:25 +0200 Subject: added test for property replacer field functionality to testbench plus some cosmetic changes --- tools/syslogd.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index 7ee5dbd7..cae07811 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -1155,6 +1155,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); @@ -1170,7 +1171,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 -- cgit From 4a8c02870a55e19c1bebfae5cb70d1ec5aa7c203 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 13 May 2009 16:00:15 +0200 Subject: moved user object destruction to queue itself So far, the consumer was responsible for destroying objects. However, this does not work well with ultra-reliable queues. This is the first move to support them. --- tools/syslogd.c | 1 - 1 file changed, 1 deletion(-) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index cae07811..88a588e9 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -1230,7 +1230,6 @@ dbgprintf("msgConsumer..MULTIQUEUE: i: %d, pMsg: %p\n", i, pMsg); parseMsg(pMsg); } processMsg(pMsg); - msgDestruct(&pMsg); } dbgprintf("DONE msgConsumer..MULTIQUEUE:\n"); -- cgit From a9c4b26d462dd3c9dbd0575a3a1acc6d8df1c3b3 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 19 May 2009 18:47:26 +0200 Subject: some cleanup --- tools/syslogd.c | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index 88a588e9..2bd43685 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -1225,13 +1225,12 @@ msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch) for(i = 0 ; i < pBatch->nElem ; i++) { pMsg = (msg_t*) pBatch->pElem[i].pUsrp; -dbgprintf("msgConsumer..MULTIQUEUE: i: %d, pMsg: %p\n", i, pMsg); + DBGPRINTF("msgConsumer processes msg %d/%d\n", i, pBatch->nElem); if((pMsg->msgFlags & NEEDS_PARSING) != 0) { parseMsg(pMsg); } processMsg(pMsg); } -dbgprintf("DONE msgConsumer..MULTIQUEUE:\n"); RETiRet; } -- cgit From 210f43137d6a077abbd8b77c1f72193dcd81cc99 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 25 May 2009 16:25:53 +0200 Subject: fixing some minor nits --- tools/syslogd.c | 1 + 1 file changed, 1 insertion(+) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index e207b5b2..31cc4a4d 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -1950,6 +1950,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) { -- cgit From d116f30a877c65b4f23dbb92601251402b0f957e Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 25 Jun 2009 12:53:00 +0200 Subject: improvements/fixes in queue termination timeout handling - bugfix: subtle (and usually irrelevant) issue in timout processing timeout could be one second too early if nanoseconds wrapped - set a more sensible timeout for shutdow, now 1.5 seconds to complete processing (this also removes those cases where the shutdown message was not written because the termination happened before it) --- tools/syslogd.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'tools/syslogd.c') diff --git a/tools/syslogd.c b/tools/syslogd.c index ace08e58..6a3fa6c9 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -302,7 +302,7 @@ static uchar *pszMainMsgQFName = NULL; /* prefix for the main message queue f static int64 iMainMsgQueMaxFileSize = 1024*1024; static int iMainMsgQPersistUpdCnt = 0; /* persist queue info every n updates */ static int bMainMsgQSyncQeueFiles = 0; /* sync queue files on every write? */ -static int iMainMsgQtoQShutdown = 0; /* queue shutdown */ +static int iMainMsgQtoQShutdown = 1500; /* queue shutdown (ms) */ static int iMainMsgQtoActShutdown = 1000; /* action shutdown (in phase 2) */ static int iMainMsgQtoEnq = 2000; /* timeout for queue enque */ static int iMainMsgQtoWrkShutdown = 60000; /* timeout for worker thread shutdown */ @@ -367,7 +367,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a iMainMsgQueueNumWorkers = 1; iMainMsgQPersistUpdCnt = 0; bMainMsgQSyncQeueFiles = 0; - iMainMsgQtoQShutdown = 0; + iMainMsgQtoQShutdown = 1500; iMainMsgQtoActShutdown = 1000; iMainMsgQtoEnq = 2000; iMainMsgQtoWrkShutdown = 60000; -- cgit