fix for bz 2816

6 files changed, 38 insertions, 46 deletions

diff --git a/runtime/stpd/ChangeLog b/runtime/stpd/ChangeLog
index a8eccab1..937668b7 100644
--- a/runtime/stpd/ChangeLog
+++ b/runtime/stpd/ChangeLog
@@ -1,3 +1,14 @@
+2006-09-25  Tom Zanussi  <zanussi@us.ibm.com>
+	
+	* librelay.c (kill_percpu_threads): Remove printf.
+	(wait_for_percpu_threads): New.
+	(process_subbufs): Remove processing, processing_mutex, exit
+	thread if exiting flag set.
+	(read_last_buffers): Removed.
+	(cleanup_and_exit): Remove call to read_last_buffers, wait for
+	threads to read flushed buffers instead.
+	(stp_main_loop): Remove mutex init.	
+
 2006-09-22  Tom Zanussi  <zanussi@us.ibm.com>
 	
 	* librelay.c (init_relayfs): Cleanup if stp_check fails.
diff --git a/runtime/stpd/librelay.c b/runtime/stpd/librelay.c
index 33dc20fa..89c3b767 100644
--- a/runtime/stpd/librelay.c
+++ b/runtime/stpd/librelay.c
@@ -79,8 +79,6 @@ static int transport_mode;
 static int ncpus;
 static int print_totals;
 static int exiting;
-static int processing;
-static pthread_mutex_t processing_mutex;
 
 /* per-cpu data */
 static int relay_file[NR_CPUS];
@@ -277,13 +275,23 @@ static int kill_percpu_threads(int n)
 		if (pthread_cancel(reader[i]) == 0)
 			killed++;
 	}
-	if (killed != n)
-		fprintf(stderr, "WARNING: couldn't kill all per-cpu threads:  %d killed, %d total\n", killed, n);
 
 	return killed;
 }
 
 /**
+ *     wait_for_percpu_threads - wait for all threads to exit
+ *     @n: number of threads to wait for
+ */
+static void wait_for_percpu_threads(int n)
+{
+	int i;
+
+	for (i = 0; i < n; i++)
+		pthread_join(reader[i], NULL);
+}
+
+/**
  *	process_subbufs - write ready subbufs to disk
  */
 static int process_subbufs(struct buf_info *info)
@@ -345,10 +353,6 @@ static void *reader_thread(void *data)
 
 		rc = read(proc_file[cpu], &status[cpu].info,
 			  sizeof(struct buf_info));
-		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
-		pthread_mutex_lock(&processing_mutex);
-		processing++;
-		pthread_mutex_unlock(&processing_mutex);
 		subbufs_consumed = process_subbufs(&status[cpu].info);
 		if (subbufs_consumed) {
 			if (subbufs_consumed > status[cpu].max_backlog)
@@ -359,34 +363,11 @@ static void *reader_thread(void *data)
 			if (write (proc_file[cpu], &consumed_info, sizeof(struct consumed_info)) < 0)
 				fprintf(stderr,"WARNING: writing consumed info failed.\n");
 		}
-		pthread_mutex_lock(&processing_mutex);
-		processing--;
-		pthread_mutex_unlock(&processing_mutex);
-		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
+		if (status[cpu].info.flushing)
+			pthread_exit(NULL);
 	} while (1);
 }
 
-static void read_last_buffers(void)
-{
-	int cpu, rc;
-	struct consumed_info consumed_info;
-	unsigned subbufs_consumed;
-
-	for (cpu = 0; cpu < ncpus; cpu++) {
-		rc = read (proc_file[cpu], &status[cpu].info, sizeof(struct buf_info));
-		subbufs_consumed = process_subbufs(&status[cpu].info);
-		if (subbufs_consumed) {
-			if (subbufs_consumed > status[cpu].max_backlog)
-				status[cpu].max_backlog = subbufs_consumed;
-			status[cpu].info.consumed += subbufs_consumed;
-			consumed_info.cpu = cpu;
-			consumed_info.consumed = subbufs_consumed;
-			if (write (proc_file[cpu], &consumed_info, sizeof(struct consumed_info)) < 0)
-				fprintf(stderr,"WARNING: writing consumed info failed.\n");
-		}
-	}
-}
-
 #define RELAYFS_MAGIC			0xF0B4A981
 #define DEBUGFS_MAGIC			0x64626720
 /**
@@ -690,18 +671,8 @@ static void cleanup_and_exit (int closed)
 		fprintf(stderr,"\nWaititing for processes to exit\n");
 	while(wait(NULL) > 0);
 
-	if (transport_mode == STP_TRANSPORT_RELAYFS && relay_file[0] > 0) {
-		kill_percpu_threads(ncpus);
-		while(1) {
-			pthread_mutex_lock(&processing_mutex);
-			if (!processing) {
-				pthread_mutex_unlock(&processing_mutex);
-				break;
-			}
-			pthread_mutex_unlock(&processing_mutex);
-		}
-		read_last_buffers();
-	}
+	if (transport_mode == STP_TRANSPORT_RELAYFS && relay_file[0] > 0)
+		wait_for_percpu_threads(ncpus);
 
 	close_proc_files();
 
@@ -768,7 +739,6 @@ int stp_main_loop(void)
 	FILE *ofp = stdout;
 
 	setvbuf(ofp, (char *)NULL, _IOLBF, 0);
-	pthread_mutex_init(&processing_mutex, NULL);
 
 	signal(SIGINT, sigproc);
 	signal(SIGTERM, sigproc);
diff --git a/runtime/transport/ChangeLog b/runtime/transport/ChangeLog
index 83ccda7d..71fc2fdb 100644
--- a/runtime/transport/ChangeLog
+++ b/runtime/transport/ChangeLog
@@ -1,3 +1,9 @@
+2006-09-25  Tom Zanussi  <zanussi@us.ibm.com>
+	
+	* procfs.c (_stp_proc_read): Set buf_info flushing flag.
+	* transport.c (_stp_cleanup_and_exit): Remove braces.
+	* transport_msgs.h: Add flushing flag to buf_info.
+
 2006-09-22  Martin Hunt  <hunt@redhat.com>
 
 	* transport.c (_stp_work_queue): Reenable some cleanup
diff --git a/runtime/transport/procfs.c b/runtime/transport/procfs.c
index 90ceb58b..b0b03a73 100644
--- a/runtime/transport/procfs.c
+++ b/runtime/transport/procfs.c
@@ -18,6 +18,7 @@ spinlock_t _stp_pool_lock = SPIN_LOCK_UNLOCKED;
 spinlock_t _stp_ready_lock = SPIN_LOCK_UNLOCKED;
 
 #ifdef STP_RELAYFS
+extern int _stp_relay_flushing;
 /* handle the per-cpu subbuf info read for relayfs */
 static ssize_t
 _stp_proc_read (struct file *file, char __user *buf, size_t count, loff_t *ppos)
@@ -38,6 +39,7 @@ _stp_proc_read (struct file *file, char __user *buf, size_t count, loff_t *ppos)
 	out.produced = atomic_read(&_stp_chan->buf[cpu]->subbufs_produced);
 	out.consumed = atomic_read(&_stp_chan->buf[cpu]->subbufs_consumed);
 #endif  /* RELAYFS_CHANNEL_VERSION >= 4 || CONFIG_RELAY */
+	out.flushing = _stp_relay_flushing;
 
 	num = sizeof(out);
 	if (copy_to_user(buf, &out, num))
diff --git a/runtime/transport/transport.c b/runtime/transport/transport.c
index 9277afca..c0684ae4 100644
--- a/runtime/transport/transport.c
+++ b/runtime/transport/transport.c
@@ -22,6 +22,7 @@
 #include "relayfs.c"
 static struct rchan *_stp_chan;
 static struct dentry *_stp_dir;
+int _stp_relay_flushing = 0;
 #endif
 
 static atomic_t _stp_start_finished = ATOMIC_INIT (0);
@@ -144,6 +145,7 @@ static void _stp_cleanup_and_exit (int dont_rmmod)
 
 #ifdef STP_RELAYFS
 		if (_stp_transport_mode == STP_TRANSPORT_RELAYFS) {
+			_stp_relay_flushing = 1;
 			relay_flush(_stp_chan);
 		}
 #endif
diff --git a/runtime/transport/transport_msgs.h b/runtime/transport/transport_msgs.h
index a060b414..6d0b2c34 100644
--- a/runtime/transport/transport_msgs.h
+++ b/runtime/transport/transport_msgs.h
@@ -24,6 +24,7 @@ struct buf_info
 	int cpu;
 	unsigned produced;
 	unsigned consumed;
+	int flushing;
 };
 
 struct consumed_info