solved design issue with queue termination

... and also improved the test suite. There is a design issue in the
v3 queue engine that manifested to some serious problems with the new
processing mode. However, in v3 shutdown may take eternally if a queue
runs in DA mode, is configured to preserve data AND the action fails and
retries immediately. There is no cure available for v3, it would
require doing much of the work we have done on the new engine. The window
of exposure, as one might guess from the description, is very small. That
is probably the reason why we have not seen it in practice.

10 files changed, 96 insertions, 31 deletions

diff --git a/tests/DiagTalker.java b/tests/DiagTalker.java
index e33a5867..85a6671e 100644
--- a/tests/DiagTalker.java
+++ b/tests/DiagTalker.java
@@ -1,3 +1,24 @@
+/* A yet very simple tool to talk to imdiag.
+ *
+ * Copyright 2009 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of rsyslog.
+ *
+ * Rsyslog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Rsyslog is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Rsyslog.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ */
 //package com.rsyslog.diag;
 import java.io.*;
 import java.net.*;
@@ -29,9 +50,14 @@ public class DiagTalker {
                                    new InputStreamReader(System.in));
 	String userInput;
 
-	while ((userInput = stdIn.readLine()) != null) {
-	    out.println(userInput);
-	    System.out.println("imdiag returns: " + in.readLine());
+	try {
+		while ((userInput = stdIn.readLine()) != null) {
+		    out.println(userInput);
+		    System.out.println("imdiag returns: " + in.readLine());
+		}
+        } catch (SocketException e) {
+            System.err.println("We had a socket exception and consider this to be OK: "
+	    			+ e.getMessage());
 	}
 
 	out.close();
diff --git a/tests/arrayqueue.sh b/tests/arrayqueue.sh
index 5b8ebb5f..7791ed57 100755
--- a/tests/arrayqueue.sh
+++ b/tests/arrayqueue.sh
@@ -17,7 +17,7 @@ sleep 4 # we need this so that rsyslogd can receive all outstanding messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 39999
+./chkseq -fwork -e 39999
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
diff --git a/tests/chkseq.c b/tests/chkseq.c
index 3203c250..5ffe855c 100644
--- a/tests/chkseq.c
+++ b/tests/chkseq.c
@@ -3,9 +3,10 @@
  * be set.
  *
  * Params
- * argv[1]	file to check
- * argv[2]	start number
- * argv[3]	end number
+ * -f<filename> MUST be given!
+ * -s<starting number> -e<ending number>
+ * default for s is 0. -e should be given (else it is also 0)
+ * -d may be specified, in which case duplicate messages are permitted.
  *
  * Part of the testbench for rsyslog.
  *
@@ -31,6 +32,7 @@
 #include "config.h"
 #include <stdio.h>
 #include <stdlib.h>
+#include <getopt.h>
 
 int main(int argc, char *argv[])
 {
@@ -38,16 +40,36 @@ int main(int argc, char *argv[])
 	int val;
 	int i;
 	int ret = 0;
-	int start, end;
+	int dupsPermitted = 0;
+	int start = 0, end = 0;
+	int opt;
+	int nDups = 0;
+	char *file = NULL;
 
-	if(argc != 4) {
-		printf("Invalid call of chkseq\n");
-		printf("Usage: chkseq file start end\n");
+	while((opt = getopt(argc, argv, "e:f:ds:")) != EOF) {
+		switch((char)opt) {
+		case 'f':
+			file = optarg;
+			break;
+                case 'd':
+			dupsPermitted = 1;
+			break;
+                case 'e':
+			end = atoi(optarg);
+			break;
+                case 's':
+			start = atoi(optarg);
+			break;
+		default:printf("Invalid call of chkseq\n");
+			printf("Usage: chkseq file -sstart -eend -d\n");
+			exit(1);
+		}
+	}
+
+	if(file == NULL) {
+		printf("file must be given!\n");
 		exit(1);
 	}
-	
-	start = atoi(argv[2]);
-	end = atoi(argv[3]);
 
 	if(start > end) {
 		printf("start must be less than or equal end!\n");
@@ -55,22 +77,35 @@ int main(int argc, char *argv[])
 	}
 
 	/* read file */
-	fp = fopen(argv[1], "r");
+	fp = fopen(file, "r");
 	if(fp == NULL) {
 		perror(argv[1]);
 		exit(1);
 	}
 
-	for(i = start ; i < end ; ++i) {
+	for(i = start ; i < end+1 ; ++i) {
 		if(fscanf(fp, "%d\n", &val) != 1) {
 			printf("scanf error in index i=%d\n", i);
 			exit(1);
 		}
 		if(val != i) {
-			printf("read value %d, but expected value %d\n", val, i);
-			exit(1);
+			if(val == i - 1 && dupsPermitted) {
+				--i;
+				++nDups;
+			} else {
+				printf("read value %d, but expected value %d\n", val, i);
+				exit(1);
+			}
 		}
 	}
 
+	if(nDups != 0)
+		printf("info: had %d duplicates (this is no error)\n", nDups);
+
+	if(i - 1 != end) {
+		printf("only %d records in file, expected %d\n", i - 1, end);
+		exit(1);
+	}
+
 	exit(ret);
 }
diff --git a/tests/da-mainmsg-q.sh b/tests/da-mainmsg-q.sh
index 91addf68..fde9e06e 100755
--- a/tests/da-mainmsg-q.sh
+++ b/tests/da-mainmsg-q.sh
@@ -52,7 +52,7 @@ sleep 1 # we need this so that rsyslogd can receive all outstanding messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 20099
+./chkseq -fwork -e20099
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
diff --git a/tests/diskqueue.sh b/tests/diskqueue.sh
index 20767a90..42018b15 100755
--- a/tests/diskqueue.sh
+++ b/tests/diskqueue.sh
@@ -26,7 +26,7 @@ $srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 19999
+./chkseq -fwork -e19999
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
diff --git a/tests/imtcp-multiport.sh b/tests/imtcp-multiport.sh
index 17480dae..73ab9558 100755
--- a/tests/imtcp-multiport.sh
+++ b/tests/imtcp-multiport.sh
@@ -21,7 +21,7 @@ $srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 9999
+./chkseq -fwork -e9999
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
@@ -46,7 +46,7 @@ $srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 9999
+./chkseq -fwork -e9999
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
@@ -71,7 +71,7 @@ $srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 9999
+./chkseq -fwork -e9999
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
diff --git a/tests/linkedlistqueue.sh b/tests/linkedlistqueue.sh
index aac1abb6..aa574bd1 100755
--- a/tests/linkedlistqueue.sh
+++ b/tests/linkedlistqueue.sh
@@ -17,7 +17,7 @@ sleep 4 # we need this so that rsyslogd can receive all outstanding messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 39999
+./chkseq -fwork -e39999
 if [ "$?" -ne "0" ]; then
  # rm -f work rsyslog.out.log
   echo "sequence error detected"
diff --git a/tests/manytcp.sh b/tests/manytcp.sh
index 06bd38b6..861f12ff 100755
--- a/tests/manytcp.sh
+++ b/tests/manytcp.sh
@@ -12,7 +12,7 @@ $srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 39999
+./chkseq -fwork -e39999
 if [ "$?" -ne "0" ]; then
   rm -f work rsyslog.out.log
   echo "sequence error detected"
diff --git a/tests/memq-persist.sh b/tests/memq-persist.sh
index 108cba57..e935d8db 100755
--- a/tests/memq-persist.sh
+++ b/tests/memq-persist.sh
@@ -10,9 +10,11 @@
 #export RSYSLOG_DEBUGLOG="log"
 echo testing memory queue persisting to disk
 $srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason
+rm -f core.*
 rm -rf test-spool
 mkdir test-spool
 rm -f work rsyslog.out.log rsyslog.out.log.save # work files
+#valgrind ../tools/rsyslogd -c4 -u2 -n -irsyslog.pid -M../runtime/.libs:../.libs -f$srcdir/testsuites/memq-persist1.conf &
 ../tools/rsyslogd -c4 -u2 -n -irsyslog.pid -M../runtime/.libs:../.libs -f$srcdir/testsuites/memq-persist1.conf &
 sleep 1
 echo "rsyslogd started with pid " `cat rsyslog.pid`
@@ -22,20 +24,22 @@ if [ "$?" -ne "0" ]; then
   echo "error during tcpflood! see rsyslog.out.log.save for what was written"
   cp rsyslog.out.log rsyslog.out.log.save
 fi
-sleep 3 # we need to wait to ensure everything is received (less 1 second would be better)
+sleep 4 # we need to wait to ensure everything is received (less 1 second would be better)
 kill `cat rsyslog.pid`
-sleep 5 # wait for engine to terminate
+echo wait for shutdown
+$srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 echo There must exist some files now:
 ls -l test-spool
 # restart engine and have rest processed
 ../tools/rsyslogd -c4 -u2 -n -irsyslog.pid -M../runtime/.libs:../.libs -f$srcdir/testsuites/memq-persist2.conf &
+sleep 1
 $srcdir/waitqueueempty.sh # wait until rsyslogd is done processing messages
 kill `cat rsyslog.pid`
 rm -f work
 sort < rsyslog.out.log > work
-./chkseq work 0 9999
+./chkseq -fwork -e9999 -d
 if [ "$?" -ne "0" ]; then
- # rm -f work rsyslog.out.log
+  rm -f work rsyslog.out.log
   echo "sequence error detected"
   exit 1
 fi
diff --git a/tests/rscript.c b/tests/rscript.c
index ce81491c..6361aec4 100644
--- a/tests/rscript.c
+++ b/tests/rscript.c
@@ -104,8 +104,8 @@ PerformTest(cstr_t *pstrIn, rsRetVal iRetExpected, cstr_t *pstrOut)
 
 	if(strcmp((char*)rsCStrGetSzStr(pstrPrg), (char*)rsCStrGetSzStr(pstrOut))) {
 		printf("error: compiled program different from expected result!\n");
-		printf("generated vmprg (%d bytes):\n%s\n", strlen((char*)rsCStrGetSzStr(pstrPrg)), rsCStrGetSzStr(pstrPrg));
-		printf("expected (%d bytes):\n%s\n", strlen((char*)rsCStrGetSzStr(pstrOut)), rsCStrGetSzStr(pstrOut));
+		printf("generated vmprg (%d bytes):\n%s\n", (int)strlen((char*)rsCStrGetSzStr(pstrPrg)), rsCStrGetSzStr(pstrPrg));
+		printf("expected (%d bytes):\n%s\n", (int)strlen((char*)rsCStrGetSzStr(pstrOut)), rsCStrGetSzStr(pstrOut));
 		ABORT_FINALIZE(RS_RET_ERR);
 	}