diff options
Diffstat (limited to 'runtime/queue.c')
| -rw-r--r-- | runtime/queue.c | 3609 |
1 files changed, 3609 insertions, 0 deletions
diff --git a/runtime/queue.c b/runtime/queue.c new file mode 100644 index 0000000..edc9d35 --- /dev/null +++ b/runtime/queue.c @@ -0,0 +1,3609 @@ +/* queue.c + * + * This file implements the queue object and its several queueing methods. + * + * File begun on 2008-01-03 by RGerhards + * + * There is some in-depth documentation available in doc/dev_queue.html + * (and in the web doc set on https://www.rsyslog.com/doc/). Be sure to read it + * if you are getting aquainted to the object. + * + * NOTE: as of 2009-04-22, I have begin to remove the qqueue* prefix from static + * function names - this makes it really hard to read and does not provide much + * benefit, at least I (now) think so... + * + * Copyright 2008-2019 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#include "config.h" + +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <signal.h> +#include <pthread.h> +#include <fcntl.h> +#include <unistd.h> +#include <sys/stat.h> /* required for HP UX */ +#include <time.h> +#include <errno.h> +#include <inttypes.h> + +#include "rsyslog.h" +#include "queue.h" +#include "stringbuf.h" +#include "srUtils.h" +#include "obj.h" +#include "wtp.h" +#include "wti.h" +#include "msg.h" +#include "obj.h" +#include "atomic.h" +#include "errmsg.h" +#include "datetime.h" +#include "unicode-helper.h" +#include "statsobj.h" +#include "parserif.h" +#include "rsconf.h" + +#ifdef OS_SOLARIS +# include <sched.h> +#endif + +/* static data */ +DEFobjStaticHelpers +DEFobjCurrIf(glbl) +DEFobjCurrIf(strm) +DEFobjCurrIf(datetime) +DEFobjCurrIf(statsobj) + +#if __GNUC__ >= 8 +#pragma GCC diagnostic ignored "-Wcast-function-type" // TODO: investigate further! +#endif /* if __GNUC__ >= 8 */ + +#ifdef ENABLE_IMDIAG +unsigned int iOverallQueueSize = 0; +#endif + +#define OVERSIZE_QUEUE_WATERMARK 500000 /* when is a queue considered to be "overly large"? */ + + +/* forward-definitions */ +static rsRetVal doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, smsg_t *pMsg); +static rsRetVal qqueueChkPersist(qqueue_t *pThis, int nUpdates); +static rsRetVal RateLimiter(qqueue_t *pThis); +static rsRetVal qqueueChkStopWrkrDA(qqueue_t *pThis); +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); +static rsRetVal qAddDirect(qqueue_t *pThis, smsg_t *pMsg); +static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis); +static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis); +static rsRetVal qDestructDisk(qqueue_t *pThis); +rsRetVal qqueueSetSpoolDir(qqueue_t *pThis, uchar *pszSpoolDir, int lenSpoolDir); + +/* some constants for queuePersist () */ +#define QUEUE_CHECKPOINT 1 +#define QUEUE_NO_CHECKPOINT 0 + +/* tables for interfacing with the v6 config system */ +static struct cnfparamdescr cnfpdescr[] = { + { "queue.filename", eCmdHdlrGetWord, 0 }, + { "queue.spooldirectory", eCmdHdlrGetWord, 0 }, + { "queue.size", eCmdHdlrSize, 0 }, + { "queue.dequeuebatchsize", eCmdHdlrInt, 0 }, + { "queue.mindequeuebatchsize", eCmdHdlrInt, 0 }, + { "queue.mindequeuebatchsize.timeout", eCmdHdlrInt, 0 }, + { "queue.maxdiskspace", eCmdHdlrSize, 0 }, + { "queue.highwatermark", eCmdHdlrInt, 0 }, + { "queue.lowwatermark", eCmdHdlrInt, 0 }, + { "queue.fulldelaymark", eCmdHdlrInt, 0 }, + { "queue.lightdelaymark", eCmdHdlrInt, 0 }, + { "queue.discardmark", eCmdHdlrInt, 0 }, + { "queue.discardseverity", eCmdHdlrFacility, 0 }, + { "queue.checkpointinterval", eCmdHdlrInt, 0 }, + { "queue.syncqueuefiles", eCmdHdlrBinary, 0 }, + { "queue.type", eCmdHdlrQueueType, 0 }, + { "queue.workerthreads", eCmdHdlrInt, 0 }, + { "queue.timeoutshutdown", eCmdHdlrInt, 0 }, + { "queue.timeoutactioncompletion", eCmdHdlrInt, 0 }, + { "queue.timeoutenqueue", eCmdHdlrInt, 0 }, + { "queue.timeoutworkerthreadshutdown", eCmdHdlrInt, 0 }, + { "queue.workerthreadminimummessages", eCmdHdlrInt, 0 }, + { "queue.maxfilesize", eCmdHdlrSize, 0 }, + { "queue.saveonshutdown", eCmdHdlrBinary, 0 }, + { "queue.dequeueslowdown", eCmdHdlrInt, 0 }, + { "queue.dequeuetimebegin", eCmdHdlrInt, 0 }, + { "queue.dequeuetimeend", eCmdHdlrInt, 0 }, + { "queue.cry.provider", eCmdHdlrGetWord, 0 }, + { "queue.samplinginterval", eCmdHdlrInt, 0 }, + { "queue.takeflowctlfrommsg", eCmdHdlrBinary, 0 } +}; +static struct cnfparamblk pblk = + { CNFPARAMBLK_VERSION, + sizeof(cnfpdescr)/sizeof(struct cnfparamdescr), + cnfpdescr + }; + +/* support to detect duplicate queue file names */ +struct queue_filename { + struct queue_filename *next; + const char *dirname; + const char *filename; +}; +struct queue_filename *queue_filename_root = NULL; + +/* debug aid */ +#if 0 +static inline void displayBatchState(batch_t *pBatch) +{ + int i; + for(i = 0 ; i < pBatch->nElem ; ++i) { + DBGPRINTF("displayBatchState %p[%d]: %d\n", pBatch, i, pBatch->eltState[i]); + } +} +#endif +static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint); + +/* do cleanup when config is loaded */ +void qqueueDoneLoadCnf(void) +{ + struct queue_filename *next, *del; + next = queue_filename_root; + while(next != NULL) { + del = next; + next = next->next; + free((void*) del->filename); + free((void*) del->dirname); + free((void*) del); + } +} + + +/*********************************************************************** + * we need a private data structure, the "to-delete" list. As C does + * not provide any partly private data structures, we implement this + * structure right here inside the module. + * Note that this list must always be kept sorted based on a unique + * dequeue ID (which is monotonically increasing). + * rgerhards, 2009-05-18 + ***********************************************************************/ + +/* generate next uniqueue dequeue ID. Note that uniqueness is only required + * on a per-queue basis and while this instance runs. So a stricly monotonically + * increasing counter is sufficient (if enough bits are used). + */ +static inline qDeqID getNextDeqID(qqueue_t *pQueue) +{ + ISOBJ_TYPE_assert(pQueue, qqueue); + return pQueue->deqIDAdd++; +} + + +/* return the top element of the to-delete list or NULL, if the + * list is empty. + */ +static toDeleteLst_t *tdlPeek(qqueue_t *pQueue) +{ + ISOBJ_TYPE_assert(pQueue, qqueue); + return pQueue->toDeleteLst; +} + + +/* remove the top element of the to-delete list. Nothing but the + * element itself is destroyed. Must not be called when the list + * is empty. + */ +static rsRetVal tdlPop(qqueue_t *pQueue) +{ + toDeleteLst_t *pRemove; + DEFiRet; + + ISOBJ_TYPE_assert(pQueue, qqueue); + assert(pQueue->toDeleteLst != NULL); + + pRemove = pQueue->toDeleteLst; + pQueue->toDeleteLst = pQueue->toDeleteLst->pNext; + free(pRemove); + + RETiRet; +} + + +/* Add a new to-delete list entry. The function allocates the data + * structure, populates it with the values provided and links the new + * element into the correct place inside the list. + */ +static rsRetVal +tdlAdd(qqueue_t *pQueue, qDeqID deqID, int nElemDeq) +{ + toDeleteLst_t *pNew; + toDeleteLst_t *pPrev; + DEFiRet; + + ISOBJ_TYPE_assert(pQueue, qqueue); + assert(pQueue->toDeleteLst != NULL); + + CHKmalloc(pNew = malloc(sizeof(toDeleteLst_t))); + pNew->deqID = deqID; + pNew->nElemDeq = nElemDeq; + + /* now find right spot */ + for( pPrev = pQueue->toDeleteLst + ; pPrev != NULL && deqID > pPrev->deqID + ; pPrev = pPrev->pNext) { + /*JUST SEARCH*/; + } + + if(pPrev == NULL) { + pNew->pNext = pQueue->toDeleteLst; + pQueue->toDeleteLst = pNew; + } else { + pNew->pNext = pPrev->pNext; + pPrev->pNext = pNew; + } + +finalize_it: + RETiRet; +} + + +/* methods */ + +static const char * +getQueueTypeName(queueType_t t) +{ + const char *r; + + switch(t) { + case QUEUETYPE_FIXED_ARRAY: + r = "FixedArray"; + break; + case QUEUETYPE_LINKEDLIST: + r = "LinkedList"; + break; + case QUEUETYPE_DISK: + r = "Disk"; + break; + case QUEUETYPE_DIRECT: + r = "Direct"; + break; + default: + r = "invalid/unknown queue mode"; + break; + } + return r; +} + +void +qqueueDbgPrint(qqueue_t *pThis) +{ + dbgoprint((obj_t*) pThis, "parameter dump:\n"); + dbgoprint((obj_t*) pThis, "queue.filename '%s'\n", + (pThis->pszFilePrefix == NULL) ? "[NONE]" : (char*)pThis->pszFilePrefix); + dbgoprint((obj_t*) pThis, "queue.size: %d\n", pThis->iMaxQueueSize); + dbgoprint((obj_t*) pThis, "queue.dequeuebatchsize: %d\n", pThis->iDeqBatchSize); + dbgoprint((obj_t*) pThis, "queue.mindequeuebatchsize: %d\n", pThis->iMinDeqBatchSize); + dbgoprint((obj_t*) pThis, "queue.mindequeuebatchsize.timeout: %d\n", pThis->toMinDeqBatchSize); + dbgoprint((obj_t*) pThis, "queue.maxdiskspace: %lld\n", pThis->sizeOnDiskMax); + dbgoprint((obj_t*) pThis, "queue.highwatermark: %d\n", pThis->iHighWtrMrk); + dbgoprint((obj_t*) pThis, "queue.lowwatermark: %d\n", pThis->iLowWtrMrk); + dbgoprint((obj_t*) pThis, "queue.fulldelaymark: %d\n", pThis->iFullDlyMrk); + dbgoprint((obj_t*) pThis, "queue.lightdelaymark: %d\n", pThis->iLightDlyMrk); + dbgoprint((obj_t*) pThis, "queue.takeflowctlfrommsg: %d\n", pThis->takeFlowCtlFromMsg); + dbgoprint((obj_t*) pThis, "queue.discardmark: %d\n", pThis->iDiscardMrk); + dbgoprint((obj_t*) pThis, "queue.discardseverity: %d\n", pThis->iDiscardSeverity); + dbgoprint((obj_t*) pThis, "queue.checkpointinterval: %d\n", pThis->iPersistUpdCnt); + dbgoprint((obj_t*) pThis, "queue.syncqueuefiles: %d\n", pThis->bSyncQueueFiles); + dbgoprint((obj_t*) pThis, "queue.type: %d [%s]\n", pThis->qType, getQueueTypeName(pThis->qType)); + dbgoprint((obj_t*) pThis, "queue.workerthreads: %d\n", pThis->iNumWorkerThreads); + dbgoprint((obj_t*) pThis, "queue.timeoutshutdown: %d\n", pThis->toQShutdown); + dbgoprint((obj_t*) pThis, "queue.timeoutactioncompletion: %d\n", pThis->toActShutdown); + dbgoprint((obj_t*) pThis, "queue.timeoutenqueue: %d\n", pThis->toEnq); + dbgoprint((obj_t*) pThis, "queue.timeoutworkerthreadshutdown: %d\n", pThis->toWrkShutdown); + dbgoprint((obj_t*) pThis, "queue.workerthreadminimummessages: %d\n", pThis->iMinMsgsPerWrkr); + dbgoprint((obj_t*) pThis, "queue.maxfilesize: %lld\n", pThis->iMaxFileSize); + dbgoprint((obj_t*) pThis, "queue.saveonshutdown: %d\n", pThis->bSaveOnShutdown); + dbgoprint((obj_t*) pThis, "queue.dequeueslowdown: %d\n", pThis->iDeqSlowdown); + dbgoprint((obj_t*) pThis, "queue.dequeuetimebegin: %d\n", pThis->iDeqtWinFromHr); + dbgoprint((obj_t*) pThis, "queue.dequeuetimeend: %d\n", pThis->iDeqtWinToHr); +} + + +/* get the physical queue size. Must only be called + * while mutex is locked! + * rgerhards, 2008-01-29 + */ +static int +getPhysicalQueueSize(qqueue_t *pThis) +{ + return (int) PREFER_FETCH_32BIT(pThis->iQueueSize); +} + + +/* get the logical queue size (that is store size minus logically dequeued elements). + * Must only be called while mutex is locked! + * rgerhards, 2009-05-19 + */ +static int +getLogicalQueueSize(qqueue_t *pThis) +{ + return pThis->iQueueSize - pThis->nLogDeq; +} + + + +/* This function drains the queue in cases where this needs to be done. The most probable + * reason is a HUP which needs to discard data (because the queue is configured to be lossy). + * During a shutdown, this is typically not needed, as the OS frees up ressources and does + * this much quicker than when we clean up ourselvs. -- rgerhards, 2008-10-21 + * This function returns void, as it makes no sense to communicate an error back, even if + * it happens. + * This functions works "around" the regular deque mechanism, because it is only used to + * clean up (in cases where message loss is acceptable). + */ +static void queueDrain(qqueue_t *pThis) +{ + smsg_t *pMsg; + assert(pThis != NULL); + + DBGOPRINT((obj_t*) pThis, "queue (type %d) will lose %d messages, destroying...\n", + pThis->qType, pThis->iQueueSize); + /* iQueueSize is not decremented by qDel(), so we need to do it ourselves */ + while(ATOMIC_DEC_AND_FETCH(&pThis->iQueueSize, &pThis->mutQueueSize) > 0) { + pThis->qDeq(pThis, &pMsg); + if(pMsg != NULL) { + msgDestruct(&pMsg); + } + pThis->qDel(pThis); + } +} + + +/* --------------- code for disk-assisted (DA) queue modes -------------------- */ + + +/* returns the number of workers that should be advised at + * this point in time. The mutex must be locked when + * ths function is called. -- rgerhards, 2008-01-25 + */ +static rsRetVal +qqueueAdviseMaxWorkers(qqueue_t *pThis) +{ + DEFiRet; + int iMaxWorkers; + + ISOBJ_TYPE_assert(pThis, qqueue); + + if(!pThis->bEnqOnly) { + if(pThis->bIsDA && getLogicalQueueSize(pThis) >= pThis->iHighWtrMrk) { + DBGOPRINT((obj_t*) pThis, "(re)activating DA worker\n"); + wtpAdviseMaxWorkers(pThis->pWtpDA, 1, DENY_WORKER_START_DURING_SHUTDOWN); + /* disk queues have always one worker */ + } + if(getLogicalQueueSize(pThis) == 0) { + iMaxWorkers = 0; + } else if(pThis->iMinMsgsPerWrkr == 0) { + iMaxWorkers = 1; + } else { + iMaxWorkers = getLogicalQueueSize(pThis) / pThis->iMinMsgsPerWrkr + 1; + } + wtpAdviseMaxWorkers(pThis->pWtpReg, iMaxWorkers, DENY_WORKER_START_DURING_SHUTDOWN); + } + + RETiRet; +} + + +/* check if we run in disk-assisted mode and record that + * setting for easy (and quick!) access in the future. This + * function must only be called from constructors and only + * from those that support disk-assisted modes (aka memory- + * based queue drivers). + * rgerhards, 2008-01-14 + */ +static rsRetVal +qqueueChkIsDA(qqueue_t *pThis) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + if(pThis->pszFilePrefix != NULL) { + pThis->bIsDA = 1; + DBGOPRINT((obj_t*) pThis, "is disk-assisted, disk will be used on demand\n"); + } else { + DBGOPRINT((obj_t*) pThis, "is NOT disk-assisted\n"); + } + + RETiRet; +} + + +/* Start disk-assisted queue mode. + * rgerhards, 2008-01-15 + */ +static rsRetVal +StartDA(qqueue_t *pThis) +{ + DEFiRet; + uchar pszDAQName[128]; + + ISOBJ_TYPE_assert(pThis, qqueue); + + /* create message queue */ + CHKiRet(qqueueConstruct(&pThis->pqDA, QUEUETYPE_DISK, pThis->iNumWorkerThreads, 0, pThis->pConsumer)); + + /* give it a name */ + snprintf((char*) pszDAQName, sizeof(pszDAQName), "%s[DA]", obj.GetName((obj_t*) pThis)); + obj.SetName((obj_t*) pThis->pqDA, pszDAQName); + + /* as the created queue is the same object class, we take the + * liberty to access its properties directly. + */ + pThis->pqDA->pqParent = pThis; + + CHKiRet(qqueueSetpAction(pThis->pqDA, pThis->pAction)); + CHKiRet(qqueueSetsizeOnDiskMax(pThis->pqDA, pThis->sizeOnDiskMax)); + CHKiRet(qqueueSetiDeqSlowdown(pThis->pqDA, pThis->iDeqSlowdown)); + CHKiRet(qqueueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize)); + CHKiRet(qqueueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(qqueueSetSpoolDir(pThis->pqDA, pThis->pszSpoolDir, pThis->lenSpoolDir)); + CHKiRet(qqueueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt)); + CHKiRet(qqueueSetbSyncQueueFiles(pThis->pqDA, pThis->bSyncQueueFiles)); + CHKiRet(qqueueSettoActShutdown(pThis->pqDA, pThis->toActShutdown)); + CHKiRet(qqueueSettoEnq(pThis->pqDA, pThis->toEnq)); + CHKiRet(qqueueSetiDeqtWinFromHr(pThis->pqDA, pThis->iDeqtWinFromHr)); + CHKiRet(qqueueSetiDeqtWinToHr(pThis->pqDA, pThis->iDeqtWinToHr)); + CHKiRet(qqueueSettoQShutdown(pThis->pqDA, pThis->toQShutdown)); + CHKiRet(qqueueSetiHighWtrMrk(pThis->pqDA, 0)); + CHKiRet(qqueueSetiDiscardMrk(pThis->pqDA, 0)); + pThis->pqDA->iDeqBatchSize = pThis->iDeqBatchSize; + pThis->pqDA->iMinDeqBatchSize = pThis->iMinDeqBatchSize; + pThis->pqDA->iMinMsgsPerWrkr = pThis->iMinMsgsPerWrkr; + pThis->pqDA->iLowWtrMrk = pThis->iLowWtrMrk; + if(pThis->useCryprov) { + /* hand over cryprov to DA queue - in-mem queue does no longer need it + * and DA queue will be kept active from now on until termination. + */ + pThis->pqDA->useCryprov = pThis->useCryprov; + pThis->pqDA->cryprov = pThis->cryprov; + pThis->pqDA->cryprovData = pThis->cryprovData; + pThis->pqDA->cryprovName = pThis->cryprovName; + pThis->pqDA->cryprovNameFull = pThis->cryprovNameFull; + /* reset memory queue parameters */ + pThis->useCryprov = 0; + /* pThis->cryprov cannot and need not be reset, is structure */ + pThis->cryprovData = NULL; + pThis->cryprovName = NULL; + pThis->cryprovNameFull = NULL; + } + + iRet = qqueueStart(runConf, pThis->pqDA); + /* file not found is expected, that means it is no previous QIF available */ + if(iRet != RS_RET_OK && iRet != RS_RET_FILE_NOT_FOUND) { + errno = 0; /* else an errno is shown in errmsg! */ + LogError(errno, iRet, "error starting up disk queue, using pure in-memory mode"); + pThis->bIsDA = 0; /* disable memory mode */ + FINALIZE; /* something is wrong */ + } + + DBGOPRINT((obj_t*) pThis, "DA queue initialized, disk queue 0x%lx\n", + qqueueGetID(pThis->pqDA)); + +finalize_it: + if(iRet != RS_RET_OK) { + if(pThis->pqDA != NULL) { + qqueueDestruct(&pThis->pqDA); + } + LogError(0, iRet, "%s: error creating disk queue - giving up.", + obj.GetName((obj_t*)pThis)); + pThis->bIsDA = 0; + } + + RETiRet; +} + + +/* initiate DA mode + * param bEnqOnly tells if the disk queue is to be run in enqueue-only mode. This may + * be needed during shutdown of memory queues which need to be persisted to disk. + * If this function fails (should not happen), DA mode is not turned on. + * rgerhards, 2008-01-16 + */ +static rsRetVal ATTR_NONNULL() +InitDA(qqueue_t *const pThis, const int bLockMutex) +{ + DEFiRet; + uchar pszBuf[64]; + size_t lenBuf; + + ISOBJ_TYPE_assert(pThis, qqueue); + if(bLockMutex == LOCK_MUTEX) { + d_pthread_mutex_lock(pThis->mut); + } + + /* check if we already have a DA worker pool. If not, initiate one. Please note that the + * pool is created on first need but never again destructed (until the queue is). This + * is intentional. We assume that when we need it once, we may also need it on another + * occasion. Ressources used are quite minimal when no worker is running. + * rgerhards, 2008-01-24 + * NOTE: this is the DA worker *pool*, not the DA queue! + */ + lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "%s:DAwpool", obj.GetName((obj_t*) pThis)); + CHKiRet(wtpConstruct (&pThis->pWtpDA)); + CHKiRet(wtpSetDbgHdr (pThis->pWtpDA, pszBuf, lenBuf)); + CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, int)) qqueueChkStopWrkrDA)); + CHKiRet(wtpSetpfGetDeqBatchSize (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, int*)) GetDeqBatchSize)); + CHKiRet(wtpSetpfDoWork (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, void *pWti)) ConsumerDA)); + CHKiRet(wtpSetpfObjProcessed (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, wti_t *pWti)) batchProcessed)); + CHKiRet(wtpSetpmutUsr (pThis->pWtpDA, pThis->mut)); + CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpDA, 1)); + CHKiRet(wtpSettoWrkShutdown (pThis->pWtpDA, pThis->toWrkShutdown)); + CHKiRet(wtpSetpUsr (pThis->pWtpDA, pThis)); + CHKiRet(wtpConstructFinalize (pThis->pWtpDA)); + /* if we reach this point, we have a "good" DA worker pool */ + + /* now construct the actual queue (if it does not already exist) */ + if(pThis->pqDA == NULL) { + CHKiRet(StartDA(pThis)); + } + +finalize_it: + if(bLockMutex == LOCK_MUTEX) { + d_pthread_mutex_unlock(pThis->mut); + } + RETiRet; +} + + +/* --------------- end code for disk-assisted queue modes -------------------- */ + + +/* Now, we define type-specific handlers. The provide a generic functionality, + * but for this specific type of queue. The mapping to these handlers happens during + * queue construction. Later on, handlers are called by pointers present in the + * queue instance object. + */ + +/* -------------------- fixed array -------------------- */ +static rsRetVal qConstructFixedArray(qqueue_t *pThis) +{ + DEFiRet; + + assert(pThis != NULL); + + if(pThis->iMaxQueueSize == 0) + ABORT_FINALIZE(RS_RET_QSIZE_ZERO); + + if((pThis->tVars.farray.pBuf = malloc(sizeof(void *) * pThis->iMaxQueueSize)) == NULL) { + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + } + + pThis->tVars.farray.deqhead = 0; + pThis->tVars.farray.head = 0; + pThis->tVars.farray.tail = 0; + + qqueueChkIsDA(pThis); + +finalize_it: + RETiRet; +} + + +static rsRetVal qDestructFixedArray(qqueue_t *pThis) +{ + DEFiRet; + + assert(pThis != NULL); + + queueDrain(pThis); /* discard any remaining queue entries */ + free(pThis->tVars.farray.pBuf); + + RETiRet; +} + + +static rsRetVal qAddFixedArray(qqueue_t *pThis, smsg_t* in) +{ + DEFiRet; + + assert(pThis != NULL); + pThis->tVars.farray.pBuf[pThis->tVars.farray.tail] = in; + pThis->tVars.farray.tail++; + if (pThis->tVars.farray.tail == pThis->iMaxQueueSize) + pThis->tVars.farray.tail = 0; + + RETiRet; +} + + +static rsRetVal qDeqFixedArray(qqueue_t *pThis, smsg_t **out) +{ + DEFiRet; + + assert(pThis != NULL); + *out = (void*) pThis->tVars.farray.pBuf[pThis->tVars.farray.deqhead]; + + pThis->tVars.farray.deqhead++; + if (pThis->tVars.farray.deqhead == pThis->iMaxQueueSize) + pThis->tVars.farray.deqhead = 0; + + RETiRet; +} + + +static rsRetVal qDelFixedArray(qqueue_t *pThis) +{ + DEFiRet; + + assert(pThis != NULL); + + pThis->tVars.farray.head++; + if (pThis->tVars.farray.head == pThis->iMaxQueueSize) + pThis->tVars.farray.head = 0; + + RETiRet; +} + + +/* -------------------- linked list -------------------- */ + + +static rsRetVal qConstructLinkedList(qqueue_t *pThis) +{ + DEFiRet; + + assert(pThis != NULL); + + pThis->tVars.linklist.pDeqRoot = NULL; + pThis->tVars.linklist.pDelRoot = NULL; + pThis->tVars.linklist.pLast = NULL; + + qqueueChkIsDA(pThis); + + RETiRet; +} + + +static rsRetVal qDestructLinkedList(qqueue_t __attribute__((unused)) *pThis) +{ + DEFiRet; + + queueDrain(pThis); /* discard any remaining queue entries */ + + /* with the linked list type, there is nothing left to do here. The + * reason is that there are no dynamic elements for the list itself. + */ + + RETiRet; +} + +static rsRetVal qAddLinkedList(qqueue_t *pThis, smsg_t* pMsg) +{ + qLinkedList_t *pEntry; + DEFiRet; + + CHKmalloc((pEntry = (qLinkedList_t*) malloc(sizeof(qLinkedList_t)))); + + pEntry->pNext = NULL; + pEntry->pMsg = pMsg; + + if(pThis->tVars.linklist.pDelRoot == NULL) { + pThis->tVars.linklist.pDelRoot = pThis->tVars.linklist.pDeqRoot = pThis->tVars.linklist.pLast + = pEntry; + } else { + pThis->tVars.linklist.pLast->pNext = pEntry; + pThis->tVars.linklist.pLast = pEntry; + } + + if(pThis->tVars.linklist.pDeqRoot == NULL) { + pThis->tVars.linklist.pDeqRoot = pEntry; + } + +finalize_it: + RETiRet; +} + + +static rsRetVal qDeqLinkedList(qqueue_t *pThis, smsg_t **ppMsg) +{ + qLinkedList_t *pEntry; + DEFiRet; + + pEntry = pThis->tVars.linklist.pDeqRoot; + if (pEntry != NULL) { + *ppMsg = pEntry->pMsg; + pThis->tVars.linklist.pDeqRoot = pEntry->pNext; + } else { + /* Check and return NULL for linklist.pDeqRoot */ + dbgprintf("qDeqLinkedList: pDeqRoot is NULL!\n"); + *ppMsg = NULL; + pThis->tVars.linklist.pDeqRoot = NULL; + } + + RETiRet; +} + + +static rsRetVal qDelLinkedList(qqueue_t *pThis) +{ + qLinkedList_t *pEntry; + DEFiRet; + + pEntry = pThis->tVars.linklist.pDelRoot; + + if(pThis->tVars.linklist.pDelRoot == pThis->tVars.linklist.pLast) { + pThis->tVars.linklist.pDelRoot = pThis->tVars.linklist.pDeqRoot = pThis->tVars.linklist.pLast = NULL; + } else { + pThis->tVars.linklist.pDelRoot = pEntry->pNext; + } + + free(pEntry); + + RETiRet; +} + + +/* -------------------- disk -------------------- */ + + +/* The following function is used to "save" ourself from being killed by + * a fatally failed disk queue. A fatal failure is, for example, if no + * data can be read or written. In that case, the disk support is disabled, + * with all on-disk structures kept as-is as much as possible. However, + * we do not really stop or destruct the in-memory disk queue object. + * Practice has shown that this may cause races during destruction which + * themselfs can lead to segfault. So we prefer to was some ressources by + * keeping the queue active. + * Instead, the queue is switched to direct mode, so that at least + * some processing can happen. Of course, this may still have lots of + * undesired side-effects, but is probably better than aborting the + * syslogd. Note that this function *must* succeed in one way or another, as + * we can not recover from failure here. But it may emit different return + * states, which can trigger different processing in the higher layers. + * rgerhards, 2011-05-03 + */ +static rsRetVal +queueSwitchToEmergencyMode(qqueue_t *pThis, rsRetVal initiatingError) +{ + pThis->iQueueSize = 0; + pThis->nLogDeq = 0; + + 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->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; + pThis->pqParent->pqDA = NULL; + /* This may have undesired side effects, not sure if I really evaluated + * all. So you know where to look at if you come to this point during + * troubleshooting ;) -- rgerhards, 2011-05-03 + */ + } + + LogError(0, initiatingError, "fatal error on disk queue '%s', " + "emergency switch to direct mode", obj.GetName((obj_t*) pThis)); + return RS_RET_ERR_QUEUE_EMERGENCY; +} + + +static rsRetVal +qqueueLoadPersStrmInfoFixup(strm_t *pStrm, qqueue_t __attribute__((unused)) *pThis) +{ + DEFiRet; + ISOBJ_TYPE_assert(pStrm, strm); + ISOBJ_TYPE_assert(pThis, qqueue); + CHKiRet(strm.SetDir(pStrm, pThis->pszSpoolDir, pThis->lenSpoolDir)); + CHKiRet(strm.SetbSync(pStrm, pThis->bSyncQueueFiles)); +finalize_it: + RETiRet; +} + + +/* The method loads the persistent queue information. + * rgerhards, 2008-01-11 + */ +static rsRetVal +qqueueTryLoadPersistedInfo(qqueue_t *pThis) +{ + DEFiRet; + strm_t *psQIF = NULL; + struct stat stat_buf; + + ISOBJ_TYPE_assert(pThis, qqueue); + + /* check if the file exists */ + if(stat((char*) pThis->pszQIFNam, &stat_buf) == -1) { + if(errno == ENOENT) { + DBGOPRINT((obj_t*) pThis, "clean startup, no .qi file found\n"); + ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); + } else { + LogError(errno, RS_RET_IO_ERROR, "queue: %s: error %d could not access .qi file", + obj.GetName((obj_t*) pThis), errno); + ABORT_FINALIZE(RS_RET_IO_ERROR); + } + } + + /* If we reach this point, we have a .qi file */ + + CHKiRet(strm.Construct(&psQIF)); + CHKiRet(strm.SettOperationsMode(psQIF, STREAMMODE_READ)); + CHKiRet(strm.SetsType(psQIF, STREAMTYPE_FILE_SINGLE)); + CHKiRet(strm.SetFName(psQIF, pThis->pszQIFNam, pThis->lenQIFNam)); + CHKiRet(strm.ConstructFinalize(psQIF)); + + /* first, we try to read the property bag for ourselfs */ + CHKiRet(obj.DeserializePropBag((obj_t*) pThis, psQIF)); + + /* then the stream objects (same order as when persisted!) */ + CHKiRet(obj.Deserialize(&pThis->tVars.disk.pWrite, (uchar*) "strm", psQIF, + (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); + CHKiRet(obj.Deserialize(&pThis->tVars.disk.pReadDel, (uchar*) "strm", psQIF, + (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); + /* create a duplicate for the read "pointer". */ + CHKiRet(strm.Dup(pThis->tVars.disk.pReadDel, &pThis->tVars.disk.pReadDeq)); + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDeq, 0)); /* deq must NOT delete the files! */ + CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pReadDeq)); + /* if we use a crypto provider, we need to amend the objects with it's info */ + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pWrite, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pWrite, pThis->cryprovData)); + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDeq, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDeq, pThis->cryprovData)); + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDel, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDel, pThis->cryprovData)); + } + + CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pWrite)); + CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pReadDel)); + CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pReadDeq)); + + /* OK, we could successfully read the file, so we now can request that it be + * deleted when we are done with the persisted information. + */ + pThis->bNeedDelQIF = 1; + LogMsg(0, RS_RET_OK, LOG_INFO, "%s: queue files exist on disk, re-starting with " + "%d messages. This will keep the disk queue file open, details: " + "https://rainer.gerhards.net/2013/07/rsyslog-why-disk-assisted-queues-keep-a-file-open.html", + objGetName((obj_t*) pThis), getLogicalQueueSize(pThis)); + +finalize_it: + if(psQIF != NULL) + strm.Destruct(&psQIF); + + if(iRet != RS_RET_OK) { + DBGOPRINT((obj_t*) pThis, "state %d reading .qi file - can not read persisted info (if any)\n", + iRet); + } + + RETiRet; +} + + +/* disk queue constructor. + * Note that we use a file limit of 10,000,000 files. That number should never pose a + * problem. If so, I guess the user has a design issue... But of course, the code can + * always be changed (though it would probably be more appropriate to increase the + * allowed file size at this point - that should be a config setting... + * rgerhards, 2008-01-10 + */ +static rsRetVal qConstructDisk(qqueue_t *pThis) +{ + DEFiRet; + int bRestarted = 0; + + assert(pThis != NULL); + + /* and now check if there is some persistent information that needs to be read in */ + iRet = qqueueTryLoadPersistedInfo(pThis); + if(iRet == RS_RET_OK) + bRestarted = 1; + else if(iRet != RS_RET_FILE_NOT_FOUND) + FINALIZE; + + if(bRestarted == 1) { + ; + } else { + CHKiRet(strm.Construct(&pThis->tVars.disk.pWrite)); + CHKiRet(strm.SetbSync(pThis->tVars.disk.pWrite, pThis->bSyncQueueFiles)); + CHKiRet(strm.SetDir(pThis->tVars.disk.pWrite, pThis->pszSpoolDir, pThis->lenSpoolDir)); + CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pWrite, 10000000)); + CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pWrite, STREAMMODE_WRITE)); + CHKiRet(strm.SetsType(pThis->tVars.disk.pWrite, STREAMTYPE_FILE_CIRCULAR)); + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pWrite, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pWrite, pThis->cryprovData)); + } + CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pWrite)); + + CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDeq)); + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDeq, 0)); + CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDeq, pThis->pszSpoolDir, pThis->lenSpoolDir)); + CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDeq, 10000000)); + CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDeq, STREAMMODE_READ)); + CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDeq, STREAMTYPE_FILE_CIRCULAR)); + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDeq, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDeq, pThis->cryprovData)); + } + CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pReadDeq)); + + CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDel)); + CHKiRet(strm.SetbSync(pThis->tVars.disk.pReadDel, pThis->bSyncQueueFiles)); + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDel, 1)); + CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDel, pThis->pszSpoolDir, pThis->lenSpoolDir)); + CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDel, 10000000)); + CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDel, STREAMMODE_READ)); + CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDel, STREAMTYPE_FILE_CIRCULAR)); + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDel, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDel, pThis->cryprovData)); + } + CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pReadDel)); + + CHKiRet(strm.SetFName(pThis->tVars.disk.pWrite, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(strm.SetFName(pThis->tVars.disk.pReadDeq, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(strm.SetFName(pThis->tVars.disk.pReadDel, pThis->pszFilePrefix, pThis->lenFilePrefix)); + } + + /* now we set (and overwrite in case of a persisted restart) some parameters which + * should always reflect the current configuration variables. Be careful by doing so, + * for example file name generation must not be changed as that would break the + * ability to read existing queue files. -- rgerhards, 2008-01-12 + */ + CHKiRet(strm.SetiMaxFileSize(pThis->tVars.disk.pWrite, pThis->iMaxFileSize)); + CHKiRet(strm.SetiMaxFileSize(pThis->tVars.disk.pReadDeq, pThis->iMaxFileSize)); + CHKiRet(strm.SetiMaxFileSize(pThis->tVars.disk.pReadDel, pThis->iMaxFileSize)); + +finalize_it: + RETiRet; +} + + +static rsRetVal qDestructDisk(qqueue_t *pThis) +{ + DEFiRet; + + assert(pThis != NULL); + + free(pThis->pszQIFNam); + if(pThis->tVars.disk.pWrite != NULL) { + int64 currOffs; + strm.GetCurrOffset(pThis->tVars.disk.pWrite, &currOffs); + if(currOffs == 0) { + /* if no data is present, we can (and must!) delete this + * file. Else we can leave garbagge after termination. + */ + strm.SetbDeleteOnClose(pThis->tVars.disk.pWrite, 1); + } + strm.Destruct(&pThis->tVars.disk.pWrite); + } + if(pThis->tVars.disk.pReadDeq != NULL) + strm.Destruct(&pThis->tVars.disk.pReadDeq); + if(pThis->tVars.disk.pReadDel != NULL) + strm.Destruct(&pThis->tVars.disk.pReadDel); + + RETiRet; +} + +static rsRetVal ATTR_NONNULL(1,2) +qAddDisk(qqueue_t *const pThis, smsg_t* pMsg) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, qqueue); + ISOBJ_TYPE_assert(pMsg, msg); + number_t nWriteCount; + const int oldfile = strmGetCurrFileNum(pThis->tVars.disk.pWrite); + + CHKiRet(strm.SetWCntr(pThis->tVars.disk.pWrite, &nWriteCount)); + CHKiRet((objSerialize(pMsg))(pMsg, pThis->tVars.disk.pWrite)); + CHKiRet(strm.Flush(pThis->tVars.disk.pWrite)); + CHKiRet(strm.SetWCntr(pThis->tVars.disk.pWrite, NULL)); /* no more counting for now... */ + + pThis->tVars.disk.sizeOnDisk += nWriteCount; + + /* we have enqueued the user element to disk. So we now need to destruct + * the in-memory representation. The instance will be re-created upon + * dequeue. -- rgerhards, 2008-07-09 + */ + msgDestruct(&pMsg); + + DBGOPRINT((obj_t*) pThis, "write wrote %lld octets to disk, queue disk size now %lld octets, EnqOnly:%d\n", + nWriteCount, pThis->tVars.disk.sizeOnDisk, pThis->bEnqOnly); + + /* Did we have a change in the on-disk file? If so, we + * should do a "robustness sync" of the .qi file to guard + * against the most harsh consequences of kill -9 and power off. + */ + int newfile; + newfile = strmGetCurrFileNum(pThis->tVars.disk.pWrite); + if(newfile != oldfile) { + DBGOPRINT((obj_t*) pThis, "current to-be-written-to file has changed from " + "number %d to number %d - requiring a .qi write for robustness\n", + oldfile, newfile); + pThis->tVars.disk.nForcePersist = 2; + } + +finalize_it: + RETiRet; +} + + +static rsRetVal +qDeqDisk(qqueue_t *pThis, smsg_t **ppMsg) +{ + DEFiRet; + iRet = objDeserializeWithMethods(ppMsg, (uchar*) "msg", 3, + pThis->tVars.disk.pReadDeq, NULL, + NULL, msgConstructForDeserializer, NULL, MsgDeserialize); + if(iRet != RS_RET_OK) { + LogError(0, iRet, "%s: qDeqDisk error happened at around offset %lld", + obj.GetName((obj_t*)pThis), + (long long) pThis->tVars.disk.pReadDeq->iCurrOffs); + } + RETiRet; +} + + +/* -------------------- direct (no queueing) -------------------- */ +static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis) +{ + return RS_RET_OK; +} + + +static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis) +{ + return RS_RET_OK; +} + +static rsRetVal qAddDirectWithWti(qqueue_t *pThis, smsg_t* pMsg, wti_t *pWti) +{ + batch_t singleBatch; + batch_obj_t batchObj; + batch_state_t batchState = BATCH_STATE_RDY; + DEFiRet; + + //TODO: init batchObj (states _OK and new fields -- CHECK) + 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 + */ + memset(&batchObj, 0, sizeof(batch_obj_t)); + memset(&singleBatch, 0, sizeof(batch_t)); + batchObj.pMsg = pMsg; + singleBatch.nElem = 1; /* there always is only one in direct mode */ + singleBatch.pElem = &batchObj; + singleBatch.eltState = &batchState; + iRet = pThis->pConsumer(pThis->pAction, &singleBatch, pWti); + msgDestruct(&pMsg); + + RETiRet; +} + +/* 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. + */ +static rsRetVal +qAddDirect(qqueue_t *pThis, smsg_t* pMsg) +{ + wti_t *pWti; + DEFiRet; + + pWti = wtiGetDummy(); + pWti->pbShutdownImmediate = &pThis->bShutdownImmediate; + iRet = qAddDirectWithWti(pThis, pMsg, pWti); + RETiRet; +} + + +/* --------------- end type-specific handlers -------------------- */ + + +/* generic code to add a queue entry + * We use some specific code to most efficiently support direct mode + * queues. This is justified in spite of the gain and the need to do some + * things truely different. -- rgerhards, 2008-02-12 + */ +static rsRetVal +qqueueAdd(qqueue_t *pThis, smsg_t *pMsg) +{ + DEFiRet; + + assert(pThis != NULL); + + static int msgCnt = 0; + + if(pThis->iSmpInterval > 0) + { + msgCnt = (msgCnt + 1) % (pThis->iSmpInterval); + if(msgCnt != 0) + { + msgDestruct(&pMsg); + goto finalize_it; + } + } + + CHKiRet(pThis->qAdd(pThis, pMsg)); + + if(pThis->qType != QUEUETYPE_DIRECT) { + ATOMIC_INC(&pThis->iQueueSize, &pThis->mutQueueSize); +# ifdef ENABLE_IMDIAG +# ifdef HAVE_ATOMIC_BUILTINS + /* mutex is never used due to conditional compilation */ + ATOMIC_INC(&iOverallQueueSize, &NULL); +# else + ++iOverallQueueSize; /* racy, but we can't wait for a mutex! */ +# endif +# endif + } + +finalize_it: + RETiRet; +} + + +/* generic code to dequeue a queue entry + */ +static rsRetVal +qqueueDeq(qqueue_t *pThis, smsg_t **ppMsg) +{ + DEFiRet; + + assert(pThis != NULL); + + /* we do NOT abort if we encounter an error, because otherwise the queue + * will not be decremented, what will most probably result in an endless loop. + * If we decrement, however, we may lose a message. But that is better than + * losing the whole process because it loops... -- rgerhards, 2008-01-03 + */ + iRet = pThis->qDeq(pThis, ppMsg); + ATOMIC_INC(&pThis->nLogDeq, &pThis->mutLogDeq); + + DBGOPRINT((obj_t*) pThis, "entry deleted, size now log %d, phys %d entries\n", + getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); + + RETiRet; +} + + +/* Try to shut down regular and DA queue workers, within the queue timeout + * period. That means processing continues as usual. This is the expected + * usual case, where during shutdown those messages remaining are being + * processed. At this point, it is acceptable that the queue can not be + * fully depleted, that case is handled in the next step. During this phase, + * we first shut down the main queue DA worker to prevent new data to arrive + * at the DA queue, and then we ask the regular workers of both the Regular + * and DA queue to try complete processing. + * rgerhards, 2009-10-14 + */ +static rsRetVal ATTR_NONNULL(1) +tryShutdownWorkersWithinQueueTimeout(qqueue_t *const pThis) +{ + struct timespec tTimeout; + rsRetVal iRetLocal; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + assert(pThis->pqParent == NULL); /* detect invalid calling sequence */ + + if(pThis->bIsDA) { + /* We need to lock the mutex, as otherwise we may have a race that prevents + * us from awaking the DA worker. */ + d_pthread_mutex_lock(pThis->mut); + + /* tell regular queue DA worker to stop shuffling messages to DA queue... */ + DBGOPRINT((obj_t*) pThis, "setting EnqOnly mode for DA worker\n"); + pThis->pqDA->bEnqOnly = 1; + wtpSetState(pThis->pWtpDA, wtpState_SHUTDOWN_IMMEDIATE); + wtpAdviseMaxWorkers(pThis->pWtpDA, 1, DENY_WORKER_START_DURING_SHUTDOWN); + DBGOPRINT((obj_t*) pThis, "awoke DA worker, told it to shut down.\n"); + + /* also tell the DA queue worker to shut down, so that it already knows... */ + wtpSetState(pThis->pqDA->pWtpReg, wtpState_SHUTDOWN); + wtpAdviseMaxWorkers(pThis->pqDA->pWtpReg, 1, DENY_WORKER_START_DURING_SHUTDOWN); + /* awake its lone worker */ + DBGOPRINT((obj_t*) pThis, "awoke DA queue regular worker, told it to shut down when done.\n"); + + d_pthread_mutex_unlock(pThis->mut); + } + + + /* first calculate absolute timeout - we need the absolute value here, because we need to coordinate + * shutdown of both the regular and DA queue on *the same* timeout. + */ + timeoutComp(&tTimeout, pThis->toQShutdown); + DBGOPRINT((obj_t*) pThis, "trying shutdown of regular workers\n"); + iRetLocal = wtpShutdownAll(pThis->pWtpReg, wtpState_SHUTDOWN, &tTimeout); + if(iRetLocal == RS_RET_TIMED_OUT) { + LogMsg(0, RS_RET_TIMED_OUT, LOG_INFO, + "%s: regular queue shutdown timed out on primary queue " + "(this is OK, timeout was %d)", + objGetName((obj_t*) pThis), pThis->toQShutdown); + } else { + DBGOPRINT((obj_t*) pThis, "regular queue workers shut down.\n"); + } + + /* OK, the worker for the regular queue is processed, on the the DA queue regular worker. */ + if(pThis->pqDA != NULL) { + DBGOPRINT((obj_t*) pThis, "we have a DA queue (0x%lx), requesting its shutdown.\n", + qqueueGetID(pThis->pqDA)); + /* we use the same absolute timeout as above, so we do not use more than the configured + * timeout interval! + */ + DBGOPRINT((obj_t*) pThis, "trying shutdown of regular worker of DA queue\n"); + iRetLocal = wtpShutdownAll(pThis->pqDA->pWtpReg, wtpState_SHUTDOWN, &tTimeout); + if(iRetLocal == RS_RET_TIMED_OUT) { + LogMsg(0, RS_RET_TIMED_OUT, LOG_INFO, + "%s: regular queue shutdown timed out on DA queue (this is OK, " + "timeout was %d)", objGetName((obj_t*) pThis), pThis->toQShutdown); + } else { + DBGOPRINT((obj_t*) pThis, "DA queue worker shut down.\n"); + } + } + + RETiRet; +} + + +/* Try to shut down regular and DA queue workers, within the action timeout + * period. This aborts processing, but at the end of the current action, in + * a well-defined manner. During this phase, we terminate all three worker + * pools, including the regular queue DA worker if it not yet has terminated. + * Not finishing processing all messages is OK (and expected) at this stage + * (they may be preserved later, depending * on bSaveOnShutdown setting). + * rgerhards, 2009-10-14 + */ +static rsRetVal +tryShutdownWorkersWithinActionTimeout(qqueue_t *pThis) +{ + struct timespec tTimeout; + rsRetVal iRetLocal; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + assert(pThis->pqParent == NULL); /* detect invalid calling sequence */ + + /* instruct workers to finish ASAP, even if still work exists */ + DBGOPRINT((obj_t*) pThis, "trying to shutdown workers within Action Timeout"); + DBGOPRINT((obj_t*) pThis, "setting EnqOnly mode\n"); + pThis->bEnqOnly = 1; + pThis->bShutdownImmediate = 1; + /* now DA queue */ + if(pThis->bIsDA) { + pThis->pqDA->bEnqOnly = 1; + pThis->pqDA->bShutdownImmediate = 1; + } + + /* now give the queue workers a last chance to gracefully shut down (based on action timeout setting) */ + timeoutComp(&tTimeout, pThis->toActShutdown); + DBGOPRINT((obj_t*) pThis, "trying immediate shutdown of regular workers (if any)\n"); + iRetLocal = wtpShutdownAll(pThis->pWtpReg, wtpState_SHUTDOWN_IMMEDIATE, &tTimeout); + if(iRetLocal == RS_RET_TIMED_OUT) { + LogMsg(0, RS_RET_TIMED_OUT, LOG_INFO, + "%s: immediate shutdown timed out on primary queue (this is acceptable and " + "triggers cancellation)", objGetName((obj_t*) pThis)); + } else if(iRetLocal != RS_RET_OK) { + LogMsg(0, iRetLocal, LOG_WARNING, + "%s: potential internal error: unexpected return state after trying " + "immediate shutdown of the primary queue in disk save mode. " + "Continuing, but results are unpredictable", objGetName((obj_t*) pThis)); + } + + if(pThis->pqDA != NULL) { + /* and now the same for the DA queue */ + DBGOPRINT((obj_t*) pThis, "trying immediate shutdown of DA queue workers\n"); + iRetLocal = wtpShutdownAll(pThis->pqDA->pWtpReg, wtpState_SHUTDOWN_IMMEDIATE, &tTimeout); + if(iRetLocal == RS_RET_TIMED_OUT) { + LogMsg(0, RS_RET_TIMED_OUT, LOG_INFO, + "%s: immediate shutdown timed out on DA queue (this is acceptable and " + "triggers cancellation)", objGetName((obj_t*) pThis)); + } else if(iRetLocal != RS_RET_OK) { + LogMsg(0, iRetLocal, LOG_WARNING, + "%s: potential internal error: unexpected return state after trying " + "immediate shutdown of the DA queue in disk save mode. " + "Continuing, but results are unpredictable", objGetName((obj_t*) pThis)); + } + + /* and now we need to terminate the DA worker itself. We always grant it a 100ms timeout, + * which should be sufficient and usually not be required (it is expected to have finished + * long before while we were processing the queue timeout in shutdown phase 1). + * rgerhards, 2009-10-14 + */ + timeoutComp(&tTimeout, 100); + DBGOPRINT((obj_t*) pThis, "trying regular shutdown of main queue DA worker pool\n"); + iRetLocal = wtpShutdownAll(pThis->pWtpDA, wtpState_SHUTDOWN_IMMEDIATE, &tTimeout); + if(iRetLocal == RS_RET_TIMED_OUT) { + LogMsg(0, iRetLocal, LOG_WARNING, + "%s: shutdown timed out on main queue DA worker pool " + "(this is not good, but possibly OK)", + objGetName((obj_t*) pThis)); + } else { + DBGOPRINT((obj_t*) pThis, "main queue DA worker pool shut down.\n"); + } + } + + RETiRet; +} + + +/* This function cancels all remaining regular workers for both the main and the DA + * queue. + * rgerhards, 2009-05-29 + */ +static rsRetVal +cancelWorkers(qqueue_t *pThis) +{ + rsRetVal iRetLocal; + DEFiRet; + + assert(pThis->qType != QUEUETYPE_DIRECT); + + /* Now queue workers should have terminated. If not, we need to cancel them as we have applied + * all timeout setting. If any worker in any queue still executes, its consumer is possibly + * long-running and cancelling is the only way to get rid of it. + */ + DBGOPRINT((obj_t*) pThis, "checking to see if we need to cancel any worker threads of the primary queue\n"); + iRetLocal = wtpCancelAll(pThis->pWtpReg, objGetName((obj_t*) pThis)); + /* ^-- returns immediately if all threads already have terminated */ + if(iRetLocal != RS_RET_OK) { + DBGOPRINT((obj_t*) pThis, "unexpected iRet state %d trying to cancel primary queue worker " + "threads, continuing, but results are unpredictable\n", iRetLocal); + } + + /* ... and now the DA queue, if it exists (should always be after the primary one) */ + if(pThis->pqDA != NULL) { + DBGOPRINT((obj_t*) pThis, "checking to see if we need to cancel any worker threads of " + "the DA queue\n"); + iRetLocal = wtpCancelAll(pThis->pqDA->pWtpReg, objGetName((obj_t*) pThis)); + /* returns immediately if all threads already have terminated */ + if(iRetLocal != RS_RET_OK) { + DBGOPRINT((obj_t*) pThis, "unexpected iRet state %d trying to cancel DA queue worker " + "threads, continuing, but results are unpredictable\n", iRetLocal); + } + + /* finally, we cancel the main queue's DA worker pool, if it still is running. It may be + * restarted later to persist the queue. But we stop it, because otherwise we get into + * big trouble when resetting the logical dequeue pointer. This operation can only be + * done when *no* worker is running. So time for a shutdown... -- rgerhards, 2009-05-28 + */ + DBGOPRINT((obj_t*) pThis, "checking to see if main queue DA worker pool needs to be cancelled\n"); + wtpCancelAll(pThis->pWtpDA, objGetName((obj_t*) pThis)); + /* returns immediately if all threads already have terminated */ + } + + RETiRet; +} + + +/* This function shuts down all worker threads and waits until they + * have terminated. If they timeout, they are cancelled. + * rgerhards, 2008-01-24 + * Please note that this function shuts down BOTH the parent AND the child queue + * in DA case. This is necessary because their timeouts are tightly coupled. Most + * importantly, the timeouts would be applied twice (or logic be extremely + * complex) if each would have its own shutdown. The function does not self check + * this condition - the caller must make sure it is not called with a parent. + * rgerhards, 2009-05-26: we do NO longer persist the queue here if bSaveOnShutdown + * is set. This must be handled by the caller. Not doing that cleans up the queue + * shutdown considerably. Also, older engines had a potential hang condition when + * the DA queue was already started and the DA worker configured for infinite + * retries and the action was during retry processing. This was a design issue, + * which is solved as of now. Note that the shutdown now may take a little bit + * longer, because we no longer can persist the queue in parallel to waiting + * on worker timeouts. + */ +rsRetVal ATTR_NONNULL(1) +qqueueShutdownWorkers(qqueue_t *const pThis) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, qqueue); + + if(pThis->qType == QUEUETYPE_DIRECT) { + FINALIZE; + } + + assert(pThis->pqParent == NULL); /* detect invalid calling sequence */ + + DBGOPRINT((obj_t*) pThis, "initiating worker thread shutdown sequence %p\n", pThis); + + CHKiRet(tryShutdownWorkersWithinQueueTimeout(pThis)); + + pthread_mutex_lock(pThis->mut); + int physQueueSize; + physQueueSize = getPhysicalQueueSize(pThis); + pthread_mutex_unlock(pThis->mut); + if(physQueueSize > 0) { + CHKiRet(tryShutdownWorkersWithinActionTimeout(pThis)); + } + + CHKiRet(cancelWorkers(pThis)); + + /* ... finally ... all worker threads have terminated :-) + * Well, more precisely, they *are in termination*. Some cancel cleanup handlers + * may still be running. Note that the main queue's DA worker may still be running. + */ + DBGOPRINT((obj_t*) pThis, "worker threads terminated, remaining queue size log %d, phys %d.\n", + getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); + +finalize_it: + RETiRet; +} + +/* Constructor for the queue object + * This constructs the data structure, but does not yet start the queue. That + * is done by queueStart(). The reason is that we want to give the caller a chance + * 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*)) +{ + DEFiRet; + qqueue_t *pThis; + const uchar *const workDir = glblGetWorkDirRaw(ourConf); + + assert(ppThis != NULL); + assert(pConsumer != NULL); + assert(iWorkerThreads >= 0); + + CHKmalloc(pThis = (qqueue_t *)calloc(1, sizeof(qqueue_t))); + + /* we have an object, so let's fill the properties */ + objConstructSetObjInfo(pThis); + + if(workDir != NULL) { + if((pThis->pszSpoolDir = ustrdup(workDir)) == NULL) + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir); + } + /* set some water marks so that we have useful defaults if none are set specifically */ + pThis->iFullDlyMrk = -1; + pThis->iLightDlyMrk = -1; + pThis->iMaxFileSize = 1024 * 1024; /* default is 1 MiB */ + pThis->iQueueSize = 0; + pThis->nLogDeq = 0; + pThis->useCryprov = 0; + pThis->takeFlowCtlFromMsg = 0; + pThis->iMaxQueueSize = iMaxQueueSize; + pThis->pConsumer = pConsumer; + pThis->iNumWorkerThreads = iWorkerThreads; + pThis->iDeqtWinToHr = 25; /* disable time-windowed dequeuing by default */ + pThis->iDeqBatchSize = 8; /* conservative default, should still provide good performance */ + pThis->iMinDeqBatchSize = 0; /* conservative default, should still provide good performance */ + pThis->isRunning = 0; + + pThis->pszFilePrefix = NULL; + pThis->qType = qType; + + + INIT_ATOMIC_HELPER_MUT(pThis->mutQueueSize); + INIT_ATOMIC_HELPER_MUT(pThis->mutLogDeq); + +finalize_it: + OBJCONSTRUCT_CHECK_SUCCESS_AND_CLEANUP + RETiRet; +} + + +/* set default inside queue object suitable for action queues. + * This shall be called directly after queue construction. This functions has + * been added in support of the new v6 config system. It expect properly pre-initialized + * objects, but we need to differentiate between ruleset main and action queues. + * In order to avoid unnecessary complexity, we provide the necessary defaults + * via specific function calls. + */ +void +qqueueSetDefaultsActionQueue(qqueue_t *pThis) +{ + pThis->qType = QUEUETYPE_DIRECT; /* type of the main message queue above */ + pThis->iMaxQueueSize = 1000; /* size of the main message queue above */ + pThis->iDeqBatchSize = 128; /* default batch size */ + pThis->iMinDeqBatchSize = 0; + pThis->toMinDeqBatchSize = 1000; + pThis->iHighWtrMrk = -1; /* high water mark for disk-assisted queues */ + pThis->iLowWtrMrk = -1; /* low water mark for disk-assisted queues */ + pThis->iDiscardMrk = -1; /* begin to discard messages */ + pThis->iDiscardSeverity = 8; /* turn off */ + pThis->iNumWorkerThreads = 1; /* number of worker threads for the mm queue above */ + pThis->iMaxFileSize = 1024*1024; + pThis->iPersistUpdCnt = 0; /* persist queue info every n updates */ + pThis->bSyncQueueFiles = 0; + pThis->toQShutdown = loadConf->globals.actq_dflt_toQShutdown; /* queue shutdown */ + pThis->toActShutdown = loadConf->globals.actq_dflt_toActShutdown; /* action shutdown (in phase 2) */ + pThis->toEnq = loadConf->globals.actq_dflt_toEnq; /* timeout for queue enque */ + pThis->toWrkShutdown = loadConf->globals.actq_dflt_toWrkShutdown; /* timeout for worker thread shutdown */ + pThis->iMinMsgsPerWrkr = -1; /* minimum messages per worker needed to start a new one */ + pThis->bSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */ + pThis->sizeOnDiskMax = 0; /* unlimited */ + pThis->iDeqSlowdown = 0; + pThis->iDeqtWinFromHr = 0; + pThis->iDeqtWinToHr = 25; /* disable time-windowed dequeuing by default */ + pThis->iSmpInterval = 0; /* disable sampling */ +} + + +/* set defaults inside queue object suitable for main/ruleset queues. + * See queueSetDefaultsActionQueue() for more details and background. + */ +void +qqueueSetDefaultsRulesetQueue(qqueue_t *pThis) +{ + pThis->qType = QUEUETYPE_FIXED_ARRAY; /* type of the main message queue above */ + pThis->iMaxQueueSize = 50000; /* size of the main message queue above */ + pThis->iDeqBatchSize = 1024; /* default batch size */ + pThis->iMinDeqBatchSize = 0; + pThis->toMinDeqBatchSize = 1000; + pThis->iHighWtrMrk = -1; /* high water mark for disk-assisted queues */ + pThis->iLowWtrMrk = -1; /* low water mark for disk-assisted queues */ + pThis->iDiscardMrk = -1; /* begin to discard messages */ + pThis->iDiscardSeverity = 8; /* turn off */ + pThis->iNumWorkerThreads = 1; /* number of worker threads for the mm queue above */ + pThis->iMaxFileSize = 16*1024*1024; + pThis->iPersistUpdCnt = 0; /* persist queue info every n updates */ + pThis->bSyncQueueFiles = 0; + pThis->toQShutdown = ourConf->globals.ruleset_dflt_toQShutdown; + pThis->toActShutdown = ourConf->globals.ruleset_dflt_toActShutdown; + pThis->toEnq = ourConf->globals.ruleset_dflt_toEnq; + pThis->toWrkShutdown = ourConf->globals.ruleset_dflt_toWrkShutdown; + pThis->iMinMsgsPerWrkr = -1; /* minimum messages per worker needed to start a new one */ + pThis->bSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */ + pThis->sizeOnDiskMax = 0; /* unlimited */ + pThis->iDeqSlowdown = 0; + pThis->iDeqtWinFromHr = 0; + pThis->iDeqtWinToHr = 25; /* disable time-windowed dequeuing by default */ + pThis->iSmpInterval = 0; /* disable sampling */ +} + + +/* This function checks if the provided message shall be discarded and does so, if needed. + * In DA mode, we do not discard any messages as we assume the disk subsystem is fast enough to + * provide real-time creation of spool files. + * Note: cached copies of iQueueSize is provided so that no mutex locks are required. + * The caller must have obtained them while the mutex was locked. Of course, these values may no + * longer be current, but that is OK for the discard check. At worst, the message is either processed + * or discarded when it should not have been. As discarding is in itself somewhat racy and erratic, + * that is no problems for us. This function MUST NOT lock the queue mutex, it could result in + * deadlocks! + * If the message is discarded, it can no longer be processed by the caller. So be sure to check + * the return state! + * rgerhards, 2008-01-24 + */ +static int qqueueChkDiscardMsg(qqueue_t *pThis, int iQueueSize, smsg_t *pMsg) +{ + DEFiRet; + rsRetVal iRetLocal; + int iSeverity; + + ISOBJ_TYPE_assert(pThis, qqueue); + + if(pThis->iDiscardMrk > 0 && iQueueSize >= pThis->iDiscardMrk) { + iRetLocal = MsgGetSeverity(pMsg, &iSeverity); + if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) { + DBGOPRINT((obj_t*) pThis, "queue nearly full (%d entries), discarded severity %d message\n", + iQueueSize, iSeverity); + STATSCOUNTER_INC(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd); + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_QUEUE_FULL); + } else { + DBGOPRINT((obj_t*) pThis, "queue nearly full (%d entries), but could not drop msg " + "(iRet: %d, severity %d)\n", iQueueSize, iRetLocal, iSeverity); + } + } + +finalize_it: + RETiRet; +} + + +/* Finally remove n elements from the queue store. + */ +static rsRetVal ATTR_NONNULL(1) +DoDeleteBatchFromQStore(qqueue_t *const pThis, const int nElem) +{ + int i; + off64_t bytesDel = 0; /* keep CLANG static anaylzer happy */ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + + /* now send delete request to storage driver */ + if(pThis->qType == QUEUETYPE_DISK) { + strmMultiFileSeek(pThis->tVars.disk.pReadDel, pThis->tVars.disk.deqFileNumOut, + pThis->tVars.disk.deqOffs, &bytesDel); + /* We need to correct the on-disk file size. This time it is a bit tricky: + * we free disk space only upon file deletion. So we need to keep track of what we + * have read until we get an out-offset that is lower than the in-offset (which + * indicates file change). Then, we can subtract the whole thing from the on-disk + * size. -- rgerhards, 2008-01-30 + */ + if(bytesDel != 0) { + pThis->tVars.disk.sizeOnDisk -= bytesDel; + DBGOPRINT((obj_t*) pThis, "doDeleteBatch: a %lld octet file has been deleted, now %lld " + "octets disk space used\n", (long long) bytesDel, pThis->tVars.disk.sizeOnDisk); + /* awake possibly waiting enq process */ + pthread_cond_signal(&pThis->notFull); /* we hold the mutex while we are in here! */ + } + } else { /* memory queue */ + for(i = 0 ; i < nElem ; ++i) { + pThis->qDel(pThis); + } + } + + /* iQueueSize is not decremented by qDel(), so we need to do it ourselves */ + ATOMIC_SUB(&pThis->iQueueSize, nElem, &pThis->mutQueueSize); +# ifdef ENABLE_IMDIAG +# ifdef HAVE_ATOMIC_BUILTINS + /* mutex is never used due to conditional compilation */ + ATOMIC_SUB(&iOverallQueueSize, nElem, &NULL); +# else + iOverallQueueSize -= nElem; /* racy, but we can't wait for a mutex! */ +# endif +# endif + ATOMIC_SUB(&pThis->nLogDeq, nElem, &pThis->mutLogDeq); + DBGPRINTF("doDeleteBatch: delete batch from store, new sizes: log %d, phys %d\n", + getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); + ++pThis->deqIDDel; /* one more batch dequeued */ + + if((pThis->qType == QUEUETYPE_DISK) && (bytesDel != 0)) { + qqueuePersist(pThis, QUEUE_CHECKPOINT); /* robustness persist .qi file */ + } + + RETiRet; +} + + +/* remove messages from the physical queue store that are fully processed. This is + * controlled via the to-delete list. + */ +static rsRetVal +DeleteBatchFromQStore(qqueue_t *pThis, batch_t *pBatch) +{ + toDeleteLst_t *pTdl; + qDeqID deqIDDel; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + assert(pBatch != NULL); + +dbgprintf("rger: deleteBatchFromQStore, nElem %d\n", (int) pBatch->nElem); + pTdl = tdlPeek(pThis); /* get current head element */ + if(pTdl == NULL) { /* to-delete list empty */ + DoDeleteBatchFromQStore(pThis, pBatch->nElem); + } else if(pBatch->deqID == pThis->deqIDDel) { + deqIDDel = pThis->deqIDDel; + pTdl = tdlPeek(pThis); + while(pTdl != NULL && deqIDDel == pTdl->deqID) { + DoDeleteBatchFromQStore(pThis, pTdl->nElemDeq); + tdlPop(pThis); + ++deqIDDel; + pTdl = tdlPeek(pThis); + } + /* old entries deleted, now delete current ones... */ + DoDeleteBatchFromQStore(pThis, pBatch->nElem); + } else { + /* can not delete, insert into to-delete list */ + DBGPRINTF("not at head of to-delete list, enqueue %d\n", (int) pBatch->deqID); + CHKiRet(tdlAdd(pThis, pBatch->deqID, pBatch->nElem)); + } + +finalize_it: + RETiRet; +} + + +/* Delete a batch of processed user objects from the queue, which includes + * destructing the objects themself. Any entries not marked as finally + * processed are enqueued again. The new enqueue is necessary because we have a + * rgerhards, 2009-05-13 + */ +static rsRetVal +DeleteProcessedBatch(qqueue_t *pThis, batch_t *pBatch) +{ + int i; + smsg_t *pMsg; + int nEnqueued = 0; + rsRetVal localRet; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + assert(pBatch != NULL); + + for(i = 0 ; i < pBatch->nElem ; ++i) { + pMsg = pBatch->pElem[i].pMsg; + DBGPRINTF("DeleteProcessedBatch: etry %d state %d\n", i, pBatch->eltState[i]); + if( pBatch->eltState[i] == BATCH_STATE_RDY + || pBatch->eltState[i] == BATCH_STATE_SUB) { + localRet = doEnqSingleObj(pThis, eFLOWCTL_NO_DELAY, MsgAddRef(pMsg)); + ++nEnqueued; + if(localRet != RS_RET_OK) { + DBGPRINTF("DeleteProcessedBatch: error %d re-enqueuing unprocessed " + "data element - discarded\n", localRet); + } + } + msgDestruct(&pMsg); + } + + DBGPRINTF("DeleteProcessedBatch: we deleted %d objects and enqueued %d objects\n", i-nEnqueued, nEnqueued); + + if(nEnqueued > 0) + qqueueChkPersist(pThis, nEnqueued); + + iRet = DeleteBatchFromQStore(pThis, pBatch); + + pBatch->nElem = pBatch->nElemDeq = 0; /* reset batch */ // TODO: more fine init, new fields! 2010-06-14 + + RETiRet; +} + + +/* dequeue as many user pointers as are available, until we hit the configured + * upper limit of pointers. Note that this function also deletes all processed + * objects from the previous batch. However, it is perfectly valid that the + * previous batch contained NO objects at all. For example, this happens + * immediately after system startup or when a queue was exhausted and the queue + * worker needed to wait for new data. + * This must only be called when the queue mutex is LOOKED, otherwise serious + * malfunction will happen. + */ +static rsRetVal ATTR_NONNULL() +DequeueConsumableElements(qqueue_t *const pThis, wti_t *const pWti, + int *const piRemainingQueueSize, int *const pSkippedMsgs) +{ + int nDequeued; + int nDiscarded; + int nDeleted; + int iQueueSize; + int keep_running = 1; + struct timespec timeout; + smsg_t *pMsg; + rsRetVal localRet; + DEFiRet; + + nDeleted = pWti->batch.nElemDeq; + DeleteProcessedBatch(pThis, &pWti->batch); + + nDequeued = nDiscarded = 0; + if(pThis->qType == QUEUETYPE_DISK) { + pThis->tVars.disk.deqFileNumIn = strmGetCurrFileNum(pThis->tVars.disk.pReadDeq); + } + + /* work-around clang static analyzer false positive, we need a const value */ + const int iMinDeqBatchSize = pThis->iMinDeqBatchSize; + if(iMinDeqBatchSize > 0) { + timeoutComp(&timeout, pThis->toMinDeqBatchSize);/* get absolute timeout */ + } + + while((iQueueSize = getLogicalQueueSize(pThis)) > 0 && nDequeued < pThis->iDeqBatchSize) { + int rd_fd = -1; + int64_t rd_offs = 0; + int wr_fd = -1; + int64_t wr_offs = 0; + if(pThis->tVars.disk.pReadDeq != NULL) { + rd_fd = strmGetCurrFileNum(pThis->tVars.disk.pReadDeq); + rd_offs = pThis->tVars.disk.pReadDeq->iCurrOffs; + } + if(pThis->tVars.disk.pWrite != NULL) { + wr_fd = strmGetCurrFileNum(pThis->tVars.disk.pWrite); + wr_offs = pThis->tVars.disk.pWrite->iCurrOffs; + } + if(rd_fd != -1 && rd_fd == wr_fd && rd_offs == wr_offs) { + DBGPRINTF("problem on disk queue '%s': " + //"queue size log %d, phys %d, but rd_fd=wr_rd=%d and offs=%lld\n", + "queue size log %d, phys %d, but rd_fd=wr_rd=%d and offs=%" PRId64 "\n", + obj.GetName((obj_t*) pThis), iQueueSize, pThis->iQueueSize, + rd_fd, rd_offs); + *pSkippedMsgs = iQueueSize; +# ifdef ENABLE_IMDIAG + iOverallQueueSize -= iQueueSize; +# endif + pThis->iQueueSize -= iQueueSize; + iQueueSize = 0; + break; + } + + localRet = qqueueDeq(pThis, &pMsg); + if(localRet == RS_RET_FILE_NOT_FOUND) { + DBGPRINTF("fatal error on disk queue '%s': file '%s' " + "not found, queue size said to be %d", + obj.GetName((obj_t*) pThis), "...", iQueueSize); + } + CHKiRet(localRet); + + /* check if we should discard this element */ + localRet = qqueueChkDiscardMsg(pThis, pThis->iQueueSize, pMsg); + if(localRet == RS_RET_QUEUE_FULL) { + ++nDiscarded; + continue; + } else if(localRet != RS_RET_OK) { + ABORT_FINALIZE(localRet); + } + + /* all well, use this element */ + pWti->batch.pElem[nDequeued].pMsg = pMsg; + pWti->batch.eltState[nDequeued] = BATCH_STATE_RDY; + ++nDequeued; + if(nDequeued < iMinDeqBatchSize && getLogicalQueueSize(pThis) == 0) { + while(!pThis->bShutdownImmediate + && keep_running + && nDequeued < iMinDeqBatchSize + && getLogicalQueueSize(pThis) == 0) { + dbgprintf("%s minDeqBatchSize doing wait, batch is %d messages, " + "queue size %d\n", obj.GetName((obj_t*) pThis), + nDequeued, getLogicalQueueSize(pThis)); + if(wtiWaitNonEmpty(pWti, timeout) == 0) { /* timeout? */ + DBGPRINTF("%s minDeqBatchSize timeout, batch is %d messages\n", + obj.GetName((obj_t*) pThis), nDequeued); + keep_running = 0; + } + } + } + if(keep_running) { + keep_running = (getLogicalQueueSize(pThis) > 0) + && (nDequeued < pThis->iDeqBatchSize); + } + } + + if(pThis->qType == QUEUETYPE_DISK) { + strm.GetCurrOffset(pThis->tVars.disk.pReadDeq, &pThis->tVars.disk.deqOffs); + pThis->tVars.disk.deqFileNumOut = strmGetCurrFileNum(pThis->tVars.disk.pReadDeq); + } + + /* it is sufficient to persist only when the bulk of work is done */ + qqueueChkPersist(pThis, nDequeued+nDiscarded+nDeleted); + + /* If messages where DISCARDED, we need to substract them from the OverallQueueSize */ +# ifdef ENABLE_IMDIAG +# ifdef HAVE_ATOMIC_BUILTINS + ATOMIC_SUB(&iOverallQueueSize, nDiscarded, &NULL); +# else + iOverallQueueSize -= nDiscarded; /* racy, but we can't wait for a mutex! */ +# endif + DBGOPRINT((obj_t*) pThis, "dequeued %d discarded %d QueueSize %d consumable elements, szlog %d sz phys %d\n", + nDequeued, nDiscarded, iOverallQueueSize, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); +# else + DBGOPRINT((obj_t*) pThis, "dequeued %d discarded %d consumable elements, szlog %d sz phys %d\n", + nDequeued, nDiscarded, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); +# endif + + pWti->batch.nElem = nDequeued; + pWti->batch.nElemDeq = nDequeued + nDiscarded; + pWti->batch.deqID = getNextDeqID(pThis); + *piRemainingQueueSize = iQueueSize; +finalize_it: + RETiRet; +} + + +/* dequeue the queued object for the queue consumers. + * rgerhards, 2008-10-21 + * I made a radical change - we now dequeue multiple elements, and store these objects in + * an array of user pointers. We expect that this increases performance. + * rgerhards, 2009-04-22 + */ +static rsRetVal +DequeueConsumable(qqueue_t *pThis, wti_t *pWti, int *const pSkippedMsgs) +{ + DEFiRet; + int iQueueSize = 0; /* keep the compiler happy... */ + + *pSkippedMsgs = 0; + /* dequeue element batch (still protected from mutex) */ + iRet = DequeueConsumableElements(pThis, pWti, &iQueueSize, pSkippedMsgs); + if(*pSkippedMsgs > 0) { + LogError(0, RS_RET_ERR, "%s: lost %d messages from diskqueue (invalid .qi file)", + obj.GetName((obj_t*)pThis), *pSkippedMsgs); + } + + /* awake some flow-controlled sources if we can do this right now */ + /* TODO: this could be done better from a performance point of view -- do it only if + * we have someone waiting for the condition (or only when we hit the watermark right + * on the nail [exact value]) -- rgerhards, 2008-03-14 + * now that we dequeue batches of pointers, this is much less an issue... + * rgerhards, 2009-04-22 + */ + if(iQueueSize < pThis->iFullDlyMrk / 2 || glbl.GetGlobalInputTermState() == 1) { + pthread_cond_broadcast(&pThis->belowFullDlyWtrMrk); + } + + if(iQueueSize < pThis->iLightDlyMrk / 2) { + pthread_cond_broadcast(&pThis->belowLightDlyWtrMrk); + } + + pthread_cond_signal(&pThis->notFull); + /* WE ARE NO LONGER PROTECTED BY THE MUTEX */ + + if(iRet != RS_RET_OK && iRet != RS_RET_DISCARDMSG) { + LogError(0, iRet, "%s: error dequeueing element - ignoring, " + "but strange things may happen", obj.GetName((obj_t*)pThis)); + } + + RETiRet; +} + + +/* The rate limiter + * + * IMPORTANT: the rate-limiter MUST unlock and re-lock the queue when + * it actually delays processing. Otherwise inputs are stalled. + * + * Here we may wait if a dequeue time window is defined or if we are + * rate-limited. TODO: If we do so, we should also look into the + * way new worker threads are spawned. Obviously, it doesn't make much + * sense to spawn additional worker threads when none of them can do any + * processing. However, it is deemed acceptable to allow this for an initial + * implementation of the timeframe/rate limiting feature. + * Please also note that these feature could also be implemented at the action + * level. However, that would limit them to be used together with actions. We have + * taken the broader approach, moving it right into the queue. This is even + * necessary if we want to prevent spawning of multiple unnecessary worker + * threads as described above. -- rgerhards, 2008-04-02 + * + * + * time window: tCurr is current time; tFrom is start time, tTo is end time (in mil 24h format). + * We may have tFrom = 4, tTo = 10 --> run from 4 to 10 hrs. nice and happy + * we may also have tFrom= 22, tTo = 4 -> run from 10pm to 4am, which is actually two + * windows: 0-4; 22-23:59 + * so when to run? Let's assume we have 3am + * + * if(tTo < tFrom) { + * if(tCurr < tTo [3 < 4] || tCurr > tFrom [3 > 22]) + * do work + * else + * sleep for tFrom - tCurr "hours" [22 - 5 --> 17] + * } else { + * if(tCurr >= tFrom [3 >= 4] && tCurr < tTo [3 < 10]) + * do work + * else + * sleep for tTo - tCurr "hours" [4 - 3 --> 1] + * } + * + * Bottom line: we need to check which type of window we have and need to adjust our + * logic accordingly. Of course, sleep calculations need to be done up to the minute, + * but you get the idea from the code above. + */ +static rsRetVal +RateLimiter(qqueue_t *pThis) +{ + DEFiRet; + int iDelay; + int iHrCurr; + time_t tCurr; + struct tm m; + + ISOBJ_TYPE_assert(pThis, qqueue); + + iDelay = 0; + if(pThis->iDeqtWinToHr != 25) { /* 25 means disabled */ + /* time calls are expensive, so only do them when needed */ + datetime.GetTime(&tCurr); + localtime_r(&tCurr, &m); + iHrCurr = m.tm_hour; + + if(pThis->iDeqtWinToHr < pThis->iDeqtWinFromHr) { + if(iHrCurr < pThis->iDeqtWinToHr || iHrCurr > pThis->iDeqtWinFromHr) { + ; /* do not delay */ + } else { + iDelay = (pThis->iDeqtWinFromHr - iHrCurr) * 3600; + /* this time, we are already into the next hour, so we need + * to subtract our current minute and seconds. + */ + iDelay -= m.tm_min * 60; + iDelay -= m.tm_sec; + } + } else { + if(iHrCurr >= pThis->iDeqtWinFromHr && iHrCurr < pThis->iDeqtWinToHr) { + ; /* do not delay */ + } else { + if(iHrCurr < pThis->iDeqtWinFromHr) { + iDelay = (pThis->iDeqtWinFromHr - iHrCurr - 1) * 3600; + /* -1 as we are already in the hour */ + iDelay += (60 - m.tm_min) * 60; + iDelay += 60 - m.tm_sec; + } else { + iDelay = (24 - iHrCurr + pThis->iDeqtWinFromHr) * 3600; + /* this time, we are already into the next hour, so we need + * to subtract our current minute and seconds. + */ + iDelay -= m.tm_min * 60; + iDelay -= m.tm_sec; + } + } + } + } + + if(iDelay > 0) { + pthread_mutex_unlock(pThis->mut); + DBGOPRINT((obj_t*) pThis, "outside dequeue time window, delaying %d seconds\n", iDelay); + srSleep(iDelay, 0); + pthread_mutex_lock(pThis->mut); + } + + RETiRet; +} + + +/* This dequeues the next batch. Note that this function must not be + * cancelled, else it will leave back an inconsistent state. + * rgerhards, 2009-05-20 + */ +static rsRetVal +DequeueForConsumer(qqueue_t *pThis, wti_t *pWti, int *const pSkippedMsgs) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + ISOBJ_TYPE_assert(pWti, wti); + + CHKiRet(DequeueConsumable(pThis, pWti, pSkippedMsgs)); + + if(pWti->batch.nElem == 0) + ABORT_FINALIZE(RS_RET_IDLE); + + +finalize_it: + RETiRet; +} + + +/* This is called when a batch is processed and the worker does not + * ask for another batch (e.g. because it is to be terminated) + * Note that we must not be terminated while we delete a processed + * batch. Otherwise, we may not complete it, and then the cancel + * handler also tries to delete the batch. But then it finds some of + * the messages already destructed. This was a bug we have seen, especially + * with disk mode, where a delete takes rather long. Anyhow, the coneptual + * problem exists in all queue modes. + * rgerhards, 2009-05-27 + */ +static rsRetVal +batchProcessed(qqueue_t *pThis, wti_t *pWti) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + ISOBJ_TYPE_assert(pWti, wti); + + int iCancelStateSave; + /* at this spot, we must not be cancelled */ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); + DeleteProcessedBatch(pThis, &pWti->batch); + qqueueChkPersist(pThis, pWti->batch.nElemDeq); + pthread_setcancelstate(iCancelStateSave, NULL); + + RETiRet; +} + + +/* This is the queue consumer in the regular (non-DA) case. It is + * protected by the queue mutex, but MUST release it as soon as possible. + * rgerhards, 2008-01-21 + */ +static rsRetVal +ConsumerReg(qqueue_t *pThis, wti_t *pWti) +{ + int iCancelStateSave; + int bNeedReLock = 0; /**< do we need to lock the mutex again? */ + int skippedMsgs = 0; /**< did the queue loose any messages (can happen with + ** disk queue if .qi file is corrupt */ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + ISOBJ_TYPE_assert(pWti, wti); + + iRet = DequeueForConsumer(pThis, pWti, &skippedMsgs); + if(iRet == RS_RET_FILE_NOT_FOUND) { + /* This is a fatal condition and means the queue is almost unusable */ + d_pthread_mutex_unlock(pThis->mut); + DBGOPRINT((obj_t*) pThis, "got 'file not found' error %d, queue defunct\n", iRet); + iRet = queueSwitchToEmergencyMode(pThis, iRet); + // TODO: think about what to return as iRet -- keep RS_RET_FILE_NOT_FOUND? + d_pthread_mutex_lock(pThis->mut); + } + if (iRet != RS_RET_OK) { + FINALIZE; + } + + /* we now have a non-idle batch of work, so we can release the queue mutex and process it */ + d_pthread_mutex_unlock(pThis->mut); + bNeedReLock = 1; + + /* report errors, now that we are outside of queue lock */ + if(skippedMsgs > 0) { + LogError(0, 0, "problem on disk queue '%s': " + "queue files contain %d messages fewer than specified " + "in .qi file -- we lost those messages. That's all we know.", + obj.GetName((obj_t*) pThis), skippedMsgs); + } + + /* at this spot, we may be cancelled */ + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave); + + + 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 + */ + if(pThis->iDeqSlowdown) { + DBGOPRINT((obj_t*) pThis, "sleeping %d microseconds as requested by config params\n", + pThis->iDeqSlowdown); + srSleep(pThis->iDeqSlowdown / 1000000, pThis->iDeqSlowdown % 1000000); + } + + /* but now cancellation is no longer permitted */ + pthread_setcancelstate(iCancelStateSave, NULL); + +finalize_it: + DBGPRINTF("regular consumer finished, iret=%d, szlog %d sz phys %d\n", iRet, + getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); + + /* now we are done, but potentially need to re-acquire the mutex */ + if(bNeedReLock) + d_pthread_mutex_lock(pThis->mut); + + RETiRet; +} + + +/* This is a special consumer to feed the disk-queue in disk-assisted mode. + * When active, our own queue more or less acts as a memory buffer to the disk. + * So this consumer just needs to drain the memory queue and submit entries + * to the disk queue. The disk queue will then call the actual consumer from + * the app point of view (we chain two queues here). + * When this method is entered, the mutex is always locked and needs to be unlocked + * as part of the processing. + * rgerhards, 2008-01-14 + */ +static rsRetVal +ConsumerDA(qqueue_t *pThis, wti_t *pWti) +{ + int i; + int iCancelStateSave; + int bNeedReLock = 0; /**< do we need to lock the mutex again? */ + int skippedMsgs = 0; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + ISOBJ_TYPE_assert(pWti, wti); + + CHKiRet(DequeueForConsumer(pThis, pWti, &skippedMsgs)); + + /* we now have a non-idle batch of work, so we can release the queue mutex and process it */ + d_pthread_mutex_unlock(pThis->mut); + bNeedReLock = 1; + + /* at this spot, we may be cancelled */ + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave); + + /* iterate over returned results and enqueue them in DA queue */ + for(i = 0 ; i < pWti->batch.nElem && !pThis->bShutdownImmediate ; i++) { + iRet = qqueueEnqMsg(pThis->pqDA, eFLOWCTL_NO_DELAY, MsgAddRef(pWti->batch.pElem[i].pMsg)); + if(iRet != RS_RET_OK) { + if(iRet == RS_RET_ERR_QUEUE_EMERGENCY) { + /* Queue emergency error occurred */ + DBGOPRINT((obj_t*) pThis, "ConsumerDA:qqueueEnqMsg caught RS_RET_ERR_QUEUE_EMERGENCY," + "aborting loop.\n"); + FINALIZE; + } else { + DBGOPRINT((obj_t*) pThis, "ConsumerDA:qqueueEnqMsg item (%d) returned " + "with error state: '%d'\n", i, iRet); + } + } + pWti->batch.eltState[i] = BATCH_STATE_COMM; /* commited to other queue! */ + } + + /* but now cancellation is no longer permitted */ + pthread_setcancelstate(iCancelStateSave, NULL); + +finalize_it: + /* Check the last return state of qqueueEnqMsg. If an error was returned, we acknowledge it only. + * Unless the error code is RS_RET_ERR_QUEUE_EMERGENCY, we reset the return state to RS_RET_OK. + * Otherwise the Caller functions would run into an infinite Loop trying to enqueue the + * same messages over and over again. + * + * However we do NOT overwrite positive return states like + * RS_RET_TERMINATE_NOW, + * RS_RET_NO_RUN, + * RS_RET_IDLE, + * RS_RET_TERMINATE_WHEN_IDLE + * These return states are important for Queue handling of the upper laying functions. + * RGer: Note that checking for iRet < 0 is a bit bold. In theory, positive iRet + * values are "OK" states, and things that the caller shall deal with. However, + * this has not been done so consistently. Andre convinced me that the current + * code is an elegant solution. However, if problems with queue workers and/or + * shutdown come up, this code here should be looked at suspiciously. In those + * cases it may work out to check all status codes explicitely, just to avoid + * a pitfall due to unexpected states being passed on to the caller. + */ + if( iRet != RS_RET_OK && + iRet != RS_RET_ERR_QUEUE_EMERGENCY && + iRet < 0) { + DBGOPRINT((obj_t*) pThis, "ConsumerDA:qqueueEnqMsg Resetting iRet from %d back to RS_RET_OK\n", iRet); + iRet = RS_RET_OK; + } else { + DBGOPRINT((obj_t*) pThis, "ConsumerDA:qqueueEnqMsg returns with iRet %d\n", iRet); + } + + /* now we are done, but potentially need to re-acquire the mutex */ + if(bNeedReLock) + d_pthread_mutex_lock(pThis->mut); + + RETiRet; +} + + +/* must only be called when the queue mutex is locked, else results + * are not stable! + */ +static rsRetVal +qqueueChkStopWrkrDA(qqueue_t *pThis) +{ + DEFiRet; + + DBGPRINTF("rger: chkStopWrkrDA called, low watermark %d, log Size %d, phys Size %d, bEnqOnly %d\n", + pThis->iLowWtrMrk, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis), pThis->bEnqOnly); + if(pThis->bEnqOnly) { + iRet = RS_RET_TERMINATE_WHEN_IDLE; + } + if(getPhysicalQueueSize(pThis) <= pThis->iLowWtrMrk) { + iRet = RS_RET_TERMINATE_NOW; + } + + RETiRet; +} + + +/* must only be called when the queue mutex is locked, else results + * are not stable! + * If we are a child, we have done our duty when the queue is empty. In that case, + * we can terminate. Version for the regular worker thread. + */ +static rsRetVal +ChkStopWrkrReg(qqueue_t *pThis) +{ + DEFiRet; + /*DBGPRINTF("XXXX: chkStopWrkrReg called, low watermark %d, log Size %d, phys Size %d, bEnqOnly %d\n", + pThis->iLowWtrMrk, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis), pThis->bEnqOnly);*/ + if(pThis->bEnqOnly) { + iRet = RS_RET_TERMINATE_NOW; + } else if(pThis->pqParent != NULL) { + iRet = RS_RET_TERMINATE_WHEN_IDLE; + } + + RETiRet; +} + + +/* return the configured "deq max at once" interval + * rgerhards, 2009-04-22 + */ +static rsRetVal +GetDeqBatchSize(qqueue_t *pThis, int *pVal) +{ + DEFiRet; + assert(pVal != NULL); + *pVal = pThis->iDeqBatchSize; + RETiRet; +} + + +/* start up the queue - it must have been constructed and parameters defined + * before. + */ +rsRetVal +qqueueStart(rsconf_t *cnf, qqueue_t *pThis) /* this is the ConstructionFinalizer */ +{ + DEFiRet; + uchar pszBuf[64]; + uchar pszQIFNam[MAXFNAME]; + int wrk; + uchar *qName; + size_t lenBuf; + + assert(pThis != NULL); + + /* do not modify the queue if it's already running(happens when dynamic config reload is invoked + * and the queue is used in the new config as well) + */ + if (pThis->isRunning) + FINALIZE; + + dbgoprint((obj_t*) pThis, "starting queue\n"); + + if(pThis->pszSpoolDir == NULL) { + /* note: we need to pick the path so late as we do not have + * the workdir during early config load + */ + if((pThis->pszSpoolDir = (uchar*) strdup((char*)glbl.GetWorkDir(cnf))) == NULL) + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir); + } + /* set type-specific handlers and other very type-specific things + * (we can not totally hide it...) + */ + switch(pThis->qType) { + case QUEUETYPE_FIXED_ARRAY: + pThis->qConstruct = qConstructFixedArray; + pThis->qDestruct = qDestructFixedArray; + pThis->qAdd = qAddFixedArray; + pThis->qDeq = qDeqFixedArray; + pThis->qDel = qDelFixedArray; + pThis->MultiEnq = qqueueMultiEnqObjNonDirect; + break; + case QUEUETYPE_LINKEDLIST: + pThis->qConstruct = qConstructLinkedList; + pThis->qDestruct = qDestructLinkedList; + pThis->qAdd = qAddLinkedList; + pThis->qDeq = qDeqLinkedList; + pThis->qDel = qDelLinkedList; + pThis->MultiEnq = qqueueMultiEnqObjNonDirect; + break; + case QUEUETYPE_DISK: + pThis->qConstruct = qConstructDisk; + pThis->qDestruct = qDestructDisk; + pThis->qAdd = qAddDisk; + pThis->qDeq = qDeqDisk; + pThis->qDel = NULL; /* delete for disk handled via special code! */ + pThis->MultiEnq = qqueueMultiEnqObjNonDirect; + /* pre-construct file name for .qi file */ + pThis->lenQIFNam = snprintf((char*)pszQIFNam, sizeof(pszQIFNam), + "%s/%s.qi", (char*) pThis->pszSpoolDir, (char*)pThis->pszFilePrefix); + pThis->pszQIFNam = ustrdup(pszQIFNam); + DBGOPRINT((obj_t*) pThis, ".qi file name is '%s', len %d\n", pThis->pszQIFNam, + (int) pThis->lenQIFNam); + break; + 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->MultiEnq = qqueueMultiEnqObjDirect; + pThis->qDel = NULL; + break; + } + + /* finalize some initializations that could not yet be done because it is + * influenced by properties which might have been set after queueConstruct () + */ + if(pThis->pqParent == NULL) { + CHKmalloc(pThis->mut = (pthread_mutex_t *) malloc (sizeof (pthread_mutex_t))); + pthread_mutex_init(pThis->mut, NULL); + } else { + /* child queue, we need to use parent's mutex */ + DBGOPRINT((obj_t*) pThis, "I am a child\n"); + pThis->mut = pThis->pqParent->mut; + } + + pthread_mutex_init(&pThis->mutThrdMgmt, NULL); + pthread_cond_init (&pThis->notFull, NULL); + pthread_cond_init (&pThis->belowFullDlyWtrMrk, NULL); + pthread_cond_init (&pThis->belowLightDlyWtrMrk, NULL); + + /* call type-specific constructor */ + CHKiRet(pThis->qConstruct(pThis)); /* this also sets bIsDA */ + + /* re-adjust some params if required */ + if(pThis->bIsDA) { + /* if we are in DA mode, we must make sure full delayable messages do not + * initiate going to disk! + */ + wrk = pThis->iHighWtrMrk - (pThis->iHighWtrMrk / 100) * 50; /* 50% of high water mark */ + if(wrk < pThis->iFullDlyMrk) + pThis->iFullDlyMrk = wrk; + } + + DBGOPRINT((obj_t*) pThis, "params: type %d, enq-only %d, disk assisted %d, spoolDir '%s', maxFileSz %lld, " + "maxQSize %d, lqsize %d, pqsize %d, child %d, full delay %d, " + "light delay %d, deq batch size %d, min deq batch size %d, " + "high wtrmrk %d, low wtrmrk %d, " + "discardmrk %d, max wrkr %d, min msgs f. wrkr %d " + "takeFlowCtlFromMsg %d\n", + pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->pszSpoolDir, + pThis->iMaxFileSize, pThis->iMaxQueueSize, + getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis), + pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk, + pThis->iDeqBatchSize, pThis->iMinDeqBatchSize, pThis->iHighWtrMrk, pThis->iLowWtrMrk, + pThis->iDiscardMrk, (int) pThis->iNumWorkerThreads, (int) pThis->iMinMsgsPerWrkr, + pThis->takeFlowCtlFromMsg); + + pThis->bQueueStarted = 1; + if(pThis->qType == QUEUETYPE_DIRECT) + FINALIZE; /* with direct queues, we are already finished... */ + + /* create worker thread pools for regular and DA operation. + */ + lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "%.*s:Reg", + (int) (sizeof(pszBuf)-16), + obj.GetName((obj_t*) pThis)); /* leave some room inside the name for suffixes */ + if(lenBuf >= sizeof(pszBuf)) { + LogError(0, RS_RET_INTERNAL_ERROR, "%s:%d debug header too long: %zd - in " + "thory this cannot happen - truncating", __FILE__, __LINE__, lenBuf); + lenBuf = sizeof(pszBuf)-1; + pszBuf[lenBuf] = '\0'; + } + CHKiRet(wtpConstruct (&pThis->pWtpReg)); + CHKiRet(wtpSetDbgHdr (pThis->pWtpReg, pszBuf, lenBuf)); + CHKiRet(wtpSetpfRateLimiter (pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) RateLimiter)); + CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, int)) ChkStopWrkrReg)); + CHKiRet(wtpSetpfGetDeqBatchSize (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, int*)) GetDeqBatchSize)); + CHKiRet(wtpSetpfDoWork (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, void *pWti)) ConsumerReg)); + CHKiRet(wtpSetpfObjProcessed (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, wti_t *pWti)) batchProcessed)); + CHKiRet(wtpSetpmutUsr (pThis->pWtpReg, pThis->mut)); + CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpReg, pThis->iNumWorkerThreads)); + CHKiRet(wtpSettoWrkShutdown (pThis->pWtpReg, pThis->toWrkShutdown)); + CHKiRet(wtpSetpUsr (pThis->pWtpReg, pThis)); + CHKiRet(wtpConstructFinalize (pThis->pWtpReg)); + + /* set up DA system if we have a disk-assisted queue */ + if(pThis->bIsDA) + InitDA(pThis, LOCK_MUTEX); /* initiate DA mode */ + + DBGOPRINT((obj_t*) pThis, "queue finished initialization\n"); + + /* if the queue already contains data, we need to start the correct number of worker threads. This can be + * the case when a disk queue has been loaded. If we did not start it here, it would never start. + */ + qqueueAdviseMaxWorkers(pThis); + + /* support statistics gathering */ + qName = obj.GetName((obj_t*)pThis); + CHKiRet(statsobj.Construct(&pThis->statsobj)); + CHKiRet(statsobj.SetName(pThis->statsobj, qName)); + CHKiRet(statsobj.SetOrigin(pThis->statsobj, (uchar*)"core.queue")); + /* we need to save the queue size, as the stats module initializes it to 0! */ + /* iQueueSize is a dual-use counter: no init, no mutex! */ + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("size"), + ctrType_Int, CTR_FLAG_NONE, &pThis->iQueueSize)); + + STATSCOUNTER_INIT(pThis->ctrEnqueued, pThis->mutCtrEnqueued); + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("enqueued"), + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrEnqueued)); + + STATSCOUNTER_INIT(pThis->ctrFull, pThis->mutCtrFull); + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("full"), + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrFull)); + + STATSCOUNTER_INIT(pThis->ctrFDscrd, pThis->mutCtrFDscrd); + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.full"), + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrFDscrd)); + STATSCOUNTER_INIT(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd); + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.nf"), + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrNFDscrd)); + + pThis->ctrMaxqsize = 0; /* no mutex needed, thus no init call */ + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("maxqsize"), + ctrType_Int, CTR_FLAG_NONE, &pThis->ctrMaxqsize)); + + CHKiRet(statsobj.ConstructFinalize(pThis->statsobj)); + +finalize_it: + if(iRet != RS_RET_OK) { + /* note: a child uses it's parent mutex, so do not delete it! */ + if(pThis->pqParent == NULL && pThis->mut != NULL) + free(pThis->mut); + } else { + pThis->isRunning = 1; + } + RETiRet; +} + + +/* persist the queue to disk (write the .qi file). If we have something to persist, we first + * save the information on the queue properties itself and then we call + * the queue-type specific drivers. + * Variable bIsCheckpoint is set to 1 if the persist is for a checkpoint, + * and 0 otherwise. + * rgerhards, 2008-01-10 + */ +static rsRetVal +qqueuePersist(qqueue_t *pThis, int bIsCheckpoint) +{ + DEFiRet; + char *tmpQIFName = NULL; + strm_t *psQIF = NULL; /* Queue Info File */ + char errStr[1024]; + + assert(pThis != NULL); + + if(pThis->qType != QUEUETYPE_DISK) { + if(getPhysicalQueueSize(pThis) > 0) { + /* This error code is OK, but we will probably not implement this any time + * The reason is that persistence happens via DA queues. But I would like to + * leave the code as is, as we so have a hook in case we need one. + * -- rgerhards, 2008-01-28 + */ + ABORT_FINALIZE(RS_RET_NOT_IMPLEMENTED); + } else + FINALIZE; /* if the queue is empty, we are happy and done... */ + } + + DBGOPRINT((obj_t*) pThis, "persisting queue to disk, %d entries...\n", getPhysicalQueueSize(pThis)); + + if((bIsCheckpoint != QUEUE_CHECKPOINT) && (getPhysicalQueueSize(pThis) == 0)) { + if(pThis->bNeedDelQIF) { + unlink((char*)pThis->pszQIFNam); + pThis->bNeedDelQIF = 0; + } + /* indicate spool file needs to be deleted */ + if(pThis->tVars.disk.pReadDel != NULL) /* may be NULL if we had a startup failure! */ + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDel, 1)); + FINALIZE; /* nothing left to do, so be happy */ + } + + int lentmpQIFName; +#ifdef _AIX + lentmpQIFName = strlen( pThis->pszQIFNam) + strlen(".tmp") + 1; + tmpQIFName = malloc(sizeof(char)*lentmpQIFName); + if(tmpQIFName == NULL) + tmpQIFName = (char*)pThis->pszQIFNam; + snprintf(tmpQIFName, lentmpQIFName, "%s.tmp", pThis->pszQIFNam); +#else + lentmpQIFName = asprintf((char **)&tmpQIFName, "%s.tmp", pThis->pszQIFNam); + if(tmpQIFName == NULL) + tmpQIFName = (char*)pThis->pszQIFNam; +#endif + + CHKiRet(strm.Construct(&psQIF)); + CHKiRet(strm.SettOperationsMode(psQIF, STREAMMODE_WRITE_TRUNC)); + CHKiRet(strm.SetbSync(psQIF, pThis->bSyncQueueFiles)); + CHKiRet(strm.SetsType(psQIF, STREAMTYPE_FILE_SINGLE)); + CHKiRet(strm.SetFName(psQIF, (uchar*) tmpQIFName, lentmpQIFName)); + CHKiRet(strm.ConstructFinalize(psQIF)); + + /* first, write the property bag for ourselfs + * And, surprisingly enough, we currently need to persist only the size of the + * queue. All the rest is re-created with then-current config parameters when the + * queue is re-created. Well, we'll also save the current queue type, just so that + * we know when somebody has changed the queue type... -- rgerhards, 2008-01-11 + */ + CHKiRet(obj.BeginSerializePropBag(psQIF, (obj_t*) pThis)); + objSerializeSCALAR(psQIF, iQueueSize, INT); + objSerializeSCALAR(psQIF, tVars.disk.sizeOnDisk, INT64); + CHKiRet(obj.EndSerialize(psQIF)); + + /* now persist the stream info */ + if(pThis->tVars.disk.pWrite != NULL) + CHKiRet(strm.Serialize(pThis->tVars.disk.pWrite, psQIF)); + if(pThis->tVars.disk.pReadDel != NULL) + CHKiRet(strm.Serialize(pThis->tVars.disk.pReadDel, psQIF)); + + strm.Destruct(&psQIF); + if(tmpQIFName != (char*)pThis->pszQIFNam) { /* pointer, not string comparison! */ + if(rename(tmpQIFName, (char*)pThis->pszQIFNam) != 0) { + rs_strerror_r(errno, errStr, sizeof(errStr)); + DBGOPRINT((obj_t*) pThis, + "FATAL error: renaming temporary .qi file failed: %s\n", + errStr); + ABORT_FINALIZE(RS_RET_RENAME_TMP_QI_ERROR); + } + } + + /* tell the input file object that it must not delete the file on close if the queue + * is non-empty - but only if we are not during a simple checkpoint + */ + if(bIsCheckpoint != QUEUE_CHECKPOINT + && pThis->tVars.disk.pReadDel != NULL) { + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDel, 0)); + } + + /* we have persisted the queue object. So whenever it comes to an empty queue, + * we need to delete the QIF. Thus, we indicte that need. + */ + pThis->bNeedDelQIF = 1; + +finalize_it: + if(tmpQIFName != (char*)pThis->pszQIFNam) /* pointer, not string comparison! */ + free(tmpQIFName); + if(psQIF != NULL) + strm.Destruct(&psQIF); + + RETiRet; +} + + +/* check if we need to persist the current queue info. If an + * error occurs, this should be ignored by caller (but we still + * abide to our regular call interface)... + * rgerhards, 2008-01-13 + * nUpdates is the number of updates since the last call to this function. + * It may be > 1 due to batches. -- rgerhards, 2009-05-12 + */ +static rsRetVal qqueueChkPersist(qqueue_t *const pThis, const int nUpdates) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, qqueue); + assert(nUpdates >= 0); + + if(nUpdates == 0) + FINALIZE; + + pThis->iUpdsSincePersist += nUpdates; + if(pThis->iPersistUpdCnt && pThis->iUpdsSincePersist >= pThis->iPersistUpdCnt) { + qqueuePersist(pThis, QUEUE_CHECKPOINT); + pThis->iUpdsSincePersist = 0; + } + +finalize_it: + RETiRet; +} + + +/* persist a queue with all data elements to disk - this is used to handle + * bSaveOnShutdown. We utilize the DA worker to do this. This must only + * be called after all workers have been shut down and if bSaveOnShutdown + * is actually set. Note that this function may potentially run long, + * depending on the queue configuration (e.g. store on remote machine). + * rgerhards, 2009-05-26 + */ +static rsRetVal +DoSaveOnShutdown(qqueue_t *pThis) +{ + struct timespec tTimeout; + rsRetVal iRetLocal; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + + /* we reduce the low water mark, otherwise the DA worker would terminate when + * it is reached. + */ + DBGOPRINT((obj_t*) pThis, "bSaveOnShutdown set, restarting DA worker...\n"); + pThis->bShutdownImmediate = 0; /* would termiante the DA worker! */ + pThis->iLowWtrMrk = 0; + wtpSetState(pThis->pWtpDA, wtpState_SHUTDOWN); /* shutdown worker (only) when done (was _IMMEDIATE!) */ + wtpAdviseMaxWorkers(pThis->pWtpDA, 1, PERMIT_WORKER_START_DURING_SHUTDOWN); /* restart DA worker */ + + DBGOPRINT((obj_t*) pThis, "waiting for DA worker to terminate...\n"); + timeoutComp(&tTimeout, QUEUE_TIMEOUT_ETERNAL); + /* and run the primary queue's DA worker to drain the queue */ + iRetLocal = wtpShutdownAll(pThis->pWtpDA, wtpState_SHUTDOWN, &tTimeout); + DBGOPRINT((obj_t*) pThis, "end queue persistence run, iRet %d, queue size log %d, phys %d\n", + iRetLocal, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis)); + if(iRetLocal != RS_RET_OK) { + DBGOPRINT((obj_t*) pThis, "unexpected iRet state %d after trying to shut down primary " + "queue in disk save mode, continuing, but results are unpredictable\n", iRetLocal); + } + + RETiRet; +} + + +/* destructor for the queue object */ +BEGINobjDestruct(qqueue) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(qqueue) + DBGOPRINT((obj_t*) pThis, "shutdown: begin to destruct queue\n"); + if(ourConf->globals.shutdownQueueDoubleSize) { + pThis->iHighWtrMrk *= 2; + pThis->iMaxQueueSize *= 2; + } + if(pThis->bQueueStarted) { + /* shut down all workers + * We do not need to shutdown workers when we are in enqueue-only mode or we are a + * direct queue - because in both cases we have none... ;) + * with a child! -- rgerhards, 2008-01-28 + */ + if(pThis->qType != QUEUETYPE_DIRECT && !pThis->bEnqOnly && pThis->pqParent == NULL + && pThis->pWtpReg != NULL) + qqueueShutdownWorkers(pThis); + + if(pThis->bIsDA && getPhysicalQueueSize(pThis) > 0){ + if(pThis->bSaveOnShutdown) { + LogMsg(0, RS_RET_TIMED_OUT, LOG_INFO, + "%s: queue holds %d messages after shutdown of workers. " + "queue.saveonshutdown is set, so data will now be spooled to disk", + objGetName((obj_t*) pThis), getPhysicalQueueSize(pThis)); + CHKiRet(DoSaveOnShutdown(pThis)); + } else { + LogMsg(0, RS_RET_TIMED_OUT, LOG_WARNING, + "%s: queue holds %d messages after shutdown of workers. " + "queue.saveonshutdown is NOT set, so data will be discarded.", + objGetName((obj_t*) pThis), getPhysicalQueueSize(pThis)); + } + } + + /* finally destruct our (regular) worker thread pool + * Note: currently pWtpReg is never NULL, but if we optimize our logic, this may happen, + * e.g. when they are not created in enqueue-only mode. We already check the condition + * as this may otherwise be very hard to find once we optimize (and have long forgotten + * about this condition here ;) + * rgerhards, 2008-01-25 + */ + if(pThis->qType != QUEUETYPE_DIRECT && pThis->pWtpReg != NULL) { + wtpDestruct(&pThis->pWtpReg); + } + + /* Now check if we actually have a DA queue and, if so, destruct it. + * Note that the wtp must be destructed first, it may be in cancel cleanup handler + * *right now* and actually *need* to access the queue object to persist some final + * data (re-queueing case). So we need to destruct the wtp first, which will make + * sure all workers have terminated. Please note that this also generates a situation + * where it is possible that the DA queue has a parent pointer but the parent has + * no WtpDA associated with it - which is perfectly legal thanks to this code here. + */ + if(pThis->pWtpDA != NULL) { + wtpDestruct(&pThis->pWtpDA); + } + if(pThis->pqDA != NULL) { + qqueueDestruct(&pThis->pqDA); + } + + /* persist the queue (we always do that - queuePersits() does cleanup if the queue is empty) + * This handler is most important for disk queues, it will finally persist the necessary + * on-disk structures. In theory, other queueing modes may implement their other (non-DA) + * methods of persisting a queue between runs, but in practice all of this is done via + * disk queues and DA mode. Anyhow, it doesn't hurt to know that we could extend it here + * if need arises (what I doubt...) -- rgerhards, 2008-01-25 + */ + CHKiRet_Hdlr(qqueuePersist(pThis, QUEUE_NO_CHECKPOINT)) { + DBGOPRINT((obj_t*) pThis, "error %d persisting queue - data lost!\n", iRet); + } + + /* finally, clean up some simple things... */ + if(pThis->pqParent == NULL) { + /* if we are not a child, we allocated our own mutex, which we now need to destroy */ + pthread_mutex_destroy(pThis->mut); + free(pThis->mut); + } + pthread_mutex_destroy(&pThis->mutThrdMgmt); + pthread_cond_destroy(&pThis->notFull); + pthread_cond_destroy(&pThis->belowFullDlyWtrMrk); + pthread_cond_destroy(&pThis->belowLightDlyWtrMrk); + + DESTROY_ATOMIC_HELPER_MUT(pThis->mutQueueSize); + DESTROY_ATOMIC_HELPER_MUT(pThis->mutLogDeq); + + /* type-specific destructor */ + iRet = pThis->qDestruct(pThis); + } + + free(pThis->pszFilePrefix); + free(pThis->pszSpoolDir); + if(pThis->useCryprov) { + pThis->cryprov.Destruct(&pThis->cryprovData); + obj.ReleaseObj(__FILE__, pThis->cryprovNameFull+2, pThis->cryprovNameFull, + (void*) &pThis->cryprov); + free(pThis->cryprovName); + free(pThis->cryprovNameFull); + } + + /* some queues do not provide stats and thus have no statsobj! */ + if(pThis->statsobj != NULL) + statsobj.Destruct(&pThis->statsobj); +ENDobjDestruct(qqueue) + + +/* set the queue's spool directory. The directory MUST NOT be NULL. + * The passed-in string is duplicated. So if the caller does not need + * it any longer, it must free it. + */ +rsRetVal +qqueueSetSpoolDir(qqueue_t *pThis, uchar *pszSpoolDir, int lenSpoolDir) +{ + DEFiRet; + + free(pThis->pszSpoolDir); + CHKmalloc(pThis->pszSpoolDir = ustrdup(pszSpoolDir)); + pThis->lenSpoolDir = lenSpoolDir; + +finalize_it: + RETiRet; +} + + +/* set the queue's file prefix + * The passed-in string is duplicated. So if the caller does not need + * it any longer, it must free it. + * rgerhards, 2008-01-09 + */ +rsRetVal +qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix) +{ + DEFiRet; + + free(pThis->pszFilePrefix); + pThis->pszFilePrefix = NULL; + + if(pszPrefix == NULL) /* just unset the prefix! */ + ABORT_FINALIZE(RS_RET_OK); + + if((pThis->pszFilePrefix = malloc(iLenPrefix + 1)) == NULL) + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + memcpy(pThis->pszFilePrefix, pszPrefix, iLenPrefix + 1); + pThis->lenFilePrefix = iLenPrefix; + +finalize_it: + RETiRet; +} + +/* set the queue's maximum file size + * rgerhards, 2008-01-09 + */ +rsRetVal +qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + + if(iMaxFileSize < 1024) { + ABORT_FINALIZE(RS_RET_VALUE_TOO_LOW); + } + + pThis->iMaxFileSize = iMaxFileSize; + +finalize_it: + RETiRet; +} + + +/* enqueue a single data object. + * Note that the queue mutex MUST already be locked when this function is called. + * rgerhards, 2009-06-16 + */ +static rsRetVal +doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, smsg_t *pMsg) +{ + DEFiRet; + int err; + struct timespec t; + + STATSCOUNTER_INC(pThis->ctrEnqueued, pThis->mutCtrEnqueued); + /* first check if we need to discard this message (which will cause CHKiRet() to exit) + */ + CHKiRet(qqueueChkDiscardMsg(pThis, pThis->iQueueSize, pMsg)); + + /* handle flow control + * There are two different flow control mechanisms: basic and advanced flow control. + * Basic flow control has always been implemented and protects the queue structures + * in that it makes sure no more data is enqueued than the queue is configured to + * support. Enhanced flow control is being added today. There are some sources which + * can easily be stopped, e.g. a file reader. This is the case because it is unlikely + * that blocking those sources will have negative effects (after all, the file is + * continued to be written). Other sources can somewhat be blocked (e.g. the kernel + * log reader or the local log stream reader): in general, nothing is lost if messages + * from these sources are not picked up immediately. HOWEVER, they can not block for + * an extended period of time, as this either causes message loss or - even worse - some + * other bad effects (e.g. unresponsive system in respect to the main system log socket). + * Finally, there are some (few) sources which can not be blocked at all. UDP syslog is + * a prime example. If a UDP message is not received, it is simply lost. So we can't + * do anything against UDP sockets that come in too fast. The core idea of advanced + * flow control is that we take into account the different natures of the sources and + * select flow control mechanisms that fit these needs. This also means, in the end + * result, that non-blockable sources like UDP syslog receive priority in the system. + * It's a side effect, but a good one ;) -- rgerhards, 2008-03-14 + */ + if(unlikely(pThis->takeFlowCtlFromMsg)) { /* recommendation is NOT to use this option */ + flowCtlType = pMsg->flowCtlType; + } + if(flowCtlType == eFLOWCTL_FULL_DELAY) { + while(pThis->iQueueSize >= pThis->iFullDlyMrk&& ! glbl.GetGlobalInputTermState()) { + /* We have a problem during shutdown if we block eternally. In that + * case, the the input thread cannot be terminated. So we wake up + * from time to time to check for termination. + * TODO/v6(at earliest): check if we could signal the condition during + * shutdown. However, this requires new queue registries and thus is + * far to much change for a stable version (and I am still not sure it + * is worth the effort, given how seldom this situation occurs and how + * few resources the wakeups need). -- rgerhards, 2012-05-03 + * In any case, this was the old code (if we do the TODO): + * pthread_cond_wait(&pThis->belowFullDlyWtrMrk, pThis->mut); + */ + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: FullDelay mark reached for full " + "delayable message - blocking, queue size is %d.\n", pThis->iQueueSize); + timeoutComp(&t, 1000); + err = pthread_cond_timedwait(&pThis->belowLightDlyWtrMrk, pThis->mut, &t); + if(err != 0 && err != ETIMEDOUT) { + /* Something is really wrong now. Report to debug log and abort the + * wait. That keeps us running, even though we may lose messages. + */ + DBGOPRINT((obj_t*) pThis, "potential program bug: pthread_cond_timedwait()" + "/fulldelay returned %d\n", err); + break; + + } + DBGPRINTF("wti worker in full delay timed out, checking termination...\n"); + } + } else if(flowCtlType == eFLOWCTL_LIGHT_DELAY && !glbl.GetGlobalInputTermState()) { + if(pThis->iQueueSize >= pThis->iLightDlyMrk) { + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: LightDelay mark reached for light " + "delayable message - blocking a bit.\n"); + timeoutComp(&t, 1000); /* 1000 millisconds = 1 second TODO: make configurable */ + err = pthread_cond_timedwait(&pThis->belowLightDlyWtrMrk, pThis->mut, &t); + if(err != 0 && err != ETIMEDOUT) { + /* Something is really wrong now. Report to debug log */ + DBGOPRINT((obj_t*) pThis, "potential program bug: pthread_cond_timedwait()" + "/lightdelay returned %d\n", err); + + } + } + } + + /* from our regular flow control settings, we are now ready to enqueue the object. + * However, we now need to do a check if the queue permits to add more data. If that + * is not the case, basic flow control enters the field, which means we wait for + * the queue to become ready or drop the new message. -- rgerhards, 2008-03-14 + */ + while( (pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize) + || ((pThis->qType == QUEUETYPE_DISK || pThis->bIsDA) && pThis->sizeOnDiskMax != 0 + && pThis->tVars.disk.sizeOnDisk > pThis->sizeOnDiskMax)) { + STATSCOUNTER_INC(pThis->ctrFull, pThis->mutCtrFull); + if(pThis->toEnq == 0 || pThis->bEnqOnly) { + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: queue FULL - configured for immediate " + "discarding QueueSize=%d MaxQueueSize=%d sizeOnDisk=%lld " + "sizeOnDiskMax=%lld\n", pThis->iQueueSize, pThis->iMaxQueueSize, + pThis->tVars.disk.sizeOnDisk, pThis->sizeOnDiskMax); + STATSCOUNTER_INC(pThis->ctrFDscrd, pThis->mutCtrFDscrd); + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_QUEUE_FULL); + } else { + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: queue FULL - waiting %dms to drain.\n", + pThis->toEnq); + if(glbl.GetGlobalInputTermState()) { + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: queue FULL, discard due to " + "FORCE_TERM.\n"); + ABORT_FINALIZE(RS_RET_FORCE_TERM); + } + timeoutComp(&t, pThis->toEnq); + const int r = pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t); + if(dbgTimeoutToStderr && r != 0) { + fprintf(stderr, "%lld: queue timeout(%dms), error %d%s, " + "lost message %s\n", (long long) time(NULL), pThis->toEnq, + r, ( r == ETIMEDOUT) ? "[ETIMEDOUT]" : "", pMsg->pszRawMsg); + } + if(r == ETIMEDOUT) { + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: cond timeout, dropping message!\n"); + STATSCOUNTER_INC(pThis->ctrFDscrd, pThis->mutCtrFDscrd); + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_QUEUE_FULL); + } else if(r != 0) { + DBGOPRINT((obj_t*) pThis, "doEnqSingleObject: cond error %d, dropping message!\n", r); + STATSCOUNTER_INC(pThis->ctrFDscrd, pThis->mutCtrFDscrd); + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_QUEUE_FULL); + } + dbgoprint((obj_t*) pThis, "doEnqSingleObject: wait solved queue full condition, enqueing\n"); + } + } + + /* and finally enqueue the message */ + CHKiRet(qqueueAdd(pThis, pMsg)); + STATSCOUNTER_SETMAX_NOMUT(pThis->ctrMaxqsize, pThis->iQueueSize); + + /* check if we had a file rollover and need to persist + * the .qi file for robustness reasons. + * Note: the n=2 write is required for closing the old file and + * the n=1 write is required after opening and writing to the new + * file. + */ + if(pThis->tVars.disk.nForcePersist > 0) { + DBGOPRINT((obj_t*) pThis, ".qi file write required for robustness reasons (n=%d)\n", + pThis->tVars.disk.nForcePersist); + pThis->tVars.disk.nForcePersist--; + qqueuePersist(pThis, QUEUE_CHECKPOINT); + } + +finalize_it: + RETiRet; +} + +/* ------------------------------ multi-enqueue functions ------------------------------ */ +/* enqueue multiple user data elements at once. The aim is to provide a faster interface + * for object submission. Uses the multi_submit_t helper object. + * Please note that this function is not cancel-safe and consequently + * sets the calling thread's cancelibility state to PTHREAD_CANCEL_DISABLE + * during its execution. If that is not done, race conditions occur if the + * thread is canceled (most important use case is input module termination). + * rgerhards, 2009-06-16 + * Note: there now exists multiple different functions implementing specially + * optimized algorithms for different config cases. -- rgerhards, 2010-06-09 + */ +/* now the function for all modes but direct */ +static rsRetVal +qqueueMultiEnqObjNonDirect(qqueue_t *pThis, multi_submit_t *pMultiSub) +{ + int iCancelStateSave; + int i; + rsRetVal localRet; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + assert(pMultiSub != NULL); + + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); + d_pthread_mutex_lock(pThis->mut); + for(i = 0 ; i < pMultiSub->nElem ; ++i) { + localRet = doEnqSingleObj(pThis, pMultiSub->ppMsgs[i]->flowCtlType, (void*)pMultiSub->ppMsgs[i]); + if(localRet != RS_RET_OK && localRet != RS_RET_QUEUE_FULL) + ABORT_FINALIZE(localRet); + } + qqueueChkPersist(pThis, pMultiSub->nElem); + +finalize_it: + /* make sure at least one worker is running. */ + qqueueAdviseMaxWorkers(pThis); + /* and release the mutex */ + d_pthread_mutex_unlock(pThis->mut); + pthread_setcancelstate(iCancelStateSave, NULL); + DBGOPRINT((obj_t*) pThis, "MultiEnqObj advised worker start\n"); + + RETiRet; +} + +/* now, the same function, but for direct mode */ +static rsRetVal +qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub) +{ + int i; + wti_t *pWti; + DEFiRet; + + pWti = wtiGetDummy(); + pWti->pbShutdownImmediate = &pThis->bShutdownImmediate; + + for(i = 0 ; i < pMultiSub->nElem ; ++i) { + CHKiRet(qAddDirectWithWti(pThis, (void*)pMultiSub->ppMsgs[i], pWti)); + } + +finalize_it: + RETiRet; +} +/* ------------------------------ END multi-enqueue functions ------------------------------ */ + + +/* enqueue a new user data element + * Enqueues the new element and awakes worker thread. + */ +rsRetVal +qqueueEnqMsg(qqueue_t *pThis, flowControl_t flowCtlType, smsg_t *pMsg) +{ + DEFiRet; + int iCancelStateSave; + ISOBJ_TYPE_assert(pThis, qqueue); + + const int isNonDirectQ = pThis->qType != QUEUETYPE_DIRECT; + + if(isNonDirectQ) { + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); + d_pthread_mutex_lock(pThis->mut); + } + + CHKiRet(doEnqSingleObj(pThis, flowCtlType, pMsg)); + + qqueueChkPersist(pThis, 1); + +finalize_it: + if(isNonDirectQ) { + /* make sure at least one worker is running. */ + qqueueAdviseMaxWorkers(pThis); + /* and release the mutex */ + d_pthread_mutex_unlock(pThis->mut); + pthread_setcancelstate(iCancelStateSave, NULL); + DBGOPRINT((obj_t*) pThis, "EnqueueMsg advised worker start\n"); + } + + RETiRet; +} + + +/* are any queue params set at all? 1 - yes, 0 - no + * We need to evaluate the param block for this function, which is somewhat + * inefficient. HOWEVER, this is only done during config load, so we really + * don't care... -- rgerhards, 2013-05-10 + */ +int +queueCnfParamsSet(struct nvlst *lst) +{ + int r; + struct cnfparamvals *pvals; + + pvals = nvlstGetParams(lst, &pblk, NULL); + r = cnfparamvalsIsSet(&pblk, pvals); + cnfparamvalsDestruct(pvals, &pblk); + return r; +} + + +static rsRetVal +initCryprov(qqueue_t *pThis, struct nvlst *lst) +{ + uchar szDrvrName[1024]; + DEFiRet; + + if(snprintf((char*)szDrvrName, sizeof(szDrvrName), "lmcry_%s", pThis->cryprovName) + == sizeof(szDrvrName)) { + LogError(0, RS_RET_ERR, "queue: crypto provider " + "name is too long: '%s' - encryption disabled", + pThis->cryprovName); + ABORT_FINALIZE(RS_RET_ERR); + } + pThis->cryprovNameFull = ustrdup(szDrvrName); + + pThis->cryprov.ifVersion = cryprovCURR_IF_VERSION; + /* The pDrvrName+2 below is a hack to obtain the object name. It + * safes us to have yet another variable with the name without "lm" in + * front of it. If we change the module load interface, we may re-think + * about this hack, but for the time being it is efficient and clean enough. + */ + if(obj.UseObj(__FILE__, szDrvrName, szDrvrName, (void*) &pThis->cryprov) + != RS_RET_OK) { + LogError(0, RS_RET_LOAD_ERROR, "queue: could not load " + "crypto provider '%s' - encryption disabled", + szDrvrName); + ABORT_FINALIZE(RS_RET_CRYPROV_ERR); + } + + if(pThis->cryprov.Construct(&pThis->cryprovData) != RS_RET_OK) { + LogError(0, RS_RET_CRYPROV_ERR, "queue: error constructing " + "crypto provider %s dataset - encryption disabled", + szDrvrName); + ABORT_FINALIZE(RS_RET_CRYPROV_ERR); + } + CHKiRet(pThis->cryprov.SetCnfParam(pThis->cryprovData, lst, CRYPROV_PARAMTYPE_DISK)); + + dbgprintf("loaded crypto provider %s, data instance at %p\n", + szDrvrName, pThis->cryprovData); + pThis->useCryprov = 1; +finalize_it: + RETiRet; +} + +/* check the the queue file name is unique. */ +static rsRetVal ATTR_NONNULL() +checkUniqueDiskFile(qqueue_t *const pThis) +{ + DEFiRet; + struct queue_filename *queue_fn_curr = queue_filename_root; + struct queue_filename *newetry = NULL; + const char *const curr_dirname = (pThis->pszSpoolDir == NULL) ? "" : (char*)pThis->pszSpoolDir; + + if(pThis->pszFilePrefix == NULL) { + FINALIZE; /* no disk queue! */ + } + + while(queue_fn_curr != NULL) { + if(!strcmp((const char*) pThis->pszFilePrefix, queue_fn_curr->filename) && + !strcmp(curr_dirname, queue_fn_curr->dirname)) { + parser_errmsg("queue directory '%s' and file name prefix '%s' already used. " + "This is not possible. Please make it unique.", + curr_dirname, pThis->pszFilePrefix); + ABORT_FINALIZE(RS_RET_ERR_QUEUE_FN_DUP); + } + queue_fn_curr = queue_fn_curr->next; + } + + /* name ok, so let's add it to the list */ + CHKmalloc(newetry = calloc(1, sizeof(struct queue_filename))); + CHKmalloc(newetry->filename = strdup((char*) pThis->pszFilePrefix)); + CHKmalloc(newetry->dirname = strdup(curr_dirname)); + newetry->next = queue_filename_root; + queue_filename_root = newetry; + +finalize_it: + if(iRet != RS_RET_OK) { + if(newetry != NULL) { + free((void*)newetry->filename); + free((void*)newetry); + } + } + RETiRet; +} + +void +qqueueCorrectParams(qqueue_t *pThis) +{ + int goodval; /* a "good value" to use for comparisons (different objects) */ + + if(pThis->iMaxQueueSize < 100 + && (pThis->qType == QUEUETYPE_LINKEDLIST || pThis->qType == QUEUETYPE_FIXED_ARRAY)) { + LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "Note: queue.size=\"%d\" is very " + "low and can lead to unpredictable results. See also " + "https://www.rsyslog.com/lower-bound-for-queue-sizes/", + pThis->iMaxQueueSize); + } + + /* we need to do a quick check if our water marks are set plausible. If not, + * we correct the most important shortcomings. + */ + goodval = (pThis->iMaxQueueSize / 100) * 60; + if(pThis->iHighWtrMrk != -1 && pThis->iHighWtrMrk < goodval) { + LogMsg(0, RS_RET_CONF_PARSE_WARNING, LOG_WARNING, "queue \"%s\": high water mark " + "is set quite low at %d. You should only set it below " + "60%% (%d) if you have a good reason for this.", + obj.GetName((obj_t*) pThis), pThis->iHighWtrMrk, goodval); + } + + if(pThis->iNumWorkerThreads > 1) { + goodval = (pThis->iMaxQueueSize / 100) * 10; + if(pThis->iMinMsgsPerWrkr != -1 && pThis->iMinMsgsPerWrkr < goodval) { + LogMsg(0, RS_RET_CONF_PARSE_WARNING, LOG_WARNING, "queue \"%s\": " + "queue.workerThreadMinimumMessage " + "is set quite low at %d. You should only set it below " + "10%% (%d) if you have a good reason for this.", + obj.GetName((obj_t*) pThis), pThis->iMinMsgsPerWrkr, goodval); + } + } + + if(pThis->iDiscardMrk > pThis->iMaxQueueSize) { + LogError(0, RS_RET_PARAM_ERROR, "error: queue \"%s\": " + "queue.discardMark %d is set larger than queue.size", + obj.GetName((obj_t*) pThis), pThis->iDiscardMrk); + } + + goodval = (pThis->iMaxQueueSize / 100) * 80; + if(pThis->iDiscardMrk != -1 && pThis->iDiscardMrk < goodval) { + LogMsg(0, RS_RET_CONF_PARSE_WARNING, LOG_WARNING, + "queue \"%s\": queue.discardMark " + "is set quite low at %d. You should only set it below " + "80%% (%d) if you have a good reason for this.", + obj.GetName((obj_t*) pThis), pThis->iDiscardMrk, goodval); + } + + if(pThis->pszFilePrefix != NULL) { /* This means we have a potential DA queue */ + if(pThis->iFullDlyMrk != -1 && pThis->iFullDlyMrk < pThis->iHighWtrMrk) { + LogMsg(0, RS_RET_CONF_WRN_FULLDLY_BELOW_HIGHWTR, LOG_WARNING, + "queue \"%s\": queue.fullDelayMark " + "is set below high water mark. This will result in DA mode " + " NOT being activated for full delayable messages: In many " + "cases this is a configuration error, please check if this " + "is really what you want", + obj.GetName((obj_t*) pThis)); + } + } + + /* now come parameter corrections and defaults */ + if(pThis->iHighWtrMrk < 2 || pThis->iHighWtrMrk > pThis->iMaxQueueSize) { + pThis->iHighWtrMrk = (pThis->iMaxQueueSize / 100) * 90; + if(pThis->iHighWtrMrk == 0) { /* guard against very low max queue sizes! */ + pThis->iHighWtrMrk = pThis->iMaxQueueSize; + } + } + if( pThis->iLowWtrMrk < 2 + || pThis->iLowWtrMrk > pThis->iMaxQueueSize + || pThis->iLowWtrMrk > pThis->iHighWtrMrk ) { + pThis->iLowWtrMrk = (pThis->iMaxQueueSize / 100) * 70; + if(pThis->iLowWtrMrk == 0) { + pThis->iLowWtrMrk = 1; + } + } + + if((pThis->iMinMsgsPerWrkr < 1 + || pThis->iMinMsgsPerWrkr > pThis->iMaxQueueSize) ) { + pThis->iMinMsgsPerWrkr = pThis->iMaxQueueSize / pThis->iNumWorkerThreads; + } + + if(pThis->iFullDlyMrk == -1 || pThis->iFullDlyMrk > pThis->iMaxQueueSize) { + pThis->iFullDlyMrk = (pThis->iMaxQueueSize / 100) * 97; + if(pThis->iFullDlyMrk == 0) { + pThis->iFullDlyMrk = + (pThis->iMaxQueueSize == 1) ? 1 : pThis->iMaxQueueSize - 1; + } + } + + if(pThis->iLightDlyMrk == 0) { + pThis->iLightDlyMrk = pThis->iMaxQueueSize; + } + + if(pThis->iLightDlyMrk == -1 || pThis->iLightDlyMrk > pThis->iMaxQueueSize) { + pThis->iLightDlyMrk = (pThis->iMaxQueueSize / 100) * 70; + if(pThis->iLightDlyMrk == 0) { + pThis->iLightDlyMrk = + (pThis->iMaxQueueSize == 1) ? 1 : pThis->iMaxQueueSize - 1; + } + } + + if(pThis->iDiscardMrk < 1 || pThis->iDiscardMrk > pThis->iMaxQueueSize) { + pThis->iDiscardMrk = (pThis->iMaxQueueSize / 100) * 98; + if(pThis->iDiscardMrk == 0) { + /* for very small queues, we disable this by default */ + pThis->iDiscardMrk = pThis->iMaxQueueSize; + } + } + + if(pThis->iMaxQueueSize > 0 && pThis->iDeqBatchSize > pThis->iMaxQueueSize) { + pThis->iDeqBatchSize = pThis->iMaxQueueSize; + } +} + +/* apply all params from param block to queue. Must be called before + * finalizing. This supports the v6 config system. Defaults were already + * set during queue creation. The pvals object is destructed by this + * function. + */ +rsRetVal +qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst) +{ + int i; + struct cnfparamvals *pvals; + int n_params_set = 0; + DEFiRet; + + pvals = nvlstGetParams(lst, &pblk, NULL); + if(pvals == NULL) { + parser_errmsg("error processing queue config parameters"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + if(Debug) { + dbgprintf("queue param blk:\n"); + cnfparamsPrint(&pblk, pvals); + } + for(i = 0 ; i < pblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + n_params_set++; + if(!strcmp(pblk.descr[i].name, "queue.filename")) { + pThis->pszFilePrefix = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); + pThis->lenFilePrefix = es_strlen(pvals[i].val.d.estr); + } else if(!strcmp(pblk.descr[i].name, "queue.cry.provider")) { + pThis->cryprovName = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(pblk.descr[i].name, "queue.spooldirectory")) { + free(pThis->pszSpoolDir); + pThis->pszSpoolDir = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); + pThis->lenSpoolDir = es_strlen(pvals[i].val.d.estr); + if(pThis->pszSpoolDir[pThis->lenSpoolDir-1] == '/') { + pThis->pszSpoolDir[pThis->lenSpoolDir-1] = '\0'; + --pThis->lenSpoolDir; + parser_errmsg("queue.spooldirectory must not end with '/', " + "corrected to '%s'", pThis->pszSpoolDir); + } + } else if(!strcmp(pblk.descr[i].name, "queue.size")) { + if(pvals[i].val.d.n > 0x7fffffff) { + parser_warnmsg("queue.size higher than maximum (2147483647) - " + "corrected to maximum"); + pvals[i].val.d.n = 0x7fffffff; + } else if(pvals[i].val.d.n > OVERSIZE_QUEUE_WATERMARK) { + parser_warnmsg("queue.size=%d is very large - is this " + "really intended? More info at " + "https://www.rsyslog.com/avoid-overly-large-in-memory-queues/", + (int) pvals[i].val.d.n); + } + pThis->iMaxQueueSize = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.dequeuebatchsize")) { + pThis->iDeqBatchSize = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.mindequeuebatchsize")) { + pThis->iMinDeqBatchSize = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.mindequeuebatchsize.timeout")) { + pThis->toMinDeqBatchSize = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.maxdiskspace")) { + pThis->sizeOnDiskMax = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.highwatermark")) { + pThis->iHighWtrMrk = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.lowwatermark")) { + pThis->iLowWtrMrk = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.fulldelaymark")) { + pThis->iFullDlyMrk = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.lightdelaymark")) { + pThis->iLightDlyMrk = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.discardmark")) { + pThis->iDiscardMrk = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.discardseverity")) { + pThis->iDiscardSeverity = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.checkpointinterval")) { + pThis->iPersistUpdCnt = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.syncqueuefiles")) { + pThis->bSyncQueueFiles = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.type")) { + pThis->qType = (queueType_t) pvals[i].val.d.n; + if(pThis->qType == QUEUETYPE_DIRECT) { + /* if we have a direct queue, we mimic this param was not set. + * Our prime intent is to make sure we detect when "real" params + * are set on a direct queue, and the type setting is obviously + * not relevant here. + */ + n_params_set--; + } + } else if(!strcmp(pblk.descr[i].name, "queue.workerthreads")) { + pThis->iNumWorkerThreads = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.timeoutshutdown")) { + pThis->toQShutdown = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.timeoutactioncompletion")) { + pThis->toActShutdown = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.timeoutenqueue")) { + pThis->toEnq = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.timeoutworkerthreadshutdown")) { + pThis->toWrkShutdown = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.workerthreadminimummessages")) { + pThis->iMinMsgsPerWrkr = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.maxfilesize")) { + pThis->iMaxFileSize = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.saveonshutdown")) { + pThis->bSaveOnShutdown = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.dequeueslowdown")) { + pThis->iDeqSlowdown = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.dequeuetimebegin")) { + pThis->iDeqtWinFromHr = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.dequeuetimeend")) { + pThis->iDeqtWinToHr = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.samplinginterval")) { + pThis->iSmpInterval = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "queue.takeflowctlfrommsg")) { + pThis->takeFlowCtlFromMsg = pvals[i].val.d.n; + } else { + DBGPRINTF("queue: program error, non-handled " + "param '%s'\n", pblk.descr[i].name); + } + } + + checkUniqueDiskFile(pThis); + + if(pThis->qType == QUEUETYPE_DIRECT) { + if(n_params_set > 0) { + LogMsg(0, RS_RET_OK, LOG_WARNING, "warning on queue '%s': " + "queue is in direct mode, but parameters have been set. " + "These PARAMETERS cannot be applied and WILL BE IGNORED.", + obj.GetName((obj_t*) pThis)); + } + } else if(pThis->qType == QUEUETYPE_DISK) { + if(pThis->pszFilePrefix == NULL) { + LogError(0, RS_RET_QUEUE_DISK_NO_FN, "error on queue '%s', disk mode selected, but " + "no queue file name given; queue type changed to 'linkedList'", + obj.GetName((obj_t*) pThis)); + pThis->qType = QUEUETYPE_LINKEDLIST; + } + } + + if(pThis->pszFilePrefix == NULL && pThis->cryprovName != NULL) { + LogError(0, RS_RET_QUEUE_CRY_DISK_ONLY, "error on queue '%s', crypto provider can " + "only be set for disk or disk assisted queue - ignored", + obj.GetName((obj_t*) pThis)); + free(pThis->cryprovName); + pThis->cryprovName = NULL; + } + + if(pThis->cryprovName != NULL) { + initCryprov(pThis, lst); + } + + cnfparamvalsDestruct(pvals, &pblk); +finalize_it: + RETiRet; +} + +/* return 1 if the content of two qqueue_t structs equal */ +int +queuesEqual(qqueue_t *pOld, qqueue_t *pNew) +{ + return ( + NUM_EQUALS(qType) && + NUM_EQUALS(iMaxQueueSize) && + NUM_EQUALS(iDeqBatchSize) && + NUM_EQUALS(iMinDeqBatchSize) && + NUM_EQUALS(toMinDeqBatchSize) && + NUM_EQUALS(sizeOnDiskMax) && + NUM_EQUALS(iHighWtrMrk) && + NUM_EQUALS(iLowWtrMrk) && + NUM_EQUALS(iFullDlyMrk) && + NUM_EQUALS(iLightDlyMrk) && + NUM_EQUALS(iDiscardMrk) && + NUM_EQUALS(iDiscardSeverity) && + NUM_EQUALS(iPersistUpdCnt) && + NUM_EQUALS(bSyncQueueFiles) && + NUM_EQUALS(iNumWorkerThreads) && + NUM_EQUALS(toQShutdown) && + NUM_EQUALS(toActShutdown) && + NUM_EQUALS(toEnq) && + NUM_EQUALS(toWrkShutdown) && + NUM_EQUALS(iMinMsgsPerWrkr) && + NUM_EQUALS(iMaxFileSize) && + NUM_EQUALS(bSaveOnShutdown) && + NUM_EQUALS(iDeqSlowdown) && + NUM_EQUALS(iDeqtWinFromHr) && + NUM_EQUALS(iDeqtWinToHr) && + NUM_EQUALS(iSmpInterval) && + NUM_EQUALS(takeFlowCtlFromMsg) && + USTR_EQUALS(pszFilePrefix) && + USTR_EQUALS(cryprovName) + ); +} + + +/* some simple object access methods */ +DEFpropSetMeth(qqueue, bSyncQueueFiles, int) +DEFpropSetMeth(qqueue, iPersistUpdCnt, int) +DEFpropSetMeth(qqueue, iDeqtWinFromHr, int) +DEFpropSetMeth(qqueue, iDeqtWinToHr, int) +DEFpropSetMeth(qqueue, toQShutdown, long) +DEFpropSetMeth(qqueue, toActShutdown, long) +DEFpropSetMeth(qqueue, toWrkShutdown, long) +DEFpropSetMeth(qqueue, toEnq, long) +DEFpropSetMeth(qqueue, iHighWtrMrk, int) +DEFpropSetMeth(qqueue, iLowWtrMrk, int) +DEFpropSetMeth(qqueue, iDiscardMrk, int) +DEFpropSetMeth(qqueue, iDiscardSeverity, int) +DEFpropSetMeth(qqueue, iLightDlyMrk, int) +DEFpropSetMeth(qqueue, iNumWorkerThreads, int) +DEFpropSetMeth(qqueue, iMinMsgsPerWrkr, int) +DEFpropSetMeth(qqueue, bSaveOnShutdown, int) +DEFpropSetMeth(qqueue, pAction, action_t*) +DEFpropSetMeth(qqueue, iDeqSlowdown, int) +DEFpropSetMeth(qqueue, iDeqBatchSize, int) +DEFpropSetMeth(qqueue, iMinDeqBatchSize, int) +DEFpropSetMeth(qqueue, sizeOnDiskMax, int64) +DEFpropSetMeth(qqueue, iSmpInterval, int) + + +/* This function can be used as a generic way to set properties. Only the subset + * of properties required to read persisted property bags is supported. This + * functions shall only be called by the property bag reader, thus it is static. + * rgerhards, 2008-01-11 + */ +#define isProp(name) !rsCStrSzStrCmp(pProp->pcsName, (uchar*) name, sizeof(name) - 1) +static rsRetVal qqueueSetProperty(qqueue_t *pThis, var_t *pProp) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, qqueue); + assert(pProp != NULL); + + if(isProp("iQueueSize")) { + pThis->iQueueSize = pProp->val.num; +# ifdef ENABLE_IMDIAG + iOverallQueueSize += pThis->iQueueSize; +# endif + } else if(isProp("tVars.disk.sizeOnDisk")) { + pThis->tVars.disk.sizeOnDisk = pProp->val.num; + } else if(isProp("qType")) { + if(pThis->qType != pProp->val.num) + ABORT_FINALIZE(RS_RET_QTYPE_MISMATCH); + } + +finalize_it: + RETiRet; +} +#undef isProp + +/* dummy */ +static rsRetVal qqueueQueryInterface(interface_t __attribute__((unused)) *i) { return RS_RET_NOT_IMPLEMENTED; } + +/* Initialize the stream class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-01-09 + */ +BEGINObjClassInit(qqueue, 1, OBJ_IS_CORE_MODULE) + /* request objects we use */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(strm, CORE_COMPONENT)); + CHKiRet(objUse(datetime, CORE_COMPONENT)); + CHKiRet(objUse(statsobj, CORE_COMPONENT)); + + /* now set our own handlers */ + OBJSetMethodHandler(objMethod_SETPROPERTY, qqueueSetProperty); +ENDObjClassInit(qqueue) |