added the capability to have ruleset-specific main message queues

This offers considerable additional flexibility AND superior performance
(in cases where multiple inputs now can avoid lock contention)

5 files changed, 73 insertions, 3 deletions

diff --git a/runtime/msg.h b/runtime/msg.h
index f7d74597..9101cef7 100644
--- a/runtime/msg.h
+++ b/runtime/msg.h
@@ -208,6 +208,16 @@ MsgSetRawMsgSize(msg_t *pMsg, size_t newLen)
 }
 
 
+/* get the ruleset that is associated with the ruleset.
+ * May be NULL. -- rgerhards, 2009-10-27
+ */
+static inline ruleset_t*
+MsgGetRuleset(msg_t *pMsg)
+{
+	return pMsg->pRuleset;
+}
+
+
 #endif /* #ifndef MSG_H_INCLUDED */
 /* vim:set ai:
  */
diff --git a/runtime/queue.h b/runtime/queue.h
index 26c57a50..93573dae 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -55,7 +55,7 @@ typedef struct qLinkedList_S {
 
 
 /* the queue object */
-typedef struct queue_s {
+struct queue_s {
 	BEGINobjInstance;
 	queueType_t	qType;
 	int	nLogDeq;	/* number of elements currently logically dequeued */
@@ -160,7 +160,7 @@ typedef struct queue_s {
 			strm_t *pReadDel; /* current file for deleting */
 		} disk;
 	} tVars;
-} qqueue_t;
+};
 
 
 /* the define below is an "eternal" timeout for the timeout settings which require a value.
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index bd1c936e..7c966276 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -107,6 +107,7 @@ typedef struct wti_s wti_t;
 typedef obj_t nsd_t;
 typedef obj_t nsdsel_t;
 typedef struct msg msg_t;
+typedef struct queue_s qqueue_t;
 typedef struct prop_s prop_t;
 typedef struct interface_s interface_t;
 typedef struct objInfo_s objInfo_t;
@@ -387,6 +388,8 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_HOST_NOT_SPECIFIED = -2151, /**< (target) host was not specified where it was needed */
 	RS_RET_ERR_LIBNET_INIT = -2152, /**< error initializing libnet */
 	RS_RET_FORCE_TERM = -2153,	/**< thread was forced to terminate by bShallShutdown, a state, not an error */
+	RS_RET_RULES_QUEUE_EXISTS = -2154,/**< we were instructed to create a new ruleset queue, but one already exists */
+	RS_RET_NO_CURR_RULESET = -2155,/**< no current ruleset exists (but one is required) */
 
 	/* RainerScript error messages (range 1000.. 1999) */
 	RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index d3de672e..3592aafd 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -40,11 +40,13 @@
 
 #include "rsyslog.h"
 #include "obj.h"
+#include "cfsysline.h"
 #include "msg.h"
 #include "ruleset.h"
 #include "rule.h"
 #include "errmsg.h"
 #include "unicode-helper.h"
+#include "dirty.h" /* for main ruleset queue creation */
 
 /* static data */
 DEFobjStaticHelpers
@@ -214,6 +216,19 @@ GetCurrent(void)
 }
 
 
+/* get main queue associated with ruleset. If no ruleset-specifc main queue
+ * is set, the primary main message queue is returned.
+ * We use a non-standard calling interface, as nothing can go wrong and it
+ * is really much more natural to return the pointer directly.
+ */
+static qqueue_t*
+GetRulesetQueue(ruleset_t *pThis)
+{
+	ISOBJ_TYPE_assert(pThis, ruleset);
+	return (pThis->pQueue == NULL) ? pMsgQueue : pThis->pQueue;
+}
+
+
 /* Find the ruleset with the given name and return a pointer to its object.
  */
 static rsRetVal
@@ -384,6 +399,41 @@ debugPrintAll(void)
 }
 
 
+/* Create a ruleset-specific "main" queue for this ruleset. If one is already
+ * defined, an error message is emitted but nothing else is done.
+ * Note: we use the main message queue parameters for queue creation and access
+ * syslogd.c directly to obtain these. This is far from being perfect, but
+ * considered acceptable for the time being.
+ * rgerhards, 2009-10-27
+ */
+static rsRetVal
+rulesetCreateQueue(void __attribute__((unused)) *pVal, int *pNewVal)
+{
+	DEFiRet;
+
+	if(pCurrRuleset == NULL) {
+		errmsg.LogError(0, RS_RET_NO_CURR_RULESET, "error: currently no specific ruleset specified, thus a "
+				"queue can not be added to it");
+		ABORT_FINALIZE(RS_RET_NO_CURR_RULESET);
+	}
+
+	if(pCurrRuleset->pQueue != NULL) {
+		errmsg.LogError(0, RS_RET_RULES_QUEUE_EXISTS, "error: ruleset already has a main queue, can not "
+				"add another one");
+		ABORT_FINALIZE(RS_RET_RULES_QUEUE_EXISTS);
+	}
+
+	if(pNewVal == 0)
+		FINALIZE; /* if it is turned off, we do not need to change anything ;) */
+
+	dbgprintf("adding a ruleset-specific \"main\" queue");
+	CHKiRet(createMainQueue(&pCurrRuleset->pQueue, UCHAR_CONSTANT("ruleset")));
+
+finalize_it:
+	RETiRet;
+}
+
+
 /* queryInterface function
  * rgerhards, 2008-02-21
  */
@@ -413,6 +463,7 @@ CODESTARTobjQueryInterface(ruleset)
 	pIf->GetRuleset = GetRuleset;
 	pIf->SetDefaultRuleset = SetDefaultRuleset;
 	pIf->SetCurrRuleset = SetCurrRuleset;
+	pIf->GetRulesetQueue = GetRulesetQueue;
 finalize_it:
 ENDobjQueryInterface(ruleset)
 
@@ -442,6 +493,9 @@ BEGINObjClassInit(ruleset, 1, OBJ_IS_CORE_MODULE) /* class, version */
 
 	/* prepare global data */
 	CHKiRet(llInit(&llRulesets, rulesetDestructForLinkedList, keyDestruct, strcasecmp));
+
+	/* config file handlers */
+	CHKiRet(regCfSysLineHdlr((uchar *)"rulesetcreatemainqueue", 0, eCmdHdlrBinary, rulesetCreateQueue, NULL, NULL));
 ENDObjClassInit(ruleset)
 
 /* vi:set ai:
diff --git a/runtime/ruleset.h b/runtime/ruleset.h
index 32571687..deea9405 100644
--- a/runtime/ruleset.h
+++ b/runtime/ruleset.h
@@ -25,6 +25,7 @@
 #ifndef INCLUDED_RULESET_H
 #define INCLUDED_RULESET_H
 
+#include "queue.h"
 #include "linkedlist.h"
 
 /* the ruleset object */
@@ -32,6 +33,7 @@ struct ruleset_s {
 	BEGINobjInstance;	/* Data to implement generic object - MUST be the first data element! */
 	linkedList_t llRules;	/* this is NOT a pointer - no typo here ;) */
 	uchar *pszName;		/* name of our ruleset */
+	qqueue_t *pQueue;	/* "main" message queue, if the ruleset has its own (else NULL) */
 };
 
 /* interfaces */
@@ -50,8 +52,9 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */
 	rsRetVal (*SetDefaultRuleset)(uchar*);
 	rsRetVal (*SetCurrRuleset)(uchar*);
 	ruleset_t* (*GetCurrent)(void);
+	qqueue_t* (*GetRulesetQueue)(ruleset_t*);
 ENDinterface(ruleset)
-#define rulesetCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+#define rulesetCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */
 
 
 /* prototypes */