internal: added ability to terminate input modules not via pthread_cancel...

... but an alternate approach via pthread_kill. This is somewhat safer as we
do not need to think about the cancel-safeness of all libraries we use.
However, not all inputs can easily supported, so this now is a feature
that can be requested by the input module (the most important ones
request it).

15 files changed, 134 insertions, 13 deletions

diff --git a/ChangeLog b/ChangeLog
index a3316ddc..f9857a75 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -11,6 +11,12 @@ Version 5.1.3  [DEVEL] (rgerhards), 2009-07-??
   sizes had been specified in the property replacer.
 - bugfix: minor static memory leak while reading configuration
   did NOT leak based on message volume
+- internal: added ability to terminate input modules not via pthread_cancel
+  but an alternate approach via pthread_kill. This is somewhat safer as we
+  do not need to think about the cancel-safeness of all libraries we use.
+  However, not all inputs can easily supported, so this now is a feature
+  that can be requested by the input module (the most important ones
+  request it).
 ---------------------------------------------------------------------------
 Version 5.1.2  [DEVEL] (rgerhards), 2009-07-08
 - bugfix: properties inputname, fromhost, fromhost-ip, msg were lost when
diff --git a/plugins/imdiag/imdiag.c b/plugins/imdiag/imdiag.c
index 7cacc44a..9602f50d 100644
--- a/plugins/imdiag/imdiag.c
+++ b/plugins/imdiag/imdiag.c
@@ -447,10 +447,17 @@ resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unus
 }
 
 
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
 
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_IMOD_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
 ENDqueryEtryPt
 
 
diff --git a/plugins/imgssapi/imgssapi.c b/plugins/imgssapi/imgssapi.c
index d8791880..f9c1f86b 100644
--- a/plugins/imgssapi/imgssapi.c
+++ b/plugins/imgssapi/imgssapi.c
@@ -674,9 +674,17 @@ CODESTARTafterRun
 ENDafterRun
 
 
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
+
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_IMOD_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
 ENDqueryEtryPt
 
 
diff --git a/plugins/imtcp/imtcp.c b/plugins/imtcp/imtcp.c
index e1f513c8..1f111ef3 100644
--- a/plugins/imtcp/imtcp.c
+++ b/plugins/imtcp/imtcp.c
@@ -250,6 +250,13 @@ CODESTARTafterRun
 ENDafterRun
 
 
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
+
 BEGINmodExit
 CODESTARTmodExit
 	if(pOurTcpsrv != NULL)
@@ -287,6 +294,7 @@ resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unus
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_IMOD_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
 ENDqueryEtryPt
 
 
diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c
index 718c3090..ea41e4f6 100644
--- a/plugins/imudp/imudp.c
+++ b/plugins/imudp/imudp.c
@@ -329,6 +329,8 @@ CODESTARTrunInput
 
 		/* wait for io to become ready */
 		nfds = select(maxfds+1, (fd_set *) &readfds, NULL, NULL, NULL);
+		if(glbl.GetGlobalInputTermState() == 1)
+			break; /* terminate input! */
 
 	       for(i = 0; nfds && i < *udpLstnSocks; i++) {
 			if(FD_ISSET(udpLstnSocks[i+1], &readfds)) {
@@ -395,9 +397,17 @@ CODESTARTmodExit
 ENDmodExit
 
 
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
+
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_IMOD_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
 ENDqueryEtryPt
 
 static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index 6d60efaf..c099be56 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -282,6 +282,8 @@ CODESTARTrunInput
 
 		/* wait for io to become ready */
 		nfds = select(maxfds+1, (fd_set *) &readfds, NULL, NULL, NULL);
+		if(glbl.GetGlobalInputTermState() == 1)
+			break; /* terminate input! */
 
 		for (i = 0; i < nfunix && nfds > 0; i++) {
 			if ((fd = funix[i]) != -1 && FD_ISSET(fd, &readfds)) {
@@ -352,9 +354,17 @@ CODESTARTmodExit
 ENDmodExit
 
 
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
+
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_IMOD_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
 ENDqueryEtryPt
 
 static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
diff --git a/runtime/glbl.c b/runtime/glbl.c
index 534dc262..f27b8e73 100644
--- a/runtime/glbl.c
+++ b/runtime/glbl.c
@@ -39,6 +39,7 @@
 #include "cfsysline.h"
 #include "glbl.h"
 #include "prop.h"
+#include "atomic.h"
 
 /* some defaults */
 #ifndef DFLT_NETSTRM_DRVR
@@ -71,6 +72,7 @@ static uchar *pszDfltNetstrmDrvr = NULL; /* module name of default netstream dri
 static uchar *pszDfltNetstrmDrvrCAF = NULL; /* default CA file for the netstrm driver */
 static uchar *pszDfltNetstrmDrvrKeyFile = NULL; /* default key file for the netstrm driver (server) */
 static uchar *pszDfltNetstrmDrvrCertFile = NULL; /* default cert file for the netstrm driver (server) */
+static int bTerminateInputs = 0;		/* global switch that inputs shall terminate ASAP (1=> terminate) */
 
 
 /* define a macro for the simple properties' set and get functions
@@ -115,6 +117,24 @@ SIMP_PROP_SET(DfltNetstrmDrvrCertFile, pszDfltNetstrmDrvrCertFile, uchar*) /* TO
 #undef SIMP_PROP_GET
 
 
+/* return global input termination status
+ * rgerhards, 2009-07-20
+ */
+static int GetGlobalInputTermState(void)
+{
+	return ATOMIC_FETCH_32BIT(bTerminateInputs);
+}
+
+
+/* set global termiantion state to "terminate". Note that this is a
+ * "once in a lifetime" action which can not be undone. -- gerhards, 2009-07-20
+ */
+static void SetGlobalInputTermination(void)
+{
+	ATOMIC_STORE_1_TO_INT(bTerminateInputs);
+}
+
+
 /* return our local hostname. if it is not set, "[localhost]" is returned
  */
 static uchar*
@@ -239,6 +259,8 @@ CODESTARTobjQueryInterface(glbl)
 	pIf->GetWorkDir = GetWorkDir;
 	pIf->GenerateLocalHostNameProperty = GenerateLocalHostNameProperty;
 	pIf->GetLocalHostNameProp = GetLocalHostNameProp;
+	pIf->SetGlobalInputTermination = SetGlobalInputTermination;
+	pIf->GetGlobalInputTermState = GetGlobalInputTermState;
 #define SIMP_PROP(name) \
 	pIf->Get##name = Get##name; \
 	pIf->Set##name = Set##name;
diff --git a/runtime/glbl.h b/runtime/glbl.h
index 8ecd8466..0d0c8210 100644
--- a/runtime/glbl.h
+++ b/runtime/glbl.h
@@ -61,9 +61,12 @@ BEGINinterface(glbl) /* name must also be changed in ENDinterface macro! */
 	/* added v3, 2009-06-30 */
 	rsRetVal (*GenerateLocalHostNameProperty)(void);
 	prop_t* (*GetLocalHostNameProp)(void);
+	/* added v4, 2009-07-20 */
+	int (*GetGlobalInputTermState)(void);
+	void (*SetGlobalInputTermination)(void);
 #undef	SIMP_PROP
 ENDinterface(glbl)
-#define glblCURR_IF_VERSION 3 /* increment whenever you change the interface structure! */
+#define glblCURR_IF_VERSION 4 /* increment whenever you change the interface structure! */
 /* version 2 had PreserveFQDN added - rgerhards, 2008-12-08 */
 
 /* the remaining prototypes */
diff --git a/runtime/module-template.h b/runtime/module-template.h
index 3e963199..d49da2c9 100644
--- a/runtime/module-template.h
+++ b/runtime/module-template.h
@@ -368,6 +368,17 @@ static rsRetVal queryEtryPt(uchar *name, rsRetVal (**pEtryPoint)())\
 		*pEtryPoint = endTransaction;\
 	}
 
+
+/* the following definition is a queryEtryPt block that must be added
+ * if a non-output module supports "isCompatibleWithFeature".
+ * rgerhards, 2009-07-20
+ */
+#define CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES \
+	  else if(!strcmp((char*) name, "isCompatibleWithFeature")) {\
+		*pEtryPoint = isCompatibleWithFeature;\
+	}
+
+
 /* the following definition is the standard block for queryEtryPt for INPUT
  * modules. This can be used if no specific handling (e.g. to cover version
  * differences) is needed.
diff --git a/runtime/modules.c b/runtime/modules.c
index 0ec6b1ce..b588909e 100644
--- a/runtime/modules.c
+++ b/runtime/modules.c
@@ -77,8 +77,9 @@ static modInfo_t *pLoadedModulesLast = NULL;	/* tail-pointer */
 uchar	*pModDir = NULL; /* read-only after startup */
 
 
-/* we provide a set of dummy functions for output modules that do not support the
- * transactional interface. As they do not do this, they commit each message they
+/* we provide a set of dummy functions for modules that do not support the
+ * some interfaces.
+ * On the commit feature: As the modules do not support it, they commit each message they
  * receive, and as such the dummies can always return RS_RET_OK without causing
  * harm. This simplifies things as in action processing we do not need to check
  * if the transactional entry points exist.
@@ -91,6 +92,11 @@ static rsRetVal dummyEndTransaction()
 {
 	return RS_RET_OK;
 }
+static rsRetVal dummyIsCompatibleWithFeature() 
+{
+dbgprintf("XXX: dummy isCompatibleWithFeature called!\n");
+	return RS_RET_INCOMPATIBLE;
+}
 
 #ifdef DEBUG
 /* we add some home-grown support to track our users (and detect who does not free us). In
@@ -428,6 +434,11 @@ doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_
 	 */
 	CHKiRet((*pNew->modQueryEtryPt)((uchar*)"modGetID", &pNew->modGetID));
 	CHKiRet((*pNew->modQueryEtryPt)((uchar*)"modExit", &pNew->modExit));
+	localRet = (*pNew->modQueryEtryPt)((uchar*)"isCompatibleWithFeature", &pNew->isCompatibleWithFeature);
+	if(localRet == RS_RET_MODULE_ENTRY_POINT_NOT_FOUND)
+		pNew->isCompatibleWithFeature = dummyIsCompatibleWithFeature;
+	else if(localRet != RS_RET_OK)
+		ABORT_FINALIZE(localRet);
 
 	/* ... and now the module-specific interfaces */
 	switch(pNew->eType) {
@@ -442,7 +453,6 @@ doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_
 			CHKiRet((*pNew->modQueryEtryPt)((uchar*)"dbgPrintInstInfo", &pNew->dbgPrintInstInfo));
 			CHKiRet((*pNew->modQueryEtryPt)((uchar*)"doAction", &pNew->mod.om.doAction));
 			CHKiRet((*pNew->modQueryEtryPt)((uchar*)"parseSelectorAct", &pNew->mod.om.parseSelectorAct));
-			CHKiRet((*pNew->modQueryEtryPt)((uchar*)"isCompatibleWithFeature", &pNew->isCompatibleWithFeature));
 			CHKiRet((*pNew->modQueryEtryPt)((uchar*)"tryResume", &pNew->tryResume));
 			/* try load optional interfaces */
 			localRet = (*pNew->modQueryEtryPt)((uchar*)"doHUP", &pNew->doHUP);
diff --git a/runtime/syslogd-types.h b/runtime/syslogd-types.h
index 4a26f993..161ee06f 100644
--- a/runtime/syslogd-types.h
+++ b/runtime/syslogd-types.h
@@ -56,7 +56,8 @@
  * applications I do not yet envision. -- rgerhards, 2007-07-24
  */
 typedef enum _syslogFeature {
-	sFEATURERepeatedMsgReduction = 1
+	sFEATURERepeatedMsgReduction = 1,
+	sFEATURENonCancelInputTermination = 2
 } syslogFeature;
 
 /* we define our own facility and severities */
diff --git a/tcpsrv.c b/tcpsrv.c
index 776dc46b..5c1a4ac8 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -549,6 +549,8 @@ Run(tcpsrv_t *pThis)
 
 		/* wait for io to become ready */
 		CHKiRet(nssel.Wait(pSel, &nfds));
+		if(glbl.GetGlobalInputTermState() == 1)
+			break; /* terminate input! */
 
 		for(i = 0 ; i < pThis->iLstnMax ; ++i) {
 			CHKiRet(nssel.IsReady(pSel, pThis->ppLstn[i], NSDSEL_RD, &bIsReady, &nfds));
@@ -580,7 +582,7 @@ finalize_it: /* this is a very special case - this time only we do not exit the
 	}
 
 	/* note that this point is usually not reached */
-	pthread_cleanup_pop(0); /* remove cleanup handler */
+	pthread_cleanup_pop(1); /* remove cleanup handler */
 
 	RETiRet;
 }
diff --git a/threads.c b/threads.c
index 13222694..20ab9f66 100644
--- a/threads.c
+++ b/threads.c
@@ -51,8 +51,7 @@ static rsRetVal thrdConstruct(thrdInfo_t **ppThis)
 
 	assert(ppThis != NULL);
 
-	if((pThis = calloc(1, sizeof(thrdInfo_t))) == NULL)
-		return RS_RET_OUT_OF_MEMORY;
+	CHKmalloc(pThis = calloc(1, sizeof(thrdInfo_t)));
 
 	/* OK, we got the element, now initialize members that should
 	 * not be zero-filled.
@@ -61,6 +60,8 @@ static rsRetVal thrdConstruct(thrdInfo_t **ppThis)
 	pthread_mutex_init (pThis->mutTermOK, NULL);
 
 	*ppThis = pThis;
+
+finalize_it:
 	RETiRet;
 }
 
@@ -91,8 +92,14 @@ rsRetVal thrdTerminate(thrdInfo_t *pThis)
 	DEFiRet;
 	assert(pThis != NULL);
 	
-	pthread_cancel(pThis->thrdID);
-	pthread_join(pThis->thrdID, NULL); /* wait for cancel to complete */
+	if(pThis->bNeedsCancel) {
+		DBGPRINTF("request term via canceling for input thread 0x%x\n", (unsigned) pThis->thrdID);
+		pthread_cancel(pThis->thrdID);
+	} else {
+		DBGPRINTF("request term via SIGTTIN for input thread 0x%x\n", (unsigned) pThis->thrdID);
+		pthread_kill(pThis->thrdID, SIGTTIN);
+	}
+	pthread_join(pThis->thrdID, NULL); /* wait for input thread to complete */
 	pThis->bIsActive = 0;
 
 	/* call cleanup function, if any */
@@ -132,6 +139,11 @@ static void* thrdStarter(void *arg)
 	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);
+
 	/* setup complete, we are now ready to execute the user code. We will not
 	 * regain control until the user code is finished, in which case we terminate
 	 * the thread.
@@ -147,7 +159,7 @@ static void* thrdStarter(void *arg)
  * executing threads. It is added at the end of the list.
  * rgerhards, 2007-12-14
  */
-rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdInfo_t *))
+rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdInfo_t *), bool bNeedsCancel)
 {
 	DEFiRet;
 	thrdInfo_t *pThis;
@@ -159,6 +171,7 @@ rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdI
 	pThis->bIsActive = 1;
 	pThis->pUsrThrdMain = thrdMain;
 	pThis->pAfterRun = afterRun;
+	pThis->bNeedsCancel = bNeedsCancel;
 	i = pthread_create(&pThis->thrdID, NULL, thrdStarter, pThis);
 	CHKiRet(llAppend(&llThrds, NULL, pThis));
 
diff --git a/threads.h b/threads.h
index 78924d95..c37157fe 100644
--- a/threads.h
+++ b/threads.h
@@ -31,6 +31,7 @@ struct thrdInfo {
 	rsRetVal (*pUsrThrdMain)(struct thrdInfo*); /* user thread main to be called in new thread */
 	rsRetVal (*pAfterRun)(struct thrdInfo*);   /* cleanup function */
 	pthread_t thrdID;
+	bool bNeedsCancel;	/* must input be terminated by pthread_cancel()? */
 };
 
 /* prototypes */
@@ -38,7 +39,7 @@ rsRetVal thrdExit(void);
 rsRetVal thrdInit(void);
 rsRetVal thrdTerminate(thrdInfo_t *pThis);
 rsRetVal thrdTerminateAll(void);
-rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdInfo_t *));
+rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdInfo_t *), bool);
 rsRetVal thrdSleep(thrdInfo_t *pThis, int iSeconds, int iuSeconds);
 
 /* macros (replace inline functions) */
diff --git a/tools/syslogd.c b/tools/syslogd.c
index ada3288e..40e9e92c 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -1664,6 +1664,7 @@ die(int sig)
 
 	/* close the inputs */
 	DBGPRINTF("Terminating input threads...\n");
+	glbl.SetGlobalInputTermination();
 	thrdTerminateAll();
 
 	/* and THEN send the termination log message (see long comment above) */
@@ -2083,6 +2084,7 @@ static rsRetVal
 runInputModules(void)
 {
 	modInfo_t *pMod;
+	int bNeedsCancel;
 
 	BEGINfunc
 	/* loop through all modules and activate them (brr...) */
@@ -2090,7 +2092,9 @@ runInputModules(void)
 	while(pMod != NULL) {
 		if(pMod->mod.im.bCanRun) {
 			/* activate here */
-			thrdCreate(pMod->mod.im.runInput, pMod->mod.im.afterRun);
+			bNeedsCancel = (pMod->isCompatibleWithFeature(sFEATURENonCancelInputTermination) == RS_RET_OK) ?
+				       0 : 1;
+			thrdCreate(pMod->mod.im.runInput, pMod->mod.im.afterRun, bNeedsCancel);
 		}
 	pMod = module.GetNxtType(pMod, eMOD_IN);
 	}
@@ -2429,6 +2433,9 @@ void sighup_handler()
 	sigaction(SIGHUP, &sigAct, NULL);
 }
 
+void sigttin_handler()
+{
+}
 
 /* this function pulls all internal messages from the buffer
  * and puts them into the processing engine.
@@ -3036,6 +3043,8 @@ doGlblProcessInit(void)
 	sigaction(SIGCHLD, &sigAct, NULL);
 	sigAct.sa_handler = Debug ? debug_switch : SIG_IGN;
 	sigaction(SIGUSR1, &sigAct, NULL);
+	sigAct.sa_handler = sigttin_handler;
+	sigaction(SIGTTIN, &sigAct, NULL); /* (ab)used to interrupt input threads */
 	sigAct.sa_handler = SIG_IGN;
 	sigaction(SIGPIPE, &sigAct, NULL);
 	sigaction(SIGXFSZ, &sigAct, NULL); /* do not abort if 2gig file limit is hit */