summaryrefslogtreecommitdiffstats
path: root/runtime
diff options
context:
space:
mode:
authorRainer Gerhards <rgerhards@adiscon.com>2009-07-20 12:18:20 +0200
committerRainer Gerhards <rgerhards@adiscon.com>2009-07-20 12:18:20 +0200
commitbf8125f4e96a011ec28cc58b225bb815f72fc53c (patch)
tree2431ecfe08b01c7ef1756cc8e2b5be9fd4a65c60 /runtime
parent4ec7b9d9ec12d91dde3d030bdaf87cfdd6b5d81d (diff)
downloadrsyslog-bf8125f4e96a011ec28cc58b225bb815f72fc53c.tar.gz
rsyslog-bf8125f4e96a011ec28cc58b225bb815f72fc53c.tar.xz
rsyslog-bf8125f4e96a011ec28cc58b225bb815f72fc53c.zip
bugfix: minor static memory leak while reading configuration
This did NOT leak based on message volume. Also, did some cleanup during the commit.
Diffstat (limited to 'runtime')
-rw-r--r--runtime/queue.c1
-rw-r--r--runtime/wti.c7
-rw-r--r--runtime/wtp.c25
-rw-r--r--runtime/wtp.h1
4 files changed, 15 insertions, 19 deletions
diff --git a/runtime/queue.c b/runtime/queue.c
index 78859e8d..7590af18 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -1956,7 +1956,6 @@ ChkStopWrkrReg(qqueue_t *pThis)
{
DEFiRet;
if(pThis->bEnqOnly) {
-dbgprintf("XXX: terminate_NOW queue:Reg worker: enqOnly! queue size %d\n", getPhysicalQueueSize(pThis));
iRet = RS_RET_TERMINATE_NOW;
} else if(pThis->pqParent != NULL) {
iRet = RS_RET_TERMINATE_WHEN_IDLE;
diff --git a/runtime/wti.c b/runtime/wti.c
index 91c63ffe..900e1cba 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -192,7 +192,7 @@ wtiWorkerCancelCleanup(void *arg)
DBGPRINTF("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis));
- /* call user supplied handler (that one e.g. requeues the element) */
+ /* call user supplied handler */
pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr, pThis->batch.pElem[0].pUsrp);
ENDfunc
@@ -215,7 +215,6 @@ doIdleProcessing(wti_t *pThis, wtp_t *pWtp, int *pbInactivityTOOccured)
pWtp->pfOnIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED);
d_pthread_mutex_lock(pWtp->pmutUsr);
-RUNLOG_VAR("%d", pThis->bAlwaysRunning);
if(pThis->bAlwaysRunning) {
/* never shut down any started worker */
d_pthread_cond_wait(pWtp->pcondBusy, pWtp->pmutUsr);
@@ -321,7 +320,6 @@ wtiSetDbgHdr(wti_t *pThis, uchar *pszMsg, size_t lenMsg)
if(pThis->pszDbgHdr != NULL) {
free(pThis->pszDbgHdr);
- pThis->pszDbgHdr = NULL;
}
if((pThis->pszDbgHdr = malloc(sizeof(uchar) * lenMsg + 1)) == NULL)
@@ -355,6 +353,5 @@ BEGINObjClassInit(wti, 1, OBJ_IS_CORE_MODULE) /* one is the object version (most
CHKiRet(objUse(glbl, CORE_COMPONENT));
ENDObjClassInit(wti)
-/*
- * vi:set ai:
+/* vi:set ai:
*/
diff --git a/runtime/wtp.c b/runtime/wtp.c
index 23726802..af4c7621 100644
--- a/runtime/wtp.c
+++ b/runtime/wtp.c
@@ -89,6 +89,8 @@ static rsRetVal NotImplementedDummy() { return RS_RET_NOT_IMPLEMENTED; }
BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! */
pthread_mutex_init(&pThis->mutWtp, NULL);
pthread_cond_init(&pThis->condThrdTrm, NULL);
+ pthread_attr_init(&pThis->attrThrd);
+ pthread_attr_setdetachstate(&pThis->attrThrd, PTHREAD_CREATE_DETACHED);
/* set all function pointers to "not implemented" dummy so that we can safely call them */
pThis->pfChkStopWrkr = NotImplementedDummy;
pThis->pfGetDeqBatchSize = NotImplementedDummy;
@@ -152,6 +154,7 @@ CODESTARTobjDestruct(wtp)
/* actual destruction */
pthread_cond_destroy(&pThis->condThrdTrm);
pthread_mutex_destroy(&pThis->mutWtp);
+ pthread_attr_destroy(&pThis->attrThrd);
free(pThis->pszDbgHdr);
ENDobjDestruct(wtp)
@@ -284,9 +287,9 @@ wtpCancelAll(wtp_t *pThis)
}
-/* cancellation cleanup handler for executing worker
- * decrements the worker counter
- * rgerhards, 2008-01-20
+/* cancellation cleanup handler for executing worker decrements the worker counter.
+ * This is also called when the the worker is normally shut down.
+ * rgerhards, 2009-07-20
*/
static void
wtpWrkrExecCancelCleanup(void *arg)
@@ -299,6 +302,7 @@ wtpWrkrExecCancelCleanup(void *arg)
pThis = pWti->pWtp;
ISOBJ_TYPE_assert(pThis, wtp);
+ /* the order of the next two statements is important! */
wtiSetState(pWti, WRKTHRD_STOPPED);
ATOMIC_DEC(pThis->iCurNumWrkThrd);
pthread_cond_broadcast(&pThis->condThrdTrm); /* activate anyone waiting on thread shutdown */
@@ -352,18 +356,16 @@ wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in
/* start a new worker */
static rsRetVal
-wtpStartWrkr(wtp_t *pThis, int bLockMutex)
+wtpStartWrkr(wtp_t *pThis)
{
- DEFVARS_mutexProtection;
wti_t *pWti;
int i;
int iState;
- pthread_attr_t attr;
DEFiRet;
ISOBJ_TYPE_assert(pThis, wtp);
- BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp, bLockMutex);
+ d_pthread_mutex_lock(&pThis->mutWtp);
/* find free spot in thread table. */
for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
@@ -381,17 +383,14 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex)
pWti = pThis->pWrkr[i];
wtiSetState(pWti, WRKTHRD_RUNNING);
- pthread_attr_init(&attr);
- pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
- iState = pthread_create(&(pWti->thrdID), &attr, wtpWorker, (void*) pWti);
- pthread_attr_destroy(&attr); /* TODO: we could globally reuse such an attribute 2009-07-08 */
+ iState = pthread_create(&(pWti->thrdID), &pThis->attrThrd, wtpWorker, (void*) pWti);
ATOMIC_INC(pThis->iCurNumWrkThrd); /* we got one more! */
DBGPRINTF("%s: started with state %d, num workers now %d\n",
wtpGetDbgHdr(pThis), iState, ATOMIC_FETCH_32BIT(pThis->iCurNumWrkThrd));
finalize_it:
- END_MTX_PROTECTED_OPERATIONS(&pThis->mutWtp);
+ d_pthread_mutex_unlock(&pThis->mutWtp);
RETiRet;
}
@@ -425,7 +424,7 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr)
DBGPRINTF("%s: high activity - starting %d additional worker thread(s).\n", wtpGetDbgHdr(pThis), nMissing);
/* start the rqtd nbr of workers */
for(i = 0 ; i < nMissing ; ++i) {
- CHKiRet(wtpStartWrkr(pThis, LOCK_MUTEX));
+ CHKiRet(wtpStartWrkr(pThis));
}
} else {
pthread_cond_signal(pThis->pcondBusy);
diff --git a/runtime/wtp.h b/runtime/wtp.h
index 358ced3a..0505b91c 100644
--- a/runtime/wtp.h
+++ b/runtime/wtp.h
@@ -55,6 +55,7 @@ struct wtp_s {
/* end sync variables */
/* user objects */
void *pUsr; /* pointer to user object (in this case, the queue the wtp belongs to) */
+ pthread_attr_t attrThrd;/* attribute for new threads (created just once and cached here) */
pthread_mutex_t *pmutUsr;
pthread_cond_t *pcondBusy; /* condition the user will signal "busy again, keep runing" on (awakes worker) */
rsRetVal (*pfChkStopWrkr)(void *pUsr, int);