summaryrefslogtreecommitdiffstats
path: root/runtime/wtp.c
diff options
context:
space:
mode:
Diffstat (limited to 'runtime/wtp.c')
-rw-r--r--runtime/wtp.c51
1 files changed, 38 insertions, 13 deletions
diff --git a/runtime/wtp.c b/runtime/wtp.c
index 060e6627..ece80911 100644
--- a/runtime/wtp.c
+++ b/runtime/wtp.c
@@ -96,6 +96,8 @@ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro!
pThis->pfGetDeqBatchSize = NotImplementedDummy;
pThis->pfDoWork = NotImplementedDummy;
pThis->pfObjProcessed = NotImplementedDummy;
+ INIT_ATOMIC_HELPER_MUT(pThis->mutCurNumWrkThrd);
+ INIT_ATOMIC_HELPER_MUT(pThis->mutWtpState);
ENDobjConstruct(wtp)
@@ -149,13 +151,15 @@ CODESTARTobjDestruct(wtp)
pthread_cond_destroy(&pThis->condThrdTrm);
pthread_mutex_destroy(&pThis->mutWtp);
pthread_attr_destroy(&pThis->attrThrd);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutCurNumWrkThrd);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutWtpState);
free(pThis->pszDbgHdr);
ENDobjDestruct(wtp)
/* Sent a specific state for the worker thread pool. -- rgerhards, 2008-01-21
- * We do not need to do atomic instructions as set operations are only
+ * 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
@@ -166,7 +170,7 @@ rsRetVal
wtpSetState(wtp_t *pThis, wtpState_t iNewState)
{
ISOBJ_TYPE_assert(pThis, wtp);
- pThis->wtpState = iNewState;
+ pThis->wtpState = iNewState; // TODO: do we need a mutex here? 2010-04-26
return RS_RET_OK;
}
@@ -186,7 +190,7 @@ wtpChkStopWrkr(wtp_t *pThis, int bLockUsrMutex)
/* 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);
+ wtpState = (wtpState_t) ATOMIC_FETCH_32BIT((int*)&pThis->wtpState, &pThis->mutWtpState);
if(wtpState == wtpState_SHUTDOWN_IMMEDIATE) {
ABORT_FINALIZE(RS_RET_TERMINATE_NOW);
@@ -216,6 +220,7 @@ wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, struct timespec *ptTimeout
{
DEFiRet;
int bTimedOut;
+ int i;
ISOBJ_TYPE_assert(pThis, wtp);
@@ -223,6 +228,10 @@ wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, struct timespec *ptTimeout
d_pthread_mutex_lock(pThis->pmutUsr);
wtpSetState(pThis, tShutdownCmd);
pthread_cond_broadcast(pThis->pcondBusy); /* wake up all workers */
+ /* awake workers in retry loop */
+ for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
+ wtiWakeupThrd(pThis->pWrkr[i]);
+ }
d_pthread_mutex_unlock(pThis->pmutUsr);
/* wait for worker thread termination */
@@ -231,12 +240,19 @@ wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, struct timespec *ptTimeout
bTimedOut = 0;
while(pThis->iCurNumWrkThrd > 0 && !bTimedOut) {
DBGPRINTF("%s: waiting %ldms on worker thread termination, %d still running\n",
- wtpGetDbgHdr(pThis), timeoutVal(ptTimeout), ATOMIC_FETCH_32BIT(pThis->iCurNumWrkThrd));
+ wtpGetDbgHdr(pThis), timeoutVal(ptTimeout),
+ ATOMIC_FETCH_32BIT(&pThis->iCurNumWrkThrd, &pThis->mutCurNumWrkThrd));
if(d_pthread_cond_timedwait(&pThis->condThrdTrm, &pThis->mutWtp, ptTimeout) != 0) {
DBGPRINTF("%s: timeout waiting on worker thread termination\n", wtpGetDbgHdr(pThis));
bTimedOut = 1; /* we exit the loop on timeout */
}
+
+ /* awake workers in retry loop */
+ for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
+ wtiWakeupThrd(pThis->pWrkr[i]);
+ }
+
}
pthread_cleanup_pop(1);
@@ -285,10 +301,11 @@ wtpWrkrExecCleanup(wti_t *pWti)
/* the order of the next two statements is important! */
wtiSetState(pWti, WRKTHRD_STOPPED);
- ATOMIC_DEC(pThis->iCurNumWrkThrd);
+ ATOMIC_DEC(&pThis->iCurNumWrkThrd, &pThis->mutCurNumWrkThrd);
- DBGPRINTF("%s: Worker thread %lx, terminated, num workers now %d\n",
- wtpGetDbgHdr(pThis), (unsigned long) pWti, ATOMIC_FETCH_32BIT(pThis->iCurNumWrkThrd));
+ DBGPRINTF("%s: Worker thread %lx, terminated, um workers now %d\n",
+ wtpGetDbgHdr(pThis), (unsigned long) pWti,
+ ATOMIC_FETCH_32BIT(&pThis->iCurNumWrkThrd, &pThis->mutCurNumWrkThrd));
ENDfunc
}
@@ -329,20 +346,28 @@ wtpWrkrExecCancelCleanup(void *arg)
static void *
wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in wtp! */
{
- uchar *pszDbgHdr;
- uchar thrdName[32] = "rs:";
wti_t *pWti = (wti_t*) arg;
wtp_t *pThis;
sigset_t sigSet;
+# if HAVE_PRCTL && defined PR_SET_NAME
+ uchar *pszDbgHdr;
+ uchar thrdName[32] = "rs:";
+# endif
BEGINfunc
ISOBJ_TYPE_assert(pWti, wti);
pThis = pWti->pWtp;
ISOBJ_TYPE_assert(pThis, wtp);
+ /* block all signals */
sigfillset(&sigSet);
pthread_sigmask(SIG_BLOCK, &sigSet, NULL);
+ /* but ignore SIGTTN, which we (ab)use to signal the thread to shutdown -- rgerhards, 2009-07-20 */
+ sigemptyset(&sigSet);
+ sigaddset(&sigSet, SIGTTIN);
+ pthread_sigmask(SIG_UNBLOCK, &sigSet, NULL);
+
# if HAVE_PRCTL && defined PR_SET_NAME
/* set thread name - we ignore if the call fails, has no harsh consequences... */
pszDbgHdr = wtpGetDbgHdr(pThis);
@@ -398,10 +423,11 @@ wtpStartWrkr(wtp_t *pThis)
pWti = pThis->pWrkr[i];
wtiSetState(pWti, WRKTHRD_RUNNING);
iState = pthread_create(&(pWti->thrdID), &pThis->attrThrd, wtpWorker, (void*) pWti);
- ATOMIC_INC(pThis->iCurNumWrkThrd); /* we got one more! */
+ ATOMIC_INC(&pThis->iCurNumWrkThrd, &pThis->mutCurNumWrkThrd); /* we got one more! */
DBGPRINTF("%s: started with state %d, num workers now %d\n",
- wtpGetDbgHdr(pThis), iState, ATOMIC_FETCH_32BIT(pThis->iCurNumWrkThrd));
+ wtpGetDbgHdr(pThis), iState,
+ ATOMIC_FETCH_32BIT(&pThis->iCurNumWrkThrd, &pThis->mutCurNumWrkThrd));
finalize_it:
d_pthread_mutex_unlock(&pThis->mutWtp);
@@ -432,7 +458,7 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
if(nMaxWrkr > pThis->iNumWorkerThreads) /* limit to configured maximum */
nMaxWrkr = pThis->iNumWorkerThreads;
- nMissing = nMaxWrkr - ATOMIC_FETCH_32BIT(pThis->iCurNumWrkThrd);
+ nMissing = nMaxWrkr - ATOMIC_FETCH_32BIT(&pThis->iCurNumWrkThrd, &pThis->mutCurNumWrkThrd);
if(nMissing > 0) {
DBGPRINTF("%s: high activity - starting %d additional worker thread(s).\n",
@@ -442,7 +468,6 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
CHKiRet(wtpStartWrkr(pThis));
}
} else {
-dbgprintf("YYY: wtpAdviseMaxWorkers, sufficient workers, just doing adivse signal cond busy\n");
pthread_cond_signal(pThis->pcondBusy);
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-23 19:52:05 +0000