1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
4 
5 #include <kcpolydb.h>
6 #include <stdio.h>
7 #include <stdlib.h>
8 
9 #include "util/histogram.h"
10 #include "util/random.h"
11 #include "util/testutil.h"
12 
13 // Comma-separated list of operations to run in the specified order
14 //   Actual benchmarks:
15 //
16 //   fillseq       -- write N values in sequential key order in async mode
17 //   fillrandom    -- write N values in random key order in async mode
18 //   overwrite     -- overwrite N values in random key order in async mode
19 //   fillseqsync   -- write N/100 values in sequential key order in sync mode
20 //   fillrandsync  -- write N/100 values in random key order in sync mode
21 //   fillrand100K  -- write N/1000 100K values in random order in async mode
22 //   fillseq100K   -- write N/1000 100K values in seq order in async mode
23 //   readseq       -- read N times sequentially
24 //   readseq100K   -- read N/1000 100K values in sequential order in async mode
25 //   readrand100K  -- read N/1000 100K values in sequential order in async mode
26 //   readrandom    -- read N times in random order
27 static const char* FLAGS_benchmarks =
28     "fillseq,"
29     "fillseqsync,"
30     "fillrandsync,"
31     "fillrandom,"
32     "overwrite,"
33     "readrandom,"
34     "readseq,"
35     "fillrand100K,"
36     "fillseq100K,"
37     "readseq100K,"
38     "readrand100K,";
39 
40 // Number of key/values to place in database
41 static int FLAGS_num = 1000000;
42 
43 // Number of read operations to do.  If negative, do FLAGS_num reads.
44 static int FLAGS_reads = -1;
45 
46 // Size of each value
47 static int FLAGS_value_size = 100;
48 
49 // Arrange to generate values that shrink to this fraction of
50 // their original size after compression
51 static double FLAGS_compression_ratio = 0.5;
52 
53 // Print histogram of operation timings
54 static bool FLAGS_histogram = false;
55 
56 // Cache size. Default 4 MB
57 static int FLAGS_cache_size = 4194304;
58 
59 // Page size. Default 1 KB
60 static int FLAGS_page_size = 1024;
61 
62 // If true, do not destroy the existing database.  If you set this
63 // flag and also specify a benchmark that wants a fresh database, that
64 // benchmark will fail.
65 static bool FLAGS_use_existing_db = false;
66 
67 // Compression flag. If true, compression is on. If false, compression
68 // is off.
69 static bool FLAGS_compression = true;
70 
71 // Use the db with the following name.
72 static const char* FLAGS_db = nullptr;
73 
DBSynchronize(kyotocabinet::TreeDB * db_)74 inline static void DBSynchronize(kyotocabinet::TreeDB* db_) {
75   // Synchronize will flush writes to disk
76   if (!db_->synchronize()) {
77     fprintf(stderr, "synchronize error: %s\n", db_->error().name());
78   }
79 }
80 
81 namespace leveldb {
82 
83 // Helper for quickly generating random data.
84 namespace {
85 class RandomGenerator {
86  private:
87   std::string data_;
88   int pos_;
89 
90  public:
RandomGenerator()91   RandomGenerator() {
92     // We use a limited amount of data over and over again and ensure
93     // that it is larger than the compression window (32KB), and also
94     // large enough to serve all typical value sizes we want to write.
95     Random rnd(301);
96     std::string piece;
97     while (data_.size() < 1048576) {
98       // Add a short fragment that is as compressible as specified
99       // by FLAGS_compression_ratio.
100       test::CompressibleString(&rnd, FLAGS_compression_ratio, 100, &piece);
101       data_.append(piece);
102     }
103     pos_ = 0;
104   }
105 
Generate(int len)106   Slice Generate(int len) {
107     if (pos_ + len > data_.size()) {
108       pos_ = 0;
109       assert(len < data_.size());
110     }
111     pos_ += len;
112     return Slice(data_.data() + pos_ - len, len);
113   }
114 };
115 
TrimSpace(Slice s)116 static Slice TrimSpace(Slice s) {
117   int start = 0;
118   while (start < s.size() && isspace(s[start])) {
119     start++;
120   }
121   int limit = s.size();
122   while (limit > start && isspace(s[limit - 1])) {
123     limit--;
124   }
125   return Slice(s.data() + start, limit - start);
126 }
127 
128 }  // namespace
129 
130 class Benchmark {
131  private:
132   kyotocabinet::TreeDB* db_;
133   int db_num_;
134   int num_;
135   int reads_;
136   double start_;
137   double last_op_finish_;
138   int64_t bytes_;
139   std::string message_;
140   Histogram hist_;
141   RandomGenerator gen_;
142   Random rand_;
143   kyotocabinet::LZOCompressor<kyotocabinet::LZO::RAW> comp_;
144 
145   // State kept for progress messages
146   int done_;
147   int next_report_;  // When to report next
148 
PrintHeader()149   void PrintHeader() {
150     const int kKeySize = 16;
151     PrintEnvironment();
152     fprintf(stdout, "Keys:       %d bytes each\n", kKeySize);
153     fprintf(stdout, "Values:     %d bytes each (%d bytes after compression)\n",
154             FLAGS_value_size,
155             static_cast<int>(FLAGS_value_size * FLAGS_compression_ratio + 0.5));
156     fprintf(stdout, "Entries:    %d\n", num_);
157     fprintf(stdout, "RawSize:    %.1f MB (estimated)\n",
158             ((static_cast<int64_t>(kKeySize + FLAGS_value_size) * num_) /
159              1048576.0));
160     fprintf(stdout, "FileSize:   %.1f MB (estimated)\n",
161             (((kKeySize + FLAGS_value_size * FLAGS_compression_ratio) * num_) /
162              1048576.0));
163     PrintWarnings();
164     fprintf(stdout, "------------------------------------------------\n");
165   }
166 
PrintWarnings()167   void PrintWarnings() {
168 #if defined(__GNUC__) && !defined(__OPTIMIZE__)
169     fprintf(
170         stdout,
171         "WARNING: Optimization is disabled: benchmarks unnecessarily slow\n");
172 #endif
173 #ifndef NDEBUG
174     fprintf(stdout,
175             "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n");
176 #endif
177   }
178 
PrintEnvironment()179   void PrintEnvironment() {
180     fprintf(stderr, "Kyoto Cabinet:    version %s, lib ver %d, lib rev %d\n",
181             kyotocabinet::VERSION, kyotocabinet::LIBVER, kyotocabinet::LIBREV);
182 
183 #if defined(__linux)
184     time_t now = time(nullptr);
185     fprintf(stderr, "Date:           %s", ctime(&now));  // ctime() adds newline
186 
187     FILE* cpuinfo = fopen("/proc/cpuinfo", "r");
188     if (cpuinfo != nullptr) {
189       char line[1000];
190       int num_cpus = 0;
191       std::string cpu_type;
192       std::string cache_size;
193       while (fgets(line, sizeof(line), cpuinfo) != nullptr) {
194         const char* sep = strchr(line, ':');
195         if (sep == nullptr) {
196           continue;
197         }
198         Slice key = TrimSpace(Slice(line, sep - 1 - line));
199         Slice val = TrimSpace(Slice(sep + 1));
200         if (key == "model name") {
201           ++num_cpus;
202           cpu_type = val.ToString();
203         } else if (key == "cache size") {
204           cache_size = val.ToString();
205         }
206       }
207       fclose(cpuinfo);
208       fprintf(stderr, "CPU:            %d * %s\n", num_cpus, cpu_type.c_str());
209       fprintf(stderr, "CPUCache:       %s\n", cache_size.c_str());
210     }
211 #endif
212   }
213 
Start()214   void Start() {
215     start_ = Env::Default()->NowMicros() * 1e-6;
216     bytes_ = 0;
217     message_.clear();
218     last_op_finish_ = start_;
219     hist_.Clear();
220     done_ = 0;
221     next_report_ = 100;
222   }
223 
FinishedSingleOp()224   void FinishedSingleOp() {
225     if (FLAGS_histogram) {
226       double now = Env::Default()->NowMicros() * 1e-6;
227       double micros = (now - last_op_finish_) * 1e6;
228       hist_.Add(micros);
229       if (micros > 20000) {
230         fprintf(stderr, "long op: %.1f micros%30s\r", micros, "");
231         fflush(stderr);
232       }
233       last_op_finish_ = now;
234     }
235 
236     done_++;
237     if (done_ >= next_report_) {
238       if (next_report_ < 1000)
239         next_report_ += 100;
240       else if (next_report_ < 5000)
241         next_report_ += 500;
242       else if (next_report_ < 10000)
243         next_report_ += 1000;
244       else if (next_report_ < 50000)
245         next_report_ += 5000;
246       else if (next_report_ < 100000)
247         next_report_ += 10000;
248       else if (next_report_ < 500000)
249         next_report_ += 50000;
250       else
251         next_report_ += 100000;
252       fprintf(stderr, "... finished %d ops%30s\r", done_, "");
253       fflush(stderr);
254     }
255   }
256 
Stop(const Slice & name)257   void Stop(const Slice& name) {
258     double finish = Env::Default()->NowMicros() * 1e-6;
259 
260     // Pretend at least one op was done in case we are running a benchmark
261     // that does not call FinishedSingleOp().
262     if (done_ < 1) done_ = 1;
263 
264     if (bytes_ > 0) {
265       char rate[100];
266       snprintf(rate, sizeof(rate), "%6.1f MB/s",
267                (bytes_ / 1048576.0) / (finish - start_));
268       if (!message_.empty()) {
269         message_ = std::string(rate) + " " + message_;
270       } else {
271         message_ = rate;
272       }
273     }
274 
275     fprintf(stdout, "%-12s : %11.3f micros/op;%s%s\n", name.ToString().c_str(),
276             (finish - start_) * 1e6 / done_, (message_.empty() ? "" : " "),
277             message_.c_str());
278     if (FLAGS_histogram) {
279       fprintf(stdout, "Microseconds per op:\n%s\n", hist_.ToString().c_str());
280     }
281     fflush(stdout);
282   }
283 
284  public:
285   enum Order { SEQUENTIAL, RANDOM };
286   enum DBState { FRESH, EXISTING };
287 
Benchmark()288   Benchmark()
289       : db_(nullptr),
290         num_(FLAGS_num),
291         reads_(FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads),
292         bytes_(0),
293         rand_(301) {
294     std::vector<std::string> files;
295     std::string test_dir;
296     Env::Default()->GetTestDirectory(&test_dir);
297     Env::Default()->GetChildren(test_dir.c_str(), &files);
298     if (!FLAGS_use_existing_db) {
299       for (int i = 0; i < files.size(); i++) {
300         if (Slice(files[i]).starts_with("dbbench_polyDB")) {
301           std::string file_name(test_dir);
302           file_name += "/";
303           file_name += files[i];
304           Env::Default()->DeleteFile(file_name.c_str());
305         }
306       }
307     }
308   }
309 
~Benchmark()310   ~Benchmark() {
311     if (!db_->close()) {
312       fprintf(stderr, "close error: %s\n", db_->error().name());
313     }
314   }
315 
Run()316   void Run() {
317     PrintHeader();
318     Open(false);
319 
320     const char* benchmarks = FLAGS_benchmarks;
321     while (benchmarks != nullptr) {
322       const char* sep = strchr(benchmarks, ',');
323       Slice name;
324       if (sep == nullptr) {
325         name = benchmarks;
326         benchmarks = nullptr;
327       } else {
328         name = Slice(benchmarks, sep - benchmarks);
329         benchmarks = sep + 1;
330       }
331 
332       Start();
333 
334       bool known = true;
335       bool write_sync = false;
336       if (name == Slice("fillseq")) {
337         Write(write_sync, SEQUENTIAL, FRESH, num_, FLAGS_value_size, 1);
338         DBSynchronize(db_);
339       } else if (name == Slice("fillrandom")) {
340         Write(write_sync, RANDOM, FRESH, num_, FLAGS_value_size, 1);
341         DBSynchronize(db_);
342       } else if (name == Slice("overwrite")) {
343         Write(write_sync, RANDOM, EXISTING, num_, FLAGS_value_size, 1);
344         DBSynchronize(db_);
345       } else if (name == Slice("fillrandsync")) {
346         write_sync = true;
347         Write(write_sync, RANDOM, FRESH, num_ / 100, FLAGS_value_size, 1);
348         DBSynchronize(db_);
349       } else if (name == Slice("fillseqsync")) {
350         write_sync = true;
351         Write(write_sync, SEQUENTIAL, FRESH, num_ / 100, FLAGS_value_size, 1);
352         DBSynchronize(db_);
353       } else if (name == Slice("fillrand100K")) {
354         Write(write_sync, RANDOM, FRESH, num_ / 1000, 100 * 1000, 1);
355         DBSynchronize(db_);
356       } else if (name == Slice("fillseq100K")) {
357         Write(write_sync, SEQUENTIAL, FRESH, num_ / 1000, 100 * 1000, 1);
358         DBSynchronize(db_);
359       } else if (name == Slice("readseq")) {
360         ReadSequential();
361       } else if (name == Slice("readrandom")) {
362         ReadRandom();
363       } else if (name == Slice("readrand100K")) {
364         int n = reads_;
365         reads_ /= 1000;
366         ReadRandom();
367         reads_ = n;
368       } else if (name == Slice("readseq100K")) {
369         int n = reads_;
370         reads_ /= 1000;
371         ReadSequential();
372         reads_ = n;
373       } else {
374         known = false;
375         if (name != Slice()) {  // No error message for empty name
376           fprintf(stderr, "unknown benchmark '%s'\n", name.ToString().c_str());
377         }
378       }
379       if (known) {
380         Stop(name);
381       }
382     }
383   }
384 
385  private:
Open(bool sync)386   void Open(bool sync) {
387     assert(db_ == nullptr);
388 
389     // Initialize db_
390     db_ = new kyotocabinet::TreeDB();
391     char file_name[100];
392     db_num_++;
393     std::string test_dir;
394     Env::Default()->GetTestDirectory(&test_dir);
395     snprintf(file_name, sizeof(file_name), "%s/dbbench_polyDB-%d.kct",
396              test_dir.c_str(), db_num_);
397 
398     // Create tuning options and open the database
399     int open_options =
400         kyotocabinet::PolyDB::OWRITER | kyotocabinet::PolyDB::OCREATE;
401     int tune_options =
402         kyotocabinet::TreeDB::TSMALL | kyotocabinet::TreeDB::TLINEAR;
403     if (FLAGS_compression) {
404       tune_options |= kyotocabinet::TreeDB::TCOMPRESS;
405       db_->tune_compressor(&comp_);
406     }
407     db_->tune_options(tune_options);
408     db_->tune_page_cache(FLAGS_cache_size);
409     db_->tune_page(FLAGS_page_size);
410     db_->tune_map(256LL << 20);
411     if (sync) {
412       open_options |= kyotocabinet::PolyDB::OAUTOSYNC;
413     }
414     if (!db_->open(file_name, open_options)) {
415       fprintf(stderr, "open error: %s\n", db_->error().name());
416     }
417   }
418 
Write(bool sync,Order order,DBState state,int num_entries,int value_size,int entries_per_batch)419   void Write(bool sync, Order order, DBState state, int num_entries,
420              int value_size, int entries_per_batch) {
421     // Create new database if state == FRESH
422     if (state == FRESH) {
423       if (FLAGS_use_existing_db) {
424         message_ = "skipping (--use_existing_db is true)";
425         return;
426       }
427       delete db_;
428       db_ = nullptr;
429       Open(sync);
430       Start();  // Do not count time taken to destroy/open
431     }
432 
433     if (num_entries != num_) {
434       char msg[100];
435       snprintf(msg, sizeof(msg), "(%d ops)", num_entries);
436       message_ = msg;
437     }
438 
439     // Write to database
440     for (int i = 0; i < num_entries; i++) {
441       const int k = (order == SEQUENTIAL) ? i : (rand_.Next() % num_entries);
442       char key[100];
443       snprintf(key, sizeof(key), "%016d", k);
444       bytes_ += value_size + strlen(key);
445       std::string cpp_key = key;
446       if (!db_->set(cpp_key, gen_.Generate(value_size).ToString())) {
447         fprintf(stderr, "set error: %s\n", db_->error().name());
448       }
449       FinishedSingleOp();
450     }
451   }
452 
ReadSequential()453   void ReadSequential() {
454     kyotocabinet::DB::Cursor* cur = db_->cursor();
455     cur->jump();
456     std::string ckey, cvalue;
457     while (cur->get(&ckey, &cvalue, true)) {
458       bytes_ += ckey.size() + cvalue.size();
459       FinishedSingleOp();
460     }
461     delete cur;
462   }
463 
ReadRandom()464   void ReadRandom() {
465     std::string value;
466     for (int i = 0; i < reads_; i++) {
467       char key[100];
468       const int k = rand_.Next() % reads_;
469       snprintf(key, sizeof(key), "%016d", k);
470       db_->get(key, &value);
471       FinishedSingleOp();
472     }
473   }
474 };
475 
476 }  // namespace leveldb
477 
main(int argc,char ** argv)478 int main(int argc, char** argv) {
479   std::string default_db_path;
480   for (int i = 1; i < argc; i++) {
481     double d;
482     int n;
483     char junk;
484     if (leveldb::Slice(argv[i]).starts_with("--benchmarks=")) {
485       FLAGS_benchmarks = argv[i] + strlen("--benchmarks=");
486     } else if (sscanf(argv[i], "--compression_ratio=%lf%c", &d, &junk) == 1) {
487       FLAGS_compression_ratio = d;
488     } else if (sscanf(argv[i], "--histogram=%d%c", &n, &junk) == 1 &&
489                (n == 0 || n == 1)) {
490       FLAGS_histogram = n;
491     } else if (sscanf(argv[i], "--num=%d%c", &n, &junk) == 1) {
492       FLAGS_num = n;
493     } else if (sscanf(argv[i], "--reads=%d%c", &n, &junk) == 1) {
494       FLAGS_reads = n;
495     } else if (sscanf(argv[i], "--value_size=%d%c", &n, &junk) == 1) {
496       FLAGS_value_size = n;
497     } else if (sscanf(argv[i], "--cache_size=%d%c", &n, &junk) == 1) {
498       FLAGS_cache_size = n;
499     } else if (sscanf(argv[i], "--page_size=%d%c", &n, &junk) == 1) {
500       FLAGS_page_size = n;
501     } else if (sscanf(argv[i], "--compression=%d%c", &n, &junk) == 1 &&
502                (n == 0 || n == 1)) {
503       FLAGS_compression = (n == 1) ? true : false;
504     } else if (strncmp(argv[i], "--db=", 5) == 0) {
505       FLAGS_db = argv[i] + 5;
506     } else {
507       fprintf(stderr, "Invalid flag '%s'\n", argv[i]);
508       exit(1);
509     }
510   }
511 
512   // Choose a location for the test database if none given with --db=<path>
513   if (FLAGS_db == nullptr) {
514     leveldb::Env::Default()->GetTestDirectory(&default_db_path);
515     default_db_path += "/dbbench";
516     FLAGS_db = default_db_path.c_str();
517   }
518 
519   leveldb::Benchmark benchmark;
520   benchmark.Run();
521   return 0;
522 }
523