8000 Fix newly-introduced issues in pgbench. · postgres/postgres@9779bda · GitHub
[go: up one dir, main page]
Skip to content 8000

Commit 9779bda

Browse files
tglsfdctglsfdc
committed
Fix newly-introduced issues in pgbench.
The result of FD_ISSET() doesn't necessarily fit in a bool, though assigning it to one might accidentally work depending on platform and which socket FD number is being inquired of. Rewrite to test it with if(), rather than making any specific assumption about the result width, to match the way every other such call in PG is written. Don't break out of the input_mask-filling loop after finding the first client that we're waiting for results from. That mostly breaks parallel query management. Also, if we choose not to call select(), be sure to clear out any bits the mask-filling loop might have set, so that we don't accidentally call doCustom for clients we don't know have input. Doing so would likely be harmless, but it's a waste of cycles and doesn't seem to be intended. Make this_usec wide enough. (Yeah, the value would usually fit in an int, but then why are we using int64 everywhere else?) Minor cosmetic adjustments, mostly comment improvements. Problems introduced by commit 12788ae. The first issue was discovered by buildfarm testing, the others by code review.
1 parent fdc9186 commit 9779bda

File tree

1 file changed

+49
-33
lines changed
< 8000 div data-hpc="true">

1 file changed

+49
-33
lines changed

‎src/bin/pgbench/pgbench.c

Lines changed: 49 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -299,7 +299,7 @@ typedef enum
299299
*/
300300
CSTATE_ABORTED,
301301
CSTATE_FINISHED
302-
} ConnectionStateEnum;
302+
} ConnectionStateEnum;
303303

304304
/*
305305
* Connection state.
@@ -4420,50 +4420,51 @@ threadRun(void *arg)
44204420
initStats(&aggs, INSTR_TIME_GET_DOUBLE(thread->start_time));
44214421
last = aggs;
44224422

4423-
/* initialize explicitely the state machines */
4423+
/* explicitly initialize the state machines */
44244424
for (i = 0; i < nstate; i++)
44254425
{
44264426
state[i].state = CSTATE_CHOOSE_SCRIPT;
44274427
}
44284428

4429+
/* loop till all clients have terminated */
44294430
while (remains > 0)
44304431
{
44314432
fd_set input_mask;
4432-
int maxsock; /* max socket number to be waited */
4433-
int64 now_usec = 0;
4433+
int maxsock; /* max socket number to be waited for */
44344434
int64 min_usec;
4435+
int64 now_usec = 0; /* set this only if needed */
44354436

4437+
/* identify which client sockets should be checked for input */
44364438
FD_ZERO(&input_mask);
4437-
44384439
maxsock = -1;
44394440
min_usec = PG_INT64_MAX;
44404441
for (i = 0; i < nstate; i++)
44414442
{
44424443
CState *st = &state[i];
4443-
int sock;
44444444

44454445
if (st->state == CSTATE_THROTTLE && timer_exceeded)
44464446
{
4447-
/* interrupt client which has not started a transaction */
4447+
/* interrupt client that has not started a transaction */
44484448
st->state = CSTATE_FINISHED;
4449-
remains--;
44504449
PQfinish(st->con);
44514450
st->con = NULL;
4452-
continue;
4451+
remains--;
44534452
}
44544453
else if (st->state == CSTATE_SLEEP || st->state == CSTATE_THROTTLE)
44554454
{
44564455
/* a nap from the script, or under throttling */
4457-
int this_usec;
4456+
int64 this_usec;
44584457

4459-
if (min_usec == PG_INT64_MAX)
4458+
/* get current time if needed */
4459+
if (now_usec == 0)
44604460
{
44614461
instr_time now;
44624462

44634463
INSTR_TIME_SET_CURRENT(now);
44644464
now_usec = INSTR_TIME_GET_MICROSEC(now);
44654465
}
44664466

4467+
/* min_usec should be the minimum delay across all clients */
44674468
this_usec = (st->state == CSTATE_SLEEP ?
44684469
st->sleep_until : st->txn_scheduled) - now_usec;
44694470
if (min_usec > this_usec)
@@ -4475,22 +4476,26 @@ threadRun(void *arg)
44754476
* waiting for result from server - nothing to do unless the
44764477
* socket is readable
44774478
*/
4478-
sock = PQsocket(st->con);
4479+
int sock = PQsocket(st->con);
4480+
44794481
if (sock < 0)
44804482
{
4481-
fprintf(stderr, "invalid socket: %s", PQerrorMessage(st->con));
4483+
fprintf(stderr, "invalid socket: %s",
4484+
PQerrorMessage(st->con));
44824485
goto done;
44834486
}
44844487

44854488
FD_SET(sock, &input_mask);
4486-
44874489
if (maxsock < sock)
44884490
maxsock = sock;
4489-
break;
44904491
}
4491-
else if (st->state != CSTATE_ABORTED && st->state != CSTATE_FINISHED)
4492+
else if (st->state != CSTATE_ABORTED &&
4493+
st->state != CSTATE_FINISHED)
44924494
{
4493-
/* the connection is ready to run */
4495+
/*
4496+
* This client thread is ready to do something, so we don't
4497+
* want to wait. No need to examine additional clients.
4498+
*/
44944499
min_usec = 0;
44954500
break;
44964501
}
@@ -4515,9 +4520,10 @@ threadRun(void *arg)
45154520
}
45164521

45174522
/*
4518-
* Sleep until we receive data from the server, or a nap-time
4519-
* specified in the script ends, or it's time to print a progress
4520-
* report.
4523+
* If no clients are ready to execute actions, sleep until we receive
4524+
* data from the server, or a nap-time specified in the script ends,
4525+
* or it's time to print a progress report. Update input_mask to show
4526+
* which client(s) received data.
45214527
*/
45224528
if (min_usec > 0 && maxsock != -1)
45234529
{
@@ -4536,21 +4542,29 @@ threadRun(void *arg)
45364542
if (nsocks < 0)
45374543
{
45384544
if (errno == EINTR)
4545+
{
4546+
/* On EINTR, go back to top of loop */
45394547
continue;
4548+
}
45404549
/* must be something wrong */
45414550
fprintf(stderr, "select() failed: %s\n", strerror(errno));
45424551
goto done;
45434552
}
45444553
}
4554+
else
4555+
{
4556+
/* If we didn't call select(), don't try to read any data */
4557+
FD_ZERO(&input_mask);
4558+
}
45454559

45464560
/* ok, advance the state machine of each connection */
45474561
for (i = 0; i < nstate; i++)
45484562
{
45494563
CState *st = &state[i];
4550-
bool ready;
45514564

4552-
if (st->state == CSTATE_WAIT_RESULT && st->con)
4565+
if (st->state == CSTATE_WAIT_RESULT)
45534566
{
4567+
/* don't call doCustom unless data is available */
45544568
int sock = PQsocket(st->con);
45554569

45564570
if (sock < 0)
@@ -4560,22 +4574,24 @@ threadRun(void *arg)
45604574
goto done;
45614575
}
45624576

4563-
ready = FD_ISSET(sock, &input_mask);
4577+
if (!FD_ISSET(sock, &input_mask))
4578+
continue;
45644579
}
4565-
else if (st->state == CSTATE_FINISHED || st->state == CSTATE_ABORTED)
4566-
ready = false;
4567-
else
4568-
ready = true;
4569-
4570-
if (ready)
4580+
else if (st->state == CSTATE_FINISHED ||
4581+
st->state == CSTATE_ABORTED)
45714582
{
4572-
doCustom(thread, st, &aggs);
4573-
if (st->state == CSTATE_FINISHED || st->state == CSTATE_ABORTED)
4574-
remains--;
4583+
/* this client is done, no need to consider it anymore */
4584+
continue;
45754585
}
4586+
4587+
doCustom(thread, st, &aggs);
4588+
4589+
/* If doCustom changed client to finished state, reduce remains */
4590+
if (st->state == CSTATE_FINISHED || st->state == CSTATE_ABORTED)
4591+
remains--;
45764592
}
45774593

4578-
/* progress report by thread 0 for all threads */
4594+
/* progress report is made by thread 0 for all threads */
45794595
if (progress && thread->tid == 0)
45804596
{
45814597
instr_time now_time;

0 commit comments

Comments
 (0)
0