Inside Bitcoin

501        method = &Benchmark::ReadWhileWriting;
502      } else if (name == Slice("compact")) {
503        method = &Benchmark::Compact;
504      } else if (name == Slice("crc32c")) {
505        method = &Benchmark::Crc32c;
506      } else if (name == Slice("snappycomp")) {
507        method = &Benchmark::SnappyCompress;
508      } else if (name == Slice("snappyuncomp")) {
509        method = &Benchmark::SnappyUncompress;
510      } else if (name == Slice("heapprofile")) {
511        HeapProfile();
512      } else if (name == Slice("stats")) {
513        PrintStats("leveldb.stats");
514      } else if (name == Slice("sstables")) {
515        PrintStats("leveldb.sstables");
516      } else {
517        if (!name.empty()) {  // No error message for empty name
518          fprintf(stderr, "unknown benchmark '%s'\n", name.ToString().c_str());
519        }
520      }
521
522      if (fresh_db) {
523        if (FLAGS_use_existing_db) {
524          fprintf(stdout, "%-12s : skipped (--use_existing_db is true)\n",
525                  name.ToString().c_str());
526          method = nullptr;
527        } else {
528          delete db_;
529          db_ = nullptr;
530          DestroyDB(FLAGS_db, Options());
531          Open();
532        }
533      }
534
535      if (method != nullptr) {
536        RunBenchmark(num_threads, name, method);
537      }
538    }
539  }
540
541 private:
542  struct ThreadArg {
543    Benchmark* bm;
544    SharedState* shared;
545    ThreadState* thread;
546    void (Benchmark::*method)(ThreadState*);
547  };
548
549  static void ThreadBody(void* v) {
550    ThreadArg* arg = reinterpret_cast<ThreadArg*>(v);
551    SharedState* shared = arg->shared;
552    ThreadState* thread = arg->thread;
553    {
554      MutexLock l(&shared->mu);
555      shared->num_initialized++;
556      if (shared->num_initialized >= shared->total) {
557        shared->cv.SignalAll();
558      }
559      while (!shared->start) {
560        shared->cv.Wait();
561      }
562    }
563
564    thread->stats.Start();
565    (arg->bm->*(arg->method))(thread);
566    thread->stats.Stop();
567
568    {
569      MutexLock l(&shared->mu);
570      shared->num_done++;
571      if (shared->num_done >= shared->total) {
572        shared->cv.SignalAll();
573      }
574    }
575  }
576
577  void RunBenchmark(int n, Slice name,
578                    void (Benchmark::*method)(ThreadState*)) {
579    SharedState shared(n);
580
581    ThreadArg* arg = new ThreadArg[n];
582    for (int i = 0; i < n; i++) {
583      arg[i].bm = this;
584      arg[i].method = method;
585      arg[i].shared = &shared;
586      arg[i].thread = new ThreadState(i);
587      arg[i].thread->shared = &shared;
588      g_env->StartThread(ThreadBody, &arg[i]);
589    }
590
591    shared.mu.Lock();
592    while (shared.num_initialized < n) {
593      shared.cv.Wait();
594    }
595
596    shared.start = true;
597    shared.cv.SignalAll();
598    while (shared.num_done < n) {
599      shared.cv.Wait();
600    }
601    shared.mu.Unlock();
602
603    for (int i = 1; i < n; i++) {
604      arg[0].thread->stats.Merge(arg[i].thread->stats);
605    }
606    arg[0].thread->stats.Report(name);
607
608    for (int i = 0; i < n; i++) {
609      delete arg[i].thread;
610    }
611    delete[] arg;
612  }
613
614  void Crc32c(ThreadState* thread) {
615    // Checksum about 500MB of data total
616    const int size = 4096;
617    const char* label = "(4K per op)";
618    std::string data(size, 'x');
619    int64_t bytes = 0;
620    uint32_t crc = 0;
621    while (bytes < 500 * 1048576) {
622      crc = crc32c::Value(data.data(), size);
623      thread->stats.FinishedSingleOp();
624      bytes += size;
625    }
626    // Print so result is not dead
627    fprintf(stderr, "... crc=0x%x\r", static_cast<unsigned int>(crc));
628
629    thread->stats.AddBytes(bytes);
630    thread->stats.AddMessage(label);
631  }
632
633  void SnappyCompress(ThreadState* thread) {
634    RandomGenerator gen;
635    Slice input = gen.Generate(Options().block_size);
636    int64_t bytes = 0;
637    int64_t produced = 0;
638    bool ok = true;
639    std::string compressed;
640    while (ok && bytes < 1024 * 1048576) {  // Compress 1G
641      ok = port::Snappy_Compress(input.data(), input.size(), &compressed);
642      produced += compressed.size();
643      bytes += input.size();
644      thread->stats.FinishedSingleOp();
645    }
646
647    if (!ok) {
648      thread->stats.AddMessage("(snappy failure)");
649    } else {
650      char buf[100];
651      snprintf(buf, sizeof(buf), "(output: %.1f%%)",
652               (produced * 100.0) / bytes);
653      thread->stats.AddMessage(buf);
654      thread->stats.AddBytes(bytes);
655    }
656  }
657
658  void SnappyUncompress(ThreadState* thread) {
659    RandomGenerator gen;
660    Slice input = gen.Generate(Options().block_size);
661    std::string compressed;
662    bool ok = port::Snappy_Compress(input.data(), input.size(), &compressed);
663    int64_t bytes = 0;
664    char* uncompressed = new char[input.size()];
665    while (ok && bytes < 1024 * 1048576) {  // Compress 1G
666      ok = port::Snappy_Uncompress(compressed.data(), compressed.size(),
667                                   uncompressed);
668      bytes += input.size();
669      thread->stats.FinishedSingleOp();
670    }
671    delete[] uncompressed;
672
673    if (!ok) {
674      thread->stats.AddMessage("(snappy failure)");
675    } else {
676      thread->stats.AddBytes(bytes);
677    }
678  }
679
680  void Open() {
681    assert(db_ == nullptr);
682    Options options;
683    options.env = g_env;
684    options.create_if_missing = !FLAGS_use_existing_db;
685    options.block_cache = cache_;
686    options.write_buffer_size = FLAGS_write_buffer_size;
687    options.max_file_size = FLAGS_max_file_size;
688    options.block_size = FLAGS_block_size;
689    options.max_open_files = FLAGS_open_files;
690    options.filter_policy = filter_policy_;
691    options.reuse_logs = FLAGS_reuse_logs;
692    Status s = DB::Open(options, FLAGS_db, &db_);
693    if (!s.ok()) {
694      fprintf(stderr, "open error: %s\n", s.ToString().c_str());
695      exit(1);
696    }
697  }
698
699  void OpenBench(ThreadState* thread) {
700    for (int i = 0; i < num_; i++) {
701      delete db_;
702      Open();
703      thread->stats.FinishedSingleOp();
704    }
705  }
706
707  void WriteSeq(ThreadState* thread) { DoWrite(thread, true); }
708
709  void WriteRandom(ThreadState* thread) { DoWrite(thread, false); }
710
711  void DoWrite(ThreadState* thread, bool seq) {
712    if (num_ != FLAGS_num) {
713      char msg[100];
714      snprintf(msg, sizeof(msg), "(%d ops)", num_);
715      thread->stats.AddMessage(msg);
716    }
717
718    RandomGenerator gen;
719    WriteBatch batch;
720    Status s;
721    int64_t bytes = 0;
722    for (int i = 0; i < num_; i += entries_per_batch_) {
723      batch.Clear();
724      for (int j = 0; j < entries_per_batch_; j++) {
725        const int k = seq ? i + j : (thread->rand.Next() % FLAGS_num);
726        char key[100];
727        snprintf(key, sizeof(key), "%016d", k);
728        batch.Put(key, gen.Generate(value_size_));
729        bytes += value_size_ + strlen(key);
730        thread->stats.FinishedSingleOp();
731      }
732      s = db_->Write(write_options_, &batch);
733      if (!s.ok()) {
734        fprintf(stderr, "put error: %s\n", s.ToString().c_str());
735        exit(1);
736      }
737    }
738    thread->stats.AddBytes(bytes);
739  }
740
741  void ReadSequential(ThreadState* thread) {
742    Iterator* iter = db_->NewIterator(ReadOptions());
743    int i = 0;
744    int64_t bytes = 0;
745    for (iter->SeekToFirst(); i < reads_ && iter->Valid(); iter->Next()) {
746      bytes += iter->key().size() + iter->value().size();
747      thread->stats.FinishedSingleOp();
748      ++i;
749    }
750    delete iter;