Imported Upstream version 5.8.2upstream/5.8.2

31 files changed, 989 insertions, 190 deletions

diff --git a/ChangeLog b/ChangeLog
index 665ed18..e438ac6 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,4 +1,20 @@
 ---------------------------------------------------------------------------
+Version 5.8.2  [V5-stable] (rgerhards), 2011-06-21
+- bugfix: problems in failover action handling
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=270
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=254
+- bugfix: mutex was invalidly left unlocked during action processing
+  At least one case where this can occur is during thread shutdown, which
+  may be initiated by lower activity. In most cases, this is quite
+  unlikely to happen. However, if it does, data structures may be 
+  corrupted which could lead to fatal failure and segfault. I detected
+  this via a testbench test, not a user report. But I assume that some
+  users may have had unreproducable aborts that were cause by this bug.
+- bugfix: memory leak in imtcp & subsystems under some circumstances
+  This leak is tied to error conditions which lead to incorrect cleanup
+  of some data structures. [backport from v6]
+- bugfix/improvement:$WorkDirectory now gracefully handles trailing slashes
+---------------------------------------------------------------------------
 Version 5.8.1  [V5-stable] (rgerhards), 2011-05-19
 - bugfix: invalid processing in QUEUE_FULL condition
   If the the multi-submit interface was used and a QUEUE_FULL condition
@@ -829,6 +845,9 @@ Version 4.7.0  [v4-devel] (rgerhards), 2010-04-14
 - imported changes from 4.5.6 and below
 ---------------------------------------------------------------------------
 Version 4.6.6  [v4-stable] (rgerhards), 2010-11-??
+- bugfix: memory leak in imtcp & subsystems under some circumstances
+  This leak is tied to error conditions which lead to incorrect cleanup
+  of some data structures. [backport from v6, limited testing under v4]
 - bugfix: invalid processing in QUEUE_FULL condition
   If the the multi-submit interface was used and a QUEUE_FULL condition
   occured, the failed message was properly destructed. However, the
@@ -881,6 +900,9 @@ Version 4.6.6  [v4-stable] (rgerhards), 2010-11-??
   discarded (due to QUEUE_FULL or similar problem)
 - bugfix: a slightly more informative error message when a TCP
   connections is aborted
+- bugfix: timestamp was incorrectly calculated for timezones with minute
+  offset
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=271
 - some improvements thanks to clang's static code analyzer
   o overall cleanup (mostly unnecessary writes and otherwise unused stuff)
   o bugfix: fixed a very remote problem in msg.c which could occur when
@@ -1519,6 +1541,9 @@ version before switching to this one.
   Thanks to Ken for providing the patch
 ---------------------------------------------------------------------------
 Version 3.22.4 [v3-stable] (rgerhards), 2010-??-??
+- bugfix: timestamp was incorrectly calculated for timezones with minute
+  offset
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=271
 - improved some code based on clang static analyzer results
 ---------------------------------------------------------------------------
 Version 3.22.3 [v3-stable] (rgerhards), 2010-11-24
diff --git a/action.c b/action.c
index c5bd03c..54a05fc 100644
--- a/action.c
+++ b/action.c
@@ -39,7 +39,35 @@
  * - processAction
  * - submitBatch
  * - tryDoAction
- * - 
+ * - ...
+ *
+ * MORE ON PROCESSING, QUEUES and FILTERING
+ * All filtering needs to be done BEFORE messages are enqueued to an
+ * action. In previous code, part of the filtering was done at the
+ * "remote end" of the action queue, which lead to problems in
+ * non-direct mode (because then things run asynchronously). In order
+ * to solve this problem once and for all, I have changed the code so
+ * that all filtering is done before enq, and processing on the
+ * dequeue side of action processing now always executes whatever is
+ * enqueued. This is the only way to handle things consistently and
+ * (as much as possible) in a queue-type agnostic way. However, it is
+ * a rather radical change, which I unfortunately needed to make from
+ * stable version 5.8.1 to 5.8.2. If new problems pop up, you now know
+ * what may be their cause. In any case, the way it is done now is the
+ * only correct one.
+ * A problem is that, under fortunate conditions, we use the current
+ * batch for the output system as well. This is very good from a performance
+ * point of view, but makes the distinction between enq and deq side of
+ * the queue a bit hard. The current idea is that the filter condition
+ * alone is checked at the deq side of the queue (seems to be unavoidable
+ * to do it that way), but all other complex conditons (like failover
+ * handling) go into the computation of the filter condition. For
+ * non-direct queues, we still enqueue only what is acutally necessary.
+ * Note that in this case the rest of the code must ensure that the filter
+ * is set to "true". While this is not perfect and not as simple as
+ * we would like to see it, it looks like the best way to tackle that
+ * beast.
+ * rgerhards, 2011-06-15
  *
  * Copyright 2007-2011 Rainer Gerhards and Adiscon GmbH.
  *
@@ -611,8 +639,8 @@ static rsRetVal actionTryResume(action_t *pThis, int *pbShutdownImmediate)
 	}
 
 	if(Debug && (pThis->eState == ACT_STATE_RTRY ||pThis->eState == ACT_STATE_SUSP)) {
-		DBGPRINTF("actionTryResume: action state: %s, next retry (if applicable): %u [now %u]\n",
-			getActStateName(pThis), (unsigned) pThis->ttResumeRtry, (unsigned) ttNow);
+		DBGPRINTF("actionTryResume: action %p state: %s, next retry (if applicable): %u [now %u]\n",
+			pThis, getActStateName(pThis), (unsigned) pThis->ttResumeRtry, (unsigned) ttNow);
 	}
 
 finalize_it:
@@ -932,27 +960,32 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem)
 	i = pBatch->iDoneUpTo;	/* all messages below that index are processed */
 	iElemProcessed = 0;
 	iCommittedUpTo = i;
+dbgprintf("XXXXX:     tryDoAction %p, pnElem %d, nElem %d\n", pAction, *pnElem, pBatch->nElem);
 	while(iElemProcessed <= *pnElem && i < pBatch->nElem) {
 		if(*(pBatch->pbShutdownImmediate))
 			ABORT_FINALIZE(RS_RET_FORCE_TERM);
+		/* NOTE: do NOT extend the filter below! Anything else must be done on the
+		 * enq side of the queue (see file header comment)! -- rgerhards, 2011-06-15
+		 */
 		if(   pBatch->pElem[i].bFilterOK
-		   && pBatch->pElem[i].state != BATCH_STATE_DISC//) {
-		   && ((pAction->bExecWhenPrevSusp  == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
+		    && pBatch->pElem[i].state != BATCH_STATE_DISC) {
 			pMsg = (msg_t*) pBatch->pElem[i].pUsrp;
 			localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams,
-						        pBatch->pbShutdownImmediate);
-			DBGPRINTF("action call returned %d\n", localRet);
+							pBatch->pbShutdownImmediate);
+			DBGPRINTF("action %p call returned %d\n", pAction, localRet);
 			/* Note: we directly modify the batch object state, because we know that
 			 * wo do not overwrite BATCH_STATE_DISC indicators!
 			 */
 			if(localRet == RS_RET_OK) {
 				/* mark messages as committed */
 				while(iCommittedUpTo <= i) {
+					pBatch->pElem[iCommittedUpTo].bPrevWasSuspended = 0; /* we had success! */
 					pBatch->pElem[iCommittedUpTo++].state = BATCH_STATE_COMM;
 				}
 			} else if(localRet == RS_RET_PREVIOUS_COMMITTED) {
 				/* mark messages as committed */
 				while(iCommittedUpTo < i) {
+					pBatch->pElem[iCommittedUpTo].bPrevWasSuspended = 0; /* we had success! */
 					pBatch->pElem[iCommittedUpTo++].state = BATCH_STATE_COMM;
 				}
 				pBatch->pElem[i].state = BATCH_STATE_SUB;
@@ -1035,6 +1068,8 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
 				bDone = 1;
 			} else {
 				/* retry with half as much. Depth is log_2 batchsize, so recursion is not too deep */
+				DBGPRINTF("submitBatch recursing trying to find and exclude the culprit "
+				          "for iRet %d\n", localRet);
 				submitBatch(pAction, pBatch, nElem / 2);
 				submitBatch(pAction, pBatch, nElem - (nElem / 2));
 				bDone = 1;
@@ -1224,11 +1259,13 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
 
 /* This function builds up a batch of messages to be (later)
  * submitted to the action queue.
- * Note: this function is also called from syslogd itself as part of its
- * flush processing. If so, pBatch will be NULL and idxBtch undefined.
+ * Important: this function MUST not be called with messages that are to
+ * be discarded due to their "prevWasSuspended" state. It will not check for
+ * this and submit all messages to the queue for execution. So these must
+ * be filtered out before calling us (what is done currently!).
  */
 rsRetVal
-actionWriteToAction(action_t *pAction, batch_t *pBatch, int idxBtch)
+actionWriteToAction(action_t *pAction)
 {
 	msg_t *pMsgSave;	/* to save current message pointer, necessary to restore
 				   it in case it needs to be updated (e.g. repeated msgs) */
@@ -1325,35 +1362,7 @@ actionWriteToAction(action_t *pAction, batch_t *pBatch, int idxBtch)
 	/* When we reach this point, we have a valid, non-disabled action.
 	 * So let's enqueue our message for execution. -- rgerhards, 2007-07-24
 	 */
-	if(   pBatch != NULL
-	   && (pAction->bExecWhenPrevSusp  == 1 && pBatch->pElem[idxBtch].bPrevWasSuspended)) {
-		/* in that case, we need to create a special batch which reflects the
-		 * suspended state. Otherwise, that information would be dropped inside
-		 * the queue engine. TODO: in later releases (v6?) create a better
-		 * solution than what we do here. However, for v5 this sounds much too
-		 * intrusive. -- rgerhardsm, 2011-03-16
-		 * (Code is copied over from queue.c and slightly modified)
-		 */
-		batch_t singleBatch;
-		batch_obj_t batchObj;
-		int i;
-		memset(&batchObj, 0, sizeof(batch_obj_t));
-		memset(&singleBatch, 0, sizeof(batch_t));
-		batchObj.state = BATCH_STATE_RDY;
-		batchObj.pUsrp = (obj_t*) pAction->f_pMsg;
-		batchObj.bPrevWasSuspended = 1;
-		batchObj.bFilterOK = 1;
-		singleBatch.nElem = 1; /* there always is only one in direct mode */
-		singleBatch.pElem = &batchObj;
-
-		iRet = qqueueEnqObjDirectBatch(pAction->pQueue, &singleBatch);
-
-		for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
-			free(batchObj.staticActStrings[i]);
-		}
-	} else { /* standard case, just submit */
-		iRet = doSubmitToActionQ(pAction, pAction->f_pMsg);
-	}
+	iRet = doSubmitToActionQ(pAction, pAction->f_pMsg);
 
 	if(iRet == RS_RET_OK)
 		pAction->f_prevcount = 0; /* message processed, so we start a new cycle */
@@ -1413,7 +1422,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch)
 		 * isolated messages), but back off so we'll flush less often in the future.
 		 */
 		if(getActNow(pAction) > REPEATTIME(pAction)) {
-			iRet = actionWriteToAction(pAction, pBatch, idxBtch);
+			iRet = actionWriteToAction(pAction);
 			BACKOFF(pAction);
 		}
 	} else {/* new message, save it */
@@ -1422,7 +1431,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch)
 		 */
 		if(pAction->f_pMsg != NULL) {
 			if(pAction->f_prevcount > 0)
-				actionWriteToAction(pAction, pBatch, idxBtch);
+				actionWriteToAction(pAction);
 				/* we do not care about iRet above - I think it's right but if we have
 				 * some troubles, you know where to look at ;) -- rgerhards, 2007-08-01
 				 */
@@ -1430,7 +1439,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch)
 		}
 		pAction->f_pMsg = MsgAddRef(pMsg);
 		/* call the output driver */
-		iRet = actionWriteToAction(pAction, pBatch, idxBtch);
+		iRet = actionWriteToAction(pAction);
 	}
 
 finalize_it:
@@ -1519,6 +1528,69 @@ finalize_it:
 }
 
 
+/* enqueue a batch in direct mode. We have put this into its own function just to avoid
+ * cluttering the actual submit function.
+ * rgerhards, 2011-06-16
+ */
+static inline rsRetVal
+doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
+{
+	sbool FilterSave[1024];
+	sbool *pFilterSave;
+	sbool bNeedSubmit;
+	sbool bModifiedFilter;
+	int i;
+	DEFiRet;
+
+	if(batchNumMsgs(pBatch) <= (int) (sizeof(FilterSave)/sizeof(sbool))) {
+		pFilterSave = FilterSave;
+	} else {
+		CHKmalloc(pFilterSave = malloc(batchNumMsgs(pBatch) * sizeof(sbool)));
+	}
+
+	/* note: for direct mode, we need to adjust the filter property. For non-direct
+	 * this is not necessary, because in that case we enqueue only what actually needs
+	 * to be processed.
+	 */
+	if(pAction->bExecWhenPrevSusp) {
+		bNeedSubmit = 0;
+		bModifiedFilter = 0;
+		for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
+			pFilterSave[i] = pBatch->pElem[i].bFilterOK;
+			if(!pBatch->pElem[i].bPrevWasSuspended) {
+				DBGPRINTF("action enq stage: change bFilterOK to 0 due to "
+					  "failover case in elem %d\n", i);
+				pBatch->pElem[i].bFilterOK = 0;
+				bModifiedFilter = 1;
+			}
+			if(pBatch->pElem[i].bFilterOK)
+				bNeedSubmit = 1;
+			DBGPRINTF("action %p[%d]: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+				   pAction, i, pBatch->pElem[i].bFilterOK,  pBatch->pElem[i].state,
+				   pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
+		}
+		if(bNeedSubmit) {
+			iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
+		} else {
+			DBGPRINTF("no need to submit batch, all bFilterOK==0\n");
+		}
+		if(bModifiedFilter) {
+			for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
+				DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+					   pAction, pBatch->pElem[i].bFilterOK,  pBatch->pElem[i].state,
+					   pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
+				/* note: clang static code analyzer reports a false positive below */
+				pBatch->pElem[i].bFilterOK = pFilterSave[i];
+			}
+		}
+	} else {
+		iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
+	}
+
+finalize_it:
+	RETiRet;
+}
+
 /* This submits the message to the action queue in case we do NOT need to handle repeat
  * message processing. That case permits us to gain lots of freedom during processing
  * and thus speed.
@@ -1531,13 +1603,19 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch)
 	DEFiRet;
 
 	DBGPRINTF("Called action(Batch), logging to %s\n", module.GetStateName(pAction->pMod));
-	if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
-		iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
-	else {  /* in this case, we do single submits to the queue. 
+
+	if(pAction->pQueue->qType == QUEUETYPE_DIRECT) {
+		iRet = doQueueEnqObjDirectBatch(pAction, pBatch);
+	} else {/* in this case, we do single submits to the queue. 
 		 * TODO: optimize this, we may do at least a multi-submit!
 		 */
 		for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
-			if(pBatch->pElem[i].bFilterOK) {
+			DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+		           pAction, pBatch->pElem[i].bFilterOK,  pBatch->pElem[i].state,
+			   pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
+			if(   pBatch->pElem[i].bFilterOK
+			   && pBatch->pElem[i].state != BATCH_STATE_DISC
+			   && (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) {
 				doSubmitToActionQ(pAction, (msg_t*)(pBatch->pElem[i].pUsrp));
 			}
 		}
@@ -1558,8 +1636,12 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
 	int i;
 	DEFiRet;
 
-	DBGPRINTF("Called action(complex case), logging to %s\n", module.GetStateName(pAction->pMod));
+	DBGPRINTF("Called action %p (complex case), logging to %s\n",
+		  pAction, module.GetStateName(pAction->pMod));
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
+		DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
+		           pAction, pBatch->pElem[i].bFilterOK,  pBatch->pElem[i].state,
+			   pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
 		if(   pBatch->pElem[i].bFilterOK
                    && pBatch->pElem[i].state != BATCH_STATE_DISC
 		   && ((pAction->bExecWhenPrevSusp  == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
diff --git a/action.h b/action.h
index 0ab8062..bae64d3 100644
--- a/action.h
+++ b/action.h
@@ -100,7 +100,7 @@ rsRetVal actionDestruct(action_t *pThis);
 rsRetVal actionDbgPrint(action_t *pThis);
 rsRetVal actionSetGlobalResumeInterval(int iNewVal);
 rsRetVal actionDoAction(action_t *pAction);
-rsRetVal actionWriteToAction(action_t *pAction, batch_t *pBatch, int idxBtch);
+rsRetVal actionWriteToAction(action_t *pAction);
 rsRetVal actionCallHUPHdlr(action_t *pAction);
 rsRetVal actionClassInit(void);
 rsRetVal addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, int bSuspended);
diff --git a/configure b/configure
index f62f453..d2e2857 100755
--- a/configure
+++ b/configure
@@ -1,6 +1,6 @@
 #! /bin/sh
 # Guess values for system-dependent variables and create Makefiles.
-# Generated by GNU Autoconf 2.66 for rsyslog 5.8.1.
+# Generated by GNU Autoconf 2.66 for rsyslog 5.8.2.
 #
 # Report bugs to <rsyslog@lists.adiscon.com>.
 #
@@ -562,8 +562,8 @@ MAKEFLAGS=
 # Identity of this package.
 PACKAGE_NAME='rsyslog'
 PACKAGE_TARNAME='rsyslog'
-PACKAGE_VERSION='5.8.1'
-PACKAGE_STRING='rsyslog 5.8.1'
+PACKAGE_VERSION='5.8.2'
+PACKAGE_STRING='rsyslog 5.8.2'
 PACKAGE_BUGREPORT='rsyslog@lists.adiscon.com'
 PACKAGE_URL=''
 
@@ -1475,7 +1475,7 @@ if test "$ac_init_help" = "long"; then
   # Omit some internal or obsolete options to make the list less imposing.
   # This message is too long to be a string in the A/UX 3.1 sh.
   cat <<_ACEOF
-\`configure' configures rsyslog 5.8.1 to adapt to many kinds of systems.
+\`configure' configures rsyslog 5.8.2 to adapt to many kinds of systems.
 
 Usage: $0 [OPTION]... [VAR=VALUE]...
 
@@ -1545,7 +1545,7 @@ fi
 
 if test -n "$ac_init_help"; then
   case $ac_init_help in
-     short | recursive ) echo "Configuration of rsyslog 5.8.1:";;
+     short | recursive ) echo "Configuration of rsyslog 5.8.2:";;
    esac
   cat <<\_ACEOF
 
@@ -1720,7 +1720,7 @@ fi
 test -n "$ac_init_help" && exit $ac_status
 if $ac_init_version; then
   cat <<\_ACEOF
-rsyslog configure 5.8.1
+rsyslog configure 5.8.2
 generated by GNU Autoconf 2.66
 
 Copyright (C) 2010 Free Software Foundation, Inc.
@@ -2299,7 +2299,7 @@ cat >config.log <<_ACEOF
 This file contains any messages produced by compilers while
 running configure, to aid debugging if configure makes a mistake.
 
-It was created by rsyslog $as_me 5.8.1, which was
+It was created by rsyslog $as_me 5.8.2, which was
 generated by GNU Autoconf 2.66.  Invocation command line was
 
   $ $0 $@
@@ -3114,7 +3114,7 @@ fi
 
 # Define the identity of the package.
  PACKAGE='rsyslog'
- VERSION='5.8.1'
+ VERSION='5.8.2'
 
 
 cat >>confdefs.h <<_ACEOF
@@ -17336,7 +17336,7 @@ cat >>$CONFIG_STATUS <<\_ACEOF || ac_write_fail=1
 # report actual input values of CONFIG_FILES etc. instead of their
 # values after options handling.
 ac_log="
-This file was extended by rsyslog $as_me 5.8.1, which was
+This file was extended by rsyslog $as_me 5.8.2, which was
 generated by GNU Autoconf 2.66.  Invocation command line was
 
   CONFIG_FILES    = $CONFIG_FILES
@@ -17402,7 +17402,7 @@ _ACEOF
 cat >>$CONFIG_STATUS <<_ACEOF || ac_write_fail=1
 ac_cs_config="`$as_echo "$ac_configure_args" | sed 's/^ //; s/[\\""\`\$]/\\\\&/g'`"
 ac_cs_version="\\
-rsyslog config.status 5.8.1
+rsyslog config.status 5.8.2
 configured by $0, generated by GNU Autoconf 2.66,
   with options \\"\$ac_cs_config\\"
 
diff --git a/configure.ac b/configure.ac
index 1faaff8..5873c66 100644
--- a/configure.ac
+++ b/configure.ac
@@ -2,7 +2,7 @@
 # Process this file with autoconf to produce a configure script.
 
 AC_PREREQ(2.61)
-AC_INIT([rsyslog],[5.8.1],[rsyslog@lists.adiscon.com])
+AC_INIT([rsyslog],[5.8.2],[rsyslog@lists.adiscon.com])
 AM_INIT_AUTOMAKE
 
 m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])])
diff --git a/doc/manual.html b/doc/manual.html
index 5c65675..db9eb90 100644
--- a/doc/manual.html
+++ b/doc/manual.html
@@ -19,7 +19,7 @@ rsyslog support</a> available directly from the source!</p>
 <p><b>Please visit the <a href="http://www.rsyslog.com/sponsors">rsyslog sponsor's page</a>
 to honor the project sponsors or become one yourself!</b> We are very grateful for any help towards the
 project goals.</p>
-<p><b>This documentation is for version 5.8.1 (stable branch) of rsyslog.</b>
+<p><b>This documentation is for version 5.8.2 (stable branch) of rsyslog.</b>
 Visit the <i><a href="http://www.rsyslog.com/status">rsyslog status page</a></i></b>
 to obtain current version information and project status.
 </p><p><b>If you like rsyslog, you might
diff --git a/doc/rsyslog_conf_global.html b/doc/rsyslog_conf_global.html
index a5d69f1..21786a7 100644
--- a/doc/rsyslog_conf_global.html
+++ b/doc/rsyslog_conf_global.html
@@ -279,7 +279,21 @@ default may change as uniprocessor systems become less common. [available since
 <li>$PreserveFQDN [on/<b>off</b>) - if set to off (legacy default to remain compatible
 to sysklogd), the domain part from a name that is within the same domain as the receiving
 system is stripped. If set to on, full names are always used.</li>
-<li>$WorkDirectory &lt;name&gt; (directory for spool and other work files)</li>
+<li>$WorkDirectory &lt;name&gt; (directory for spool and other work files.
+Do <b>not</b> use trailing slashes)</li>
+<li>$UDPServerAddress &lt;IP&gt; (imudp) -- local IP
+address (or name) the UDP listens should bind to</li>
+<li>$UDPServerRun &lt;port&gt; (imudp) -- former
+-r&lt;port&gt; option, default 514, start UDP server on this
+port, "*" means all addresses</li>
+<li>$UDPServerTimeRequery &lt;nbr-of-times&gt; (imudp) -- this is a performance
+optimization. Getting the system time is very costly. With this setting, imudp can
+be instructed to obtain the precise time only once every n-times. This logic is
+only activated if messages come in at a very fast rate, so doing less frequent
+time calls should usually be acceptable. The default value is two, because we have
+seen that even without optimization the kernel often returns twice the identical time.
+You can set this value as high as you like, but do so at your own risk. The higher
+the value, the less precise the timestamp.
 <li><a href="droppriv.html">$PrivDropToGroup</a></li>
 <li><a href="droppriv.html">$PrivDropToGroupID</a></li>
 <li><a href="droppriv.html">$PrivDropToUser</a></li>
diff --git a/plugins/imrelp/imrelp.c b/plugins/imrelp/imrelp.c
index 13fd442..20c4f45 100644
--- a/plugins/imrelp/imrelp.c
+++ b/plugins/imrelp/imrelp.c
@@ -60,8 +60,6 @@ static prop_t *pInputName = NULL;	/* there is only one global inputName for all
 
 
 /* config settings */
-static int iTCPSessMax = 200; /* max number of sessions */
-
 
 /* ------------------------------ callbacks ------------------------------ */
 #if 0
@@ -173,7 +171,6 @@ ENDmodExit
 static rsRetVal
 resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
 {
-	iTCPSessMax = 200;
 	return RS_RET_OK;
 }
 
@@ -197,8 +194,6 @@ CODEmodInit_QueryRegCFSLineHdlr
 	/* register config file handlers */
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputrelpserverrun", 0, eCmdHdlrGetWord,
 				   addListener, NULL, STD_LOADABLE_MODULE_ID));
-	CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputrelpmaxsessions", 0, eCmdHdlrInt,
-				   NULL, &iTCPSessMax, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler,
 		resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
 ENDmodInit
diff --git a/runtime/datetime.c b/runtime/datetime.c
index de26762..f81180e 100644
--- a/runtime/datetime.c
+++ b/runtime/datetime.c
@@ -122,7 +122,7 @@ static void getCurrTime(struct syslogTime *t, time_t *ttSeconds)
 	else
 		t->OffsetMode = '+';
 	t->OffsetHour = lBias / 3600;
-	t->OffsetMinute = lBias % 3600;
+	t->OffsetMinute = (lBias % 3600) / 60;
 	t->timeType = TIME_TYPE_RFC5424; /* we have a high precision timestamp */
 }
 
diff --git a/runtime/glbl.c b/runtime/glbl.c
index ec4992c..dea5a17 100644
--- a/runtime/glbl.c
+++ b/runtime/glbl.c
@@ -159,8 +159,29 @@ static void SetGlobalInputTermination(void)
  */
 static rsRetVal setWorkDir(void __attribute__((unused)) *pVal, uchar *pNewVal)
 {
-	DEFiRet;
+	size_t lenDir;
+	int i;
 	struct stat sb;
+	DEFiRet;
+
+	/* remove trailing slashes */
+	lenDir = ustrlen(pNewVal);
+	i = lenDir - 1;
+	while(i > 0 && pNewVal[i] == '/') {
+		--i;
+	}
+
+	if(i < 0) {
+		errmsg.LogError(0, RS_RET_ERR_WRKDIR, "$WorkDirectory: empty value "
+				"- directive ignored");
+		ABORT_FINALIZE(RS_RET_ERR_WRKDIR);
+	}
+
+	if(i != (int) lenDir - 1) {
+		pNewVal[i+1] = '\0';
+		errmsg.LogError(0, RS_RET_WRN_WRKDIR, "$WorkDirectory: trailing slashes "
+			"removed, new value is '%s'", pNewVal);
+	}
 
 	if(stat((char*) pNewVal, &sb) != 0) {
 		errmsg.LogError(0, RS_RET_ERR_WRKDIR, "$WorkDirectory: %s can not be "
diff --git a/runtime/nsd_gtls.c b/runtime/nsd_gtls.c
index 152dc8d..ca4b292 100644
--- a/runtime/nsd_gtls.c
+++ b/runtime/nsd_gtls.c
@@ -1121,6 +1121,7 @@ gtlsEndSess(nsd_gtls_t *pThis)
 			}
 		}
 		gnutls_deinit(pThis->sess);
+		pThis->bHaveSess = 0;
 	}
 	RETiRet;
 }
@@ -1174,6 +1175,8 @@ CODESTARTobjDestruct(nsd_gtls)
 		gnutls_x509_crt_deinit(pThis->ourCert);
 	if(pThis->bOurKeyIsInit)
 		gnutls_x509_privkey_deinit(pThis->ourKey);
+	if(pThis->bHaveSess)
+		gnutls_deinit(pThis->sess);
 ENDobjDestruct(nsd_gtls)
 
 
diff --git a/runtime/nsd_ptcp.c b/runtime/nsd_ptcp.c
index c891523..69eb768 100644
--- a/runtime/nsd_ptcp.c
+++ b/runtime/nsd_ptcp.c
@@ -334,6 +334,12 @@ AcceptConnReq(nsd_t *pNsd, nsd_t **ppNew)
 
 	iNewSock = accept(pThis->sock, (struct sockaddr*) &addr, &addrlen);
 	if(iNewSock < 0) {
+		if(Debug) {
+			char errStr[1024];
+			rs_strerror_r(errno, errStr, sizeof(errStr));
+			dbgprintf("nds_ptcp: error accepting connection on socket %d, errno %d: %s\n",
+				  pThis->sock, errno, errStr);
+		}
 		ABORT_FINALIZE(RS_RET_ACCEPT_ERR);
 	}
 
diff --git a/runtime/nsdsel_gtls.c b/runtime/nsdsel_gtls.c
index 1a389a0..aff55af 100644
--- a/runtime/nsdsel_gtls.c
+++ b/runtime/nsdsel_gtls.c
@@ -177,6 +177,7 @@ doRetry(nsd_gtls_t *pNsd)
 finalize_it:
 	if(iRet != RS_RET_OK && iRet != RS_RET_CLOSED && iRet != RS_RET_RETRY)
 		pNsd->bAbortConn = 1; /* request abort */
+dbgprintf("XXXXXX: doRetry: iRet %d, pNsd->bAbortConn %d\n", iRet, pNsd->bAbortConn);
 	RETiRet;
 }
 
diff --git a/runtime/queue.c b/runtime/queue.c
index 88e01a7..00eb76c 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -1678,6 +1678,7 @@ static rsRetVal
 ConsumerReg(qqueue_t *pThis, wti_t *pWti)
 {
 	int iCancelStateSave;
+	int bNeedReLock = 0;	/**< do we need to lock the mutex again? */
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, qqueue);
@@ -1687,6 +1688,7 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti)
 
 	/* we now have a non-idle batch of work, so we can release the queue mutex and process it */
 	d_pthread_mutex_unlock(pThis->mut);
+	bNeedReLock = 1;
 
 	/* at this spot, we may be cancelled */
 	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave);
@@ -1706,12 +1708,14 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti)
 	/* but now cancellation is no longer permitted */
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
 
-	/* now we are done, but need to re-aquire the mutex */
-	d_pthread_mutex_lock(pThis->mut);
-
 finalize_it:
-	dbgprintf("regular consumer finished, iret=%d, szlog %d sz phys %d\n", iRet,
+	DBGPRINTF("regular consumer finished, iret=%d, szlog %d sz phys %d\n", iRet,
 	          getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis));
+
+	/* now we are done, but potentially need to re-aquire the mutex */
+	if(bNeedReLock)
+		d_pthread_mutex_lock(pThis->mut);
+
 	RETiRet;
 }
 
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index d63dbe4..52b29ac 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -342,6 +342,7 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_ERR_HDFS_OPEN = -2179, /**< error during hdfsOpen (e.g. file does not exist) */
 	RS_RET_FILE_NOT_SPECIFIED = -2180, /**< file name not configured where this was required */
 	RS_RET_ERR_WRKDIR = -2181, /**< problems with the rsyslog working directory */
+	RS_RET_WRN_WRKDIR = -2182, /**< correctable problems with the rsyslog working directory */
 
 	/* RainerScript error messages (range 1000.. 1999) */
 	RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/runtime/rule.c b/runtime/rule.c
index d5f18e7..0776e2d 100644
--- a/runtime/rule.c
+++ b/runtime/rule.c
@@ -266,6 +266,7 @@ static rsRetVal
 processBatch(rule_t *pThis, batch_t *pBatch)
 {
 	int i;
+	rsRetVal localRet;
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, rule);
@@ -273,9 +274,13 @@ processBatch(rule_t *pThis, batch_t *pBatch)
 
 	/* first check the filters and reset status variables */
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
-		CHKiRet(shouldProcessThisMessage(pThis, (msg_t*)(pBatch->pElem[i].pUsrp),
-						 &(pBatch->pElem[i].bFilterOK)));
-		// TODO: really abort on error? 2010-06-10
+		localRet = shouldProcessThisMessage(pThis, (msg_t*)(pBatch->pElem[i].pUsrp),
+						    &(pBatch->pElem[i].bFilterOK));
+		if(localRet != RS_RET_OK) {
+			DBGPRINTF("processBatch: iRet %d returned from shouldProcessThisMessage, "
+			          "ignoring message\n", localRet);
+			pBatch->pElem[i].bFilterOK = 0;
+		}
 		if(pBatch->pElem[i].bFilterOK) {
 			/* re-init only when actually needed (cache write cost!) */
 			pBatch->pElem[i].bPrevWasSuspended = 0;
diff --git a/tcpsrv.c b/tcpsrv.c
index 9972a13..0b82251 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -556,6 +556,7 @@ RunSelect(tcpsrv_t *pThis)
 	int bIsReady;
 	tcps_sess_t *pNewSess;
 	nssel_t *pSel = NULL;
+	rsRetVal localRet;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
@@ -604,8 +605,8 @@ RunSelect(tcpsrv_t *pThis)
 		while(nfds && iTCPSess != -1) {
 			if(glbl.GetGlobalInputTermState() == 1)
 				ABORT_FINALIZE(RS_RET_FORCE_TERM);
-			CHKiRet(nssel.IsReady(pSel, pThis->pSessions[iTCPSess]->pStrm, NSDSEL_RD, &bIsReady, &nfds));
-			if(bIsReady) {
+			localRet = nssel.IsReady(pSel, pThis->pSessions[iTCPSess]->pStrm, NSDSEL_RD, &bIsReady, &nfds);
+			if(bIsReady || localRet != RS_RET_OK) {
 				doReceive(pThis, &pThis->pSessions[iTCPSess], NULL);
 				--nfds; /* indicate we have processed one */
 			}
@@ -618,7 +619,9 @@ finalize_it: /* this is a very special case - this time only we do not exit the
 	      * crashed, which made sense (the rest of the engine was not prepared for
 	      * that) -- rgerhards, 2008-05-19
 	      */
-		/*EMPTY*/;
+		if(pSel != NULL) { /* cleanup missing? happens during err exit! */
+			nssel.Destruct(&pSel);
+		}
 	}
 
 	/* note that this point is usually not reached */
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 930aa30..77381d2 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -18,6 +18,7 @@ TESTS +=  \
 	imtcp_conndrop.sh \
 	imtcp_addtlframedelim.sh \
 	sndrcv.sh \
+	sndrcv_failover.sh \
 	sndrcv_gzip.sh \
 	sndrcv_udp.sh \
 	sndrcv_udp_nonstdpt.sh \
@@ -52,6 +53,8 @@ TESTS +=  \
 	discard-rptdmsg.sh \
 	discard-allmark.sh \
 	discard.sh \
+	failover-async.sh \
+	failover-double.sh \
 	failover-basic.sh \
 	failover-rptd.sh \
 	failover-no-rptd.sh \
@@ -96,6 +99,13 @@ TESTS +=  \
 	imptcp_conndrop.sh 
 endif
 
+if ENABLE_GNUTLS
+if HAVE_VALGRIND
+# This test does not work on v5 as we keep DH params
+#TESTS += manytcp-too-few-tls.sh
+endif
+endif
+
 if ENABLE_OMUXSOCK
 TESTS += uxsock_simple.sh
 endif
@@ -279,6 +289,10 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \
 	   failover-basic.sh \
 	   failover-basic-vg.sh \
 	   testsuites/failover-basic.conf \
+	   failover-async.sh \
+	   testsuites/failover-async.conf \
+	   failover-double.sh \
+	   testsuites/failover-double.conf \
 	   discard-rptdmsg.sh \
 	   discard-rptdmsg-vg.sh \
 	   testsuites/discard-rptdmsg.conf \
@@ -301,6 +315,10 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \
 	   threadingmqaq.sh \
 	   testsuites/threadingmqaq.conf \
 	   sndrcv_drvr.sh \
+	   sndrcv_drvr_noexit.sh \
+	   sndrcv_failover.sh \
+	   testsuites/sndrcv_failover_sender.conf \
+	   testsuites/sndrcv_failover_rcvr.conf \
 	   sndrcv.sh \
 	   testsuites/sndrcv_sender.conf \
 	   testsuites/sndrcv_rcvr.conf \
@@ -426,7 +444,8 @@ uxsockrcvr_SOURCES = uxsockrcvr.c
 uxsockrcvr_LDADD = $(SOL_LIBS)
 
 tcpflood_SOURCES = tcpflood.c
-tcpflood_LDADD = $(SOL_LIBS)
+tcpflood_CPPFLAGS =  $(PTHREADS_CFLAGS)
+tcpflood_LDADD = $(SOL_LIBS) $(PTHREADS_LIBS)
 
 syslog_caller_SOURCES = syslog_caller.c
 syslog_caller_LDADD = $(SOL_LIBS)
diff --git a/tests/Makefile.in b/tests/Makefile.in
index 8b629e5..9acbda9 100644
--- a/tests/Makefile.in
+++ b/tests/Makefile.in
@@ -62,6 +62,7 @@ host_triplet = @host@
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	imtcp_conndrop.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	imtcp_addtlframedelim.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	sndrcv.sh \
+@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	sndrcv_failover.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	sndrcv_gzip.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	sndrcv_udp.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	sndrcv_udp_nonstdpt.sh \
@@ -96,6 +97,8 @@ host_triplet = @host@
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	discard-rptdmsg.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	discard-allmark.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	discard.sh \
+@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-async.sh \
+@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-double.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-basic.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-rptd.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-no-rptd.sh \
@@ -130,6 +133,9 @@ host_triplet = @host@
 @ENABLE_IMPTCP_TRUE@@ENABLE_TESTBENCH_TRUE@	imptcp_addtlframedelim.sh \
 @ENABLE_IMPTCP_TRUE@@ENABLE_TESTBENCH_TRUE@	imptcp_conndrop.sh 
 
+
+# This test does not work on v5 as we keep DH params
+#TESTS += manytcp-too-few-tls.sh
 @ENABLE_OMUXSOCK_TRUE@@ENABLE_TESTBENCH_TRUE@am__append_7 = uxsock_simple.sh
 @ENABLE_OMUDPSPOOF_TRUE@@ENABLE_TESTBENCH_TRUE@am__append_8 = sndrcv_omudpspoof.sh \
 @ENABLE_OMUDPSPOOF_TRUE@@ENABLE_TESTBENCH_TRUE@	 sndrcv_omudpspoof_nonstdpt.sh
@@ -217,9 +223,9 @@ syslog_caller_DEPENDENCIES = $(am__DEPENDENCIES_1)
 am_syslog_inject_OBJECTS = syslog_inject.$(OBJEXT)
 syslog_inject_OBJECTS = $(am_syslog_inject_OBJECTS)
 syslog_inject_DEPENDENCIES = $(am__DEPENDENCIES_1)
-am_tcpflood_OBJECTS = tcpflood.$(OBJEXT)
+am_tcpflood_OBJECTS = tcpflood-tcpflood.$(OBJEXT)
 tcpflood_OBJECTS = $(am_tcpflood_OBJECTS)
-tcpflood_DEPENDENCIES = $(am__DEPENDENCIES_1)
+tcpflood_DEPENDENCIES = $(am__DEPENDENCIES_1) $(am__DEPENDENCIES_1)
 am_uxsockrcvr_OBJECTS = uxsockrcvr.$(OBJEXT)
 uxsockrcvr_OBJECTS = $(am_uxsockrcvr_OBJECTS)
 uxsockrcvr_DEPENDENCIES = $(am__DEPENDENCIES_1)
@@ -555,6 +561,10 @@ EXTRA_DIST = 1.rstest 2.rstest 3.rstest err1.rstest \
 	   failover-basic.sh \
 	   failover-basic-vg.sh \
 	   testsuites/failover-basic.conf \
+	   failover-async.sh \
+	   testsuites/failover-async.conf \
+	   failover-double.sh \
+	   testsuites/failover-double.conf \
 	   discard-rptdmsg.sh \
 	   discard-rptdmsg-vg.sh \
 	   testsuites/discard-rptdmsg.conf \
@@ -577,6 +587,10 @@ EXTRA_DIST = 1.rstest 2.rstest 3.rstest err1.rstest \
 	   threadingmqaq.sh \
 	   testsuites/threadingmqaq.conf \
 	   sndrcv_drvr.sh \
+	   sndrcv_drvr_noexit.sh \
+	   sndrcv_failover.sh \
+	   testsuites/sndrcv_failover_sender.conf \
+	   testsuites/sndrcv_failover_rcvr.conf \
 	   sndrcv.sh \
 	   testsuites/sndrcv_sender.conf \
 	   testsuites/sndrcv_rcvr.conf \
@@ -700,7 +714,8 @@ chkseq_SOURCES = chkseq.c
 uxsockrcvr_SOURCES = uxsockrcvr.c
 uxsockrcvr_LDADD = $(SOL_LIBS)
 tcpflood_SOURCES = tcpflood.c
-tcpflood_LDADD = $(SOL_LIBS)
+tcpflood_CPPFLAGS = $(PTHREADS_CFLAGS)
+tcpflood_LDADD = $(SOL_LIBS) $(PTHREADS_LIBS)
 syslog_caller_SOURCES = syslog_caller.c
 syslog_caller_LDADD = $(SOL_LIBS)
 syslog_inject_SOURCES = syslog_inject.c
@@ -825,7 +840,7 @@ distclean-compile:
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/rt_init-runtime-dummy.Po@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/syslog_caller.Po@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/syslog_inject.Po@am__quote@
-@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/tcpflood.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/tcpflood-tcpflood.Po@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/uxsockrcvr.Po@am__quote@
 
 .c.o:
@@ -932,6 +947,22 @@ rt_init-runtime-dummy.obj: runtime-dummy.c
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
 @am__fastdepCC_FALSE@	$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(rt_init_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -c -o rt_init-runtime-dummy.obj `if test -f 'runtime-dummy.c'; then $(CYGPATH_W) 'runtime-dummy.c'; else $(CYGPATH_W) '$(srcdir)/runtime-dummy.c'; fi`
 
+tcpflood-tcpflood.o: tcpflood.c
+@am__fastdepCC_TRUE@	$(AM_V_CC)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(tcpflood_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -MT tcpflood-tcpflood.o -MD -MP -MF $(DEPDIR)/tcpflood-tcpflood.Tpo -c -o tcpflood-tcpflood.o `test -f 'tcpflood.c' || echo '$(srcdir)/'`tcpflood.c
+@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/tcpflood-tcpflood.Tpo $(DEPDIR)/tcpflood-tcpflood.Po
+@am__fastdepCC_FALSE@	$(AM_V_CC) @AM_BACKSLASH@
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	source='tcpflood.c' object='tcpflood-tcpflood.o' libtool=no @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCC_FALSE@	$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(tcpflood_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -c -o tcpflood-tcpflood.o `test -f 'tcpflood.c' || echo '$(srcdir)/'`tcpflood.c
+
+tcpflood-tcpflood.obj: tcpflood.c
+@am__fastdepCC_TRUE@	$(AM_V_CC)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(tcpflood_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -MT tcpflood-tcpflood.obj -MD -MP -MF $(DEPDIR)/tcpflood-tcpflood.Tpo -c -o tcpflood-tcpflood.obj `if test -f 'tcpflood.c'; then $(CYGPATH_W) 'tcpflood.c'; else $(CYGPATH_W) '$(srcdir)/tcpflood.c'; fi`
+@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/tcpflood-tcpflood.Tpo $(DEPDIR)/tcpflood-tcpflood.Po
+@am__fastdepCC_FALSE@	$(AM_V_CC) @AM_BACKSLASH@
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	source='tcpflood.c' object='tcpflood-tcpflood.obj' libtool=no @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCC_FALSE@	$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(tcpflood_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -c -o tcpflood-tcpflood.obj `if test -f 'tcpflood.c'; then $(CYGPATH_W) 'tcpflood.c'; else $(CYGPATH_W) '$(srcdir)/tcpflood.c'; fi`
+
 mostlyclean-libtool:
 	-rm -f *.lo
 
diff --git a/tests/diag.sh b/tests/diag.sh
index d1242fb..617118b 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 nologfuncflow nostdout"
+#export RSYSLOG_DEBUG="debug nologfuncflow noprintmutexaction stdout"
 #export RSYSLOG_DEBUGLOG="log"
 case $1 in
    'init')	$srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason
@@ -22,7 +22,7 @@ case $1 in
 		rm -f work rsyslog.out.log rsyslog2.out.log rsyslog.out.log.save # common work files
 		rm -rf test-spool test-logdir
 		rm -f rsyslog.out.*.log work-presort rsyslog.pipe
-		rm -f rsyslog.input
+		rm -f rsyslog.input rsyslog.empty
 		rm -f core.* vgcore.*
 		mkdir test-spool
 		;;
@@ -31,7 +31,7 @@ case $1 in
 		rm -f work rsyslog.out.log rsyslog2.out.log rsyslog.out.log.save # common work files
 		rm -rf test-spool test-logdir 
 		rm -f rsyslog.out.*.log rsyslog.random.data work-presort rsyslog.pipe
-		rm -f rsyslog.input stat-file1
+		rm -f rsyslog.input stat-file1 #rsyslog.empty
 		echo  -------------------------------------------------------------------------------
 		;;
    'startup')   # start rsyslogd with default params. $2 is the config file name to use
diff --git a/tests/failover-async.sh b/tests/failover-async.sh
new file mode 100755
index 0000000..5fc393d
--- /dev/null
+++ b/tests/failover-async.sh
@@ -0,0 +1,12 @@
+# This file is part of the rsyslog project, released under GPLv3
+echo ===============================================================================
+echo \[failover-async.sh\]: async test for failover functionality
+source $srcdir/diag.sh init
+source $srcdir/diag.sh startup failover-async.conf
+source $srcdir/diag.sh injectmsg  0 5000
+echo doing shutdown
+source $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+source $srcdir/diag.sh wait-shutdown 
+source $srcdir/diag.sh seq-check  0 4999
+source $srcdir/diag.sh exit
diff --git a/tests/failover-double.sh b/tests/failover-double.sh
new file mode 100755
index 0000000..172b91d
--- /dev/null
+++ b/tests/failover-double.sh
@@ -0,0 +1,12 @@
+# This file is part of the rsyslog project, released under GPLv3
+echo ===============================================================================
+echo \[failover-double.sh\]: test for double failover functionality
+source $srcdir/diag.sh init
+source $srcdir/diag.sh startup failover-double.conf
+source $srcdir/diag.sh injectmsg  0 5000
+echo doing shutdown
+source $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+source $srcdir/diag.sh wait-shutdown 
+source $srcdir/diag.sh seq-check  0 4999
+source $srcdir/diag.sh exit
diff --git a/tests/sndrcv_drvr.sh b/tests/sndrcv_drvr.sh
index 3d61306..f909264 100755
--- a/tests/sndrcv_drvr.sh
+++ b/tests/sndrcv_drvr.sh
@@ -1,50 +1,2 @@
-# This is test driver for testing two rsyslog instances. It can be
-# utilized by any test that just needs two instances with different
-# config files, where messages are injected in instance TWO and 
-# (with whatever rsyslog mechanism) being relayed over to instance ONE,
-# where they are written to the log file. After the run, the completeness
-# of that log file is checked.
-# The code is almost the same, but the config files differ (probably greatly)
-# for different test cases. As such, this driver needs to be called with the
-# config file name ($2). From that name, the sender and receiver config file
-# names are automatically generated. 
-# So: $1 config file name, $2 number of messages
-#
-# A note on TLS testing: the current testsuite (in git!) already contains
-# TLS test cases. However, getting these test cases correct is not simple.
-# That's not a problem with the code itself, but rater a problem with
-# synchronization in the test environment. So I have deciced to keep the
-# TLS tests in, but not yet actually utilize them. This is most probably
-# left as an excercise for future (devel) releases. -- rgerhards, 2009-11-11
-#
-# added 2009-11-11 by Rgerhards
-# This file is part of the rsyslog project, released  under GPLv3
-# uncomment for debugging support:
-source $srcdir/diag.sh init
-# start up the instances
-#export RSYSLOG_DEBUG="debug nostdout noprintmutexaction"
-#export RSYSLOG_DEBUGLOG="log"
-source $srcdir/diag.sh startup $1_rcvr.conf 
-source $srcdir/diag.sh wait-startup
-#export RSYSLOG_DEBUGLOG="log2"
-#valgrind="valgrind"
-source $srcdir/diag.sh startup $1_sender.conf 2
-source $srcdir/diag.sh wait-startup 2
-# may be needed by TLS (once we do it): sleep 30
-
-# now inject the messages into instance 2. It will connect to instance 1,
-# and that instance will record the data.
-source $srcdir/diag.sh tcpflood -m$2 -i1
-sleep 2 # make sure all data is received in input buffers
-# shut down sender when everything is sent, receiver continues to run concurrently
-# may be needed by TLS (once we do it): sleep 60
-source $srcdir/diag.sh shutdown-when-empty 2
-source $srcdir/diag.sh wait-shutdown 2
-# now it is time to stop the receiver as well
-source $srcdir/diag.sh shutdown-when-empty
-source $srcdir/diag.sh wait-shutdown
-
-# may be needed by TLS (once we do it): sleep 60
-# do the final check
-source $srcdir/diag.sh seq-check 1 $2
+source $srcdir/sndrcv_drvr_noexit.sh $1 $2
 source $srcdir/diag.sh exit
diff --git a/tests/sndrcv_drvr_noexit.sh b/tests/sndrcv_drvr_noexit.sh
new file mode 100755
index 0000000..899eace
--- /dev/null
+++ b/tests/sndrcv_drvr_noexit.sh
@@ -0,0 +1,49 @@
+# This is test driver for testing two rsyslog instances. It can be
+# utilized by any test that just needs two instances with different
+# config files, where messages are injected in instance TWO and 
+# (with whatever rsyslog mechanism) being relayed over to instance ONE,
+# where they are written to the log file. After the run, the completeness
+# of that log file is checked.
+# The code is almost the same, but the config files differ (probably greatly)
+# for different test cases. As such, this driver needs to be called with the
+# config file name ($2). From that name, the sender and receiver config file
+# names are automatically generated. 
+# So: $1 config file name, $2 number of messages
+#
+# A note on TLS testing: the current testsuite (in git!) already contains
+# TLS test cases. However, getting these test cases correct is not simple.
+# That's not a problem with the code itself, but rater a problem with
+# synchronization in the test environment. So I have deciced to keep the
+# TLS tests in, but not yet actually utilize them. This is most probably
+# left as an excercise for future (devel) releases. -- rgerhards, 2009-11-11
+#
+# added 2009-11-11 by Rgerhards
+# This file is part of the rsyslog project, released  under GPLv3
+# uncomment for debugging support:
+source $srcdir/diag.sh init
+# start up the instances
+#export RSYSLOG_DEBUG="debug nostdout noprintmutexaction"
+#export RSYSLOG_DEBUGLOG="log"
+source $srcdir/diag.sh startup $1_rcvr.conf 
+source $srcdir/diag.sh wait-startup
+#export RSYSLOG_DEBUGLOG="log2"
+#valgrind="valgrind"
+source $srcdir/diag.sh startup $1_sender.conf 2
+source $srcdir/diag.sh wait-startup 2
+# may be needed by TLS (once we do it): sleep 30
+
+# now inject the messages into instance 2. It will connect to instance 1,
+# and that instance will record the data.
+source $srcdir/diag.sh tcpflood -m$2 -i1
+sleep 2 # make sure all data is received in input buffers
+# shut down sender when everything is sent, receiver continues to run concurrently
+# may be needed by TLS (once we do it): sleep 60
+source $srcdir/diag.sh shutdown-when-empty 2
+source $srcdir/diag.sh wait-shutdown 2
+# now it is time to stop the receiver as well
+source $srcdir/diag.sh shutdown-when-empty
+source $srcdir/diag.sh wait-shutdown
+
+# may be needed by TLS (once we do it): sleep 60
+# do the final check
+source $srcdir/diag.sh seq-check 1 $2
diff --git a/tests/sndrcv_failover.sh b/tests/sndrcv_failover.sh
new file mode 100755
index 0000000..4c5e183
--- /dev/null
+++ b/tests/sndrcv_failover.sh
@@ -0,0 +1,21 @@
+# This tests failover capabilities. Data is sent to local port 13516, where
+# no process shall listen. Then it fails over to a second instance, then to
+# a file. The second instance is started. So all data should be received
+# there and none be logged to the file.
+# This builds on the basic sndrcv.sh test, but adds a first, failing,
+# location to the conf file.
+# added 2011-06-20 by Rgerhards
+# This file is part of the rsyslog project, released  under GPLv3
+echo ===============================================================================
+echo \[sndrcv_failover.sh\]: testing failover capabilities for tcp sending
+source $srcdir/sndrcv_drvr_noexit.sh sndrcv_failover 50000
+ls -l rsyslog.empty
+if [[ -s rsyslog.empty ]] ; then
+  echo "FAIL: rsyslog.empty has data. Failover handling failed. Data is written"
+  echo "      even though the previous action (in a failover chain!) properly"
+  echo "      worked."
+  exit 1
+else
+  echo "rsyslog.empty is empty - OK"
+fi ;
+source $srcdir/diag.sh exit
diff --git a/tests/tcpflood.c b/tests/tcpflood.c
index 3e7053c..49b1e9e 100644
--- a/tests/tcpflood.c
+++ b/tests/tcpflood.c
@@ -16,6 +16,7 @@
  *      bytes as forth. Add -r to randomize the amount of extra
  *      data included in the range 1..(value of -d).
  * -r	randomize amount of extra data added (-d must be > 0)
+ * -s	(silent) do not show progress indicator (never done on non-tty)
  * -f	support for testing dynafiles. If given, include a dynafile ID
  *      in the range 0..(f-1) as the SECOND field, shifting all field values
  *      one field to the right. Zero (default) disables this functionality.
@@ -32,6 +33,24 @@
  * -D	randomly drop and re-establish connections. Useful for stress-testing
  *      the TCP receiver.
  * -F	USASCII value for frame delimiter (in octet-stuffing mode), default LF
+ * -R	number of times the test shall be run (very useful for gathering performance
+ *      data and other repetitive things). Default: 1
+ * -S   number of seconds to sleep between different runs (-R) Default: 30
+ * -X   generate sTats data records. Default: off
+ * -e   encode output in CSV (not yet everywhere supported)
+ *      for performance data:
+ *      each inidividual line has the runtime of one test
+ *      the last line has 0 in field 1, followed by numberRuns,TotalRuntime,
+ *      Average,min,max
+ * -T   transport to use. Currently supported: "udp", "tcp" (default)
+ *      Note: UDP supports a single target port, only
+ * -W	wait time between sending batches of messages, in microseconds (Default: 0)
+ * -b   number of messages within a batch (default: 100,000,000 millions)
+ * -Y	use multiple threads, one per connection (which means 1 if one only connection
+ *  	is configured!)
+ * -z	private key file for TLS mode
+ * -Z	cert (public key) file for TLS mode
+ * -L	loglevel to use for GnuTLS troubleshooting (0-off to 10-all, 0 default)
  *
  * Part of the testbench for rsyslog.
  *
@@ -66,7 +85,15 @@
 #include <unistd.h>
 #include <string.h>
 #include <netinet/in.h>
+#include <pthread.h>
 #include <sys/resource.h>
+#include <sys/time.h>
+#include <errno.h>
+#ifdef ENABLE_GNUTLS
+#	include <gnutls/gnutls.h>
+#	include <gcrypt.h>
+	GCRY_THREAD_OPTION_PTHREAD_IMPL;
+#endif
 
 #define EXIT_FAILURE 1
 #define INVALID_SOCKET -1
@@ -74,6 +101,7 @@
 #define NETTEST_INPUT_CONF_FILE "nettest.input.conf" /* name of input file, must match $IncludeConfig in .conf files */
 
 #define MAX_EXTRADATA_LEN 100*1024
+#define MAX_SENDBUF 2 * MAX_EXTRADATA_LEN
 
 static char *targetIP = "127.0.0.1";
 static char *msgPRI = "167";
@@ -83,10 +111,11 @@ static int dynFileIDs = 0;
 static int extraDataLen = 0; /* amount of extra data to add to message */
 static int bRandomizeExtraData = 0; /* randomize amount of extra data added */
 static int numMsgsToSend; /* number of messages to send */
-static int numConnections = 1; /* number of connections to create */
+static unsigned numConnections = 1; /* number of connections to create */
 static int *sockArray;  /* array of sockets to use */
 static int msgNum = 0;	/* initial message number to start with */
 static int bShowProgress = 1; /* show progress messages */
+static int bSilent = 0; /* completely silent operation */
 static int bRandConnDrop = 0; /* randomly drop connections? */
 static char *MsgToSend = NULL; /* if non-null, this is the actual message to send */
 static int bBinaryFile = 0;	/* is -I file binary */
@@ -95,6 +124,75 @@ static int numFileIterations = 1;/* how often is file data to be sent? */
 static char frameDelim = '\n';	/* default frame delimiter */
 FILE *dataFP = NULL;		/* file pointer for data file, if used */
 static long nConnDrops = 0;	/* counter: number of time connection was dropped (-D option) */
+static int numRuns = 1;		/* number of times the test shall be run */
+static int sleepBetweenRuns = 30; /* number of seconds to sleep between runs */
+static int bStatsRecords = 0;	/* generate stats records */
+static int bCSVoutput = 0;	/* generate output in CSV (where applicable) */
+static long long batchsize = 100000000ll;
+static int waittime = 0;
+static int runMultithreaded = 0; /* run tests in multithreaded mode */
+static int numThrds = 1;	/* number of threads to use */
+static char *tlsCertFile = NULL;
+static char *tlsKeyFile = NULL;
+static int tlsLogLevel = 0;
+
+#ifdef ENABLE_GNUTLS
+static gnutls_session_t *sessArray;	/* array of TLS sessions to use */
+static gnutls_certificate_credentials tlscred;
+#endif
+
+/* variables for managing multi-threaded operations */
+int runningThreads;		/* number of threads currently running */
+int doRun;			/* shall sender thread begin to run? */
+pthread_mutex_t thrdMgmt;	/* mutex for controling startup/shutdown */
+pthread_cond_t condStarted;
+pthread_cond_t condDoRun;
+
+/* the following struct provides information for a generator instance (thread) */
+struct instdata {
+	/* lower and upper bounds for the thread in question */
+	unsigned long long lower;
+	unsigned long long numMsgs; /* number of messages to send */
+	unsigned long long numSent; /* number of messages already sent */
+	unsigned idx;	/**< index of fd to be used for sending */
+	pthread_t thread; /**< thread processing this instance */
+} *instarray = NULL;
+
+/* the following structure is used to gather performance data */
+struct runstats {
+	unsigned long long totalRuntime;
+	unsigned long minRuntime;
+	unsigned long maxRuntime;
+	int numRuns;
+};
+
+static int udpsock;			/* socket for sending in UDP mode */
+static struct sockaddr_in udpRcvr;	/* remote receiver in UDP mode */
+
+static enum { TP_UDP, TP_TCP, TP_TLS } transport = TP_TCP;
+
+/* forward definitions */
+static void initTLSSess(int);
+static int sendTLS(int i, char *buf, int lenBuf);
+static void closeTLSSess(int __attribute__((unused)) i);
+
+/* prepare send subsystem for UDP send */
+static inline int
+setupUDP(void)
+{
+	if((udpsock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1)
+		return 1;
+
+	memset((char *) &udpRcvr, 0, sizeof(udpRcvr));
+	udpRcvr.sin_family = AF_INET;
+	udpRcvr.sin_port = htons(targetPort);
+	if(inet_aton(targetIP, &udpRcvr.sin_addr)==0) {
+		fprintf(stderr, "inet_aton() failed\n");
+		return(1);
+	}
+
+	return 0;
+}
 
 
 /* open a single tcp connection
@@ -150,12 +248,18 @@ int openConn(int *fd)
  */
 int openConnections(void)
 {
-	int i;
+	unsigned i;
 	char msgBuf[128];
 	size_t lenMsg;
 
+	if(transport == TP_UDP)
+		return setupUDP();
+
 	if(bShowProgress)
 		write(1, "      open connections", sizeof("      open connections")-1);
+#	ifdef ENABLE_GNUTLS
+	sessArray = calloc(numConnections, sizeof(gnutls_session_t));
+#	endif
 	sockArray = calloc(numConnections, sizeof(int));
 	for(i = 0 ; i < numConnections ; ++i) {
 		if(i % 10 == 0) {
@@ -166,9 +270,14 @@ int openConnections(void)
 			printf("error in trying to open connection i=%d\n", i);
 			return 1;
 		}
+		if(transport == TP_TLS) {
+			initTLSSess(i);
+		}
+	}
+	if(bShowProgress) {
+		lenMsg = sprintf(msgBuf, "\r%5.5d open connections\n", i);
+		write(1, msgBuf, lenMsg);
 	}
-	lenMsg = sprintf(msgBuf, "\r%5.5d open connections\n", i);
-	write(1, msgBuf, lenMsg);
 
 	return 0;
 }
@@ -183,11 +292,14 @@ int openConnections(void)
  */
 void closeConnections(void)
 {
-	int i;
+	unsigned i;
 	size_t lenMsg;
 	struct linger ling;
 	char msgBuf[128];
 
+	if(transport == TP_UDP)
+		return;
+
 	if(bShowProgress)
 		write(1, "      close connections", sizeof("      close connections")-1);
 	for(i = 0 ; i < numConnections ; ++i) {
@@ -204,11 +316,15 @@ void closeConnections(void)
 			ling.l_onoff = 1;
 			ling.l_linger = 1;
 			setsockopt(sockArray[i], SOL_SOCKET, SO_LINGER, &ling, sizeof(ling));
+			if(transport == TP_TLS)
+				closeTLSSess(i);
 			close(sockArray[i]);
 		}
 	}
-	lenMsg = sprintf(msgBuf, "\r%5.5d close connections\n", i);
-	write(1, msgBuf, lenMsg);
+	if(bShowProgress) {
+		lenMsg = sprintf(msgBuf, "\r%5.5d close connections\n", i);
+		write(1, msgBuf, lenMsg);
+	}
 
 }
 
@@ -218,12 +334,11 @@ void closeConnections(void)
  * of constructing test messages. -- rgerhards, 2010-03-31
  */
 static inline void
-genMsg(char *buf, size_t maxBuf, int *pLenBuf)
+genMsg(char *buf, size_t maxBuf, int *pLenBuf, struct instdata *inst)
 {
 	int edLen; /* actual extra data length to use */
 	char extraData[MAX_EXTRADATA_LEN + 1];
 	char dynFileIDBuf[128] = "";
-	static int numMsgsGen = 0;
 	int done;
 
 	if(dataFP != NULL) {
@@ -262,11 +377,9 @@ genMsg(char *buf, size_t maxBuf, int *pLenBuf)
 		/* use fixed message format from command line */
 		*pLenBuf = snprintf(buf, maxBuf, "%s\n", MsgToSend);
 	}
+	++inst->numSent;
 
-	if(numMsgsGen++ >= numMsgsToSend)
-		*pLenBuf = 0; /* indicate end of run */
-
-finalize_it: ;
+finalize_it: /*EMPTY to keep the compiler happy */;
 }
 
 /* send messages to the tcp connections we keep open. We use
@@ -277,50 +390,72 @@ finalize_it: ;
  * last. All messages in between are sent over random connections.
  * Note that message numbers start at 0.
  */
-int sendMessages(void)
+int sendMessages(struct instdata *inst)
 {
-	int i = 0;
+	unsigned i = 0;
 	int socknum;
 	int lenBuf;
-	int lenSend;
-	char *statusText;
+	int lenSend = 0;
+	char *statusText = "";
 	char buf[MAX_EXTRADATA_LEN + 1024];
+	char sendBuf[MAX_SENDBUF];
+	int offsSendBuf = 0;
 
-	if(dataFile == NULL) {
-		printf("Sending %d messages.\n", numMsgsToSend);
-		statusText = "messages";
-	} else {
-		printf("Sending file '%s' %d times.\n", dataFile, numFileIterations);
-		statusText = "kb";
+	if(!bSilent) {
+		if(dataFile == NULL) {
+			printf("Sending %llu messages.\n", inst->numMsgs);
+			statusText = "messages";
+		} else {
+			printf("Sending file '%s' %d times.\n", dataFile,
+			       numFileIterations);
+			statusText = "kb";
+		}
 	}
 	if(bShowProgress)
 		printf("\r%8.8d %s sent", 0, statusText);
-	while(1) { /* broken inside loop! */
-		if(i < numConnections)
-			socknum = i;
-		else if(i >= numMsgsToSend - numConnections)
-			socknum = i - (numMsgsToSend - numConnections);
-		else {
-			int rnd = rand();
-			socknum = rnd % numConnections;
+	while(i < inst->numMsgs) {
+		if(runMultithreaded) {
+			socknum = inst->idx;
+		} else {
+			if(i < numConnections)
+				socknum = i;
+			else if(i >= inst->numMsgs - numConnections) {
+				socknum = i - (inst->numMsgs - numConnections);
+			} else {
+				int rnd = rand();
+				socknum = rnd % numConnections;
+			}
 		}
-		genMsg(buf, sizeof(buf), &lenBuf); /* generate the message to send according to params */
-		if(lenBuf == 0)
-			break; /* end of processing! */
-		if(sockArray[socknum] == -1) {
-			/* connection was dropped, need to re-establish */
-			if(openConn(&(sockArray[socknum])) != 0) {
-				printf("error in trying to re-open connection %d\n", socknum);
-				exit(1);
+		genMsg(buf, sizeof(buf), &lenBuf, inst); /* generate the message to send according to params */
+		if(transport == TP_TCP) {
+			if(sockArray[socknum] == -1) {
+				/* connection was dropped, need to re-establish */
+				if(openConn(&(sockArray[socknum])) != 0) {
+					printf("error in trying to re-open connection %d\n", socknum);
+					exit(1);
+				}
+			}
+			lenSend = send(sockArray[socknum], buf, lenBuf, 0);
+		} else if(transport == TP_UDP) {
+			lenSend = sendto(udpsock, buf, lenBuf, 0, &udpRcvr, sizeof(udpRcvr));
+		} else if(transport == TP_TLS) {
+			if(offsSendBuf + lenBuf < MAX_SENDBUF) {
+				memcpy(sendBuf+offsSendBuf, buf, lenBuf);
+				offsSendBuf += lenBuf;
+				lenSend = lenBuf; /* simulate "good" call */
+			} else {
+				lenSend = sendTLS(socknum, sendBuf, offsSendBuf);
+				lenSend = (lenSend == offsSendBuf) ? lenBuf : -1;
+				memcpy(sendBuf, buf, lenBuf);
+				offsSendBuf = lenBuf;
 			}
 		}
-		lenSend = send(sockArray[socknum], buf, lenBuf, 0);
 		if(lenSend != lenBuf) {
 			printf("\r%5.5d\n", i);
 			fflush(stdout);
 			perror("send test data");
-			printf("send() failed at socket %d, index %d, msgNum %d\n",
-				sockArray[socknum], i, msgNum);
+			printf("send() failed at socket %d, index %d, msgNum %lld\n",
+				sockArray[socknum], i, inst->numSent);
 			fflush(stderr);
 			return(1);
 		}
@@ -328,7 +463,7 @@ int sendMessages(void)
 			if(bShowProgress)
 				printf("\r%8.8d", i);
 		}
-		if(bRandConnDrop) {
+		if(!runMultithreaded && bRandConnDrop) {
 			/* if we need to randomly drop connections, see if we 
 			 * are a victim
 			 */
@@ -338,14 +473,333 @@ int sendMessages(void)
 				sockArray[socknum] = -1;
 			}
 		}
+		if(inst->numSent % batchsize == 0) {
+			usleep(waittime);
+		}
 		++msgNum;
 		++i;
 	}
-	printf("\r%8.8d %s sent\n", i, statusText);
+	if(transport == TP_TLS && offsSendBuf != 0) {
+		/* send remaining buffer */
+		lenSend = sendTLS(socknum, sendBuf, offsSendBuf);
+	}
+	if(!bSilent)
+		printf("\r%8.8d %s sent\n", i, statusText);
+
+	return 0;
+}
+
+
+/* this is the thread that starts a generator
+ */
+static void *
+thrdStarter(void *arg)
+{
+	struct instdata *inst = (struct instdata*) arg;
+	pthread_mutex_lock(&thrdMgmt);
+	runningThreads++;
+	pthread_cond_signal(&condStarted);
+	while(doRun == 0) {
+		pthread_cond_wait(&condDoRun, &thrdMgmt);
+	}
+	pthread_mutex_unlock(&thrdMgmt);
+	if(sendMessages(inst) != 0) {
+		printf("error sending messages\n");
+	}
+	return NULL;
+}
+
+
+/* This function initializes the actual traffic generators. The function sets up all required
+ * parameter blocks and starts threads. It returns when all threads are ready to run
+ * and the main task must just enable them.
+ */
+static inline void
+prepareGenerators()
+{
+	int i;
+	long long msgsThrd;
+	long long starting = 0;
+	
+	if(runMultithreaded) {
+		bSilent = 1;
+		numThrds = numConnections;
+	} else {
+		numThrds = 1;
+	}
+
+	runningThreads = 0;
+	doRun = 0;
+	pthread_mutex_init(&thrdMgmt, NULL);
+	pthread_cond_init(&condStarted, NULL);
+	pthread_cond_init(&condDoRun, NULL);
+
+	if(instarray != NULL) {
+		free(instarray);
+	}
+	instarray = calloc(numThrds, sizeof(struct instdata));
+	msgsThrd = numMsgsToSend / numThrds;
+
+	for(i = 0 ; i < numThrds ; ++i)  {
+		instarray[i].lower = starting;
+		instarray[i].numMsgs = msgsThrd;
+		instarray[i].numSent = 0;
+		instarray[i].idx = i;
+		pthread_create(&(instarray[i].thread), NULL, thrdStarter, instarray + i); 
+		/*printf("started thread %x\n", (unsigned) instarray[i].thread);*/
+		starting += msgsThrd;
+	}
+}
+
+/* Let all generators run. Threads must have been started. Here we wait until
+ * all threads are initialized and then broadcast that they can begin to run.
+ */
+static inline void
+runGenerators()
+{
+	pthread_mutex_lock(&thrdMgmt);
+	while(runningThreads != numThrds){
+		pthread_cond_wait(&condStarted, &thrdMgmt);
+	}
+	doRun = 1;
+	pthread_cond_broadcast(&condDoRun);
+	pthread_mutex_unlock(&thrdMgmt);
+}
+
+
+/* Wait for all traffic generators to stop.
+ */
+static inline void
+waitGenerators()
+{
+	int i;
+	for(i = 0 ; i < numThrds ; ++i)  {
+		pthread_join(instarray[i].thread, NULL);
+		/*printf("thread %x stopped\n", (unsigned) instarray[i].thread);*/
+	}
+	pthread_mutex_destroy(&thrdMgmt);
+	pthread_cond_destroy(&condStarted);
+	pthread_cond_destroy(&condDoRun);
+}
+
+/* functions related to computing statistics on the runtime of a test. This is
+ * a separate function primarily not to mess up the test driver.
+ * rgerhards, 2010-12-08
+ */
+static inline void
+endTiming(struct timeval *tvStart, struct runstats *stats)
+{
+	long sec, usec;
+	unsigned long runtime;
+	struct timeval tvEnd;
+
+	gettimeofday(&tvEnd, NULL);
+	if(tvStart->tv_usec > tvEnd.tv_usec) {
+		tvEnd.tv_sec--;
+		tvEnd.tv_usec += 1000000;
+	}
+
+	sec = tvEnd.tv_sec - tvStart->tv_sec;
+	usec = tvEnd.tv_usec - tvStart->tv_usec;
+
+	runtime = sec * 1000 + (usec / 1000);
+	stats->totalRuntime += runtime;
+	if(runtime < stats->minRuntime)
+		stats->minRuntime = runtime;
+	if(runtime > stats->maxRuntime)
+		stats->maxRuntime = runtime;
+
+	if(!bSilent || bStatsRecords) {
+		if(bCSVoutput) {
+			printf("%ld.%3.3ld\n", runtime / 1000, runtime % 1000);
+		} else {
+			printf("runtime: %ld.%3.3ld\n", runtime / 1000, runtime % 1000);
+		}
+	}
+}
+
+
+/* generate stats summary record at end of run
+ */
+static inline void
+genStats(struct runstats *stats)
+{
+	long unsigned avg;
+	avg = stats->totalRuntime / stats->numRuns;
+
+	if(bCSVoutput) {
+		printf("#numRuns,TotalRuntime,AvgRuntime,MinRuntime,MaxRuntime\n");
+		printf("%d,%llu.%3.3d,%lu.%3.3lu,%lu.%3.3lu,%lu.%3.3lu\n",
+			stats->numRuns,
+		        stats->totalRuntime / 1000, (int) stats->totalRuntime % 1000,
+		        avg / 1000, avg % 1000,
+		        stats->minRuntime / 1000, stats->minRuntime % 1000,
+		        stats->maxRuntime / 1000, stats->maxRuntime % 1000);
+	} else {
+		printf("Runs:     %d\n",   stats->numRuns);
+		printf("Runtime:\n");
+		printf("  total:  %llu.%3.3d\n", stats->totalRuntime / 1000,
+						 (int) stats->totalRuntime % 1000);
+		printf("  avg:    %lu.%3.3lu\n",  avg / 1000, avg % 1000);
+		printf("  min:    %lu.%3.3lu\n",  stats->minRuntime / 1000, stats->minRuntime % 1000);
+		printf("  max:    %lu.%3.3lu\n",  stats->maxRuntime / 1000, stats->maxRuntime % 1000);
+		printf("All times are wallclock time.\n");
+	}
+}
+
+
+/* Run the actual test. This function handles various meta-parameters, like
+ * a specified number of iterations, performance measurement and so on...
+ * rgerhards, 2010-12-08
+ */
+static int
+runTests(void)
+{
+	struct timeval tvStart;
+	struct runstats stats;
+	int run;
+
+	stats.totalRuntime = 0;
+	stats.minRuntime = (unsigned long long) 0xffffffffffffffffll;
+	stats.maxRuntime = 0;
+	stats.numRuns = numRuns;
+	run = 1;
+	while(1) { /* loop broken inside */
+		if(!bSilent)
+			printf("starting run %d\n", run);
+		prepareGenerators();
+		gettimeofday(&tvStart, NULL);
+		runGenerators();
+		waitGenerators();
+		endTiming(&tvStart, &stats);
+		if(run == numRuns)
+			break;
+		if(!bSilent)
+			printf("sleeping %d seconds before next run\n", sleepBetweenRuns);
+		sleep(sleepBetweenRuns);
+		++run;
+	}
+
+	if(bStatsRecords) {
+		genStats(&stats);
+	}
 
 	return 0;
 }
 
+#	if defined(ENABLE_GNUTLS)
+/* This defines a log function to be provided to GnuTLS. It hopefully
+ * helps us track down hard to find problems.
+ * rgerhards, 2008-06-20
+ */
+static void tlsLogFunction(int level, const char *msg)
+{
+	printf("GnuTLS (level %d): %s", level, msg);
+
+}
+
+
+/* global init GnuTLS
+ */
+static void
+initTLS(void)
+{
+	int r;
+
+	/* order of gcry_control and gnutls_global_init matters! */
+	gcry_control(GCRYCTL_SET_THREAD_CBS, &gcry_threads_pthread);
+	gnutls_global_init();
+	/* set debug mode, if so required by the options */
+	if(tlsLogLevel > 0) {
+		gnutls_global_set_log_function(tlsLogFunction);
+		gnutls_global_set_log_level(tlsLogLevel);
+	}
+
+	r = gnutls_certificate_allocate_credentials(&tlscred);
+	if(r != GNUTLS_E_SUCCESS) {
+		printf("error allocating credentials\n");
+		gnutls_perror(r);
+		exit(1);
+	}
+	r = gnutls_certificate_set_x509_key_file(tlscred, tlsCertFile, tlsKeyFile, GNUTLS_X509_FMT_PEM);
+	if(r != GNUTLS_E_SUCCESS) {
+		printf("error setting certificate files -- have you mixed up key and certificate?\n");
+		printf("If in doubt, try swapping the files in -z/-Z\n");
+		printf("Certifcate is: '%s'\n", tlsCertFile);
+		printf("Key        is: '%s'\n", tlsKeyFile);
+		gnutls_perror(r);
+		r = gnutls_certificate_set_x509_key_file(tlscred, tlsKeyFile, tlsCertFile,
+							 GNUTLS_X509_FMT_PEM);
+		if(r == GNUTLS_E_SUCCESS) {
+			printf("Tried swapping files, this seems to work "
+			       "(but results may be unpredictable!)\n");
+		} else {
+			exit(1);
+		}
+	}
+}
+
+
+static void
+initTLSSess(int i)
+{
+	int r;
+	gnutls_init(sessArray + i, GNUTLS_CLIENT);
+
+	/* Use default priorities */
+	gnutls_set_default_priority(sessArray[i]);
+
+	/* put our credentials to the current session */
+	r = gnutls_credentials_set(sessArray[i], GNUTLS_CRD_CERTIFICATE, tlscred);
+	if(r != GNUTLS_E_SUCCESS) {
+		fprintf (stderr, "Setting credentials failed\n");
+		gnutls_perror(r);
+		exit(1);
+	}
+
+	/* NOTE: the following statement generates a cast warning, but there seems to
+	 * be no way around it with current GnuTLS. Do NOT try to "fix" the situation!
+	 */
+	gnutls_transport_set_ptr(sessArray[i], (gnutls_transport_ptr_t) sockArray[i]);
+
+	/* Perform the TLS handshake */
+	r = gnutls_handshake(sessArray[i]);
+	if(r < 0) {
+		fprintf (stderr, "TLS Handshake failed\n");
+		gnutls_perror(r);
+		exit(1);
+	}
+}
+
+static int
+sendTLS(int i, char *buf, int lenBuf)
+{
+	int lenSent;
+	int r;
+
+	lenSent = 0;
+	while(lenSent != lenBuf) {
+		r = gnutls_record_send(sessArray[i], buf + lenSent, lenBuf - lenSent);
+		if(r < 0)
+			break;
+		lenSent += r;
+	}
+
+	return lenSent;
+}
+
+static void
+closeTLSSess(int i)
+{
+	gnutls_bye(sessArray[i], GNUTLS_SHUT_RDWR);
+	gnutls_deinit(sessArray[i]);
+}
+#	else	/* NO TLS available */
+static void initTLS(void) {}
+static void initTLSSess(int __attribute__((unused)) i) {}
+static int sendTLS(int i, char *buf, int lenBuf) { return 0; }
+static void closeTLSSess(int __attribute__((unused)) i) {}
+#	endif
 
 /* Run the test.
  * rgerhards, 2009-04-03
@@ -370,18 +824,17 @@ int main(int argc, char *argv[])
 
 	setvbuf(stdout, buf, _IONBF, 48);
 	
-	if(!isatty(1))
-		bShowProgress = 0;
-
-	while((opt = getopt(argc, argv, "f:F:t:p:c:C:m:i:I:P:d:Dn:M:rB")) != -1) {
+	while((opt = getopt(argc, argv, "b:ef:F:t:p:c:C:m:i:I:P:d:Dn:L:M:rsBR:S:T:XW:Yz:Z:")) != -1) {
 		switch (opt) {
+		case 'b':	batchsize = atoll(optarg);
+				break;
 		case 't':	targetIP = optarg;
 				break;
 		case 'p':	targetPort = atoi(optarg);
 				break;
 		case 'n':	numTargetPorts = atoi(optarg);
 				break;
-		case 'c':	numConnections = atoi(optarg);
+		case 'c':	numConnections = (unsigned) atoi(optarg);
 				break;
 		case 'C':	numFileIterations = atoi(optarg);
 				break;
@@ -406,6 +859,8 @@ int main(int argc, char *argv[])
 				break;
 		case 'F':	frameDelim = atoi(optarg);
 				break;
+		case 'L':	tlsLogLevel = atoi(optarg);
+				break;
 		case 'M':	MsgToSend = optarg;
 				break;
 		case 'I':	dataFile = optarg;
@@ -414,20 +869,61 @@ int main(int argc, char *argv[])
 				 */
 				numMsgsToSend = 1000000;
 				break;
+		case 's':	bSilent = 1;
+				break;
 		case 'B':	bBinaryFile = 1;
 				break;
+		case 'R':	numRuns = atoi(optarg);
+				break;
+		case 'S':	sleepBetweenRuns = atoi(optarg);
+				break;
+		case 'X':	bStatsRecords = 1;
+				break;
+		case 'e':	bCSVoutput = 1;
+				break;
+		case 'T':	if(!strcmp(optarg, "udp")) {
+					transport = TP_UDP;
+				} else if(!strcmp(optarg, "tcp")) {
+					transport = TP_TCP;
+				} else if(!strcmp(optarg, "tls")) {
+#					if defined(ENABLE_GNUTLS)
+						transport = TP_TLS;
+#					else
+						fprintf(stderr, "compiled without TLS support!\n", optarg);
+						exit(1);
+#					endif
+				} else {
+					fprintf(stderr, "unknown transport '%s'\n", optarg);
+					exit(1);
+				}
+				break;
+		case 'W':	waittime = atoi(optarg);
+				break;
+		case 'Y':	runMultithreaded = 1;
+				break;
+		case 'z':	tlsKeyFile = optarg;
+				break;
+		case 'Z':	tlsCertFile = optarg;
+				break;
 		default:	printf("invalid option '%c' or value missing - terminating...\n", opt);
 				exit (1);
 				break;
 		}
 	}
 
+	if(bStatsRecords && waittime) {
+		fprintf(stderr, "warning: generating performance stats and using a waittime "
+				"is somewhat contradictory!\n");
+	}
+
+	if(!isatty(1) || bSilent)
+		bShowProgress = 0;
+
 	if(numConnections > 20) {
 		/* if we use many (whatever this means, 20 is randomly picked)
 		 * connections, we need to make sure we have a high enough
 		 * limit. -- rgerhards, 2010-03-25
 		 */
-		struct rlimit maxFiles;
 		maxFiles.rlim_cur = numConnections + 20;
 		maxFiles.rlim_max = numConnections + 20;
 		if(setrlimit(RLIMIT_NOFILE, &maxFiles) < 0) {
@@ -446,22 +942,27 @@ int main(int argc, char *argv[])
 		}
 	}
 
+	if(transport == TP_TLS) {
+		initTLS();
+	}
+
 	if(openConnections() != 0) {
 		printf("error opening connections\n");
 		exit(1);
 	}
 
-	if(sendMessages() != 0) {
-		printf("error sending messages\n");
+	if(runTests() != 0) {
+		printf("error running tests\n");
 		exit(1);
 	}
 
 	closeConnections(); /* this is important so that we do not finish too early! */
 
-	if(nConnDrops > 0)
+	if(nConnDrops > 0 && !bSilent)
 		printf("-D option initiated %ld connection closures\n", nConnDrops);
 
-	printf("End of tcpflood Run\n");
+	if(!bSilent)
+		printf("End of tcpflood Run\n");
 
 	exit(ret);
 }
diff --git a/tests/testsuites/failover-async.conf b/tests/testsuites/failover-async.conf
new file mode 100644
index 0000000..76445de
--- /dev/null
+++ b/tests/testsuites/failover-async.conf
@@ -0,0 +1,9 @@
+# see the equally-named .sh file for details
+$IncludeConfig diag-common.conf
+
+$template outfmt,"%msg:F,58:2%\n"
+# note: the target server shall not be available!
+
+$ActionQueueType LinkedList
+:msg, contains, "msgnum:" @@127.0.0.1:13514
+& ./rsyslog.out.log;outfmt
diff --git a/tests/testsuites/failover-double.conf b/tests/testsuites/failover-double.conf
new file mode 100644
index 0000000..a999132
--- /dev/null
+++ b/tests/testsuites/failover-double.conf
@@ -0,0 +1,9 @@
+$IncludeConfig diag-common.conf
+
+$template outfmt,"%msg:F,58:2%\n"
+
+:msg, contains, "msgnum:" @@127.0.0.1:13516
+$ActionExecOnlyWhenPreviousIsSuspended on
+&	@@127.0.0.1:1234
+&	./rsyslog.out.log;outfmt
+$ActionExecOnlyWhenPreviousIsSuspended off
diff --git a/tests/testsuites/sndrcv_failover_rcvr.conf b/tests/testsuites/sndrcv_failover_rcvr.conf
new file mode 100644
index 0000000..6f7ce34
--- /dev/null
+++ b/tests/testsuites/sndrcv_failover_rcvr.conf
@@ -0,0 +1,11 @@
+# see equally-named shell file for details
+# rgerhards, 2009-11-11
+$IncludeConfig diag-common.conf
+
+$ModLoad ../plugins/imtcp/.libs/imtcp
+# then SENDER sends to this port (not tcpflood!)
+$InputTCPServerRun 13515
+
+$template outfmt,"%msg:F,58:2%\n"
+$template dynfile,"rsyslog.out.log" # trick to use relative path names!
+:msg, contains, "msgnum:" ?dynfile;outfmt
diff --git a/tests/testsuites/sndrcv_failover_sender.conf b/tests/testsuites/sndrcv_failover_sender.conf
new file mode 100644
index 0000000..b8e7c18
--- /dev/null
+++ b/tests/testsuites/sndrcv_failover_sender.conf
@@ -0,0 +1,13 @@
+# see tcpsndrcv.sh for details
+# rgerhards, 2009-11-11
+$IncludeConfig diag-common2.conf
+
+$ModLoad ../plugins/imtcp/.libs/imtcp
+# this listener is for message generation by the test framework!
+$InputTCPServerRun 13514
+
+*.*	@@127.0.0.1:13516 # this must be DEAD
+$ActionExecOnlyWhenPreviousIsSuspended on
+&	@@127.0.0.1:13515
+&	./rsyslog.empty
+$ActionExecOnlyWhenPreviousIsSuspended off
diff --git a/tools/syslogd.c b/tools/syslogd.c
index 487ab36..096f930 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -799,7 +799,7 @@ DEFFUNC_llExecFunc(flushRptdMsgsActions)
 		DBGPRINTF("flush %s: repeated %d times, %d sec.\n",
 		    module.GetStateName(pAction->pMod), pAction->f_prevcount,
 		    repeatinterval[pAction->f_repeatcount]);
-		actionWriteToAction(pAction, NULL, 0);
+		actionWriteToAction(pAction);
 		BACKOFF(pAction);
 	}
 	UnlockObj(pAction);