Re: Fork-based version of pgbench - Mailing list pgsql-patches

On Thu, 1 Dec 2005, Tom Lane wrote:

> Argh, I'm an idiot ... a big part of the problem with the original
> fork-based pgbench is that I'd rearranged the startup code without
> noticing a data dependency.  You can't initialize the default scripts
> until you've gotten the correct value of "tps" by inspecting the
> database.  What was happening was that it was making tps = 1 always,
> which meant that every transaction tried to update the bid = 1 row
> of "branches", which resulted in strict serialization of transactions.
> Performance of the attached version is markedly better ;-)
>

I've threaded it in Win32 ... however, it does not support more than
MAXIMUM_WAIT_OBJECTS number of clients (which is at least 64 after win2k).
I think it is acceptable though. If you really need more than this number
of clients, then start several instances of pgbench.

Regards,
Qingqing

---

*** pgbench.c    Thu Dec  1 18:47:10 2005
--- pgbench-patched.c    Thu Dec  1 18:43:08 2005
***************
*** 75,80 ****
--- 75,81 ----
  bool        use_log;            /* log transaction latencies to a file */

  int            is_connect;            /* establish connection for each transaction */
+ int            debug = 0;            /* debug flag */

  char       *pghost = "";
  char       *pgport = NULL;
***************
*** 377,384 ****
  /*
   * Run a single client process.  Result is 0 if OK, 1 if error
   */
  static int
! doClient(int id, int debug)
  {
      PGconn       *con = NULL;        /* connection handle to DB */
      VariableSet    variables;
--- 378,390 ----
  /*
   * Run a single client process.  Result is 0 if OK, 1 if error
   */
+ #ifdef WIN32
+ static unsigned int __stdcall
+ doClient(void *arg)
+ #else
  static int
! doClient(int id)
! #endif
  {
      PGconn       *con = NULL;        /* connection handle to DB */
      VariableSet    variables;
***************
*** 389,394 ****
--- 395,404 ----
      struct timeval txn_begin;    /* used for measuring latencies */
      PGresult   *res;

+ #ifdef WIN32
+     int    id = (int)arg;
+ #endif
+
      variables.variables = NULL;
      variables.nvariables = 0;

***************
*** 525,530 ****
--- 535,541 ----
      /* Done with this client */
      if (con)
          PQfinish(con);
+
      return 0;
  }

***************
*** 967,973 ****
      int            is_init_mode = 0;        /* initialize mode? */
      int            is_no_vacuum = 0;        /* no vacuum at all before testing? */
      int            is_full_vacuum = 0;        /* do full vacuum before testing? */
-     int            debug = 0;        /* debug flag */
      int            ttype = 0;        /* transaction type. 0: TPC-B, 1: SELECT only,
                                   * 2: skip update of branches and tellers,
                                   * 3: custom */
--- 978,983 ----
***************
*** 979,984 ****
--- 989,997 ----
      struct timeval tv1;            /* start up time */
      struct timeval tv2;            /* end time */

+ #ifdef WIN32
+     HANDLE        hThreads[MAXIMUM_WAIT_OBJECTS];
+ #endif
  #if !(defined(__CYGWIN__) || defined(__MINGW32__))
      struct rlimit rlim;
  #endif
***************
*** 1030,1035 ****
--- 1043,1058 ----
                              nclients);
                      exit(1);
                  }
+ #ifdef WIN32
+                 /* Check if the number is beyond our capacity */
+                 if (nclients > MAXIMUM_WAIT_OBJECTS)
+                 {
+                     fprintf(stderr, "At most %d number of clients are supported "
+                                     "in one pgbench process. If you want more, start "
+                                     "another one", MAXIMUM_WAIT_OBJECTS);
+                     exit(1);
+                 }
+ #endif
  #if !(defined(__CYGWIN__) || defined(__MINGW32__))
  #ifdef RLIMIT_NOFILE            /* most platform uses RLIMIT_NOFILE */
                  if (getrlimit(RLIMIT_NOFILE, &rlim) == -1)
***************
*** 1260,1265 ****
--- 1283,1299 ----
      remains = 0;
      while (remains < nclients)
      {
+ #ifdef WIN32
+         hThreads[remains] = _beginthreadex(NULL, 0, doClient, (void *)remains, 0, NULL);
+         if (hThreads[remains] == 0)
+         {
+             /* create thread failed */
+             fprintf(stderr, "create thread failed: %d\n", (int)GetLastError());
+             exit(1);
+         }
+         else
+             remains++;
+ #else
          pid_t        result;

          result = fork();
***************
*** 1272,1284 ****
          else if (result == 0)
          {
              /* fork succeeded, in child */
!             exit(doClient(remains, debug));
          }
          else
          {
              /* fork succeeded, in parent */
              remains++;
          }
      }

      /* Wait for all the clients to finish */
--- 1306,1319 ----
          else if (result == 0)
          {
              /* fork succeeded, in child */
!             exit(doClient(remains));
          }
          else
          {
              /* fork succeeded, in parent */
              remains++;
          }
+ #endif
      }

      /* Wait for all the clients to finish */
***************
*** 1287,1293 ****
--- 1322,1348 ----
      {
          int        status;

+ #ifdef WIN32
+         DWORD    ret, dwstatus;
+         int        index;
+
+         ret = WaitForMultipleObjects(nclients, hThreads, FALSE, INFINITE);
+         switch(ret)
+         {
+         case WAIT_FAILED:
+             status = -1;
+             break;
+         default:
+             index = ret - WAIT_OBJECT_0;
+             if (!GetExitCodeThread(hThreads[index], &dwstatus))
+                 status = -1;
+             else
+                 status = dwstatus;
+             break;
+         }
+ #else
          if (wait(&status) != (pid_t) -1)
+ #endif
          {
              if (status != 0)
                  nfailed++;