performance-optimized imudp

1 files changed, 44 insertions, 32 deletions

diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c
index d4c9d759..f4830a47 100644
--- a/plugins/imudp/imudp.c
+++ b/plugins/imudp/imudp.c
@@ -179,39 +179,51 @@ CODESTARTrunInput
 	       for (i = 0; nfds && i < *udpLstnSocks; i++) {
 		       if (FD_ISSET(udpLstnSocks[i+1], &readfds)) {
 			       socklen = sizeof(frominet);
-			       l = recvfrom(udpLstnSocks[i+1], (char*) pRcvBuf, iMaxLine, 0,
-					    (struct sockaddr *)&frominet, &socklen);
-			       if (l > 0) {
-				       if(net.cvthname(&frominet, fromHost, fromHostFQDN, fromHostIP) == RS_RET_OK) {
-					       dbgprintf("Message from inetd socket: #%d, host: %s\n",
-						       udpLstnSocks[i+1], fromHost);
-					       /* Here we check if a host is permitted to send us
-						* syslog messages. If it isn't, we do not further
-						* process the message but log a warning (if we are
-						* configured to do this).
-						* rgerhards, 2005-09-26
-						*/
-					       if(net.isAllowedSender(net.pAllowedSenders_UDP,
-						  (struct sockaddr *)&frominet, (char*)fromHostFQDN)) {
-						       parseAndSubmitMessage(fromHost, fromHostIP, pRcvBuf, l,
-						       MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_NO_DELAY, (uchar*)"imudp");
-					       } else {
-						       dbgprintf("%s is not an allowed sender\n", (char*)fromHostFQDN);
-						       if(glbl.GetOption_DisallowWarning) {
-							       errmsg.LogError(0, NO_ERRCODE, "UDP message from disallowed sender %s discarded",
-									  (char*)fromHost);
-						       }	
+			        do {
+					/* we now try to read from the file descriptor until there
+					 * is no more data. This is done in the hope to get better performance
+					 * out of the system. However, this also means that a descriptor
+					 * monopolizes processing while it contains data. This can lead to
+					 * data loss in other descriptors. However, if the system is incapable of
+					 * handling the workload, we will loss data in any case. So it doesn't really
+					 * matter where the actual loss occurs - it is always random, because we depend
+					 * on scheduling order. -- rgerhards, 2008-10-02
+					 */
+					l = recvfrom(udpLstnSocks[i+1], (char*) pRcvBuf, iMaxLine, 0,
+						    (struct sockaddr *)&frominet, &socklen);
+					if(l > 0) {
+					       if(net.cvthname(&frominet, fromHost, fromHostFQDN, fromHostIP) == RS_RET_OK) {
+						       dbgprintf("Message from inetd socket: #%d, host: %s\n",
+							       udpLstnSocks[i+1], fromHost);
+						       /* Here we check if a host is permitted to send us
+							* syslog messages. If it isn't, we do not further
+							* process the message but log a warning (if we are
+							* configured to do this).
+							* rgerhards, 2005-09-26
+							*/
+						       if(net.isAllowedSender(net.pAllowedSenders_UDP,
+							  (struct sockaddr *)&frominet, (char*)fromHostFQDN)) {
+							       parseAndSubmitMessage(fromHost, fromHostIP, pRcvBuf, l,
+							       MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_NO_DELAY, (uchar*)"imudp");
+						       } else {
+							       dbgprintf("%s is not an allowed sender\n", (char*)fromHostFQDN);
+							       if(glbl.GetOption_DisallowWarning) {
+								       errmsg.LogError(0, NO_ERRCODE, "UDP message from disallowed sender %s discarded",
+										  (char*)fromHost);
+							       }	
+						       }
 					       }
-				       }
-			       } else if (l < 0 && errno != EINTR && errno != EAGAIN) {
-					char errStr[1024];
-					rs_strerror_r(errno, errStr, sizeof(errStr));
-					dbgprintf("INET socket error: %d = %s.\n", errno, errStr);
-					       errmsg.LogError(errno, NO_ERRCODE, "recvfrom inet");
-					       /* should be harmless */
-					       sleep(1);
-				       }
-				--nfds; /* indicate we have processed one */
+					} else if(l < 0 && errno != EINTR && errno != EAGAIN) {
+						char errStr[1024];
+						rs_strerror_r(errno, errStr, sizeof(errStr));
+						dbgprintf("INET socket error: %d = %s.\n", errno, errStr);
+						       errmsg.LogError(errno, NO_ERRCODE, "recvfrom inet");
+						       /* should be harmless */
+						       sleep(1);
+					}
+				} while(l > 0); /* Warning: do ... while()! */
+
+				--nfds; /* indicate we have processed one descriptor */
 			}
 	       }
 	       /* end of a run, back to loop for next recv() */