on the way to a real input module interface and threading class...

7 files changed, 189 insertions, 13 deletions

diff --git a/module-template.h b/module-template.h
index d54f5f83..4bce7f8d 100644
--- a/module-template.h
+++ b/module-template.h
@@ -28,6 +28,7 @@
 
 #include "modules.h"
 #include "objomsr.h"
+#include "threads.h"
 
 /* macro to define standard output-module static data members
  */
@@ -445,6 +446,24 @@ static rsRetVal modExit(void)\
 	return iRet;\
 }
 
+
+/* runInput()
+ * This is the main function for input modules. It is used to gather data from the
+ * input source and submit it to the message queue. Each runInput() instance has its own
+ * thread. This is handled by the rsyslog engine. It needs to spawn off new threads only
+ * if there is a module-internal need to do so.
+ */
+#define BEGINrunInput \
+static rsRetVal runInput(void)\
+{\
+	DEFiRet;
+
+#define CODESTARTrunInput 
+
+#define ENDrunInput \
+	return iRet;\
+}
+
 /*
  * vi:set ai:
  */
diff --git a/modules.c b/modules.c
index b7ca71eb..6869ede6 100644
--- a/modules.c
+++ b/modules.c
@@ -160,13 +160,19 @@ modInfo_t *modGetNxt(modInfo_t *pThis)
 
 
 /* this function is like modGetNxt(), but it returns pointers to
- * output modules only. As we currently deal just with output modules,
+ * modules of specific type only. As we currently deal just with output modules,
  * it is a dummy, to be filled with real code later.
  * rgerhards, 2007-07-24
  */
-modInfo_t *omodGetNxt(modInfo_t *pThis)
+modInfo_t *modGetNxtType(modInfo_t *pThis, eModType_t rqtdType)
 {
-	return(modGetNxt(pThis));
+	modInfo_t *pMod = pThis;
+
+	do {
+		pMod = modGetNxt(pMod);
+	} while(!(pMod == NULL || pMod->eType == rqtdType)); /* warning: do ... while() */
+
+	return pMod;
 }
 
 
diff --git a/modules.h b/modules.h
index 5b745047..914a65fd 100644
--- a/modules.h
+++ b/modules.h
@@ -73,6 +73,9 @@ typedef struct moduleInfo {
 	 * can allocate instance memory in this call.
 	 */
 	rsRetVal (*createInstance)();
+	/* input module specific members */
+	/* TODO: do a union with members, pass pointer to msg submit function to IM  rger, 2007-12-14 */
+	rsRetVal (*runInput)(void);		/* function to gather input and submit to queue */
 	union	{
 		struct {/* data for input modules */
 			/* input modules come after output modules are finished, I am
@@ -91,7 +94,7 @@ typedef struct moduleInfo {
 
 /* prototypes */
 rsRetVal doModInit(rsRetVal (*modInit)(), uchar *name, void *pModHdlr);
-modInfo_t *omodGetNxt(modInfo_t *pThis);
+modInfo_t *modGetNxtType(modInfo_t *pThis, eModType_t rqtdType);
 uchar *modGetName(modInfo_t *pThis);
 uchar *modGetStateName(modInfo_t *pThis);
 void modPrintList(void);
diff --git a/plugins/immark/immark.c b/plugins/immark/immark.c
index d70069c5..a43ea61e 100644
--- a/plugins/immark/immark.c
+++ b/plugins/immark/immark.c
@@ -39,10 +39,16 @@
 #include <pthread.h>
 #include "rsyslog.h"
 #include "syslogd.h"
+#include "cfsysline.h"
 #include "module-template.h"
 
 MODULE_TYPE_INPUT
 
+/* Module static data */
+/* TODO: this needs a lot of work ;) */
+DEF_OMOD_STATIC_DATA
+static int bDoMarkMessages = 1;
+
 typedef struct _instanceData {
 } instanceData;
 
@@ -58,9 +64,8 @@ typedef struct _instanceData {
  * (and pre 1.20.2 releases of rsyslog) did in mark procesing. They simply
  * do not belong here.
  */
-rsRetVal
-immark_runInput(void)
-{
+BEGINrunInput
+CODESTARTrunInput
 	struct timeval tvSelectTimeout;
 	sigset_t sigSet;
 	sigfillset(&sigSet);
@@ -81,7 +86,8 @@ dbgprintf("immark post select, doing mark, bFinished: %d\n", bFinished);
 	}
 fprintf(stderr, "immark: finished!\n");
 	return RS_RET_OK;
-}
+ENDrunInput
+
 
 BEGINfreeInstance
 CODESTARTfreeInstance
@@ -107,6 +113,8 @@ ENDqueryEtryPt
 BEGINmodInit()
 CODESTARTmodInit
 	*ipIFVersProvided = 1; /* so far, we only support the initial definition */
+CODEmodInit_QueryRegCFSLineHdlr
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"markmessages", 0, eCmdHdlrBinary, NULL, &bDoMarkMessages, STD_LOADABLE_MODULE_ID));
 ENDmodInit
 #endif /* #if 0 */
 /*
diff --git a/syslogd.c b/syslogd.c
index 1a09ed3b..1da475e5 100644
--- a/syslogd.c
+++ b/syslogd.c
@@ -4227,10 +4227,34 @@ finalize_it:
 }
 
 
+/* Start the input modules. This function will probably undergo big changes
+ * while we implement the input module interface. For now, it does the most
+ * important thing to get at least my poor initial input modules up and
+ * running. Almost no config option is taken.
+ * rgerhards, 2007-12-14
+ */
+static rsRetVal
+startInputModules(void)
+{
+	DEFiRet;
+	modInfo_t *pMod;
+
+	/* loop through all modules and activate them (brr...) */
+	pMod = modGetNxtType(NULL, eMOD_IN);
+	while(pMod != NULL) {
+		/* activate here */
+		pMod = modGetNxtType(pMod, eMOD_IN);
+	}
+
+	return iRet;
+}
+
+
 /* INIT -- Initialize syslogd from configuration table
  * init() is called at initial startup AND each time syslogd is HUPed
  */
-static void init(void)
+static void
+init(void)
 {
 	DEFiRet;
 	register int i;
@@ -4422,6 +4446,13 @@ static void init(void)
 
 	Initialized = 1;
 
+	/* the output part and the queue is now ready to run. So it is a good time
+	 * now to start the inputs. Please note that the net code above should be
+	 * shuffled to down here once we have everything in input modules.
+	 * rgerhards, 2007-12-14
+	 */
+	startInputModules();
+
 	if(Debug) {
 		dbgPrintInitInfo();
 	}
@@ -4449,7 +4480,7 @@ static void init(void)
 	sigAct.sa_handler = sighup_handler;
 	sigaction(SIGHUP, &sigAct, NULL);
 
-	dbgprintf(" restarted.\n");
+	dbgprintf(" (re)started.\n");
 }
 
 
@@ -5058,7 +5089,7 @@ static rsRetVal cflineDoAction(uchar **p, action_t **ppAction)
 	assert(ppAction != NULL);
 
 	/* loop through all modules and see if one picks up the line */
-	pMod = omodGetNxt(NULL);
+	pMod = modGetNxtType(NULL, eMOD_OUT);
 	while(pMod != NULL) {
 		iRet = pMod->mod.om.parseSelectorAct(p, &pModData, &pOMSR);
 		dbgprintf("tried selector action for %s: %d\n", modGetName(pMod), iRet);
@@ -5085,7 +5116,7 @@ static rsRetVal cflineDoAction(uchar **p, action_t **ppAction)
 			dbgprintf("error %d parsing config line\n", (int) iRet);
 			break;
 		}
-		pMod = omodGetNxt(pMod);
+		pMod = modGetNxtType(pMod, eMOD_OUT);
 	}
 
 	*ppAction = pAction;
diff --git a/threads.c b/threads.c
index c3aee4b8..d5168d02 100644
--- a/threads.c
+++ b/threads.c
@@ -27,17 +27,111 @@
 #include "config.h"
 #include "rsyslog.h"
 
-#include <pthread.h>
 #include <stdlib.h>
 #include <string.h>
+#include <signal.h>
+#include <pthread.h>
 #include <assert.h>
 
 #include "syslogd.h"
+#include "linkedlist.h"
 #include "threads.h"
 
+/* static data */
 int iMainMsgQueueSize;
 msgQueue *pMsgQueue = NULL;
 
+/* linked list of currently-known threads */
+static linkedList_t llThrds;
+
+/* methods */
+
+/* Construct a new thread object
+ */
+static rsRetVal thrdConstruct(thrdInfo_t **pThis)
+{
+	thrdInfo_t *pNew;
+
+	if((pNew = calloc(1, sizeof(thrdInfo_t))) == NULL)
+		return RS_RET_OUT_OF_MEMORY;
+
+	/* OK, we got the element, now initialize members that should
+	 * not be zero-filled.
+	 */
+
+	*pThis = pNew;
+	return RS_RET_OK;
+}
+
+
+/* Destructs a thread object. The object must not be linked to the
+ * linked list of threads. Please note that the thread should have been
+ * stopped before. If not, we try to do it.
+ */
+static rsRetVal thrdDestruct(thrdInfo_t *pThis)
+{
+	assert(pThis != NULL);
+
+	if(pThis->bIsActive == 1) {
+		thrdTerminate(pThis);
+	}
+	free(pThis);
+
+	return RS_RET_OK;
+}
+
+
+/* terminate a thread gracefully. It's termination sync state is taken into
+ * account.
+ */
+rsRetVal thrdTerminate(thrdInfo_t *pThis)
+{
+	assert(pThis != NULL);
+	
+	if(pThis->eTermTool == eTermSync_SIGNAL) {
+		pthread_kill(pThis->thrdID, SIGUSR2);
+		pthread_join(pThis->thrdID, NULL);
+		/* TODO: TIMEOUT! */
+	} else if(pThis->eTermTool == eTermSync_NONE) {
+		pthread_cancel(pThis->thrdID);
+	}
+	pThis->bIsActive = 0;
+	
+	return RS_RET_OK;
+}
+
+
+/* initialize the thread-support subsystem
+ * must be called once at the start of the program
+ */
+rsRetVal thrdInit(void)
+{
+	DEFiRet;
+
+	iRet = llInit(&llThrds, thrdDestruct, NULL, NULL);
+
+	return iRet;
+}
+
+
+/* de-initialize the thread subsystem
+ * must be called once at the end of the program
+ */
+rsRetVal thrdExit(void)
+{
+	DEFiRet;
+
+	iRet = llDestroy(&llThrds);
+
+	return iRet;
+}
+
+
+
+/* queue functions (may be migrated to some other file...)
+ */
+
+
 msgQueue *queueInit (void)
 {
 	msgQueue *q;
diff --git a/threads.h b/threads.h
index 5f3faab7..dd2b17f7 100644
--- a/threads.h
+++ b/threads.h
@@ -23,6 +23,20 @@
 #ifndef THREADS_H_INCLUDED
 #define THREADS_H_INCLUDED
 
+
+/* type of sync tools for terminating the thread */
+typedef enum eTermSyncType {
+	eTermSync_NONE = 0,	/* no cleanup necessary, just cancel thread */
+	eTermSync_SIGNAL	/* termination via pthread_kill() */
+} eTermSyncType_t;
+
+/* the thread object */
+typedef struct thrdInfo {
+	eTermSyncType_t	eTermTool;
+	int bIsActive;		/* Is thread running? */
+	pthread_t thrdID;
+} thrdInfo_t;
+
 /* this is the first approach to a queue, this time with static
  * memory.
  */
@@ -35,6 +49,7 @@ typedef struct {
 } msgQueue;
 
 /* prototypes */
+rsRetVal thrdTerminate(thrdInfo_t *pThis);
 msgQueue *queueInit (void);
 void queueDelete (msgQueue *q);
 void queueAdd (msgQueue *q, void* in);