@@ -299,7 +299,7 @@ typedef enum
299
299
*/
300
300
CSTATE_ABORTED ,
301
301
CSTATE_FINISHED
302
- } ConnectionStateEnum ;
302
+ } ConnectionStateEnum ;
303
303
304
304
/*
305
305
* Connection state.
@@ -4420,50 +4420,51 @@ threadRun(void *arg)
4420
4420
initStats (& aggs , INSTR_TIME_GET_DOUBLE (thread -> start_time ));
4421
4421
last = aggs ;
4422
4422
4423
- /* initialize explicitely the state machines */
4423
+ /* explicitly initialize the state machines */
4424
4424
for (i = 0 ; i < nstate ; i ++ )
4425
4425
{
4426
4426
state [i ].state = CSTATE_CHOOSE_SCRIPT ;
4427
4427
}
4428
4428
4429
+ /* loop till all clients have terminated */
4429
4430
while (remains > 0 )
4430
4431
{
4431
4432
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 */
4434
4434
int64 min_usec ;
4435
+ int64 now_usec = 0 ; /* set this only if needed */
4435
4436
4437
+ /* identify which client sockets should be checked for input */
4436
4438
FD_ZERO (& input_mask );
4437
-
4438
4439
maxsock = -1 ;
4439
4440
min_usec = PG_INT64_MAX ;
4440
4441
for (i = 0 ; i < nstate ; i ++ )
4441
4442
{
4442
4443
CState * st = & state [i ];
4443
- int sock ;
4444
4444
4445
4445
if (st -> state == CSTATE_THROTTLE && timer_exceeded )
4446
4446
{
4447
- /* interrupt client which has not started a transaction */
4447
+ /* interrupt client that has not started a transaction */
4448
4448
st -> state = CSTATE_FINISHED ;
4449
- remains -- ;
4450
4449
PQfinish (st -> con );
4451
4450
st -> con = NULL ;
4452
- continue ;
4451
+ remains -- ;
4453
4452
}
4454
4453
else if (st -> state == CSTATE_SLEEP || st -> state == CSTATE_THROTTLE )
4455
4454
{
4456
4455
/* a nap from the script, or under throttling */
4457
- int this_usec ;
4456
+ int64 this_usec ;
4458
4457
4459
- if (min_usec == PG_INT64_MAX )
4458
+ /* get current time if needed */
4459
+ if (now_usec == 0 )
4460
4460
{
4461
4461
instr_time now ;
4462
4462
4463
4463
INSTR_TIME_SET_CURRENT (now );
4464
4464
now_usec = INSTR_TIME_GET_MICROSEC (now );
4465
4465
}
4466
4466
4467
+ /* min_usec should be the minimum delay across all clients */
4467
4468
this_usec = (st -> state == CSTATE_SLEEP ?
4468
4469
st -> sleep_until : st -> txn_scheduled ) - now_usec ;
4469
4470
if (min_usec > this_usec )
@@ -4475,22 +4476,26 @@ threadRun(void *arg)
4475
4476
* waiting for result from server - nothing to do unless the
4476
4477
* socket is readable
4477
4478
*/
4478
- sock = PQsocket (st -> con );
4479
+ int sock = PQsocket (st -> con );
4480
+
4479
4481
if (sock < 0 )
4480
4482
{
4481
- fprintf (stderr , "invalid socket: %s" , PQerrorMessage (st -> con ));
4483
+ fprintf (stderr , "invalid socket: %s" ,
4484
+ PQerrorMessage (st -> con ));
4482
4485
goto done ;
4483
4486
}
4484
4487
4485
4488
FD_SET (sock , & input_mask );
4486
-
4487
4489
if (maxsock < sock )
4488
4490
maxsock = sock ;
4489
- break ;
4490
4491
}
4491
- else if (st -> state != CSTATE_ABORTED && st -> state != CSTATE_FINISHED )
4492
+ else if (st -> state != CSTATE_ABORTED &&
4493
+ st -> state != CSTATE_FINISHED )
4492
4494
{
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
+ */
4494
4499
min_usec = 0 ;
4495
4500
break ;
4496
4501
}
@@ -4515,9 +4520,10 @@ threadRun(void *arg)
4515
4520
}
4516
4521
4517
4522
/*
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.
4521
4527
*/
4522
4528
if (min_usec > 0 && maxsock != -1 )
4523
4529
{
@@ -4536,21 +4542,29 @@ threadRun(void *arg)
4536
4542
if (nsocks < 0 )
4537
4543
{
4538
4544
if (errno == EINTR )
4545
+ {
4546
+ /* On EINTR, go back to top of loop */
4539
4547
continue ;
4548
+ }
4540
4549
/* must be something wrong */
4541
4550
fprintf (stderr , "select() failed: %s\n" , strerror (errno ));
4542
4551
goto done ;
4543
4552
}
4544
4553
}
4554
+ else
4555
+ {
4556
+ /* If we didn't call select(), don't try to read any data */
4557
+ FD_ZERO (& input_mask );
4558
+ }
4545
4559
4546
4560
/* ok, advance the state machine of each connection */
4547
4561
for (i = 0 ; i < nstate ; i ++ )
4548
4562
{
4549
4563
CState * st = & state [i ];
4550
- bool ready ;
4551
4564
4552
- if (st -> state == CSTATE_WAIT_RESULT && st -> con )
4565
+ if (st -> state == CSTATE_WAIT_RESULT )
4553
4566
{
4567
+ /* don't call doCustom unless data is available */
4554
4568
int sock = PQsocket (st -> con );
4555
4569
4556
4570
if (sock < 0 )
@@ -4560,22 +4574,24 @@ threadRun(void *arg)
4560
4574
goto done ;
4561
4575
}
4562
4576
4563
- ready = FD_ISSET (sock , & input_mask );
4577
+ if (!FD_ISSET (sock , & input_mask ))
4578
+ continue ;
4564
4579
}
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 )
4571
4582
{
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 ;
4575
4585
}
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 -- ;
4576
4592
}
4577
4593
4578
- /* progress report by thread 0 for all threads */
4594
+ /* progress report is made by thread 0 for all threads */
4579
4595
if (progress && thread -> tid == 0 )
4580
4596
{
4581
4597
instr_time now_time ;
0 commit comments