pg.ddx.io pgsql-hackers@postgresql.org mailing list archive help / color / mirror / Atom feed From: Tomas Vondra <tomas.vondra@2ndquadrant.com>
To: pgsql-hackers@postgresql.org
Subject: PATCH: pgbench - merging transaction logs
Date: Sun, 08 Mar 2015 21:29:15 +0100
Message-ID: <54FCB11B.5050907@2ndquadrant.com> (raw )
List-Unsubscribe: <mailto:majordomo@postgresql.org?body=unsub%20pgsql-hackers>
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
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
Attachments:
[text/x-diff] pgbench-merge-logs-v1.patch (12.1K, ../54FCB11B.5050907@2ndquadrant.com/2-pgbench-merge-logs-v1.patch)
download | inline diff:
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.
*/
view thread (31+ messages) latest in thread
Message-ID: <54FCB11B.5050907@2ndquadrant.com>
Permalink: ../54FCB11B.5050907@2ndquadrant.com/
Also on: postgresql.org/message-id/54FCB11B.5050907@2ndquadrant.com
copy link · copy postgr.es · copy git am
reply Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: tomas.vondra@2ndquadrant.com
Subject: Re: PATCH: pgbench - merging transaction logs
In-Reply-To: <54FCB11B.5050907@2ndquadrant.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox