Merge branch 'v4-devel' into beta

6 files changed, 61 insertions, 22 deletions

diff --git a/ChangeLog b/ChangeLog
index 10d3da0e..b8e884e3 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -3,10 +3,14 @@ Version 4.5.2  [DEVEL] (rgerhards), 2009-07-??
 - legacy syslog parser changed so that it now accepts date stamps in
   wrong case. Some devices seem to create them and I do not see any harm
   in supporting that.
+- added $InputTCPMaxListeners directive - permits to specify how many 
+  TCP servers shall be possible (default is 20).
 - bugfix: memory leak with some input modules. Those inputs that
   use parseAndSubmitMsg() leak two small memory blocks with every message.
   Typically, those process only relatively few messages, so the issue 
   does most probably not have any effect in practice.
+- bugfix: if tcp listen port could not be created, no error message was
+  emitted
 ---------------------------------------------------------------------------
 Version 4.5.1  [DEVEL] (rgerhards), 2009-07-15
 - CONFIG CHANGE: $HUPisRestart default is now "off". We are doing this
diff --git a/doc/imtcp.html b/doc/imtcp.html
index 9ea7efa1..5217634e 100644
--- a/doc/imtcp.html
+++ b/doc/imtcp.html
@@ -43,15 +43,19 @@ can be found at the <a href="http://www.rsyslog.com/Article321.phtml">Cisco tcp
 page.
 <li>$InputTCPServerRun &lt;port&gt;<br>
 Starts a TCP server on selected port</li>
-<li><ul><li>$InputTCPMaxSessions &lt;number&gt;</li></ul>
-Sets the maximum number of sessions supported</li><li>$InputTCPServerStreamDriverMode &lt;number&gt;<br>
+<li>$InputTCPMaxListeners &lt;number&gt;<br>
+Sets the maximum number of listeners (server ports) supported. Default is 20. This must be set before the first $InputTCPServerRun directive.</li>
+<li>$InputTCPMaxSessions &lt;number&gt;<br>
+Sets the maximum number of sessions supported. Default is 200. This must be set before the first $InputTCPServerRun directive</li>
+<li>$InputTCPServerStreamDriverMode &lt;number&gt;<br>
 Sets the driver mode for the currently selected <a href="netstream.html">network stream driver</a>. &lt;number&gt; is driver specifc.</li>
 <li>$InputTCPServerInputName &lt;name&gt;<br>
 Sets a name for the inputname property. If no name is set "imtcp" is used by default. Setting a 
 name is not strictly necessary, but can be useful to apply filtering based on which input
 the message was received from.
 <li>$InputTCPServerStreamDriverAuthMode &lt;mode-string&gt;<br>
-Sets the authentication mode for the currently selected <a href="netstream.html">network stream driver</a>. &lt;mode-string&gt; is driver specifc.</li><li>$InputTCPServerStreamDriverPermittedPeer &lt;id-string&gt;<br>
+Sets the authentication mode for the currently selected <a href="netstream.html">network stream driver</a>. &lt;mode-string&gt; is driver specifc.</li>
+<li>$InputTCPServerStreamDriverPermittedPeer &lt;id-string&gt;<br>
 Sets permitted peer IDs. Only these peers are able to connect to the
 listener. &lt;id-string&gt; semantics depend on the currently selected
 AuthMode and&nbsp; <a href="netstream.html">network stream driver</a>. PermittedPeers may not be set in anonymous modes.</li>
diff --git a/plugins/imtcp/imtcp.c b/plugins/imtcp/imtcp.c
index e1f513c8..01c4bc33 100644
--- a/plugins/imtcp/imtcp.c
+++ b/plugins/imtcp/imtcp.c
@@ -82,6 +82,7 @@ static permittedPeers_t *pPermPeersRoot = NULL;
 
 /* config settings */
 static int iTCPSessMax = 200; /* max number of sessions */
+static int iTCPLstnMax = 20; /* max number of sessions */
 static int iStrmDrvrMode = 0; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
 static int iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER; /* addtl frame delimiter, e.g. for netscreen, default none */
 static uchar *pszStrmDrvrAuthMode = NULL; /* authentication mode to use */
@@ -188,6 +189,7 @@ static rsRetVal addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVa
 	if(pOurTcpsrv == NULL) {
 		CHKiRet(tcpsrv.Construct(&pOurTcpsrv));
 		CHKiRet(tcpsrv.SetSessMax(pOurTcpsrv, iTCPSessMax));
+		CHKiRet(tcpsrv.SetLstnMax(pOurTcpsrv, iTCPLstnMax));
 		CHKiRet(tcpsrv.SetCBIsPermittedHost(pOurTcpsrv, isPermittedHost));
 		CHKiRet(tcpsrv.SetCBRcvData(pOurTcpsrv, doRcvData));
 		CHKiRet(tcpsrv.SetCBOpenLstnSocks(pOurTcpsrv, doOpenLstnSocks));
@@ -273,6 +275,7 @@ static rsRetVal
 resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
 {
 	iTCPSessMax = 200;
+	iTCPLstnMax = 20;
 	iStrmDrvrMode = 0;
 	iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
 	free(pszInputName);
@@ -308,6 +311,8 @@ CODEmodInit_QueryRegCFSLineHdlr
 				   addTCPListener, NULL, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpmaxsessions"), 0, eCmdHdlrInt,
 				   NULL, &iTCPSessMax, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpmaxlisteners"), 0, eCmdHdlrInt,
+				   NULL, &iTCPLstnMax, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpserverstreamdrivermode"), 0,
 				   eCmdHdlrInt, NULL, &iStrmDrvrMode, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpserverstreamdriverauthmode"), 0,
diff --git a/runtime/stream.c b/runtime/stream.c
index 605a9771..1b9beb5b 100644
--- a/runtime/stream.c
+++ b/runtime/stream.c
@@ -628,7 +628,8 @@ static rsRetVal strmConstructFinalize(strm_t *pThis)
 		pthread_cond_init(&pThis->notEmpty, 0);
 		pthread_cond_init(&pThis->isEmpty, 0);
 		pThis->iCnt = pThis->iEnq = pThis->iDeq = 0;
-		for(i = 0 ; i < STREAM_ASYNC_NUMBUFS ; ++i) {
+		//for(i = 0 ; i < STREAM_ASYNC_NUMBUFS ; ++i) {
+		for(i = 0 ; i < 1 ; ++i) { // HOTFIX!!!
 			CHKmalloc(pThis->asyncBuf[i].pBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize));
 		}
 		pThis->pIOBuf = pThis->asyncBuf[0].pBuf;
@@ -844,7 +845,10 @@ dbgprintf("XXX: doAsyncWriteInternal: strm %p, len %ld\n", pThis, (long) lenBuf)
 		d_pthread_cond_wait(&pThis->notFull, &pThis->mut);
 
 	pThis->asyncBuf[pThis->iEnq % STREAM_ASYNC_NUMBUFS].lenBuf = lenBuf;
-	pThis->pIOBuf = pThis->asyncBuf[++pThis->iEnq % STREAM_ASYNC_NUMBUFS].pBuf;
+	pThis->asyncBuf[pThis->iEnq % STREAM_ASYNC_NUMBUFS].pBuf = pThis->pIOBuf;
+	//pThis->pIOBuf = pThis->asyncBuf[++pThis->iEnq % STREAM_ASYNC_NUMBUFS].pBuf;
+	++pThis->iEnq;
+	CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize));
 
 	pThis->bDoTimedWait = 0; /* everything written, no need to timeout partial buffer writes */
 	if(++pThis->iCnt == 1)
@@ -938,6 +942,8 @@ asyncWriterThread(void *pPtr)
 		iDeq = pThis->iDeq++ % STREAM_ASYNC_NUMBUFS;
 		doWriteInternal(pThis, pThis->asyncBuf[iDeq].pBuf, pThis->asyncBuf[iDeq].lenBuf);
 		// TODO: error check????? 2009-07-06
+		free(pThis->asyncBuf[iDeq].pBuf);
+		pThis->asyncBuf[iDeq].pBuf = NULL;
 
 		--pThis->iCnt;
 		if(pThis->iCnt < STREAM_ASYNC_NUMBUFS) {
diff --git a/tcpsrv.c b/tcpsrv.c
index e8ea2b98..0be723d1 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -261,7 +261,7 @@ static void deinit_tcp_listener(tcpsrv_t *pThis)
 	}
 
 	/* finally close our listen streams */
-	for(i = 0 ; i < pThis->iLstnMax ; ++i) {
+	for(i = 0 ; i < pThis->iLstnCurr ; ++i) {
 		netstrm.Destruct(pThis->ppLstn + i);
 	}
 }
@@ -280,12 +280,12 @@ addTcpLstn(void *pUsr, netstrm_t *pLstn)
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 	ISOBJ_TYPE_assert(pLstn, netstrm);
 
-	if(pThis->iLstnMax >= TCPLSTN_MAX_DEFAULT)
+	if(pThis->iLstnCurr >= pThis->iLstnMax)
 		ABORT_FINALIZE(RS_RET_MAX_LSTN_REACHED);
 
-	pThis->ppLstn[pThis->iLstnMax] = pLstn;
-	pThis->ppLstnPort[pThis->iLstnMax] = pPortList;
-	++pThis->iLstnMax;
+	pThis->ppLstn[pThis->iLstnCurr] = pLstn;
+	pThis->ppLstnPort[pThis->iLstnCurr] = pPortList;
+	++pThis->iLstnCurr;
 
 finalize_it:
 	RETiRet;
@@ -327,15 +327,19 @@ finalize_it:
 static rsRetVal
 create_tcp_socket(tcpsrv_t *pThis)
 {
-	tcpLstnPortList_t *pEntry;
 	DEFiRet;
+	rsRetVal localRet;
+	tcpLstnPortList_t *pEntry;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
 	/* init all configured ports */
 	pEntry = pThis->pLstnPorts;
 	while(pEntry != NULL) {
-		CHKiRet(initTCPListener(pThis, pEntry));
+		localRet = initTCPListener(pThis, pEntry);
+		if(localRet != RS_RET_OK) {
+			errmsg.LogError(0, localRet, "Could not create tcp listener, ignoring port %s.", pEntry->pszPort);
+		}
 		pEntry = pEntry->pNext;
 	}
 
@@ -481,7 +485,6 @@ Run(tcpsrv_t *pThis)
 	 * this thread. Thus, we also need to instantiate a cancel cleanup handler
 	 * to prevent us from leaking anything. -- rgerharsd, 20080-04-24
 	 */
-RUNLOG_STR("XXXX: tcp server runs\n");
 	pthread_cleanup_push(RunCancelCleanup, (void*) &pSel);
 	while(1) {
 		CHKiRet(nssel.Construct(&pSel));
@@ -489,7 +492,7 @@ RUNLOG_STR("XXXX: tcp server runs\n");
 		CHKiRet(nssel.ConstructFinalize(pSel));
 
 		/* Add the TCP listen sockets to the list of read descriptors. */
-		for(i = 0 ; i < pThis->iLstnMax ; ++i) {
+		for(i = 0 ; i < pThis->iLstnCurr ; ++i) {
 			CHKiRet(nssel.Add(pSel, pThis->ppLstn[i], NSDSEL_RD));
 		}
 
@@ -502,11 +505,10 @@ RUNLOG_STR("XXXX: tcp server runs\n");
 			iTCPSess = TCPSessGetNxtSess(pThis, iTCPSess);
 		}
 
-RUNLOG_STR("XXXX: tcp server select\n");
 		/* wait for io to become ready */
 		CHKiRet(nssel.Wait(pSel, &nfds));
 
-		for(i = 0 ; i < pThis->iLstnMax ; ++i) {
+		for(i = 0 ; i < pThis->iLstnCurr ; ++i) {
 			CHKiRet(nssel.IsReady(pSel, pThis->ppLstn[i], NSDSEL_RD, &bIsReady, &nfds));
 			if(bIsReady) {
 				dbgprintf("New connect on NSD %p.\n", pThis->ppLstn[i]);
@@ -514,7 +516,6 @@ RUNLOG_STR("XXXX: tcp server select\n");
 				--nfds; /* indicate we have processed one */
 			}
 		}
-RUNLOG_STR("XXXX: tcp server post select\n");
 
 		/* now check the sessions */
 		iTCPSess = TCPSessGetNxtSess(pThis, -1);
@@ -578,7 +579,8 @@ finalize_it: /* this is a very special case - this time only we do not exit the
 
 /* Standard-Constructor */
 BEGINobjConstruct(tcpsrv) /* be sure to specify the object type also in END macro! */
-	pThis->iSessMax = TCPSESS_MAX_DEFAULT; /* TODO: useful default ;) */
+	pThis->iSessMax = TCPSESS_MAX_DEFAULT;
+	pThis->iLstnMax = TCPLSTN_MAX_DEFAULT;
 	pThis->addtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
 	pThis->OnMsgReceive = NULL;
 ENDobjConstruct(tcpsrv)
@@ -602,8 +604,8 @@ tcpsrvConstructFinalize(tcpsrv_t *pThis)
 	CHKiRet(netstrms.ConstructFinalize(pThis->pNS));
 
 	/* set up listeners */
-	CHKmalloc(pThis->ppLstn = calloc(TCPLSTN_MAX_DEFAULT, sizeof(netstrm_t*)));
-	CHKmalloc(pThis->ppLstnPort = calloc(TCPLSTN_MAX_DEFAULT, sizeof(tcpLstnPortList_t*)));
+	CHKmalloc(pThis->ppLstn = calloc(pThis->iLstnMax, sizeof(netstrm_t*)));
+	CHKmalloc(pThis->ppLstnPort = calloc(pThis->iLstnMax, sizeof(tcpLstnPortList_t*)));
 	iRet = pThis->OpenLstnSocks(pThis);
 
 finalize_it:
@@ -820,6 +822,20 @@ SetDrvrPermPeers(tcpsrv_t *pThis, permittedPeers_t *pPermPeers)
  * -------------------------------------------------------------------------- */
 
 
+/* set max number of listeners
+ * this must be called before ConstructFinalize, or it will have no effect!
+ * rgerhards, 2009-08-17
+ */
+static rsRetVal
+SetLstnMax(tcpsrv_t *pThis, int iMax)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, tcpsrv);
+	pThis->iLstnMax = iMax;
+	RETiRet;
+}
+
+
 /* set max number of sessions
  * this must be called before ConstructFinalize, or it will have no effect!
  * rgerhards, 2009-04-09
@@ -862,6 +878,7 @@ CODESTARTobjQueryInterface(tcpsrv)
 	pIf->SetInputName = SetInputName;
 	pIf->SetAddtlFrameDelim = SetAddtlFrameDelim;
 	pIf->SetSessMax = SetSessMax;
+	pIf->SetLstnMax = SetLstnMax;
 	pIf->SetDrvrMode = SetDrvrMode;
 	pIf->SetDrvrAuthMode = SetDrvrAuthMode;
 	pIf->SetDrvrPermPeers = SetDrvrPermPeers;
diff --git a/tcpsrv.h b/tcpsrv.h
index 70682398..64065aab 100644
--- a/tcpsrv.h
+++ b/tcpsrv.h
@@ -54,9 +54,10 @@ struct tcpsrv_s {
 	uchar *pszInputName;	/**< value to be used as input name */
 	ruleset_t *pRuleset;	/**< ruleset to bind to */
 	permittedPeers_t *pPermPeers;/**< driver's permitted peers */
-	int iLstnMax;		/**< max nbr of listeners currently supported */
+	int iLstnCurr;		/**< max nbr of listeners currently supported */
 	netstrm_t **ppLstn;	/**< our netstream listners */
 	tcpLstnPortList_t **ppLstnPort; /**< pointer to relevant listen port description */
+	int iLstnMax;		/**< max number of listners supported */
 	int iSessMax;		/**< max number of sessions supported */
 	tcpLstnPortList_t *pLstnPorts;	/**< head pointer for listen ports */
 	int addtlFrameDelim;	/**< additional frame delimiter for plain TCP syslog framing (e.g. to handle NetScreen) */
@@ -111,8 +112,10 @@ BEGINinterface(tcpsrv) /* name must also be changed in ENDinterface macro! */
 	/* added v6 */
 	rsRetVal (*SetOnMsgReceive)(tcpsrv_t *pThis, rsRetVal (*OnMsgReceive)(tcps_sess_t*, uchar*, int)); /* 2009-05-24 */
 	rsRetVal (*SetRuleset)(tcpsrv_t *pThis, ruleset_t*); /* 2009-06-12 */
+	/* added v7 */
+	rsRetVal (*SetLstnMax)(tcpsrv_t *pThis, int iMaxLstn);	/* 2009-08-17 */
 ENDinterface(tcpsrv)
-#define tcpsrvCURR_IF_VERSION 6 /* increment whenever you change the interface structure! */
+#define tcpsrvCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */
 /* change for v4:
  * - SetAddtlFrameDelim() added -- rgerhards, 2008-12-10
  * - SetInputName() added -- rgerhards, 2008-12-10