@@ -136,6 +136,12 @@ int unlogged_tables = 0;
136
136
*/
137
137
double sample_rate = 0.0 ;
138
138
139
+ /*
140
+ * When threads are throttled to a given rate limit, this is the target delay
141
+ * to reach that rate in usec. 0 is the default and means no throttling.
142
+ */
143
+ int64 throttle_delay = 0 ;
144
+
139
145
/*
140
146
* tablespace selection
141
147
*/
@@ -202,11 +208,13 @@ typedef struct
202
208
int listen ; /* 0 indicates that an async query has been
203
209
* sent */
204
210
int sleeping ; /* 1 indicates that the client is napping */
211
+ bool throttling ; /* whether nap is for throttling */
205
212
int64 until ; /* napping until (usec) */
206
213
Variable * variables ; /* array of variable definitions */
207
214
int nvariables ;
208
215
instr_time txn_begin ; /* used for measuring transaction latencies */
209
216
instr_time stmt_begin ; /* used for measuring statement latencies */
217
+ bool is_throttled ; /* whether transaction throttling is done */
210
218
int use_file ; /* index in sql_files for this client */
211
219
bool prepared [MAX_FILES ];
212
220
} CState ;
@@ -224,6 +232,9 @@ typedef struct
224
232
instr_time * exec_elapsed ; /* time spent executing cmds (per Command) */
225
233
int * exec_count ; /* number of cmd executions (per Command) */
226
234
unsigned short random_state [3 ]; /* separate randomness for each thread */
235
+ int64 throttle_trigger ; /* previous/next throttling (us) */
236
+ int64 throttle_lag ; /* total transaction lag behind throttling */
237
+ int64 throttle_lag_max ; /* max transaction lag */
227
238
} TState ;
228
239
229
240
#define INVALID_THREAD ((pthread_t) 0)
@@ -232,6 +243,8 @@ typedef struct
232
243
{
233
244
instr_time conn_time ;
234
245
int xacts ;
246
+ int64 throttle_lag ;
247
+ int64 throttle_lag_max ;
235
248
} TResult ;
236
249
237
250
/*
@@ -356,6 +369,7 @@ usage(void)
356
369
" -N, --skip-some-updates skip updates of pgbench_tellers and pgbench_branches\n"
357
370
" -P, --progress=NUM show thread progress report every NUM seconds\n"
358
371
" -r, --report-latencies report average latency per command\n"
372
+ " -R, --rate=SPEC target rate in transactions per second\n"
359
373
" -s, --scale=NUM report this scale factor in output\n"
360
374
" -S, --select-only perform SELECT-only transactions\n"
361
375
" -t, --transactions number of transactions each client runs "
@@ -898,17 +912,62 @@ doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile, AggVa
898
912
{
899
913
PGresult * res ;
900
914
Command * * commands ;
915
+ bool trans_needs_throttle = false;
901
916
902
917
top :
903
918
commands = sql_files [st -> use_file ];
904
919
920
+ /*
921
+ * Handle throttling once per transaction by sleeping. It is simpler
922
+ * to do this here rather than at the end, because so much complicated
923
+ * logic happens below when statements finish.
924
+ */
925
+ if (throttle_delay && ! st -> is_throttled )
926
+ {
927
+ /*
928
+ * Use inverse transform sampling to randomly generate a delay, such
929
+ * that the series of delays will approximate a Poisson distribution
930
+ * centered on the throttle_delay time.
931
+ *
932
+ * 1000 implies a 6.9 (-log(1/1000)) to 0.0 (log 1.0) delay multiplier.
933
+ *
934
+ * If transactions are too slow or a given wait is shorter than
935
+ * a transaction, the next transaction will start right away.
936
+ */
937
+ int64 wait = (int64 )
938
+ throttle_delay * - log (getrand (thread , 1 , 1000 )/1000.0 );
939
+
940
+ thread -> throttle_trigger += wait ;
941
+
942
+ st -> until = thread -> throttle_trigger ;
943
+ st -> sleeping = 1 ;
944
+ st -> throttling = true;
945
+ st -> is_throttled = true;
946
+ if (debug )
947
+ fprintf (stderr , "client %d throttling " INT64_FORMAT " us\n" ,
948
+ st -> id , wait );
949
+ }
950
+
905
951
if (st -> sleeping )
906
952
{ /* are we sleeping? */
907
953
instr_time now ;
954
+ int64 now_us ;
908
955
909
956
INSTR_TIME_SET_CURRENT (now );
910
- if (st -> until <= INSTR_TIME_GET_MICROSEC (now ))
957
+ now_us = INSTR_TIME_GET_MICROSEC (now );
958
+ if (st -> until <= now_us )
959
+ {
911
960
st -> sleeping = 0 ; /* Done sleeping, go ahead with next command */
961
+ if (st -> throttling )
962
+ {
963
+ /* Measure lag of throttled transaction relative to target */
964
+ int64 lag = now_us - st -> until ;
965
+ thread -> throttle_lag += lag ;
966
+ if (lag > thread -> throttle_lag_max )
967
+ thread -> throttle_lag_max = lag ;
968
+ st -> throttling = false;
969
+ }
970
+ }
912
971
else
913
972
return true; /* Still sleeping, nothing to do here */
914
973
}
@@ -1095,6 +1154,15 @@ doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile, AggVa
1095
1154
st -> state = 0 ;
1096
1155
st -> use_file = (int ) getrand (thread , 0 , num_files - 1 );
1097
1156
commands = sql_files [st -> use_file ];
1157
+ st -> is_throttled = false;
1158
+ /*
1159
+ * No transaction is underway anymore, which means there is nothing
1160
+ * to listen to right now. When throttling rate limits are active,
1161
+ * a sleep will happen next, as the next transaction starts. And
1162
+ * then in any case the next SQL command will set listen back to 1.
1163
+ */
1164
+ st -> listen = 0 ;
1165
+ trans_needs_throttle = (throttle_delay > 0 );
1098
1166
}
1099
1167
}
1100
1168
@@ -1113,6 +1181,16 @@ doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile, AggVa
1113
1181
INSTR_TIME_ACCUM_DIFF (* conn_time , end , start );
1114
1182
}
1115
1183
1184
+ /*
1185
+ * This ensures that a throttling delay is inserted before proceeding
1186
+ * with sql commands, after the first transaction. The first transaction
1187
+ * throttling is performed when first entering doCustom.
1188
+ */
1189
+ if (trans_needs_throttle ) {
1190
+ trans_needs_throttle = false;
1191
+ goto top ;
1192
+ }
1193
+
1116
1194
/* Record transaction start time if logging is enabled */
1117
1195
if (logfile && st -> state == 0 )
1118
1196
INSTR_TIME_SET_CURRENT (st -> txn_begin );
@@ -2017,7 +2095,8 @@ process_builtin(char *tb)
2017
2095
static void
2018
2096
printResults (int ttype , int normal_xacts , int nclients ,
2019
2097
TState * threads , int nthreads ,
2020
- instr_time total_time , instr_time conn_total_time )
2098
+ instr_time total_time , instr_time conn_total_time ,
2099
+ int64 throttle_lag , int64 throttle_lag_max )
2021
2100
{
2022
2101
double time_include ,
2023
2102
tps_include ,
@@ -2055,6 +2134,19 @@ printResults(int ttype, int normal_xacts, int nclients,
2055
2134
printf ("number of transactions actually processed: %d\n" ,
2056
2135
normal_xacts );
2057
2136
}
2137
+
2138
+ if (throttle_delay )
2139
+ {
2140
+ /*
2141
+ * Report average transaction lag under rate limit throttling. This
2142
+ * is the delay between scheduled and actual start times for the
2143
+ * transaction. The measured lag may be caused by thread/client load,
2144
+ * the database load, or the Poisson throttling process.
2145
+ */
2146
+ printf ("average rate limit schedule lag: %.3f ms (max %.3f ms)\n" ,
2147
+ 0.001 * throttle_lag / normal_xacts , 0.001 * throttle_lag_max );
2148
+ }
2149
+
2058
2150
printf ("tps = %f (including connections establishing)\n" , tps_include );
2059
2151
printf ("tps = %f (excluding connections establishing)\n" , tps_exclude );
2060
2152
@@ -2140,6 +2232,7 @@ main(int argc, char **argv)
2140
2232
{"unlogged-tables" , no_argument , & unlogged_tables , 1 },
2141
2233
{"sampling-rate" , required_argument , NULL , 4 },
2142
2234
{"aggregate-interval" , required_argument , NULL , 5 },
2235
+ {"rate" , required_argument , NULL , 'R' },
2143
2236
{NULL , 0 , NULL , 0 }
2144
2237
};
2145
2238
@@ -2162,6 +2255,8 @@ main(int argc, char **argv)
2162
2255
instr_time total_time ;
2163
2256
instr_time conn_total_time ;
2164
2257
int total_xacts ;
2258
+ int64 throttle_lag = 0 ;
2259
+ int64 throttle_lag_max = 0 ;
2165
2260
2166
2261
int i ;
2167
2262
@@ -2206,7 +2301,7 @@ main(int argc, char **argv)
2206
2301
state = (CState * ) pg_malloc (sizeof (CState ));
2207
2302
memset (state , 0 , sizeof (CState ));
2208
2303
2209
- while ((c = getopt_long (argc , argv , "ih:nvp:dqSNc:j:Crs:t:T:U:lf:D:F:M:P:" , long_options , & optindex )) != -1 )
2304
+ while ((c = getopt_long (argc , argv , "ih:nvp:dqSNc:j:Crs:t:T:U:lf:D:F:M:P:R: " , long_options , & optindex )) != -1 )
2210
2305
{
2211
2306
switch (c )
2212
2307
{
@@ -2371,6 +2466,19 @@ main(int argc, char **argv)
2371
2466
exit (1 );
2372
2467
}
2373
2468
break ;
2469
+ case 'R' :
2470
+ {
2471
+ /* get a double from the beginning of option value */
2472
+ double throttle_value = atof (optarg );
2473
+ if (throttle_value <= 0.0 )
2474
+ {
2475
+ fprintf (stderr , "invalid rate limit: %s\n" , optarg );
2476
+ exit (1 );
2477
+ }
2478
+ /* Invert rate limit into a time offset */
2479
+ throttle_delay = (int64 ) (1000000.0 / throttle_value );
2480
+ }
2481
+ break ;
2374
2482
case 0 :
2375
2483
/* This covers long options which take no argument. */
2376
2484
break ;
@@ -2408,6 +2516,9 @@ main(int argc, char **argv)
2408
2516
}
2409
2517
}
2410
2518
2519
+ /* compute a per thread delay */
2520
+ throttle_delay *= nthreads ;
2521
+
2411
2522
if (argc > optind )
2412
2523
dbName = argv [optind ];
2413
2524
else
@@ -2721,6 +2832,9 @@ main(int argc, char **argv)
2721
2832
TResult * r = (TResult * ) ret ;
2722
2833
2723
2834
total_xacts += r -> xacts ;
2835
+ throttle_lag += r -> throttle_lag ;
2836
+ if (r -> throttle_lag_max > throttle_lag_max )
2837
+ throttle_lag_max = r -> throttle_lag_max ;
2724
2838
INSTR_TIME_ADD (conn_total_time , r -> conn_time );
2725
2839
free (ret );
2726
2840
}
@@ -2731,7 +2845,7 @@ main(int argc, char **argv)
2731
2845
INSTR_TIME_SET_CURRENT (total_time );
2732
2846
INSTR_TIME_SUBTRACT (total_time , start_time );
2733
2847
printResults (ttype , total_xacts , nclients , threads , nthreads ,
2734
- total_time , conn_total_time );
2848
+ total_time , conn_total_time , throttle_lag , throttle_lag_max );
2735
2849
2736
2850
return 0 ;
2737
2851
}
@@ -2756,6 +2870,17 @@ threadRun(void *arg)
2756
2870
2757
2871
AggVals aggs ;
2758
2872
2873
+ /*
2874
+ * Initialize throttling rate target for all of the thread's clients. It
2875
+ * might be a little more accurate to reset thread->start_time here too.
2876
+ * The possible drift seems too small relative to typical throttle delay
2877
+ * times to worry about it.
2878
+ */
2879
+ INSTR_TIME_SET_CURRENT (start );
2880
+ thread -> throttle_trigger = INSTR_TIME_GET_MICROSEC (start );
2881
+ thread -> throttle_lag = 0 ;
2882
+ thread -> throttle_lag_max = 0 ;
2883
+
2759
2884
result = pg_malloc (sizeof (TResult ));
2760
2885
2761
2886
INSTR_TIME_SET_ZERO (result -> conn_time );
@@ -2831,25 +2956,38 @@ threadRun(void *arg)
2831
2956
Command * * commands = sql_files [st -> use_file ];
2832
2957
int sock ;
2833
2958
2834
- if (st -> sleeping )
2959
+ if (st -> con == NULL )
2835
2960
{
2836
- int this_usec ;
2837
-
2838
- if (min_usec == INT64_MAX )
2961
+ continue ;
2962
+ }
2963
+ else if (st -> sleeping )
2964
+ {
2965
+ if (st -> throttling && timer_exceeded )
2839
2966
{
2840
- instr_time now ;
2841
-
2842
- INSTR_TIME_SET_CURRENT (now );
2843
- now_usec = INSTR_TIME_GET_MICROSEC (now );
2967
+ /* interrupt client which has not started a transaction */
2968
+ remains -- ;
2969
+ st -> sleeping = 0 ;
2970
+ st -> throttling = false;
2971
+ PQfinish (st -> con );
2972
+ st -> con = NULL ;
2973
+ continue ;
2844
2974
}
2975
+ else /* just a nap from the script */
2976
+ {
2977
+ int this_usec ;
2845
2978
2846
- this_usec = st -> until - now_usec ;
2847
- if (min_usec > this_usec )
2848
- min_usec = this_usec ;
2849
- }
2850
- else if (st -> con == NULL )
2851
- {
2852
- continue ;
2979
+ if (min_usec == INT64_MAX )
2980
+ {
2981
+ instr_time now ;
2982
+
2983
+ INSTR_TIME_SET_CURRENT (now );
2984
+ now_usec = INSTR_TIME_GET_MICROSEC (now );
2985
+ }
2986
+
2987
+ this_usec = st -> until - now_usec ;
2988
+ if (min_usec > this_usec )
2989
+ min_usec = this_usec ;
2990
+ }
2853
2991
}
2854
2992
else if (commands [st -> state ]-> type == META_COMMAND )
2855
2993
{
@@ -2986,6 +3124,8 @@ threadRun(void *arg)
2986
3124
result -> xacts = 0 ;
2987
3125
for (i = 0 ; i < nstate ; i ++ )
2988
3126
result -> xacts += state [i ].cnt ;
3127
+ result -> throttle_lag = thread -> throttle_lag ;
3128
+ result -> throttle_lag_max = thread -> throttle_lag_max ;
2989
3129
INSTR_TIME_SET_CURRENT (end );
2990
3130
INSTR_TIME_ACCUM_DIFF (result -> conn_time , end , start );
2991
3131
if (logfile )
0 commit comments