Re: Ragged latency log data in multi-threaded pgbench

Takahiro Itagaki wrote:
> The log filenames are "pgbench_log.<main-process-id>.<thread-serial-number>"
> for each thread, but the first thread (including single-threaded) still uses
> "pgbench_log.<main-process-id>" for the name because of compatibility.
>

Attached is an updated version that I think is ready to commit.  Only
changes are docs--I rewrote those to improve the wording some.  The code
looked and tested fine to me.  I just added support for the new format
to pgbench-tools and am back to happily running large batches of tests
using it again.

I confirmed a few things:

-On my CentOS system, the original problem is masked if you have
"--enable-thread-safety" on; the multi-threaded output shows up without
any broken lines into the single file.  As I suspected it's only the
multi-process implementation that shows the issue here.  Since Tom
points out that's luck rather than something that should be relied upon,
I don't think that actually changes what to do here, it just explains
why this wasn't obvious in earlier testing--normally I have thread
safety on nowadays.

-Patch corrects the problem.  I took a build without thread safety on,
demonstrated the issue with its pgbench.  Apply the patch, rebuild just
pgbench, run again; new multiple log files have no issue.

-It's easy to convert existing scripts to utilize the new multiple log
format.  Right now the current idiom you're forced into using when
running pgbench scripts is to track the PID it's run as, then use
something like:

mv pgbench_log.${PID} pgbench.log

To convert to a stable filename for later processing.  Now you just use
something like this instead:

cat pgbench_log.${PID}* > pgbench.log
rm -f pgbench_log.${PID}*

And that works fine.

--
Greg Smith  2ndQuadrant US  Baltimore, MD
PostgreSQL Training, Services and Support
greg@2ndQuadrant.com   www.2ndQuadrant.us

diff --git a/contrib/pgbench/pgbench.c b/contrib/pgbench/pgbench.c
index 0019db4..28a8c84 100644
--- a/contrib/pgbench/pgbench.c
+++ b/contrib/pgbench/pgbench.c
@@ -131,11 +131,9 @@ int            fillfactor = 100;
 #define ntellers    10
 #define naccounts    100000

-FILE       *LOGFILE = NULL;
-
 bool        use_log;            /* log transaction latencies to a file */
-
-int            is_connect;            /* establish connection for each transaction */
+bool        is_connect;            /* establish connection for each transaction */
+int            main_pid;            /* main process id used in log filename */

 char       *pghost = "";
 char       *pgport = "";
@@ -183,6 +181,7 @@ typedef struct
  */
 typedef struct
 {
+    int            tid;            /* thread id */
     pthread_t    thread;            /* thread handle */
     CState       *state;            /* array of CState */
     int            nstate;            /* length of state[] */
@@ -741,7 +740,7 @@ clientDone(CState *st, bool ok)

 /* return false iff client should be disconnected */
 static bool
-doCustom(CState *st, instr_time *conn_time)
+doCustom(CState *st, instr_time *conn_time, FILE *log_file)
 {
     PGresult   *res;
     Command   **commands;
@@ -778,7 +777,7 @@ top:
         /*
          * transaction finished: record the time it took in the log
          */
-        if (use_log && commands[st->state + 1] == NULL)
+        if (log_file && commands[st->state + 1] == NULL)
         {
             instr_time    now;
             instr_time    diff;
@@ -791,12 +790,12 @@ top:

 #ifndef WIN32
             /* This is more than we really ought to know about instr_time */
-            fprintf(LOGFILE, "%d %d %.0f %d %ld %ld\n",
+            fprintf(log_file, "%d %d %.0f %d %ld %ld\n",
                     st->id, st->cnt, usec, st->use_file,
                     (long) now.tv_sec, (long) now.tv_usec);
 #else
             /* On Windows, instr_time doesn't provide a timestamp anyway */
-            fprintf(LOGFILE, "%d %d %.0f %d 0 0\n",
+            fprintf(log_file, "%d %d %.0f %d 0 0\n",
                     st->id, st->cnt, usec, st->use_file);
 #endif
         }
@@ -857,7 +856,7 @@ top:
         INSTR_TIME_ACCUM_DIFF(*conn_time, end, start);
     }

-    if (use_log && st->state == 0)
+    if (log_file && st->state == 0)
         INSTR_TIME_SET_CURRENT(st->txn_begin);

     if (commands[st->state]->type == SQL_COMMAND)
@@ -1833,7 +1832,7 @@ main(int argc, char **argv)
                 }
                 break;
             case 'C':
-                is_connect = 1;
+                is_connect = true;
                 break;
             case 's':
                 scale_given = true;
@@ -1955,6 +1954,12 @@ main(int argc, char **argv)
         exit(1);
     }

+    /*
+     * save main process id in the global variable because process id will be
+     * changed after fork.
+     */
+    main_pid = (int) getpid();
+
     if (nclients > 1)
     {
         state = (CState *) realloc(state, sizeof(CState) * nclients);
@@ -1980,20 +1985,6 @@ main(int argc, char **argv)
         }
     }

-    if (use_log)
-    {
-        char        logpath[64];
-
-        snprintf(logpath, 64, "pgbench_log.%d", (int) getpid());
-        LOGFILE = fopen(logpath, "w");
-
-        if (LOGFILE == NULL)
-        {
-            fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
-            exit(1);
-        }
-    }
-
     if (debug)
     {
         if (duration <= 0)
@@ -2111,6 +2102,7 @@ main(int argc, char **argv)
     threads = (TState *) malloc(sizeof(TState) * nthreads);
     for (i = 0; i < nthreads; i++)
     {
+        threads[i].tid = i;
         threads[i].state = &state[nclients / nthreads * i];
         threads[i].nstate = nclients / nthreads;
         INSTR_TIME_SET_CURRENT(threads[i].start_time);
@@ -2159,8 +2151,6 @@ main(int argc, char **argv)
     INSTR_TIME_SET_CURRENT(total_time);
     INSTR_TIME_SUBTRACT(total_time, start_time);
     printResults(ttype, total_xacts, nclients, nthreads, total_time, conn_total_time);
-    if (LOGFILE)
-        fclose(LOGFILE);

     return 0;
 }
@@ -2171,6 +2161,7 @@ threadRun(void *arg)
     TState       *thread = (TState *) arg;
     CState       *state = thread->state;
     TResult    *result;
+    FILE       *log_file = NULL;        /* per-thread log file */
     instr_time    start,
                 end;
     int            nstate = thread->nstate;
@@ -2180,7 +2171,25 @@ threadRun(void *arg)
     result = malloc(sizeof(TResult));
     INSTR_TIME_SET_ZERO(result->conn_time);

-    if (is_connect == 0)
+    /* open log file if requested */
+    if (use_log)
+    {
+        char        logpath[64];
+
+        if (thread->tid == 0)
+            snprintf(logpath, 64, "pgbench_log.%d", main_pid);
+        else
+            snprintf(logpath, 64, "pgbench_log.%d.%d", main_pid, thread->tid);
+        log_file = fopen(logpath, "w");
+
+        if (log_file == NULL)
+        {
+            fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
+            goto done;
+        }
+    }
+
+    if (!is_connect)
     {
         /* make connections to the database */
         for (i = 0; i < nstate; i++)
@@ -2202,7 +2211,7 @@ threadRun(void *arg)
         int            prev_ecnt = st->ecnt;

         st->use_file = getrand(0, num_files - 1);
-        if (!doCustom(st, &result->conn_time))
+        if (!doCustom(st, &result->conn_time, log_file))
             remains--;            /* I've aborted */

         if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND)
@@ -2304,7 +2313,7 @@ threadRun(void *arg)
             if (st->con && (FD_ISSET(PQsocket(st->con), &input_mask)
                             || commands[st->state]->type == META_COMMAND))
             {
-                if (!doCustom(st, &result->conn_time))
+                if (!doCustom(st, &result->conn_time, log_file))
                     remains--;    /* I've aborted */
             }

@@ -2326,6 +2335,8 @@ done:
         result->xacts += state[i].cnt;
     INSTR_TIME_SET_CURRENT(end);
     INSTR_TIME_ACCUM_DIFF(result->conn_time, end, start);
+    if (log_file)
+        fclose(log_file);
     return result;
 }

diff --git a/doc/src/sgml/pgbench.sgml b/doc/src/sgml/pgbench.sgml
index 72fb0e5..a141395 100644
--- a/doc/src/sgml/pgbench.sgml
+++ b/doc/src/sgml/pgbench.sgml
@@ -551,6 +551,16 @@ END;
    taken by each transaction to a logfile.  The logfile will be named
    <filename>pgbench_log.<replaceable>nnn</></filename>, where
    <replaceable>nnn</> is the PID of the pgbench process.
+   If the <literal>-j</> option is 2 or higher, creating multiple worker
+   threads, each will have its own log file. The first worker will use the
+   the same name for its log file as in the standard single worker case.
+   The additional log files for the other workers will be named
+   <filename>pgbench_log.<replaceable>nnn</>.<replaceable>mmm</></filename>,
+   where <replaceable>mmm</> is a sequential number for each worker starting
+   with 1.
+  </para>
+
+  <para>
    The format of the log is:

    <programlisting>