imfile: added $InputFileMaxLinesAtOnce directive

4 files changed, 21 insertions, 3 deletions

diff --git a/ChangeLog b/ChangeLog
index 0d35ce0f..d139f5b3 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,4 @@
+- imfile: added $InputFileMaxLinesAtOnce directive
 - enhanced imfile to support input batching
 ---------------------------------------------------------------------------
 Version 5.8.1  [V5-stable] (rgerhards), 2011-05-19
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/plugins/imfile/imfile.c b/plugins/imfile/imfile.c
index 415eb137..9bc84220 100644
--- a/plugins/imfile/imfile.c
+++ b/plugins/imfile/imfile.c
@@ -72,6 +72,7 @@ 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) */
@@ -93,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 */
@@ -210,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);
@@ -224,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!) */
@@ -233,7 +239,6 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
 			pThis->nRecords = 0;
 		}
 	}
-	/* NOTE: This is usually not reached due to loop exit via CHKiRet() only! */
 
 finalize_it:
 	if(pThis->multiSub.nElem > 0) {
@@ -484,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;
 }
@@ -527,6 +533,7 @@ static rsRetVal addMonitor(void __attribute__((unused)) *pVal, uchar *pNewVal)
 		pThis->multiSub.nElem = 0;
 		pThis->iSeverity = iSeverity;
 		pThis->iFacility = iFacility;
+		pThis->maxLinesAtOnce = maxLinesAtOnce;
 		pThis->iPersistStateInterval = iPersistStateInterval;
 		pThis->nRecords = 0;
 		pThis->readMode = readMode;
@@ -604,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/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>