summaryrefslogtreecommitdiffstats
path: root/queue.c
diff options
context:
space:
mode:
Diffstat (limited to 'queue.c')
-rw-r--r--queue.c1579
1 files changed, 418 insertions, 1161 deletions
diff --git a/queue.c b/queue.c
index 89e0168f..c3666db2 100644
--- a/queue.c
+++ b/queue.c
@@ -52,491 +52,23 @@
#include "stringbuf.h"
#include "srUtils.h"
#include "obj.h"
+#include "wtp.h"
+#include "wti.h"
/* static data */
DEFobjStaticHelpers
-/* debug aides */
-#if 0
-#define d_pthread_mutex_lock(x) {dbgprintf("mutex %p lock %s, %s(), line %d\n", x, __FILE__, __func__, __LINE__); \
- pthread_mutex_lock(x); \
- if(1)dbgprintf("mutex %p lock aquired %s, %s(), line %d\n", x, __FILE__, __func__, __LINE__); \
- }
-#define d_pthread_mutex_unlock(x) {dbgprintf("mutex %p UNlock %s, %s(), line %d\n", x ,__FILE__, __func__, __LINE__);\
- pthread_mutex_unlock(x); \
- if(1)dbgprintf("mutex %p UNlock done %s, %s(), line %d\n", x, __FILE__, __func__, __LINE__); \
- }
-#else
-#define d_pthread_mutex_lock(x) pthread_mutex_lock(x)
-#define d_pthread_mutex_unlock(x) pthread_mutex_unlock(x)
-#endif
-
-
/* forward-definitions */
rsRetVal queueChkPersist(queue_t *pThis);
-static void *queueWorker(void *arg);
-static rsRetVal queueChkWrkThrdChanges(queue_t *pThis);
static rsRetVal queueSetEnqOnly(queue_t *pThis, int bEnqOnly);
+static int queueChkStopWrkrDA(queue_t *pThis);
+static int queueIsIdleDA(queue_t *pThis);
+static rsRetVal queueConsumerDA(queue_t *pThis, wti_t *pWti, int iCancelStateSave);
+static rsRetVal queueConsumerCancelCleanup(void *arg1, void *arg2);
/* methods */
-/* cancellation cleanup handler - frees provided mutex
- * rgerhards, 2008-01-14
- */
-static void queueMutexCleanup(void *arg)
-{
- assert(arg != NULL);
- d_pthread_mutex_unlock((pthread_mutex_t*) arg);
-}
-
-
-/* get the current worker state. For simplicity and speed, we have
- * NOT used our regular calling interface this time. I hope that won't
- * bite in the long term... -- rgerhards, 2008-01-17
- */
-static inline qWrkCmd_t
-qWrkrGetState(qWrkThrd_t *pThis)
-{
- assert(pThis != NULL);
- return pThis->tCurrCmd;
-}
-
-
-/* indicate worker thread startup
- * (it would be best if we could do this with an atomic operation)
- * rgerhards, 2008-01-19
- */
-static void
-queueWrkrThrdStartupIndication(queue_t *pThis)
-{
- int iCancelStateSave;
-
- ISOBJ_TYPE_assert(pThis, queue);
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
- pThis->iCurNumWrkThrd++;
- d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
-}
-
-
-/* indicate worker thread shutdown
- * (it would be best if we could do this with an atomic operation)
- * rgerhards, 2008-01-19
- */
-static void
-queueWrkrThrdShutdownIndication(queue_t *pThis)
-{
- int iCancelStateSave;
-
- ISOBJ_TYPE_assert(pThis, queue);
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
- pThis->iCurNumWrkThrd--;
- pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
- d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
-}
-
-
-/* send a command to a specific thread
- */
-static rsRetVal
-qWrkrSetState(qWrkThrd_t *pThis, qWrkCmd_t tCmd)
-{
- DEFiRet;
-
- assert(pThis != NULL);
-
-dbgprintf("Queue 0x%lx: trying to send command %d to thread %d\n", queueGetID(pThis->pQueue), tCmd, pThis->iThrd);
- if(pThis->tCurrCmd == eWRKTHRD_SHUTDOWN_IMMEDIATE && tCmd != eWRKTHRD_TERMINATING)
- FINALIZE;
-
- dbgprintf("Queue 0x%lx: sending command %d to thread %d\n", queueGetID(pThis->pQueue), tCmd, pThis->iThrd);
-
- /* change some admin structures */
- switch(tCmd) {
- case eWRKTHRD_TERMINATING:
- pthread_cond_destroy(&pThis->condInitDone);
- pthread_mutex_destroy(&pThis->mut);
- dbgprintf("Queue 0x%lx/w%d: thread terminating with %d entries left in queue, %d workers running.\n",
- queueGetID(pThis->pQueue), pThis->iThrd, pThis->pQueue->iQueueSize,
- pThis->pQueue->iCurNumWrkThrd);
- break;
- case eWRKTHRD_RUN_CREATED:
- pthread_cond_init(&pThis->condInitDone, NULL);
- pthread_mutex_init(&pThis->mut, NULL);
- break;
- case eWRKTHRD_RUN_INIT:
- break;
- case eWRKTHRD_RUNNING:
- pthread_cond_signal(&pThis->condInitDone);
- break;
- /* these cases just to satisfy the compiler, we do (yet) not act an them: */
- case eWRKTHRD_STOPPED:
- case eWRKTHRD_SHUTDOWN:
- case eWRKTHRD_SHUTDOWN_IMMEDIATE:
- /* DO NOTHING */
- break;
- }
-
- pThis->tCurrCmd = tCmd;
-
-finalize_it:
- return iRet;
-}
-
-/* send a command to a specific active thread. If the thread is not
- * active, the command is not sent.
- */
-static inline rsRetVal
-queueTellActWrkThrd(queue_t *pThis, int iIdx, qWrkCmd_t tCmd)
-{
- DEFiRet;
-
- ISOBJ_TYPE_assert(pThis, queue);
- assert(iIdx >= 0 && iIdx <= pThis->iNumWorkerThreads);
-
- if(pThis->pWrkThrds[iIdx].tCurrCmd >= eWRKTHRD_RUN_CREATED) {
- qWrkrSetState(&pThis->pWrkThrds[iIdx], tCmd);
- } else {
- dbgprintf("Queue 0x%lx: command %d NOT sent to inactive thread %d\n", queueGetID(pThis), tCmd, iIdx);
- }
-
- return iRet;
-}
-
-
-/* Finalize construction of a wWrkrThrd_t "object"
- * rgerhards, 2008-01-17
- */
-static inline rsRetVal
-qWrkrConstructFinalize(qWrkThrd_t *pThis, queue_t *pQueue, int i)
-{
- assert(pThis != NULL);
- ISOBJ_TYPE_assert(pQueue, queue);
-
- dbgprintf("Queue 0x%lx: finalizing construction of worker %d instance data\n", queueGetID(pQueue), i);
-
- /* initialize our thread instance descriptor */
- pThis = pQueue->pWrkThrds + i;
- pThis->pQueue = pQueue;
- pThis->iThrd = i;
- pThis->pUsr = NULL;
-
- qWrkrSetState(pThis, eWRKTHRD_STOPPED);
-
- return RS_RET_OK;
-}
-
-
-/* Waits until the specified worker thread
- * changed to full running state (aka have started up). This function
- * MUST NOT be called while the queue mutex is locked as it does
- * this itself. The wait is without timeout.
- * rgerhards, 2008-01-17
- */
-static inline rsRetVal
-qWrkrWaitStartup(qWrkThrd_t *pThis)
-{
- int iCancelStateSave;
-
- assert(pThis != NULL);
-
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(pThis->pQueue->mut);
- if((pThis->tCurrCmd == eWRKTHRD_RUN_CREATED) || (pThis->tCurrCmd == eWRKTHRD_RUN_CREATED)) {
- dbgprintf("Queue 0x%lx: waiting on worker thread %d startup\n", queueGetID(pThis->pQueue),
- pThis->iThrd);
- pthread_cond_wait(&pThis->condInitDone, pThis->pQueue->mut);
-dbgprintf("worker startup done!\n");
- }
- d_pthread_mutex_unlock(pThis->pQueue->mut);
- pthread_setcancelstate(iCancelStateSave, NULL);
-
- return RS_RET_OK;
-}
-
-
-/* waits until all worker threads that a currently initializing are fully started up
- * rgerhards, 2008-01-18
- */
-static rsRetVal
-qWrkrWaitAllWrkrStartup(queue_t *pThis)
-{
- DEFiRet;
- int i;
-
- for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
- qWrkrWaitStartup(pThis->pWrkThrds + i);
- }
-
- return iRet;
-}
-
-
-/* initialize the qWrkThrd_t structure - this MUST be called right after
- * startup of a worker thread. -- rgerhards, 2008-01-17
- */
-static inline rsRetVal
-qWrkrInit(qWrkThrd_t **ppThis, queue_t *pQueue)
-{
- qWrkThrd_t *pThis;
- int i;
-
- assert(ppThis != NULL);
- ISOBJ_TYPE_assert(pQueue, queue);
-
- /* find myself in the queue's thread table */
- for(i = 0 ; i <= pQueue->iNumWorkerThreads ; ++i)
- if(pQueue->pWrkThrds[i].thrdID == pthread_self())
- break;
-dbgprintf("Queue %p, worker start, thrd id found %x, idx %d, self %x\n", pQueue,
- (unsigned) pQueue->pWrkThrds[i].thrdID, i, (unsigned) pthread_self());
- assert(pQueue->pWrkThrds[i].thrdID == pthread_self());
-
- /* initialize our thread instance descriptor */
- pThis = pQueue->pWrkThrds + i;
- pThis->pQueue = pQueue;
- pThis->iThrd = i;
- pThis->pUsr = NULL;
-
- *ppThis = pThis;
- qWrkrSetState(pThis, eWRKTHRD_RUN_INIT);
-
- return RS_RET_OK;
-}
-
-
-/* join a specific worker thread
- */
-static inline rsRetVal
-queueJoinWrkThrd(queue_t *pThis, int iIdx)
-{
- DEFiRet;
-
- ISOBJ_TYPE_assert(pThis, queue);
- assert(iIdx >= 0 && iIdx <= pThis->iNumWorkerThreads);
- 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);
- qWrkrSetState(&pThis->pWrkThrds[iIdx], eWRKTHRD_STOPPED); /* back to virgin... */
- pThis->pWrkThrds[iIdx].thrdID = 0; /* invalidate the thread ID so that we do not accidently find reused ones */
- dbgprintf("Queue 0x%lx: thread %d state %d, has stopped\n", queueGetID(pThis), iIdx,
- pThis->pWrkThrds[iIdx].tCurrCmd);
-
- return iRet;
-}
-
-
-/* Starts a worker thread (on a specific index [i]!)
- */
-static inline rsRetVal
-queueStrtWrkThrd(queue_t *pThis, int i)
-{
- DEFiRet;
- int iState;
-
- ISOBJ_TYPE_assert(pThis, queue);
- assert(i >= 0 && i <= pThis->iNumWorkerThreads);
- assert(pThis->pWrkThrds[i].tCurrCmd < eWRKTHRD_RUN_CREATED);
-
- qWrkrSetState(&pThis->pWrkThrds[i], eWRKTHRD_RUN_CREATED);
- iState = pthread_create(&(pThis->pWrkThrds[i].thrdID), NULL, queueWorker, (void*) pThis);
- dbgprintf("Queue 0x%lx: starting Worker thread %x, index %d with state %d.\n",
- (unsigned long) pThis, (unsigned) pThis->pWrkThrds[i].thrdID, i, iState);
-
- return iRet;
-}
-
-
-/* start the DA worker thread (if not already running)
- */
-static inline rsRetVal
-queueStrtDAWrkr(queue_t *pThis)
-{
- DEFiRet;
- int iCancelStateSave;
-
- ISOBJ_TYPE_assert(pThis, queue);
-
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->pWrkThrds[0].mut);
- if(pThis->pWrkThrds[0].tCurrCmd == eWRKTHRD_STOPPED) {
- iRet = queueStrtWrkThrd(pThis, 0);
- }
- d_pthread_mutex_unlock(&pThis->pWrkThrds[0].mut);
- pthread_setcancelstate(iCancelStateSave, NULL);
-
- return iRet;
-}
-
-/* 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
-queueStrtNewWrkThrd(queue_t *pThis)
-{
- DEFiRet;
- int i;
- int iStartingUp;
- int iState;
-
- ISOBJ_TYPE_assert(pThis, queue);
-
- /* 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) {
-dbgprintf("Queue %p: search thrd tbl slot: i %d, CuccCmd %d\n", pThis, i, pThis->pWrkThrds[i].tCurrCmd);
- if(pThis->pWrkThrds[i].tCurrCmd == eWRKTHRD_STOPPED) {
- break;
- } else if(pThis->pWrkThrds[i].tCurrCmd == eWRKTHRD_RUN_CREATED) {
- iStartingUp = i;
- break;
- }
- }
-
-dbgprintf("Queue %p: after thrd search: i %d, iStartingUp %d\n", pThis, 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! */
-
- qWrkrSetState(&pThis->pWrkThrds[i], eWRKTHRD_RUN_CREATED);
- 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 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
- */
-static inline rsRetVal
-queueTellActWrkThrds(queue_t *pThis, int iStartIdx, qWrkCmd_t tCmd)
-{
- DEFiRet;
- int i;
-
- 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 >= eWRKTHRD_TERMINATING)
- queueTellActWrkThrd(pThis, i, tCmd);
-
- return iRet;
-}
-
-
-/* compute an absolute time timeout suitable for calls to pthread_cond_timedwait()
- * rgerhards, 2008-01-14
- */
-static rsRetVal
-queueTimeoutComp(struct timespec *pt, int iTimeout)
-{
- assert(pt != NULL);
- /* compute timeout */
- clock_gettime(CLOCK_REALTIME, pt);
- pt->tv_nsec += (iTimeout % 1000) * 1000000; /* think INTEGER arithmetic! */
- if(pt->tv_nsec > 999999999) { /* overrun? */
- pt->tv_nsec -= 1000000000;
- ++pt->tv_sec;
- }
- pt->tv_sec += iTimeout / 1000;
- return RS_RET_OK; /* so far, this is static... */
-}
-
-
-/* wake up all worker threads. Param bWithDAWrk tells if the DA worker
- * is to be awaken, too. It needs special handling because it waits on
- * two different conditions depending on processing state.
- * rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueWakeupWrkThrds(queue_t *pThis, int bWithDAWrk)
-{
- DEFiRet;
-
- ISOBJ_TYPE_assert(pThis, queue);
-
- pthread_cond_broadcast(&pThis->notEmpty);
- if(bWithDAWrk && pThis->qRunsDA != QRUNS_REGULAR) {
- /* if running disk-assisted, workers may wait on that condition, too */
- pthread_cond_broadcast(&pThis->condDA);
- }
-
- return iRet;
-}
-
-
-/* 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;
- int iCancelStateSave;
-
- ISOBJ_TYPE_assert(pThis, queue);
-
- /* process any pending thread requests */
- queueChkWrkThrdChanges(pThis);
-
- if(pThis->bEnqOnly == 1)
- FINALIZE; /* in enqueue-only mode we have no workers */
-
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
- pthread_cleanup_push(queueMutexCleanup, &pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
- /* check if we need to start up another worker */
- if(pThis->qRunsDA == QRUNS_REGULAR) {
- if(pThis->iCurNumWrkThrd < pThis->iNumWorkerThreads) {
-dbgprintf("Queue %p: less than max workers are running, qsize %d, workers %d, qRunsDA: %d\n",
- pThis, pThis->iQueueSize, pThis->iCurNumWrkThrd, pThis->qRunsDA);
- /* 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);
- }
- }
- } else {
- if(pThis->iCurNumWrkThrd == 0 && pThis->bEnqOnly == 0) {
-dbgprintf("Queue %p: DA worker is no longer running, restarting, qsize %d, workers %d, qRunsDA: %d\n",
- pThis, pThis->iQueueSize, pThis->iCurNumWrkThrd, pThis->qRunsDA);
- /* DA worker has timed out and needs to be restarted */
- iRet = queueStrtDAWrkr(pThis);
- }
- }
- pthread_cleanup_pop(1);
-
-finalize_it:
- return iRet;
-}
-
/* --------------- code for disk-assisted (DA) queue modes -------------------- */
@@ -555,7 +87,7 @@ queueTurnOffDAMode(queue_t *pThis)
DEFiRet;
ISOBJ_TYPE_assert(pThis, queue);
- assert(pThis->qRunsDA != QRUNS_REGULAR);
+ assert(pThis->bRunsDA);
/* if we need to pull any data that we still need from the (child) disk queue,
* now would be the time to do so. At present, we do not need this, but I'd like to
@@ -572,62 +104,18 @@ queueTurnOffDAMode(queue_t *pThis)
dbgprintf("Queue 0x%lx: disk-assistance being been turned off, bEnqOnly %d, bQueInDestr %d, NumWrkd %d\n",
queueGetID(pThis),
pThis->bEnqOnly,pThis->bQueueInDestruction,pThis->iCurNumWrkThrd);
- // TODO: think about this code - there is a race
- if(pThis->bEnqOnly == 0 && pThis->bQueueInDestruction == 0 && pThis->iCurNumWrkThrd < 2)
- queueStrtNewWrkThrd(pThis);
- pThis->qRunsDA = QRUNS_REGULAR; /* tell the world we are back in non-DA mode */
+ // TODO: mutex?
+ pThis->bRunsDA = 0; /* tell the world we are back in non-DA mode */
/* we destruct the queue object, which will also shutdown the queue worker. As the queue is empty,
* this will be quick.
*/
queueDestruct(&pThis->pqDA); /* and now we are ready to destruct the DA queue */
- /* now free the remaining resources */
- pthread_mutex_destroy(&pThis->mutDA);
- pthread_cond_destroy(&pThis->condDA);
-
- queueTellActWrkThrd(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);
- return iRet;
-}
-
-/* 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)
-{
- DEFiRet;
- int i;
-
- ISOBJ_TYPE_assert(pThis, queue);
-
- if(pThis->bThrdStateChanged == 0)
- FINALIZE;
-
- /* go through all threads (including DA thread) */
- for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
- switch(pThis->pWrkThrds[i].tCurrCmd) {
- case eWRKTHRD_TERMINATING:
- queueJoinWrkThrd(pThis, i);
- break;
- /* these cases just to satisfy the compiler, we do not act an them: */
- case eWRKTHRD_STOPPED:
- case eWRKTHRD_RUN_CREATED:
- case eWRKTHRD_RUN_INIT:
- case eWRKTHRD_RUNNING:
- case eWRKTHRD_SHUTDOWN:
- case eWRKTHRD_SHUTDOWN_IMMEDIATE:
- /* DO NOTHING */
- break;
- }
- }
-
-finalize_it:
- return iRet;
+ RETiRet;
}
@@ -644,6 +132,7 @@ queueChkIsDA(queue_t *pThis)
DEFiRet;
ISOBJ_TYPE_assert(pThis, queue);
+RUNLOG_VAR("%s", pThis->pszFilePrefix);
if(pThis->pszFilePrefix != NULL) {
pThis->bIsDA = 1;
dbgprintf("Queue 0x%lx: is disk-assisted, disk will be used on demand\n", queueGetID(pThis));
@@ -651,7 +140,7 @@ queueChkIsDA(queue_t *pThis)
dbgprintf("Queue 0x%lx: is NOT disk-assisted\n", queueGetID(pThis));
}
- return iRet;
+ RETiRet;
}
@@ -660,11 +149,11 @@ queueChkIsDA(queue_t *pThis)
* chore of this function is to create the DA queue object. If that function fails,
* the DA worker should return with an appropriate state, which in turn should lead to
* a re-set to non-DA mode in the Enq process. The queue mutex must be locked when this
- * function is called, else a race on pThis->qRunsDA may happen.
+ * function is called, else a race on pThis->bRunsDA may happen.
* rgerhards, 2008-01-15
*/
static rsRetVal
-queueStrtDA(queue_t *pThis)
+queueStartDA(queue_t *pThis)
{
DEFiRet;
@@ -682,30 +171,19 @@ dbgprintf("Queue %p: queueSTrtDA after child queue construct, q %p\n", pThis, pT
/* as the created queue is the same object class, we take the
* liberty to access its properties directly.
*/
- pThis->pqDA->condSignalOnEmpty = &pThis->condDA;
- pThis->pqDA->mutSignalOnEmpty = &pThis->mutDA;
- pThis->pqDA->condSignalOnEmpty2 = &pThis->notEmpty;
- pThis->pqDA->bSignalOnEmpty = 2;
pThis->pqDA->pqParent = pThis;
-dbgprintf("Queue %p: queueSTrtDA after assign\n", pThis);
CHKiRet(queueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize));
CHKiRet(queueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix));
CHKiRet(queueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt));
CHKiRet(queueSettoActShutdown(pThis->pqDA, pThis->toActShutdown));
CHKiRet(queueSettoEnq(pThis->pqDA, pThis->toEnq));
-dbgprintf("Queue %p: queueSTrtDA 10\n", pThis);
CHKiRet(queueSetEnqOnly(pThis->pqDA, pThis->bDAEnqOnly));
-dbgprintf("Queue %p: queueSTrtDA 15\n", pThis);
CHKiRet(queueSetiHighWtrMrk(pThis->pqDA, 0));
-dbgprintf("Queue %p: queueSTrtDA 20\n", pThis);
CHKiRet(queueSetiDiscardMrk(pThis->pqDA, 0));
-dbgprintf("Queue %p: queueSTrtDA 25\n", pThis);
if(pThis->toQShutdown == 0) {
-dbgprintf("Queue %p: queueSTrtDA 30a\n", pThis);
CHKiRet(queueSettoQShutdown(pThis->pqDA, 0)); /* if the user really wants... */
} else {
-dbgprintf("Queue %p: queueSTrtDA 30b\n", pThis);
/* we use the shortest possible shutdown (0 is endless!) because when we run on disk AND
* have an obviously large backlog, we can't finish it in any case. So there is no point
* in holding shutdown longer than necessary. -- rgerhards, 2008-01-15
@@ -713,30 +191,21 @@ dbgprintf("Queue %p: queueSTrtDA 30b\n", pThis);
CHKiRet(queueSettoQShutdown(pThis->pqDA, 1));
}
-dbgprintf("Queue %p: queueSTrtDA pre start\n", pThis);
+dbgprintf("Queue %p: queueStartDA pre start\n", pThis);
iRet = queueStart(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)
FINALIZE; /* something is wrong */
- /* tell our fellow workers to shut down
- * NOTE: we do NOT join them by intension! If we did, we would hold draining
- * the queue until some potentially long-running actions are finished. Having
- * the ability to immediatly drain the queue was the primary intension of
- * 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(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,
+ * extremely unlikely that any regular worker is sleeping on empty queue. HOWEVER,
* we want to be on the safe side, and so we awake anyone that is waiting
* on one. So even if the scheduler plays badly with us, things should be
* quite well. -- rgerhards, 2008-01-15
*/
- queueWakeupWrkThrds(pThis, 0); /* awake all workers, but not ourselves ;) */
+ wtpWakeupWrkr(pThis->pWtpReg); /* awake all workers, but not ourselves ;) */
- pThis->qRunsDA = QRUNS_DA; /* we are now in DA mode! */
+ pThis->bRunsDA = 1; /* we are now in DA mode! */
dbgprintf("Queue 0x%lx: is now running in disk assisted mode, disk queue 0x%lx\n",
queueGetID(pThis), queueGetID(pThis->pqDA));
@@ -751,31 +220,63 @@ finalize_it:
pThis->bIsDA = 0;
}
- return iRet;
+ 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 inline rsRetVal
-queueInitDA(queue_t *pThis, int bEnqOnly)
+queueInitDA(queue_t *pThis, int bEnqOnly, int bLockMutex)
{
DEFiRet;
+ DEFVARS_mutexProtection;
+ uchar pszBuf[64];
+ size_t lenBuf;
+
+ BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, bLockMutex);
+ /* 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
+ */
+ if(pThis->pWtpDA == NULL) {
+ lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "Queue 0x%lx/DA", (unsigned long) pThis);
+ CHKiRet(wtpConstruct (&pThis->pWtpDA));
+ CHKiRet(wtpSetDbgHdr (pThis->pWtpDA, pszBuf, lenBuf));
+ CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpDA, queueChkStopWrkrDA));
+ CHKiRet(wtpSetpfIsIdle (pThis->pWtpDA, queueIsIdleDA));
+ CHKiRet(wtpSetpfDoWork (pThis->pWtpDA, queueConsumerDA));
+ CHKiRet(wtpSetpfOnWorkerCancel (pThis->pWtpDA, queueConsumerCancelCleanup));
+ CHKiRet(wtpSetpfOnWorkerStartup (pThis->pWtpDA, queueStartDA));
+ CHKiRet(wtpSetpfOnWorkerShutdown(pThis->pWtpDA, queueTurnOffDAMode));
+ CHKiRet(wtpSetpmutUsr (pThis->pWtpDA, pThis->mut));
+ CHKiRet(wtpSetpcondBusy (pThis->pWtpDA, &pThis->notEmpty));
+ CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpDA, 1));
+ CHKiRet(wtpSetpUsr (pThis->pWtpDA, pThis));
+ CHKiRet(wtpConstructFinalize (pThis->pWtpDA));
+ }
+ /* if we reach this point, we have a "good" DA worker pool */
/* indicate we now run in DA mode - this is reset by the DA worker if it fails */
- pThis->qRunsDA = QRUNS_DA_INIT;
+ pThis->bRunsDA = 1;
pThis->bDAEnqOnly = bEnqOnly;
- /* now we must start our DA worker thread - it does the rest of the initialization
- * In enqueue-only mode, we do not start any workers.
+ /* now we must now adivse the wtp that we need one worker. If none is yet active,
+ * that will also start one up. If we forgot that step, everything would be stalled
+ * until the next enqueue request.
*/
if(pThis->bEnqOnly == 0)
- iRet = queueStrtDAWrkr(pThis);
+ wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* one worker only for disk queues! */
- return iRet;
+finalize_it:
+ END_MTX_PROTECTED_OPERATIONS(pThis->mut);
+ RETiRet;
}
@@ -783,7 +284,7 @@ queueInitDA(queue_t *pThis, int bEnqOnly)
* keep it running if we are already in it.
* rgerhards, 2008-01-14
*/
-static rsRetVal
+static inline rsRetVal
queueChkStrtDA(queue_t *pThis)
{
DEFiRet;
@@ -794,35 +295,31 @@ queueChkStrtDA(queue_t *pThis)
if(pThis->iQueueSize != pThis->iHighWtrMrk)
ABORT_FINALIZE(RS_RET_OK);
- if(pThis->qRunsDA != QRUNS_REGULAR) {
+dbgprintf("Queue %p: chkStartDA\n", pThis);
+ if(pThis->bRunsDA) {
/* then we need to signal that we are at the high water mark again. If that happens
* on our way down the queue, that doesn't matter, because then nobody is waiting
* on the condition variable.
+ * (Remember that a DA queue stops draining the queue once it has reached the low
+ * water mark and restarts it when the high water mark is reached again - this is
+ * what this code here is responsible for. Please note that all workers may have been
+ * terminated due to the inactivity timeout, thus we need to advise the pool that
+ * we need at least one).
*/
dbgprintf("Queue 0x%lx: %d entries - passed high water mark in DA mode, send notify\n",
queueGetID(pThis), pThis->iQueueSize);
- //pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- // TODO: mutex call order check! this must aquire the queue mutex
- //d_pthread_mutex_lock(&pThis->mutDA);
- pthread_cond_signal(&pThis->condDA);
- //d_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) {
- /* if we reach this point, we are NOT currently running in DA mode. */
+ wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* run again [see comment above] ;) */
+ } else {
+ /* this is the case when we are currently not running in DA mode. So it is time
+ * to turn it back on.
+ */
dbgprintf("Queue 0x%lx: %d entries - passed high water mark for disk-assisted mode, initiating...\n",
queueGetID(pThis), pThis->iQueueSize);
-
- queueInitDA(pThis, QUEUE_MODE_ENQDEQ); /* initiate DA mode */
+ queueInitDA(pThis, QUEUE_MODE_ENQDEQ, MUTEX_ALREADY_LOCKED); /* initiate DA mode */
}
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -855,7 +352,7 @@ static rsRetVal qConstructFixedArray(queue_t *pThis)
queueChkIsDA(pThis);
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -868,7 +365,7 @@ static rsRetVal qDestructFixedArray(queue_t *pThis)
if(pThis->tVars.farray.pBuf != NULL)
free(pThis->tVars.farray.pBuf);
- return iRet;
+ RETiRet;
}
static rsRetVal qAddFixedArray(queue_t *pThis, void* in)
@@ -881,7 +378,7 @@ static rsRetVal qAddFixedArray(queue_t *pThis, void* in)
if (pThis->tVars.farray.tail == pThis->iMaxQueueSize)
pThis->tVars.farray.tail = 0;
- return iRet;
+ RETiRet;
}
static rsRetVal qDelFixedArray(queue_t *pThis, void **out)
@@ -895,7 +392,7 @@ static rsRetVal qDelFixedArray(queue_t *pThis, void **out)
if (pThis->tVars.farray.head == pThis->iMaxQueueSize)
pThis->tVars.farray.head = 0;
- return iRet;
+ RETiRet;
}
@@ -911,7 +408,7 @@ static rsRetVal qConstructLinkedList(queue_t *pThis)
queueChkIsDA(pThis);
- return iRet;
+ RETiRet;
}
@@ -925,7 +422,7 @@ static rsRetVal qDestructLinkedList(queue_t __attribute__((unused)) *pThis)
* dynamic left with the linked list.
*/
- return iRet;
+ RETiRet;
}
static rsRetVal qAddLinkedList(queue_t *pThis, void* pUsr)
@@ -949,7 +446,7 @@ static rsRetVal qAddLinkedList(queue_t *pThis, void* pUsr)
}
finalize_it:
- return iRet;
+ RETiRet;
}
static rsRetVal qDelLinkedList(queue_t *pThis, void **ppUsr)
@@ -971,7 +468,7 @@ static rsRetVal qDelLinkedList(queue_t *pThis, void **ppUsr)
}
free(pEntry);
- return iRet;
+ RETiRet;
}
@@ -986,7 +483,7 @@ queueLoadPersStrmInfoFixup(strm_t *pStrm, queue_t *pThis)
ISOBJ_TYPE_assert(pThis, queue);
CHKiRet(strmSetDir(pStrm, glblGetWorkDir(), strlen((char*)glblGetWorkDir())));
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -1024,7 +521,7 @@ queueHaveQIF(queue_t *pThis)
/* If we reach this point, we have a .qi file */
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -1092,7 +589,7 @@ finalize_it:
queueGetID(pThis), iRet);
}
- return iRet;
+ RETiRet;
}
@@ -1150,7 +647,7 @@ CHKiRet(strmSetiMaxFileSize(pThis->tVars.disk.pWrite, pThis->iMaxFileSize));
CHKiRet(strmSetiMaxFileSize(pThis->tVars.disk.pRead, pThis->iMaxFileSize));
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -1166,7 +663,7 @@ static rsRetVal qDestructDisk(queue_t *pThis)
if(pThis->pszSpoolDir != NULL)
free(pThis->pszSpoolDir);
- return iRet;
+ RETiRet;
}
static rsRetVal qAddDisk(queue_t *pThis, void* pUsr)
@@ -1179,7 +676,7 @@ static rsRetVal qAddDisk(queue_t *pThis, void* pUsr)
CHKiRet(strmFlush(pThis->tVars.disk.pWrite));
finalize_it:
- return iRet;
+ RETiRet;
}
static rsRetVal qDelDisk(queue_t *pThis, void **ppUsr)
@@ -1216,7 +713,7 @@ static rsRetVal qAddDirect(queue_t *pThis, void* pUsr)
* queue anything ;)
*/
- return iRet;
+ RETiRet;
}
static rsRetVal qDelDirect(queue_t __attribute__((unused)) *pThis, __attribute__((unused)) void **out)
@@ -1228,6 +725,7 @@ static rsRetVal qDelDirect(queue_t __attribute__((unused)) *pThis, __attribute__
/* --------------- end type-specific handlers -------------------- */
+
/* generic code to add a queue entry */
static rsRetVal
queueAdd(queue_t *pThis, void *pUsr)
@@ -1242,7 +740,7 @@ queueAdd(queue_t *pThis, void *pUsr)
dbgprintf("Queue 0x%lx: entry added, size now %d entries\n", queueGetID(pThis), pThis->iQueueSize);
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -1265,121 +763,31 @@ queueDel(queue_t *pThis, void *pUsr)
dbgprintf("Queue 0x%lx: entry deleted, state %d, size now %d entries\n",
queueGetID(pThis), iRet, pThis->iQueueSize);
- return iRet;
-}
-
-
-/* Send a shutdown command to all workers and awake them. This function
- * does NOT wait for them to terminate. Set bIncludeDAWRk to send the
- * termination command to the DA worker, too (else this does not happen).
- * rgerhards, 2008-01-16
- */
-static inline rsRetVal
-queueWrkThrdReqTrm(queue_t *pThis, qWrkCmd_t tShutdownCmd, int bIncludeDAWrk)
-{
- DEFiRet;
-
- if(bIncludeDAWrk) {
- queueTellActWrkThrds(pThis, 0, tShutdownCmd);/* first tell the workers our request */
- queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
- } else {
- queueTellActWrkThrds(pThis, 1, tShutdownCmd);/* first tell the workers our request */
- queueWakeupWrkThrds(pThis, 0); /* awake all workers but not DA-worker */
- }
-
- return iRet;
-}
-
-
-/* Send a shutdown command to all workers and see if they terminate.
- * A timeout may be specified.
- * rgerhards, 2008-01-14
- */
-static rsRetVal
-queueWrkThrdTrm(queue_t *pThis, qWrkCmd_t tShutdownCmd, long iTimeout)
-{
- DEFiRet;
- int bTimedOut;
- struct timespec t;
- int iCancelStateSave;
-
- queueTellActWrkThrds(pThis, 0, tShutdownCmd);/* first tell the workers our request */
- queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
- /* race: must make sure all are running! */
- queueTimeoutComp(&t, iTimeout);/* get timeout */
-
- /* and wait for their termination */
-dbgprintf("Queue %p: waiting for mutex %p\n", pThis, &pThis->mutThrdMgmt);
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
- pthread_cleanup_push(queueMutexCleanup, &pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
- bTimedOut = 0;
- while(pThis->iCurNumWrkThrd > 0 && !bTimedOut) {
- dbgprintf("Queue 0x%lx: waiting %ldms on worker thread termination, %d still running\n",
- queueGetID(pThis), iTimeout, pThis->iCurNumWrkThrd);
-
- if(pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mutThrdMgmt, &t) != 0) {
- dbgprintf("Queue 0x%lx: timeout waiting on worker thread termination\n", queueGetID(pThis));
- bTimedOut = 1; /* we exit the loop on timeout */
- }
- }
- pthread_cleanup_pop(1);
-
- if(bTimedOut)
- iRet = RS_RET_TIMED_OUT;
-
- return iRet;
+ RETiRet;
}
-/* Unconditionally cancel all running worker threads.
- * rgerhards, 2008-01-14
- */
-static rsRetVal
-queueWrkThrdCancel(queue_t *pThis)
-{
- DEFiRet;
- int i;
- // TODO: we need to implement peek(), without it (today!) we lose one message upon
- // worker cancellation! -- rgerhards, 2008-01-14
-
- /* process any pending thread requests so that we know who actually is still running */
- queueChkWrkThrdChanges(pThis);
-
- /* awake the workers one more time, just to be sure */
- queueWakeupWrkThrds(pThis, 1); /* awake all workers including DA-worker */
-
- /* first tell the workers our request */
- for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
- if(pThis->pWrkThrds[i].tCurrCmd >= eWRKTHRD_TERMINATING) {
- dbgprintf("Queue 0x%lx: canceling worker thread %d\n", queueGetID(pThis), i);
- pthread_cancel(pThis->pWrkThrds[i].thrdID);
- }
- }
-
- return iRet;
-}
-
-
-/* Worker thread management function carried out when the main
- * worker is about to terminate.
+/* This function shuts down all worker threads and waits until they
+ * have terminated. If they timeout, they are cancelled. Parameters have been set
+ * before this function is called so that DA queues will be fully persisted to
+ * disk (if configured to do so).
+ * rgerhards, 2008-01-24
*/
static rsRetVal queueShutdownWorkers(queue_t *pThis)
{
DEFiRet;
int i;
- assert(pThis != NULL);
+ ISOBJ_TYPE_assert(pThis, queue);
- dbgprintf("Queue 0x%lx: initiating worker thread shutdown sequence\n", (unsigned long) pThis);
+ dbgprintf("Queue 0x%lx: initiating worker thread shutdown sequence\n", queueGetID(pThis));
/* even if the timeout count is set to 0 (run endless), we still call the queueWrkThrdTrm(). This
* is necessary so that all threads get sent the termination command. With a timeout of 0, however,
* the function returns immediate with RS_RET_TIMED_OUT. We catch that state and accept it as
* good.
*/
- iRet = queueWrkThrdTrm(pThis, eWRKTHRD_SHUTDOWN, pThis->toQShutdown);
+ wtpShutdownAll(pThis->pWtpReg, wtpState_SHUTDOWN, pThis->toQShutdown);
if(iRet == RS_RET_TIMED_OUT) {
if(pThis->toQShutdown == 0) {
iRet = RS_RET_OK;
@@ -1387,418 +795,69 @@ 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, eWRKTHRD_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
+ iRet = wtpShutdownAll(pThis->pWtpReg, wtpState_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
}
}
if(iRet != RS_RET_OK) { /* this is true on actual error on first try or timeout and error on second */
/* still didn't work out - so we now need to cancel the workers */
dbgprintf("Queue 0x%lx: worker threads could not be shutdown, now canceling them\n", (unsigned long) pThis);
- iRet = queueWrkThrdCancel(pThis);
+ iRet = wtpCancelAll(pThis->pWtpReg);
+ }
+
+ // TODO: do it just once but right ;)
+ if(pThis->pWtpDA != NULL) {
+ wtpShutdownAll(pThis->pWtpDA, pThis->toQShutdown, pThis->toQShutdown);
+ if(iRet == RS_RET_TIMED_OUT) {
+ if(pThis->toQShutdown == 0) {
+ iRet = RS_RET_OK;
+ } else {
+ /* 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 = wtpShutdownAll(pThis->pWtpDA, wtpState_SHUTDOWN_IMMEDIATE, pThis->toActShutdown);
+ }
+ }
+
+ if(iRet != RS_RET_OK) { /* this is true on actual error on first try or timeout and error on second */
+ /* still didn't work out - so we now need to cancel the workers */
+ dbgprintf("Queue 0x%lx: worker threads could not be shutdown, now canceling them\n", (unsigned long) pThis);
+ iRet = wtpCancelAll(pThis->pWtpDA);
+ }
}
+
/* finally join the threads
* In case of a cancellation, this may actually take some time. This is also
* needed to clean up the thread descriptors, even with a regular termination.
* And, most importantly, this is needed if we have an indifitite termination
* time set (timeout == 0)! -- rgerhards, 2008-01-14
*/
+#if 0 // totally wrong, we must implement something along these lines in wtp!
+RUNLOG;
for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
- if(pThis->pWrkThrds[i].tCurrCmd != eWRKTHRD_STOPPED) {
- queueJoinWrkThrd(pThis, i);
+ if(pThis->pWtpReg->pWrkr[i]->tCurrCmd != eWRKTHRD_STOPPED) {
+ wtiJoinThrd(pThis->pWtpReg->pWrkr[i]);
}
}
- dbgprintf("Queue 0x%lx: worker threads terminated, remaining queue size %d.\n",
- queueGetID(pThis), pThis->iQueueSize);
-
- return iRet;
-}
-
-/* This is a special consumer to feed the disk-queue in disk-assited 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).
- * rgerhards, 2008-01-14
- */
-static inline rsRetVal
-queueDAConsumer(queue_t *pThis, qWrkThrd_t *pWrkrInst)
-{
- DEFiRet;
- int iCancelStateSave;
-
- ISOBJ_TYPE_assert(pThis, queue);
- assert(pThis->qRunsDA != QRUNS_REGULAR);
- ISOBJ_assert(pWrkrInst->pUsr);
-
-dbgprintf("Queue %p/w%d: queueDAConsumer, queue size %d\n", pThis, pWrkrInst->iThrd, pThis->iQueueSize);/* dirty iQueueSize! */
- CHKiRet(queueEnqObj(pThis->pqDA, pWrkrInst->pUsr));
-
- /* We check if we reached the low water mark (but only if we are not in shutdown mode)
- * Note that the child queue now in almost all cases is non-empty, because we just enqueued
- * a message. Note that we need a quick check below to see if we are still in running state.
- * If not, we do not go into the wait, because that's not a good thing to do. We do not
- * do a full termination check, as this is done when we go back to the main worker loop.
- * We need to re-aquire the queue mutex here, because we need to have a consistent
- * access to the queue's admin data.
- */
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-dbgprintf("pre mutex lock (think about CLEANUP!)\n");
- d_pthread_mutex_lock(pThis->mut);
- pthread_cleanup_push(queueMutexCleanup, pThis->mut);
- pthread_setcancelstate(iCancelStateSave, NULL);
-dbgprintf("mutex locked (think about CLEANUP!)\n");
- if(pThis->iQueueSize <= pThis->iLowWtrMrk && pWrkrInst->tCurrCmd == eWRKTHRD_RUNNING) {
- dbgprintf("Queue 0x%lx/w%d: %d entries - passed low water mark in DA mode, sleeping\n",
- queueGetID(pThis), pWrkrInst->iThrd, pThis->iQueueSize);
- /* wait for either passing the high water mark or the child disk queue drain */
- pthread_cond_wait(&pThis->condDA, pThis->mut);
- }
- pthread_cleanup_pop(1); /* release mutex in an atomic way via cleanup handler */
-
-finalize_it:
-dbgprintf("DAConsumer returns with iRet %d\n", iRet);
- return iRet;
-}
-
-
-/* This is a helper for queueWorker () it either calls the configured
- * consumer or the DA-consumer (if in disk-assisted mode). It is
- * protected by the queue mutex, but MUST release it as soon as possible.
- * Most importantly, it must release it before the consumer is called.
- * rgerhards, 2008-01-14
- */
-static inline rsRetVal
-queueWorkerChkAndCallConsumer(queue_t *pThis, qWrkThrd_t *pWrkrInst, int iCancelStateSave)
-{
- DEFiRet;
- rsRetVal iRetLocal;
- int iSeverity;
- int iQueueSize;
- void *pUsr;
- int qRunsDA;
- int iMyThrdIndx;
-
- ISOBJ_TYPE_assert(pThis, queue);
- assert(pWrkrInst != NULL);
-
- iMyThrdIndx = pWrkrInst->iThrd;
-
- /* dequeue element (still protected from mutex) */
- iRet = queueDel(pThis, &pUsr);
- queueChkPersist(pThis); // when we support peek(), we must do this down after the del!
- iQueueSize = pThis->iQueueSize; /* cache this for after mutex release */
- pWrkrInst->pUsr = pUsr; /* save it for the cancel cleanup handler */
- qRunsDA = pThis->qRunsDA;
- d_pthread_mutex_unlock(pThis->mut);
- pthread_cond_signal(&pThis->notFull);
- pthread_setcancelstate(iCancelStateSave, NULL);
- /* WE ARE NO LONGER PROTECTED FROM THE MUTEX */
-
- /* do actual processing (the lengthy part, runs in parallel)
- * If we had a problem while dequeing, we do not call the consumer,
- * but we otherwise ignore it. This is in the hopes that it will be
- * self-healing. However, this is really not a good thing.
- * rgerhards, 2008-01-03
- */
- if(iRet != RS_RET_OK)
- FINALIZE;
-
- /* call consumer depending on queue mode (in DA mode, we have just one thread, so it can not change) */
- if(qRunsDA == QRUNS_DA) {
- queueDAConsumer(pThis, pWrkrInst);
- } else {
- /* we are running in normal, non-disk-assisted mode
- * do a quick check if we need to drain the queue. It is OK to use the cached
- * iQueueSize here, because it does not hurt if it is slightly wrong.
- */
- if(pThis->iDiscardMrk > 0 && iQueueSize >= pThis->iDiscardMrk) {
- iRetLocal = objGetSeverity(pUsr, &iSeverity);
- if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
- dbgprintf("Queue 0x%lx/w%d: dequeue/queue nearly full (%d entries), "
- "discarded severity %d message\n",
- queueGetID(pThis), iMyThrdIndx, iQueueSize, iSeverity);
- objDestruct(pUsr);
- }
- } else {
- dbgprintf("Queue 0x%lx/w%d: worker executes consumer...\n",
- queueGetID(pThis), iMyThrdIndx);
- iRetLocal = pThis->pConsumer(pUsr);
- if(iRetLocal != RS_RET_OK) {
- dbgprintf("Queue 0x%lx/w%d: Consumer returned iRet %d\n",
- queueGetID(pThis), iMyThrdIndx, iRetLocal);
+RUNLOG;
+ if(pThis->pWtpDA != NULL) {
+ for(i = 0 ; i <= pThis->iNumWorkerThreads ; ++i) {
+ if(pThis->pWtpDA->pWrkr[i]->tCurrCmd != eWRKTHRD_STOPPED) {
+ wtiJoinThrd(pThis->pWtpDA->pWrkr[i]);
}
}
}
+#endif
-finalize_it:
- if(iRet != RS_RET_OK) {
- dbgprintf("Queue 0x%lx/w%d: error %d dequeueing element - ignoring, but strange things "
- "may happen\n", queueGetID(pThis), iMyThrdIndx, iRet);
- }
-dbgprintf("CallConsumer returns %d\n", iRet);
- return iRet;
-}
-
-
-
-/* cancellation cleanup handler for queueWorker ()
- * Updates admin structure and frees ressources.
- * rgerhards, 2008-01-16
- */
-static void queueWorkerCancelCleanup(void *arg)
-{
- qWrkThrd_t *pWrkrInst = (qWrkThrd_t*) arg;
- queue_t *pThis;
- int iCancelStateSave;
-
- assert(pWrkrInst != NULL);
- ISOBJ_TYPE_assert(pWrkrInst->pQueue, queue);
- pThis = pWrkrInst->pQueue;
-
- dbgprintf("Queue 0x%lx/w%d: cancelation cleanup handler called (NOT FULLY IMPLEMENTED, one msg lost!)\n",
- queueGetID(pThis), pWrkrInst->iThrd);
-
- /* TODO: re-enqueue the data element! */
-
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
- qWrkrSetState(&pThis->pWrkThrds[pWrkrInst->iThrd], eWRKTHRD_TERMINATING);
- pThis->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
-
- dbgprintf("Queue 0x%lx/w%d: thread CANCELED with %d entries left in queue, %d workers running.\n",
- queueGetID(pThis), pWrkrInst->iThrd, pThis->iQueueSize, pThis->iCurNumWrkThrd - 1);
-
- pThis->iCurNumWrkThrd--;
- pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
- d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
-}
-
-
-/* This function is created to keep the code in queueWorker () short. Thus it
- * also does not abide to the usual calling conventions used in rsyslog. It is more
- * like a macro. Its sole purpose is to have a handy shortcut for the queue
- * termination condition. For the same reason, the calling parameters are a bit
- * more verbose than the need to be in theory. The reasoning is the Worker has
- * everything handy and so we do not need to access it from memory (OK, the
- * optimized would probably have created the same code, but why not do it
- * optimal right away...). The function returns 0 if the worker should terminate
- * and something else if it should continue to run.
- * rgerhards, 2008-01-18
- */
-static inline int
-queueWorkerRemainActive(queue_t *pThis, qWrkThrd_t *pWrkrInst)
-{
- register int b; /* this is a boolean! */
-
- /* first check the usual termination condition that applies to all workers */
- b = ( (qWrkrGetState(pWrkrInst) == eWRKTHRD_RUNNING)
- || ((qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN) && (pThis->iQueueSize > 0)));
-dbgprintf("Queue %p/w%d: chk 1 pre empty queue, qsize %d (high wtr %d), cont run: %d, cmd %d, DA qsize %d\n", pThis,
- pWrkrInst->iThrd,
- pThis->iQueueSize, pThis->iHighWtrMrk, b, qWrkrGetState(pWrkrInst),
- (pThis->pqDA == NULL) ? -1 : pThis->pqDA->iQueueSize);
- if(b && pWrkrInst->iThrd == 0 && pThis->qRunsDA == QRUNS_DA) {
- b = pThis->iQueueSize >= pThis->iHighWtrMrk || pThis->pqDA->iQueueSize != 0;
- }
+ dbgprintf("Queue 0x%lx: worker threads terminated, remaining queue size %d.\n",
+ queueGetID(pThis), pThis->iQueueSize);
-dbgprintf("Queue %p/w%d: pre empty queue, qsize %d, cont run: %d\n", pThis, pWrkrInst->iThrd, pThis->iQueueSize, b);
- return b;
+ RETiRet;
}
-/* Each queue has at least one associated worker (consumer) thread. It will pull
- * the message from the queue and pass it to a user-defined function.
- * This function was provided on construction. It MUST be thread-safe.
- * Worker thread 0 is always reserved for disk-assisted mode (if the queue
- * is not DA, this worker will be dormant). All other workers are for
- * regular operations mode. Workers are started and stopped as need arises.
- * rgerhards, 2008-01-15
- */
-static void *
-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;
- qWrkThrd_t *pWrkrInst; /* for cleanup handler */
- int bContinueRun;
-
- ISOBJ_TYPE_assert(pThis, queue);
-
- sigfillset(&sigSet);
- pthread_sigmask(SIG_BLOCK, &sigSet, NULL);
-
- /* do some one-time thread initialization */
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
-
- /* initialize our thread instance descriptor */
- qWrkrInit(&pWrkrInst, pThis);
-
- iMyThrdIndx = pWrkrInst->iThrd;
-
- dbgprintf("Queue 0x%lx/w%d: worker thread startup.\n", queueGetID(pThis), iMyThrdIndx);
-
-dbgprintf("qRunsDA %d, check against %d\n", pThis->qRunsDA, QRUNS_DA);
- if((iMyThrdIndx == 0) && (pThis->qRunsDA != QRUNS_DA)) { /* 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. */
- queueTellActWrkThrd(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
- */
- pThis->iCurNumWrkThrd++;
- if(qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_INIT)
- qWrkrSetState(pWrkrInst, eWRKTHRD_RUNNING); /* we are running now! */
-
- pthread_cleanup_push(queueWorkerCancelCleanup, pWrkrInst);
-
- d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
- /* end one-time stuff */
-
- /* now we have our identity, on to real processing */
- bContinueRun = 1; /* we need this variable, because we need to check the actual termination condition
- * while protected by mutex */
- while(bContinueRun) {
-dbgprintf("Queue 0x%lx/w%d: start worker run, queue cmd currently %d\n",
- queueGetID(pThis), iMyThrdIndx, pThis->pWrkThrds[iMyThrdIndx].tCurrCmd);
- /* process any pending thread requests */
- queueChkWrkThrdChanges(pThis);
-
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
- d_pthread_mutex_lock(pThis->mut);
-dbgprintf("pthis 2: %p\n", pThis);
-
- if((bContinueRun = queueWorkerRemainActive(pThis, pWrkrInst)) == 0) {
-dbgprintf("pthis 2a: %p\n", pThis);
- d_pthread_mutex_unlock(pThis->mut);
- pthread_setcancelstate(iCancelStateSave, NULL);
-dbgprintf("pthis 2b: %p\n", pThis);
- continue; /* and break loop */
- }
-
- /* if we reach this point, we are still protected by the mutex */
-dbgprintf("pthis 3: %p\n", pThis);
-
- if(pThis->iQueueSize == 0) {
- dbgprintf("Queue 0x%lx/w%d: queue EMPTY, waiting for next message.\n",
- queueGetID(pThis), iMyThrdIndx);
- /* check if the parent DA worker is running and, if not, initiate it. Thanks
- * to queueStrtDAWrkr (), we do not actually need to check (that routines does
- * that for us, but we need to aquire the parent queue's mutex to call it.
- */
- if(pThis->pqParent != NULL) {
- dbgprintf("Queue %p: pre start parent %p worker\n", pThis, pThis->pqParent);
- queueStrtDAWrkr(pThis->pqParent);
- }
-
- if(pThis->bSignalOnEmpty > 0) {
- /* we need to signal our parent queue that we are empty */
- dbgprintf("Queue %p/w%d: signal parent we are empty\n", pThis, iMyThrdIndx);
- pthread_cond_signal(pThis->condSignalOnEmpty);
- dbgprintf("Queue %p/w%d: signaling parent empty done\n", pThis, iMyThrdIndx);
- }
- if(pThis->bSignalOnEmpty > 1) {
- /* no mutex associated with this condition, it's just a try (but needed
- * to wakeup a parent worker if e.g. the queue was restarted from disk) */
- pthread_cond_signal(pThis->condSignalOnEmpty2);
- }
- /* 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, worker shutdown %d\n", pThis, iMyThrdIndx, pThis->toWrkShutdown);
- /* DA worker and first worker never have an inactivity timeout */
- if(pWrkrInst->iThrd < 2 || pThis->toWrkShutdown == -1) {
- //xxx 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.
- */
- qWrkrSetState(pWrkrInst, eWRKTHRD_SHUTDOWN);
- }
- }
- dbgprintf("Queue %p/w%d: post condwait ->notEmpty\n", pThis, iMyThrdIndx);
- d_pthread_mutex_unlock(pThis->mut);
- pthread_setcancelstate(iCancelStateSave, NULL);
- pthread_testcancel(); /* see big comment below */
- pthread_yield(); /* see big comment below */
- continue; /* request next iteration */
- }
-
- /* if we reach this point, we have a non-empty queue (and are still protected by mutex) */
- queueWorkerChkAndCallConsumer(pThis, pWrkrInst, 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
- * we may never get cancelled if we do not create a cancellation point ourselfs.
- */
- pthread_testcancel();
- /* We now yield to give the other threads a chance to obtain the mutex. If we do not
- * do that, this thread may very well aquire the mutex again before another thread
- * has even a chance to run. The reason is that mutex operations are free to be
- * implemented in the quickest possible way (and they typically are!). That is, the
- * mutex lock/unlock most probably just does an atomic memory swap and does not necessarily
- * schedule other threads waiting on the same mutex. That can lead to the same thread
- * aquiring the mutex ever and ever again while all others are starving for it. We
- * have exactly seen this behaviour when we deliberately introduced a long-running
- * test action which basically did a sleep. I understand that with real actions the
- * likelihood of this starvation condition is very low - but it could still happen
- * 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();
- if(Debug && (qWrkrGetState(pWrkrInst) == 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);
- }
-
- /* indicate termination */
- pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
-dbgprintf("Queue %p: worker waiting for mutex\n", pThis);
- d_pthread_mutex_lock(&pThis->mutThrdMgmt);
- /* check if we are the DA worker and, if so, switch back to regular mode */
- if(pWrkrInst->iThrd == 0) {
- queueTurnOffDAMode(pThis);
- }
- pthread_cleanup_pop(0); /* remove cleanup handler */
-
- pThis->iCurNumWrkThrd--; /* one less ;) */
- /* if we ever need finalize_it, here would be the place for it! */
- if(qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN ||
- qWrkrGetState(pWrkrInst) == eWRKTHRD_SHUTDOWN_IMMEDIATE ||
- qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_INIT ||
- qWrkrGetState(pWrkrInst) == eWRKTHRD_RUN_CREATED) {
- /* in shutdown case, we need to flag termination. All other commands
- * have a meaning to the thread harvester, so we can not overwrite them
- */
-dbgprintf("Queue 0x%lx/w%d: setting termination state\n", queueGetID(pThis), iMyThrdIndx);
- qWrkrSetState(pWrkrInst, eWRKTHRD_TERMINATING);
- }
- pThis->bThrdStateChanged = 1; /* indicate change, so harverster will be called */
- pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
- d_pthread_mutex_unlock(&pThis->mutThrdMgmt);
- pthread_setcancelstate(iCancelStateSave, NULL);
-
- pthread_exit(0);
-}
-
/* Constructor for the queue object
* This constructs the data structure, but does not yet start the queue. That
@@ -1811,6 +870,8 @@ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads,
DEFiRet;
queue_t *pThis;
+int *pBoom = NULL;
+//*pBoom = 'A';
assert(ppThis != NULL);
assert(pConsumer != NULL);
assert(iWorkerThreads >= 0);
@@ -1866,7 +927,205 @@ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads,
finalize_it:
OBJCONSTRUCT_CHECK_SUCCESS_AND_CLEANUP
- return iRet;
+ RETiRet;
+}
+
+
+/* cancellation cleanup handler for queueWorker ()
+ * Updates admin structure and frees ressources.
+ * rgerhards, 2008-01-16
+ */
+static rsRetVal
+queueConsumerCancelCleanup(void *arg1, void *arg2)
+{
+ queue_t *pThis = (queue_t*) arg1;
+ wti_t *pWti = (wti_t*) arg2;
+
+ ISOBJ_TYPE_assert(pThis, queue);
+ ISOBJ_TYPE_assert(pWti, wti);
+
+ dbgprintf("Queue 0x%lx: cancelation cleanup handler consumer called (NOT FULLY IMPLEMENTED, one msg lost!)\n",
+ queueGetID(pThis));
+
+ /* TODO: re-enqueue the data element! */
+
+ return RS_RET_OK;
+}
+
+
+
+/* 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 and bRunsDA are 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 queueChkDiscardMsg(queue_t *pThis, int iQueueSize, int bRunsDA, void *pUsr)
+{
+ DEFiRet;
+ rsRetVal iRetLocal;
+ int iSeverity;
+
+ ISOBJ_TYPE_assert(pThis, queue);
+ ISOBJ_assert(pUsr);
+
+ if(pThis->iDiscardMrk > 0 && iQueueSize >= pThis->iDiscardMrk && bRunsDA == 0) {
+ iRetLocal = objGetSeverity(pUsr, &iSeverity);
+ if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
+ dbgprintf("Queue 0x%lx: queue nearly full (%d entries), discarded severity %d message\n",
+ queueGetID(pThis), iQueueSize, iSeverity);
+ objDestruct(pUsr);
+ ABORT_FINALIZE(RS_RET_QUEUE_FULL);
+ } else {
+ dbgprintf("Queue 0x%lx: queue nearly full (%d entries), but could not drop msg "
+ "(iRet: %d, severity %d)\n", queueGetID(pThis), iQueueSize,
+ iRetLocal, iSeverity);
+ }
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* dequeue the queued object for the queue consumers.
+ * rgerhards, 2008-10-21
+ */
+static rsRetVal
+queueDequeueConsumable(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
+{
+ DEFiRet;
+ void *pUsr;
+ int iQueueSize;
+ int bRunsDA; /* cache for early mutex release */
+
+ /* dequeue element (still protected from mutex) */
+ iRet = queueDel(pThis, &pUsr);
+ queueChkPersist(pThis); // when we support peek(), we must do this down after the del!
+ iQueueSize = pThis->iQueueSize; /* cache this for after mutex release */
+ bRunsDA = pThis->bRunsDA; /* cache this for after mutex release */
+ pWti->pUsrp = pUsr; /* save it for the cancel cleanup handler */
+ d_pthread_mutex_unlock(pThis->mut);
+ pthread_cond_signal(&pThis->notFull);
+ pthread_setcancelstate(iCancelStateSave, NULL);
+ /* WE ARE NO LONGER PROTECTED BY THE MUTEX */
+
+ /* do actual processing (the lengthy part, runs in parallel)
+ * If we had a problem while dequeing, we do not call the consumer,
+ * but we otherwise ignore it. This is in the hopes that it will be
+ * self-healing. However, this is really not a good thing.
+ * rgerhards, 2008-01-03
+ */
+ if(iRet != RS_RET_OK)
+ FINALIZE;
+
+ /* we are running in normal, non-disk-assisted mode do a quick check if we need to drain the queue.
+ * 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: It is OK to use the cached iQueueSize here, because it does not hurt if it is slightly wrong.
+ */
+ CHKiRet(queueChkDiscardMsg(pThis, iQueueSize, bRunsDA, pUsr));
+
+finalize_it:
+ if(iRet != RS_RET_OK && iRet != RS_RET_DISCARDMSG) {
+ dbgprintf("Queue 0x%lx/w?: error %d dequeueing element - ignoring, but strange things "
+ "may happen\n", queueGetID(pThis), iRet);
+ }
+ 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
+queueConsumerReg(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, queue);
+ ISOBJ_TYPE_assert(pWti, wti);
+
+ CHKiRet(queueDequeueConsumable(pThis, pWti, iCancelStateSave));
+ CHKiRet(pThis->pConsumer(pWti->pUsrp));
+
+finalize_it:
+dbgprintf("Queue %p: regular consumer returns %d\n", pThis, iRet);
+ RETiRet;
+}
+
+
+/* This is a special consumer to feed the disk-queue in disk-assited 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
+queueConsumerDA(queue_t *pThis, wti_t *pWti, int iCancelStateSave)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, queue);
+ ISOBJ_TYPE_assert(pWti, wti);
+
+dbgprintf("Queue %p/w?: queueDAConsumer, queue size %d\n", pThis, pThis->iQueueSize);/* dirty iQueueSize! */
+ CHKiRet(queueDequeueConsumable(pThis, pWti, iCancelStateSave));
+ CHKiRet(queueEnqObj(pThis->pqDA, pWti->pUsrp));
+
+finalize_it:
+dbgprintf("DAConsumer returns with iRet %d\n", iRet);
+ RETiRet;
+}
+
+
+/* must only be called when the queue mutex is locked, else results
+ * are not stable!
+ * Version when running in DA mode.
+ */
+static int
+queueChkStopWrkrDA(queue_t *pThis)
+{
+ return pThis->bEnqOnly || !pThis->bRunsDA;
+}
+
+/* must only be called when the queue mutex is locked, else results
+ * are not stable!
+ * Version when running in non-DA mode.
+ */
+static int
+queueChkStopWrkrReg(queue_t *pThis)
+{
+ return pThis->bEnqOnly || pThis->bRunsDA;
+}
+
+
+/* must only be called when the queue mutex is locked, else results
+ * are not stable! DA version
+ */
+static int
+queueIsIdleDA(queue_t *pThis)
+{
+ return (pThis->iQueueSize == 0 || (pThis->bRunsDA && pThis->iQueueSize <= pThis->iLowWtrMrk));
+}
+/* must only be called when the queue mutex is locked, else results
+ * are not stable! Regular version
+ */
+static int
+queueIsIdleReg(queue_t *pThis)
+{
+ return (pThis->iQueueSize == 0);
}
@@ -1878,7 +1137,8 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */
DEFiRet;
rsRetVal iRetLocal;
int bInitialized = 0; /* is queue already initialized? */
- int i;
+ uchar pszBuf[64];
+ size_t lenBuf;
assert(pThis != NULL);
@@ -1888,7 +1148,7 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */
if(pThis->pqParent == NULL) {
dbgprintf("Queue %p: no parent, alloc mutex\n", pThis);
pThis->mut = (pthread_mutex_t *) malloc (sizeof (pthread_mutex_t));
- pthread_mutex_init (pThis->mut, NULL);
+ pthread_mutex_init(pThis->mut, NULL);
} else {
/* child queue, we need to use parent's mutex */
pThis->mut = pThis->pqParent->mut;
@@ -1909,15 +1169,23 @@ dbgprintf("Queue %p: post mutexes, mut %p\n", pThis, pThis->mut);
if(pThis->qType == QUEUETYPE_DIRECT)
FINALIZE; /* with direct queues, we are already finished... */
- /* initialize worker thread instances
- * TODO: move to separate function
+ /* create worker thread pools for regular operation. The DA pool is created on an as-needed
+ * basis, which potentially means never under most circumstances.
*/
- if((pThis->pWrkThrds = calloc(pThis->iNumWorkerThreads + 1, sizeof(qWrkThrd_t))) == NULL)
- ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
- for(i = 0 ; i < pThis->iNumWorkerThreads + 1 ; ++i) {
- qWrkrConstructFinalize(&pThis->pWrkThrds[i], pThis, i);
- }
-
+ lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "Queue 0x%lx/Reg", (unsigned long) pThis);
+ CHKiRet(wtpConstruct (&pThis->pWtpReg));
+ CHKiRet(wtpSetDbgHdr (pThis->pWtpReg, pszBuf, lenBuf));
+ CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpReg, queueChkStopWrkrReg));
+ CHKiRet(wtpSetpfIsIdle (pThis->pWtpReg, queueIsIdleReg));
+ CHKiRet(wtpSetpfDoWork (pThis->pWtpReg, queueConsumerReg));
+ CHKiRet(wtpSetpfOnWorkerCancel (pThis->pWtpReg, queueConsumerCancelCleanup));
+ CHKiRet(wtpSetpmutUsr (pThis->pWtpReg, pThis->mut));
+ CHKiRet(wtpSetpcondBusy (pThis->pWtpReg, &pThis->notEmpty));
+ CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpReg, pThis->iNumWorkerThreads));
+ CHKiRet(wtpSetpUsr (pThis->pWtpReg, pThis));
+ CHKiRet(wtpConstructFinalize (pThis->pWtpReg));
+
+ /* initialize worker thread instances */
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
@@ -1927,7 +1195,11 @@ dbgprintf("Queue %p: post mutexes, mut %p\n", pThis, pThis->mut);
dbgprintf("Queue 0x%lx: on-disk queue present, needs to be reloaded\n",
queueGetID(pThis));
- queueInitDA(pThis, QUEUE_MODE_ENQDEQ); /* initiate DA mode */
+ queueInitDA(pThis, QUEUE_MODE_ENQDEQ, LOCK_MUTEX); /* initiate DA mode */
+ /* we need to start the DA worker thread so that messages will be processed. So
+ * we advise the worker pool there is at least one needed. The wtp does the rest...
+ */
+ wtpAdviseMaxWorkers(pThis->pWtpDA, 1);
bInitialized = 1; /* we are done */
} else {
// TODO: use logerror? -- rgerhards, 2008-01-16
@@ -1938,16 +1210,14 @@ dbgprintf("Queue %p: post mutexes, mut %p\n", pThis, pThis->mut);
if(!bInitialized) {
dbgprintf("Queue 0x%lx: queue starts up without (loading) any disk state\n", queueGetID(pThis));
- /* fire up the worker threads */
- if(pThis->bEnqOnly == 0)
- queueStrtNewWrkThrd(pThis);
- // TODO: preforked workers! queueStrtAllWrkThrds(pThis);
+ /* we do not fire up any worker threads here, this happens automatically when they are needed */
+ // TODO: preforked workers? queueStrtAllWrkThrds(pThis);
}
pThis->bQueueStarted = 1;
finalize_it:
dbgprintf("queueStart() exit, iret %d\n", iRet);
- return iRet;
+ RETiRet;
}
@@ -2023,7 +1293,7 @@ finalize_it:
if(psQIF != NULL)
strmDestruct(&psQIF);
- return iRet;
+ RETiRet;
}
@@ -2043,15 +1313,16 @@ rsRetVal queueChkPersist(queue_t *pThis)
pThis->iUpdsSincePersist = 0;
}
- return iRet;
+ RETiRet;
}
/* destructor for the queue object */
rsRetVal queueDestruct(queue_t **ppThis)
{
- queue_t *pThis;
DEFiRet;
+ queue_t *pThis;
+ DEFVARS_mutexProtection;
assert(ppThis != NULL);
pThis = *ppThis;
@@ -2064,52 +1335,38 @@ pThis->bSaveOnShutdown = 1; // TODO: Test remove
/* we do not need to take care of any messages left in queue if we are in enqueue only mode */
if(!pThis->bEnqOnly) {
/* in regular mode, need look at termination */
+ BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, LOCK_MUTEX); /* some workers may be running in parallel! */
/* optimize parameters for shutdown of DA-enabled queues */
if(pThis->bIsDA && pThis->iQueueSize > 0) { // TODO: atomic iQueueSize!
dbgprintf("IsDA queue, modifying params for draining\n");
pThis->iHighWtrMrk = 1; /* make sure we drain */
pThis->iLowWtrMrk = 0; /* disable low water mark algo */
- if(pThis->qRunsDA == QRUNS_REGULAR) {
+ if(pThis->bRunsDA == 0) {
if(pThis->iQueueSize > 0) {
- queueInitDA(pThis, QUEUE_MODE_ENQONLY); /* initiate DA mode */
+ queueInitDA(pThis, QUEUE_MODE_ENQONLY, MUTEX_ALREADY_LOCKED); /* initiate DA mode */
}
} else {
queueSetEnqOnly(pThis->pqDA, QUEUE_MODE_ENQONLY); /* turn on enqueue-only mode */
+ /* worker may have been waited on low water mark, reactivate */
+ wtpAdviseMaxWorkers(pThis->pWtpDA, 1);
}
if(pThis->bSaveOnShutdown) {
dbgprintf("bSaveOnShutdown set, eternal timeout set\n");
pThis->toQShutdown = QUEUE_TIMEOUT_ETERNAL;
}
- /* now we need to activate workers (read doc/dev_queue.html) */
+ END_MTX_PROTECTED_OPERATIONS(pThis->mut);
}
-
- /* wait until all pending workers are started up */
- qWrkrWaitAllWrkrStartup(pThis);
-
- // We need to startup a worker if we are in non-DA mode and the queue is not empty and not in enque-only mode */
- dbgprintf("Queue %p: queueDestruct probing if any regular workers need to be started, CurWrkr %d, qsize %d, qRunsDA %d\n",
- pThis, pThis->iCurNumWrkThrd, pThis->iQueueSize, pThis->qRunsDA);
- d_pthread_mutex_lock(pThis->mut);
- dbgprintf("queueDestruct mutex locked\n");
- if(pThis->iCurNumWrkThrd == 0 && pThis->iQueueSize > 0 && !pThis->bEnqOnly) {
- dbgprintf("Queue %p: queueDestruct must start regular workers!\n", pThis);
- // TODO check mutex call order - doies function aquire mutex?
- queueStrtNewWrkThrd(pThis);
- }
- d_pthread_mutex_unlock(pThis->mut);
- dbgprintf("queueDestruct mutex unlocked\n");
-
- /* wait again in case a new worker was started */
- qWrkrWaitAllWrkrStartup(pThis);
}
- /* terminate our own worker threads */
- if(pThis->pWrkThrds != NULL) {
+ /* at this point, the queue is either empty with all workers being idle (or deact) or the queue
+ * is full and all workers are running. We now need to wait for everyone to become idle.
+ */
+ if(pThis->qType != QUEUETYPE_DIRECT) {
queueShutdownWorkers(pThis);
}
- /* if still running DA, terminate disk queue */
- if(pThis->qRunsDA != QRUNS_REGULAR)
+ /* if still running DA, terminate disk queue (note that the DA queue is NULL if it was never used) */
+ if(pThis->bRunsDA && pThis->pqDA != NULL)
queueDestruct(&pThis->pqDA);
/* persist the queue (we always do that - queuePersits() does cleanup if the queue is empty) */
@@ -2118,9 +1375,10 @@ pThis->bSaveOnShutdown = 1; // TODO: Test remove
}
/* ... then free resources */
- if(pThis->pWrkThrds != NULL) {
- free(pThis->pWrkThrds);
- pThis->pWrkThrds = NULL;
+ if(pThis->qType != QUEUETYPE_DIRECT) {
+ wtpDestruct(&pThis->pWtpReg);
+ if(pThis->pWtpDA != NULL)
+ wtpDestruct(&pThis->pWtpDA);
}
if(pThis->pqParent == NULL) {
@@ -2141,7 +1399,7 @@ pThis->bSaveOnShutdown = 1; // TODO: Test remove
free(pThis);
*ppThis = NULL;
- return iRet;
+ RETiRet;
}
@@ -2167,7 +1425,7 @@ queueSetFilePrefix(queue_t *pThis, uchar *pszPrefix, size_t iLenPrefix)
pThis->lenFilePrefix = iLenPrefix;
finalize_it:
- return iRet;
+ RETiRet;
}
/* set the queue's maximum file size
@@ -2187,7 +1445,7 @@ queueSetMaxFileSize(queue_t *pThis, size_t iMaxFileSize)
pThis->iMaxFileSize = iMaxFileSize;
finalize_it:
- return iRet;
+ RETiRet;
}
@@ -2203,65 +1461,64 @@ queueEnqObj(queue_t *pThis, void *pUsr)
{
DEFiRet;
int iCancelStateSave;
+ int iMaxWorkers;
int i;
struct timespec t;
- int iSeverity = 8;
- rsRetVal iRetLocal;
ISOBJ_TYPE_assert(pThis, queue);
- /* process any pending thread requests */
- queueChkWrkThrdChanges(pThis);
-
+dbgprintf("Queue %p: EnqObj() 1\n", pThis);
/* 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, 2008-01-08
*/
- if(pThis->pWrkThrds != NULL) {
+ if(pThis->qType != QUEUETYPE_DIRECT) {
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
d_pthread_mutex_lock(pThis->mut);
}
- /* first check if we can discard anything */
- if(pThis->iDiscardMrk > 0 && pThis->iQueueSize >= pThis->iDiscardMrk) {
- iRetLocal = objGetSeverity(pUsr, &iSeverity);
- if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
- dbgprintf("Queue 0x%lx: queue nearly full (%d entries), discarded severity %d message\n",
- queueGetID(pThis), pThis->iQueueSize, iSeverity);
- objDestruct(pUsr);
- ABORT_FINALIZE(RS_RET_QUEUE_FULL);
- } else {
- dbgprintf("Queue 0x%lx: queue nearly full (%d entries), but could not drop msg "
- "(iRet: %d, severity %d)\n", queueGetID(pThis), pThis->iQueueSize,
- iRetLocal, iSeverity);
- }
- }
+ /* first check if we need to discard this message (which will cause CHKiRet() to exit) */
+ CHKiRet(queueChkDiscardMsg(pThis, pThis->iQueueSize, pThis->bRunsDA, pUsr));
+dbgprintf("Queue %p: EnqObj() 10\n", pThis);
/* then check if we need to add an assistance disk queue */
if(pThis->bIsDA)
CHKiRet(queueChkStrtDA(pThis));
- /* re-process any new pending thread requests and see if we need to start workers */
- queueChkAndStrtWrk(pThis);
+RUNLOG_VAR("%d", pThis->bIsDA);
+ /* make sure at least one worker is running. */
+ if(pThis->bRunsDA) {
+RUNLOG;
+ wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* disk queues have always one worker */
+ } else {
+ if(pThis->qType == QUEUETYPE_DISK || pThis->iMinMsgsPerWrkr == 0) {
+ iMaxWorkers = 1;
+ } else {
+ iMaxWorkers = pThis->iQueueSize / pThis->iMinMsgsPerWrkr + 1;
+ }
+RUNLOG;
+ wtpAdviseMaxWorkers(pThis->pWtpReg, iMaxWorkers);
+ }
- /* and finally (try to) enqueue what is left over */
+ /* wait for the queue to be ready... */
while(pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize) {
dbgprintf("Queue 0x%lx: enqueueMsg: queue FULL - waiting to drain.\n", queueGetID(pThis));
- queueTimeoutComp(&t, pThis->toEnq);
+ timeoutComp(&t, pThis->toEnq);
if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
dbgprintf("Queue 0x%lx: enqueueMsg: cond timeout, dropping message!\n", queueGetID(pThis));
objDestruct(pUsr);
ABORT_FINALIZE(RS_RET_QUEUE_FULL);
}
}
+
+ /* and finally enqueue the message */
CHKiRet(queueAdd(pThis, pUsr));
queueChkPersist(pThis);
finalize_it:
- /* now awake sleeping worker threads */
- if(pThis->pWrkThrds != NULL) {
+ if(pThis->qType != QUEUETYPE_DIRECT) {
d_pthread_mutex_unlock(pThis->mut);
i = pthread_cond_signal(&pThis->notEmpty);
dbgprintf("Queue 0x%lx: EnqueueMsg signaled condition (%d)\n", (unsigned long) pThis, i);
@@ -2269,7 +1526,7 @@ finalize_it:
}
- return iRet;
+ RETiRet;
}
@@ -2307,7 +1564,8 @@ queueSetEnqOnly(queue_t *pThis, int bEnqOnly)
/* this means we need to terminate all workers - that's it... */
dbgprintf("Queue 0x%lx: switching to enqueue-only mode, terminating all worker threads\n",
queueGetID(pThis));
- queueWrkThrdReqTrm(pThis, eWRKTHRD_SHUTDOWN_IMMEDIATE, 0);
+ wtpWakeupAllWrkr(pThis->pWtpDA);
+ wtpWakeupAllWrkr(pThis->pWtpReg);
} else {
/* switch back to regular mode */
ABORT_FINALIZE(RS_RET_NOT_IMPLEMENTED); /* we don't need this so far... */
@@ -2321,7 +1579,7 @@ finalize_it:
d_pthread_mutex_unlock(pThis->mut);
pthread_setcancelstate(iCancelStateSave, NULL);
}
- return iRet;
+ RETiRet;
}
@@ -2360,19 +1618,18 @@ static rsRetVal queueSetProperty(queue_t *pThis, property_t *pProp)
}
finalize_it:
- return iRet;
+ RETiRet;
}
#undef isProp
-
-
/* 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(queue, 1)
- //OBJSetMethodHandler(objMethod_SERIALIZE, strmSerialize);
OBJSetMethodHandler(objMethod_SETPROPERTY, queueSetProperty);
//OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strmConstructFinalize);
+//fprintf(stdout, "queueChkStopWrkrReg: %p\n", queueChkStopWrkrReg);
+//fprintf(stdout, "queueChkStopWrkrDA: %p\n", queueChkStopWrkrDA);
ENDObjClassInit(queue)
/*