Imported Upstream version 5.8.0upstream/5.8.0

19 files changed, 338 insertions, 159 deletions

diff --git a/ChangeLog b/ChangeLog
index e4c2588..3300c73 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,4 +1,17 @@
 ---------------------------------------------------------------------------
+Version 5.8.0  [V5-stable] (rgerhards), 2011-04-12
+
+This is the new v5-stable branch, importing all feature from the 5.7.x
+versions. To see what has changed in regard to the previous v5-stable,
+check the Changelog for 5.7.x below.
+
+- bugfix: race condition in deferred name resolution
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=238
+  Special thanks to Marcin for his persistence in helping to solve this
+  bug.
+- bugfix: DA queue was never shutdown once it was started
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=241
+---------------------------------------------------------------------------
 Version 5.7.10  [V5-BETA] (rgerhards), 2011-03-29
 - bugfix: ompgsql did not work properly with ANSI SQL strings
   closes: http://bugzilla.adiscon.com/show_bug.cgi?id=229
@@ -150,7 +163,7 @@ Version 5.7.0  [V5-DEVEL] (rgerhards), 2010-09-16
     thanks to Lennart Poettering for this patch
   * sd-systemd API added as part of rsyslog runtime library
 ---------------------------------------------------------------------------
-Version 5.6.5  [V5-STABLE] (rgerhards), 2011-03-??
+Version 5.6.5  [V5-STABLE] (rgerhards), 2011-03-22
 - bugfix: failover did not work correctly if repeated msg reduction was on
   affected directive was: $ActionExecOnlyWhenPreviousIsSuspended on
 - bugfix: omlibdbi did not use password from rsyslog.con
diff --git a/configure b/configure
index d07725b..dde2576 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.65 for rsyslog 5.7.10.
+# Generated by GNU Autoconf 2.65 for rsyslog 5.8.0.
 #
 # Report bugs to <rsyslog@lists.adiscon.com>.
 #
@@ -701,8 +701,8 @@ MAKEFLAGS=
 # Identity of this package.
 PACKAGE_NAME='rsyslog'
 PACKAGE_TARNAME='rsyslog'
-PACKAGE_VERSION='5.7.10'
-PACKAGE_STRING='rsyslog 5.7.10'
+PACKAGE_VERSION='5.8.0'
+PACKAGE_STRING='rsyslog 5.8.0'
 PACKAGE_BUGREPORT='rsyslog@lists.adiscon.com'
 PACKAGE_URL=''
 
@@ -1608,7 +1608,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.7.10 to adapt to many kinds of systems.
+\`configure' configures rsyslog 5.8.0 to adapt to many kinds of systems.
 
 Usage: $0 [OPTION]... [VAR=VALUE]...
 
@@ -1678,7 +1678,7 @@ fi
 
 if test -n "$ac_init_help"; then
   case $ac_init_help in
-     short | recursive ) echo "Configuration of rsyslog 5.7.10:";;
+     short | recursive ) echo "Configuration of rsyslog 5.8.0:";;
    esac
   cat <<\_ACEOF
 
@@ -1848,7 +1848,7 @@ fi
 test -n "$ac_init_help" && exit $ac_status
 if $ac_init_version; then
   cat <<\_ACEOF
-rsyslog configure 5.7.10
+rsyslog configure 5.8.0
 generated by GNU Autoconf 2.65
 
 Copyright (C) 2009 Free Software Foundation, Inc.
@@ -2423,7 +2423,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.7.10, which was
+It was created by rsyslog $as_me 5.8.0, which was
 generated by GNU Autoconf 2.65.  Invocation command line was
 
   $ $0 $@
@@ -3231,7 +3231,7 @@ fi
 
 # Define the identity of the package.
  PACKAGE='rsyslog'
- VERSION='5.7.10'
+ VERSION='5.8.0'
 
 
 cat >>confdefs.h <<_ACEOF
@@ -17140,7 +17140,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.7.10, which was
+This file was extended by rsyslog $as_me 5.8.0, which was
 generated by GNU Autoconf 2.65.  Invocation command line was
 
   CONFIG_FILES    = $CONFIG_FILES
@@ -17206,7 +17206,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.7.10
+rsyslog config.status 5.8.0
 configured by $0, generated by GNU Autoconf 2.65,
   with options \\"\$ac_cs_config\\"
 
diff --git a/configure.ac b/configure.ac
index 2e56349..d608296 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.7.10],[rsyslog@lists.adiscon.com])
+AC_INIT([rsyslog],[5.8.0],[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 f28d18d..e1f0ebf 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.7.10 (beta branch) of rsyslog.</b>
+<p><b>This documentation is for version 5.8.0 (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
@@ -110,5 +110,7 @@ any restriction as long as your license is GPLv3 compatible. If your license is
 you may even be still permitted to use rsyslog source code. However, then you need to look at the way
 <a href="licensing.html">rsyslog is licensed</a>.</p>
 <p>Feedback is always welcome, but if you have a support question, please do not
-mail Rainer directly (<a href="free_support.html">why not?</a>).
+mail Rainer directly (<a href="free_support.html">why not?</a>) - use the
+<a href="http://lists.adiscon.net/mailman/listinfo/rsyslog">rsyslogmailing list</a>
+or <a href="http://kb.monitorware.com/rsyslog-f40.html">rsyslog formum</a> instead.
 </body></html>
diff --git a/doc/rsyslog_conf_global.html b/doc/rsyslog_conf_global.html
index 7d121a4..a5d69f1 100644
--- a/doc/rsyslog_conf_global.html
+++ b/doc/rsyslog_conf_global.html
@@ -93,7 +93,7 @@ default 60000 (1 minute)]</li>
 <li>$ActionQueueWorkerThreadMinumumMessages &lt;number&gt;, default 100</li>
 <li><a href="rsconf1_actionresumeinterval.html">$ActionResumeInterval</a></li>
 <li>$ActionResumeRetryCount &lt;number&gt; [default 0, -1 means eternal]</li>
-<li>$ActionSendResendLastMsgOnReconn &lt;[on/<b>off</b>]&gt; specifies if the last message is to be resend when a connecition broken and has been reconnedcted. May increase reliability, but comes at the risk of message duplication.
+<li>$ActionSendResendLastMsgOnReconnect &lt;[on/<b>off</b>]&gt; specifies if the last message is to be resend when a connecition breaks and has been reconnected. May increase reliability, but comes at the risk of message duplication.
 <li>$ActionSendStreamDriver &lt;driver basename&gt; just like $DefaultNetstreamDriver, but for the specific action</li>
 <li>$ActionSendStreamDriverMode &lt;mode&gt;, default 0, mode to use with the stream driver (driver-specific)</li>
 <li>$ActionSendStreamDriverAuthMode &lt;mode&gt;,&nbsp; authentication mode to use with the stream driver. Note that this directive requires TLS
diff --git a/doc/rsyslog_tls.html b/doc/rsyslog_tls.html
index bb312c7..286660d 100644
--- a/doc/rsyslog_tls.html
+++ b/doc/rsyslog_tls.html
@@ -162,25 +162,11 @@ similar "smart" command on the client. It should show up in the
 respective server log file. If you dig out your sniffer, you should see
 that the traffic on the wire is actually protected.</p>
 <h3>Limitations</h3>
-<p>The current implementation has a number of limitations. These
-are
-being worked on. Most importantly, neither the client nor the server
-are authenticated. So while the message transfer is encrypted, you can
-not be sure which peer you are talking to. Please note that this is a
-limitation found in most real-world SSL syslog systems. Of course, that
-is not an excuse for not yet providing this feature - but it tells you
-that it is acceptable and can be worked around by proper firewalling,
-ACLs and other organizational measures. Mutual authentication will be
-added shortly to rsyslog.</p>
-<p>Secondly, the plain tcp syslog listener
-can currently listen to a single port, in a single mode. So if you use
-a TLS-based listener, you can not run unencrypted syslog on the same
-instance at the same time. A work-around is to run a second rsyslogd
-instance. This limitation, too, is scheduled to be removed soon.</p>
 <p>The
 RELP transport can currently not be protected by TLS. A work-around is
 to use stunnel. TLS support for RELP will be added once plain TCP
-syslog has sufficiently matured.</p>
+syslog has sufficiently matured and there either is some time left to do this
+or we find a sponsor ;).</p>
 <h2>Certificates</h2>
 <p>In order to be really secure, certificates are needed. This is
 a short summary on how to generate the necessary certificates with
diff --git a/plugins/imdiag/imdiag.c b/plugins/imdiag/imdiag.c
index 404cebc..ed6ef50 100644
--- a/plugins/imdiag/imdiag.c
+++ b/plugins/imdiag/imdiag.c
@@ -205,7 +205,7 @@ doInjectMsg(int iNum)
 	DEFiRet;
 
 	snprintf((char*)szMsg, sizeof(szMsg)/sizeof(uchar),
-		 "<167>Mar  1 01:00:00 172.20.245.8 tag msgnum:%8.8d:\n", iNum);
+		 "<167>Mar  1 01:00:00 172.20.245.8 tag msgnum:%8.8d:", iNum);
 
 	datetime.getCurrTime(&stTime, &ttGenTime);
 	/* we now create our own message object and submit it to the queue */
@@ -247,6 +247,7 @@ injectMsg(uchar *pszCmd, tcps_sess_t *pSess)
 	}
 
 	CHKiRet(sendResponse(pSess, "%d messages injected\n", nMsgs));
+	DBGPRINTF("imdiag: %d messages injected\n", nMsgs);
 
 finalize_it:
 	RETiRet;
@@ -279,6 +280,7 @@ waitMainQEmpty(tcps_sess_t *pSess)
 	}
 
 	CHKiRet(sendResponse(pSess, "mainqueue empty\n"));
+	DBGPRINTF("imdiag: mainqueue empty\n");
 
 finalize_it:
 	RETiRet;
@@ -314,6 +316,7 @@ OnMsgReceived(tcps_sess_t *pSess, uchar *pRcv, int iLenMsg)
 	if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("getmainmsgqueuesize"))) {
 		CHKiRet(diagGetMainMsgQSize(&iMsgQueueSize));
 		CHKiRet(sendResponse(pSess, "%d\n", iMsgQueueSize));
+		DBGPRINTF("imdiag: %d messages in main queue\n", iMsgQueueSize);
 	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("waitmainqueueempty"))) {
 		CHKiRet(waitMainQEmpty(pSess));
 	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("injectmsg"))) {
diff --git a/runtime/msg.c b/runtime/msg.c
index e8e6096..d1e67aa 100644
--- a/runtime/msg.c
+++ b/runtime/msg.c
@@ -287,6 +287,121 @@ static pthread_mutex_t mutTrimCtr;	 /* mutex to handle malloc trim */
 static int getAPPNAMELen(msg_t *pM, sbool bLockMutex);
 
 
+/* The following functions will support advanced output module
+ * multithreading, once this is implemented. Currently, we
+ * include them as hooks only. The idea is that we need to guard
+ * some msg objects data fields against concurrent access if
+ * we run on multiple threads. Please note that in any case this
+ * is not necessary for calls from INPUT modules, because they
+ * construct the message object and do this serially. Only when
+ * the message is in the processing queue, multiple threads may
+ * access a single object. Consequently, there are no guard functions
+ * for "set" methods, as these are called during input. Only "get"
+ * functions that modify important structures have them.
+ * rgerhards, 2007-07-20
+ * We now support locked and non-locked operations, depending on
+ * the configuration of rsyslog. To support this, we use function
+ * pointers. Initially, we start in non-locked mode. There, all
+ * locking operations call into dummy functions. When locking is
+ * enabled, the function pointers are changed to functions doing
+ * actual work. We also introduced another MsgPrepareEnqueue() function
+ * which initializes the locking structures, if needed. This is
+ * necessary because internal messages during config file startup
+ * processing are always created in non-locking mode. So we can
+ * not initialize locking structures during constructions. We now
+ * postpone this until when the message is fully constructed and
+ * enqueued. Then we know the status of locking. This has a nice
+ * side effect, and that is that during the initial creation of
+ * the Msg object no locking needs to be done, which results in better
+ * performance. -- rgerhards, 2008-01-05
+ */
+static void (*funcLock)(msg_t *pMsg);
+static void (*funcUnlock)(msg_t *pMsg);
+static void (*funcDeleteMutex)(msg_t *pMsg);
+void (*funcMsgPrepareEnqueue)(msg_t *pMsg);
+#if 1 /* This is a debug aid */
+#define MsgLock(pMsg) 	funcLock(pMsg)
+#define MsgUnlock(pMsg) funcUnlock(pMsg)
+#else
+#define MsgLock(pMsg) 	{dbgprintf("MsgLock line %d\n - ", __LINE__); funcLock(pMsg);; }
+#define MsgUnlock(pMsg) {dbgprintf("MsgUnlock line %d - ", __LINE__); funcUnlock(pMsg); }
+#endif
+
+/* the next function is a dummy to be used by the looking functions
+ * when the class is not yet running in an environment where locking
+ * is necessary. Please note that the need to lock can (and will) change
+ * during a single run. Typically, this is depending on the operation mode
+ * of the message queues (which is operator-configurable). -- rgerhards, 2008-01-05
+ */
+static void MsgLockingDummy(msg_t __attribute__((unused)) *pMsg)
+{
+	/* empty be design */
+}
+
+
+/* The following function prepares a message for enqueue into the queue. This is
+ * where a message may be accessed by multiple threads. This implementation here
+ * is the version for multiple concurrent acces. It initializes the locking
+ * structures.
+ * TODO: change to an iRet interface! -- rgerhards, 2008-07-14
+ */
+static void MsgPrepareEnqueueLockingCase(msg_t *pThis)
+{
+	BEGINfunc
+	assert(pThis != NULL);
+	pthread_mutex_init(&pThis->mut, NULL);
+	pThis->bDoLock = 1;
+	ENDfunc
+}
+
+
+/* ... and now the locking and unlocking implementations: */
+static void MsgLockLockingCase(msg_t *pThis)
+{
+	/* DEV debug only! dbgprintf("MsgLock(0x%lx)\n", (unsigned long) pThis); */
+	assert(pThis != NULL);
+	if(pThis->bDoLock == 1) /* TODO: this is a testing hack, we should find a way with better performance! -- rgerhards, 2009-01-27 */
+		pthread_mutex_lock(&pThis->mut);
+}
+
+static void MsgUnlockLockingCase(msg_t *pThis)
+{
+	/* DEV debug only! dbgprintf("MsgUnlock(0x%lx)\n", (unsigned long) pThis); */
+	assert(pThis != NULL);
+	if(pThis->bDoLock == 1) /* TODO: this is a testing hack, we should find a way with better performance! -- rgerhards, 2009-01-27 */
+		pthread_mutex_unlock(&pThis->mut);
+}
+
+/* delete the mutex object on message destruction (locking case)
+ */
+static void MsgDeleteMutexLockingCase(msg_t *pThis)
+{
+	assert(pThis != NULL);
+	pthread_mutex_destroy(&pThis->mut);
+}
+
+/* enable multiple concurrent access on the message object
+ * This works on a class-wide basis and can bot be undone.
+ * That is, if it is once enabled, it can not be disabled during
+ * the same run. When this function is called, no other thread
+ * must manipulate message objects. Then we would have race conditions,
+ * but guarding against this is counter-productive because it
+ * would cost additional time. Plus, it would be a programming error.
+ * rgerhards, 2008-01-05
+ */
+rsRetVal MsgEnableThreadSafety(void)
+{
+	DEFiRet;
+	funcLock = MsgLockLockingCase;
+	funcUnlock = MsgUnlockLockingCase;
+	funcMsgPrepareEnqueue = MsgPrepareEnqueueLockingCase;
+	funcDeleteMutex = MsgDeleteMutexLockingCase;
+	RETiRet;
+}
+
+/* end locking functions */
+
+
 static inline int getProtocolVersion(msg_t *pM)
 {
 	return(pM->iProtocolVersion);
@@ -306,6 +421,7 @@ resolveDNS(msg_t *pMsg) {
 	uchar fromHostFQDN[NI_MAXHOST];
 	DEFiRet;
 
+	MsgLock(pMsg);
 	CHKiRet(objUse(net, CORE_COMPONENT));
 	if(pMsg->msgFlags & NEEDS_DNSRESOL) {
 		localRet = net.cvthname(pMsg->rcvFrom.pfrominet, fromHost, fromHostFQDN, fromHostIP);
@@ -315,6 +431,7 @@ resolveDNS(msg_t *pMsg) {
 		}
 	}
 finalize_it:
+	MsgUnlock(pMsg);
 	if(iRet != RS_RET_OK) {
 		/* best we can do: remove property */
 		MsgSetRcvFromStr(pMsg, UCHAR_CONSTANT(""), 0, &propFromHost);
@@ -535,121 +652,6 @@ uchar *propIDToName(propid_t propID)
 }
 
 
-/* The following functions will support advanced output module
- * multithreading, once this is implemented. Currently, we
- * include them as hooks only. The idea is that we need to guard
- * some msg objects data fields against concurrent access if
- * we run on multiple threads. Please note that in any case this
- * is not necessary for calls from INPUT modules, because they
- * construct the message object and do this serially. Only when
- * the message is in the processing queue, multiple threads may
- * access a single object. Consequently, there are no guard functions
- * for "set" methods, as these are called during input. Only "get"
- * functions that modify important structures have them.
- * rgerhards, 2007-07-20
- * We now support locked and non-locked operations, depending on
- * the configuration of rsyslog. To support this, we use function
- * pointers. Initially, we start in non-locked mode. There, all
- * locking operations call into dummy functions. When locking is
- * enabled, the function pointers are changed to functions doing
- * actual work. We also introduced another MsgPrepareEnqueue() function
- * which initializes the locking structures, if needed. This is
- * necessary because internal messages during config file startup
- * processing are always created in non-locking mode. So we can
- * not initialize locking structures during constructions. We now
- * postpone this until when the message is fully constructed and
- * enqueued. Then we know the status of locking. This has a nice
- * side effect, and that is that during the initial creation of
- * the Msg object no locking needs to be done, which results in better
- * performance. -- rgerhards, 2008-01-05
- */
-static void (*funcLock)(msg_t *pMsg);
-static void (*funcUnlock)(msg_t *pMsg);
-static void (*funcDeleteMutex)(msg_t *pMsg);
-void (*funcMsgPrepareEnqueue)(msg_t *pMsg);
-#if 1 /* This is a debug aid */
-#define MsgLock(pMsg) 	funcLock(pMsg)
-#define MsgUnlock(pMsg) funcUnlock(pMsg)
-#else
-#define MsgLock(pMsg) 	{dbgprintf("MsgLock line %d\n - ", __LINE__); funcLock(pMsg);; }
-#define MsgUnlock(pMsg) {dbgprintf("MsgUnlock line %d - ", __LINE__); funcUnlock(pMsg); }
-#endif
-
-/* the next function is a dummy to be used by the looking functions
- * when the class is not yet running in an environment where locking
- * is necessary. Please note that the need to lock can (and will) change
- * during a single run. Typically, this is depending on the operation mode
- * of the message queues (which is operator-configurable). -- rgerhards, 2008-01-05
- */
-static void MsgLockingDummy(msg_t __attribute__((unused)) *pMsg)
-{
-	/* empty be design */
-}
-
-
-/* The following function prepares a message for enqueue into the queue. This is
- * where a message may be accessed by multiple threads. This implementation here
- * is the version for multiple concurrent acces. It initializes the locking
- * structures.
- * TODO: change to an iRet interface! -- rgerhards, 2008-07-14
- */
-static void MsgPrepareEnqueueLockingCase(msg_t *pThis)
-{
-	BEGINfunc
-	assert(pThis != NULL);
-	pthread_mutex_init(&pThis->mut, NULL);
-	pThis->bDoLock = 1;
-	ENDfunc
-}
-
-
-/* ... and now the locking and unlocking implementations: */
-static void MsgLockLockingCase(msg_t *pThis)
-{
-	/* DEV debug only! dbgprintf("MsgLock(0x%lx)\n", (unsigned long) pThis); */
-	assert(pThis != NULL);
-	if(pThis->bDoLock == 1) /* TODO: this is a testing hack, we should find a way with better performance! -- rgerhards, 2009-01-27 */
-		pthread_mutex_lock(&pThis->mut);
-}
-
-static void MsgUnlockLockingCase(msg_t *pThis)
-{
-	/* DEV debug only! dbgprintf("MsgUnlock(0x%lx)\n", (unsigned long) pThis); */
-	assert(pThis != NULL);
-	if(pThis->bDoLock == 1) /* TODO: this is a testing hack, we should find a way with better performance! -- rgerhards, 2009-01-27 */
-		pthread_mutex_unlock(&pThis->mut);
-}
-
-/* delete the mutex object on message destruction (locking case)
- */
-static void MsgDeleteMutexLockingCase(msg_t *pThis)
-{
-	assert(pThis != NULL);
-	pthread_mutex_destroy(&pThis->mut);
-}
-
-/* enable multiple concurrent access on the message object
- * This works on a class-wide basis and can bot be undone.
- * That is, if it is once enabled, it can not be disabled during
- * the same run. When this function is called, no other thread
- * must manipulate message objects. Then we would have race conditions,
- * but guarding against this is counter-productive because it
- * would cost additional time. Plus, it would be a programming error.
- * rgerhards, 2008-01-05
- */
-rsRetVal MsgEnableThreadSafety(void)
-{
-	DEFiRet;
-	funcLock = MsgLockLockingCase;
-	funcUnlock = MsgUnlockLockingCase;
-	funcMsgPrepareEnqueue = MsgPrepareEnqueueLockingCase;
-	funcDeleteMutex = MsgDeleteMutexLockingCase;
-	RETiRet;
-}
-
-/* end locking functions */
-
-
 /* This is common code for all Constructors. It is defined in an
  * inline'able function so that we can save a function call in the
  * actual constructors (otherwise, the msgConstruct would need
diff --git a/runtime/queue.c b/runtime/queue.c
index ef6e843..50ae307 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -1774,9 +1774,13 @@ qqueueChkStopWrkrDA(qqueue_t *pThis)
 {
 	DEFiRet;
 
+//DBGPRINTF("XXXX: chkStopWrkrDA called, low watermark %d, phys Size %d\n", pThis->iLowWtrMrk, getPhysicalQueueSize(pThis));
 	if(pThis->bEnqOnly) {
 		iRet = RS_RET_TERMINATE_WHEN_IDLE;
 	}
+	if(getPhysicalQueueSize(pThis) <= pThis->iLowWtrMrk) {
+		iRet = RS_RET_TERMINATE_NOW;
+	}
 
 	RETiRet;
 }
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index c9c64a3..5ee2a55 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -223,7 +223,7 @@ processBatch(batch_t *pBatch)
 	DEFiRet;
 	assert(pBatch != NULL);
 
-dbgprintf("ZZZ: processBatch: batch of %d elements must be processed\n", pBatch->nElem);
+	DBGPRINTF("processBatch: batch of %d elements must be processed\n", pBatch->nElem);
 	if(pBatch->bSingleRuleset) {
 		pThis = batchGetRuleset(pBatch);
 		if(pThis == NULL)
@@ -235,7 +235,7 @@ dbgprintf("ZZZ: processBatch: batch of %d elements must be processed\n", pBatch-
 	}
 
 finalize_it:
-dbgprintf("ruleset.ProcessMsg() returns %d\n", iRet);
+	DBGPRINTF("ruleset.ProcessMsg() returns %d\n", iRet);
 	RETiRet;
 }
 
diff --git a/tests/Makefile.am b/tests/Makefile.am
index ebbe0cf..930aa30 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -56,8 +56,7 @@ TESTS +=  \
 	failover-rptd.sh \
 	failover-no-rptd.sh \
 	failover-no-basic.sh \
-	queue-persist.sh 
-	queue-persist.sh \
+	rcvr_fail_restore.sh \
 	linkedlistqueue.sh
 endif
 
@@ -289,6 +288,9 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \
 	   diag.sh \
 	   testsuites/diag-common.conf \
 	   testsuites/diag-common2.conf \
+           rcvr_fail_restore.sh \
+	   testsuites/rcvr_fail_restore_rcvr.conf \
+	   testsuites/rcvr_fail_restore_sender.conf \
            daqueue-persist.sh \
 	   daqueue-persist-drvr.sh \
 	   queue-persist.sh \
diff --git a/tests/Makefile.in b/tests/Makefile.in
index 32472e8..7322ba3 100644
--- a/tests/Makefile.in
+++ b/tests/Makefile.in
@@ -100,7 +100,8 @@ host_triplet = @host@
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-rptd.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-no-rptd.sh \
 @ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	failover-no-basic.sh \
-@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	queue-persist.sh 
+@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	rcvr_fail_restore.sh \
+@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	linkedlistqueue.sh
 
 @ENABLE_TESTBENCH_TRUE@@HAVE_VALGRIND_TRUE@am__append_2 = \
 @ENABLE_TESTBENCH_TRUE@@HAVE_VALGRIND_TRUE@	discard-rptdmsg-vg.sh \
@@ -562,6 +563,9 @@ EXTRA_DIST = 1.rstest 2.rstest 3.rstest err1.rstest \
 	   diag.sh \
 	   testsuites/diag-common.conf \
 	   testsuites/diag-common2.conf \
+           rcvr_fail_restore.sh \
+	   testsuites/rcvr_fail_restore_rcvr.conf \
+	   testsuites/rcvr_fail_restore_sender.conf \
            daqueue-persist.sh \
 	   daqueue-persist-drvr.sh \
 	   queue-persist.sh \
@@ -1226,8 +1230,6 @@ uninstall-am:
 	mostlyclean-generic mostlyclean-libtool pdf pdf-am ps ps-am \
 	tags uninstall uninstall-am
 
-@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	queue-persist.sh \
-@ENABLE_IMDIAG_TRUE@@ENABLE_TESTBENCH_TRUE@	linkedlistqueue.sh
 
 # Tell versions [3.59,3.63) of GNU make to not export all variables.
 # Otherwise a system limit (for SysV at least) may be exceeded.
diff --git a/tests/diag.sh b/tests/diag.sh
index 8fbe42e..d1242fb 100755
--- a/tests/diag.sh
+++ b/tests/diag.sh
@@ -84,15 +84,27 @@ case $1 in
 			exit 1
 		fi
 		;;
+   'get-mainqueuesize') # show the current main queue size
+		if [ "$2" == "2" ]
+		then
+			echo getmainmsgqueuesize | ./diagtalker -p13501
+		else
+			echo getmainmsgqueuesize | ./diagtalker
+		fi
+		;;
    'wait-queueempty') # wait for main message queue to be empty. $2 is the instance.
 		if [ "$2" == "2" ]
 		then
-			echo WaitMainQueueEmpty | ./diagtalker
+			echo WaitMainQueueEmpty | ./diagtalker -p13501
 		else
 			echo WaitMainQueueEmpty | ./diagtalker
 		fi
 		;;
    'shutdown-when-empty') # shut rsyslogd down when main queue is empty. $2 is the instance.
+		if [ "$2" == "2" ]
+		then
+		   echo Shutting down instance 2
+		fi
    		$srcdir/diag.sh wait-queueempty $2
 		kill `cat rsyslog$2.pid`
 		# note: we do not wait for the actual termination!
diff --git a/tests/diagtalker.c b/tests/diagtalker.c
index 6a721e4..0d728bd 100644
--- a/tests/diagtalker.c
+++ b/tests/diagtalker.c
@@ -1,7 +1,7 @@
 /* A yet very simple tool to talk to imdiag (this replaces the
  * previous Java implementation in order to get fewer dependencies).
  *
- * Copyright 2010 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2010,2011 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -126,7 +126,7 @@ doProcessing()
 		len = strlen(line);
 		sendCmd(fd, line, len);
 		waitRsp(fd, line, sizeof(line));
-		printf("imdiag: %s", line);
+		printf("imdiag[%d]: %s", targetPort, line);
 	}
 }
 
@@ -139,7 +139,7 @@ int main(int argc, char *argv[])
 	int ret = 0;
 	int opt;
 
-	while((opt = getopt(argc, argv, "f:t:p:c:C:m:i:I:P:d:n:M:rB")) != -1) {
+	while((opt = getopt(argc, argv, "t:p:")) != -1) {
 		switch (opt) {
 		case 't':	targetIP = optarg;
 				break;
diff --git a/tests/rcvr_fail_restore.sh b/tests/rcvr_fail_restore.sh
new file mode 100755
index 0000000..79486f1
--- /dev/null
+++ b/tests/rcvr_fail_restore.sh
@@ -0,0 +1,122 @@
+# Copyright (C) 2011 by Rainer Gerhards
+# This file is part of the rsyslog project, released  under GPLv3
+echo ===============================================================================
+echo \[rcvr_fail_restore.sh\]: test failed receiver restore case
+source $srcdir/diag.sh init
+#
+# STEP1: start both instances and send 1000 messages.
+# Note: receiver is instance 2, sender instance 1.
+#
+# start up the instances. Note that the envrionment settings can be changed to
+# set instance-specific debugging parameters!
+#export RSYSLOG_DEBUG="debug nostdout"
+#export RSYSLOG_DEBUGLOG="log2"
+source $srcdir/diag.sh startup rcvr_fail_restore_rcvr.conf 2
+#export RSYSLOG_DEBUGLOG="log"
+#valgrind="valgrind"
+source $srcdir/diag.sh startup rcvr_fail_restore_sender.conf
+# re-set params so that new instances do not thrash it...
+#unset RSYSLOG_DEBUG
+#unset RSYSLOG_DEBUGLOG
+
+# now inject the messages into instance 2. It will connect to instance 1,
+# and that instance will record the data.
+source $srcdir/diag.sh injectmsg  1 1000
+source $srcdir/diag.sh wait-queueempty
+./msleep 1000 # let things settle down a bit
+
+#
+# Step 2: shutdown receiver, then send some more data, which then
+# needs to go into the queue.
+#
+
+source $srcdir/diag.sh shutdown-when-empty 2
+source $srcdir/diag.sh wait-shutdown 2
+
+source $srcdir/diag.sh injectmsg  1001 10000
+./msleep 3000 # make sure some retries happen (retry interval is set to 3 second)
+source $srcdir/diag.sh get-mainqueuesize
+ls -l test-spool
+
+#
+# Step 3: restart receiver, wait that the sender drains its queue
+#
+#export RSYSLOG_DEBUGLOG="log2"
+source $srcdir/diag.sh startup rcvr_fail_restore_rcvr.conf 2
+echo waiting for sender to drain queue [may need a short while]
+source $srcdir/diag.sh wait-queueempty
+ls -l test-spool
+OLDFILESIZE=$(stat -c%s test-spool/mainq.00000001)
+echo file size to expect is $OLDFILESIZE
+
+
+#
+# Step 4: send new data. Queue files are not permitted to grow now
+# (but one file continous to exist).
+#
+source $srcdir/diag.sh injectmsg  11001 10
+source $srcdir/diag.sh wait-queueempty
+
+# at this point, the queue file shall not have grown. Note
+# that we MUST NOT shut down the instance right now, because it
+# would clean up the queue files! So we need to do our checks
+# first (here!).
+ls -l test-spool
+NEWFILESIZE=$(stat -c%s test-spool/mainq.00000001)
+if [ $NEWFILESIZE != $OLDFILESIZE ]
+then
+   echo file sizes do not match, expected $OLDFILESIZE, actual $NEWFILESIZE
+   echo this means that data has been written to the queue file where it
+   echo no longer should be written.
+   # abort will happen below, because we must ensure proper system shutdown
+   # HOWEVER, during actual testing it may be useful to do an exit here (so
+   # that e.g. the debug log is pointed right at the correct spot).
+   # exit 1
+fi
+
+#
+# We now do an extra test (so this is two in one ;)) to see if the DA
+# queue can be reactivated after its initial shutdown. In essence, we 
+# redo steps 2 and 3.
+#
+# Step 5: stop receiver again, then send some more data, which then
+# needs to go into the queue.
+#
+echo "*** done primary test *** now checking if DA can be restarted"
+source $srcdir/diag.sh shutdown-when-empty 2
+source $srcdir/diag.sh wait-shutdown 2
+
+source $srcdir/diag.sh injectmsg  11011 10000
+sleep 1 # we need to wait, otherwise we may be so fast that the receiver
+# comes up before we have finally suspended the action
+source $srcdir/diag.sh get-mainqueuesize
+ls -l test-spool
+
+#
+# Step 6: restart receiver, wait that the sender drains its queue
+#
+source $srcdir/diag.sh startup rcvr_fail_restore_rcvr.conf 2
+echo waiting for sender to drain queue [may need a short while]
+source $srcdir/diag.sh wait-queueempty
+ls -l test-spool
+
+#
+# Queue file size checks done. Now it is time to terminate the system
+# and see if everything could be received (the usual check, done here
+# for completeness, more or less as a bonus).
+#
+source $srcdir/diag.sh shutdown-when-empty
+source $srcdir/diag.sh wait-shutdown
+
+# now it is time to stop the receiver as well
+source $srcdir/diag.sh shutdown-when-empty 2
+source $srcdir/diag.sh wait-shutdown 2
+
+# now abort test if we need to (due to filesize predicate)
+if [ $NEWFILESIZE != $OLDFILESIZE ]
+then
+   exit 1
+fi
+# do the final check
+source $srcdir/diag.sh seq-check 1 21010
+source $srcdir/diag.sh exit
diff --git a/tests/testsuites/diag-common.conf b/tests/testsuites/diag-common.conf
index 9e9e28f..d76e2d1 100644
--- a/tests/testsuites/diag-common.conf
+++ b/tests/testsuites/diag-common.conf
@@ -13,4 +13,7 @@ $IMDiagServerRun 13500
 $template startupfile,"rsyslogd.started" # trick to use relative path names!
 :syslogtag, contains, "rsyslogd"  ?startupfile
 
-$ErrorMessagesToStderr off
+# I have disabled the directive below, so that we see errors in testcase
+# creation. I am not sure why it was present in the first place, so for
+# now I just leave it commented out -- rgerhards, 2011-03-30
+#$ErrorMessagesToStderr off
diff --git a/tests/testsuites/random.conf b/tests/testsuites/random.conf
index a7079df..74ec61a 100644
--- a/tests/testsuites/random.conf
+++ b/tests/testsuites/random.conf
@@ -4,6 +4,11 @@
 # rgerhards, 2010-04-01
 $IncludeConfig diag-common.conf
 
+# The random data will generate TCP framing error messages. We will
+# not clutter the test output with them. So we disable error messages
+# to stderr.
+$ErrorMessagesToStderr off
+
 $ModLoad ../plugins/imtcp/.libs/imtcp
 $MainMsgQueueTimeoutShutdown 10000
 $InputTCPServerRun 13514
diff --git a/tests/testsuites/rcvr_fail_restore_rcvr.conf b/tests/testsuites/rcvr_fail_restore_rcvr.conf
new file mode 100644
index 0000000..38ebad2
--- /dev/null
+++ b/tests/testsuites/rcvr_fail_restore_rcvr.conf
@@ -0,0 +1,8 @@
+$IncludeConfig diag-common2.conf
+
+$ModLoad ../plugins/imtcp/.libs/imtcp
+# then SENDER sends to this port (not tcpflood!)
+$InputTCPServerRun 13515
+
+$template outfmt,"%msg:F,58:2%\n"
+:msg, contains, "msgnum:" ./rsyslog.out.log;outfmt
diff --git a/tests/testsuites/rcvr_fail_restore_sender.conf b/tests/testsuites/rcvr_fail_restore_sender.conf
new file mode 100644
index 0000000..6b11aa4
--- /dev/null
+++ b/tests/testsuites/rcvr_fail_restore_sender.conf
@@ -0,0 +1,15 @@
+$IncludeConfig diag-common.conf
+
+$ModLoad ../plugins/imtcp/.libs/imtcp
+# this listener is for message generation by the test framework!
+$InputTCPServerRun 13514
+
+$WorkDirectory test-spool
+$MainMsgQueueMaxFileSize 1g
+$MainMsgQueueFileName mainq
+
+# we use the shortest resume interval a) to let the test not run too long 
+# and b) make sure some retries happen before the reconnect
+$ActionResumeInterval 1
+$ActionResumeRetryCount -1
+*.*	@@127.0.0.1:13515