use arena to allocate memtable's bloomfilter and hashskiplist's buckets_

Summary:
    Bloomfilter and hashskiplist's buckets_ allocated by memtable's arena
    DynamicBloom: pass arena via constructor, allocate space in SetTotalBits
    HashSkipListRep: allocate space of buckets_ using arena.
       do not delete it in deconstructor because arena would take care of it.
    Several test files are changed.

Test Plan:
    make all check

Reviewers: ljin, haobo, yhchiang, sdong

Reviewed By: sdong

Subscribers: igor, dhruba

Differential Revision: https://reviews.facebook.net/D19335

db/c_test.c

@@ -495,7 +495,7 @@ int main(int argc, char** argv) {
     rocksdb_filterpolicy_t* policy = rocksdb_filterpolicy_create_bloom(10);
     rocksdb_options_set_filter_policy(options, policy);
     rocksdb_options_set_prefix_extractor(options, rocksdb_slicetransform_create_fixed_prefix(3));
-    rocksdb_options_set_hash_skip_list_rep(options, 50000, 4, 4);
+    rocksdb_options_set_hash_skip_list_rep(options, 5000, 4, 4);
     rocksdb_options_set_plain_table_factory(options, 4, 10, 0.75, 16);
 
     db = rocksdb_open(options, dbname, &err);

db/db_test.cc

@@ -436,7 +436,8 @@ class DBTest {
     switch (option_config_) {
       case kHashSkipList:
         options.prefix_extractor.reset(NewFixedPrefixTransform(1));
-        options.memtable_factory.reset(NewHashSkipListRepFactory());
+        options.memtable_factory.reset(
+            NewHashSkipListRepFactory(16));
         break;
       case kPlainTableFirstBytePrefix:
         options.table_factory.reset(new PlainTableFactory());
@@ -6691,7 +6692,7 @@ TEST(DBTest, PrefixScan) {
   options.disable_auto_compactions = true;
   options.max_background_compactions = 2;
   options.create_if_missing = true;
-  options.memtable_factory.reset(NewHashSkipListRepFactory());
+  options.memtable_factory.reset(NewHashSkipListRepFactory(16));
 
   // 11 RAND I/Os
   DestroyAndReopen(&options);
@@ -6848,7 +6849,7 @@ TEST(DBTest, TailingIteratorPrefixSeek) {
   options.create_if_missing = true;
   options.disable_auto_compactions = true;
   options.prefix_extractor.reset(NewFixedPrefixTransform(2));
-  options.memtable_factory.reset(NewHashSkipListRepFactory());
+  options.memtable_factory.reset(NewHashSkipListRepFactory(16));
   DestroyAndReopen(&options);
   CreateAndReopenWithCF({"pikachu"}, &options);
 

db/memtable.cc

@@ -55,6 +55,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp, const Options& options)
   assert(!should_flush_);
   if (prefix_extractor_ && options.memtable_prefix_bloom_bits > 0) {
     prefix_bloom_.reset(new DynamicBloom(
+        &arena_,
         options.memtable_prefix_bloom_bits, options.bloom_locality,
         options.memtable_prefix_bloom_probes, nullptr,
         options.memtable_prefix_bloom_huge_page_tlb_size,

table/plain_table_reader.cc

@@ -333,7 +333,7 @@ void PlainTableReader::AllocateIndexAndBloom(int num_prefixes,
     uint32_t bloom_total_bits = num_prefixes * bloom_bits_per_key;
     if (bloom_total_bits > 0) {
       enable_bloom_ = true;
-      bloom_.SetTotalBits(bloom_total_bits, options_.bloom_locality,
+      bloom_.SetTotalBits(&arena_, bloom_total_bits, options_.bloom_locality,
                           huge_page_tlb_size, options_.info_log.get());
     }
   }
@@ -465,7 +465,7 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
         table_properties_->num_entries * bloom_bits_per_key;
     if (num_bloom_bits > 0) {
       enable_bloom_ = true;
-      bloom_.SetTotalBits(num_bloom_bits, options_.bloom_locality,
+      bloom_.SetTotalBits(&arena_, num_bloom_bits, options_.bloom_locality,
                           huge_page_tlb_size, options_.info_log.get());
     }
   }

util/dynamic_bloom.cc

@@ -32,12 +32,13 @@ uint32_t GetTotalBitsForLocality(uint32_t total_bits) {
 }
 }
 
-DynamicBloom::DynamicBloom(uint32_t total_bits, uint32_t locality,
+DynamicBloom::DynamicBloom(Arena* arena, uint32_t total_bits, uint32_t locality,
                            uint32_t num_probes,
                            uint32_t (*hash_func)(const Slice& key),
-                           size_t huge_page_tlb_size, Logger* logger)
+                           size_t huge_page_tlb_size,
+                           Logger* logger)
     : DynamicBloom(num_probes, hash_func) {
-  SetTotalBits(total_bits, locality, huge_page_tlb_size, logger);
+  SetTotalBits(arena, total_bits, locality, huge_page_tlb_size, logger);
 }
 
 DynamicBloom::DynamicBloom(uint32_t num_probes,
@@ -47,8 +48,10 @@ DynamicBloom::DynamicBloom(uint32_t num_probes,
       kNumProbes(num_probes),
       hash_func_(hash_func == nullptr ? &BloomHash : hash_func) {}
 
-void DynamicBloom::SetTotalBits(uint32_t total_bits, uint32_t locality,
-                                size_t huge_page_tlb_size, Logger* logger) {
+void DynamicBloom::SetTotalBits(Arena* arena,
+                                uint32_t total_bits, uint32_t locality,
+                                size_t huge_page_tlb_size,
+                                Logger* logger) {
   kTotalBits = (locality > 0) ? GetTotalBitsForLocality(total_bits)
                               : (total_bits + 7) / 8 * 8;
   kNumBlocks = (locality > 0) ? (kTotalBits / (CACHE_LINE_SIZE * 8)) : 0;
@@ -60,8 +63,9 @@ void DynamicBloom::SetTotalBits(uint32_t total_bits, uint32_t locality,
   if (kNumBlocks > 0) {
     sz += CACHE_LINE_SIZE - 1;
   }
+  assert(arena);
   raw_ = reinterpret_cast<unsigned char*>(
-      arena_.AllocateAligned(sz, huge_page_tlb_size, logger));
+      arena->AllocateAligned(sz, huge_page_tlb_size, logger));
   memset(raw_, 0, sz);
   if (kNumBlocks > 0 && (reinterpret_cast<uint64_t>(raw_) % CACHE_LINE_SIZE)) {
     data_ = raw_ + CACHE_LINE_SIZE -

util/dynamic_bloom.h

@@ -18,6 +18,7 @@ class Logger;
 
 class DynamicBloom {
  public:
+  // arena: pass arena to bloom filter, hence trace the usage of memory
   // total_bits: fixed total bits for the bloom
   // num_probes: number of hash probes for a single key
   // locality:  If positive, optimize for cache line locality, 0 otherwise.
@@ -27,7 +28,8 @@ class DynamicBloom {
   //                      it to be allocated, like:
   //                         sysctl -w vm.nr_hugepages=20
   //                     See linux doc Documentation/vm/hugetlbpage.txt
-  explicit DynamicBloom(uint32_t total_bits, uint32_t locality = 0,
+  explicit DynamicBloom(Arena* arena,
+                        uint32_t total_bits, uint32_t locality = 0,
                         uint32_t num_probes = 6,
                         uint32_t (*hash_func)(const Slice& key) = nullptr,
                         size_t huge_page_tlb_size = 0,
@@ -36,7 +38,7 @@ class DynamicBloom {
   explicit DynamicBloom(uint32_t num_probes = 6,
                         uint32_t (*hash_func)(const Slice& key) = nullptr);
 
-  void SetTotalBits(uint32_t total_bits, uint32_t locality,
+  void SetTotalBits(Arena* arena, uint32_t total_bits, uint32_t locality,
                     size_t huge_page_tlb_size, Logger* logger);
 
   ~DynamicBloom() {}
@@ -63,8 +65,6 @@ class DynamicBloom {
   uint32_t (*hash_func_)(const Slice& key);
   unsigned char* data_;
   unsigned char* raw_;
-
-  Arena arena_;
 };
 
 inline void DynamicBloom::Add(const Slice& key) { AddHash(hash_func_(key)); }

util/dynamic_bloom_test.cc

@@ -40,17 +40,19 @@ class DynamicBloomTest {
 };
 
 TEST(DynamicBloomTest, EmptyFilter) {
-  DynamicBloom bloom1(100, 0, 2);
+  Arena arena;
+  DynamicBloom bloom1(&arena, 100, 0, 2);
   ASSERT_TRUE(!bloom1.MayContain("hello"));
   ASSERT_TRUE(!bloom1.MayContain("world"));
 
-  DynamicBloom bloom2(CACHE_LINE_SIZE * 8 * 2 - 1, 1, 2);
+  DynamicBloom bloom2(&arena, CACHE_LINE_SIZE * 8 * 2 - 1, 1, 2);
   ASSERT_TRUE(!bloom2.MayContain("hello"));
   ASSERT_TRUE(!bloom2.MayContain("world"));
 }
 
 TEST(DynamicBloomTest, Small) {
-  DynamicBloom bloom1(100, 0, 2);
+  Arena arena;
+  DynamicBloom bloom1(&arena, 100, 0, 2);
   bloom1.Add("hello");
   bloom1.Add("world");
   ASSERT_TRUE(bloom1.MayContain("hello"));
@@ -58,7 +60,7 @@ TEST(DynamicBloomTest, Small) {
   ASSERT_TRUE(!bloom1.MayContain("x"));
   ASSERT_TRUE(!bloom1.MayContain("foo"));
 
-  DynamicBloom bloom2(CACHE_LINE_SIZE * 8 * 2 - 1, 1, 2);
+  DynamicBloom bloom2(&arena, CACHE_LINE_SIZE * 8 * 2 - 1, 1, 2);
   bloom2.Add("hello");
   bloom2.Add("world");
   ASSERT_TRUE(bloom2.MayContain("hello"));
@@ -94,13 +96,14 @@ TEST(DynamicBloomTest, VaryingLengths) {
   for (uint32_t enable_locality = 0; enable_locality < 2; ++enable_locality) {
     for (uint32_t num = 1; num <= 10000; num = NextNum(num)) {
       uint32_t bloom_bits = 0;
+      Arena arena;
       if (enable_locality == 0) {
         bloom_bits = std::max(num * FLAGS_bits_per_key, 64U);
       } else {
         bloom_bits = std::max(num * FLAGS_bits_per_key,
                               enable_locality * CACHE_LINE_SIZE * 8);
       }
-      DynamicBloom bloom(bloom_bits, enable_locality, num_probes);
+      DynamicBloom bloom(&arena, bloom_bits, enable_locality, num_probes);
       for (uint64_t i = 0; i < num; i++) {
         bloom.Add(Key(i, buffer));
         ASSERT_TRUE(bloom.MayContain(Key(i, buffer)));
@@ -148,10 +151,11 @@ TEST(DynamicBloomTest, perf) {
   }
 
   for (uint64_t m = 1; m <= 8; ++m) {
+    Arena arena;
     const uint64_t num_keys = m * 8 * 1024 * 1024;
     fprintf(stderr, "testing %" PRIu64 "M keys\n", m * 8);
 
-    DynamicBloom std_bloom(num_keys * 10, 0, num_probes);
+    DynamicBloom std_bloom(&arena, num_keys * 10, 0, num_probes);
 
     timer.Start();
     for (uint64_t i = 1; i <= num_keys; ++i) {
@@ -175,7 +179,7 @@ TEST(DynamicBloomTest, perf) {
     ASSERT_TRUE(count == num_keys);
 
     // Locality enabled version
-    DynamicBloom blocked_bloom(num_keys * 10, 1, num_probes);
+    DynamicBloom blocked_bloom(&arena, num_keys * 10, 1, num_probes);
 
       timer.Start();
       for (uint64_t i = 1; i <= num_keys; ++i) {

util/hash_skiplist_rep.cc

@@ -229,7 +229,9 @@ HashSkipListRep::HashSkipListRep(const MemTableRep::KeyComparator& compare,
       transform_(transform),
       compare_(compare),
       arena_(arena) {
-  buckets_ = new port::AtomicPointer[bucket_size];
+  auto mem =
+      arena->AllocateAligned(sizeof(port::AtomicPointer) * bucket_size);
+  buckets_ = new (mem) port::AtomicPointer[bucket_size];
 
   for (size_t i = 0; i < bucket_size_; ++i) {
     buckets_[i].NoBarrier_Store(nullptr);
@@ -237,7 +239,6 @@ HashSkipListRep::HashSkipListRep(const MemTableRep::KeyComparator& compare,
 }
 
 HashSkipListRep::~HashSkipListRep() {
-  delete[] buckets_;
 }
 
 HashSkipListRep::Bucket* HashSkipListRep::GetInitializedBucket(
@@ -271,7 +272,7 @@ bool HashSkipListRep::Contains(const char* key) const {
 }
 
 size_t HashSkipListRep::ApproximateMemoryUsage() {
-  return sizeof(buckets_);
+  return 0;
 }
 
 void HashSkipListRep::Get(const LookupKey& k, void* callback_args,