diff -cpr head/contrib/pgbench/pgbench.c pgbench-mt/contrib/pgbench/pgbench.c
*** head/contrib/pgbench/pgbench.c	2009-06-11 23:48:51.000000000 +0900
--- pgbench-mt/contrib/pgbench/pgbench.c	2009-07-24 10:11:07.625570376 +0900
***************
*** 30,35 ****
--- 30,36 ----
  
  #include "libpq-fe.h"
  #include "pqsignal.h"
+ #include "portability/instr_time.h"
  
  #include <ctype.h>
  
***************
*** 55,60 ****
--- 56,95 ----
  #include <sys/resource.h>		/* for getrlimit */
  #endif
  
+ #ifndef INT64_MAX
+ #define INT64_MAX	INT64CONST(0x7FFFFFFFFFFFFFFF)
+ #endif
+ 
+ /*
+  * Multi-platform pthread implementations
+  */
+ 
+ #ifdef WIN32
+ /* Use native win32 threads on Windows */
+ typedef struct win32_pthread   *pthread_t;
+ typedef int						pthread_attr_t;
+ static int pthread_create(pthread_t *thread, pthread_attr_t *attr, void * (*start_routine)(void *), void *arg);
+ static int pthread_join(pthread_t th, void **thread_return);
+ 
+ #elif defined(ENABLE_THREAD_SAFETY)
+ /* Use platform-dependent pthread */
+ #include <pthread.h>
+ 
+ #else
+ 
+ #include <sys/wait.h>
+ /* Use emulation with fork. Rename pthread idendifiers to avoid conflictions */
+ #define pthread_t				pg_pthread_t
+ #define pthread_attr_t			pg_pthread_attr_t
+ #define pthread_create			pg_pthread_create
+ #define pthread_join			pg_pthread_join
+ typedef struct fork_pthread	   *pthread_t;
+ typedef int						pthread_attr_t;
+ static int pthread_create(pthread_t *thread, pthread_attr_t *attr, void * (*start_routine)(void *), void *arg);
+ static int pthread_join(pthread_t th, void **thread_return);
+ 
+ #endif
+ 
  extern char *optarg;
  extern int	optind;
  
*************** extern int	optind;
*** 71,77 ****
  
  #define DEFAULT_NXACTS	10		/* default nxacts */
  
- int			nclients = 1;		/* default number of simulated clients */
  int			nxacts = 0;			/* number of transactions per client */
  int			duration = 0;		/* duration in seconds */
  
--- 106,111 ----
*************** FILE	   *LOGFILE = NULL;
*** 99,106 ****
  
  bool		use_log;			/* log transaction latencies to a file */
  
- int			remains;			/* number of remaining clients */
- 
  int			is_connect;			/* establish connection  for each transaction */
  
  char	   *pghost = "";
--- 133,138 ----
*************** typedef struct
*** 135,149 ****
  	int			listen;			/* 0 indicates that an async query has been
  								 * sent */
  	int			sleeping;		/* 1 indicates that the client is napping */
! 	struct timeval until;		/* napping until */
  	Variable   *variables;		/* array of variable definitions */
  	int			nvariables;
! 	struct timeval txn_begin;	/* used for measuring latencies */
  	int			use_file;		/* index in sql_files for this client */
  	bool		prepared[MAX_FILES];
  } CState;
  
  /*
   * queries read from files
   */
  #define SQL_COMMAND		1
--- 167,200 ----
  	int			listen;			/* 0 indicates that an async query has been
  								 * sent */
  	int			sleeping;		/* 1 indicates that the client is napping */
! 	int64		until;			/* napping until (usec) */
  	Variable   *variables;		/* array of variable definitions */
  	int			nvariables;
! 	instr_time	txn_begin;		/* used for measuring latencies */
  	int			use_file;		/* index in sql_files for this client */
  	bool		prepared[MAX_FILES];
  } CState;
  
  /*
+  * Thread state and result
+  */
+ typedef struct
+ {
+ 	pthread_t		thread;		/* thread handle */
+ 	CState		   *state;		/* array of CState */
+ 	int				nstate;		/* length of state */
+ 	instr_time		start_time;	/* thread start time */
+ } TState;
+ 
+ #define INVALID_THREAD		((pthread_t) 0)
+ 
+ typedef struct
+ {
+ 	instr_time		conn_time;
+ 	int				xacts;
+ } TResult;
+ 
+ /*
   * queries read from files
   */
  #define SQL_COMMAND		1
*************** typedef struct
*** 168,175 ****
  	char	   *argv[MAX_ARGS]; /* command list */
  } Command;
  
! Command   **sql_files[MAX_FILES];		/* SQL script files */
! int			num_files;			/* number of script files */
  
  /* default scenario */
  static char *tpc_b = {
--- 219,227 ----
  	char	   *argv[MAX_ARGS]; /* command list */
  } Command;
  
! static Command	  **sql_files[MAX_FILES];	/* SQL script files */
! static int			num_files;				/* number of script files */
! static int			debug = 0;				/* debug flag */
  
  /* default scenario */
  static char *tpc_b = {
*************** static char *select_only = {
*** 212,255 ****
  	"SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
  };
  
- /* Connection overhead time */
- static struct timeval conn_total_time = {0, 0};
- 
  /* Function prototypes */
  static void setalarm(int seconds);
! 
! 
! /* Calculate total time */
! static void
! addTime(struct timeval * t1, struct timeval * t2, struct timeval * result)
! {
! 	int			sec = t1->tv_sec + t2->tv_sec;
! 	int			usec = t1->tv_usec + t2->tv_usec;
! 
! 	if (usec >= 1000000)
! 	{
! 		usec -= 1000000;
! 		sec++;
! 	}
! 	result->tv_sec = sec;
! 	result->tv_usec = usec;
! }
! 
! /* Calculate time difference */
! static void
! diffTime(struct timeval * t1, struct timeval * t2, struct timeval * result)
! {
! 	int			sec = t1->tv_sec - t2->tv_sec;
! 	int			usec = t1->tv_usec - t2->tv_usec;
! 
! 	if (usec < 0)
! 	{
! 		usec += 1000000;
! 		sec--;
! 	}
! 	result->tv_sec = sec;
! 	result->tv_usec = usec;
! }
  
  static void
  usage(const char *progname)
--- 264,272 ----
  	"SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
  };
  
  /* Function prototypes */
  static void setalarm(int seconds);
! static void* threadRun(void *arg);
  
  static void
  usage(const char *progname)
*************** usage(const char *progname)
*** 267,272 ****
--- 284,290 ----
  		   "  -D VARNAME=VALUE\n"
  		   "               define variable for use by custom script\n"
  		   "  -f FILENAME  read transaction script from FILENAME\n"
+ 		   "  -j NUM       number of threads (default: 1)\n"
  		   "  -l           write transaction times to log file\n"
  		   "  -M {simple|extended|prepared}\n"
  		   "               protocol for submitting queries to server (default: simple)\n"
*************** discard_response(CState *state)
*** 376,404 ****
  	} while (res);
  }
  
- /* check to see if the SQL result was good */
- static int
- check(CState *state, PGresult *res, int n)
- {
- 	CState	   *st = &state[n];
- 
- 	switch (PQresultStatus(res))
- 	{
- 		case PGRES_COMMAND_OK:
- 		case PGRES_TUPLES_OK:
- 			/* OK */
- 			break;
- 		default:
- 			fprintf(stderr, "Client %d aborted in state %d: %s",
- 					n, st->state, PQerrorMessage(st->con));
- 			remains--;			/* I've aborted */
- 			PQfinish(st->con);
- 			st->con = NULL;
- 			return (-1);
- 	}
- 	return (0);					/* OK */
- }
- 
  static int
  compareVariables(const void *v1, const void *v2)
  {
--- 394,399 ----
*************** preparedStatementName(char *buffer, int 
*** 595,605 ****
  	sprintf(buffer, "P%d_%d", file, state);
  }
  
! static void
! doCustom(CState *state, int n, int debug)
  {
  	PGresult   *res;
- 	CState	   *st = &state[n];
  	Command   **commands;
  
  top:
--- 590,613 ----
  	sprintf(buffer, "P%d_%d", file, state);
  }
  
! static bool
! clientDone(CState *st, bool ok)
! {
! 	(void) ok;	/* unused */
! 
! 	if (st->con != NULL)
! 	{
! 		PQfinish(st->con);
! 		st->con = NULL;
! 	}
! 	return false;	/* always false */
! }
! 
! /* return false iff client should be disconnected */
! static bool
! doCustom(CState *st, instr_time *conn_time)
  {
  	PGresult   *res;
  	Command   **commands;
  
  top:
*************** top:
*** 607,622 ****
  
  	if (st->sleeping)
  	{							/* are we sleeping? */
! 		int			usec;
! 		struct timeval now;
  
! 		gettimeofday(&now, NULL);
! 		usec = (st->until.tv_sec - now.tv_sec) * 1000000 +
! 			st->until.tv_usec - now.tv_usec;
! 		if (usec <= 0)
  			st->sleeping = 0;	/* Done sleeping, go ahead with next command */
  		else
! 			return;				/* Still sleeping, nothing to do here */
  	}
  
  	if (st->listen)
--- 615,627 ----
  
  	if (st->sleeping)
  	{							/* are we sleeping? */
! 		instr_time	now;
  
! 		INSTR_TIME_SET_CURRENT(now);
! 		if (st->until <= INSTR_TIME_GET_MICROSEC(now))
  			st->sleeping = 0;	/* Done sleeping, go ahead with next command */
  		else
! 			return true;		/* Still sleeping, nothing to do here */
  	}
  
  	if (st->listen)
*************** top:
*** 624,640 ****
  		if (commands[st->state]->type == SQL_COMMAND)
  		{
  			if (debug)
! 				fprintf(stderr, "client %d receiving\n", n);
  			if (!PQconsumeInput(st->con))
  			{					/* there's something wrong */
! 				fprintf(stderr, "Client %d aborted in state %d. Probably the backend died while processing.\n", n, st->state);
! 				remains--;		/* I've aborted */
! 				PQfinish(st->con);
! 				st->con = NULL;
! 				return;
  			}
  			if (PQisBusy(st->con))
! 				return;			/* don't have the whole result yet */
  		}
  
  		/*
--- 629,642 ----
  		if (commands[st->state]->type == SQL_COMMAND)
  		{
  			if (debug)
! 				fprintf(stderr, "client %d receiving\n", st->id);
  			if (!PQconsumeInput(st->con))
  			{					/* there's something wrong */
! 				fprintf(stderr, "Client %d aborted in state %d. Probably the backend died while processing.\n", st->id, st->state);
! 				return clientDone(st, false);
  			}
  			if (PQisBusy(st->con))
! 				return true;	/* don't have the whole result yet */
  		}
  
  		/*
*************** top:
*** 642,666 ****
  		 */
  		if (use_log && commands[st->state + 1] == NULL)
  		{
! 			double		diff;
! 			struct timeval now;
! 
! 			gettimeofday(&now, NULL);
! 			diff = (int) (now.tv_sec - st->txn_begin.tv_sec) * 1000000.0 +
! 				(int) (now.tv_usec - st->txn_begin.tv_usec);
! 
! 			fprintf(LOGFILE, "%d %d %.0f %d %ld %ld\n",
! 					st->id, st->cnt, diff, st->use_file,
! 					(long) now.tv_sec, (long) now.tv_usec);
  		}
  
  		if (commands[st->state]->type == SQL_COMMAND)
  		{
  			res = PQgetResult(st->con);
! 			if (check(state, res, n))
  			{
! 				PQclear(res);
! 				return;
  			}
  			PQclear(res);
  			discard_response(st);
--- 644,678 ----
  		 */
  		if (use_log && commands[st->state + 1] == NULL)
  		{
! 			instr_time	diff;
! 			double		sec;
! 			double		msec;
! 			double		usec;
! 
! 			INSTR_TIME_SET_CURRENT(diff);
! 			INSTR_TIME_SUBTRACT(diff, st->txn_begin);
! 			sec = INSTR_TIME_GET_DOUBLE(diff);
! 			msec = INSTR_TIME_GET_MILLISEC(diff);
! 			usec = (double) INSTR_TIME_GET_MICROSEC(diff);
! 
! 			fprintf(LOGFILE, "%d %d %.0f %d %.0f %.0f\n",
! 					st->id, st->cnt, usec, st->use_file,
! 					sec, usec - sec * 1000.0);
  		}
  
  		if (commands[st->state]->type == SQL_COMMAND)
  		{
  			res = PQgetResult(st->con);
! 			switch (PQresultStatus(res))
  			{
! 				case PGRES_COMMAND_OK:
! 				case PGRES_TUPLES_OK:
! 					break;	/* OK */
! 				default:
! 					fprintf(stderr, "Client %d aborted in state %d: %s",
! 						st->id, st->state, PQerrorMessage(st->con));
! 					PQclear(res);
! 					return clientDone(st, false);
  			}
  			PQclear(res);
  			discard_response(st);
*************** top:
*** 676,690 ****
  
  			++st->cnt;
  			if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
! 			{
! 				remains--;		/* I've done */
! 				if (st->con != NULL)
! 				{
! 					PQfinish(st->con);
! 					st->con = NULL;
! 				}
! 				return;
! 			}
  		}
  
  		/* increment state counter */
--- 688,694 ----
  
  			++st->cnt;
  			if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
! 				return clientDone(st, true);	/* exit success */
  		}
  
  		/* increment state counter */
*************** top:
*** 699,725 ****
  
  	if (st->con == NULL)
  	{
! 		struct timeval t1,
! 					t2,
! 					t3;
  
! 		gettimeofday(&t1, NULL);
  		if ((st->con = doConnect()) == NULL)
  		{
! 			fprintf(stderr, "Client %d aborted in establishing connection.\n",
! 					n);
! 			remains--;			/* I've aborted */
! 			PQfinish(st->con);
! 			st->con = NULL;
! 			return;
  		}
! 		gettimeofday(&t2, NULL);
! 		diffTime(&t2, &t1, &t3);
! 		addTime(&conn_total_time, &t3, &conn_total_time);
  	}
  
  	if (use_log && st->state == 0)
! 		gettimeofday(&(st->txn_begin), NULL);
  
  	if (commands[st->state]->type == SQL_COMMAND)
  	{
--- 703,722 ----
  
  	if (st->con == NULL)
  	{
! 		instr_time	start, end;
  
! 		INSTR_TIME_SET_CURRENT(start);
  		if ((st->con = doConnect()) == NULL)
  		{
! 			fprintf(stderr, "Client %d aborted in establishing connection.\n", st->id);
! 			return clientDone(st, false);
  		}
! 		INSTR_TIME_SET_CURRENT(end);
! 		INSTR_TIME_ACCUM_DIFF(*conn_time, end, start);
  	}
  
  	if (use_log && st->state == 0)
! 		INSTR_TIME_SET_CURRENT(st->txn_begin);
  
  	if (commands[st->state]->type == SQL_COMMAND)
  	{
*************** top:
*** 735,745 ****
  			{
  				fprintf(stderr, "out of memory\n");
  				st->ecnt++;
! 				return;
  			}
  
  			if (debug)
! 				fprintf(stderr, "client %d sending %s\n", n, sql);
  			r = PQsendQuery(st->con, sql);
  			free(sql);
  		}
--- 732,742 ----
  			{
  				fprintf(stderr, "out of memory\n");
  				st->ecnt++;
! 				return true;
  			}
  
  			if (debug)
! 				fprintf(stderr, "client %d sending %s\n", st->id, sql);
  			r = PQsendQuery(st->con, sql);
  			free(sql);
  		}
*************** top:
*** 751,757 ****
  			getQueryParams(st, command, params);
  
  			if (debug)
! 				fprintf(stderr, "client %d sending %s\n", n, sql);
  			r = PQsendQueryParams(st->con, sql, command->argc - 1,
  								  NULL, params, NULL, NULL, 0);
  		}
--- 748,754 ----
  			getQueryParams(st, command, params);
  
  			if (debug)
! 				fprintf(stderr, "client %d sending %s\n", st->id, sql);
  			r = PQsendQueryParams(st->con, sql, command->argc - 1,
  								  NULL, params, NULL, NULL, 0);
  		}
*************** top:
*** 785,791 ****
  			preparedStatementName(name, st->use_file, st->state);
  
  			if (debug)
! 				fprintf(stderr, "client %d sending %s\n", n, name);
  			r = PQsendQueryPrepared(st->con, name, command->argc - 1,
  									params, NULL, NULL, 0);
  		}
--- 782,788 ----
  			preparedStatementName(name, st->use_file, st->state);
  
  			if (debug)
! 				fprintf(stderr, "client %d sending %s\n", st->id, name);
  			r = PQsendQueryPrepared(st->con, name, command->argc - 1,
  									params, NULL, NULL, 0);
  		}
*************** top:
*** 795,801 ****
  		if (r == 0)
  		{
  			if (debug)
! 				fprintf(stderr, "client %d cannot send %s\n", n, command->argv[0]);
  			st->ecnt++;
  		}
  		else
--- 792,798 ----
  		if (r == 0)
  		{
  			if (debug)
! 				fprintf(stderr, "client %d cannot send %s\n", st->id, command->argv[0]);
  			st->ecnt++;
  		}
  		else
*************** top:
*** 809,815 ****
  
  		if (debug)
  		{
! 			fprintf(stderr, "client %d executing \\%s", n, argv[0]);
  			for (i = 1; i < argc; i++)
  				fprintf(stderr, " %s", argv[i]);
  			fprintf(stderr, "\n");
--- 806,812 ----
  
  		if (debug)
  		{
! 			fprintf(stderr, "client %d executing \\%s", st->id, argv[0]);
  			for (i = 1; i < argc; i++)
  				fprintf(stderr, " %s", argv[i]);
  			fprintf(stderr, "\n");
*************** top:
*** 828,834 ****
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
  					st->ecnt++;
! 					return;
  				}
  				min = atoi(var);
  			}
--- 825,831 ----
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
  					st->ecnt++;
! 					return true;
  				}
  				min = atoi(var);
  			}
*************** top:
*** 850,856 ****
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[3]);
  					st->ecnt++;
! 					return;
  				}
  				max = atoi(var);
  			}
--- 847,853 ----
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[3]);
  					st->ecnt++;
! 					return true;
  				}
  				max = atoi(var);
  			}
*************** top:
*** 861,867 ****
  			{
  				fprintf(stderr, "%s: invalid maximum number %d\n", argv[0], max);
  				st->ecnt++;
! 				return;
  			}
  
  #ifdef DEBUG
--- 858,864 ----
  			{
  				fprintf(stderr, "%s: invalid maximum number %d\n", argv[0], max);
  				st->ecnt++;
! 				return true;
  			}
  
  #ifdef DEBUG
*************** top:
*** 873,879 ****
  			{
  				fprintf(stderr, "%s: out of memory\n", argv[0]);
  				st->ecnt++;
! 				return;
  			}
  
  			st->listen = 1;
--- 870,876 ----
  			{
  				fprintf(stderr, "%s: out of memory\n", argv[0]);
  				st->ecnt++;
! 				return true;
  			}
  
  			st->listen = 1;
*************** top:
*** 891,897 ****
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
  					st->ecnt++;
! 					return;
  				}
  				ope1 = atoi(var);
  			}
--- 888,894 ----
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
  					st->ecnt++;
! 					return true;
  				}
  				ope1 = atoi(var);
  			}
*************** top:
*** 908,914 ****
  					{
  						fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[4]);
  						st->ecnt++;
! 						return;
  					}
  					ope2 = atoi(var);
  				}
--- 905,911 ----
  					{
  						fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[4]);
  						st->ecnt++;
! 						return true;
  					}
  					ope2 = atoi(var);
  				}
*************** top:
*** 927,933 ****
  					{
  						fprintf(stderr, "%s: division by zero\n", argv[0]);
  						st->ecnt++;
! 						return;
  					}
  					snprintf(res, sizeof(res), "%d", ope1 / ope2);
  				}
--- 924,930 ----
  					{
  						fprintf(stderr, "%s: division by zero\n", argv[0]);
  						st->ecnt++;
! 						return true;
  					}
  					snprintf(res, sizeof(res), "%d", ope1 / ope2);
  				}
*************** top:
*** 935,941 ****
  				{
  					fprintf(stderr, "%s: unsupported operator %s\n", argv[0], argv[3]);
  					st->ecnt++;
! 					return;
  				}
  			}
  
--- 932,938 ----
  				{
  					fprintf(stderr, "%s: unsupported operator %s\n", argv[0], argv[3]);
  					st->ecnt++;
! 					return true;
  				}
  			}
  
*************** top:
*** 943,949 ****
  			{
  				fprintf(stderr, "%s: out of memory\n", argv[0]);
  				st->ecnt++;
! 				return;
  			}
  
  			st->listen = 1;
--- 940,946 ----
  			{
  				fprintf(stderr, "%s: out of memory\n", argv[0]);
  				st->ecnt++;
! 				return true;
  			}
  
  			st->listen = 1;
*************** top:
*** 952,958 ****
  		{
  			char	   *var;
  			int			usec;
! 			struct timeval now;
  
  			if (*argv[1] == ':')
  			{
--- 949,955 ----
  		{
  			char	   *var;
  			int			usec;
! 			instr_time now;
  
  			if (*argv[1] == ':')
  			{
*************** top:
*** 960,966 ****
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[1]);
  					st->ecnt++;
! 					return;
  				}
  				usec = atoi(var);
  			}
--- 957,963 ----
  				{
  					fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[1]);
  					st->ecnt++;
! 					return true;
  				}
  				usec = atoi(var);
  			}
*************** top:
*** 977,985 ****
  			else
  				usec *= 1000000;
  
! 			gettimeofday(&now, NULL);
! 			st->until.tv_sec = now.tv_sec + (now.tv_usec + usec) / 1000000;
! 			st->until.tv_usec = (now.tv_usec + usec) % 1000000;
  			st->sleeping = 1;
  
  			st->listen = 1;
--- 974,981 ----
  			else
  				usec *= 1000000;
  
! 			INSTR_TIME_SET_CURRENT(now);
! 			st->until = INSTR_TIME_GET_MICROSEC(now) + usec;
  			st->sleeping = 1;
  
  			st->listen = 1;
*************** top:
*** 987,1004 ****
  
  		goto top;
  	}
  }
  
  /* discard connections */
  static void
! disconnect_all(CState *state)
  {
  	int			i;
  
! 	for (i = 0; i < nclients; i++)
  	{
  		if (state[i].con)
  			PQfinish(state[i].con);
  	}
  }
  
--- 983,1005 ----
  
  		goto top;
  	}
+ 
+ 	return true;
  }
  
  /* discard connections */
  static void
! disconnect_all(CState *state, int length)
  {
  	int			i;
  
! 	for (i = 0; i < length; i++)
  	{
  		if (state[i].con)
+ 		{
  			PQfinish(state[i].con);
+ 			state[i].con = NULL;
+ 		}
  	}
  }
  
*************** process_commands(char *buf)
*** 1264,1269 ****
--- 1265,1288 ----
  				return NULL;
  			}
  
+ 			/*
+ 			 * Split argument into number and unit for "sleep 1ms" or so.
+ 			 * We don't have to terminate the number argument with null
+ 			 * because it will parsed with atoi, that ignores trailing
+ 			 * non-digit characters.
+ 			 */
+ 			if (my_commands->argv[1][0] != ':')
+ 			{
+ 				char	*c = my_commands->argv[1];
+ 				while (isdigit(*c)) { c++; }
+ 				if (*c)
+ 				{
+ 					my_commands->argv[2] = c;
+ 					if (my_commands->argc < 3)
+ 						my_commands->argc = 3;
+ 				}
+ 			}
+ 
  			if (my_commands->argc >= 3)
  			{
  				if (pg_strcasecmp(my_commands->argv[2], "us") != 0 &&
*************** process_builtin(char *tb)
*** 1450,1474 ****
  
  /* print out results */
  static void
! printResults(
! 			 int ttype, CState *state,
! 			 struct timeval * start_time, struct timeval * end_time)
  {
! 	double		t1,
! 				t2;
! 	int			i;
! 	int			normal_xacts = 0;
  	char	   *s;
  
! 	for (i = 0; i < nclients; i++)
! 		normal_xacts += state[i].cnt;
! 
! 	t1 = (end_time->tv_sec - start_time->tv_sec) * 1000000.0 + (end_time->tv_usec - start_time->tv_usec);
! 	t1 = normal_xacts * 1000000.0 / t1;
! 
! 	t2 = (end_time->tv_sec - start_time->tv_sec - conn_total_time.tv_sec) * 1000000.0 +
! 		(end_time->tv_usec - start_time->tv_usec - conn_total_time.tv_usec);
! 	t2 = normal_xacts * 1000000.0 / t2;
  
  	if (ttype == 0)
  		s = "TPC-B (sort of)";
--- 1469,1486 ----
  
  /* print out results */
  static void
! printResults(int ttype, int normal_xacts, int nclients, int nthreads,
! 			 instr_time total_time, instr_time conn_total_time)
  {
! 	double		time_include,
! 				tps_include,
! 				tps_exclude;
  	char	   *s;
  
! 	time_include = INSTR_TIME_GET_DOUBLE(total_time);
! 	tps_include = normal_xacts / time_include;
! 	tps_exclude = normal_xacts / (time_include -
! 		(INSTR_TIME_GET_DOUBLE(conn_total_time) / nthreads));
  
  	if (ttype == 0)
  		s = "TPC-B (sort of)";
*************** printResults(
*** 1483,1488 ****
--- 1495,1501 ----
  	printf("scaling factor: %d\n", scale);
  	printf("query mode: %s\n", QUERYMODE[querymode]);
  	printf("number of clients: %d\n", nclients);
+ 	printf("number of threads: %d\n", nthreads);
  	if (duration <= 0)
  	{
  		printf("number of transactions per client: %d\n", nxacts);
*************** printResults(
*** 1495,1502 ****
  		printf("number of transactions actually processed: %d\n",
  			   normal_xacts);
  	}
! 	printf("tps = %f (including connections establishing)\n", t1);
! 	printf("tps = %f (excluding connections establishing)\n", t2);
  }
  
  
--- 1508,1515 ----
  		printf("number of transactions actually processed: %d\n",
  			   normal_xacts);
  	}
! 	printf("tps = %f (including connections establishing)\n", tps_include);
! 	printf("tps = %f (excluding connections establishing)\n", tps_exclude);
  }
  
  
*************** int
*** 1504,1532 ****
  main(int argc, char **argv)
  {
  	int			c;
  	int			is_init_mode = 0;		/* initialize mode? */
  	int			is_no_vacuum = 0;		/* no vacuum at all before testing? */
  	int			do_vacuum_accounts = 0; /* do vacuum accounts 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 */
  	char	   *filename = NULL;
  	bool		scale_given = false;
  
  	CState	   *state;			/* status of clients */
  
! 	struct timeval start_time;	/* start up time */
! 	struct timeval end_time;	/* end time */
  
  	int			i;
  
- 	fd_set		input_mask;
- 	int			nsocks;			/* return from select(2) */
- 	int			maxsock;		/* max socket number to be waited */
- 	struct timeval now;
- 	struct timeval timeout;
- 	int			min_usec;
- 
  #ifdef HAVE_GETRLIMIT
  	struct rlimit rlim;
  #endif
--- 1517,1542 ----
  main(int argc, char **argv)
  {
  	int			c;
+ 	int			nclients = 1;		/* default number of simulated clients */
+ 	int			nthreads = 1;		/* default number of threads */
  	int			is_init_mode = 0;		/* initialize mode? */
  	int			is_no_vacuum = 0;		/* no vacuum at all before testing? */
  	int			do_vacuum_accounts = 0; /* do vacuum accounts before testing? */
  	int			ttype = 0;		/* transaction type. 0: TPC-B, 1: SELECT only,
  								 * 2: skip update of branches and tellers */
  	char	   *filename = NULL;
  	bool		scale_given = false;
  
  	CState	   *state;			/* status of clients */
+ 	TState	   *threads;		/* array of thread */
  
! 	instr_time	start_time;		/* start up time */
! 	instr_time	total_time;
! 	instr_time	conn_total_time;
! 	int			total_xacts;
  
  	int			i;
  
  #ifdef HAVE_GETRLIMIT
  	struct rlimit rlim;
  #endif
*************** main(int argc, char **argv)
*** 1576,1582 ****
  
  	memset(state, 0, sizeof(*state));
  
! 	while ((c = getopt(argc, argv, "ih:nvp:dSNc:Cs:t:T:U:lf:D:F:M:")) != -1)
  	{
  		switch (c)
  		{
--- 1586,1592 ----
  
  	memset(state, 0, sizeof(*state));
  
! 	while ((c = getopt(argc, argv, "ih:nvp:dSNc:Cs:t:T:U:lf:D:F:M:j:")) != -1)
  	{
  		switch (c)
  		{
*************** main(int argc, char **argv)
*** 1629,1634 ****
--- 1639,1652 ----
  				}
  #endif   /* HAVE_GETRLIMIT */
  				break;
+ 			case 'j':	/* jobs */
+ 				nthreads = atoi(optarg);
+ 				if (nthreads <= 0)
+ 				{
+ 					fprintf(stderr, "invalid number of threads: %d\n", nthreads);
+ 					exit(1);
+ 				}
+ 				break;
  			case 'C':
  				is_connect = 1;
  				break;
*************** main(int argc, char **argv)
*** 1749,1755 ****
  	if (nxacts <= 0 && duration <= 0)
  		nxacts = DEFAULT_NXACTS;
  
! 	remains = nclients;
  
  	if (nclients > 1)
  	{
--- 1767,1777 ----
  	if (nxacts <= 0 && duration <= 0)
  		nxacts = DEFAULT_NXACTS;
  
! 	if (nclients % nthreads != 0)
! 	{
! 		fprintf(stderr, "number of clients (%d) must be a multiple number of threads (%d)\n", nclients, nthreads);
! 		exit(1);
! 	}
  
  	if (nclients > 1)
  	{
*************** main(int argc, char **argv)
*** 1767,1772 ****
--- 1789,1795 ----
  		{
  			int			j;
  
+ 			state[i].id = i;
  			for (j = 0; j < state[0].nvariables; j++)
  			{
  				if (putVariable(&state[i], state[0].variables[j].name, state[0].variables[j].value) == false)
*************** main(int argc, char **argv)
*** 1876,1908 ****
  	PQfinish(con);
  
  	/* set random seed */
! 	gettimeofday(&start_time, NULL);
! 	srandom((unsigned int) start_time.tv_usec);
! 
! 	/* get start up time */
! 	gettimeofday(&start_time, NULL);
! 
! 	/* set alarm if duration is specified. */
! 	if (duration > 0)
! 		setalarm(duration);
! 
! 	if (is_connect == 0)
! 	{
! 		struct timeval t,
! 					now;
! 
! 		/* make connections to the database */
! 		for (i = 0; i < nclients; i++)
! 		{
! 			state[i].id = i;
! 			if ((state[i].con = doConnect()) == NULL)
! 				exit(1);
! 		}
! 		/* time after connections set up */
! 		gettimeofday(&now, NULL);
! 		diffTime(&now, &start_time, &t);
! 		addTime(&conn_total_time, &t, &conn_total_time);
! 	}
  
  	/* process bultin SQL scripts */
  	switch (ttype)
--- 1899,1906 ----
  	PQfinish(con);
  
  	/* set random seed */
! 	INSTR_TIME_SET_CURRENT(start_time);
! 	srandom((unsigned int) INSTR_TIME_GET_MICROSEC(start_time));
  
  	/* process bultin SQL scripts */
  	switch (ttype)
*************** main(int argc, char **argv)
*** 1926,2065 ****
  			break;
  	}
  
  	/* send start up queries in async manner */
! 	for (i = 0; i < nclients; i++)
  	{
! 		Command   **commands = sql_files[state[i].use_file];
! 		int			prev_ecnt = state[i].ecnt;
  
! 		state[i].use_file = getrand(0, num_files - 1);
! 		doCustom(state, i, debug);
  
! 		if (state[i].ecnt > prev_ecnt && commands[state[i].state]->type == META_COMMAND)
  		{
! 			fprintf(stderr, "Client %d aborted in state %d. Execution meta-command failed.\n", i, state[i].state);
  			remains--;			/* I've aborted */
! 			PQfinish(state[i].con);
! 			state[i].con = NULL;
  		}
  	}
  
! 	for (;;)
  	{
! 		if (remains <= 0)
! 		{						/* all done ? */
! 			disconnect_all(state);
! 			/* get end time */
! 			gettimeofday(&end_time, NULL);
! 			printResults(ttype, state, &start_time, &end_time);
! 			if (LOGFILE)
! 				fclose(LOGFILE);
! 			exit(0);
! 		}
  
  		FD_ZERO(&input_mask);
  
  		maxsock = -1;
! 		min_usec = -1;
! 		for (i = 0; i < nclients; i++)
  		{
! 			Command   **commands = sql_files[state[i].use_file];
  
! 			if (state[i].sleeping)
  			{
  				int			this_usec;
- 				int			sock = PQsocket(state[i].con);
  
! 				if (min_usec < 0)
  				{
! 					gettimeofday(&now, NULL);
! 					min_usec = 0;
  				}
  
! 				this_usec = (state[i].until.tv_sec - now.tv_sec) * 1000000 +
! 					state[i].until.tv_usec - now.tv_usec;
! 
! 				if (this_usec > 0 && (min_usec == 0 || this_usec < min_usec))
  					min_usec = this_usec;
- 
- 				FD_SET		(sock, &input_mask);
- 
- 				if (maxsock < sock)
- 					maxsock = sock;
  			}
! 			else if (state[i].con && commands[state[i].state]->type != META_COMMAND)
  			{
! 				int			sock = PQsocket(state[i].con);
! 
! 				if (sock < 0)
! 				{
! 					disconnect_all(state);
! 					exit(1);
! 				}
! 				FD_SET		(sock, &input_mask);
  
! 				if (maxsock < sock)
! 					maxsock = sock;
  			}
  		}
  
! 		if (maxsock != -1)
  		{
! 			if (min_usec >= 0)
  			{
  				timeout.tv_sec = min_usec / 1000000;
  				timeout.tv_usec = min_usec % 1000000;
! 
! 				nsocks = select(maxsock + 1, &input_mask, (fd_set *) NULL,
! 								(fd_set *) NULL, &timeout);
  			}
  			else
! 				nsocks = select(maxsock + 1, &input_mask, (fd_set *) NULL,
! 								(fd_set *) NULL, (struct timeval *) NULL);
  			if (nsocks < 0)
  			{
  				if (errno == EINTR)
  					continue;
  				/* must be something wrong */
- 				disconnect_all(state);
  				fprintf(stderr, "select failed: %s\n", strerror(errno));
! 				exit(1);
! 			}
! #ifdef NOT_USED
! 			else if (nsocks == 0)
! 			{					/* timeout */
! 				fprintf(stderr, "select timeout\n");
! 				for (i = 0; i < nclients; i++)
! 				{
! 					fprintf(stderr, "client %d:state %d cnt %d ecnt %d listen %d\n",
! 							i, state[i].state, state[i].cnt, state[i].ecnt, state[i].listen);
! 				}
! 				exit(0);
  			}
- #endif
  		}
  
  		/* ok, backend returns reply */
! 		for (i = 0; i < nclients; i++)
  		{
! 			Command   **commands = sql_files[state[i].use_file];
! 			int			prev_ecnt = state[i].ecnt;
  
! 			if (state[i].con && (FD_ISSET(PQsocket(state[i].con), &input_mask)
! 						  || commands[state[i].state]->type == META_COMMAND))
  			{
! 				doCustom(state, i, debug);
  			}
  
! 			if (state[i].ecnt > prev_ecnt && commands[state[i].state]->type == META_COMMAND)
  			{
! 				fprintf(stderr, "Client %d aborted in state %d. Execution of meta-command failed.\n", i, state[i].state);
  				remains--;		/* I've aborted */
! 				PQfinish(state[i].con);
! 				state[i].con = NULL;
  			}
  		}
  	}
  }
  
  
--- 1924,2150 ----
  			break;
  	}
  
+ 	/* get start up time */
+ 	INSTR_TIME_SET_CURRENT(start_time);
+ 
+ 	/* set alarm if duration is specified. */
+ 	if (duration > 0)
+ 		setalarm(duration);
+ 
+ 	/* start threads */
+ 	threads = (TState *) malloc(sizeof(TState) * nthreads);
+ 	for (i = 0; i < nthreads; i++)
+ 	{
+ 		threads[i].state = &state[nclients / nthreads * i];
+ 		threads[i].nstate = nclients / nthreads;
+ 		INSTR_TIME_SET_CURRENT(threads[i].start_time);
+ 
+ 		/* the first thread (i = 0) is executed by main thread */
+ 		if (i > 0)
+ 		{
+ 			int err = pthread_create(&threads[i].thread, NULL, threadRun, &threads[i]);
+ 			if (err != 0 || threads[i].thread == INVALID_THREAD)
+ 			{
+ 				fprintf(stderr, "cannot create thread: %s\n", strerror(err));
+ 				exit(1);
+ 			}
+ 		}
+ 		else
+ 		{
+ 			threads[i].thread = INVALID_THREAD;
+ 		}
+ 	}
+ 
+ 	/* wait for threads and accumulate results */
+ 	total_xacts = 0;
+ 	INSTR_TIME_SET_ZERO(conn_total_time);
+ 	for (i = 0; i < nthreads; i++)
+ 	{
+ 		void *ret = NULL;
+ 
+ 		if (threads[i].thread == INVALID_THREAD)
+ 			ret = threadRun(&threads[i]);
+ 		else
+ 			pthread_join(threads[i].thread, &ret);
+ 
+ 		if (ret != NULL)
+ 		{
+ 			TResult *r = (TResult *) ret;
+ 			total_xacts += r->xacts;
+ 			INSTR_TIME_ADD(conn_total_time, r->conn_time);
+ 			free(ret);
+ 		}
+ 	}
+ 	disconnect_all(state, nclients);
+ 
+ 	/* get end time */
+ 	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;
+ }
+ 
+ static void *
+ threadRun(void *arg)
+ {
+ 	TState	   *thread = (TState *) arg;
+ 	CState	   *state = thread->state;
+ 	TResult	   *result;
+ 	instr_time	start, end;
+ 	int			nstate = thread->nstate;
+ 	int			remains = nstate;	/* number of remaining clients */
+ 	int			i;
+ 
+ 	result = malloc(sizeof(TResult));
+ 	INSTR_TIME_SET_ZERO(result->conn_time);
+ 
+ 	if (is_connect == 0)
+ 	{
+ 		/* make connections to the database */
+ 		for (i = 0; i < nstate; i++)
+ 		{
+ 			if ((state[i].con = doConnect()) == NULL)
+ 				goto done;
+ 		}
+ 	}
+ 
+ 	/* time after thread and connections set up */
+ 	INSTR_TIME_SET_CURRENT(result->conn_time);
+ 	INSTR_TIME_SUBTRACT(result->conn_time, thread->start_time);
+ 
  	/* send start up queries in async manner */
! 	for (i = 0; i < nstate; i++)
  	{
! 		CState	   *st = &state[i];
! 		Command   **commands = sql_files[st->use_file];
! 		int			prev_ecnt = st->ecnt;
  
! 		st->use_file = getrand(0, num_files - 1);
! 		if (!doCustom(st, &result->conn_time))
! 			remains--;		/* I've aborted */
  
! 		if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND)
  		{
! 			fprintf(stderr, "Client %d aborted in state %d. Execution meta-command failed.\n", i, st->state);
  			remains--;			/* I've aborted */
! 			PQfinish(st->con);
! 			st->con = NULL;
  		}
  	}
  
! 	while (remains > 0)
  	{
! 		fd_set			input_mask;
! 		int				maxsock;		/* max socket number to be waited */
! 		int64			now_usec = 0;
! 		int64			min_usec;
  
  		FD_ZERO(&input_mask);
  
  		maxsock = -1;
! 		min_usec = INT64_MAX;
! 		for (i = 0; i < nstate; i++)
  		{
! 			CState	   *st = &state[i];
! 			Command   **commands = sql_files[st->use_file];
! 			int			sock;
  
! 			if (st->sleeping)
  			{
  				int			this_usec;
  
! 				if (min_usec == INT64_MAX)
  				{
! 					instr_time	now;
! 					INSTR_TIME_SET_CURRENT(now);
! 					now_usec = INSTR_TIME_GET_MICROSEC(now);
  				}
  
! 				this_usec = st->until - now_usec;
! 				if (min_usec > this_usec)
  					min_usec = this_usec;
  			}
! 			else if (st->con == NULL)
  			{
! 				continue;
! 			}
! 			else if (commands[st->state]->type == META_COMMAND)
! 			{
! 				min_usec = 0;	/* the connection is ready to run */
! 				break;
! 			}
  
! 			sock = PQsocket(st->con);
! 			if (sock < 0)
! 			{
! 				fprintf(stderr, "bad socket: %s\n", strerror(errno));
! 				goto done;
  			}
+ 
+ 			FD_SET(sock, &input_mask);
+ 			if (maxsock < sock)
+ 				maxsock = sock;
  		}
  
! 		if (min_usec > 0 && maxsock != -1)
  		{
! 			int		nsocks;			/* return from select(2) */
! 
! 			if (min_usec != INT64_MAX)
  			{
+ 				struct timeval	timeout;
  				timeout.tv_sec = min_usec / 1000000;
  				timeout.tv_usec = min_usec % 1000000;
! 				nsocks = select(maxsock + 1, &input_mask, NULL, NULL, &timeout);
  			}
  			else
! 				nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
  			if (nsocks < 0)
  			{
  				if (errno == EINTR)
  					continue;
  				/* must be something wrong */
  				fprintf(stderr, "select failed: %s\n", strerror(errno));
! 				goto done;
  			}
  		}
  
  		/* ok, backend returns reply */
! 		for (i = 0; i < nstate; i++)
  		{
! 			CState	   *st = &state[i];
! 			Command   **commands = sql_files[st->use_file];
! 			int			prev_ecnt = st->ecnt;
  
! 			if (st->con && (FD_ISSET(PQsocket(st->con), &input_mask)
! 						  || commands[st->state]->type == META_COMMAND))
  			{
! 				if (!doCustom(st, &result->conn_time))
! 					remains--;		/* I've aborted */
  			}
  
! 			if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND)
  			{
! 				fprintf(stderr, "Client %d aborted in state %d. Execution of meta-command failed.\n", i, st->state);
  				remains--;		/* I've aborted */
! 				PQfinish(st->con);
! 				st->con = NULL;
  			}
  		}
  	}
+ 
+ done:
+ 	INSTR_TIME_SET_CURRENT(start);
+ 	disconnect_all(state, nstate);
+ 	result->xacts = 0;
+ 	for (i = 0; i < nstate; i++)
+ 		result->xacts += state[i].cnt;
+ 	INSTR_TIME_SET_CURRENT(end);
+ 	INSTR_TIME_ACCUM_DIFF(result->conn_time, end, start);
+ 	return result;
  }
  
  
*************** setalarm(int seconds)
*** 2081,2086 ****
--- 2166,2252 ----
  	pqsignal(SIGALRM, handle_sig_alarm);
  	alarm(seconds);
  }
+ 
+ #ifndef ENABLE_THREAD_SAFETY
+ 
+ /*
+  * implements pthread using fork.
+  */
+ 
+ typedef struct fork_pthread
+ {
+ 	pid_t	pid;
+ 	int		pipes[2];
+ } fork_pthread;
+ 
+ static int
+ pthread_create(pthread_t *thread,
+ 			   pthread_attr_t *attr,
+ 			   void * (*start_routine)(void *),
+ 			   void *arg)
+ {
+ 	fork_pthread   *th;
+ 	void		   *ret;
+ 
+ 	th = (fork_pthread *) malloc(sizeof(fork_pthread));
+ 	pipe(th->pipes);
+ 
+ 	th->pid = fork();
+ 	if (th->pid == -1)	/* error */
+ 	{
+ 		free(th);
+ 		return errno;
+ 	}
+ 	if (th->pid != 0)	/* parent process */
+ 	{
+ 		close(th->pipes[1]);
+ 		*thread = th;
+ 		return 0;
+ 	}
+ 
+ 	/* child process */
+ 	close(th->pipes[0]);
+ 
+ 	/* set alarm again because the child does not inherit timers */
+ 	if (duration > 0)
+ 		setalarm(duration);
+ 
+ 	ret = start_routine(arg);
+ 	write(th->pipes[1], ret, sizeof(TResult));
+ 	close(th->pipes[1]);
+ 	free(th);
+ 	exit(0);
+ }
+ 
+ static int
+ pthread_join(pthread_t th, void **thread_return)
+ {
+ 	int		status;
+ 
+ 	while (waitpid(th->pid, &status, 0) != th->pid)
+ 	{
+ 		if (errno != EINTR)
+ 			return errno;
+ 	}
+ 
+ 	if (thread_return != NULL)
+ 	{
+ 		/* assume result is TResult */
+ 		*thread_return = malloc(sizeof(TResult));
+ 		if (read(th->pipes[0], *thread_return, sizeof(TResult)) != sizeof(TResult))
+ 		{
+ 			free(*thread_return);
+ 			*thread_return = NULL;
+ 		}
+ 	}
+ 	close(th->pipes[0]);
+ 
+ 	free(th);
+ 	return 0;
+ }
+ 
+ #endif
+ 
  #else							/* WIN32 */
  
  static VOID CALLBACK
*************** setalarm(int seconds)
*** 2107,2110 ****
--- 2273,2342 ----
  	}
  }
  
+ /* partial pthread implementation for Windows */
+ 
+ typedef struct win32_pthread
+ {
+ 	HANDLE		handle;
+ 	void	   *(*routine)(void *);
+ 	void	   *arg;
+ 	void	   *result;
+ } win32_pthread;
+ 
+ static unsigned __stdcall
+ win32_pthread_run(void *arg)
+ {
+ 	win32_pthread *th = (win32_pthread *) arg;
+ 
+ 	th->result = th->routine(th->arg);
+ 
+ 	return 0;
+ }
+ 
+ static int
+ pthread_create(pthread_t *thread,
+ 			   pthread_attr_t *attr,
+ 			   void * (*start_routine)(void *),
+ 			   void *arg)
+ {
+ 	int				save_errno;
+ 	win32_pthread   *th;
+ 
+ 	th = (win32_pthread *) malloc(sizeof(win32_pthread));
+ 	th->routine = start_routine;
+ 	th->arg = arg;
+ 	th->result = NULL;
+ 
+ 	th->handle = (HANDLE) _beginthreadex(NULL, 0, win32_pthread_run, th, 0, NULL);
+ 	if (th->handle == NULL)
+ 	{
+ 		save_errno = errno;
+ 		free(th);
+ 		return save_errno;
+ 	}
+ 
+ 	*thread = th;
+ 	return 0;
+ }
+ 
+ static int
+ pthread_join(pthread_t th, void **thread_return)
+ {
+ 	if (th == NULL || th->handle == NULL)
+ 		return errno = EINVAL;
+ 
+ 	if (WaitForSingleObject(th->handle, INFINITE) != WAIT_OBJECT_0)
+ 	{
+ 		_dosmaperr(GetLastError());
+ 		return errno;
+ 	}
+ 
+ 	if (thread_return)
+ 		*thread_return = th->result;
+ 
+ 	CloseHandle(th->handle);
+ 	free(th);
+ 	return 0;
+ }
+ 
  #endif   /* WIN32 */
diff -cpr head/doc/src/sgml/pgbench.sgml pgbench-mt/doc/src/sgml/pgbench.sgml
*** head/doc/src/sgml/pgbench.sgml	2009-05-08 07:01:18.000000000 +0900
--- pgbench-mt/doc/src/sgml/pgbench.sgml	2009-07-24 10:08:24.778572416 +0900
*************** pgbench <optional> <replaceable>options<
*** 172,177 ****
--- 172,185 ----
        </entry>
       </row>
       <row>
+       <entry><literal>-j</literal> <replaceable>threads</></entry>
+       <entry>
+        Number of worker threads. Clients are equally-divided into those
+        threads and executed in it. The number of clients must be a multiple
+        number of threads. Default is 1.
+       </entry>
+      </row>
+      <row>
        <entry><literal>-t</literal> <replaceable>transactions</></entry>
        <entry>
         Number of transactions each client runs.  Default is 10.