Re: Fork-based version of pgbench

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++;