folly/test/ProducerConsumerQueueBenchmark.cpp

   1 /*
   2  * Copyright 2014 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 // @author: Bert Maher <bertrand@fb.com>
  18
  19 #include <thread>
  20 #include <iostream>
  21 #include <stdio.h>
  22 #include <pthread.h>
  23
  24 #include <gflags/gflags.h>
  25 #include <glog/logging.h>
  26 #include "folly/Benchmark.h"
  27 #include "folly/stats/Histogram.h"
  28 #include "folly/stats/Histogram-defs.h"
  29 #include "folly/ProducerConsumerQueue.h"
  30
  31 namespace {
  32
  33 using namespace folly;
  34
  35 typedef int ThroughputType;
  36 typedef ProducerConsumerQueue<ThroughputType> ThroughputQueueType;
  37
  38 typedef long LatencyType;
  39 typedef ProducerConsumerQueue<LatencyType> LatencyQueueType;
  40
  41 template<class QueueType>
  42 struct ThroughputTest {
  43   explicit ThroughputTest(size_t size, int iters, int cpu0, int cpu1)
  44   : queue_(size),
  45     done_(false),
  46     iters_(iters),
  47     cpu0_(cpu0),
  48     cpu1_(cpu1)
  49     { }
  50
  51   void producer() {
  52     if (cpu0_ > -1) {
  53       cpu_set_t cpuset;
  54       CPU_ZERO(&cpuset);
  55       CPU_SET(cpu0_, &cpuset);
  56       pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
  57     }
  58     for (int i = 0; i < iters_; ++i) {
  59       ThroughputType item = i;
  60       while (!queue_.write((ThroughputType) item)) {
  61       }
  62     }
  63   }
  64
  65   void consumer() {
  66     if (cpu1_ > -1) {
  67       cpu_set_t cpuset;
  68       CPU_ZERO(&cpuset);
  69       CPU_SET(cpu1_, &cpuset);
  70       pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
  71     }
  72     for (int i = 0; i < iters_; ++i) {
  73       ThroughputType item = 0;
  74       while (!queue_.read(item)) {
  75       }
  76       doNotOptimizeAway(item);
  77     }
  78   }
  79
  80   QueueType queue_;
  81   std::atomic<bool> done_;
  82   int iters_;
  83   int cpu0_;
  84   int cpu1_;
  85 };
  86
  87 template<class QueueType>
  88 struct LatencyTest {
  89   explicit LatencyTest(size_t size, int iters, int cpu0, int cpu1)
  90   : queue_(size),
  91     done_(false),
  92     iters_(iters),
  93     cpu0_(cpu0),
  94     cpu1_(cpu1),
  95     hist_(1, 0, 30)
  96     {
  97       computeTimeCost();
  98     }
  99
 100   void computeTimeCost() {
 101     int iterations = 1000;
 102     timespec start, end;
 103     clock_gettime(CLOCK_REALTIME, &start);
 104     for (int i = 0; i < iterations; ++i) {
 105       timespec tv;
 106       clock_gettime(CLOCK_REALTIME, &tv);
 107     }
 108     clock_gettime(CLOCK_REALTIME, &end);
 109     time_cost_ = 2 * detail::timespecDiff(end, start) / iterations;
 110   }
 111
 112   void producer() {
 113     if (cpu0_ > -1) {
 114       cpu_set_t cpuset;
 115       CPU_ZERO(&cpuset);
 116       CPU_SET(cpu0_, &cpuset);
 117       pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
 118     }
 119     for (int i = 0; i < iters_; ++i) {
 120       timespec sleeptime, sleepstart;
 121       clock_gettime(CLOCK_REALTIME, &sleepstart);
 122       do {
 123         clock_gettime(CLOCK_REALTIME, &sleeptime);
 124       } while (detail::timespecDiff(sleeptime, sleepstart) < 1000000);
 125
 126       timespec tv;
 127       clock_gettime(CLOCK_REALTIME, &tv);
 128       while (!queue_.write((LatencyType) tv.tv_nsec)) {
 129       }
 130     }
 131   }
 132
 133   void consumer() {
 134     if (cpu1_ > -1) {
 135       cpu_set_t cpuset;
 136       CPU_ZERO(&cpuset);
 137       CPU_SET(cpu1_, &cpuset);
 138       pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
 139     }
 140     for (int i = 0; i < iters_; ++i) {
 141       long enqueue_nsec;
 142       while (!queue_.read(enqueue_nsec)) {
 143       }
 144
 145       timespec tv;
 146       clock_gettime(CLOCK_REALTIME, &tv);
 147       int diff = tv.tv_nsec - enqueue_nsec - time_cost_;
 148       if (diff < 0) {
 149         continue;
 150       }
 151
 152       // Naive log-scale bucketing.
 153       int bucket;
 154       for (bucket = 0;
 155            bucket <= 30 && (1 << bucket) <= diff;
 156            ++bucket) {
 157       }
 158       hist_.addValue(bucket - 1);
 159     }
 160   }
 161
 162   void printHistogram() {
 163     hist_.toTSV(std::cout);
 164   }
 165
 166   QueueType queue_;
 167   std::atomic<bool> done_;
 168   int time_cost_;
 169   int iters_;
 170   int cpu0_;
 171   int cpu1_;
 172   Histogram<int> hist_;
 173 };
 174
 175 void BM_ProducerConsumer(int iters, int size) {
 176   BenchmarkSuspender susp;
 177   CHECK_GT(size, 0);
 178   ThroughputTest<ThroughputQueueType> *test =
 179     new ThroughputTest<ThroughputQueueType>(size, iters, -1, -1);
 180   susp.dismiss();
 181
 182   std::thread producer( [test] { test->producer(); } );
 183   std::thread consumer( [test] { test->consumer(); } );
 184
 185   producer.join();
 186   test->done_ = true;
 187   consumer.join();
 188   delete test;
 189 }
 190
 191 void BM_ProducerConsumerAffinity(int iters, int size) {
 192   BenchmarkSuspender susp;
 193   CHECK_GT(size, 0);
 194   ThroughputTest<ThroughputQueueType> *test =
 195     new ThroughputTest<ThroughputQueueType>(size, iters, 0, 1);
 196   susp.dismiss();
 197
 198   std::thread producer( [test] { test->producer(); } );
 199   std::thread consumer( [test] { test->consumer(); } );
 200
 201   producer.join();
 202   test->done_ = true;
 203   consumer.join();
 204   delete test;
 205 }
 206
 207 void BM_ProducerConsumerLatency(int iters, int size) {
 208   BenchmarkSuspender susp;
 209   CHECK_GT(size, 0);
 210   LatencyTest<LatencyQueueType> *test =
 211     new LatencyTest<LatencyQueueType>(size, 100000, 0, 1);
 212   susp.dismiss();
 213
 214   std::thread producer( [test] { test->producer(); } );
 215   std::thread consumer( [test] { test->consumer(); } );
 216
 217   producer.join();
 218   test->done_ = true;
 219   consumer.join();
 220   test->printHistogram();
 221   delete test;
 222 }
 223
 224
 225 BENCHMARK_DRAW_LINE();
 226
 227 BENCHMARK_PARAM(BM_ProducerConsumer, 1048574);
 228 BENCHMARK_PARAM(BM_ProducerConsumerAffinity, 1048574);
 229 BENCHMARK_PARAM(BM_ProducerConsumerLatency, 1048574);
 230
 231 }
 232
 233 int main(int argc, char** argv) {
 234   google::InitGoogleLogging(argv[0]);
 235   google::ParseCommandLineFlags(&argc, &argv, true);
 236
 237   runBenchmarks();
 238   return 0;
 239 }
 240
 241 #if 0
 242 /*
 243 Benchmark on Intel(R) Xeon(R) CPU E5-2660 0 @ 2.20GHz
 244 Latency histogram:
 245   log(nsec)
 246   min  max     count
 247   6    7       5124
 248   7    8       4799
 249   8    9       49
 250   9    10      2
 251   10   11      1
 252   11   12      5
 253   12   13      3
 254   13   14      9
 255   14   15      8
 256 ============================================================================
 257 folly/test/ProducerConsumerQueueBenchmark.cpp   relative  time/iter  iters/s
 258 ============================================================================
 259 ----------------------------------------------------------------------------
 260 BM_ProducerConsumer(1048574)                                 7.52ns  132.90M
 261 BM_ProducerConsumerAffinity(1048574)                         8.28ns  120.75M
 262 BM_ProducerConsumerLatency(1048574)                          10.00s   99.98m
 263 ============================================================================
 264 */
 265 #endif