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)

9 files changed, 99 insertions, 36 deletions

diff --git a/source3/Makefile.in b/source3/Makefile.in
index 2b490da4f6..a27aa85f04 100644
--- a/source3/Makefile.in
+++ b/source3/Makefile.in
@@ -291,6 +291,9 @@ SMBTORTURE_OBJ = utils/torture.o utils/nbio.o $(LIBSMB_OBJ) $(PARAM_OBJ) \
 MASKTEST_OBJ = utils/masktest.o $(LIBSMB_OBJ) $(PARAM_OBJ) \
                  $(UBIQX_OBJ) $(LIB_OBJ)
 
+MSGTEST_OBJ = utils/msgtest.o $(LIBSMB_OBJ) $(PARAM_OBJ) \
+                 $(UBIQX_OBJ) $(LIB_OBJ)
+
 LOCKTEST_OBJ = utils/locktest.o $(LOCKING_OBJ) $(LIBSMB_OBJ) $(PARAM_OBJ) \
                  $(UBIQX_OBJ) $(LIB_OBJ)
 
@@ -359,6 +362,8 @@ smbtorture : CHECK bin/smbtorture
 
 masktest : CHECK bin/masktest
 
+msgtest : CHECK bin/msgtest
+
 locktest : CHECK bin/locktest
 
 locktest2 : CHECK bin/locktest2
@@ -514,6 +519,10 @@ bin/masktest: $(MASKTEST_OBJ) bin/.dummy
 	@echo Linking $@
 	@$(CC) $(FLAGS) -o $@ $(MASKTEST_OBJ) $(LDFLAGS) $(LIBS)
 
+bin/msgtest: $(MSGTEST_OBJ) bin/.dummy
+	@echo Linking $@
+	@$(CC) $(FLAGS) -o $@ $(MSGTEST_OBJ) $(LDFLAGS) $(LIBS)
+
 bin/locktest: $(LOCKTEST_OBJ) bin/.dummy
 	@echo Linking $@
 	@$(CC) $(FLAGS) -o $@ $(LOCKTEST_OBJ) $(LDFLAGS) $(LIBS)
diff --git a/source3/include/proto.h b/source3/include/proto.h
index 16545b156d..d8d7180cc1 100644
--- a/source3/include/proto.h
+++ b/source3/include/proto.h
@@ -153,12 +153,12 @@ void mdfour(unsigned char *out, unsigned char *in, int n);
 void ping_message(int msg_type, pid_t src, void *buf, size_t len);
 void debuglevel_message(int msg_type, pid_t src, void *buf, size_t len);
 BOOL message_init(void);
-BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len);
+BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len, BOOL duplicates_allowed);
 void message_dispatch(void);
 void message_register(int msg_type, 
 		      void (*fn)(int msg_type, pid_t pid, void *buf, size_t len));
 void message_deregister(int msg_type);
-BOOL message_send_all(int msg_type, void *buf, size_t len);
+BOOL message_send_all(int msg_type, void *buf, size_t len, BOOL duplicates_allowed);
 
 /*The following definitions come from  lib/ms_fnmatch.c  */
 
diff --git a/source3/lib/debug.c b/source3/lib/debug.c
index cdcd44955b..9d520b6c2f 100644
--- a/source3/lib/debug.c
+++ b/source3/lib/debug.c
@@ -135,7 +135,7 @@ send a "set debug level" message
 ****************************************************************************/
 void debug_message_send(pid_t pid, int level)
 {
-	message_send_pid(pid, MSG_DEBUG, &level, sizeof(int));
+	message_send_pid(pid, MSG_DEBUG, &level, sizeof(int), False);
 }
 
 
diff --git a/source3/lib/messages.c b/source3/lib/messages.c
index 126ca768b7..0e2dfeffd6 100644
--- a/source3/lib/messages.c
+++ b/source3/lib/messages.c
@@ -71,7 +71,7 @@ a useful function for testing the message system
 void ping_message(int msg_type, pid_t src, void *buf, size_t len)
 {
 	DEBUG(1,("INFO: Received PING message from PID %d\n",src));
-	message_send_pid(src, MSG_PONG, buf, len);
+	message_send_pid(src, MSG_PONG, buf, len, True);
 }
 
 /****************************************************************************
@@ -83,7 +83,7 @@ void debuglevel_message(int msg_type, pid_t src, void *buf, size_t len)
 	
 	DEBUG(1,("INFO: Received REQ_DEBUGLEVEL message from PID %d\n",src));
         level = DEBUGLEVEL;
-	message_send_pid(src, MSG_DEBUGLEVEL, &level, sizeof(int));
+	message_send_pid(src, MSG_DEBUGLEVEL, &level, sizeof(int), True);
 }
 
 /****************************************************************************
@@ -148,13 +148,23 @@ static BOOL message_notify(pid_t pid)
 /****************************************************************************
 send a message to a particular pid
 ****************************************************************************/
-BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len)
+BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 {
 	TDB_DATA kbuf;
 	TDB_DATA dbuf;
 	struct message_rec rec;
 	void *p;
 
+	/*
+	 * Do an early check for process exists - saves adding into a tdb
+	 * and deleting again if the target is not present. JRA.
+	 */
+
+	if (kill(pid, 0) == -1) {
+		DEBUG(2,("message_send_pid: pid %d doesn't exist\n", (int)pid));
+		return False;
+	}
+
 	rec.msg_version = MESSAGE_VERSION;
 	rec.msg_type = msg_type;
 	rec.dest = pid;
@@ -183,6 +193,30 @@ BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len)
 		goto ok;
 	}
 
+	if (!duplicates_allowed) {
+		char *ptr;
+		struct message_rec *prec;
+		
+		for(ptr = (char *)dbuf.dptr, prec = (struct message_rec *)ptr; ptr < dbuf.dptr + dbuf.dsize;
+					ptr += (sizeof(rec) + prec->len), prec = (struct message_rec *)ptr) {
+
+			/*
+			 * First check if the message header matches, then, if it's a non-zero
+			 * sized message, check if the data matches. If so it's a duplicate and
+			 * we can discard it. JRA.
+			 */
+
+			if (!memcmp(ptr, &rec, sizeof(rec))) {
+				if (!len || (len && !memcmp( ptr + sizeof(rec), (char *)buf, len))) {
+					DEBUG(10,("message_send_pid: discarding duplicate message.\n"));
+					free(dbuf.dptr);
+					tdb_unlockchain(tdb, kbuf);
+					return True;
+				}
+			}
+		}
+	}
+
 	/* we're adding to an existing entry */
 	p = (void *)malloc(dbuf.dsize + len + sizeof(rec));
 	if (!p) goto failed;
@@ -323,6 +357,7 @@ static struct {
 	int msg_type;
 	void *buf;
 	size_t len;
+	BOOL duplicates;
 } msg_all;
 
 /****************************************************************************
@@ -335,7 +370,7 @@ static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void
 	memcpy(&crec, dbuf.dptr, sizeof(crec));
 
 	if (crec.cnum == -1) return 0;
-	message_send_pid(crec.pid, msg_all.msg_type, msg_all.buf, msg_all.len);
+	message_send_pid(crec.pid, msg_all.msg_type, msg_all.buf, msg_all.len, msg_all.duplicates);
 	return 0;
 }
 
@@ -344,7 +379,7 @@ this is a useful function for sending messages to all smbd processes.
 It isn't very efficient, but should be OK for the sorts of applications that 
 use it. When we need efficient broadcast we can add it.
 ****************************************************************************/
-BOOL message_send_all(int msg_type, void *buf, size_t len)
+BOOL message_send_all(int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 {
 	TDB_CONTEXT *the_tdb;
 
@@ -357,6 +392,7 @@ BOOL message_send_all(int msg_type, void *buf, size_t len)
 	msg_all.msg_type = msg_type;
 	msg_all.buf = buf;
 	msg_all.len = len;
+	msg_all.duplicates = duplicates_allowed;
 
 	tdb_traverse(the_tdb, traverse_fn, NULL);
 	tdb_close(the_tdb);
diff --git a/source3/printing/printing.c b/source3/printing/printing.c
index 2d98aa9974..1eb6c27555 100644
--- a/source3/printing/printing.c
+++ b/source3/printing/printing.c
@@ -576,8 +576,7 @@ BOOL print_job_delete(struct current_user *user, int jobid, int *errcode)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	return !print_job_exists(jobid);
 }
@@ -627,8 +626,7 @@ BOOL print_job_pause(struct current_user *user, int jobid, int *errcode)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	/* how do we tell if this succeeded? */
 
@@ -678,8 +676,7 @@ BOOL print_job_resume(struct current_user *user, int jobid, int *errcode)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	return True;
 }
@@ -942,8 +939,7 @@ BOOL print_job_end(int jobid)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	return True;
 }
@@ -1125,8 +1121,7 @@ BOOL print_queue_pause(struct current_user *user, int snum, int *errcode)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	return True;
 }
@@ -1159,8 +1154,7 @@ BOOL print_queue_resume(struct current_user *user, int snum, int *errcode)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	return True;
 }
@@ -1189,8 +1183,7 @@ BOOL print_queue_purge(struct current_user *user, int snum, int *errcode)
 
 	printer_name = PRINTERNAME(snum);
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer_name, 
-			 strlen(printer_name) + 1);
+	message_send_all(MSG_PRINTER_NOTIFY, printer_name, strlen(printer_name) + 1, False);
 
 	return True;
 }
diff --git a/source3/profile/profile.c b/source3/profile/profile.c
index 80584adfa2..dce1d78a9b 100644
--- a/source3/profile/profile.c
+++ b/source3/profile/profile.c
@@ -86,7 +86,7 @@ void reqprofile_message(int msg_type, pid_t src, void *buf, size_t len)
 	level = 0;
 #endif
 	DEBUG(1,("INFO: Received REQ_PROFILELEVEL message from PID %d\n",src));
-	message_send_pid(src, MSG_PROFILELEVEL, &level, sizeof(int));
+	message_send_pid(src, MSG_PROFILELEVEL, &level, sizeof(int), True);
 }
 
 /*******************************************************************
diff --git a/source3/rpc_server/srv_spoolss_nt.c b/source3/rpc_server/srv_spoolss_nt.c
index 6a3b72ffa7..30acc14904 100644
--- a/source3/rpc_server/srv_spoolss_nt.c
+++ b/source3/rpc_server/srv_spoolss_nt.c
@@ -640,7 +640,7 @@ static BOOL srv_spoolss_sendnotify(POLICY_HND *handle)
 	/*srv_spoolss_receive_message(printer);*/
 	DEBUG(10,("srv_spoolss_sendnotify: Sending message about printer %s\n", printer ));
 
-	message_send_all(MSG_PRINTER_NOTIFY, printer, strlen(printer) + 1); /* Null terminate... */
+	message_send_all(MSG_PRINTER_NOTIFY, printer, strlen(printer) + 1, False); /* Null terminate... */
 
 	return True;
 }	
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) {