From ee4aed1713bb968afa6db992f9e2e6c00d6c9350 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 10 Aug 2010 16:18:21 +0200 Subject: added tests for imptcp and fixed some problems with it it now also works reliably in edge-triggered mode --- plugins/imptcp/imptcp.c | 59 ++++++++++++++++++++++++++------------ tests/Makefile.am | 6 ++++ tests/diag.sh | 2 +- tests/imptcp_large.sh | 14 +++++++++ tests/testsuites/imptcp_large.conf | 16 +++++++++++ 5 files changed, 78 insertions(+), 19 deletions(-) create mode 100755 tests/imptcp_large.sh create mode 100644 tests/testsuites/imptcp_large.conf diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c index 5aeb0192..80df959c 100644 --- a/plugins/imptcp/imptcp.c +++ b/plugins/imptcp/imptcp.c @@ -434,6 +434,8 @@ AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP) iNewSock = accept(sock, (struct sockaddr*) &addr, &addrlen); if(iNewSock < 0) { + if(errno == EAGAIN || errno == EWOULDBLOCK) + ABORT_FINALIZE(RS_RET_NO_MORE_DATA); ABORT_FINALIZE(RS_RET_ACCEPT_ERR); } @@ -747,6 +749,8 @@ addLstn(ptcpsrv_t *pSrv, int sock) /* add to start of server's listener list */ pLstn->prev = NULL; pLstn->next = pSrv->pLstn; + if(pSrv->pLstn != NULL) + pSrv->pLstn->prev = pLstn; pSrv->pLstn = pLstn; iRet = addEPollSock(epolld_lstn, pLstn, sock, &pLstn->epd); @@ -777,6 +781,8 @@ addSess(ptcpsrv_t *pSrv, int sock, prop_t *peerName, prop_t *peerIP) /* add to start of server's listener list */ pSess->prev = NULL; pSess->next = pSrv->pSess; + if(pSrv->pSess != NULL) + pSrv->pSess->prev = pSess; pSrv->pSess = pSess; iRet = addEPollSock(epolld_sess, pSess, sock, &pSess->epd); @@ -801,6 +807,9 @@ closeSess(ptcpsess_t *pSess) close(sock); /* finally unlink session from structures */ +//fprintf(stderr, "closing session %d next %p, prev %p\n", pSess->sock, pSess->next, pSess->prev); +//dbgprintf("imptcp: pSess->next %p\n", pSess->next); +//dbgprintf("imptcp: pSess->prev %p\n", pSess->prev); if(pSess->next != NULL) pSess->next->prev = pSess->prev; if(pSess->prev == NULL) { @@ -921,11 +930,19 @@ lstnActivity(ptcplstn_t *pLstn) int newSock; prop_t *peerName; prop_t *peerIP; + rsRetVal localRet; +int iac = 0; DEFiRet; DBGPRINTF("imptcp: new connection on listen socket %d\n", pLstn->sock); - CHKiRet(AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP)); - CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP)); + while(1) { + localRet = AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP); +//if(iac++ > 0) fprintf(stderr, "%d accepts in a row!\n", iac); + if(localRet == RS_RET_NO_MORE_DATA) + break; + CHKiRet(localRet); + CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP)); + } finalize_it: RETiRet; @@ -940,24 +957,32 @@ sessActivity(ptcpsess_t *pSess) { int lenRcv; int lenBuf; - char rcvBuf[4096]; + char rcvBuf[128*1024]; DEFiRet; +int iac = 0; DBGPRINTF("imptcp: new activity on session socket %d\n", pSess->sock); - lenBuf = sizeof(rcvBuf); - lenRcv = recv(pSess->sock, rcvBuf, lenBuf, 0); - - if(lenRcv > 0) { - /* have data, process it */ - DBGPRINTF("imtcp: data(%d) on socket %d: %s\n", lenBuf, pSess->sock, rcvBuf); - CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv)); - } else if (lenRcv == 0) { - /* session was closed, do clean-up */ - CHKiRet(closeSess(pSess)); - } else { - DBGPRINTF("imtcp: error on session socket %d - closed.\n", pSess->sock); - closeSess(pSess); /* try clean-up by dropping session */ + while(1) { + lenBuf = sizeof(rcvBuf); + lenRcv = recv(pSess->sock, rcvBuf, lenBuf, 0); +//if(iac++ > 1) fprintf(stderr, "\n%d recv in a row!\n", iac-1); + + if(lenRcv > 0) { + /* have data, process it */ + DBGPRINTF("imtcp: data(%d) on socket %d: %s\n", lenBuf, pSess->sock, rcvBuf); + CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv)); + } else if (lenRcv == 0) { + /* session was closed, do clean-up */ + CHKiRet(closeSess(pSess)); + break; + } else { + if(errno == EAGAIN || errno == EWOULDBLOCK) + break; + DBGPRINTF("imtcp: error on session socket %d - closed.\n", pSess->sock); + closeSess(pSess); /* try clean-up by dropping session */ + break; + } } finalize_it: @@ -1127,8 +1152,6 @@ CODEmodInit_QueryRegCFSLineHdlr addTCPListener, NULL, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpservernotifyonconnectionclose"), 0, eCmdHdlrBinary, NULL, &cs.bEmitMsgOnClose, STD_LOADABLE_MODULE_ID)); - //CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverstreamdriverpermittedpeer"), 0, - //eCmdHdlrGetWord, setPermittedPeer, NULL, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserveraddtlframedelimiter"), 0, eCmdHdlrInt, NULL, &cs.iAddtlFrameDelim, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverinputname"), 0, diff --git a/tests/Makefile.am b/tests/Makefile.am index 0045f00a..b9b1ede7 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -7,6 +7,8 @@ TESTS = $(TESTRUNS) cfg.sh \ diskqueue.sh \ diskqueue-fsync.sh \ manytcp.sh \ + manyptcp.sh \ + imptcp_large.sh \ sndrcv.sh \ sndrcv_gzip.sh \ asynwr_simple.sh \ @@ -136,6 +138,10 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \ testsuites/imtcp-multiport.conf \ manytcp.sh \ testsuites/manytcp.conf \ + manyptcp.sh \ + testsuites/manyptcp.conf \ + imptcp_large.sh \ + testsuites/imptcp_large.conf \ inputname.sh \ testsuites/inputname_imtcp.conf \ testsuites/1.inputname_imtcp_12514 \ diff --git a/tests/diag.sh b/tests/diag.sh index 51ad5f6a..8659aa17 100755 --- a/tests/diag.sh +++ b/tests/diag.sh @@ -10,7 +10,7 @@ #valgrind="valgrind --tool=helgrind --log-fd=1" #valgrind="valgrind --tool=exp-ptrcheck --log-fd=1" #set -o xtrace -#export RSYSLOG_DEBUG="debug nostdout printmutexaction" +#export RSYSLOG_DEBUG="debug nostdout noprintmutexaction" #export RSYSLOG_DEBUGLOG="log" case $1 in 'init') $srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason diff --git a/tests/imptcp_large.sh b/tests/imptcp_large.sh new file mode 100755 index 00000000..f6eee895 --- /dev/null +++ b/tests/imptcp_large.sh @@ -0,0 +1,14 @@ +# Test imptcp with large messages +# added 2010-08-10 by Rgerhards +# +# This file is part of the rsyslog project, released under GPLv3 +cat rsyslog.action.1.include +source $srcdir/diag.sh init +source $srcdir/diag.sh startup imptcp_large.conf +# send 4000 messages of 10.000bytes plus header max, randomized +source $srcdir/diag.sh tcpflood -c5 -m20000 -r -d10000 -P129 +sleep 2 # due to large messages, we need this time for the tcp receiver to settle... +source $srcdir/diag.sh shutdown-when-empty # shut down rsyslogd when done processing messages +source $srcdir/diag.sh wait-shutdown # and wait for it to terminate +source $srcdir/diag.sh seq-check 0 19999 -E +source $srcdir/diag.sh exit diff --git a/tests/testsuites/imptcp_large.conf b/tests/testsuites/imptcp_large.conf new file mode 100644 index 00000000..677e33f6 --- /dev/null +++ b/tests/testsuites/imptcp_large.conf @@ -0,0 +1,16 @@ +# simple async writing test +# rgerhards, 2010-03-09 +$MaxMessageSize 10k +$IncludeConfig diag-common.conf + +$ModLoad ../plugins/imptcp/.libs/imptcp +$MainMsgQueueTimeoutShutdown 10000 +$InputPTCPServerRun 13514 + +$template outfmt,"%msg:F,58:2%,%msg:F,58:3%,%msg:F,58:4%\n" +$template dynfile,"rsyslog.out.log" # trick to use relative path names! +$OMFileFlushOnTXEnd off +$OMFileFlushInterval 2 +$OMFileIOBufferSize 256k +$IncludeConfig rsyslog.action.1.include +local0.* ?dynfile;outfmt -- cgit