diff options
Diffstat (limited to 'runtime/wti.c')
| -rw-r--r-- | runtime/wti.c | 81 |
1 files changed, 40 insertions, 41 deletions
diff --git a/runtime/wti.c b/runtime/wti.c index 343a7227..abdf4add 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -39,15 +39,22 @@ #include <pthread.h> #include <errno.h> +#ifdef OS_SOLARIS +# include <sched.h> +#endif + #include "rsyslog.h" #include "stringbuf.h" #include "srUtils.h" #include "wtp.h" #include "wti.h" #include "obj.h" +#include "glbl.h" +#include "atomic.h" /* static data */ DEFobjStaticHelpers +DEFobjCurrIf(glbl) /* forward-definitions */ @@ -99,6 +106,7 @@ rsRetVal wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex) { DEFiRet; + qWrkCmd_t tCurrCmd; DEFVARS_mutexProtection; ISOBJ_TYPE_assert(pThis, wti); @@ -106,13 +114,14 @@ wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex) BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); + tCurrCmd = pThis->tCurrCmd; /* all worker states must be followed sequentially, only termination can be set in any state */ - if( (bActiveOnly && (pThis->tCurrCmd < eWRKTHRD_RUN_CREATED)) - || (pThis->tCurrCmd > tCmd && !(tCmd == eWRKTHRD_TERMINATING || tCmd == eWRKTHRD_STOPPED))) { - dbgprintf("%s: command %d can not be accepted in current %d processing state - ignored\n", - wtiGetDbgHdr(pThis), tCmd, pThis->tCurrCmd); + if( (bActiveOnly && (tCurrCmd < eWRKTHRD_RUN_CREATED)) + || (tCurrCmd > tCmd && !(tCmd == eWRKTHRD_TERMINATING || tCmd == eWRKTHRD_STOPPED))) { + DBGPRINTF("%s: command %d can not be accepted in current %d processing state - ignored\n", + wtiGetDbgHdr(pThis), tCmd, tCurrCmd); } else { - dbgprintf("%s: receiving command %d\n", wtiGetDbgHdr(pThis), tCmd); + DBGPRINTF("%s: receiving command %d\n", wtiGetDbgHdr(pThis), tCmd); /* we could replace this with a simple if, but we leave the switch in in case we need * to add something at a later stage. -- rgerhards, 2008-09-30 */ @@ -136,7 +145,12 @@ wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex) /* DO NOTHING */ break; } - pThis->tCurrCmd = tCmd; /* apply the new state */ + /* apply the new state */ + unsigned val = ATOMIC_CAS_VAL(pThis->tCurrCmd, tCurrCmd, tCmd); + if(val != tCurrCmd) { + DBGPRINTF("wtiSetState PROBLEM, tCurrCmd %d overwritten with %d, wanted to set %d\n", tCurrCmd, val, tCmd); + } + } END_MTX_PROTECTED_OPERATIONS(&pThis->mut); @@ -144,7 +158,7 @@ wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex) } -/* Cancel the thread. If the thread is already cancelled or termination, +/* Cancel the thread. If the thread is already cancelled or terminated, * we do not again cancel it. But it is save and legal to call wtiCancelThrd() in * such situations. * rgerhards, 2008-02-26 @@ -158,11 +172,13 @@ wtiCancelThrd(wti_t *pThis) d_pthread_mutex_lock(&pThis->mut); + wtiProcessThrdChanges(pThis, MUTEX_ALREADY_LOCKED); /* process state change, so that we have current state vars */ + if(pThis->tCurrCmd >= eWRKTHRD_TERMINATING) { - dbgoprint((obj_t*) pThis, "canceling worker thread\n"); + dbgoprint((obj_t*) pThis, "canceling worker thread, curr stat %d\n", pThis->tCurrCmd); pthread_cancel(pThis->thrdID); wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED); - pThis->pWtp->bThrdStateChanged = 1; /* indicate change, so harverster will be called */ + ATOMIC_STORE_1_TO_INT(pThis->pWtp->bThrdStateChanged); /* indicate change, so harverster will be called */ } d_pthread_mutex_unlock(&pThis->mut); @@ -194,8 +210,7 @@ CODESTARTobjDestruct(wti) pthread_cond_destroy(&pThis->condExitDone); pthread_mutex_destroy(&pThis->mut); - if(pThis->pszDbgHdr != NULL) - free(pThis->pszDbgHdr); + free(pThis->pszDbgHdr); ENDobjDestruct(wti) @@ -303,7 +318,7 @@ wtiWorkerCancelCleanup(void *arg) pWtp = pThis->pWtp; ISOBJ_TYPE_assert(pWtp, wtp); - dbgprintf("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis)); /* call user supplied handler (that one e.g. requeues the element) */ pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr, pThis->pUsrp); @@ -312,7 +327,7 @@ wtiWorkerCancelCleanup(void *arg) d_pthread_mutex_lock(&pWtp->mut); wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED); /* TODO: sync access? I currently think it is NOT needed -- rgerhards, 2008-01-28 */ - pWtp->bThrdStateChanged = 1; /* indicate change, so harverster will be called */ + ATOMIC_STORE_1_TO_INT(pWtp->bThrdStateChanged); /* indicate change, so harverster will be called */ d_pthread_mutex_unlock(&pWtp->mut); pthread_setcancelstate(iCancelStateSave, NULL); @@ -321,27 +336,6 @@ wtiWorkerCancelCleanup(void *arg) /* generic worker thread framework - * - * Some special comments below, so that they do not clutter the main function code: - * - * On the use of pthread_testcancel(): - * 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. - * - * On the use of pthread_yield(): - * We 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 */ #pragma GCC diagnostic ignored "-Wempty-body" rsRetVal @@ -369,10 +363,6 @@ wtiWorker(wti_t *pThis) while(1) { /* loop will be broken below - need to do mutex locks */ /* process any pending thread requests */ wtpProcessThrdChanges(pWtp); - pthread_testcancel(); /* see big comment in function header */ -# if !defined(__hpux) /* pthread_yield is missing there! */ - pthread_yield(); /* see big comment in function header */ -# endif /* if we have a rate-limiter set for this worker pool, let's call it. Please * keep in mind that the rate-limiter may hold us for an extended period @@ -395,7 +385,7 @@ wtiWorker(wti_t *pThis) /* if we reach this point, we are still protected by the mutex */ if(pWtp->pfIsIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED)) { - dbgprintf("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis)); pWtp->pfOnIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED); if(pWtp->toWrkShutdown == -1) { @@ -404,7 +394,7 @@ wtiWorker(wti_t *pThis) } else { timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */ if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) { - dbgprintf("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); bInactivityTOOccured = 1; /* indicate we had a timeout */ } } @@ -424,7 +414,7 @@ wtiWorker(wti_t *pThis) pWtp->pfOnWorkerShutdown(pWtp->pUsr); wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED); - pWtp->bThrdStateChanged = 1; /* indicate change, so harverster will be called */ + ATOMIC_STORE_1_TO_INT(pWtp->bThrdStateChanged); /* indicate change, so harverster will be called */ d_pthread_mutex_unlock(&pThis->mut); pthread_setcancelstate(iCancelStateSave, NULL); @@ -470,6 +460,14 @@ finalize_it: /* dummy */ rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } +/* exit our class + */ +BEGINObjClassExit(wti, OBJ_IS_CORE_MODULE) /* CHANGE class also in END MACRO! */ +CODESTARTObjClassExit(nsdsel_gtls) + /* release objects we no longer need */ + objRelease(glbl, CORE_COMPONENT); +ENDObjClassExit(wti) + /* Initialize the wti class. Must be called as the very first method * before anything else is called inside this class. @@ -477,6 +475,7 @@ rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } */ BEGINObjClassInit(wti, 1, OBJ_IS_CORE_MODULE) /* one is the object version (most important for persisting) */ /* request objects we use */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); ENDObjClassInit(wti) /* |