Function TEST

be/src/kudu/rpc/service_queue-test.cc:100–148

}

TEST(TestServiceQueue, LifoServiceQueuePerf) {
  LifoServiceQueue queue(FLAGS_max_queue_size);
  vector<std::thread> producers;
  vector<std::thread> consumers;

  for (int i = 0; i < FLAGS_num_producers; i++) {
    producers.emplace_back(&ProducerThread<LifoServiceQueue>, &queue);
  }

  for (int i = 0; i < FLAGS_num_consumers; i++) {
    consumers.emplace_back(&ConsumerThread<LifoServiceQueue>, &queue);
  }

  int seconds = AllowSlowTests() ? 10 : 1;
  uint64_t total_sample = 0;
  uint64_t total_queue_len = 0;
  uint64_t total_idle_workers = 0;
  Stopwatch sw(Stopwatch::ALL_THREADS);
  sw.start();
  int32_t before = total;

  for (int i = 0; i < seconds * 50; i++) {
    SleepFor(MonoDelta::FromMilliseconds(20));
    total_sample++;
    total_queue_len += queue.estimated_queue_length();
    total_idle_workers += queue.estimated_idle_worker_count();
  }

  sw.stop();
  int32_t delta = total - before;

  queue.Shutdown();
  for (int i = 0; i < FLAGS_num_producers; i++) {
    producers[i].join();
  }
  for (int i = 0; i < FLAGS_num_consumers; i++) {
    consumers[i].join();
  }

  float reqs_per_second = static_cast<float>(delta / sw.elapsed().wall_seconds());
  float user_cpu_micros_per_req = static_cast<float>(sw.elapsed().user / 1000.0 / delta);
  float sys_cpu_micros_per_req = static_cast<float>(sw.elapsed().system / 1000.0 / delta);

  LOG(INFO) << "Reqs/sec:         " << (int32_t)reqs_per_second;
  LOG(INFO) << "User CPU per req: " << user_cpu_micros_per_req << "us";
  LOG(INFO) << "Sys CPU per req:  " << sys_cpu_micros_per_req << "us";
  LOG(INFO) << "Avg rpc queue length: " << total_queue_len / static_cast<double>(total_sample);
  LOG(INFO) << "Avg idle workers:     " << total_idle_workers / static_cast<double>(total_sample);
}

} // namespace rpc
} // namespace kudu