Merge branch 'c8'

Conflicts:
	ChangeLog
	doc/imudp.html
	plugins/imudp/imudp.c

7 files changed, 65 insertions, 9 deletions

diff --git a/ChangeLog b/ChangeLog
index 04eea243..d38fdef9 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,5 +1,6 @@
 ---------------------------------------------------------------------------
 Version 7.5.3 [devel] 2013-07-??
+- imudp: add ability to specify SO_RCVBUF size (rcvbufSize parameter)
 - impstats: add process resource usage counters [via getrusage()]
 - librelp 1.2.0 is now required
 - make use of new librelp generic error reporting facility
diff --git a/doc/imudp.html b/doc/imudp.html
index a8dbca31..043d774c 100644
--- a/doc/imudp.html
+++ b/doc/imudp.html
@@ -75,6 +75,17 @@ are defined for a single input and each of the inputnames shall be unique.
 Note that there currently is no differentiation between IPv4/v6 listeners on
 the same port.
 </li>
+<li><b>rcvbufSize</b> [size] - (available since 7.5.3)
+This request a socket receive buffer of specific size from the operating system.
+It is an expert parameter, which should only be changed for a good reason. Note that
+setting this parameter disables Linux auto-tuning, which usually works pretty well.
+The default value is 0, which means "keep the OS buffer size unchanged". This is a size
+value. So in addition to pure integer values, sizes like "256k", "1m" and the like can
+be specified. Note that setting very large sizes may require root or other special
+privileges. Also note that the OS may slightly adjust the value or shrink it to a
+system-set max value if the user is not sufficiently privileged. Technically, this
+parameter will result in a setsockopt() call with SO_RCVBUF (and SO_RCVBUFFORCE if it
+is available).
 </ul>
 <b>Caveats/Known Bugs:</b>
 <ul>
@@ -91,6 +102,13 @@ user account.
 input(type="imudp" port="514")
 </textarea>
 
+<p>The following sample is mostly equivalent to the first one, but request a
+larger rcvuf size. Note that 1m most probably will not be honored by the OS
+until the user is sufficiently privileged.</p>
+<textarea rows="3" cols="60">module(load="imudp") # needs to be done just once
+input(type="imudp" port="514" rcvbufSize="1m")
+</textarea>
+
 <p>In the next example, we set up three listeners at ports 10514, 10515 and 10516
 and assign a listner name of "udp" to it, followed by the port number:
 </p>
@@ -110,7 +128,6 @@ input(type="imudp" port=["10514","10515","10516"]
       inputname="" inputname.appendPort="on")
 </textarea>
 
-
 <p><b>Legacy Configuration Directives</b>:</p>
 <p>Multiple receivers may be configured by specifying
 $UDPServerRun multiple times.
@@ -129,7 +146,7 @@ equivalent to: SchedulingPolicy
 <li>$IMUDPSchedulingPriority &lt;number&gt; Available since 4.7.4+, 5.7.3+, 6.1.3+.<br>
 equivalent to: SchedulingPriority 
 </ul>
-<p><b>Sample:</b></p>
+<p><b>Legacy Sample:</b></p>
 <p>This sets up an UPD server on port 514:<br>
 </p>
 <textarea rows="3" cols="60">$ModLoad imudp # needs to be done just once
diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c
index 312645bd..386ef219 100644
--- a/plugins/imudp/imudp.c
+++ b/plugins/imudp/imudp.c
@@ -112,6 +112,7 @@ struct instanceConf_s {
 	ruleset_t *pBindRuleset;	/* ruleset to bind listener to (use system default if unspecified) */
 	int ratelimitInterval;
 	int ratelimitBurst;
+	int rcvbuf;			/* 0 means: do not set, keep OS default */
 	struct instanceConf_s *next;
 	sbool bAppendPortToInpname;
 };
@@ -146,9 +147,10 @@ static struct cnfparamdescr inppdescr[] = {
 	{ "inputname", eCmdHdlrGetWord, 0 },
 	{ "inputname.appendport", eCmdHdlrBinary, 0 },
 	{ "address", eCmdHdlrString, 0 },
-	{ "ruleset", eCmdHdlrString, 0 },
 	{ "ratelimit.interval", eCmdHdlrInt, 0 },
-	{ "ratelimit.burst", eCmdHdlrInt, 0 }
+	{ "ratelimit.burst", eCmdHdlrInt, 0 },
+	{ "rcvbufsize", eCmdHdlrSize, 0 },
+	{ "ruleset", eCmdHdlrString, 0 }
 };
 static struct cnfparamblk inppblk =
 	{ CNFPARAMBLK_VERSION,
@@ -177,6 +179,7 @@ createInstance(instanceConf_t **pinst)
 	inst->bAppendPortToInpname = 0;
 	inst->ratelimitBurst = 10000; /* arbitrary high limit */
 	inst->ratelimitInterval = 0; /* off */
+	inst->rcvbuf = 0;
 
 	/* node created, let's add to config */
 	if(loadModConf->tail == NULL) {
@@ -252,7 +255,7 @@ addListner(instanceConf_t *inst)
 
 	DBGPRINTF("Trying to open syslog UDP ports at %s:%s.\n", bindName, inst->pszBindPort);
 
-	newSocks = net.create_udp_socket(bindAddr, port, 1);
+	newSocks = net.create_udp_socket(bindAddr, port, 1, inst->rcvbuf);
 	if(newSocks != NULL) {
 		/* we now need to add the new sockets to the existing set */
 		/* ready to copy */
@@ -729,6 +732,8 @@ createListner(es_str_t *port, struct cnfparamvals *pvals)
 			inst->ratelimitBurst = (int) pvals[i].val.d.n;
 		} else if(!strcmp(inppblk.descr[i].name, "ratelimit.interval")) {
 			inst->ratelimitInterval = (int) pvals[i].val.d.n;
+		} else if(!strcmp(inppblk.descr[i].name, "rcvbufsize")) {
+			inst->rcvbuf = (int) pvals[i].val.d.n;
 		} else {
 			dbgprintf("imudp: program error, non-handled "
 			  "param '%s'\n", inppblk.descr[i].name);
diff --git a/plugins/omudpspoof/omudpspoof.c b/plugins/omudpspoof/omudpspoof.c
index c80f0e57..058c4252 100644
--- a/plugins/omudpspoof/omudpspoof.c
+++ b/plugins/omudpspoof/omudpspoof.c
@@ -563,7 +563,7 @@ static rsRetVal doTryResume(instanceData *pData)
 	}
 	DBGPRINTF("%s found, resuming.\n", pData->host);
 	pData->f_addr = res;
-	pData->pSockArray = net.create_udp_socket((uchar*)pData->host, NULL, 0);
+	pData->pSockArray = net.create_udp_socket((uchar*)pData->host, NULL, 0, 0);
 
 finalize_it:
 	if(iRet != RS_RET_OK) {
diff --git a/runtime/net.c b/runtime/net.c
index 13391cc0..7c180b19 100644
--- a/runtime/net.c
+++ b/runtime/net.c
@@ -1189,12 +1189,16 @@ void closeUDPListenSockets(int *pSockArr)
  * hostname and/or pszPort may be NULL, but not both!
  * bIsServer indicates if a server socket should be created
  * 1 - server, 0 - client
+ * param rcvbuf indicates desired rcvbuf size; 0 means OS default
  */
-int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer)
+int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer, int rcvbuf)
 {
         struct addrinfo hints, *res, *r;
         int error, maxs, *s, *socks, on = 1;
 	int sockflags;
+	int actrcvbuf;
+	socklen_t optlen;
+	char errStr[1024];
 
 	assert(!((pszPort == NULL) && (hostname == NULL)));
         memset(&hints, 0, sizeof(hints));
@@ -1297,6 +1301,35 @@ int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer)
 			continue;
 		}
 
+		if(rcvbuf != 0) {
+#			if defined(SO_RCVBUFFORCE)
+			if(setsockopt(*s, SOL_SOCKET, SO_RCVBUFFORCE, &rcvbuf, sizeof(rcvbuf)) < 0)
+#			endif
+			{
+				/* if we fail, try to do it the regular way. Experiments show that at 
+				 * least some platforms do not return an error here, but silently set
+				 * it to the max permitted value. So we do our error check a bit
+				 * differently by querying the size below.
+				 */
+				setsockopt(*s, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf));
+			}
+		}
+
+		if(Debug || rcvbuf != 0) {
+			optlen = sizeof(actrcvbuf);
+			if(getsockopt(*s, SOL_SOCKET, SO_RCVBUF, &actrcvbuf, &optlen) == 0) {
+				dbgprintf("socket %d, actual rcvbuf size %d\n", *s, actrcvbuf);
+				if(rcvbuf != 0 && actrcvbuf/2 != rcvbuf) {
+					errmsg.LogError(errno, NO_ERRCODE,
+						"cannot set rcvbuf size %d for socket %d, value now is %d",
+						rcvbuf, *s, actrcvbuf/2);
+				}
+			} else {
+				dbgprintf("could not obtain rcvbuf size for socket %d: %s\n",
+					*s, rs_strerror_r(errno, errStr, sizeof(errStr)));
+			}
+		}
+
 		if(bIsServer) {
 			/* rgerhards, 2007-06-22: if we run on a kernel that does not support
 			 * the IPV6_V6ONLY socket option, we need to use a work-around. On such
diff --git a/runtime/net.h b/runtime/net.h
index b196116b..d7a7b512 100644
--- a/runtime/net.h
+++ b/runtime/net.h
@@ -137,7 +137,7 @@ BEGINinterface(net) /* name must also be changed in ENDinterface macro! */
 	void (*PrintAllowedSenders)(int iListToPrint);
 	void (*clearAllowedSenders)(uchar*);
 	void (*debugListenInfo)(int fd, char *type);
-	int *(*create_udp_socket)(uchar *hostname, uchar *LogPort, int bIsServer);
+	int *(*create_udp_socket)(uchar *hostname, uchar *LogPort, int bIsServer, int rcvbuf);
 	void (*closeUDPListenSockets)(int *finet);
 	int (*isAllowedSender)(uchar *pszType, struct sockaddr *pFrom, const char *pszFromHost); /* deprecated! */
 	rsRetVal (*getLocalHostname)(uchar**);
diff --git a/tools/omfwd.c b/tools/omfwd.c
index 179e9614..a75f04c4 100644
--- a/tools/omfwd.c
+++ b/tools/omfwd.c
@@ -675,7 +675,7 @@ static rsRetVal doTryResume(instanceData *pData)
 		pData->f_addr = res;
 		pData->bIsConnected = 1;
 		if(pData->pSockArray == NULL) {
-			pData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0);
+			pData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0, 0);
 		}
 	} else {
 		CHKiRet(TCPSendInit((void*)pData));