[PATCH 1/3] Signal-based file switching support for relay/ring buffer.

[Masami Hiramatsu <mhiramat@redhat.com>]

* runtime/staprun/relay_old.c (switch_outfile): New function for file
  switching.
  (reader_thread): Use switch_oldoutfile and block SIGUSR1 and SIGUSR2
  in default.
  (switchfile_handler): Send SIGUSR2 signal to reader threads for file
  switching.
  (init_relayfs): Assign switchfile_handler to SIGUSR1.
---

 runtime/staprun/relay.c |   60 ++++++++++++++++++++++++++++++++++++++---------
 1 files changed, 49 insertions(+), 11 deletions(-)

diff --git a/runtime/staprun/relay.c b/runtime/staprun/relay.c
index f4aa139..362a251 100644
--- a/runtime/staprun/relay.c
+++ b/runtime/staprun/relay.c
@@ -15,6 +15,7 @@
 int out_fd[NR_CPUS];
 static pthread_t reader[NR_CPUS];
 static int relay_fd[NR_CPUS];
+static int switch_file[NR_CPUS];
 static int bulkmode = 0;
 static volatile int stop_threads = 0;
 static time_t *time_backlog[NR_CPUS];
@@ -107,6 +108,21 @@ static int open_outfile(int fnum, int cpu, int remove_file)
 	return 0;
 }
 
+static int switch_outfile(int cpu, int *fnum)
+{
+	int remove_file = 0;
+
+	dbug(3, "thread %d switching file\n", cpu);
+	close(out_fd[cpu]);
+	*fnum += 1;
+	if (fnum_max && *fnum >= fnum_max)
+		remove_file = 1;
+	if (open_outfile(*fnum, cpu, remove_file) < 0) {
+		perr("Couldn't open file for cpu %d, exiting.", cpu);
+		return -1;
+	}
+	return 0;
+}
 /**
  *	reader_thread - per-cpu channel buffer reader
  */
@@ -122,9 +138,9 @@ static void *reader_thread(void *data)
 	struct sigaction sa;
 	off_t wsize = 0;
 	int fnum = 0;
-	int remove_file = 0;
 
 	sigemptyset(&sigs);
+	sigaddset(&sigs,SIGUSR1);
 	sigaddset(&sigs,SIGUSR2);
 	pthread_sigmask(SIG_BLOCK, &sigs, NULL);
 
@@ -156,6 +172,7 @@ static void *reader_thread(void *data)
 	pollfd.events = POLLIN;
 
         do {
+		dbug(3, "thread %d start ppoll\n", cpu);
                 rc = ppoll(&pollfd, 1, timeout, &sigs);
                 if (rc < 0) {
 			dbug(3, "cpu=%d poll=%d errno=%d\n", cpu, rc, errno);
@@ -164,25 +181,27 @@ static void *reader_thread(void *data)
 				goto error_out;
                         }
                 }
+		dbug(3, "thread %d end ppoll\n", cpu);
+		if (switch_file[cpu]) {
+			switch_file[cpu] = 0;
+			if (switch_outfile(cpu, &fnum) < 0)
+				goto error_out;
+			wsize = 0;
+		}
+
 		while ((rc = read(relay_fd[cpu], buf, sizeof(buf))) > 0) {
-			wsize += rc;
 			/* Switching file */
-			if (fsize_max && wsize > fsize_max) {
-				close(out_fd[cpu]);
-				fnum++;
-				if (fnum_max && fnum == fnum_max)
-					remove_file = 1;
-				if (open_outfile(fnum, cpu, remove_file) < 0) {
-					perr("Couldn't open file for cpu %d, exiting.", cpu);
+			if (fsize_max && wsize + rc > fsize_max) {
+				if (switch_outfile(cpu, &fnum) < 0)
 					goto error_out;
-				}
-				wsize = rc;
+				wsize = 0;
 			}
 			if (write(out_fd[cpu], buf, rc) != rc) {
 				if (errno != EPIPE)
 					perr("Couldn't write to output %d for cpu %d, exiting.", out_fd[cpu], cpu);
 				goto error_out;
 			}
+			wsize += rc;
 		}
         } while (!stop_threads);
 	dbug(3, "exiting thread for cpu %d\n", cpu);
@@ -195,6 +214,19 @@ error_out:
 	return(NULL);
 }
 
+static void switchfile_handler(int sig)
+{
+	int i;
+	dbug(3, "file switching signal %d received\n", sig);
+	for (i = 0; i < ncpus; i++) {
+		if (reader[i]) {
+			switch_file[i] = 1;
+			pthread_kill(reader[i], SIGUSR2);
+		} else
+			break;
+	}
+}
+
 /**
  *	init_relayfs - create files and threads for relayfs processing
  *
@@ -308,6 +340,12 @@ int init_relayfs(void)
 		
 	}
 	if (!load_only) {
+		struct sigaction sa;
+
+		sa.sa_handler = switchfile_handler;
+		sa.sa_flags = 0;
+		sigemptyset(&sa.sa_mask);
+		sigaction(SIGUSR1, &sa, NULL);
 		dbug(2, "starting threads\n");
 		for (i = 0; i < ncpus; i++) {
 			if (pthread_create(&reader[i], NULL, reader_thread,


-- 
Masami Hiramatsu

Software Engineer
Hitachi Computer Products (America), Inc.
Software Solutions Division

e-mail: mhiramat@redhat.com