From: hadaq <hadaq>
Date: Wed, 28 Jul 2010 12:31:46 +0000 (+0000)
Subject: netmem 100% CPU usage solved. recvGeneric in assembleMsg() without malloc/free. Sergey.
X-Git-Url: https://jspc29.x-matter.uni-frankfurt.de/git/?a=commitdiff_plain;h=7a305bf8a4a8eb890fbab4f2936cb9bcd7c8a7f4;p=daqdata.git

netmem 100% CPU usage solved. recvGeneric in assembleMsg() without malloc/free. Sergey.
---

diff --git a/hadaq/nettrans.c b/hadaq/nettrans.c
index 1ac15c4..77d98fc 100644
--- a/hadaq/nettrans.c
+++ b/hadaq/nettrans.c
@@ -1,4 +1,4 @@
-static char *rcsId = "$Header: /misc/hadesprojects/daq/cvsroot/eventbuilder/hadaq/nettrans.c,v 6.41 2008-06-24 18:05:09 hadaq Exp $";
+static char *rcsId = "$Header: /misc/hadesprojects/daq/cvsroot/eventbuilder/hadaq/nettrans.c,v 6.42 2010-07-28 12:31:46 hadaq Exp $";
 
 
 #define _GNU_SOURCE
@@ -51,7 +51,7 @@ static int openUdp(NetTrans *my, unsigned long addr, int port, int fl)
 			syslog(LOG_WARNING, "setsockopt(..., SO_RCVBUF, ...): %s", strerror(errno));
 		}
 
-		if (getsockopt(my->fd, SOL_SOCKET, SO_RCVBUF, &rcvBufLenRet, (socklen_t *)&rcvBufLenLen) == -1) {
+		if (getsockopt(my->fd, SOL_SOCKET, SO_RCVBUF, &rcvBufLenRet, (socklen_t *) &rcvBufLenLen) == -1) {
 			syslog(LOG_WARNING, "getsockopt(..., SO_RCVBUF, ...): %s", strerror(errno));
 		}
 		if (rcvBufLenRet < rcvBufLenReq) {
@@ -124,6 +124,7 @@ int closeGeneric(NetTrans *my)
 	switch (my->type) {
 	case NetTransType_udp:
 		close(my->fd);
+		free(my->pkt_null);
 		retVal = 0;
 		break;
 	}
@@ -172,7 +173,7 @@ int recvGeneric(NetTrans *my)
 
 	switch (my->type) {
 	case NetTransType_udp:
-		retVal = recvfrom(my->fd, my->pkt, my->mtuSize, 0, &my->spec.sa, (socklen_t *)&socklen);
+		retVal = recvfrom(my->fd, my->pkt, my->mtuSize, 0, &my->spec.sa, (socklen_t *) &socklen);
 		break;
 	}
 
@@ -219,6 +220,9 @@ NetTrans *NetTrans_create(const char *name, size_t bandwidth, Worker *worker)
 
 	instance++;
 
+	/* Allocate memory for discarded packets when shmem buffers are full */
+	my->pkt_null = malloc(my->mtuSize);
+
 	return my;
 }
 
@@ -305,20 +309,32 @@ int assembleMsg(NetTrans *my, void *hadTu, size_t size)
 	/* BUGBUG quickhack for flushing the input data.  Properly in one of */
 	/* the parent functions this should be handled, but this would need */
 	/* better sunc. between nettrans & shmtrans */
+
 	if (hadTu == NULL) {
-/* 		my->pkt = malloc(my->mtuSize); */
-/* 		if (0 > recvGeneric(my)) { */
-/* 			abort(); */
-/* 		} */
+		/* 
+		 *  We come to this point if the buffer is full and there is 
+		 *  no place to store next hadTuQueue. We have either to discard
+		 *  the network packet or to sleep. Sleep is important otherwise
+		 *  assembleMsg() will be called again and again (leading to 100% CPU load)
+		 *  by netmem since there is data in the socket to be read out. 
+		 */
+
+		my->pkt = my->pkt_null;
+		if (0 > recvGeneric(my)) {
+			abort();
+		}
 		my->offset = 0;
 		(*my->pktsDiscarded)++;
-/* 		free(my->pkt); */
+
 		return 0;
 	}
+
 	my->pkt = (char *) hadTu + my->offset;
+
 	if (0 > recvGeneric(my)) {
 		abort();
 	}
+
 	my->offset += my->pktSize;
 	(*my->pktsReceived)++;
 
@@ -337,8 +353,10 @@ int assembleMsg(NetTrans *my, void *hadTu, size_t size)
 			(*my->bytesReceived) += my->offset;
 			retVal = 1;
 		}
+
 		my->offset = 0;
 	}
+
 	return retVal;
 }
 
@@ -350,10 +368,10 @@ void NetTrans_recv(NetTrans *my, void *hadTu, size_t size)
 
 unsigned long NetTrans_multiRecv(NetTrans *my[], void *hadTu[], size_t size[], int nrOfMsgs)
 {
-  /*
-   *  This original function has a limitation of maximum 32 subsystems. 
-   *  This limitation is due to a bit mask msgIsComplete of an unsigned long type. 
-   */
+	/*
+	 *  This original function has a limitation of maximum 32 subsystems. 
+	 *  This limitation is due to a bit mask msgIsComplete of an unsigned long type. 
+	 */
 	int i;
 	unsigned long msgIsComplete;
 	fd_set fdSetS, *fdSet = &fdSetS;
@@ -363,20 +381,20 @@ unsigned long NetTrans_multiRecv(NetTrans *my[], void *hadTu[], size_t size[], i
 	msgIsComplete = 0;
 
 	/* do { */
-		FD_ZERO(fdSet);
-		for (i = 0; i < nrOfMsgs; i++) {
-			FD_SET(my[i]->fd, fdSet);
-		}
-		if (0 > select(getdtablesize(), fdSet, NULL, NULL, NULL)) {
-			abort();
-		}
-		for (i = 0; i < nrOfMsgs; i++) {
-			if (FD_ISSET(my[i]->fd, fdSet)) {
-				if (assembleMsg(my[i], hadTu[i], size[i])) {
-					msgIsComplete |= 1 << i;
-				}
+	FD_ZERO(fdSet);
+	for (i = 0; i < nrOfMsgs; i++) {
+		FD_SET(my[i]->fd, fdSet);
+	}
+	if (0 > select(getdtablesize(), fdSet, NULL, NULL, NULL)) {
+		abort();
+	}
+	for (i = 0; i < nrOfMsgs; i++) {
+		if (FD_ISSET(my[i]->fd, fdSet)) {
+			if (assembleMsg(my[i], hadTu[i], size[i])) {
+				msgIsComplete |= 1 << i;
 			}
 		}
+	}
 	/* } while (0 && msgIsComplete == 0); */
 
 	return msgIsComplete;
@@ -384,23 +402,23 @@ unsigned long NetTrans_multiRecv(NetTrans *my[], void *hadTu[], size_t size[], i
 
 fd_set *NetTrans_multiReceive(NetTrans *my[], void *hadTu[], size_t size[], int nrOfMsgs)
 {
-  /*
-   *  This function does not have a limitation of a number of subsystems 
-   *  due to a bit mask msgIsComplete. Instead the function returns fdSet itself 
-   *  which should be later checked with FD_ISSET. S.Y. 
-   */
+	/*
+	 *  This function does not have a limitation of a number of subsystems 
+	 *  due to a bit mask msgIsComplete. Instead the function returns fdSet itself 
+	 *  which should be later checked with FD_ISSET. S.Y. 
+	 */
 	int i;
 	fd_set fdSetS, *fdSet = &fdSetS;
 
 	FD_ZERO(fdSet);
 
 	for (i = 0; i < nrOfMsgs; i++) {
-	  FD_SET(my[i]->fd, fdSet);
+		FD_SET(my[i]->fd, fdSet);
 	}
 
 	if (0 > select(getdtablesize(), fdSet, NULL, NULL, NULL)) {
-	  printf("nettranc::NetTrans_multiReceive: abort after select\n");
-	  abort();
+		printf("nettranc::NetTrans_multiReceive: abort after select\n");
+		abort();
 	}
 
 	return fdSet;