improved tcpsrv performance by enabling multiple-entry epoll

so far, we always pulled a single event from the epoll interface.
Now 128, what should result in performance improvement (less API
calls) on busy systems. Most importantly affects imtcp.

6 files changed, 49 insertions, 28 deletions

diff --git a/ChangeLog b/ChangeLog
index d095097d..8dae0f8f 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -2,6 +2,10 @@
 Version 6.1.3  [DEVEL] (rgerhards), 2010-12-??
 - added $IMUDPSchedulingPolicy and $IMUDPSchedulingPriority config settings
 - added $LocalHostName config directive
+- improved tcpsrv performance by enabling multiple-entry epoll
+  so far, we always pulled a single event from the epoll interface. 
+  Now 128, what should result in performance improvement (less API
+  calls) on busy systems. Most importantly affects imtcp.
 - bugfix: fixed build problems on some platforms
   namely those that have 32bit atomic operations but not 64 bit ones
 - bugfix: local hostname was pulled too-early, so that some config 
diff --git a/runtime/nsd.h b/runtime/nsd.h
index e5b9320b..1d44a14c 100644
--- a/runtime/nsd.h
+++ b/runtime/nsd.h
@@ -92,7 +92,7 @@ BEGINinterface(nsdpoll) /* name must also be changed in ENDinterface macro! */
 	rsRetVal (*Construct)(nsdpoll_t **ppThis);
 	rsRetVal (*Destruct)(nsdpoll_t **ppThis);
 	rsRetVal (*Ctl)(nsdpoll_t *pNsdpoll, nsd_t *pNsd, int id, void *pUsr, int mode, int op);
-	rsRetVal (*Wait)(nsdpoll_t *pNsdpoll, int timeout, int *idRdy, void **ppUsr);
+	rsRetVal (*Wait)(nsdpoll_t *pNsdpoll, int timeout, int *numReady, int idRdy[], void *ppUsr[]);
 ENDinterface(nsdpoll)
 #define nsdpollCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
 
diff --git a/runtime/nsdpoll_ptcp.c b/runtime/nsdpoll_ptcp.c
index bc374c60..f22c6cb6 100644
--- a/runtime/nsdpoll_ptcp.c
+++ b/runtime/nsdpoll_ptcp.c
@@ -199,23 +199,26 @@ finalize_it:
 /* Wait for io to become ready. After the successful call, idRdy contains the
  * id set by the caller for that i/o event, ppUsr is a pointer to a location
  * where the user pointer shall be stored.
- * TODO: this is a trivial implementation that only polls one event at a time. We
- *       may later extend it to poll for multiple events, what would cause less 
- *       overhead.
+ * numEntries contains the maximum number of entries on entry and the actual
+ * number of entries actually read on exit.
  * rgerhards, 2009-11-18
  */
 static rsRetVal
-Wait(nsdpoll_t *pNsdpoll, int timeout, int *idRdy, void **ppUsr) {
+Wait(nsdpoll_t *pNsdpoll, int timeout, int *numEntries, int idRdy[], void *ppUsr[]) {
 	nsdpoll_ptcp_t *pThis = (nsdpoll_ptcp_t*) pNsdpoll;
 	nsdpoll_epollevt_lst_t *pOurEvt;
-	struct epoll_event event;
+	struct epoll_event event[128];
 	int nfds;
+	int i;
 	DEFiRet;
 
 	assert(idRdy != NULL);
 	assert(ppUsr != NULL);
 
-	nfds = epoll_wait(pThis->efd, &event, 1, timeout);
+	if(*numEntries > 128)
+		*numEntries = 128;
+	DBGPRINTF("doing epoll_wait for max %d events\n", *numEntries);
+	nfds = epoll_wait(pThis->efd, event, *numEntries, timeout);
 	if(nfds == -1) {
 		if(errno == EINTR) {
 			ABORT_FINALIZE(RS_RET_EINTR);
@@ -227,10 +230,15 @@ Wait(nsdpoll_t *pNsdpoll, int timeout, int *idRdy, void **ppUsr) {
 		ABORT_FINALIZE(RS_RET_TIMEOUT);
 	}
 
-	/* we got a valid event, so tell the caller... */
-	pOurEvt = (nsdpoll_epollevt_lst_t*) event.data.u64;
-	*idRdy = pOurEvt->id;
-	*ppUsr = pOurEvt->pUsr;
+	/* we got valid events, so tell the caller... */
+dbgprintf("epoll returned %d entries\n", nfds);
+	for(i = 0 ; i < nfds ; ++i) {
+		pOurEvt = (nsdpoll_epollevt_lst_t*) event[i].data.u64;
+		idRdy[i] = pOurEvt->id;
+		ppUsr[i] = pOurEvt->pUsr;
+dbgprintf("epoll push ppusr[%d]: %p\n", i, pOurEvt->pUsr);
+	}
+	*numEntries = nfds;
 
 finalize_it:
 	RETiRet;
diff --git a/runtime/nspoll.c b/runtime/nspoll.c
index 64927280..c36375fd 100644
--- a/runtime/nspoll.c
+++ b/runtime/nspoll.c
@@ -129,11 +129,11 @@ finalize_it:
 /* Carries out the actual wait (all done in lower layers)
  */
 static rsRetVal
-Wait(nspoll_t *pThis, int timeout, int *idRdy, void **ppUsr) {
+Wait(nspoll_t *pThis, int timeout, int *numEntries, int idRdy[], void *ppUsr[]) {
 	DEFiRet;
 	ISOBJ_TYPE_assert(pThis, nspoll);
 	assert(idRdy != NULL);
-	iRet = pThis->Drvr.Wait(pThis->pDrvrData, timeout, idRdy, ppUsr);
+	iRet = pThis->Drvr.Wait(pThis->pDrvrData, timeout, numEntries, idRdy, ppUsr);
 	RETiRet;
 }
 
diff --git a/runtime/nspoll.h b/runtime/nspoll.h
index a77759c0..4b066577 100644
--- a/runtime/nspoll.h
+++ b/runtime/nspoll.h
@@ -50,11 +50,12 @@ BEGINinterface(nspoll) /* name must also be changed in ENDinterface macro! */
 	rsRetVal (*Construct)(nspoll_t **ppThis);
 	rsRetVal (*ConstructFinalize)(nspoll_t *pThis);
 	rsRetVal (*Destruct)(nspoll_t **ppThis);
-	rsRetVal (*Wait)(nspoll_t *pNsdpoll, int timeout, int *idRdy, void **ppUsr);
+	rsRetVal (*Wait)(nspoll_t *pNsdpoll, int timeout, int *numEntries, int idRdy[], void *ppUsr[]);
 	rsRetVal (*Ctl)(nspoll_t *pNsdpoll, netstrm_t *pStrm, int id, void *pUsr, int mode, int op);
 	rsRetVal (*IsEPollSupported)(void); /* static method */
 ENDinterface(nspoll)
-#define nspollCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+#define nspollCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */
+/* interface change in v2 is that wait supports multiple return objects */
 
 /* prototypes */
 PROTOTYPEObj(nspoll);
diff --git a/tcpsrv.c b/tcpsrv.c
index fbb9446d..3a2cf81d 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -637,9 +637,12 @@ Run(tcpsrv_t *pThis)
 {
 	DEFiRet;
 	int i;
+	int retIDs[128]; /* 128 is currently fixed num of concurrent requests */
+	void *pUsr[128];
+	int numEntries;
 	tcps_sess_t *pNewSess;
 	nspoll_t *pPoll = NULL;
-	void *pUsr;
+	int currIdx;
 	rsRetVal localRet;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
@@ -659,7 +662,7 @@ Run(tcpsrv_t *pThis)
 		FINALIZE;
 	}
 
-	dbgprintf("tcpsrv uses epoll() interface, nsdpol driver found\n");
+	dbgprintf("tcpsrv uses epoll() interface, nsdpoll driver found\n");
 
 	/* flag that we are in epoll mode */
 	pThis->bUsingEPoll = TRUE;
@@ -672,7 +675,8 @@ Run(tcpsrv_t *pThis)
 	}
 
 	while(1) {
-		localRet = nspoll.Wait(pPoll, -1, &i, &pUsr);
+		numEntries = sizeof(retIDs)/sizeof(int);
+		localRet = nspoll.Wait(pPoll, -1, &numEntries, retIDs, pUsr);
 		if(glbl.GetGlobalInputTermState() == 1)
 			break; /* terminate input! */
 
@@ -683,16 +687,20 @@ Run(tcpsrv_t *pThis)
 		if(localRet != RS_RET_OK)
 			continue;
 
-		dbgprintf("poll returned with i %d, pUsr %p\n", i, pUsr);
-
-		if(pUsr == pThis->ppLstn) {
-			DBGPRINTF("New connect on NSD %p.\n", pThis->ppLstn[i]);
-			SessAccept(pThis, pThis->ppLstnPort[i], &pNewSess, pThis->ppLstn[i]);
-			CHKiRet(nspoll.Ctl(pPoll, pNewSess->pStrm, 0, pNewSess, NSDPOLL_IN, NSDPOLL_ADD));
-			DBGPRINTF("New session created with NSD %p.\n", pNewSess);
-		} else {
-			pNewSess = (tcps_sess_t*) pUsr;
-			doReceive(pThis, &pNewSess, pPoll);
+		dbgprintf("poll returned with %d entries.\n", numEntries);
+
+		for(i = 0 ; i < numEntries ; i++) {
+			currIdx = retIDs[i];
+			dbgprintf("tcpsrv processing i %d, pUsr %p\n", currIdx, pUsr[i]);
+			if(pUsr[i] == pThis->ppLstn) {
+				DBGPRINTF("New connect on NSD %p.\n", pThis->ppLstn[currIdx]);
+				SessAccept(pThis, pThis->ppLstnPort[currIdx], &pNewSess, pThis->ppLstn[currIdx]);
+				CHKiRet(nspoll.Ctl(pPoll, pNewSess->pStrm, 0, pNewSess, NSDPOLL_IN, NSDPOLL_ADD));
+				DBGPRINTF("New session created with NSD %p.\n", pNewSess);
+			} else {
+				pNewSess = (tcps_sess_t*) pUsr[i];
+				doReceive(pThis, &pNewSess, pPoll);
+			}
 		}
 	}