rsyslog/rsyslog
1
0
Fork 0
mirror of https://github.com/rsyslog/rsyslog.git synced 2026-06-19 17:32:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

removed mutex locks

Browse Source 
... by utilizing that we need to modify a state variable only in
a sequential way during shutdown.
This commit is contained in:
Rainer Gerhards 2009-07-17 19:59:22 +02:00
parent 4c9eded44d
commit ac186f1c3d
4 changed files with 36 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
runtime/atomic.h
View File

@ -51,7 +51,6 @@
# define ATOMIC_STORE_1_TO_32BIT(data) __sync_lock_test_and_set(&(data), 1) # define ATOMIC_STORE_1_TO_32BIT(data) __sync_lock_test_and_set(&(data), 1)
# define ATOMIC_STORE_0_TO_INT(data) __sync_fetch_and_and(&(data), 0) # define ATOMIC_STORE_0_TO_INT(data) __sync_fetch_and_and(&(data), 0)
# define ATOMIC_STORE_1_TO_INT(data) __sync_fetch_and_or(&(data), 1) # define ATOMIC_STORE_1_TO_INT(data) __sync_fetch_and_or(&(data), 1)
# define ATOMIC_STORE_INT_TO_INT(data, val) __sync_fetch_and_or(&(data), (val))
# define ATOMIC_CAS(data, oldVal, newVal) __sync_bool_compare_and_swap(&(data), (oldVal), (newVal)); # define ATOMIC_CAS(data, oldVal, newVal) __sync_bool_compare_and_swap(&(data), (oldVal), (newVal));
# define ATOMIC_CAS_VAL(data, oldVal, newVal) __sync_val_compare_and_swap(&(data), (oldVal), (newVal)); # define ATOMIC_CAS_VAL(data, oldVal, newVal) __sync_val_compare_and_swap(&(data), (oldVal), (newVal));
#else #else

2
runtime/wti.c
View File

@ -250,7 +250,7 @@ wtiWorker(wti_t *pThis)
d_pthread_mutex_lock(pWtp->pmutUsr); d_pthread_mutex_lock(pWtp->pmutUsr);
/* first check if we are in shutdown process (but evaluate a bit later) */ /* first check if we are in shutdown process (but evaluate a bit later) */
terminateRet = wtpChkStopWrkr(pWtp, LOCK_MUTEX, MUTEX_ALREADY_LOCKED); terminateRet = wtpChkStopWrkr(pWtp, MUTEX_ALREADY_LOCKED);
if(terminateRet == RS_RET_TERMINATE_NOW) { if(terminateRet == RS_RET_TERMINATE_NOW) {
/* we now need to free the old batch */ /* we now need to free the old batch */
localRet = pWtp->pfObjProcessed(pWtp->pUsr, pThis); localRet = pWtp->pfObjProcessed(pWtp->pUsr, pThis);

63
runtime/wtp.c
View File

@ -87,7 +87,7 @@ static rsRetVal NotImplementedDummy() { return RS_RET_NOT_IMPLEMENTED; }
/* Standard-Constructor for the wtp object /* Standard-Constructor for the wtp object
*/ */
BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! */ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! */
pthread_mutex_init(&pThis->mut, NULL); pthread_mutex_init(&pThis->mutWtp, NULL);
pthread_cond_init(&pThis->condThrdTrm, NULL); pthread_cond_init(&pThis->condThrdTrm, NULL);
/* set all function pointers to "not implemented" dummy so that we can safely call them */ /* set all function pointers to "not implemented" dummy so that we can safely call them */
pThis->pfChkStopWrkr = NotImplementedDummy; pThis->pfChkStopWrkr = NotImplementedDummy;
@ -151,7 +151,7 @@ CODESTARTobjDestruct(wtp)
/* actual destruction */ /* actual destruction */
pthread_cond_destroy(&pThis->condThrdTrm); pthread_cond_destroy(&pThis->condThrdTrm);
pthread_mutex_destroy(&pThis->mut); pthread_mutex_destroy(&pThis->mutWtp);
free(pThis->pszDbgHdr); free(pThis->pszDbgHdr);
ENDobjDestruct(wtp) ENDobjDestruct(wtp)
@ -187,19 +187,20 @@ wtpWakeupAllWrkr(wtp_t *pThis)
} }
/* Sent a specific state for the worker thread pool. /* Sent a specific state for the worker thread pool. -- rgerhards, 2008-01-21
* rgerhards, 2008-01-21 * We do not need to do atomic instructions as set operations are only
* called when terminating the pool, and then in strict sequence. So we
* can never overwrite each other. On the other hand, it also doesn't
* matter if the read operation obtains an older value, as we then simply
* do one more iteration, what is perfectly legal (during shutdown
* they are awoken in any case). -- rgerhards, 2009-07-20
*/ */
rsRetVal rsRetVal
wtpSetState(wtp_t *pThis, wtpState_t iNewState) wtpSetState(wtp_t *pThis, wtpState_t iNewState)
{ {
DEFiRet;
ISOBJ_TYPE_assert(pThis, wtp); ISOBJ_TYPE_assert(pThis, wtp);
pThis->wtpState = iNewState; pThis->wtpState = iNewState;
/* TODO: must wakeup workers? seen to be not needed -- rgerhards, 2008-01-28 */ return RS_RET_OK;
RETiRet;
} }
@ -209,17 +210,20 @@ wtpSetState(wtp_t *pThis, wtpState_t iNewState)
* rgerhards, 2008-01-21 * rgerhards, 2008-01-21
*/ */
rsRetVal rsRetVal
wtpChkStopWrkr(wtp_t *pThis, int bLockMutex, int bLockUsrMutex) wtpChkStopWrkr(wtp_t *pThis, int bLockUsrMutex)
{ {
DEFiRet; DEFiRet;
DEFVARS_mutexProtection; wtpState_t wtpState;
ISOBJ_TYPE_assert(pThis, wtp); ISOBJ_TYPE_assert(pThis, wtp);
/* we need a consistent value, but it doesn't really matter if it is changed
* right after the fetch - then we simply do one more iteration in the worker
*/
wtpState = ATOMIC_FETCH_32BIT(pThis->wtpState);
BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); if(wtpState == wtpState_SHUTDOWN_IMMEDIATE) {
if(pThis->wtpState == wtpState_SHUTDOWN_IMMEDIATE) {
ABORT_FINALIZE(RS_RET_TERMINATE_NOW); ABORT_FINALIZE(RS_RET_TERMINATE_NOW);
} else if(pThis->wtpState == wtpState_SHUTDOWN) { } else if(wtpState == wtpState_SHUTDOWN) {
ABORT_FINALIZE(RS_RET_TERMINATE_WHEN_IDLE); ABORT_FINALIZE(RS_RET_TERMINATE_WHEN_IDLE);
} }
@ -229,7 +233,6 @@ wtpChkStopWrkr(wtp_t *pThis, int bLockMutex, int bLockUsrMutex)
} }
finalize_it: finalize_it:
END_MTX_PROTECTED_OPERATIONS(&pThis->mut);
RETiRet; RETiRet;
} }
@ -251,14 +254,14 @@ wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, struct timespec *ptTimeout
wtpWakeupAllWrkr(pThis); wtpWakeupAllWrkr(pThis);
/* wait for worker thread termination */ /* wait for worker thread termination */
d_pthread_mutex_lock(&pThis->mut); d_pthread_mutex_lock(&pThis->mutWtp);
pthread_cleanup_push(mutexCancelCleanup, &pThis->mut); pthread_cleanup_push(mutexCancelCleanup, &pThis->mutWtp);
bTimedOut = 0; bTimedOut = 0;
while(pThis->iCurNumWrkThrd > 0 && !bTimedOut) { while(pThis->iCurNumWrkThrd > 0 && !bTimedOut) {
dbgprintf("%s: waiting %ldms on worker thread termination, %d still running\n", dbgprintf("%s: waiting %ldms on worker thread termination, %d still running\n",
wtpGetDbgHdr(pThis), timeoutVal(ptTimeout), pThis->iCurNumWrkThrd); wtpGetDbgHdr(pThis), timeoutVal(ptTimeout), pThis->iCurNumWrkThrd);
if(d_pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mut, ptTimeout) != 0) { if(d_pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mutWtp, ptTimeout) != 0) {
dbgprintf("%s: timeout waiting on worker thread termination\n", wtpGetDbgHdr(pThis)); dbgprintf("%s: timeout waiting on worker thread termination\n", wtpGetDbgHdr(pThis));
bTimedOut = 1; /* we exit the loop on timeout */ bTimedOut = 1; /* we exit the loop on timeout */
} }
@ -284,12 +287,12 @@ wtpCancelAll(wtp_t *pThis)
ISOBJ_TYPE_assert(pThis, wtp); ISOBJ_TYPE_assert(pThis, wtp);
d_pthread_mutex_lock(&pThis->mut); d_pthread_mutex_lock(&pThis->mutWtp);
/* go through all workers and cancel those that are active */ /* go through all workers and cancel those that are active */
for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) { for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
wtiCancelThrd(pThis->pWrkr[i]); wtiCancelThrd(pThis->pWrkr[i]);
} }
d_pthread_mutex_unlock(&pThis->mut); d_pthread_mutex_unlock(&pThis->mutWtp);
RETiRet; RETiRet;
} }
@ -305,9 +308,9 @@ wtpSetInactivityGuard(wtp_t *pThis, int bNewState, int bLockMutex)
DEFiRet; DEFiRet;
DEFVARS_mutexProtection; DEFVARS_mutexProtection;
BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp, bLockMutex);
pThis->bInactivityGuard = bNewState; pThis->bInactivityGuard = bNewState;
END_MTX_PROTECTED_OPERATIONS(&pThis->mut); END_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp);
RETiRet; RETiRet;
} }
@ -330,11 +333,11 @@ wtpWrkrExecCancelCleanup(void *arg)
// TODO: the mutex_lock is dangerous, if we are cancelled within some function // TODO: the mutex_lock is dangerous, if we are cancelled within some function
// that already has the mutex locked... // that already has the mutex locked...
d_pthread_mutex_lock(&pThis->mut); d_pthread_mutex_lock(&pThis->mutWtp);
pThis->iCurNumWrkThrd--; pThis->iCurNumWrkThrd--;
wtiSetState(pWti, WRKTHRD_STOPPED); wtiSetState(pWti, WRKTHRD_STOPPED);
pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */ pthread_cond_signal(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
d_pthread_mutex_unlock(&pThis->mut); d_pthread_mutex_unlock(&pThis->mutWtp);
DBGPRINTF("%s: Worker thread %lx, terminated, num workers now %d\n", DBGPRINTF("%s: Worker thread %lx, terminated, num workers now %d\n",
wtpGetDbgHdr(pThis), (unsigned long) pWti, pThis->iCurNumWrkThrd); wtpGetDbgHdr(pThis), (unsigned long) pWti, pThis->iCurNumWrkThrd);
@ -396,7 +399,7 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex)
ISOBJ_TYPE_assert(pThis, wtp); ISOBJ_TYPE_assert(pThis, wtp);
BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp, bLockMutex);
pThis->iCurNumWrkThrd++; pThis->iCurNumWrkThrd++;
@ -420,7 +423,7 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex)
wtpGetDbgHdr(pThis), iState, pThis->iCurNumWrkThrd); wtpGetDbgHdr(pThis), iState, pThis->iCurNumWrkThrd);
finalize_it: finalize_it:
END_MTX_PROTECTED_OPERATIONS(&pThis->mut); END_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp);
RETiRet; RETiRet;
} }
@ -445,7 +448,7 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
if(nMaxWrkr == 0) if(nMaxWrkr == 0)
FINALIZE; FINALIZE;
d_pthread_mutex_lock(&pThis->mut); d_pthread_mutex_lock(&pThis->mutWtp);
if(nMaxWrkr > pThis->iNumWorkerThreads) /* limit to configured maximum */ if(nMaxWrkr > pThis->iNumWorkerThreads) /* limit to configured maximum */
nMaxWrkr = pThis->iNumWorkerThreads; nMaxWrkr = pThis->iNumWorkerThreads;
@ -466,7 +469,7 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
finalize_it: finalize_it:
d_pthread_mutex_unlock(&pThis->mut); d_pthread_mutex_unlock(&pThis->mutWtp);
RETiRet; RETiRet;
} }
@ -504,9 +507,9 @@ wtpGetCurNumWrkr(wtp_t *pThis, int bLockMutex)
BEGINfunc BEGINfunc
ISOBJ_TYPE_assert(pThis, wtp); ISOBJ_TYPE_assert(pThis, wtp);
BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp, bLockMutex);
iNumWrkr = pThis->iCurNumWrkThrd; iNumWrkr = pThis->iCurNumWrkThrd;
END_MTX_PROTECTED_OPERATIONS(&pThis->mut); END_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp);
ENDfunc ENDfunc
return iNumWrkr; return iNumWrkr;

4
runtime/wtp.h
View File

@ -51,7 +51,7 @@ struct wtp_s {
bool bInactivityGuard;/* prevents inactivity due to race condition */ bool bInactivityGuard;/* prevents inactivity due to race condition */
rsRetVal (*pConsumer)(void *); /* user-supplied consumer function for dewtpd messages */ rsRetVal (*pConsumer)(void *); /* user-supplied consumer function for dewtpd messages */
/* synchronization variables */ /* synchronization variables */
pthread_mutex_t mut; /* mutex for the wtp's thread management */ pthread_mutex_t mutWtp; /* mutex for the wtp's thread management */
pthread_cond_t condThrdTrm;/* signalled when threads terminate */ pthread_cond_t condThrdTrm;/* signalled when threads terminate */
/* end sync variables */ /* end sync variables */
/* user objects */ /* user objects */
@ -82,7 +82,7 @@ rsRetVal wtpDestruct(wtp_t **ppThis);
rsRetVal wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr); rsRetVal wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr);
rsRetVal wtpProcessThrdChanges(wtp_t *pThis); rsRetVal wtpProcessThrdChanges(wtp_t *pThis);
rsRetVal wtpSetInactivityGuard(wtp_t *pThis, int bNewState, int bLockMutex); rsRetVal wtpSetInactivityGuard(wtp_t *pThis, int bNewState, int bLockMutex);
rsRetVal wtpChkStopWrkr(wtp_t *pThis, int bLockMutex, int bLockUsrMutex); rsRetVal wtpChkStopWrkr(wtp_t *pThis, int bLockUsrMutex);
rsRetVal wtpSetState(wtp_t *pThis, wtpState_t iNewState); rsRetVal wtpSetState(wtp_t *pThis, wtpState_t iNewState);
rsRetVal wtpWakeupWrkr(wtp_t *pThis); rsRetVal wtpWakeupWrkr(wtp_t *pThis);
rsRetVal wtpWakeupAllWrkr(wtp_t *pThis); rsRetVal wtpWakeupAllWrkr(wtp_t *pThis);