summaryrefslogtreecommitdiffstats
path: root/runtime/queue.c
diff options
context:
space:
mode:
Diffstat (limited to 'runtime/queue.c')
-rw-r--r--runtime/queue.c178
1 files changed, 113 insertions, 65 deletions
diff --git a/runtime/queue.c b/runtime/queue.c
index bb9ea060..bb40e540 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -127,7 +127,7 @@ static struct cnfparamblk pblk =
};
/* debug aid */
-static void displayBatchState(batch_t *pBatch)
+static inline void displayBatchState(batch_t *pBatch)
{
int i;
for(i = 0 ; i < pBatch->nElem ; ++i) {
@@ -232,12 +232,19 @@ getQueueTypeName(queueType_t t)
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 = "unknown queue type";
+ break;
}
return r;
}
@@ -976,6 +983,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
{
batch_t singleBatch;
batch_obj_t batchObj;
+ sbool active = 1;
int i;
DEFiRet;
@@ -994,9 +1002,9 @@ static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr)
memset(&singleBatch, 0, sizeof(batch_t));
batchObj.state = BATCH_STATE_RDY;
batchObj.pUsrp = (obj_t*) pUsr;
- batchObj.bFilterOK = 1;
singleBatch.nElem = 1; /* there always is only one in direct mode */
singleBatch.pElem = &batchObj;
+ singleBatch.active = &active;
iRet = pThis->pConsumer(pThis->pUsr, &singleBatch, &pThis->bShutdownImmediate);
/* delete the batch string params: TODO: create its own "class" for this */
for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
@@ -1374,7 +1382,7 @@ finalize_it:
}
-/* set default inisde queue object suitable for action queues.
+/* 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.
@@ -1408,6 +1416,36 @@ qqueueSetDefaultsActionQueue(qqueue_t *pThis)
}
+/* 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->iHighWtrMrk = 45000; /* high water mark for disk-assisted queues */
+ pThis->iLowWtrMrk = 20000; /* low water mark for disk-assisted queues */
+ pThis->iDiscardMrk = 49500; /* 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 = 1500; /* queue shutdown */
+ pThis->toActShutdown = 1000; /* action shutdown (in phase 2) */
+ pThis->toEnq = 2000; /* timeout for queue enque */
+ pThis->toWrkShutdown = 60000; /* timeout for worker thread shutdown */
+ pThis->iMinMsgsPerWrkr = 1000; /* 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 */
+}
+
+
/* 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.
@@ -1596,7 +1634,6 @@ DequeueConsumableElements(qqueue_t *pThis, wti_t *pWti, int *piRemainingQueueSiz
/* all well, use this element */
pWti->batch.pElem[nDequeued].pUsrp = pUsr;
pWti->batch.pElem[nDequeued].state = BATCH_STATE_RDY;
- pWti->batch.pElem[nDequeued].bFilterOK = 1; // TODO: think again if we can handle that with more performance
++nDequeued;
}
@@ -2075,6 +2112,7 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk,
pThis->iDeqBatchSize);
+ pThis->bQueueStarted = 1;
if(pThis->qType == QUEUETYPE_DIRECT)
FINALIZE; /* with direct queues, we are already finished... */
@@ -2105,7 +2143,6 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
* the case when a disk queue has been loaded. If we did not start it here, it would never start.
*/
qqueueAdviseMaxWorkers(pThis);
- pThis->bQueueStarted = 1;
/* support statistics gathering */
qName = obj.GetName((obj_t*)pThis);
@@ -2303,73 +2340,75 @@ DoSaveOnShutdown(qqueue_t *pThis)
/* destructor for the queue object */
BEGINobjDestruct(qqueue) /* be sure to specify the object type also in END and CODESTART macros! */
CODESTARTobjDestruct(qqueue)
- /* 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)
- ShutdownWorkers(pThis);
+ 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)
+ ShutdownWorkers(pThis);
- if(pThis->bIsDA && getPhysicalQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) {
- CHKiRet(DoSaveOnShutdown(pThis));
- }
+ if(pThis->bIsDA && getPhysicalQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) {
+ CHKiRet(DoSaveOnShutdown(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);
- }
+ /* 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);
- }
+ /* 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);
- }
+ /* 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->notEmpty);
- pthread_cond_destroy(&pThis->belowFullDlyWtrMrk);
- pthread_cond_destroy(&pThis->belowLightDlyWtrMrk);
+ /* 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->notEmpty);
+ pthread_cond_destroy(&pThis->belowFullDlyWtrMrk);
+ pthread_cond_destroy(&pThis->belowLightDlyWtrMrk);
- DESTROY_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
- DESTROY_ATOMIC_HELPER_MUT(pThis->mutLogDeq);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutLogDeq);
- /* type-specific destructor */
- iRet = pThis->qDestruct(pThis);
+ /* type-specific destructor */
+ iRet = pThis->qDestruct(pThis);
+ }
free(pThis->pszFilePrefix);
free(pThis->pszSpoolDir);
@@ -2669,6 +2708,15 @@ qqueueDoCnfParams(struct nvlst *lst, struct cnfparamvals **ppvals)
return RS_RET_OK;
}
+
+/* are any queue params set at all? 1 - yes, 0 - no */
+int
+queueCnfParamsSet(struct cnfparamvals *pvals)
+{
+ return cnfparamvalsIsSet(&pblk, pvals);
+}
+
+
/* 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
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-14 17:01:16 +0000