summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--queue.c324
-rw-r--r--queue.h21
-rw-r--r--rsyslog.h1
-rw-r--r--syslogd.c2
4 files changed, 223 insertions, 125 deletions
diff --git a/queue.c b/queue.c
index 00f6e63f..acfdefb3 100644
--- a/queue.c
+++ b/queue.c
@@ -1,3 +1,4 @@
+// TODO: "preforked" worker threads
// TODO: do an if(debug) in dbgrintf - performanc ein release build!
// TODO: peekmsg() on first entry, with new/inprogress/deleted entry, destruction in
// call consumer state. Facilitates retaining messages in queue until action could
@@ -53,6 +54,7 @@ DEFobjStaticHelpers
rsRetVal queueChkPersist(queue_t *pThis);
static void *queueWorker(void *arg);
static rsRetVal queueGetQueueSize(queue_t *pThis, int *piQueueSize);
+static rsRetVal queueChkWrkThrdChanges(queue_t *pThis);
/* methods */
@@ -67,7 +69,7 @@ queueTellActWrkThrd(queue_t *pThis, int iIdx, qWrkCmd_t tCmd)
ISOBJ_TYPE_assert(pThis, queue);
assert(iIdx >= 0 && iIdx <= pThis->iNumWorkerThreads);
- if(pThis->pWrkThrds[iIdx].tCurrCmd >= eWRKTHRDCMD_RUN) {
+ if(pThis->pWrkThrds[iIdx].tCurrCmd >= eWRKTHRD_RUN_INIT) {
dbgprintf("Queue 0x%lx: sending command %d to thread %d\n", queueGetID(pThis), tCmd, iIdx);
pThis->pWrkThrds[iIdx].tCurrCmd = tCmd;
} else {
@@ -104,12 +106,12 @@ queueJoinWrkThrd(queue_t *pThis, int iIdx)
ISOBJ_TYPE_assert(pThis, queue);
assert(iIdx >= 0 && iIdx <= pThis->iNumWorkerThreads);
- assert(pThis->pWrkThrds[iIdx].tCurrCmd != eWRKTHRDCMD_NEVER_RAN);
+ assert(pThis->pWrkThrds[iIdx].tCurrCmd != eWRKTHRD_STOPPED);
dbgprintf("Queue 0x%lx: thread %d state %d, waiting for exit\n", queueGetID(pThis), iIdx,
pThis->pWrkThrds[iIdx].tCurrCmd);
pthread_join(pThis->pWrkThrds[iIdx].thrdID, NULL);
- pThis->pWrkThrds[iIdx].tCurrCmd = eWRKTHRDCMD_NEVER_RAN; /* back to virgin... */
+ pThis->pWrkThrds[iIdx].tCurrCmd = eWRKTHRD_STOPPED; /* back to virgin... */
dbgprintf("Queue 0x%lx: thread %d state %d, has exited\n", queueGetID(pThis), iIdx,
pThis->pWrkThrds[iIdx].tCurrCmd);
@@ -127,9 +129,9 @@ queueStrtWrkThrd(queue_t *pThis, int i)
ISOBJ_TYPE_assert(pThis, queue);
assert(i >= 0 && i <= pThis->iNumWorkerThreads);
- assert(pThis->pWrkThrds[i].tCurrCmd < eWRKTHRDCMD_RUN);
+ assert(pThis->pWrkThrds[i].tCurrCmd < eWRKTHRD_RUN_INIT);
- queueTellWrkThrd(pThis, i, eWRKTHRDCMD_RUN);
+ queueTellWrkThrd(pThis, i, eWRKTHRD_RUN_INIT);
iState = pthread_create(&(pThis->pWrkThrds[i].thrdID), NULL, queueWorker, (void*) pThis);
dbgprintf("Queue 0x%lx: Worker thread %x, index %d started with state %d.\n",
(unsigned long) pThis, (unsigned) pThis->pWrkThrds[i].thrdID, i, iState);
@@ -138,37 +140,59 @@ queueStrtWrkThrd(queue_t *pThis, int i)
}
-/* send a command to all active worker threads. A start index can be
- * given. Usually, this is 0 or 1. Thread 0 is reserved to disk-assisted
- * mode and this start index take care of the special handling it needs to
- * receive. -- rgerhards, 2008-01-16
+/* Starts a *new* worker thread. Function searches itself for a free index spot. It must only
+ * be called when we have less than max workers active. Pending wrkr thread requests MUST have
+ * been processed before calling this function. -- rgerhards, 2008-01-16
*/
static inline rsRetVal
-queueTellActWrkThrds(queue_t *pThis, int iStartIdx, qWrkCmd_t tCmd)
+queueStrtNewWrkThrd(queue_t *pThis)
{
DEFiRet;
int i;
+ int iStartingUp;
+ int iState;
ISOBJ_TYPE_assert(pThis, queue);
- assert(iStartIdx == 0 || iStartIdx == 1);
- /* tell the workers our request */
- for(i = iStartIdx ; i <= pThis->iNumWorkerThreads ; ++i)
- if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRDCMD_TERMINATED)
- queueTellActWrkThrd(pThis, i, tCmd);
+ /* find free spot in thread table. If we find at least one worker that is in initializiation,
+ * we do NOT start a new one. Let's give the other one a chance, first.
+ */
+ iStartingUp = -1;
+ for(i = 1 ; i <= pThis->iNumWorkerThreads ; ++i)
+ if(pThis->pWrkThrds[i].tCurrCmd == eWRKTHRD_STOPPED) {
+ break;
+ } else if(pThis->pWrkThrds[i].tCurrCmd == eWRKTHRD_RUN_INIT) {
+ iStartingUp = i;
+ break;
+ }
+dbgprintf("after thrd search: i %d, iStartingUp %d\n", i, iStartingUp);
+ if(iStartingUp > -1)
+ ABORT_FINALIZE(RS_RET_ALREADY_STARTING);
+
+ assert(i <= pThis->iNumWorkerThreads); /* now there must be a free spot, else something is really wrong! */
+
+ queueTellWrkThrd(pThis, i, eWRKTHRD_RUN_INIT);
+ iState = pthread_create(&(pThis->pWrkThrds[i].thrdID), NULL, queueWorker, (void*) pThis);
+ dbgprintf("Queue 0x%lx: Worker thread %x, index %d started with state %d.\n",
+ (unsigned long) pThis, (unsigned) pThis->pWrkThrds[i].thrdID, i, iState);
+ /* we try to give the starting worker a little boost. It won't help much as we still
+ * hold the queue's mutex, but at least it has a chance to start on a single-CPU system.
+ */
+ pthread_yield();
+
+finalize_it:
return iRet;
}
-/* send a command to all worker threads. A start index can be
+/* send a command to all active worker threads. A start index can be
* given. Usually, this is 0 or 1. Thread 0 is reserved to disk-assisted
* mode and this start index take care of the special handling it needs to
- * receive.
- * TODO: check if we run into trouble with inactive worker threads
+ * receive. -- rgerhards, 2008-01-16
*/
static inline rsRetVal
-queueTellWrkThrds(queue_t *pThis, int iStartIdx, qWrkCmd_t tCmd)
+queueTellActWrkThrds(queue_t *pThis, int iStartIdx, qWrkCmd_t tCmd)
{
DEFiRet;
int i;
@@ -178,25 +202,27 @@ queueTellWrkThrds(queue_t *pThis, int iStartIdx, qWrkCmd_t tCmd)
/* tell the workers our request */
for(i = iStartIdx ; i <= pThis->iNumWorkerThreads ; ++i)
- if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRDCMD_TERMINATED)
- queueTellWrkThrd(pThis, i, tCmd);
+ if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRD_TERMINATED)
+ queueTellActWrkThrd(pThis, i, tCmd);
return iRet;
}
-/* start all regular worker threads
- * rgerhards, 2008-01-15
+
+/* This once was used to start all regular worker threads. Now, we have
+ * dynamic grow of the worker thread pool, based on needs. This function is
+ * still preserved, but it now does not start all but only worker 1, which
+ * is always present.
+ * rgerhards, 2008-01-16
*/
static inline rsRetVal
queueStrtAllWrkThrds(queue_t *pThis)
{
DEFiRet;
- int i;
- /* fire up the worker threads */
- for(i = 1 ; i <= pThis->iNumWorkerThreads ; ++i) {
- queueStrtWrkThrd(pThis, i);
- }
+ ISOBJ_TYPE_assert(pThis, queue);
+ assert(pThis->pWrkThrds[1].tCurrCmd < eWRKTHRD_RUN_INIT);
+ //iRet = queueStrtWrkThrd(pThis, 1);
return iRet;
}
@@ -243,6 +269,41 @@ queueWakeupWrkThrds(queue_t *pThis, int bWithDAWrk)
}
+/* This function Checks if (another) worker threads needs to be started. It
+ * must be called while the caller holds a lock on the queue mutex. So it must not
+ * do anything that either reaquires the mutex or forces somebody else to aquire
+ * it (that would lead to a deadlock).
+ * rgerhards, 2008-01-16
+ */
+static inline rsRetVal
+queueChkAndStrtWrk(queue_t *pThis)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, queue);
+
+ /* process any pending thread requests */
+ queueChkWrkThrdChanges(pThis);
+
+ /* check if we need to start up another worker (only in regular mode) */
+ if(pThis->qRunsDA == QRUNS_REGULAR) {
+ if(pThis->iCurNumWrkThrd < pThis->iNumWorkerThreads) {
+dbgprintf("Queue %p: less than max workers are running, qsize %d, workers %d\n",
+ pThis, pThis->iQueueSize, pThis->iCurNumWrkThrd);
+ /* check if we satisfy the min nbr of messages per worker to start a new one */
+ if(pThis->iCurNumWrkThrd == 0 ||
+ pThis->iQueueSize / pThis->iCurNumWrkThrd > pThis->iMinMsgsPerWrkr) {
+ dbgprintf("Queue 0x%lx: high activity - starting additional worker thread.\n",
+ queueGetID(pThis));
+ queueStrtNewWrkThrd(pThis);
+ }
+ }
+ }
+
+ return iRet;
+}
+
+
/* --------------- code for disk-assisted (DA) queue modes -------------------- */
@@ -280,7 +341,7 @@ queueTurnOffDAMode(queue_t *pThis)
pthread_mutex_destroy(&pThis->mutDA);
pthread_cond_destroy(&pThis->condDA);
- queueTellWrkThrd(pThis, 0, eWRKTHRDCMD_SHUTDOWN_IMMEDIATE);/* finally, tell ourselves to shutdown */
+ queueTellWrkThrd(pThis, 0, eWRKTHRD_SHUTDOWN_IMMEDIATE);/* finally, tell ourselves to shutdown */
dbgprintf("Queue 0x%lx: disk-assistance has been turned off, disk queue was empty (iRet %d)\n",
queueGetID(pThis), iRet);
@@ -289,6 +350,7 @@ queueTurnOffDAMode(queue_t *pThis)
/* check if we had any worker thread changes and, if so, act
* on them. At a minimum, terminated threads are harvested (joined).
+ * This function MUST NEVER block on the queue mutex!
*/
static rsRetVal
queueChkWrkThrdChanges(queue_t *pThis)
@@ -304,14 +366,15 @@ queueChkWrkThrdChanges(queue_t *pThis)
/* go through all threads (including DA thread) */
for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
switch(pThis->pWrkThrds[i].tCurrCmd) {
- case eWRKTHRDCMD_TERMINATED:
+ case eWRKTHRD_TERMINATED:
queueJoinWrkThrd(pThis, i);
break;
/* these cases just to satisfy the compiler, we do not act an them: */
- case eWRKTHRDCMD_NEVER_RAN:
- case eWRKTHRDCMD_RUN:
- case eWRKTHRDCMD_SHUTDOWN:
- case eWRKTHRDCMD_SHUTDOWN_IMMEDIATE:
+ case eWRKTHRD_STOPPED:
+ case eWRKTHRD_RUN_INIT:
+ case eWRKTHRD_RUNNING:
+ case eWRKTHRD_SHUTDOWN:
+ case eWRKTHRD_SHUTDOWN_IMMEDIATE:
/* DO NOTHING */
break;
}
@@ -373,7 +436,7 @@ dbgprintf("Queue %p/w%d: queueDAConsumer, queue size %d\n", pThis, iMyThrdIndx,
* Note that the child queue now in almost all cases is non-empty, because we just enqueued
* a message.
*/
- if(iQueueSize <= pThis->iLowWtrMrk && pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_RUN) {
+ if(iQueueSize <= pThis->iLowWtrMrk && pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_RUNNING) {
dbgprintf("Queue 0x%lx/w%d: %d entries - passed low water mark in DA mode, sleeping\n",
queueGetID(pThis), iMyThrdIndx, iQueueSize);
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
@@ -463,7 +526,7 @@ queueStrtDA(queue_t *pThis)
* reserving worker thread 0 for DA queues. So if we would join the other
* workers, we would screw up and do against our design goal.
*/
- CHKiRet(queueTellWrkThrds(pThis, 1, eWRKTHRDCMD_SHUTDOWN_IMMEDIATE));
+ CHKiRet(queueTellActWrkThrds(pThis, 1, eWRKTHRD_SHUTDOWN_IMMEDIATE));
/* as we are right now starting DA mode because we are so busy, it is
* extremely unlikely that any worker is sleeping on empty queue. HOWEVER,
@@ -493,6 +556,24 @@ finalize_it:
}
+/* initiate DA mode
+ * rgerhards, 2008-01-16
+ */
+static inline rsRetVal
+queueInitDA(queue_t *pThis)
+{
+ DEFiRet;
+
+ /* indicate we now run in DA mode - this is reset by the DA worker if it fails */
+ pThis->qRunsDA = QRUNS_DA_INIT;
+
+ /* now we must start our DA worker thread - it does the rest of the initialization */
+ iRet = queueStrtWrkThrd(pThis, 0);
+
+ return iRet;
+}
+
+
/* check if we need to start disk assisted mode and send some signals to
* keep it running if we are already in it.
* rgerhards, 2008-01-14
@@ -522,25 +603,18 @@ queueChkStrtDA(queue_t *pThis)
pthread_mutex_unlock(&pThis->mutDA);
pthread_setcancelstate(iCancelStateSave, NULL);
queueChkWrkThrdChanges(pThis); /* the queue mode may have changed while we waited, so check! */
-
- /* we need to re-check if we run disk-assisted, because that status may have changed
- * in our high water mark processing.
- */
- if(pThis->qRunsDA != QRUNS_REGULAR)
- FINALIZE;
}
- /* if we reach this point, we are NOT currently running in DA mode.
- * TODO: split this function, I think that would make the code easier
- * to read. -- rgerhards, 2008-10-15
+ /* we need to re-check if we run disk-assisted, because that status may have changed
+ * in our high water mark processing.
*/
- dbgprintf("Queue 0x%lx: %d entries - passed high water mark for disk-assisted mode, initiating...\n",
- queueGetID(pThis), pThis->iQueueSize);
-
- pThis->qRunsDA = QRUNS_DA_INIT; /* indicate we now run in DA mode - this is reset by the DA worker if it fails */
+ if(pThis->qRunsDA == QRUNS_REGULAR) {
+ /* if we reach this point, we are NOT currently running in DA mode. */
+ dbgprintf("Queue 0x%lx: %d entries - passed high water mark for disk-assisted mode, initiating...\n",
+ queueGetID(pThis), pThis->iQueueSize);
- /* now we must start our DA worker thread - it does the rest of the initialization */
- CHKiRet(queueStrtWrkThrd(pThis, 0));
+ queueInitDA(pThis); /* initiate DA mode */
+ }
finalize_it:
return iRet;
@@ -1082,7 +1156,7 @@ queueWrkThrdCancel(queue_t *pThis)
/* first tell the workers our request */
for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i)
- if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRDCMD_TERMINATED) {
+ if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRD_TERMINATED) {
dbgprintf("Queue 0x%lx: canceling worker thread %d\n", queueGetID(pThis), i);
pthread_cancel(pThis->pWrkThrds[i].thrdID);
}
@@ -1108,7 +1182,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis)
* the function returns immediate with RS_RET_TIMED_OUT. We catch that state and accept it as
* good.
*/
- iRet = queueWrkThrdTrm(pThis, eWRKTHRDCMD_SHUTDOWN, pThis->toQShutdown);
+ iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN, pThis->toQShutdown);
if(iRet == RS_RET_TIMED_OUT) {
if(pThis->toQShutdown == 0) {
iRet = RS_RET_OK;
@@ -1116,7 +1190,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis)
/* OK, we now need to try force the shutdown */
dbgprintf("Queue 0x%lx: regular worker shutdown timed out, now trying immediate\n",
queueGetID(pThis));
- iRet = queueWrkThrdTrm(pThis, eWRKTHRDCMD_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
+ iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
}
}
@@ -1133,7 +1207,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis)
* time set (timeout == 0)! -- rgerhards, 2008-01-14
*/
for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
- if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRDCMD_NEVER_RAN) {
+ if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRD_STOPPED) {
queueJoinWrkThrd(pThis, i);
}
}
@@ -1142,8 +1216,8 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis)
* terminated now. For simplicity, we simply overwrite the states.
*/
for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
- if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRDCMD_NEVER_RAN) {
- pThis->pWrkThrds[i].tCurrCmd = eWRKTHRDCMD_TERMINATED;
+ if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRD_STOPPED) {
+ pThis->pWrkThrds[i].tCurrCmd = eWRKTHRD_TERMINATED;
}
}
@@ -1161,7 +1235,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis)
* rgerhards, 2008-01-14
*/
static inline rsRetVal
-queueWorkerCallConsumer(queue_t *pThis, int iMyThrdIndx, int iCancelStateSave)
+queueWorkerChkAndCallConsumer(queue_t *pThis, int iMyThrdIndx, int iCancelStateSave)
{
DEFiRet;
rsRetVal iRetLocal;
@@ -1172,7 +1246,10 @@ queueWorkerCallConsumer(queue_t *pThis, int iMyThrdIndx, int iCancelStateSave)
/* first check if we have still something to process */
- if(pThis->iQueueSize == 0) {
+ if(pThis->iQueueSize == 0 ||
+ ( (pThis->pWrkThrds[iMyThrdIndx].tCurrCmd != eWRKTHRD_RUNNING)
+ && (pThis->pWrkThrds[iMyThrdIndx].tCurrCmd != eWRKTHRD_SHUTDOWN)
+ )) {
pthread_mutex_unlock(pThis->mut);
pthread_setcancelstate(iCancelStateSave, NULL);
FINALIZE;
@@ -1242,6 +1319,7 @@ queueWorker(void *arg)
{
queue_t *pThis = (queue_t*) arg;
sigset_t sigSet;
+ struct timespec t;
int iMyThrdIndx; /* index for this thread in queue thread table */
int iCancelStateSave;
@@ -1267,20 +1345,32 @@ queueWorker(void *arg)
if(iMyThrdIndx == 0) { /* are we the DA worker? */
if(queueStrtDA(pThis) != RS_RET_OK) { /* then fully initialize the DA queue! */
/* if we could not init the DA queue, we have nothing to do, so shut down. */
- queueTellWrkThrd(pThis, 0, eWRKTHRDCMD_SHUTDOWN_IMMEDIATE);
+ queueTellWrkThrd(pThis, 0, eWRKTHRD_SHUTDOWN_IMMEDIATE);
}
}
+ /* finally change to RUNNING state. We need to check if we actually should still run,
+ * because someone may have requested us to shut down even before we got a chance to do
+ * our init. That would be a bad race... -- rgerhards, 2008-01-16
+ */
+ if(pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_RUN_INIT)
+ pThis->pWrkThrds[iMyThrdIndx].tCurrCmd = eWRKTHRD_RUNNING; /* we are running now! */
+
pthread_mutex_unlock(pThis->mut);
pthread_setcancelstate(iCancelStateSave, NULL);
/* end one-time stuff */
/* now we have our identity, on to real processing */
- while(pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_RUN
- || (pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_SHUTDOWN && pThis->iQueueSize > 0)) {
+ while(pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_RUNNING
+ || (pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_SHUTDOWN && pThis->iQueueSize > 0)) {
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
pthread_mutex_lock(pThis->mut);
- while(pThis->iQueueSize == 0 && pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_RUN) {
+
+ /* process any pending thread requests */
+ queueChkWrkThrdChanges(pThis);
+
+dbgprintf("Queue %p/w%d: pre empty queue, qsize %d\n", pThis, iMyThrdIndx, pThis->iQueueSize);
+ while(pThis->iQueueSize == 0 && pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_RUNNING) {
dbgprintf("Queue 0x%lx/w%d: queue EMPTY, waiting for next message.\n",
queueGetID(pThis), iMyThrdIndx);
if(pThis->bSignalOnEmpty > 0) {
@@ -1304,11 +1394,26 @@ dbgprintf("Queue %p/w%d: signaling parent empty done\n", pThis, iMyThrdIndx);
/* If we arrive here, we have the regular case, where we can safely assume that
* iQueueSize and tCmd have not changed since the while().
*/
-dbgprintf("Queue %p/w%d: pre condwait ->notEmpty\n", pThis, iMyThrdIndx);
- pthread_cond_wait(pThis->notEmpty, pThis->mut);
+dbgprintf("Queue %p/w%d: pre condwait ->notEmpty, worker shutdown %d\n", pThis, iMyThrdIndx, pThis->toWrkShutdown);
+ if(pThis->toWrkShutdown == -1) {
+dbgprintf("worker never times out!\n");
+ /* never shut down any started worker */
+ pthread_cond_wait(pThis->notEmpty, pThis->mut);
+ } else {
+ queueTimeoutComp(&t, pThis->toWrkShutdown);/* get absolute timeout */
+ if(pthread_cond_timedwait (pThis->notEmpty, pThis->mut, &t) != 0) {
+ dbgprintf("Queue 0x%lx/w%d: inactivity timeout, worker terminating...\n",
+ queueGetID(pThis), iMyThrdIndx);
+ /* we use SHUTDOWN (and not SHUTDOWN_IMMEDIATE) so that the worker
+ * does not terminate if in the mean time a new message arrived.
+ */
+ pThis->pWrkThrds[iMyThrdIndx].tCurrCmd = eWRKTHRD_SHUTDOWN;
+ }
+ }
dbgprintf("Queue %p/w%d: post condwait ->notEmpty\n", pThis, iMyThrdIndx);
}
- queueWorkerCallConsumer(pThis, iMyThrdIndx, iCancelStateSave);
+
+ queueWorkerChkAndCallConsumer(pThis, iMyThrdIndx, iCancelStateSave);
/* Now make sure we can get canceled - it is not specified if pthread_setcancelstate() is
* a cancellation point in itself. As we run most of the time without cancel enabled, I fear
@@ -1328,10 +1433,10 @@ dbgprintf("Queue %p/w%d: post condwait ->notEmpty\n", pThis, iMyThrdIndx);
* and would be very hard to debug. The yield() is a sure fix, its performance overhead
* should be well accepted given the above facts. -- rgerhards, 2008-01-10
*/
+ pthread_yield();
dbgprintf("Queue 0x%lx/w%d: end worker run, queue cmd currently %d\n",
queueGetID(pThis), iMyThrdIndx, pThis->pWrkThrds[iMyThrdIndx].tCurrCmd);
- pthread_yield();
- if(Debug && (pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_SHUTDOWN) && pThis->iQueueSize > 0)
+ if(Debug && (pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_SHUTDOWN) && pThis->iQueueSize > 0)
dbgprintf("Queue 0x%lx/w%d: worker does not yet terminate because it still has "
" %d messages to process.\n", queueGetID(pThis), iMyThrdIndx, pThis->iQueueSize);
}
@@ -1340,12 +1445,12 @@ dbgprintf("Queue 0x%lx/w%d: end worker run, queue cmd currently %d\n",
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
pthread_mutex_lock(pThis->mut);
pThis->iCurNumWrkThrd--;
- if(pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_SHUTDOWN ||
- pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRDCMD_SHUTDOWN_IMMEDIATE) {
+ if(pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_SHUTDOWN ||
+ pThis->pWrkThrds[iMyThrdIndx].tCurrCmd == eWRKTHRD_SHUTDOWN_IMMEDIATE) {
/* in shutdown case, we need to flag termination. All other commands
* have a meaning to the thread harvester, so we can not overwrite them
*/
- pThis->pWrkThrds[iMyThrdIndx].tCurrCmd = eWRKTHRDCMD_TERMINATED;
+ pThis->pWrkThrds[iMyThrdIndx].tCurrCmd = eWRKTHRD_TERMINATED;
}
pThis->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
pthread_cond_signal(&pThis->condThrdTrm); /* important for shutdown situation */
@@ -1440,7 +1545,6 @@ finalize_it:
rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */
{
DEFiRet;
- int i;
rsRetVal iRetLocal;
int bInitialized = 0; /* is queue already initialized? */
@@ -1452,61 +1556,42 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */
dbgprintf("Queue 0x%lx: type %d, disk assisted %d, maxFileSz %ld starting\n", queueGetID(pThis), pThis->qType,
pThis->bIsDA, pThis->iMaxFileSize);
- if(pThis->qType != QUEUETYPE_DIRECT) {
- if((pThis->pWrkThrds = calloc(pThis->iNumWorkerThreads + 1, sizeof(qWrkThrd_t))) == NULL)
- ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
-
- if(pThis->bIsDA) {
- /* If we are disk-assisted, we need to check if there is a QIF file
- * which we need to load. -- rgerhards, 2008-01-15
- */
- iRetLocal = queueHaveQIF(pThis);
- if(iRetLocal == RS_RET_OK) {
- dbgprintf("Queue 0x%lx: on-disk queue present, needs to be reloaded\n",
- queueGetID(pThis));
+ if(pThis->qType == QUEUETYPE_DIRECT)
+ FINALIZE; /* with direct queues, we are already finished... */
- /* indicate we now run in DA mode - this is reset by the DA worker if it fails */
- pThis->qRunsDA = QRUNS_DA_INIT;
-
- /* now we must start our DA worker thread - it does the rest of the initialization */
- CHKiRet(queueStrtWrkThrd(pThis, 0));
- bInitialized = 1;
- }
- }
+ if((pThis->pWrkThrds = calloc(pThis->iNumWorkerThreads + 1, sizeof(qWrkThrd_t))) == NULL)
+ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
- if(!bInitialized) {
- dbgprintf("Queue 0x%lx: queue starts up without loading any disk state\n", queueGetID(pThis));
- /* worker 0 is reserved for disk-assisted mode, so do not start */
- queueTellWrkThrd(pThis, 0, eWRKTHRDCMD_NEVER_RAN);
+ if(pThis->bIsDA) {
+ /* If we are disk-assisted, we need to check if there is a QIF file
+ * which we need to load. -- rgerhards, 2008-01-15
+ */
+ iRetLocal = queueHaveQIF(pThis);
+ if(iRetLocal == RS_RET_OK) {
+ dbgprintf("Queue 0x%lx: on-disk queue present, needs to be reloaded\n",
+ queueGetID(pThis));
- /* fire up the worker threads */
- for(i = 1 ; i <= pThis->iNumWorkerThreads ; ++i) {
- queueStrtWrkThrd(pThis, i);
- }
+ queueInitDA(pThis); /* initiate DA mode */
+ bInitialized = 1; /* we are done */
+ } else {
+ // TODO: use logerror? -- rgerhards, 2008-01-16
+ dbgprintf("Queue 0x%lx: error %d trying to access on-disk queue files, starting without them. "
+ "Some data may be lost\n", queueGetID(pThis), iRetLocal);
}
}
-finalize_it:
- return iRet;
-}
-
-
-#if 0
-/* persist disk status on disk. This is necessary if we run either
- * a disk queue or in a disk assisted mode.
- */
-static rsRetVal queuePersistDskFilInfo(queue_t *pThis)
-{
- DEFiRet;
-
- assert(pThis != NULL);
+ if(!bInitialized) {
+ dbgprintf("Queue 0x%lx: queue starts up without (loading) any disk state\n", queueGetID(pThis));
+ /* worker 0 is reserved for disk-assisted mode, so do not start */
+ queueTellWrkThrd(pThis, 0, eWRKTHRD_STOPPED);
+ /* fire up the worker threads */
+ queueStrtAllWrkThrds(pThis);
+ }
finalize_it:
return iRet;
}
-#endif
-
/* persist the queue to disk. If we have something to persist, we first
@@ -1619,7 +1704,7 @@ rsRetVal queueDestruct(queue_t *pThis)
* leave it for the time being. -- rgerhards, 2008-01-16
*/
if(pThis->qRunsDA != QRUNS_REGULAR)
- queueWrkThrdReqTrm(pThis->pqDA, eWRKTHRDCMD_SHUTDOWN_IMMEDIATE, 0);
+ queueWrkThrdReqTrm(pThis->pqDA, eWRKTHRD_SHUTDOWN_IMMEDIATE, 0);
/* then, terminate our own worker threads */
if(pThis->pWrkThrds != NULL) {
@@ -1760,6 +1845,9 @@ queueEnqObj(queue_t *pThis, void *pUsr)
if(pThis->bIsDA)
CHKiRet(queueChkStrtDA(pThis));
+ /* re-process any new pending thread requests and see if we need to start workers */
+ queueChkAndStrtWrk(pThis);
+
/* and finally (try to) enqueue what is left over */
while(pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize) {
dbgprintf("Queue 0x%lx: enqueueMsg: queue FULL - waiting to drain.\n", queueGetID(pThis));
@@ -1790,12 +1878,14 @@ finalize_it:
DEFpropSetMeth(queue, iPersistUpdCnt, int);
DEFpropSetMeth(queue, toQShutdown, long);
DEFpropSetMeth(queue, toActShutdown, long);
+DEFpropSetMeth(queue, toWrkShutdown, long);
DEFpropSetMeth(queue, toEnq, long);
DEFpropSetMeth(queue, iHighWtrMrk, int);
DEFpropSetMeth(queue, iLowWtrMrk, int);
DEFpropSetMeth(queue, iDiscardMrk, int);
DEFpropSetMeth(queue, iDiscardSeverity, int);
DEFpropSetMeth(queue, bIsDA, int);
+DEFpropSetMeth(queue, iMinMsgsPerWrkr, int);
/* get the size of this queue. The important thing about this get method is that it
diff --git a/queue.h b/queue.h
index 7b0970c8..03a3517b 100644
--- a/queue.h
+++ b/queue.h
@@ -60,14 +60,16 @@ typedef struct qLinkedList_S {
void *pUsr;
} qLinkedList_t;
+/* commands and states for worker threads. */
typedef enum {
- eWRKTHRDCMD_NEVER_RAN = 0, /* granted, that's more a state than a cmd - thread is dead... */
- eWRKTHRDCMD_TERMINATED = 1, /* granted, that's more a state than a cmd - thread is dead... */
- /* ALL active states MUST be numerically higher than eWRKTHRDCMD_TERMINATED and NONE must be lower! */
- eWRKTHRDCMD_RUN = 2,
- eWRKTHRDCMD_SHUTDOWN = 3,
- eWRKTHRDCMD_SHUTDOWN_IMMEDIATE = 4
-} qWrkCmd_t; /* commands for queue worker threads */
+ eWRKTHRD_STOPPED = 0, /* worker thread is not running (either actually never ran or was shut down) */
+ eWRKTHRD_TERMINATED = 1,/* worker thread has shut down, but some finalzing is still needed */
+ /* ALL active states MUST be numerically higher than eWRKTHRD_TERMINATED and NONE must be lower! */
+ eWRKTHRD_RUN_INIT = 2, /* worker thread is initializing, but not yet fully running */
+ eWRKTHRD_RUNNING = 3, /* worker thread is up and running and shall continue to do so */
+ eWRKTHRD_SHUTDOWN = 4, /* worker thread is running but shall terminate when queue is empty */
+ eWRKTHRD_SHUTDOWN_IMMEDIATE = 5/* worker thread is running but shall terminate even if queue is full */
+} qWrkCmd_t;
typedef struct qWrkThrd_s {
pthread_t thrdID; /* thread ID */
@@ -82,17 +84,18 @@ typedef struct queue_s {
int iMaxQueueSize; /* how large can the queue grow? */
int iNumWorkerThreads;/* number of worker threads to use */
int iCurNumWrkThrd;/* current number of active worker threads */
+ int iMinMsgsPerWrkr;/* minimum nbr of msgs per worker thread, if more, a new worker is started until max wrkrs */
qWrkThrd_t *pWrkThrds;/* array with control structure for the worker thread(s) associated with this queue */
int iUpdsSincePersist;/* nbr of queue updates since the last persist call */
int iPersistUpdCnt; /* persits queue info after this nbr of updates - 0 -> persist only on shutdown */
int iHighWtrMrk; /* high water mark for disk-assisted memory queues */
- //int bWasBelowHighWtr;/* when running in DA mode: queue was below high wtr mrk at least once */
int iLowWtrMrk; /* low water mark for disk-assisted memory queues */
int iDiscardMrk; /* if the queue is above this mark, low-severity messages are discarded */
int iDiscardSeverity;/* messages of this severity above are discarded on too-full queue */
int bNeedDelQIF; /* does the QIF file need to be deleted when queue becomes empty? */
int toQShutdown; /* timeout for regular queue shutdown in ms */
int toActShutdown; /* timeout for long-running action shutdown in ms */
+ int toWrkShutdown; /* timeout for idle workers in ms, -1 means indefinite (0 is immediate) */
int toEnq; /* enqueue timeout */
rsRetVal (*pConsumer)(void *); /* user-supplied consumer function for dequeued messages */
/* type-specific handlers (set during construction) */
@@ -161,11 +164,13 @@ PROTOTYPEObjClassInit(queue);
PROTOTYPEpropSetMeth(queue, iPersistUpdCnt, int);
PROTOTYPEpropSetMeth(queue, toQShutdown, long);
PROTOTYPEpropSetMeth(queue, toActShutdown, long);
+PROTOTYPEpropSetMeth(queue, toWrkShutdown, long);
PROTOTYPEpropSetMeth(queue, toEnq, long);
PROTOTYPEpropSetMeth(queue, iHighWtrMrk, int);
PROTOTYPEpropSetMeth(queue, iLowWtrMrk, int);
PROTOTYPEpropSetMeth(queue, iDiscardMrk, int);
PROTOTYPEpropSetMeth(queue, iDiscardSeverity, int);
+PROTOTYPEpropSetMeth(queue, iMinMsgsPerWrkr, int);
#define queueGetID(pThis) ((unsigned long) pThis)
#endif /* #ifndef QUEUE_H_INCLUDED */
diff --git a/rsyslog.h b/rsyslog.h
index a10cdf86..e37263e3 100644
--- a/rsyslog.h
+++ b/rsyslog.h
@@ -111,6 +111,7 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth
RS_RET_FILE_NOT_FOUND = -2040, /**< file not found */
RS_RET_TIMED_OUT = -2041, /**< timeout occured (not necessarily an error) */
RS_RET_QSIZE_ZERO = -2042, /**< queue size is zero where this is not supported */
+ RS_RET_ALREADY_STARTING = -2043, /**< something (a thread?) is already starting - not necessarily an error */
RS_RET_OK_DELETE_LISTENTRY = 1, /**< operation successful, but callee requested the deletion of an entry (special state) */
RS_RET_TERMINATE_NOW = 2, /**< operation successful, function is requested to terminate (mostly used with threads) */
RS_RET_NO_RUN = 3, /**< operation successful, but function does not like to be executed */
diff --git a/syslogd.c b/syslogd.c
index 4dcfa337..99f90402 100644
--- a/syslogd.c
+++ b/syslogd.c
@@ -3401,11 +3401,13 @@ init(void)
setQPROP(queueSetiPersistUpdCnt, "$MainMsgQueueCheckpointInterval", iMainMsgQPersistUpdCnt);
setQPROP(queueSettoQShutdown, "$MainMsgQueueTimeoutShutdown", iMainMsgQtoQShutdown );
setQPROP(queueSettoActShutdown, "$MainMsgQueueTimeoutActionCompletion", iMainMsgQtoActShutdown);
+ setQPROP(queueSettoWrkShutdown, "$MainMsgQueueTimeoutWorkerThreadShutdown", 1000); // TODO: implement config directive!
setQPROP(queueSettoEnq, "$MainMsgQueueTimeoutEnqueue", iMainMsgQtoEnq);
setQPROP(queueSetiHighWtrMrk, "$MainMsgQueueHighWaterMark", iMainMsgQHighWtrMark);
setQPROP(queueSetiLowWtrMrk, "$MainMsgQueueLowWaterMark", iMainMsgQLowWtrMark);
setQPROP(queueSetiDiscardMrk, "$MainMsgQueueDiscardMark", iMainMsgQDiscardMark);
setQPROP(queueSetiDiscardSeverity, "$MainMsgQueueDiscardSeverity", iMainMsgQDiscardSeverity);
+ setQPROP(queueSetiMinMsgsPerWrkr, "$MainMsgQueueWorkerThreadMinimumMessages", 100); // TODO: implement config directive!
# undef setQPROP
# undef setQPROPstr
generated by cgit (git 2.34.1) at 2024-11-16 02:13:13 +0000