Message-ID: <54FCB11B.5050907@2ndquadrant.com>
Date: Sun, 08 Mar 2015 21:29:15 +0100
From: Tomas Vondra <tomas.vondra@2ndquadrant.com>
Organization: 2ndQuadrant
User-Agent: Mozilla/5.0 (X11; Linux x86_64;
 rv:31.0) Gecko/20100101 Thunderbird/31.5.0
MIME-Version: 1.0
To: pgsql-hackers@postgresql.org
Subject: PATCH: pgbench - merging transaction logs
Content-Type: multipart/mixed; boundary="------------070903050900080704050808"
Precedence: bulk
Sender: pgsql-hackers-owner@postgresql.org

This is a multi-part message in MIME format.
--------------070903050900080704050808
Content-Type: text/plain; charset=iso-8859-2
Content-Transfer-Encoding: 7bit

Hi there,

attached is a patch implementing merging of pgbench logs. These logs are
written by each thread, so with N threads you get N files with names

    pgbench_log.PID
    pgbench_log.PID.1
    ...
    pgbench_log.PID.N

Before analyzing these logs, these files need to be combined. I usually
ended up wrinting ad-hoc scripts doing that, lost them, written them
again and so on over and over again.

The other disadvantage of the external scripts is that you have to pass
all the info about the logs (whether the logs are aggregated, whther
there's throttling, etc.).

So integrating this into pgbench directly seems like a better approach,
and the attached patch implements that.

With '-m' or '--merge-logs' on the command-line, the logs are merged at
the end, using a simple 2-way merge to keep the log sorted by the time
field. It should work with all the other options that influence the log
format (--rate, --aggregate-interval and --latency-limit).

I'll add this to CF 2016-06.

-- 
Tomas Vondra                http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services

--------------070903050900080704050808
Content-Type: text/x-diff;
 name="pgbench-merge-logs-v1.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="pgbench-merge-logs-v1.patch"

diff --git a/contrib/pgbench/pgbench.c b/contrib/pgbench/pgbench.c
index 706fdf5..d6ec87e 100644
--- a/contrib/pgbench/pgbench.c
+++ b/contrib/pgbench/pgbench.c
@@ -157,6 +157,11 @@ char	   *tablespace = NULL;
 char	   *index_tablespace = NULL;
 
 /*
+ * merge logs (transaction logs, aggregated logs) at the end
+ */
+bool		merge_logs = false;
+
+/*
  * end of configurable parameters
  *********************************************************************/
 
@@ -367,6 +372,10 @@ static void *threadRun(void *arg);
 static void doLog(TState *thread, CState *st, FILE *logfile, instr_time *now,
 	  AggVals *agg, bool skipped);
 
+static void merge_log_files(int agg_interval, int nthreads);
+static void merge_aggregated_logs(FILE *infile_a, FILE *infile_b, FILE *outfile);
+static void merge_simple_logs(FILE *infile_a, FILE *infile_b, FILE *outfile);
+
 static void
 usage(void)
 {
@@ -408,6 +417,7 @@ usage(void)
 		   "  -v, --vacuum-all         vacuum all four standard tables before tests\n"
 		   "  --aggregate-interval=NUM aggregate data over NUM seconds\n"
 		   "  --sampling-rate=NUM      fraction of transactions to log (e.g. 0.01 for 1%%)\n"
+		   "  -m, --merge-logs         merge logs produced by multiple threads\n"
 		   "\nCommon options:\n"
 		   "  -d, --debug              print debugging output\n"
 	  "  -h, --host=HOSTNAME      database server host or socket directory\n"
@@ -2733,6 +2743,7 @@ main(int argc, char **argv)
 		{"aggregate-interval", required_argument, NULL, 5},
 		{"rate", required_argument, NULL, 'R'},
 		{"latency-limit", required_argument, NULL, 'L'},
+		{"merge-logs", no_argument, NULL, 'm'},
 		{NULL, 0, NULL, 0}
 	};
 
@@ -2808,7 +2819,7 @@ main(int argc, char **argv)
 	state = (CState *) pg_malloc(sizeof(CState));
 	memset(state, 0, sizeof(CState));
 
-	while ((c = getopt_long(argc, argv, "ih:nvp:dqSNc:j:Crs:t:T:U:lf:D:F:M:P:R:L:", long_options, &optindex)) != -1)
+	while ((c = getopt_long(argc, argv, "ih:nvp:dqSNc:j:Crs:t:T:U:lf:D:F:M:P:R:L:m", long_options, &optindex)) != -1)
 	{
 		switch (c)
 		{
@@ -3017,6 +3028,10 @@ main(int argc, char **argv)
 					latency_limit = (int64) (limit_ms * 1000);
 				}
 				break;
+			case 'm':
+				printf("merge logs\n");
+				merge_logs = true;
+				break;
 			case 0:
 				/* This covers long options which take no argument. */
 				if (foreign_keys || unlogged_tables)
@@ -3137,6 +3152,12 @@ main(int argc, char **argv)
 		exit(1);
 	}
 
+	if (merge_logs && (! use_log))
+	{
+		fprintf(stderr, "log merging is allowed only when actually logging transactions\n");
+		exit(1);
+	}
+
 	/*
 	 * is_latencies only works with multiple threads in thread-based
 	 * implementations, not fork-based ones, because it supposes that the
@@ -3418,6 +3439,10 @@ main(int argc, char **argv)
 				 throttle_lag, throttle_lag_max, throttle_latency_skipped,
 				 latency_late);
 
+	/* Merge logs, if needed */
+	if (merge_logs)
+		merge_log_files(agg_interval, nthreads);
+
 	return 0;
 }
 
@@ -3783,6 +3808,339 @@ done:
 	return result;
 }
 
+static void
+merge_log_files(int agg_interval, int nthreads)
+{
+	int i;
+
+	/* we can do this as 2-way merges (all the logs are sorted by timestamp) */
+	for (i = 1; i < nthreads; i++)
+	{
+		char	logpath[64];
+		char	logpath_new[64];
+
+		/* input and output files */
+		FILE   *infile_a, *infile_b;
+		FILE   *outfile;
+
+		/* the first input is always the 'main' logfile */
+		snprintf(logpath, sizeof(logpath), "pgbench_log.%d", main_pid);
+		infile_a = fopen(logpath, "r");
+
+		if (infile_b == NULL)
+		{
+			fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
+			return;
+		}
+
+		snprintf(logpath, sizeof(logpath), "pgbench_log.%d.%d", main_pid, i);
+		infile_b = fopen(logpath, "r");
+
+		if (infile_b == NULL)
+		{
+			fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
+			return;
+		}
+
+		snprintf(logpath, sizeof(logpath), "pgbench_log.%d.tmp", main_pid);
+		outfile = fopen(logpath, "w");
+
+		if (outfile == NULL)
+		{
+			fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
+			return;
+		}
+
+		if (agg_interval > 0)
+			merge_aggregated_logs(infile_a, infile_b, outfile);
+		else
+			merge_simple_logs(infile_a, infile_b, outfile);
+
+		fclose(infile_a);
+		fclose(infile_b);
+		fclose(outfile);
+
+		snprintf(logpath_new, sizeof(logpath), "pgbench_log.%d", main_pid);
+		unlink(logpath_new);
+
+		snprintf(logpath, sizeof(logpath), "pgbench_log.%d.%d", main_pid, i);
+		unlink(logpath);
+
+		snprintf(logpath, sizeof(logpath), "pgbench_log.%d.tmp", main_pid);
+		rename(logpath, logpath_new);
+	}
+}
+
+static void
+merge_simple_logs(FILE *infile_a, FILE *infile_b, FILE *outfile)
+{
+	bool	fetch_a = true, fetch_b = true;
+	bool	done_a = false, done_b = false;
+
+	long	tv_sec_a, tv_sec_b;
+	long	tv_usec_a, tv_usec_b;
+	int		id_a, id_b;
+	int		cnt_a, cnt_b;
+	int		use_file_a, use_file_b;
+	double	lag_a, lag_b;
+	
+	/* latency may be 'skipped' in some cases */
+	char	latency_a[64], latency_b[64];
+
+	/* repeat until we exhaust data from both files */
+	while (! (done_a && done_b))
+	{
+		if (fetch_a)
+		{
+			if (throttle_delay)
+			{
+				if (fscanf(infile_a, "%d %d %s %d %ld %ld %lf",
+					&id_a, &cnt_a, latency_a, &use_file_a,
+					&tv_sec_a, &tv_usec_a, &lag_a) != 7)
+					done_a = true;
+			}
+			else
+				if (fscanf(infile_a, "%d %d %s %d %ld %ld",
+					&id_a, &cnt_a, latency_a, &use_file_a,
+					&tv_sec_a, &tv_usec_a) != 6)
+					done_a = true;
+
+			fetch_a = false;
+		}
+
+		if (fetch_b)
+		{
+			if (throttle_delay)
+			{
+				if (fscanf(infile_b, "%d %d %s %d %ld %ld %lf",
+					&id_b, &cnt_b, latency_b, &use_file_b,
+					&tv_sec_b, &tv_usec_b, &lag_b) != 7)
+					done_b = true;
+			}
+			else
+				if (fscanf(infile_b, "%d %d %s %d %ld %ld",
+					&id_b, &cnt_b, latency_b, &use_file_b,
+					&tv_sec_b, &tv_usec_b) != 6)
+					done_b = true;
+
+			fetch_b = false;
+		}
+
+		/* both files completed */
+		if (done_a && done_b)
+			break;
+
+		if ((!done_a) && (! done_b))
+		{
+			/* 'a' before 'b' (or at the same time) */
+			if ((tv_sec_a < tv_sec_b) || ((tv_sec_a == tv_sec_b) && (tv_usec_a <= tv_usec_b)))
+			{
+				if (throttle_delay)
+					fprintf(outfile, "%d %d %s %d %ld %ld %.0f\n",
+							id_a, cnt_a, latency_a, use_file_a,
+							tv_sec_a, tv_usec_a, lag_a);
+				else
+					fprintf(outfile, "%d %d %s %d %ld %ld\n",
+							id_a, cnt_a, latency_a, use_file_a,
+							tv_sec_a, tv_usec_a);
+				fetch_a = true;
+			}
+			else /* b after a */
+			{
+				if (throttle_delay)
+					fprintf(outfile, "%d %d %s %d %ld %ld %.0f\n",
+							id_b, cnt_b, latency_b, use_file_b,
+							tv_sec_b, tv_usec_b, lag_b);
+				else
+					fprintf(outfile, "%d %d %s %d %ld %ld\n",
+							id_b, cnt_b, latency_b, use_file_b,
+							tv_sec_b, tv_usec_b);
+				fetch_b = true;
+			}
+		}
+		else if (!done_a)
+		{
+			if (throttle_delay)
+				fprintf(outfile, "%d %d %s %d %ld %ld %.0f\n",
+						id_a, cnt_a, latency_a, use_file_a,
+						tv_sec_a, tv_usec_a, lag_a);
+			else
+				fprintf(outfile, "%d %d %s %d %ld %ld\n",
+						id_a, cnt_a, latency_a, use_file_a,
+						tv_sec_a, tv_usec_a);
+			fetch_a = true;
+		}
+		else if (!done_b)
+		{
+			if (throttle_delay)
+				fprintf(outfile, "%d %d %s %d %ld %ld %.0f\n",
+						id_b, cnt_b, latency_b, use_file_b,
+						tv_sec_b, tv_usec_b, lag_b);
+			else
+				fprintf(outfile, "%d %d %s %d %ld %ld\n",
+						id_b, cnt_b, latency_b, use_file_b,
+						tv_sec_b, tv_usec_b);
+			fetch_b = true;
+		}
+	}
+}
+
+static void
+merge_aggregated_logs(FILE *infile_a, FILE *infile_b, FILE *outfile)
+{
+	bool	fetch_a = true, fetch_b = true;
+	bool	done_a = false, done_b = false;
+
+	long	start_time_a, start_time_b;
+	int		cnt_a, cnt_b;
+
+	double	min_latency_a, min_latency_b;
+	double	max_latency_a, max_latency_b;
+	double	sum_latency_a, sum_latency_b;
+	double	sum2_latency_a, sum2_latency_b;
+
+	/* used with 'throttle_delay' */
+	double	min_lag_a, min_lag_b;
+	double	max_lag_a, max_lag_b;
+	double	sum_lag_a, sum_lag_b;
+	double	sum2_lag_a, sum2_lag_b;
+
+	/* repeat until we exhaust data from both files */
+	while (! (done_a && done_b))
+	{
+		if (fetch_a)
+		{
+			if (throttle_delay)
+			{
+				if (fscanf(infile_a, "%ld %d %lf %lf %lf %lf %lf %lf %lf %lf",
+					&start_time_a, &cnt_a, &sum_latency_a, &sum2_latency_a,
+					&min_latency_a, &max_latency_a, &sum_lag_a, &sum2_lag_a,
+					&min_lag_a, &max_lag_a) != 10)
+					done_a = true;
+			}
+			else if (fscanf(infile_a, "%ld %d %lf %lf %lf %lf",
+					&start_time_a, &cnt_a, &sum_latency_a, &sum2_latency_a,
+					&min_latency_a, &max_latency_a) != 6)
+					done_a = true;
+
+			fetch_a = false;
+		}
+
+		if (fetch_b)
+		{
+			if (throttle_delay)
+			{
+				if (fscanf(infile_b, "%ld %d %lf %lf %lf %lf %lf %lf %lf %lf",
+					&start_time_b, &cnt_b, &sum_latency_b, &sum2_latency_b,
+					&min_latency_b, &max_latency_b, &sum_lag_b, &sum2_lag_b,
+					&min_lag_b, &max_lag_b) != 10)
+					done_b = true;
+			}
+			else if (fscanf(infile_b, "%ld %d %lf %lf %lf %lf",
+					&start_time_b, &cnt_b, &sum_latency_b, &sum2_latency_b,
+					&min_latency_b, &max_latency_b) != 6)
+					done_b = true;
+
+			fetch_b = false;
+		}
+
+		/* both files completed */
+		if (done_a && done_b)
+			break;
+
+		if ((!done_a) && (! done_b))
+		{
+			/* we need to decide whether to merge the data (may contain gaps) */
+			if (start_time_a == start_time_b)
+			{
+				if (throttle_delay)
+					fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f %lf %lf %lf %lf\n",
+							start_time_a, (cnt_a + cnt_b),
+							(sum_latency_a + sum_latency_b), (sum2_latency_a + sum2_latency_b),
+							(min_latency_a < min_latency_b) ? min_latency_a : min_latency_b,
+							(max_latency_a > max_latency_b) ? max_latency_a : max_latency_b,
+							(sum_lag_a + sum_lag_b), (sum2_lag_a + sum2_lag_b),
+							(min_lag_a < min_lag_b) ? min_lag_a : min_lag_b,
+							(max_lag_a > max_lag_b) ? max_lag_a : max_lag_b);
+				else
+					fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f\n",
+							start_time_a, cnt_a + cnt_b,
+							sum_latency_a + sum_latency_b,
+							sum2_latency_a + sum2_latency_b,
+							(min_latency_a < min_latency_b) ? min_latency_a : min_latency_b,
+							(max_latency_a > max_latency_b) ? max_latency_a : max_latency_b);
+
+				fetch_a = true;
+				fetch_b = true;
+			}
+			else if (start_time_a < start_time_b)
+			{
+				if (throttle_delay)
+					fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f %lf %lf %lf %lf\n",
+							start_time_a, cnt_a,
+							sum_latency_a, sum2_latency_a,
+							min_latency_a, max_latency_a,
+							sum_lag_a, sum2_lag_a,
+							min_lag_a, max_lag_a);
+				else
+					fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f\n",
+							start_time_a, cnt_a,
+							sum_latency_a, sum2_latency_a,
+							min_latency_a, max_latency_a);
+				fetch_a = true;
+			}
+			else
+			{
+				if (throttle_delay)
+					fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f %lf %lf %lf %lf\n",
+							start_time_b, cnt_b,
+							sum_latency_b, sum2_latency_b,
+							min_latency_b, max_latency_b,
+							sum_lag_b, sum2_lag_b,
+							min_lag_b, max_lag_b);
+				else
+					fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f\n",
+							start_time_b, cnt_b,
+							sum_latency_b, sum2_latency_b,
+							min_latency_b, max_latency_b);
+				fetch_b = true;
+			}
+		}
+		else if (!done_a)
+		{
+			if (throttle_delay)
+				fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f %lf %lf %lf %lf\n",
+						start_time_a, cnt_a,
+						sum_latency_a, sum2_latency_a,
+						min_latency_a, max_latency_a,
+						sum_lag_a, sum2_lag_a,
+						min_lag_a, max_lag_a);
+			else
+				fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f\n",
+						start_time_a, cnt_a,
+						sum_latency_a, sum2_latency_a,
+						min_latency_a, max_latency_a);
+			fetch_a = true;
+		}
+		else if (!done_b)
+		{
+			if (throttle_delay)
+				fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f %lf %lf %lf %lf\n",
+						start_time_b, cnt_b,
+						sum_latency_b, sum2_latency_b,
+						min_latency_b, max_latency_b,
+						sum_lag_b, sum2_lag_b,
+						min_lag_b, max_lag_b);
+			else
+				fprintf(outfile, "%ld %d %.0f %.0f %.0f %.0f\n",
+						start_time_b, cnt_b,
+						sum_latency_b, sum2_latency_b,
+						min_latency_b, max_latency_b);
+			fetch_b = true;
+		}
+	}
+}
+
 /*
  * Support for duration option: set timer_exceeded after so many seconds.
  */

--------------070903050900080704050808
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0


-- 
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

--------------070903050900080704050808--