1 files changed, 65 insertions, 3 deletions

diff --git a/runtime/wti.c b/runtime/wti.c
index f91fb5a9..a9154670 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -44,12 +44,15 @@
 #include "wti.h"
 #include "obj.h"
 #include "glbl.h"
+#include "action.h"
 #include "atomic.h"
 
 /* static data */
 DEFobjStaticHelpers
 DEFobjCurrIf(glbl)
 
+pthread_key_t thrd_wti_key;
+
 /* forward-definitions */
 
 /* methods */
@@ -171,6 +174,8 @@ BEGINobjDestruct(wti) /* be sure to specify the object type also in END and CODE
 CODESTARTobjDestruct(wti)
 	/* actual destruction */
 	batchFree(&pThis->batch);
+	free(pThis->actWrkrInfo);
+	pthread_cond_destroy(&pThis->pcondBusy);
 	DESTROY_ATOMIC_HELPER_MUT(pThis->mutIsRunning);
 
 	free(pThis->pszDbgHdr);
@@ -181,6 +186,7 @@ ENDobjDestruct(wti)
  */
 BEGINobjConstruct(wti) /* be sure to specify the object type also in END macro! */
 	INIT_ATOMIC_HELPER_MUT(pThis->mutIsRunning);
+	pthread_cond_init(&pThis->pcondBusy, NULL);
 ENDobjConstruct(wti)
 
 
@@ -195,11 +201,20 @@ wtiConstructFinalize(wti_t *pThis)
 
 	ISOBJ_TYPE_assert(pThis, wti);
 
-	DBGPRINTF("%s: finalizing construction of worker instance data\n", wtiGetDbgHdr(pThis));
+	DBGPRINTF("%s: finalizing construction of worker instance data (for %d actions)\n",
+		  wtiGetDbgHdr(pThis), iActionNbr);
 
 	/* initialize our thread instance descriptor (no concurrency here) */
 	pThis->bIsRunning = RSFALSE; 
 
+	/* must use calloc as we need zero-init */
+	CHKmalloc(pThis->actWrkrInfo = calloc(iActionNbr, sizeof(actWrkrInfo_t)));
+
+	if(pThis->pWtp == NULL) {
+		dbgprintf("wtiConstructFinalize: pWtp not set, this may be intentional\n");
+		FINALIZE;
+	}
+
 	/* we now alloc the array for user pointers. We obtain the max from the queue itself. */
 	CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize));
 	CHKiRet(batchInit(&pThis->batch, iDeqBatchSize));
@@ -249,10 +264,10 @@ doIdleProcessing(wti_t *pThis, wtp_t *pWtp, int *pbInactivityTOOccured)
 
 	if(pThis->bAlwaysRunning) {
 		/* never shut down any started worker */
-		d_pthread_cond_wait(pWtp->pcondBusy, pWtp->pmutUsr);
+		d_pthread_cond_wait(&pThis->pcondBusy, pWtp->pmutUsr);
 	} else {
 		timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */
-		if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) {
+		if(d_pthread_cond_timedwait(&pThis->pcondBusy, pWtp->pmutUsr, &t) != 0) {
 			DBGPRINTF("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis));
 			*pbInactivityTOOccured = 1; /* indicate we had a timeout */
 		}
@@ -277,6 +292,7 @@ wtiWorker(wti_t *pThis)
 	rsRetVal localRet;
 	rsRetVal terminateRet;
 	int iCancelStateSave;
+	int i;
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, wti);
@@ -286,6 +302,7 @@ wtiWorker(wti_t *pThis)
 	dbgSetThrdName(pThis->pszDbgHdr);
 	pthread_cleanup_push(wtiWorkerCancelCleanup, pThis);
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
+dbgprintf("DDDD: wti %p: worker starting\n", pThis);
 
 	/* now we have our identity, on to real processing */
 	while(1) { /* loop will be broken below - need to do mutex locks */
@@ -332,9 +349,20 @@ wtiWorker(wti_t *pThis)
 		bInactivityTOOccured = 0; /* reset for next run */
 	}
 
+	DBGPRINTF("DDDD: wti %p: worker cleanup up action instances\n", pThis);
+	for(i = 0 ; i < iActionNbr ; ++i) {
+		dbgprintf("wti %p, action %d, ptr %p\n", pThis, i, pThis->actWrkrInfo[i].actWrkrData);
+		if(pThis->actWrkrInfo[i].actWrkrData != NULL) {
+			dbgprintf("DDDD: calling freeWrkrData!\n");
+			pThis->actWrkrInfo[i].pAction->pMod->mod.om.freeWrkrInstance(pThis->actWrkrInfo[i].actWrkrData);
+			pThis->actWrkrInfo[i].actWrkrData = NULL; /* re-init for next activation */
+		}
+	}
+
 	/* indicate termination */
 	pthread_cleanup_pop(0); /* remove cleanup handler */
 	pthread_setcancelstate(iCancelStateSave, NULL);
+dbgprintf("DDDD: wti %p: worker exiting\n", pThis);
 
 	RETiRet;
 }
@@ -374,6 +402,33 @@ finalize_it:
 }
 
 
+/* This function returns (and creates if necessary) a dummy wti suitable
+ * for use by the rule engine. It is intended to be used for direct-mode
+ * main queues (folks, don't do that!). Once created, data is stored in
+ * thread-specific storage.
+ * Note: we do NOT do error checking -- if this functions fails, all the
+ * rest will fail as well... (also, it will only fail under OOM, so...).
+ * Memleak: we leak pWti's when run in direct mode. However, this is only
+ * a cosmetic leak, as we need them until all inputs are terminated,
+ * what means essentially until rsyslog itself is terminated. So we
+ * don't care -- it's just not nice in valgrind, but that's it.
+ */
+wti_t *
+wtiGetDummy(void)
+{
+	wti_t *pWti;
+
+	pWti = (wti_t*) pthread_getspecific(thrd_wti_key);
+	if(pWti == NULL) {
+		wtiConstruct(&pWti);
+		wtiConstructFinalize(pWti);
+		if(pthread_setspecific(thrd_wti_key, pWti) != 0) {
+			DBGPRINTF("wtiGetDummy: error setspecific thrd_wti_key\n");
+		}
+	}
+	return pWti;
+}
+
 /* dummy */
 rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; }
 
@@ -383,6 +438,7 @@ 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);
+	pthread_key_delete(thrd_wti_key);
 ENDObjClassExit(wti)
 
 
@@ -391,8 +447,14 @@ ENDObjClassExit(wti)
  * rgerhards, 2008-01-09
  */
 BEGINObjClassInit(wti, 1, OBJ_IS_CORE_MODULE) /* one is the object version (most important for persisting) */
+	int r;
 	/* request objects we use */
 	CHKiRet(objUse(glbl, CORE_COMPONENT));
+	r = pthread_key_create(&thrd_wti_key, NULL);
+	if(r != 0) {
+		dbgprintf("wti.c: pthread_key_create failed\n");
+		iRet = RS_RET_ERR;
+	}
 ENDObjClassInit(wti)
 
 /* vi:set ai: