Skip to content

rdkafka_performance: Add warmup flag #5348

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Stephan Dollberg (StephanDollberg) wants to merge 1 commit into
base: master
Choose a base branch
from
Open

rdkafka_performance: Add warmup flag #5348

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 44 additions & 1 deletion examples/rdkafka_performance.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ static int partition_cnt = 0;
static int eof_cnt = 0;
static int with_dr = 1;
static int read_hdrs = 0;
static int warmup_cnt = 0;


static void stop(int sig) {
Expand Down Expand Up @@ -511,6 +512,27 @@ print_stats(rd_kafka_t *rk, int mode, int otype, const char *compression) {
int extra_of = 0;
*extra = '\0';

if (warmup_cnt > 0) {
if ((int)cnt.msgs < warmup_cnt)
return;

/* Warmup count reached: flush pending delivery reports
* so that memset can cleanly zero all counters without
* late-arriving warmup DRs polluting benchmark stats. */
if (mode == 'P' && rk)
rd_kafka_flush(rk, 10000);
if (msgcnt != -1)
msgcnt -= (int)cnt.msgs;
memset(&cnt, 0, sizeof(cnt));
cnt.t_start = rd_clock();
cnt.t_last = cnt.t_start;
warmup_cnt = 0;
if (verbosity >= 1)
printf(
"%% Warmup complete, collecting "
"benchmark stats\n");
}

if (!(otype & _OTYPE_FORCE) &&
(((otype & _OTYPE_SUMMARY) && verbosity == 0) ||
cnt.t_last + dispintvl > now))
Expand Down Expand Up @@ -894,7 +916,7 @@ int main(int argc, char **argv) {

while ((opt = getopt(argc, argv,
"PCG:t:p:b:s:k:c:fi:MDd:m:S:x:"
"R:a:z:o:X:B:eT:Y:qvIur:lA:OwNH:")) != -1) {
"R:a:z:o:X:B:eT:Y:qvIur:w:lA:OwNH:")) != -1) {
switch (opt) {
case 'G':
if (rd_kafka_conf_set(conf, "group.id", optarg, errstr,
Expand Down Expand Up @@ -1098,6 +1120,15 @@ int main(int argc, char **argv) {
rate_sleep = (int)(1000000.0 / dtmp);
break;

case 'w':
warmup_cnt = atoi(optarg);
if (warmup_cnt <= 0) {
fprintf(stderr, "%% Invalid warmup count: %s\n",
optarg);
exit(1);
}
break;

case 'l':
latency_mode = 1;
break;
Expand Down Expand Up @@ -1178,6 +1209,8 @@ int main(int argc, char **argv) {
" -v Increase verbosity (default 1)\n"
" -u Output stats in table format\n"
" -r <rate> Producer msg/s limit\n"
" -w <cnt> Warmup message count. The first <cnt>\n"
" messages are excluded from stats.\n"
" -l Latency measurement.\n"
" Needs two matching instances, one\n"
" consumer and one producer, both\n"
Expand Down Expand Up @@ -1332,6 +1365,16 @@ int main(int argc, char **argv) {
exit(1);
}

if (warmup_cnt > 0) {
if (msgcnt != -1)
msgcnt += warmup_cnt;
if (verbosity >= 1)
printf(
"%% Warming up for %d messages before "
"collecting benchmark stats\n",
warmup_cnt);
}

if (mode == 'P') {
/*
* Producer
Expand Down