Merge branch 'v5-stable-imfile-batches' into v5-devel

Conflicts:
	ChangeLog
	configure.ac

8 files changed, 48 insertions, 9 deletions

diff --git a/ChangeLog b/ChangeLog
index a9084201..1150de9a 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,5 +1,7 @@
 ---------------------------------------------------------------------------
 Version 5.9.0  [V5-DEVEL] (rgerhards), 2011-03-??
+- imfile: added $InputFileMaxLinesAtOnce directive
+- enhanced imfile to support input batching
 - added capability for imtcp and imptcp to activate keep-alive packets
   at the socket layer. This has not been added to imttcp, as the latter is
   only an experimental module, and one which did not prove to be useful.
@@ -29,7 +31,17 @@ Version 5.9.0  [V5-DEVEL] (rgerhards), 2011-03-??
   affected directive was: $ActionExecOnlyWhenPreviousIsSuspended on
   closes: http://bugzilla.adiscon.com/show_bug.cgi?id=236
 ---------------------------------------------------------------------------
-Version 5.8.1  [V5-stable] (rgerhards), 2011-04-??
+Version 5.8.1  [V5-stable] (rgerhards), 2011-05-19
+- bugfix: invalid processing in QUEUE_FULL condition
+  If the the multi-submit interface was used and a QUEUE_FULL condition
+  occured, the failed message was properly destructed. However, the
+  rest of the input batch, if it existed, was not processed. So this
+  lead to potential loss of messages and a memory leak. The potential
+  loss of messages was IMHO minor, because they would have been dropped
+  in most cases due to the queue remaining full, but very few lucky ones
+  from the batch may have made it. Anyhow, this has now been changed so
+  that the rest of the batch is properly tried to be enqueued and, if
+  not possible, destructed.
 - new module mmsnmptrapd, a sample message modification module
   This can be useful to reformat snmptrapd messages and also serves as
   a sample for how to write message modification modules using the
diff --git a/configure.ac b/configure.ac
index fb0cff8d..1faaff89 100644
--- a/configure.ac
+++ b/configure.ac
@@ -2,7 +2,7 @@
 # Process this file with autoconf to produce a configure script.
 
 AC_PREREQ(2.61)
-AC_INIT([rsyslog],[5.8.1-pre3],[rsyslog@lists.adiscon.com])
+AC_INIT([rsyslog],[5.8.1],[rsyslog@lists.adiscon.com])
 AM_INIT_AUTOMAKE
 
 m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])])
diff --git a/doc/imfile.html b/doc/imfile.html
index 60726ceb..60bbbeea 100644
--- a/doc/imfile.html
+++ b/doc/imfile.html
@@ -98,9 +98,16 @@ performance, especially when set to a low value. Frequently writing the state
 file is very time consuming.
 <li><b>$InputFileReadMode</b> [mode]</b><br>
 Available in 5.7.5+
+<li><b>$InputFileMaxLinesAtOnce</b> [number]</b><br>
+Available in 5.9.0+
 <br>
-Mode to be used when reading lines. 0 (the default) means that each line is forwarded
-as its own log message.
+This is useful if multiple files need to be monitored. If set to 0, each file
+will be fully processed and then processing switches to the next file
+(this was the default in previous versions). If it is set, a maximum of
+[number] lines is processed in sequence for each file, and then the file is
+switched. This provides a kind of mutiplexing the load of multiple files and
+probably leads to a more natural distribution of events when multiple busy files
+are monitored. The default is 10240.
 <li>$InputFileBindRuleset &lt;ruleset&gt;<br>
 Available in 5.7.5+, 6.1.5+
 Binds the listener to a specific <a href="multi_ruleset.html">ruleset</a>.</li>
diff --git a/doc/manual.html b/doc/manual.html
index e1f0ebf9..5c656752 100644
--- a/doc/manual.html
+++ b/doc/manual.html
@@ -19,7 +19,7 @@ rsyslog support</a> available directly from the source!</p>
 <p><b>Please visit the <a href="http://www.rsyslog.com/sponsors">rsyslog sponsor's page</a>
 to honor the project sponsors or become one yourself!</b> We are very grateful for any help towards the
 project goals.</p>
-<p><b>This documentation is for version 5.8.0 (stable branch) of rsyslog.</b>
+<p><b>This documentation is for version 5.8.1 (stable branch) of rsyslog.</b>
 Visit the <i><a href="http://www.rsyslog.com/status">rsyslog status page</a></i></b>
 to obtain current version information and project status.
 </p><p><b>If you like rsyslog, you might
diff --git a/plugins/imfile/imfile.c b/plugins/imfile/imfile.c
index cac3a55d..9bc84220 100644
--- a/plugins/imfile/imfile.c
+++ b/plugins/imfile/imfile.c
@@ -64,6 +64,7 @@ DEFobjCurrIf(strm)
 DEFobjCurrIf(prop)
 DEFobjCurrIf(ruleset)
 
+#define NUM_MULTISUB 1024 /* max number of submits -- TODO: make configurable */
 typedef struct fileInfo_s {
 	uchar *pszFileName;
 	uchar *pszTag;
@@ -71,11 +72,13 @@ typedef struct fileInfo_s {
 	uchar *pszStateFile; /* file in which state between runs is to be stored */
 	int iFacility;
 	int iSeverity;
+	int maxLinesAtOnce;
 	int nRecords; /**< How many records did we process before persisting the stream? */
 	int iPersistStateInterval; /**< how often should state be persisted? (0=on close only) */
 	strm_t *pStrm;	/* its stream (NULL if not assigned) */
 	int readMode;	/* which mode to use in ReadMulteLine call? */
 	ruleset_t *pRuleset;	/* ruleset to bind listener to (use system default if unspecified) */
+	multi_submit_t multiSub;
 } fileInfo_t;
 
 
@@ -91,6 +94,7 @@ static int iPersistStateInterval = 0;	/* how often if state file to be persisted
 static int iFacility = 128; /* local0 */
 static int iSeverity = 5;  /* notice, as of rfc 3164 */
 static int readMode = 0;  /* mode to use for ReadMultiLine call */
+static int maxLinesAtOnce = 10240;	/* how many lines to process in a row? */
 static ruleset_t *pBindRuleset = NULL;	/* ruleset to bind listener to (use system default if unspecified) */
 
 static int iFilPtr = 0;		/* number of files to be monitored; pointer to next free spot during config */
@@ -122,7 +126,9 @@ static rsRetVal enqLine(fileInfo_t *pInfo, cstr_t *cstrLine)
 	pMsg->iFacility = LOG_FAC(pInfo->iFacility);
 	pMsg->iSeverity = LOG_PRI(pInfo->iSeverity);
 	MsgSetRuleset(pMsg, pInfo->pRuleset);
-	CHKiRet(submitMsg(pMsg));
+	pInfo->multiSub.ppMsgs[pInfo->multiSub.nElem++] = pMsg;
+	if(pInfo->multiSub.nElem == pInfo->multiSub.maxElem)
+		CHKiRet(multiSubmitMsg(&pInfo->multiSub));
 finalize_it:
 	RETiRet;
 }
@@ -206,6 +212,7 @@ static void pollFileCancelCleanup(void *pArg)
 static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
 {
 	cstr_t *pCStr = NULL;
+	int nProcessed = 0;
 	DEFiRet;
 
 	ASSERT(pbHadFileData != NULL);
@@ -220,7 +227,10 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
 
 	/* loop below will be exited when strmReadLine() returns EOF */
 	while(glbl.GetGlobalInputTermState() == 0) {
+		if(pThis->maxLinesAtOnce != 0 && nProcessed >= pThis->maxLinesAtOnce)
+			break;
 		CHKiRet(strm.ReadLine(pThis->pStrm, &pCStr, pThis->readMode));
+		++nProcessed;
 		*pbHadFileData = 1; /* this is just a flag, so set it and forget it */
 		CHKiRet(enqLine(pThis, pCStr)); /* process line */
 		rsCStrDestruct(&pCStr); /* discard string (must be done by us!) */
@@ -231,6 +241,10 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
 	}
 
 finalize_it:
+	if(pThis->multiSub.nElem > 0) {
+		/* submit everything that was not yet submitted */
+		CHKiRet(multiSubmitMsg(&pThis->multiSub));
+	}
 		; /*EMPTY STATEMENT - needed to keep compiler happy - see below! */
 	/* Note: the problem above is that pthread:cleanup_pop() is a macro which
 	 * evaluates to something like "} while(0);". So the code would become
@@ -475,6 +489,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
 	iSeverity = 5;  /* notice, as of rfc 3164 */
 	readMode = 0;
 	pBindRuleset = NULL;
+	maxLinesAtOnce = 10240;
 
 	RETiRet;
 }
@@ -513,8 +528,12 @@ static rsRetVal addMonitor(void __attribute__((unused)) *pVal, uchar *pNewVal)
 			pThis->pszStateFile = (uchar*) strdup((char*) pszStateFile);
 		}
 
+		CHKmalloc(pThis->multiSub.ppMsgs = MALLOC(NUM_MULTISUB * sizeof(msg_t*)));
+		pThis->multiSub.maxElem = NUM_MULTISUB;
+		pThis->multiSub.nElem = 0;
 		pThis->iSeverity = iSeverity;
 		pThis->iFacility = iFacility;
+		pThis->maxLinesAtOnce = maxLinesAtOnce;
 		pThis->iPersistStateInterval = iPersistStateInterval;
 		pThis->nRecords = 0;
 		pThis->readMode = readMode;
@@ -592,6 +611,8 @@ CODEmodInit_QueryRegCFSLineHdlr
 	  	NULL, &iPollInterval, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilereadmode", 0, eCmdHdlrInt,
 	  	NULL, &readMode, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilemaxlinesatonce", 0, eCmdHdlrSize,
+	  	NULL, &maxLinesAtOnce, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilepersiststateinterval", 0, eCmdHdlrInt,
 	  	NULL, &iPersistStateInterval, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilebindruleset", 0, eCmdHdlrGetWord,
diff --git a/plugins/mmsnmptrapd/mmsnmptrapd.c b/plugins/mmsnmptrapd/mmsnmptrapd.c
index 9311d5c6..b78046ee 100644
--- a/plugins/mmsnmptrapd/mmsnmptrapd.c
+++ b/plugins/mmsnmptrapd/mmsnmptrapd.c
@@ -37,9 +37,6 @@
 #include <errno.h>
 #include <unistd.h>
 #include <ctype.h>
-#include <liblognorm.h>
-#include <libestr.h>
-#include <libee/libee.h>
 #include "conf.h"
 #include "msg.h"
 #include "syslogd-types.h"
diff --git a/plugins/omlibdbi/omlibdbi.c b/plugins/omlibdbi/omlibdbi.c
index 7fcf9631..e6f3fbd9 100644
--- a/plugins/omlibdbi/omlibdbi.c
+++ b/plugins/omlibdbi/omlibdbi.c
@@ -43,6 +43,7 @@
 #include "dirty.h"
 #include "syslogd-types.h"
 #include "cfsysline.h"
+#include "conf.h"
 #include "srUtils.h"
 #include "template.h"
 #include "module-template.h"
diff --git a/tests/inputfilegen.c b/tests/inputfilegen.c
index 26fb79af..0ff8d049 100644
--- a/tests/inputfilegen.c
+++ b/tests/inputfilegen.c
@@ -1,5 +1,6 @@
 /* generate an input file suitable for use by the testbench
  * Copyright (C) 2011 by Rainer Gerhards and Adiscon GmbH.
+ * usage: ./inputfilegen num-lines > file
  * Part of rsyslog, licensed under GPLv3
  */
 #include <stdio.h>