Merge branch 'v5-beta'

Conflicts:
	ChangeLog
	configure.ac
	doc/manual.html
	runtime/glbl.c
	tools/omdiscard.c

8 files changed, 54 insertions, 6 deletions

diff --git a/runtime/datetime.h b/runtime/datetime.h
index 82bd415b..70bbf416 100644
--- a/runtime/datetime.h
+++ b/runtime/datetime.h
@@ -28,6 +28,7 @@
 
 /* the datetime object */
 typedef struct datetime_s {
+    	char dummy;
 } datetime_t;
 
 
diff --git a/runtime/errmsg.h b/runtime/errmsg.h
index 799954fb..ac7018b3 100644
--- a/runtime/errmsg.h
+++ b/runtime/errmsg.h
@@ -30,6 +30,7 @@
 
 /* the errmsg object */
 typedef struct errmsg_s {
+    	char dummy;
 } errmsg_t;
 
 
diff --git a/runtime/expr.h b/runtime/expr.h
index 974b71ec..1afe1a1f 100644
--- a/runtime/expr.h
+++ b/runtime/expr.h
@@ -30,6 +30,7 @@
 
 /* a node inside an expression tree */
 typedef struct exprNode_s {
+    	char dummy;
 } exprNode_t;
 
 
diff --git a/runtime/glbl.c b/runtime/glbl.c
index 7dc17df4..0114b1ac 100644
--- a/runtime/glbl.c
+++ b/runtime/glbl.c
@@ -31,6 +31,9 @@
 #include "config.h"
 #include <stdlib.h>
 #include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
 #include <assert.h>
 
 #include "rsyslog.h"
@@ -40,6 +43,7 @@
 #include "glbl.h"
 #include "prop.h"
 #include "atomic.h"
+#include "errmsg.h"
 
 /* some defaults */
 #ifndef DFLT_NETSTRM_DRVR
@@ -49,6 +53,7 @@
 /* static data */
 DEFobjStaticHelpers
 DEFobjCurrIf(prop)
+DEFobjCurrIf(errmsg)
 
 /* static data
  * For this object, these variables are obviously what makes the "meat" of the
@@ -147,6 +152,35 @@ static void SetGlobalInputTermination(void)
 }
 
 
+/* This function is used to set the global work directory name. 
+ * It verifies that the provided directory actually exists and
+ * emits an error message if not.
+ * rgerhards, 2011-02-16
+ */
+static rsRetVal setWorkDir(void __attribute__((unused)) *pVal, uchar *pNewVal)
+{
+	DEFiRet;
+	struct stat sb;
+
+	if(stat((char*) pNewVal, &sb) != 0) {
+		errmsg.LogError(0, RS_RET_ERR_WRKDIR, "$WorkDirectory: %s can not be "
+				"accessed, probably does not exist - directive ignored", pNewVal);
+		ABORT_FINALIZE(RS_RET_ERR_WRKDIR);
+	}
+
+	if(!S_ISDIR(sb.st_mode)) {
+		errmsg.LogError(0, RS_RET_ERR_WRKDIR, "$WorkDirectory: %s not a directory - directive ignored", 
+				pNewVal);
+		ABORT_FINALIZE(RS_RET_ERR_WRKDIR);
+	}
+
+	free(pszWorkDir);
+	pszWorkDir = pNewVal;
+
+finalize_it:
+	RETiRet;
+}
+
 /* return our local hostname. if it is not set, "[localhost]" is returned
  */
 static uchar*
@@ -354,9 +388,10 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
 BEGINAbstractObjClassInit(glbl, 1, OBJ_IS_CORE_MODULE) /* class, version */
 	/* request objects we use */
 	CHKiRet(objUse(prop, CORE_COMPONENT));
+	CHKiRet(objUse(errmsg, CORE_COMPONENT));
 
 	/* register config handlers (TODO: we need to implement a way to unregister them) */
-	CHKiRet(regCfSysLineHdlr((uchar *)"workdirectory", 0, eCmdHdlrGetWord, NULL, &pszWorkDir, NULL, eConfObjGlobal));
+	CHKiRet(regCfSysLineHdlr((uchar *)"workdirectory", 0, eCmdHdlrGetWord, setWorkDir, NULL, NULL, eConfObjGlobal));
 	CHKiRet(regCfSysLineHdlr((uchar *)"dropmsgswithmaliciousdnsptrrecords", 0, eCmdHdlrBinary, NULL, &bDropMalPTRMsgs, NULL, eConfObjGlobal));
 	CHKiRet(regCfSysLineHdlr((uchar *)"defaultnetstreamdriver", 0, eCmdHdlrGetWord, NULL, &pszDfltNetstrmDrvr, NULL, eConfObjGlobal));
 	CHKiRet(regCfSysLineHdlr((uchar *)"defaultnetstreamdrivercafile", 0, eCmdHdlrGetWord, NULL, &pszDfltNetstrmDrvrCAF, NULL, eConfObjGlobal));
diff --git a/runtime/modules.h b/runtime/modules.h
index 9569512b..a78cd6f1 100644
--- a/runtime/modules.h
+++ b/runtime/modules.h
@@ -123,6 +123,7 @@ struct modInfo_s {
 			rsRetVal (*restoreScope)(void);
 		} om;
 		struct { /* data for library modules */
+		    	char dummy;
 		} lm;
 		struct { /* data for parser modules */
 			rsRetVal (*parse)(msg_t*);
diff --git a/runtime/queue.c b/runtime/queue.c
index e4922f37..76327f6a 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -246,6 +246,7 @@ qqueueAdviseMaxWorkers(qqueue_t *pThis)
 
 	if(!pThis->bEnqOnly) {
 		if(pThis->bIsDA && getLogicalQueueSize(pThis) >= pThis->iHighWtrMrk) {
+			DBGOPRINT((obj_t*) pThis, "(re)activating DA worker\n");
 			wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* disk queues have always one worker */
 		} else {
 			if(getLogicalQueueSize(pThis) == 0) {
@@ -1211,7 +1212,6 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread
 	/* set some water marks so that we have useful defaults if none are set specifically */
 	pThis->iFullDlyMrk  = iMaxQueueSize - (iMaxQueueSize / 100) *  3; /* default 97% */
 	pThis->iLightDlyMrk = iMaxQueueSize - (iMaxQueueSize / 100) * 30; /* default 70% */
-
 	pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir);
 	pThis->iMaxFileSize = 1024 * 1024; /* default is 1 MiB */
 	pThis->iQueueSize = 0;
@@ -1497,7 +1497,7 @@ DequeueConsumable(qqueue_t *pThis, wti_t *pWti)
 	 * now that we dequeue batches of pointers, this is much less an issue...
 	 * rgerhards, 2009-04-22
 	 */
-	if(iQueueSize < pThis->iFullDlyMrk / 2) {
+	if(iQueueSize < pThis->iFullDlyMrk / 2 || glbl.GetGlobalInputTermState() == 1) {
 		pthread_cond_broadcast(&pThis->belowFullDlyWtrMrk);
 	}
 
@@ -1819,6 +1819,7 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 {
 	DEFiRet;
 	uchar pszBuf[64];
+	int wrk;
 	uchar *qName;
 	size_t lenBuf;
 
@@ -1841,7 +1842,6 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 	}
 
 	pthread_mutex_init(&pThis->mutThrdMgmt, NULL);
-	pthread_cond_init (&pThis->condDAReady, NULL);
 	pthread_cond_init (&pThis->notFull, NULL);
 	pthread_cond_init (&pThis->notEmpty, NULL);
 	pthread_cond_init (&pThis->belowFullDlyWtrMrk, NULL);
@@ -1850,6 +1850,16 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 	/* call type-specific constructor */
 	CHKiRet(pThis->qConstruct(pThis)); /* this also sets bIsDA */
 
+	/* re-adjust some params if required */
+	if(pThis->bIsDA) {
+		/* if we are in DA mode, we must make sure full delayable messages do not
+		 * initiate going to disk!
+		 */
+		wrk = pThis->iHighWtrMrk - (pThis->iHighWtrMrk / 100) * 50; /* 50% of high water mark */
+		if(wrk < pThis->iFullDlyMrk)
+			pThis->iFullDlyMrk = wrk;
+	}
+
 	DBGOPRINT((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, maxFileSz %lld, lqsize %d, pqsize %d, child %d, "
 				  "full delay %d, light delay %d, deq batch size %d starting\n",
 		  pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->iMaxFileSize,
@@ -2129,7 +2139,6 @@ CODESTARTobjDestruct(qqueue)
 		free(pThis->mut);
 	}
 	pthread_mutex_destroy(&pThis->mutThrdMgmt);
-	pthread_cond_destroy(&pThis->condDAReady);
 	pthread_cond_destroy(&pThis->notFull);
 	pthread_cond_destroy(&pThis->notEmpty);
 	pthread_cond_destroy(&pThis->belowFullDlyWtrMrk);
diff --git a/runtime/queue.h b/runtime/queue.h
index 38e248cd..97057180 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -124,7 +124,6 @@ struct queue_s {
 	pthread_cond_t notFull, notEmpty;
 	pthread_cond_t belowFullDlyWtrMrk; /* below eFLOWCTL_FULL_DELAY watermark */
 	pthread_cond_t belowLightDlyWtrMrk; /* below eFLOWCTL_FULL_DELAY watermark */
-	pthread_cond_t condDAReady;/* signalled when the DA queue is fully initialized and ready for processing */
 	int bThrdStateChanged;		/* at least one thread state has changed if 1 */
 	/* end sync variables */
 	/* the following variables are always present, because they
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index be4707c4..9d3d0289 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -347,6 +347,7 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_ERR_HDFS_WRITE = -2178, /**< error writing to HDFS */
 	RS_RET_ERR_HDFS_OPEN = -2179, /**< error during hdfsOpen (e.g. file does not exist) */
 	RS_RET_FILE_NOT_SPECIFIED = -2180, /**< file name not configured where this was required */
+	RS_RET_ERR_WRKDIR = -2181, /**< problems with the rsyslog working directory */
 
 	RS_RET_INVLD_CONF_OBJ= -2200,	/**< invalid config object (e.g. $Begin conf statement) */
 	RS_RET_ERR_LIBEE_INIT = -2201,	/**< cannot obtain libee ctx */