Fix for a problem with the new messaging system. If a sender is using the

messaging system as a notification mechanism, and the speed of notification
greatly exceeds the speed of message recovery, then you get a massively (>75Mb)
growing tdb. If the message is a simple notification, then the message is
static, and you only need one of them in transit to a target process at
any one time.
This patch adds a BOOL "allow_duplicates" to the message_send_XX primitives.
If set to False, then before sending a message the sender checks the existing
message queue for a target pid for a duplicate of this message, and doesn't
add to it if one already exists.
Also added code into msgtest.c to test this.
Jeremy.
(This used to be commit 3aa7995660395ecb85c8e35b638fa9fbbb952558)

2 files changed, 37 insertions, 12 deletions

diff --git a/source3/utils/msgtest.c b/source3/utils/msgtest.c
index 858166e697..3fbf95af8f 100644
--- a/source3/utils/msgtest.c
+++ b/source3/utils/msgtest.c
@@ -36,12 +36,12 @@ void pong_message(int msg_type, pid_t src, void *buf, size_t len)
 	pong_count++;
 }
 
-
  int main(int argc, char *argv[])
 {
 	pid_t pid;
 	int i, n;
 	static pstring servicesf = CONFIGFILE;
+	char buf[12];
 
 	TimeInit();
 	setup_logging(argv[0],True);
@@ -52,13 +52,18 @@ void pong_message(int msg_type, pid_t src, void *buf, size_t len)
 
 	message_init();
 
+	if (argc != 3) {
+		fprintf(stderr, "%s: Usage - %s pid count\n", argv[0], argv[0]);
+		exit(1);
+	}
+
 	pid = atoi(argv[1]);
 	n = atoi(argv[2]);
 
 	message_register(MSG_PONG, pong_message);
 
 	for (i=0;i<n;i++) {
-		message_send_pid(pid, MSG_PING, NULL, 0);
+		message_send_pid(pid, MSG_PING, NULL, 0, True);
 	}
 
 	while (pong_count < i) {
@@ -66,6 +71,26 @@ void pong_message(int msg_type, pid_t src, void *buf, size_t len)
 		msleep(1);
 	}
 
+	/* Now test that the duplicate filtering code works. */
+	pong_count = 0;
+
+	safe_strcpy(buf, "1234567890", sizeof(buf)-1);
+
+	for (i=0;i<n;i++) {
+		message_send_pid(getpid(), MSG_PING, NULL, 0, False);
+		message_send_pid(getpid(), MSG_PING, buf, 11, False);
+	}
+
+	for (i=0;i<n;i++) {
+		message_dispatch();
+		msleep(1);
+	}
+
+	if (pong_count != 2) {
+		fprintf(stderr, "Duplicate filter failed (%d).\n", pong_count);
+		exit(1);
+	}
+
 	return (0);
 }
 
diff --git a/source3/utils/smbcontrol.c b/source3/utils/smbcontrol.c
index a273f82151..815f9fddff 100644
--- a/source3/utils/smbcontrol.c
+++ b/source3/utils/smbcontrol.c
@@ -115,13 +115,13 @@ void profilelevel_function(int msg_type, pid_t src, void *buf, size_t len)
 /****************************************************************************
 send a message to a named destination
 ****************************************************************************/
-static BOOL send_message(char *dest, int msg_type, void *buf, int len)
+static BOOL send_message(char *dest, int msg_type, void *buf, int len, BOOL duplicates)
 {
 	pid_t pid;
 
 	/* "smbd" is the only broadcast operation */
 	if (strequal(dest,"smbd")) {
-		return message_send_all(msg_type, buf, len);
+		return message_send_all(msg_type, buf, len, duplicates);
 	} else if (strequal(dest,"nmbd")) {
 		pid = pidfile_pid(dest);
 		if (pid == 0) {
@@ -136,7 +136,7 @@ static BOOL send_message(char *dest, int msg_type, void *buf, int len)
 		}		
 	} 
 
-	return message_send_pid(pid, msg_type, buf, len);
+	return message_send_pid(pid, msg_type, buf, len, duplicates);
 }
 
 /****************************************************************************
@@ -174,7 +174,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 			return(False);
 		}
 		v = atoi(params);
-		send_message(dest, MSG_DEBUG, &v, sizeof(int));
+		send_message(dest, MSG_DEBUG, &v, sizeof(int), False);
 		break;
 
 	case MSG_PROFILE:
@@ -195,7 +195,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 			"MSG_PROFILE parameter must be off, count, on, or flush\n");
 		    return(False);
 		}
-		send_message(dest, MSG_PROFILE, &v, sizeof(int));
+		send_message(dest, MSG_PROFILE, &v, sizeof(int), False);
 		break;
 
 	case MSG_FORCE_ELECTION:
@@ -203,7 +203,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 			fprintf(stderr,"force-election can only be sent to nmbd\n");
 			return(False);
 		}
-		send_message(dest, MSG_FORCE_ELECTION, NULL, 0);
+		send_message(dest, MSG_FORCE_ELECTION, NULL, 0, False);
 		break;
 
 	case MSG_REQ_PROFILELEVEL:
@@ -212,7 +212,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 		    profilelevel_registered = True;
 		}
 		got_level = False;
-		retval = send_message(dest, MSG_REQ_PROFILELEVEL, NULL, 0);
+		retval = send_message(dest, MSG_REQ_PROFILELEVEL, NULL, 0, True);
 		if (retval) {
 			timeout_start = time(NULL);
 			while (!got_level) {
@@ -231,7 +231,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 		    debuglevel_registered = True;
 		}
 		got_level = False;
-		retval = send_message(dest, MSG_REQ_DEBUGLEVEL, NULL, 0);
+		retval = send_message(dest, MSG_REQ_DEBUGLEVEL, NULL, 0, True);
 		if (retval) {
 			timeout_start = time(NULL);
 			while (!got_level) {
@@ -254,7 +254,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 			return (False);
 		}
 		retval = send_message(dest, MSG_PRINTER_NOTIFY, params,
-				      strlen(params) + 1);
+				      strlen(params) + 1, False);
 		break;
 
 	case MSG_PING:
@@ -269,7 +269,7 @@ static BOOL do_command(char *dest, char *msg_name, char *params)
 		n = atoi(params);
 		pong_count = 0;
 		for (i=0;i<n;i++) {
-			retval = send_message(dest, MSG_PING, NULL, 0);
+			retval = send_message(dest, MSG_PING, NULL, 0, True);
 			if (retval == False) break;
 		}
 		if (retval) {