WritePrepared: Fix bug in searching in non-cached snapshots (#4639 )

Summary: When evicting an entry form the commit_cache, it is verified against the list of old snapshots to see if it overlaps with any. The list of old snapshots is split into two lists: an efficient concurrent cache and an slow vector protected by a lock. The patch fixes a bug that would stop the search in the cache if it finds any and yet would not include the larger snapshots in the slower list. An extra info log entry is also removed. The condition to trigger that although very rare is still feasible and should not spam the LOG when that happens. Fixes #4621 Pull Request resolved: https://github.com/facebook/rocksdb/pull/4639 Differential Revision: D12934989 Pulled By: maysamyabandeh fbshipit-source-id: 4e0fe8147ba292b554ae78e94c21c2ef31e03e2d
Revert "Introduce CacheAllocator, a custom allocator for cache blocks (#4437 )"
2026-07-07 14:47:40 +08:00 · 2018-11-06 13:49:35 -08:00 · 2018-10-31 14:55:11 -07:00 · 2018-10-31 13:02:53 -07:00 · 2018-10-30 12:37:06 -07:00 · 2018-10-19 16:38:03 -07:00
42 changed files with 711 additions and 419 deletions
@@ -1,5 +1,19 @@
 # Rocksdb Change Log
-## Unreleased
+# Unreleased
+### Bug Fixes
+* Fix the bug that WriteBatchWithIndex's SeekForPrev() doesn't see the entries with the same key.
+* Fix a bug in WritePrepared txns where if the number of old snapshots goes beyond the snapshot cache size (128 default) the rest will not be checked when evicting a commit entry from the commit cache.
+
+## 5.17.1 (10/16/2018)
+### Bug Fixes
+* Fix slow flush/compaction when DB contains many snapshots. The problem became noticeable to us in DBs with 100,000+ snapshots, though it will affect others at different thresholds.
+* Properly set the stop key for a truncated manual CompactRange
+* Fix corner case where a write group leader blocked due to write stall blocks other writers in queue with WriteOptions::no_slowdown set.
+
+### New Features
+* Introduced CacheAllocator, which lets the user specify custom allocator for memory in block cache.
+
+## 5.17.0 (10/05/2018)
 ### Public API Change
 * `OnTableFileCreated` will now be called for empty files generated during compaction. In that case, `TableFileCreationInfo::file_path` will be "(nil)" and `TableFileCreationInfo::file_size` will be zero.
 * Add `FlushOptions::allow_write_stall`, which controls whether Flush calls start working immediately, even if it causes user writes to stall, or will wait until flush can be performed without causing write stall (similar to `CompactRangeOptions::allow_write_stall`). Note that the default value is false, meaning we add delay to Flush calls until stalling can be avoided when possible. This is behavior change compared to previous RocksDB versions, where Flush calls didn't check if they might cause stall or not.
@@ -7,14 +21,25 @@

 ### New Features
 * TransactionOptions::skip_concurrency_control allows pessimistic transactions to skip the overhead of concurrency control. Could be used for optimizing certain transactions or during recovery.
-* Introduced CacheAllocator, which lets the user specify custom allocator for memory in block cache.

 ### Bug Fixes
 * Avoid creating empty SSTs and subsequently deleting them in certain cases during compaction.
 * Sync CURRENT file contents during checkpoint.
-* Fix format_version 4 bug with partitioned filters
+
+## 5.16.3 (10/1/2018)
+### Bug Fixes
 * Fix crash caused when `CompactFiles` run with `CompactionOptions::compression == CompressionType::kDisableCompressionOption`. Now that setting causes the compression type to be chosen according to the column family-wide compression options.

+## 5.16.2 (9/21/2018)
+### Bug Fixes
+* Fix bug in partition filters with format_version=4.
+
+## 5.16.1 (9/17/2018)
+### Bug Fixes
+* Remove trace_analyzer_tool from rocksdb_lib target in TARGETS file.
+* Fix RocksDB Java build and tests.
+* Remove sync point in Block destructor.
+
 ## 5.16.0 (8/21/2018)
 ### Public API Change
 * The merge operands are passed to `MergeOperator::ShouldMerge` in the reversed order relative to how they were merged (passed to FullMerge or FullMergeV2) for performance reasons
@@ -461,10 +461,8 @@ std::string LRUCacheShard::GetPrintableOptions() const {
 }

 LRUCache::LRUCache(size_t capacity, int num_shard_bits,
-                   bool strict_capacity_limit, double high_pri_pool_ratio,
-                   std::shared_ptr<CacheAllocator> allocator)
-    : ShardedCache(capacity, num_shard_bits, strict_capacity_limit,
-                   std::move(allocator)) {
+                   bool strict_capacity_limit, double high_pri_pool_ratio)
+    : ShardedCache(capacity, num_shard_bits, strict_capacity_limit) {
  num_shards_ = 1 << num_shard_bits;
  shards_ = reinterpret_cast<LRUCacheShard*>(
      port::cacheline_aligned_alloc(sizeof(LRUCacheShard) * num_shards_));
@@ -539,14 +537,12 @@ double LRUCache::GetHighPriPoolRatio() {
 std::shared_ptr<Cache> NewLRUCache(const LRUCacheOptions& cache_opts) {
  return NewLRUCache(cache_opts.capacity, cache_opts.num_shard_bits,
                     cache_opts.strict_capacity_limit,
-                     cache_opts.high_pri_pool_ratio,
-                     cache_opts.cache_allocator);
+                     cache_opts.high_pri_pool_ratio);
 }

-std::shared_ptr<Cache> NewLRUCache(
-    size_t capacity, int num_shard_bits, bool strict_capacity_limit,
-    double high_pri_pool_ratio,
-    std::shared_ptr<CacheAllocator> cache_allocator) {
+std::shared_ptr<Cache> NewLRUCache(size_t capacity, int num_shard_bits,
+                                   bool strict_capacity_limit,
+                                   double high_pri_pool_ratio) {
  if (num_shard_bits >= 20) {
    return nullptr;  // the cache cannot be sharded into too many fine pieces
  }
@@ -558,8 +554,7 @@ std::shared_ptr<Cache> NewLRUCache(
    num_shard_bits = GetDefaultCacheShardBits(capacity);
  }
  return std::make_shared<LRUCache>(capacity, num_shard_bits,
-                                    strict_capacity_limit, high_pri_pool_ratio,
-                                    std::move(cache_allocator));
+                                    strict_capacity_limit, high_pri_pool_ratio);
 }

 }  // namespace rocksdb
@@ -279,8 +279,7 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard : public CacheShard {
 class LRUCache : public ShardedCache {
 public:
  LRUCache(size_t capacity, int num_shard_bits, bool strict_capacity_limit,
-           double high_pri_pool_ratio,
-           std::shared_ptr<CacheAllocator> cache_allocator = nullptr);
+           double high_pri_pool_ratio);
  virtual ~LRUCache();
  virtual const char* Name() const override { return "LRUCache"; }
  virtual CacheShard* GetShard(int shard) override;
@@ -20,10 +20,8 @@
 namespace rocksdb {

 ShardedCache::ShardedCache(size_t capacity, int num_shard_bits,
-                           bool strict_capacity_limit,
-                           std::shared_ptr<CacheAllocator> allocator)
-    : Cache(std::move(allocator)),
-      num_shard_bits_(num_shard_bits),
+                           bool strict_capacity_limit)
+    : num_shard_bits_(num_shard_bits),
      capacity_(capacity),
      strict_capacity_limit_(strict_capacity_limit),
      last_id_(1) {}
@@ -144,9 +142,6 @@ std::string ShardedCache::GetPrintableOptions() const {
             strict_capacity_limit_);
    ret.append(buffer);
  }
-  snprintf(buffer, kBufferSize, "    cache_allocator : %s\n",
-           cache_allocator() ? cache_allocator()->Name() : "None");
-  ret.append(buffer);
  ret.append(GetShard(0)->GetPrintableOptions());
  return ret;
 }
@@ -47,8 +47,7 @@ class CacheShard {
 // Keys are sharded by the highest num_shard_bits bits of hash value.
 class ShardedCache : public Cache {
 public:
-  ShardedCache(size_t capacity, int num_shard_bits, bool strict_capacity_limit,
-               std::shared_ptr<CacheAllocator> cache_allocator = nullptr);
+  ShardedCache(size_t capacity, int num_shard_bits, bool strict_capacity_limit);
  virtual ~ShardedCache() = default;
  virtual const char* Name() const override = 0;
  virtual CacheShard* GetShard(int shard) = 0;
@@ -77,6 +77,12 @@ CompactionIterator::CompactionIterator(
    earliest_snapshot_ = snapshots_->at(0);
    latest_snapshot_ = snapshots_->back();
  }
+#ifndef NDEBUG
+  // findEarliestVisibleSnapshot assumes this ordering.
+  for (size_t i = 1; i < snapshots_->size(); ++i) {
+    assert(snapshots_->at(i - 1) <= snapshots_->at(i));
+  }
+#endif
  if (compaction_filter_ != nullptr) {
    if (compaction_filter_->IgnoreSnapshots()) {
      ignore_snapshots_ = true;
@@ -628,18 +634,23 @@ void CompactionIterator::PrepareOutput() {
 inline SequenceNumber CompactionIterator::findEarliestVisibleSnapshot(
    SequenceNumber in, SequenceNumber* prev_snapshot) {
  assert(snapshots_->size());
-  SequenceNumber prev = kMaxSequenceNumber;
-  for (const auto cur : *snapshots_) {
-    assert(prev == kMaxSequenceNumber || prev <= cur);
-    if (cur >= in && (snapshot_checker_ == nullptr ||
-                      snapshot_checker_->IsInSnapshot(in, cur))) {
-      *prev_snapshot = prev == kMaxSequenceNumber ? 0 : prev;
+  auto snapshots_iter = std::lower_bound(
+      snapshots_->begin(), snapshots_->end(), in);
+  if (snapshots_iter == snapshots_->begin()) {
+    *prev_snapshot = 0;
+  } else {
+    *prev_snapshot = *std::prev(snapshots_iter);
+    assert(*prev_snapshot < in);
+  }
+  for (; snapshots_iter != snapshots_->end(); ++snapshots_iter) {
+    auto cur = *snapshots_iter;
+    assert(in <= cur);
+    if (snapshot_checker_ == nullptr ||
+        snapshot_checker_->IsInSnapshot(in, cur)) {
      return cur;
    }
-    prev = cur;
-    assert(prev < kMaxSequenceNumber);
+    *prev_snapshot = cur;
  }
-  *prev_snapshot = prev;
  return kMaxSequenceNumber;
 }

@@ -219,7 +219,8 @@ void CompactionPicker::GetRange(const std::vector<CompactionInputFiles>& inputs,

 bool CompactionPicker::ExpandInputsToCleanCut(const std::string& /*cf_name*/,
                                              VersionStorageInfo* vstorage,
-                                              CompactionInputFiles* inputs) {
+                                              CompactionInputFiles* inputs,
+                                              InternalKey** next_smallest) {
  // This isn't good compaction
  assert(!inputs->empty());

@@ -242,7 +243,8 @@ bool CompactionPicker::ExpandInputsToCleanCut(const std::string& /*cf_name*/,
    GetRange(*inputs, &smallest, &largest);
    inputs->clear();
    vstorage->GetOverlappingInputs(level, &smallest, &largest, &inputs->files,
-                                   hint_index, &hint_index);
+                                   hint_index, &hint_index, true,
+                                   next_smallest);
  } while (inputs->size() > old_size);

  // we started off with inputs non-empty and the previous loop only grew
@@ -649,7 +651,6 @@ Compaction* CompactionPicker::CompactRange(
      uint64_t s = inputs[i]->compensated_file_size;
      total += s;
      if (total >= limit) {
-        **compaction_end = inputs[i + 1]->smallest;
        covering_the_whole_range = false;
        inputs.files.resize(i + 1);
        break;
@@ -658,7 +659,10 @@ Compaction* CompactionPicker::CompactRange(
  }
  assert(output_path_id < static_cast<uint32_t>(ioptions_.cf_paths.size()));

-  if (ExpandInputsToCleanCut(cf_name, vstorage, &inputs) == false) {
+  InternalKey key_storage;
+  InternalKey* next_smallest = &key_storage;
+  if (ExpandInputsToCleanCut(cf_name, vstorage, &inputs, &next_smallest) ==
+      false) {
    // manual compaction is now multi-threaded, so it can
    // happen that ExpandWhileOverlapping fails
    // we handle it higher in RunManualCompaction
@@ -666,8 +670,10 @@ Compaction* CompactionPicker::CompactRange(
    return nullptr;
  }

-  if (covering_the_whole_range) {
+  if (covering_the_whole_range || !next_smallest) {
    *compaction_end = nullptr;
+  } else {
+    **compaction_end = *next_smallest;
  }

  CompactionInputFiles output_level_inputs;
@@ -151,7 +151,8 @@ class CompactionPicker {
  // Will return false if it is impossible to apply this compaction.
  bool ExpandInputsToCleanCut(const std::string& cf_name,
                              VersionStorageInfo* vstorage,
-                              CompactionInputFiles* inputs);
+                              CompactionInputFiles* inputs,
+                              InternalKey** next_smallest = nullptr);

  // Returns true if any one of the parent files are being compacted
  bool IsRangeInCompaction(VersionStorageInfo* vstorage,
@@ -3961,6 +3961,50 @@ INSTANTIATE_TEST_CASE_P(
                      CompactionPri::kOldestSmallestSeqFirst,
                      CompactionPri::kMinOverlappingRatio));

+class NoopMergeOperator : public MergeOperator {
+ public:
+  NoopMergeOperator() {}
+
+  virtual bool FullMergeV2(const MergeOperationInput& /*merge_in*/,
+                           MergeOperationOutput* merge_out) const override {
+    std::string val("bar");
+    merge_out->new_value = val;
+    return true;
+  }
+
+  virtual const char* Name() const override { return "Noop"; }
+};
+
+TEST_F(DBCompactionTest, PartialManualCompaction) {
+  Options opts = CurrentOptions();
+  opts.num_levels = 3;
+  opts.level0_file_num_compaction_trigger = 10;
+  opts.compression = kNoCompression;
+  opts.merge_operator.reset(new NoopMergeOperator());
+  opts.target_file_size_base = 10240;
+  DestroyAndReopen(opts);
+
+  Random rnd(301);
+  for (auto i = 0; i < 8; ++i) {
+    for (auto j = 0; j < 10; ++j) {
+      Merge("foo", RandomString(&rnd, 1024));
+    }
+    Flush();
+  }
+
+  MoveFilesToLevel(2);
+
+  std::string prop;
+  EXPECT_TRUE(dbfull()->GetProperty(DB::Properties::kLiveSstFilesSize, &prop));
+  uint64_t max_compaction_bytes = atoi(prop.c_str()) / 2;
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"max_compaction_bytes", std::to_string(max_compaction_bytes)}}));
+
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+  dbfull()->CompactRange(cro, nullptr, nullptr);
+}
+
 #endif // !defined(ROCKSDB_LITE)
 }  // namespace rocksdb

@@ -813,6 +813,7 @@ class DBImpl : public DB {
  friend struct SuperVersion;
  friend class CompactedDBImpl;
  friend class DBTest_ConcurrentFlushWAL_Test;
+  friend class DBTest_MixedSlowdownOptionsStop_Test;
 #ifndef NDEBUG
  friend class DBTest2_ReadCallbackTest_Test;
  friend class WriteCallbackTest_WriteWithCallbackTest_Test;
@@ -1162,10 +1162,14 @@ Status DBImpl::DelayWrite(uint64_t num_bytes,
    uint64_t delay = write_controller_.GetDelay(env_, num_bytes);
    if (delay > 0) {
      if (write_options.no_slowdown) {
-        return Status::Incomplete();
+        return Status::Incomplete("Write stall");
      }
      TEST_SYNC_POINT("DBImpl::DelayWrite:Sleep");

+      // Notify write_thread_ about the stall so it can setup a barrier and
+      // fail any pending writers with no_slowdown
+      write_thread_.BeginWriteStall();
+      TEST_SYNC_POINT("DBImpl::DelayWrite:BeginWriteStallDone");
      mutex_.Unlock();
      // We will delay the write until we have slept for delay ms or
      // we don't need a delay anymore
@@ -1182,6 +1186,7 @@ Status DBImpl::DelayWrite(uint64_t num_bytes,
        env_->SleepForMicroseconds(kDelayInterval);
      }
      mutex_.Lock();
+      write_thread_.EndWriteStall();
    }

    // Don't wait if there's a background error, even if its a soft error. We
@@ -1190,11 +1195,16 @@ Status DBImpl::DelayWrite(uint64_t num_bytes,
    // indefinitely
    while (error_handler_.GetBGError().ok() && write_controller_.IsStopped()) {
      if (write_options.no_slowdown) {
-        return Status::Incomplete();
+        return Status::Incomplete("Write stall");
      }
      delayed = true;
+
+      // Notify write_thread_ about the stall so it can setup a barrier and
+      // fail any pending writers with no_slowdown
+      write_thread_.BeginWriteStall();
      TEST_SYNC_POINT("DBImpl::DelayWrite:Wait");
      bg_cv_.Wait();
+      write_thread_.EndWriteStall();
    }
  }
  assert(!delayed || !write_options.no_slowdown);
@@ -262,6 +262,196 @@ TEST_F(DBTest, SkipDelay) {
  }
 }

+TEST_F(DBTest, MixedSlowdownOptions) {
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.write_buffer_size = 100000;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<port::Thread> threads;
+  std::atomic<int> thread_num(0);
+
+  std::function<void()> write_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = false;
+    ASSERT_OK(dbfull()->Put(wo, key, "bar"));
+  };
+  std::function<void()> write_no_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = true;
+    ASSERT_NOK(dbfull()->Put(wo, key, "bar"));
+  };
+  // Use a small number to ensure a large delay that is still effective
+  // when we do Put
+  // TODO(myabandeh): this is time dependent and could potentially make
+  // the test flaky
+  auto token = dbfull()->TEST_write_controler().GetDelayToken(1);
+  std::atomic<int> sleep_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:BeginWriteStallDone",
+      [&](void* /*arg*/) {
+        sleep_count.fetch_add(1);
+        if (threads.empty()) {
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_slowdown_func);
+          }
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_no_slowdown_func);
+          }
+        }
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  WriteOptions wo;
+  wo.sync = false;
+  wo.disableWAL = false;
+  wo.no_slowdown = false;
+  dbfull()->Put(wo, "foo", "bar");
+  // We need the 2nd write to trigger delay. This is because delay is
+  // estimated based on the last write size which is 0 for the first write.
+  ASSERT_OK(dbfull()->Put(wo, "foo2", "bar2"));
+          token.reset();
+
+  for (auto& t : threads) {
+    t.join();
+  }
+  ASSERT_GE(sleep_count.load(), 1);
+
+  wo.no_slowdown = true;
+  ASSERT_OK(dbfull()->Put(wo, "foo3", "bar"));
+}
+
+TEST_F(DBTest, MixedSlowdownOptionsInQueue) {
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.write_buffer_size = 100000;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<port::Thread> threads;
+  std::atomic<int> thread_num(0);
+
+  std::function<void()> write_no_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = true;
+    ASSERT_NOK(dbfull()->Put(wo, key, "bar"));
+  };
+  // Use a small number to ensure a large delay that is still effective
+  // when we do Put
+  // TODO(myabandeh): this is time dependent and could potentially make
+  // the test flaky
+  auto token = dbfull()->TEST_write_controler().GetDelayToken(1);
+  std::atomic<int> sleep_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:Sleep",
+      [&](void* /*arg*/) {
+        sleep_count.fetch_add(1);
+        if (threads.empty()) {
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_no_slowdown_func);
+          }
+          // Sleep for 2s to allow the threads to insert themselves into the
+          // write queue
+          env_->SleepForMicroseconds(3000000ULL);
+        }
+      });
+  std::atomic<int> wait_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:Wait",
+      [&](void* /*arg*/) { wait_count.fetch_add(1); });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  WriteOptions wo;
+  wo.sync = false;
+  wo.disableWAL = false;
+  wo.no_slowdown = false;
+  dbfull()->Put(wo, "foo", "bar");
+  // We need the 2nd write to trigger delay. This is because delay is
+  // estimated based on the last write size which is 0 for the first write.
+  ASSERT_OK(dbfull()->Put(wo, "foo2", "bar2"));
+          token.reset();
+
+  for (auto& t : threads) {
+    t.join();
+  }
+  ASSERT_EQ(sleep_count.load(), 1);
+  ASSERT_GE(wait_count.load(), 0);
+}
+
+TEST_F(DBTest, MixedSlowdownOptionsStop) {
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.write_buffer_size = 100000;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<port::Thread> threads;
+  std::atomic<int> thread_num(0);
+
+  std::function<void()> write_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = false;
+    ASSERT_OK(dbfull()->Put(wo, key, "bar"));
+  };
+  std::function<void()> write_no_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = true;
+    ASSERT_NOK(dbfull()->Put(wo, key, "bar"));
+  };
+  std::function<void()> wakeup_writer = [&]() {
+    dbfull()->mutex_.Lock();
+    dbfull()->bg_cv_.SignalAll();
+    dbfull()->mutex_.Unlock();
+  };
+  // Use a small number to ensure a large delay that is still effective
+  // when we do Put
+  // TODO(myabandeh): this is time dependent and could potentially make
+  // the test flaky
+  auto token = dbfull()->TEST_write_controler().GetStopToken();
+  std::atomic<int> wait_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:Wait",
+      [&](void* /*arg*/) {
+        wait_count.fetch_add(1);
+        if (threads.empty()) {
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_slowdown_func);
+          }
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_no_slowdown_func);
+          }
+          // Sleep for 2s to allow the threads to insert themselves into the
+          // write queue
+          env_->SleepForMicroseconds(3000000ULL);
+        }
+        token.reset();
+        threads.emplace_back(wakeup_writer);
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  WriteOptions wo;
+  wo.sync = false;
+  wo.disableWAL = false;
+  wo.no_slowdown = false;
+  dbfull()->Put(wo, "foo", "bar");
+  // We need the 2nd write to trigger delay. This is because delay is
+  // estimated based on the last write size which is 0 for the first write.
+  ASSERT_OK(dbfull()->Put(wo, "foo2", "bar2"));
+          token.reset();
+
+  for (auto& t : threads) {
+    t.join();
+  }
+  ASSERT_GE(wait_count.load(), 1);
+
+  wo.no_slowdown = true;
+  ASSERT_OK(dbfull()->Put(wo, "foo3", "bar"));
+}
 #ifndef ROCKSDB_LITE

 TEST_F(DBTest, LevelLimitReopen) {
@@ -18,9 +18,11 @@ namespace rocksdb {
 #ifdef ROCKSDB_JEMALLOC
 #ifdef __FreeBSD__
 #include <malloc_np.h>
-#define je_malloc_stats_print malloc_stats_print
 #else
 #include "jemalloc/jemalloc.h"
+#ifdef JEMALLOC_NO_RENAME
+#define malloc_stats_print je_malloc_stats_print
+#endif
 #endif

 typedef struct {
@@ -48,7 +50,7 @@ void DumpMallocStats(std::string* stats) {
  std::unique_ptr<char[]> buf{new char[kMallocStatusLen + 1]};
  mstat.cur = buf.get();
  mstat.end = buf.get() + kMallocStatusLen;
-  je_malloc_stats_print(GetJemallocStatus, &mstat, "");
+  malloc_stats_print(GetJemallocStatus, &mstat, "");
  stats->append(buf.get());
 }
 #else
@@ -2028,7 +2028,7 @@ bool VersionStorageInfo::OverlapInLevel(int level,
 void VersionStorageInfo::GetOverlappingInputs(
    int level, const InternalKey* begin, const InternalKey* end,
    std::vector<FileMetaData*>* inputs, int hint_index, int* file_index,
-    bool expand_range) const {
+    bool expand_range, InternalKey** next_smallest) const {
  if (level >= num_non_empty_levels_) {
    // this level is empty, no overlapping inputs
    return;
@@ -2040,11 +2040,17 @@ void VersionStorageInfo::GetOverlappingInputs(
  }
  const Comparator* user_cmp = user_comparator_;
  if (level > 0) {
-    GetOverlappingInputsRangeBinarySearch(level, begin, end, inputs,
-                                          hint_index, file_index);
+    GetOverlappingInputsRangeBinarySearch(level, begin, end, inputs, hint_index,
+                                          file_index, false, next_smallest);
    return;
  }

+  if (next_smallest) {
+    // next_smallest key only makes sense for non-level 0, where files are
+    // non-overlapping
+    *next_smallest = nullptr;
+  }
+
  Slice user_begin, user_end;
  if (begin != nullptr) {
    user_begin = begin->user_key();
@@ -2198,7 +2204,7 @@ int sstableKeyCompare(const Comparator* user_cmp,
 void VersionStorageInfo::GetOverlappingInputsRangeBinarySearch(
    int level, const InternalKey* begin, const InternalKey* end,
    std::vector<FileMetaData*>* inputs, int hint_index, int* file_index,
-    bool within_interval) const {
+    bool within_interval, InternalKey** next_smallest) const {
  assert(level > 0);
  int min = 0;
  int mid = 0;
@@ -2234,6 +2240,9 @@ void VersionStorageInfo::GetOverlappingInputsRangeBinarySearch(

  // If there were no overlapping files, return immediately.
  if (!foundOverlap) {
+    if (next_smallest) {
+      next_smallest = nullptr;
+    }
    return;
  }
  // returns the index where an overlap is found
@@ -2254,6 +2263,15 @@ void VersionStorageInfo::GetOverlappingInputsRangeBinarySearch(
  for (int i = start_index; i <= end_index; i++) {
    inputs->push_back(files_[level][i]);
  }
+
+  if (next_smallest != nullptr) {
+    // Provide the next key outside the range covered by inputs
+    if (++end_index < static_cast<int>(files_[level].size())) {
+      **next_smallest = files_[level][end_index]->smallest;
+    } else {
+      *next_smallest = nullptr;
+    }
+  }
 }

 // Store in *start_index and *end_index the range of all files in
@@ -188,9 +188,11 @@ class VersionStorageInfo {
      std::vector<FileMetaData*>* inputs,
      int hint_index = -1,        // index of overlap file
      int* file_index = nullptr,  // return index of overlap file
-      bool expand_range = true)   // if set, returns files which overlap the
-      const;                      // range and overlap each other. If false,
+      bool expand_range = true,   // if set, returns files which overlap the
+                                  // range and overlap each other. If false,
                                  // then just files intersecting the range
+      InternalKey** next_smallest = nullptr)  // if non-null, returns the
+      const;  // smallest key of next file not included
  void GetCleanInputsWithinInterval(
      int level, const InternalKey* begin,  // nullptr means before all keys
      const InternalKey* end,               // nullptr means after all keys
@@ -200,14 +202,15 @@ class VersionStorageInfo {
      const;

  void GetOverlappingInputsRangeBinarySearch(
-      int level,           // level > 0
+      int level,                 // level > 0
      const InternalKey* begin,  // nullptr means before all keys
      const InternalKey* end,    // nullptr means after all keys
      std::vector<FileMetaData*>* inputs,
      int hint_index,                // index of overlap file
      int* file_index,               // return index of overlap file
-      bool within_interval = false)  // if set, force the inputs within interval
-      const;
+      bool within_interval = false,  // if set, force the inputs within interval
+      InternalKey** next_smallest = nullptr)  // if non-null, returns the
+      const;  // smallest key of next file not included

  void ExtendFileRangeOverlappingInterval(
      int level,
@@ -24,7 +24,10 @@ WriteThread::WriteThread(const ImmutableDBOptions& db_options)
      enable_pipelined_write_(db_options.enable_pipelined_write),
      newest_writer_(nullptr),
      newest_memtable_writer_(nullptr),
-      last_sequence_(0) {}
+      last_sequence_(0),
+      write_stall_dummy_(),
+      stall_mu_(),
+      stall_cv_(&stall_mu_) {}

 uint8_t WriteThread::BlockingAwaitState(Writer* w, uint8_t goal_mask) {
  // We're going to block.  Lazily create the mutex.  We guarantee
@@ -219,6 +222,28 @@ bool WriteThread::LinkOne(Writer* w, std::atomic<Writer*>* newest_writer) {
  assert(w->state == STATE_INIT);
  Writer* writers = newest_writer->load(std::memory_order_relaxed);
  while (true) {
+    // If write stall in effect, and w->no_slowdown is not true,
+    // block here until stall is cleared. If its true, then return
+    // immediately
+    if (writers == &write_stall_dummy_) {
+      if (w->no_slowdown) {
+        w->status = Status::Incomplete("Write stall");
+        SetState(w, STATE_COMPLETED);
+        return false;
+      }
+      // Since no_slowdown is false, wait here to be notified of the write
+      // stall clearing
+      {
+        MutexLock lock(&stall_mu_);
+        writers = newest_writer->load(std::memory_order_relaxed);
+        if (writers == &write_stall_dummy_) {
+          stall_cv_.Wait();
+          // Load newest_writers_ again since it may have changed
+          writers = newest_writer->load(std::memory_order_relaxed);
+          continue;
+        }
+      }
+    }
    w->link_older = writers;
    if (newest_writer->compare_exchange_weak(writers, w)) {
      return (writers == nullptr);
@@ -303,12 +328,44 @@ void WriteThread::CompleteFollower(Writer* w, WriteGroup& write_group) {
  SetState(w, STATE_COMPLETED);
 }

+void WriteThread::BeginWriteStall() {
+  LinkOne(&write_stall_dummy_, &newest_writer_);
+
+  // Walk writer list until w->write_group != nullptr. The current write group
+  // will not have a mix of slowdown/no_slowdown, so its ok to stop at that
+  // point
+  Writer* w = write_stall_dummy_.link_older;
+  Writer* prev = &write_stall_dummy_;
+  while (w != nullptr && w->write_group == nullptr) {
+    if (w->no_slowdown) {
+      prev->link_older = w->link_older;
+      w->status = Status::Incomplete("Write stall");
+      SetState(w, STATE_COMPLETED);
+      w = prev->link_older;
+    } else {
+      prev = w;
+      w = w->link_older;
+    }
+  }
+}
+
+void WriteThread::EndWriteStall() {
+  MutexLock lock(&stall_mu_);
+
+  assert(newest_writer_.load(std::memory_order_relaxed) == &write_stall_dummy_);
+  newest_writer_.exchange(write_stall_dummy_.link_older);
+
+  // Wake up writers
+  stall_cv_.SignalAll();
+}
+
 static WriteThread::AdaptationContext jbg_ctx("JoinBatchGroup");
 void WriteThread::JoinBatchGroup(Writer* w) {
  TEST_SYNC_POINT_CALLBACK("WriteThread::JoinBatchGroup:Start", w);
  assert(w->batch != nullptr);

  bool linked_as_leader = LinkOne(w, &newest_writer_);
+
  if (linked_as_leader) {
    SetState(w, STATE_GROUP_LEADER);
  }
@@ -342,6 +342,13 @@ class WriteThread {
    return last_sequence_;
  }

+  // Insert a dummy writer at the tail of the write queue to indicate a write
+  // stall, and fail any writers in the queue with no_slowdown set to true
+  void BeginWriteStall();
+
+  // Remove the dummy writer and wake up waiting writers
+  void EndWriteStall();
+
 private:
  // See AwaitState.
  const uint64_t max_yield_usec_;
@@ -365,6 +372,17 @@ class WriteThread {
  // is not necessary visible to reads because the writer can be ongoing.
  SequenceNumber last_sequence_;

+  // A dummy writer to indicate a write stall condition. This will be inserted
+  // at the tail of the writer queue by the leader, so newer writers can just
+  // check for this and bail
+  Writer write_stall_dummy_;
+
+  // Mutex and condvar for writers to block on a write stall. During a write
+  // stall, writers with no_slowdown set to false will wait on this rather
+  // on the writer queue
+  port::Mutex stall_mu_;
+  port::CondVar stall_cv_;
+
  // Waits for w->state & goal_mask using w->StateMutex().  Returns
  // the state that satisfies goal_mask.
  uint8_t BlockingAwaitState(Writer* w, uint8_t goal_mask);
@@ -25,7 +25,6 @@
 #include <stdint.h>
 #include <memory>
 #include <string>
-#include "rocksdb/cache_allocator.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
@@ -59,20 +58,13 @@ struct LRUCacheOptions {
  // BlockBasedTableOptions::cache_index_and_filter_blocks_with_high_priority.
  double high_pri_pool_ratio = 0.0;

-  // If non-nullptr will use this allocator instead of system allocator when
-  // allocating memory for cache blocks. Call this method before you start using
-  // the cache!
-  std::shared_ptr<CacheAllocator> cache_allocator;
-
  LRUCacheOptions() {}
  LRUCacheOptions(size_t _capacity, int _num_shard_bits,
-                  bool _strict_capacity_limit, double _high_pri_pool_ratio,
-                  std::shared_ptr<CacheAllocator> _cache_allocator = nullptr)
+                  bool _strict_capacity_limit, double _high_pri_pool_ratio)
      : capacity(_capacity),
        num_shard_bits(_num_shard_bits),
        strict_capacity_limit(_strict_capacity_limit),
-        high_pri_pool_ratio(_high_pri_pool_ratio),
-        cache_allocator(std::move(_cache_allocator)) {}
+        high_pri_pool_ratio(_high_pri_pool_ratio) {}
 };

 // Create a new cache with a fixed size capacity. The cache is sharded
@@ -83,10 +75,10 @@ struct LRUCacheOptions {
 // high_pri_pool_pct.
 // num_shard_bits = -1 means it is automatically determined: every shard
 // will be at least 512KB and number of shard bits will not exceed 6.
-extern std::shared_ptr<Cache> NewLRUCache(
-    size_t capacity, int num_shard_bits = -1,
-    bool strict_capacity_limit = false, double high_pri_pool_ratio = 0.0,
-    std::shared_ptr<CacheAllocator> cache_allocator = nullptr);
+extern std::shared_ptr<Cache> NewLRUCache(size_t capacity,
+                                          int num_shard_bits = -1,
+                                          bool strict_capacity_limit = false,
+                                          double high_pri_pool_ratio = 0.0);

 extern std::shared_ptr<Cache> NewLRUCache(const LRUCacheOptions& cache_opts);

@@ -99,15 +91,13 @@ extern std::shared_ptr<Cache> NewClockCache(size_t capacity,
                                            int num_shard_bits = -1,
                                            bool strict_capacity_limit = false);

-
 class Cache {
 public:
  // Depending on implementation, cache entries with high priority could be less
  // likely to get evicted than low priority entries.
  enum class Priority { HIGH, LOW };

-  Cache(std::shared_ptr<CacheAllocator> allocator = nullptr)
-      : cache_allocator_(std::move(allocator)) {}
+  Cache() {}

  // Destroys all existing entries by calling the "deleter"
  // function that was passed via the Insert() function.
@@ -238,14 +228,10 @@ class Cache {
  virtual void TEST_mark_as_data_block(const Slice& /*key*/,
                                       size_t /*charge*/) {}

-  CacheAllocator* cache_allocator() const { return cache_allocator_.get(); }
-
 private:
  // No copying allowed
  Cache(const Cache&);
  Cache& operator=(const Cache&);
-
-  std::shared_ptr<CacheAllocator> cache_allocator_;
 };

 }  // namespace rocksdb
@@ -1,29 +0,0 @@
-// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#pragma once
-
-// CacheAllocator is an interface that a client can implement to supply custom
-// cache allocation and deallocation methods. See rocksdb/cache.h for more
-// information.
-// All methods should be thread-safe.
-class CacheAllocator {
- public:
-  virtual ~CacheAllocator() = default;
-
-  // Name of the cache allocator, printed in the log
-  virtual const char* Name() const = 0;
-
-  // Allocate a block of at least size size
-  virtual void* Allocate(size_t size) = 0;
-  // Deallocate previously allocated block
-  virtual void Deallocate(void* p) = 0;
-  // Returns the memory size of the block allocated at p. The default
-  // implementation that just returns the original allocation_size is fine.
-  virtual size_t UsableSize(void* /*p*/, size_t allocation_size) const {
-    // default implementation just returns the allocation size
-    return allocation_size;
-  }
-};
@@ -2,8 +2,8 @@
 //  This source code is licensed under both the GPLv2 (found in the
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
-#ifndef ROCKSDB_LITE
 #pragma once
+#ifndef ROCKSDB_LITE
 #include <string>
 #include <vector>
 #include "rocksdb/db.h"
@@ -3,10 +3,10 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).

-#ifndef ROCKSDB_LITE
-
 #pragma once

+#ifndef ROCKSDB_LITE
+
 #include <memory>
 #include <string>

@@ -5,8 +5,8 @@
 #pragma once

 #define ROCKSDB_MAJOR 5
-#define ROCKSDB_MINOR 16
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_MINOR 17
+#define ROCKSDB_PATCH 1

 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these
@@ -39,7 +39,6 @@
 #include "table/full_filter_block.h"
 #include "table/table_builder.h"

-#include "util/cache_allocator.h"
 #include "util/coding.h"
 #include "util/compression.h"
 #include "util/crc32c.h"
@@ -655,8 +654,7 @@ Status BlockBasedTableBuilder::InsertBlockInCache(const Slice& block_contents,

    size_t size = block_contents.size();

-    auto ubuf =
-        AllocateBlock(size + 1, block_cache_compressed->cache_allocator());
+    std::unique_ptr<char[]> ubuf(new char[size + 1]);
    memcpy(ubuf.get(), block_contents.data(), size);
    ubuf[size] = type;

@@ -80,12 +80,11 @@ Status ReadBlockFromFile(
    std::unique_ptr<Block>* result, const ImmutableCFOptions& ioptions,
    bool do_uncompress, const Slice& compression_dict,
    const PersistentCacheOptions& cache_options, SequenceNumber global_seqno,
-    size_t read_amp_bytes_per_bit, CacheAllocator* allocator = nullptr,
-    const bool immortal_file = false) {
+    size_t read_amp_bytes_per_bit, const bool immortal_file = false) {
  BlockContents contents;
-  BlockFetcher block_fetcher(
-      file, prefetch_buffer, footer, options, handle, &contents, ioptions,
-      do_uncompress, compression_dict, cache_options, allocator, immortal_file);
+  BlockFetcher block_fetcher(file, prefetch_buffer, footer, options, handle,
+                             &contents, ioptions, do_uncompress,
+                             compression_dict, cache_options, immortal_file);
  Status s = block_fetcher.ReadBlockContents();
  if (s.ok()) {
    result->reset(new Block(std::move(contents), global_seqno,
@@ -95,13 +94,6 @@ Status ReadBlockFromFile(
  return s;
 }

-inline CacheAllocator* GetCacheAllocator(
-    const BlockBasedTableOptions& table_options) {
-  return table_options.block_cache.get()
-             ? table_options.block_cache->cache_allocator()
-             : nullptr;
-}
-
 // Delete the resource that is held by the iterator.
 template <class ResourceType>
 void DeleteHeldResource(void* arg, void* /*ignored*/) {
@@ -1158,8 +1150,7 @@ Status BlockBasedTable::ReadMetaBlock(Rep* rep,
      rep->footer.metaindex_handle(), &meta, rep->ioptions,
      true /* decompress */, Slice() /*compression dict*/,
      rep->persistent_cache_options, kDisableGlobalSequenceNumber,
-      0 /* read_amp_bytes_per_bit */,
-      GetCacheAllocator(rep->table_options));
+      0 /* read_amp_bytes_per_bit */);

  if (!s.ok()) {
    ROCKS_LOG_ERROR(rep->ioptions.info_log,
@@ -1182,7 +1173,7 @@ Status BlockBasedTable::GetDataBlockFromCache(
    const ImmutableCFOptions& ioptions, const ReadOptions& read_options,
    BlockBasedTable::CachableEntry<Block>* block, uint32_t format_version,
    const Slice& compression_dict, size_t read_amp_bytes_per_bit, bool is_index,
-    GetContext* get_context, CacheAllocator* allocator) {
+    GetContext* get_context) {
  Status s;
  Block* compressed_block = nullptr;
  Cache::Handle* block_cache_compressed_handle = nullptr;
@@ -1239,7 +1230,7 @@ Status BlockBasedTable::GetDataBlockFromCache(
                                          compression_dict);
  s = UncompressBlockContents(uncompresssion_ctx, compressed_block->data(),
                              compressed_block->size(), &contents,
-                              format_version, ioptions, allocator);
+                              format_version, ioptions);

  // Insert uncompressed block into block cache
  if (s.ok()) {
@@ -1301,8 +1292,7 @@ Status BlockBasedTable::PutDataBlockToCache(
    const ReadOptions& /*read_options*/, const ImmutableCFOptions& ioptions,
    CachableEntry<Block>* block, Block* raw_block, uint32_t format_version,
    const Slice& compression_dict, size_t read_amp_bytes_per_bit, bool is_index,
-    Cache::Priority priority, GetContext* get_context,
-    CacheAllocator* allocator) {
+    Cache::Priority priority, GetContext* get_context) {
  assert(raw_block->compression_type() == kNoCompression ||
         block_cache_compressed != nullptr);

@@ -1315,7 +1305,7 @@ Status BlockBasedTable::PutDataBlockToCache(
                                           compression_dict);
    s = UncompressBlockContents(uncompression_ctx, raw_block->data(),
                                raw_block->size(), &contents, format_version,
-                                ioptions, allocator);
+                                ioptions);
  }
  if (!s.ok()) {
    delete raw_block;
@@ -1412,8 +1402,7 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
  BlockFetcher block_fetcher(rep->file.get(), prefetch_buffer, rep->footer,
                             ReadOptions(), filter_handle, &block,
                             rep->ioptions, false /* decompress */,
-                             dummy_comp_dict, rep->persistent_cache_options,
-                             GetCacheAllocator(rep->table_options));
+                             dummy_comp_dict, rep->persistent_cache_options);
  Status s = block_fetcher.ReadBlockContents();

  if (!s.ok()) {
@@ -1711,9 +1700,7 @@ TBlockIter* BlockBasedTable::NewDataBlockIterator(
          &block_value, rep->ioptions, rep->blocks_maybe_compressed,
          compression_dict, rep->persistent_cache_options,
          is_index ? kDisableGlobalSequenceNumber : rep->global_seqno,
-          rep->table_options.read_amp_bytes_per_bit,
-          GetCacheAllocator(rep->table_options),
-          rep->immortal_table);
+          rep->table_options.read_amp_bytes_per_bit, rep->immortal_table);
    }
    if (s.ok()) {
      block.value = block_value.release();
@@ -1805,8 +1792,7 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
    s = GetDataBlockFromCache(
        key, ckey, block_cache, block_cache_compressed, rep->ioptions, ro,
        block_entry, rep->table_options.format_version, compression_dict,
-        rep->table_options.read_amp_bytes_per_bit, is_index, get_context,
-        GetCacheAllocator(rep->table_options));
+        rep->table_options.read_amp_bytes_per_bit, is_index, get_context);

    if (block_entry->value == nullptr && !no_io && ro.fill_cache) {
      std::unique_ptr<Block> raw_block;
@@ -1818,9 +1804,7 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
            block_cache_compressed == nullptr && rep->blocks_maybe_compressed,
            compression_dict, rep->persistent_cache_options,
            is_index ? kDisableGlobalSequenceNumber : rep->global_seqno,
-            rep->table_options.read_amp_bytes_per_bit,
-            GetCacheAllocator(rep->table_options),
-            rep->immortal_table);
+            rep->table_options.read_amp_bytes_per_bit, rep->immortal_table);
      }

      if (s.ok()) {
@@ -1833,7 +1817,7 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
                            .cache_index_and_filter_blocks_with_high_priority
                ? Cache::Priority::HIGH
                : Cache::Priority::LOW,
-            get_context, GetCacheAllocator(rep->table_options));
+            get_context);
      }
    }
  }
@@ -2540,12 +2524,11 @@ Status BlockBasedTable::VerifyChecksumInBlocks(
    BlockHandle handle = index_iter->value();
    BlockContents contents;
    Slice dummy_comp_dict;
-    BlockFetcher block_fetcher(
-        rep_->file.get(), nullptr /* prefetch buffer */, rep_->footer,
-        ReadOptions(), handle, &contents, rep_->ioptions,
-        false /* decompress */, dummy_comp_dict /*compression dict*/,
-        rep_->persistent_cache_options,
-        GetCacheAllocator(rep_->table_options));
+    BlockFetcher block_fetcher(rep_->file.get(), nullptr /* prefetch buffer */,
+                               rep_->footer, ReadOptions(), handle, &contents,
+                               rep_->ioptions, false /* decompress */,
+                               dummy_comp_dict /*compression dict*/,
+                               rep_->persistent_cache_options);
    s = block_fetcher.ReadBlockContents();
    if (!s.ok()) {
      break;
@@ -2567,12 +2550,11 @@ Status BlockBasedTable::VerifyChecksumInBlocks(
    s = handle.DecodeFrom(&input);
    BlockContents contents;
    Slice dummy_comp_dict;
-    BlockFetcher block_fetcher(
-        rep_->file.get(), nullptr /* prefetch buffer */, rep_->footer,
-        ReadOptions(), handle, &contents, rep_->ioptions,
-        false /* decompress */, dummy_comp_dict /*compression dict*/,
-        rep_->persistent_cache_options,
-        GetCacheAllocator(rep_->table_options));
+    BlockFetcher block_fetcher(rep_->file.get(), nullptr /* prefetch buffer */,
+                               rep_->footer, ReadOptions(), handle, &contents,
+                               rep_->ioptions, false /* decompress */,
+                               dummy_comp_dict /*compression dict*/,
+                               rep_->persistent_cache_options);
    s = block_fetcher.ReadBlockContents();
    if (!s.ok()) {
      break;
@@ -2876,8 +2858,7 @@ Status BlockBasedTable::DumpTable(WritableFile* out_file,
              rep_->file.get(), nullptr /* prefetch_buffer */, rep_->footer,
              ReadOptions(), handle, &block, rep_->ioptions,
              false /*decompress*/, dummy_comp_dict /*compression dict*/,
-              rep_->persistent_cache_options,
-              GetCacheAllocator(rep_->table_options));
+              rep_->persistent_cache_options);
          s = block_fetcher.ReadBlockContents();
          if (!s.ok()) {
            rep_->filter.reset(new BlockBasedFilterBlockReader(
@@ -303,8 +303,7 @@ class BlockBasedTable : public TableReader {
      const ImmutableCFOptions& ioptions, const ReadOptions& read_options,
      BlockBasedTable::CachableEntry<Block>* block, uint32_t format_version,
      const Slice& compression_dict, size_t read_amp_bytes_per_bit,
-      bool is_index = false, GetContext* get_context = nullptr,
-      CacheAllocator* allocator = nullptr);
+      bool is_index = false, GetContext* get_context = nullptr);

  // Put a raw block (maybe compressed) to the corresponding block caches.
  // This method will perform decompression against raw_block if needed and then
@@ -323,7 +322,7 @@ class BlockBasedTable : public TableReader {
      CachableEntry<Block>* block, Block* raw_block, uint32_t format_version,
      const Slice& compression_dict, size_t read_amp_bytes_per_bit,
      bool is_index = false, Cache::Priority pri = Cache::Priority::LOW,
-      GetContext* get_context = nullptr, CacheAllocator* allocator = nullptr);
+      GetContext* get_context = nullptr);

  // Calls (*handle_result)(arg, ...) repeatedly, starting with the entry found
  // after a call to Seek(key), until handle_result returns false.
@@ -17,9 +17,8 @@
 #include "rocksdb/env.h"
 #include "table/block.h"
 #include "table/block_based_table_reader.h"
-#include "table/format.h"
 #include "table/persistent_cache_helper.h"
-#include "util/cache_allocator.h"
+#include "table/format.h"
 #include "util/coding.h"
 #include "util/compression.h"
 #include "util/crc32c.h"
@@ -108,11 +107,9 @@ bool BlockFetcher::TryGetCompressedBlockFromPersistentCache() {
  if (cache_options_.persistent_cache &&
      cache_options_.persistent_cache->IsCompressed()) {
    // lookup uncompressed cache mode p-cache
-    std::unique_ptr<char[]> raw_data;
    status_ = PersistentCacheHelper::LookupRawPage(
-        cache_options_, handle_, &raw_data, block_size_ + kBlockTrailerSize);
+        cache_options_, handle_, &heap_buf_, block_size_ + kBlockTrailerSize);
    if (status_.ok()) {
-      heap_buf_ = CacheAllocationPtr(raw_data.release());
      used_buf_ = heap_buf_.get();
      slice_ = Slice(heap_buf_.get(), block_size_);
      return true;
@@ -135,7 +132,7 @@ void BlockFetcher::PrepareBufferForBlockFromFile() {
    // trivially allocated stack buffer instead of needing a full malloc()
    used_buf_ = &stack_buf_[0];
  } else {
-    heap_buf_ = AllocateBlock(block_size_ + kBlockTrailerSize, allocator_);
+    heap_buf_.reset(new char[block_size_ + kBlockTrailerSize]);
    used_buf_ = heap_buf_.get();
  }
 }
@@ -173,7 +170,7 @@ void BlockFetcher::GetBlockContents() {
    // or heap provided. Refer to https://github.com/facebook/rocksdb/pull/4096
    if (got_from_prefetch_buffer_ || used_buf_ == &stack_buf_[0]) {
      assert(used_buf_ != heap_buf_.get());
-      heap_buf_ = AllocateBlock(block_size_ + kBlockTrailerSize, allocator_);
+      heap_buf_.reset(new char[block_size_ + kBlockTrailerSize]);
      memcpy(heap_buf_.get(), used_buf_, block_size_ + kBlockTrailerSize);
    }
    *contents_ = BlockContents(std::move(heap_buf_), block_size_, true,
@@ -231,9 +228,9 @@ Status BlockFetcher::ReadBlockContents() {
  if (do_uncompress_ && compression_type != kNoCompression) {
    // compressed page, uncompress, update cache
    UncompressionContext uncompression_ctx(compression_type, compression_dict_);
-    status_ = UncompressBlockContents(uncompression_ctx, slice_.data(),
-                                      block_size_, contents_, footer_.version(),
-                                      ioptions_, allocator_);
+    status_ =
+        UncompressBlockContents(uncompression_ctx, slice_.data(), block_size_,
+                                contents_, footer_.version(), ioptions_);
  } else {
    GetBlockContents();
  }
@@ -11,8 +11,6 @@
 #include "table/block.h"
 #include "table/format.h"

-#include "util/cache_allocator.h"
-
 namespace rocksdb {
 class BlockFetcher {
 public:
@@ -28,7 +26,6 @@ class BlockFetcher {
               BlockContents* contents, const ImmutableCFOptions& ioptions,
               bool do_uncompress, const Slice& compression_dict,
               const PersistentCacheOptions& cache_options,
-               CacheAllocator* allocator = nullptr,
               const bool immortal_source = false)
      : file_(file),
        prefetch_buffer_(prefetch_buffer),
@@ -40,8 +37,7 @@ class BlockFetcher {
        do_uncompress_(do_uncompress),
        immortal_source_(immortal_source),
        compression_dict_(compression_dict),
-        cache_options_(cache_options),
-        allocator_(allocator) {}
+        cache_options_(cache_options) {}
  Status ReadBlockContents();

 private:
@@ -58,12 +54,11 @@ class BlockFetcher {
  const bool immortal_source_;
  const Slice& compression_dict_;
  const PersistentCacheOptions& cache_options_;
-  CacheAllocator* allocator_;
  Status status_;
  Slice slice_;
  char* used_buf_ = nullptr;
  size_t block_size_;
-  CacheAllocationPtr heap_buf_;
+  std::unique_ptr<char[]> heap_buf_;
  char stack_buf_[kDefaultStackBufferSize];
  bool got_from_prefetch_buffer_ = false;
  rocksdb::CompressionType compression_type;
@@ -19,7 +19,6 @@
 #include "table/block_based_table_reader.h"
 #include "table/block_fetcher.h"
 #include "table/persistent_cache_helper.h"
-#include "util/cache_allocator.h"
 #include "util/coding.h"
 #include "util/compression.h"
 #include "util/crc32c.h"
@@ -280,9 +279,8 @@ Status ReadFooterFromFile(RandomAccessFileReader* file,
 Status UncompressBlockContentsForCompressionType(
    const UncompressionContext& uncompression_ctx, const char* data, size_t n,
    BlockContents* contents, uint32_t format_version,
-    const ImmutableCFOptions& ioptions,
-    CacheAllocator* allocator) {
-  CacheAllocationPtr ubuf;
+    const ImmutableCFOptions& ioptions) {
+  std::unique_ptr<char[]> ubuf;

  assert(uncompression_ctx.type() != kNoCompression &&
         "Invalid compression type");
@@ -298,7 +296,7 @@ Status UncompressBlockContentsForCompressionType(
      if (!Snappy_GetUncompressedLength(data, n, &ulength)) {
        return Status::Corruption(snappy_corrupt_msg);
      }
-      ubuf = AllocateBlock(ulength, allocator);
+      ubuf.reset(new char[ulength]);
      if (!Snappy_Uncompress(data, n, ubuf.get())) {
        return Status::Corruption(snappy_corrupt_msg);
      }
@@ -306,10 +304,9 @@ Status UncompressBlockContentsForCompressionType(
      break;
    }
    case kZlibCompression:
-      ubuf = Zlib_Uncompress(
+      ubuf.reset(Zlib_Uncompress(
          uncompression_ctx, data, n, &decompress_size,
-          GetCompressFormatForVersion(kZlibCompression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kZlibCompression, format_version)));
      if (!ubuf) {
        static char zlib_corrupt_msg[] =
          "Zlib not supported or corrupted Zlib compressed block contents";
@@ -319,10 +316,9 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kBZip2Compression:
-      ubuf = BZip2_Uncompress(
+      ubuf.reset(BZip2_Uncompress(
          data, n, &decompress_size,
-          GetCompressFormatForVersion(kBZip2Compression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kBZip2Compression, format_version)));
      if (!ubuf) {
        static char bzip2_corrupt_msg[] =
          "Bzip2 not supported or corrupted Bzip2 compressed block contents";
@@ -332,10 +328,9 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kLZ4Compression:
-      ubuf = LZ4_Uncompress(
+      ubuf.reset(LZ4_Uncompress(
          uncompression_ctx, data, n, &decompress_size,
-          GetCompressFormatForVersion(kLZ4Compression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kLZ4Compression, format_version)));
      if (!ubuf) {
        static char lz4_corrupt_msg[] =
          "LZ4 not supported or corrupted LZ4 compressed block contents";
@@ -345,10 +340,9 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kLZ4HCCompression:
-      ubuf = LZ4_Uncompress(
+      ubuf.reset(LZ4_Uncompress(
          uncompression_ctx, data, n, &decompress_size,
-          GetCompressFormatForVersion(kLZ4HCCompression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kLZ4HCCompression, format_version)));
      if (!ubuf) {
        static char lz4hc_corrupt_msg[] =
          "LZ4HC not supported or corrupted LZ4HC compressed block contents";
@@ -358,8 +352,6 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kXpressCompression:
-      // XPRESS allocates memory internally, thus no support for custom
-      // allocator.
      ubuf.reset(XPRESS_Uncompress(data, n, &decompress_size));
      if (!ubuf) {
        static char xpress_corrupt_msg[] =
@@ -371,8 +363,7 @@ Status UncompressBlockContentsForCompressionType(
      break;
    case kZSTD:
    case kZSTDNotFinalCompression:
-      ubuf = ZSTD_Uncompress(uncompression_ctx, data, n, &decompress_size,
-                             allocator);
+      ubuf.reset(ZSTD_Uncompress(uncompression_ctx, data, n, &decompress_size));
      if (!ubuf) {
        static char zstd_corrupt_msg[] =
            "ZSTD not supported or corrupted ZSTD compressed block contents";
@@ -405,13 +396,11 @@ Status UncompressBlockContentsForCompressionType(
 Status UncompressBlockContents(const UncompressionContext& uncompression_ctx,
                               const char* data, size_t n,
                               BlockContents* contents, uint32_t format_version,
-                               const ImmutableCFOptions& ioptions,
-                               CacheAllocator* allocator) {
+                               const ImmutableCFOptions& ioptions) {
  assert(data[n] != kNoCompression);
  assert(data[n] == uncompression_ctx.type());
-  return UncompressBlockContentsForCompressionType(uncompression_ctx, data, n,
-                                                   contents, format_version,
-                                                   ioptions, allocator);
+  return UncompressBlockContentsForCompressionType(
+      uncompression_ctx, data, n, contents, format_version, ioptions);
 }

 }  // namespace rocksdb
@@ -26,7 +26,6 @@
 #include "port/port.h"  // noexcept
 #include "table/persistent_cache_options.h"
 #include "util/file_reader_writer.h"
-#include "util/cache_allocator.h"

 namespace rocksdb {

@@ -193,7 +192,7 @@ struct BlockContents {
  Slice data;     // Actual contents of data
  bool cachable;  // True iff data can be cached
  CompressionType compression_type;
-  CacheAllocationPtr allocation;
+  std::unique_ptr<char[]> allocation;

  BlockContents() : cachable(false), compression_type(kNoCompression) {}

@@ -201,28 +200,16 @@ struct BlockContents {
                CompressionType _compression_type)
      : data(_data), cachable(_cachable), compression_type(_compression_type) {}

-  BlockContents(CacheAllocationPtr&& _data, size_t _size, bool _cachable,
+  BlockContents(std::unique_ptr<char[]>&& _data, size_t _size, bool _cachable,
                CompressionType _compression_type)
      : data(_data.get(), _size),
        cachable(_cachable),
        compression_type(_compression_type),
        allocation(std::move(_data)) {}

-  BlockContents(std::unique_ptr<char[]>&& _data, size_t _size, bool _cachable,
-                CompressionType _compression_type)
-      : data(_data.get(), _size),
-        cachable(_cachable),
-        compression_type(_compression_type) {
-    allocation.reset(_data.release());
-  }
-
  // The additional memory space taken by the block data.
  size_t usable_size() const {
    if (allocation.get() != nullptr) {
-      auto allocator = allocation.get_deleter().allocator;
-      if (allocator) {
-        return allocator->UsableSize(allocation.get(), data.size());
-      }
 #ifdef ROCKSDB_MALLOC_USABLE_SIZE
      return malloc_usable_size(allocation.get());
 #else
@@ -265,7 +252,7 @@ extern Status ReadBlockContents(
 extern Status UncompressBlockContents(
    const UncompressionContext& uncompression_ctx, const char* data, size_t n,
    BlockContents* contents, uint32_t compress_format_version,
-    const ImmutableCFOptions& ioptions, CacheAllocator* allocator = nullptr);
+    const ImmutableCFOptions& ioptions);

 // This is an extension to UncompressBlockContents that accepts
 // a specific compression type. This is used by un-wrapped blocks
@@ -273,7 +260,7 @@ extern Status UncompressBlockContents(
 extern Status UncompressBlockContentsForCompressionType(
    const UncompressionContext& uncompression_ctx, const char* data, size_t n,
    BlockContents* contents, uint32_t compress_format_version,
-    const ImmutableCFOptions& ioptions, CacheAllocator* allocator = nullptr);
+    const ImmutableCFOptions& ioptions);

 // Implementation details follow.  Clients should ignore,

@@ -153,8 +153,8 @@ class PlainTableReader: public TableReader {
  DynamicBloom bloom_;
  PlainTableReaderFileInfo file_info_;
  Arena arena_;
-  CacheAllocationPtr index_block_alloc_;
-  CacheAllocationPtr bloom_block_alloc_;
+  std::unique_ptr<char[]> index_block_alloc_;
+  std::unique_ptr<char[]> bloom_block_alloc_;

  const ImmutableCFOptions& ioptions_;
  uint64_t file_size_;
@@ -2477,78 +2477,6 @@ TEST_P(BlockBasedTableTest, BlockCacheLeak) {
  c.ResetTableReader();
 }

-namespace {
-class CustomCacheAllocator : public CacheAllocator {
- public:
-  virtual const char* Name() const override { return "CustomCacheAllocator"; }
-
-  void* Allocate(size_t size) override {
-    ++numAllocations;
-    auto ptr = new char[size + 16];
-    memcpy(ptr, "cache_allocator_", 16);  // mangle first 16 bytes
-    return reinterpret_cast<void*>(ptr + 16);
-  }
-  void Deallocate(void* p) override {
-    ++numDeallocations;
-    char* ptr = reinterpret_cast<char*>(p) - 16;
-    delete[] ptr;
-  }
-
-  std::atomic<int> numAllocations;
-  std::atomic<int> numDeallocations;
-};
-}  // namespace
-
-TEST_P(BlockBasedTableTest, CacheAllocator) {
-  auto custom_cache_allocator = std::make_shared<CustomCacheAllocator>();
-  {
-    Options opt;
-    unique_ptr<InternalKeyComparator> ikc;
-    ikc.reset(new test::PlainInternalKeyComparator(opt.comparator));
-    opt.compression = kNoCompression;
-    BlockBasedTableOptions table_options;
-    table_options.block_size = 1024;
-    LRUCacheOptions lruOptions;
-    lruOptions.cache_allocator = custom_cache_allocator;
-    lruOptions.capacity = 16 * 1024 * 1024;
-    lruOptions.num_shard_bits = 4;
-    table_options.block_cache = NewLRUCache(std::move(lruOptions));
-    opt.table_factory.reset(NewBlockBasedTableFactory(table_options));
-
-    TableConstructor c(BytewiseComparator(),
-                       true /* convert_to_internal_key_ */);
-    c.Add("k01", "hello");
-    c.Add("k02", "hello2");
-    c.Add("k03", std::string(10000, 'x'));
-    c.Add("k04", std::string(200000, 'x'));
-    c.Add("k05", std::string(300000, 'x'));
-    c.Add("k06", "hello3");
-    c.Add("k07", std::string(100000, 'x'));
-    std::vector<std::string> keys;
-    stl_wrappers::KVMap kvmap;
-    const ImmutableCFOptions ioptions(opt);
-    const MutableCFOptions moptions(opt);
-    c.Finish(opt, ioptions, moptions, table_options, *ikc, &keys, &kvmap);
-
-    unique_ptr<InternalIterator> iter(
-        c.NewIterator(moptions.prefix_extractor.get()));
-    iter->SeekToFirst();
-    while (iter->Valid()) {
-      iter->key();
-      iter->value();
-      iter->Next();
-    }
-    ASSERT_OK(iter->status());
-  }
-
-  // out of scope, block cache should have been deleted, all allocations
-  // deallocated
-  EXPECT_EQ(custom_cache_allocator->numAllocations.load(),
-            custom_cache_allocator->numDeallocations.load());
-  // make sure that allocations actually happened through the cache allocator
-  EXPECT_GT(custom_cache_allocator->numAllocations.load(), 0);
-}
-
 TEST_P(BlockBasedTableTest, NewIndexIteratorLeak) {
  // A regression test to avoid data race described in
  // https://github.com/facebook/rocksdb/issues/1267
@@ -3040,7 +3040,7 @@ void VerifyDBFromDB(std::string& truth_db_name) {
    int64_t bytes = 0;
    int decompress_size;
    while (ok && bytes < 1024 * 1048576) {
-      CacheAllocationPtr uncompressed;
+      char *uncompressed = nullptr;
      switch (FLAGS_compression_type_e) {
        case rocksdb::kSnappyCompression: {
          // get size and allocate here to make comparison fair
@@ -3050,44 +3050,45 @@ void VerifyDBFromDB(std::string& truth_db_name) {
            ok = false;
            break;
          }
-          uncompressed = AllocateBlock(ulength, nullptr);
+          uncompressed = new char[ulength];
          ok = Snappy_Uncompress(compressed.data(), compressed.size(),
-                                 uncompressed.get());
+                                 uncompressed);
          break;
        }
      case rocksdb::kZlibCompression:
        uncompressed = Zlib_Uncompress(uncompression_ctx, compressed.data(),
                                       compressed.size(), &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kBZip2Compression:
        uncompressed = BZip2_Uncompress(compressed.data(), compressed.size(),
                                        &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kLZ4Compression:
        uncompressed = LZ4_Uncompress(uncompression_ctx, compressed.data(),
                                      compressed.size(), &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kLZ4HCCompression:
        uncompressed = LZ4_Uncompress(uncompression_ctx, compressed.data(),
                                      compressed.size(), &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kXpressCompression:
-        uncompressed.reset(XPRESS_Uncompress(
-            compressed.data(), compressed.size(), &decompress_size));
-        ok = uncompressed.get() != nullptr;
+        uncompressed = XPRESS_Uncompress(compressed.data(), compressed.size(),
+          &decompress_size);
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kZSTD:
        uncompressed = ZSTD_Uncompress(uncompression_ctx, compressed.data(),
                                       compressed.size(), &decompress_size);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      default:
        ok = false;
      }
+      delete[] uncompressed;
      bytes += input.size();
      thread->stats.FinishedOps(nullptr, nullptr, 1, kUncompress);
    }
@@ -1,38 +0,0 @@
-//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-//
-
-#pragma once
-
-#include "rocksdb/cache_allocator.h"
-
-namespace rocksdb {
-
-struct CustomDeleter {
-  CustomDeleter(CacheAllocator* a = nullptr) : allocator(a) {}
-
-  void operator()(char* ptr) const {
-    if (allocator) {
-      allocator->Deallocate(reinterpret_cast<void*>(ptr));
-    } else {
-      delete[] ptr;
-    }
-  }
-
-  CacheAllocator* allocator;
-};
-
-using CacheAllocationPtr = std::unique_ptr<char[], CustomDeleter>;
-
-inline CacheAllocationPtr AllocateBlock(size_t size,
-                                        CacheAllocator* allocator) {
-  if (allocator) {
-    auto block = reinterpret_cast<char*>(allocator->Allocate(size));
-    return CacheAllocationPtr(block, allocator);
-  }
-  return CacheAllocationPtr(new char[size]);
-}
-
-}  // namespace rocksdb
@@ -14,8 +14,6 @@
 #include <string>

 #include "rocksdb/options.h"
-#include "rocksdb/table.h"
-#include "util/cache_allocator.h"
 #include "util/coding.h"
 #include "util/compression_context_cache.h"

@@ -497,10 +495,11 @@ inline bool Zlib_Compress(const CompressionContext& ctx,
 // header in varint32 format
 // @param compression_dict Data for presetting the compression library's
 //    dictionary.
-inline CacheAllocationPtr Zlib_Uncompress(
-    const UncompressionContext& ctx, const char* input_data,
-    size_t input_length, int* decompress_size, uint32_t compress_format_version,
-    CacheAllocator* allocator = nullptr, int windowBits = -14) {
+inline char* Zlib_Uncompress(const UncompressionContext& ctx,
+                             const char* input_data, size_t input_length,
+                             int* decompress_size,
+                             uint32_t compress_format_version,
+                             int windowBits = -14) {
 #ifdef ZLIB
  uint32_t output_len = 0;
  if (compress_format_version == 2) {
@@ -542,9 +541,9 @@ inline CacheAllocationPtr Zlib_Uncompress(
  _stream.next_in = (Bytef*)input_data;
  _stream.avail_in = static_cast<unsigned int>(input_length);

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];

-  _stream.next_out = (Bytef*)output.get();
+  _stream.next_out = (Bytef*)output;
  _stream.avail_out = static_cast<unsigned int>(output_len);

  bool done = false;
@@ -562,17 +561,19 @@ inline CacheAllocationPtr Zlib_Uncompress(
        size_t old_sz = output_len;
        uint32_t output_len_delta = output_len / 5;
        output_len += output_len_delta < 10 ? 10 : output_len_delta;
-        auto tmp = AllocateBlock(output_len, allocator);
-        memcpy(tmp.get(), output.get(), old_sz);
-        output = std::move(tmp);
+        char* tmp = new char[output_len];
+        memcpy(tmp, output, old_sz);
+        delete[] output;
+        output = tmp;

        // Set more output.
-        _stream.next_out = (Bytef*)(output.get() + old_sz);
+        _stream.next_out = (Bytef*)(output + old_sz);
        _stream.avail_out = static_cast<unsigned int>(output_len - old_sz);
        break;
      }
      case Z_BUF_ERROR:
      default:
+        delete[] output;
        inflateEnd(&_stream);
        return nullptr;
    }
@@ -659,9 +660,9 @@ inline bool BZip2_Compress(const CompressionContext& /*ctx*/,
 // block header
 // compress_format_version == 2 -- decompressed size is included in the block
 // header in varint32 format
-inline CacheAllocationPtr BZip2_Uncompress(
-    const char* input_data, size_t input_length, int* decompress_size,
-    uint32_t compress_format_version, CacheAllocator* allocator = nullptr) {
+inline char* BZip2_Uncompress(const char* input_data, size_t input_length,
+                              int* decompress_size,
+                              uint32_t compress_format_version) {
 #ifdef BZIP2
  uint32_t output_len = 0;
  if (compress_format_version == 2) {
@@ -689,9 +690,9 @@ inline CacheAllocationPtr BZip2_Uncompress(
  _stream.next_in = (char*)input_data;
  _stream.avail_in = static_cast<unsigned int>(input_length);

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];

-  _stream.next_out = (char*)output.get();
+  _stream.next_out = (char*)output;
  _stream.avail_out = static_cast<unsigned int>(output_len);

  bool done = false;
@@ -708,16 +709,18 @@ inline CacheAllocationPtr BZip2_Uncompress(
        assert(compress_format_version != 2);
        uint32_t old_sz = output_len;
        output_len = output_len * 1.2;
-        auto tmp = AllocateBlock(output_len, allocator);
-        memcpy(tmp.get(), output.get(), old_sz);
-        output = std::move(tmp);
+        char* tmp = new char[output_len];
+        memcpy(tmp, output, old_sz);
+        delete[] output;
+        output = tmp;

        // Set more output.
-        _stream.next_out = (char*)(output.get() + old_sz);
+        _stream.next_out = (char*)(output + old_sz);
        _stream.avail_out = static_cast<unsigned int>(output_len - old_sz);
        break;
      }
      default:
+        delete[] output;
        BZ2_bzDecompressEnd(&_stream);
        return nullptr;
    }
@@ -811,12 +814,10 @@ inline bool LZ4_Compress(const CompressionContext& ctx,
 // header in varint32 format
 // @param compression_dict Data for presetting the compression library's
 //    dictionary.
-inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
-                                         const char* input_data,
-                                         size_t input_length,
-                                         int* decompress_size,
-                                         uint32_t compress_format_version,
-                                         CacheAllocator* allocator = nullptr) {
+inline char* LZ4_Uncompress(const UncompressionContext& ctx,
+                            const char* input_data, size_t input_length,
+                            int* decompress_size,
+                            uint32_t compress_format_version) {
 #ifdef LZ4
  uint32_t output_len = 0;
  if (compress_format_version == 2) {
@@ -836,7 +837,7 @@ inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
    input_data += 8;
  }

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];
 #if LZ4_VERSION_NUMBER >= 10400  // r124+
  LZ4_streamDecode_t* stream = LZ4_createStreamDecode();
  if (ctx.dict().size()) {
@@ -844,16 +845,17 @@ inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
                        static_cast<int>(ctx.dict().size()));
  }
  *decompress_size = LZ4_decompress_safe_continue(
-      stream, input_data, output.get(), static_cast<int>(input_length),
+      stream, input_data, output, static_cast<int>(input_length),
      static_cast<int>(output_len));
  LZ4_freeStreamDecode(stream);
 #else   // up to r123
-  *decompress_size = LZ4_decompress_safe(input_data, output.get(),
-                                         static_cast<int>(input_length),
-                                         static_cast<int>(output_len));
+  *decompress_size =
+      LZ4_decompress_safe(input_data, output, static_cast<int>(input_length),
+                          static_cast<int>(output_len));
 #endif  // LZ4_VERSION_NUMBER >= 10400

  if (*decompress_size < 0) {
+    delete[] output;
    return nullptr;
  }
  assert(*decompress_size == static_cast<int>(output_len));
@@ -864,7 +866,6 @@ inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
  (void)input_length;
  (void)decompress_size;
  (void)compress_format_version;
-  (void)allocator;
  return nullptr;
 #endif
 }
@@ -1027,11 +1028,9 @@ inline bool ZSTD_Compress(const CompressionContext& ctx, const char* input,

 // @param compression_dict Data for presetting the compression library's
 //    dictionary.
-inline CacheAllocationPtr ZSTD_Uncompress(const UncompressionContext& ctx,
-                                          const char* input_data,
-                                          size_t input_length,
-                                          int* decompress_size,
-                                          CacheAllocator* allocator = nullptr) {
+inline char* ZSTD_Uncompress(const UncompressionContext& ctx,
+                             const char* input_data, size_t input_length,
+                             int* decompress_size) {
 #ifdef ZSTD
  uint32_t output_len = 0;
  if (!compression::GetDecompressedSizeInfo(&input_data, &input_length,
@@ -1039,17 +1038,17 @@ inline CacheAllocationPtr ZSTD_Uncompress(const UncompressionContext& ctx,
    return nullptr;
  }

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];
  size_t actual_output_length;
 #if ZSTD_VERSION_NUMBER >= 500  // v0.5.0+
  ZSTD_DCtx* context = ctx.GetZSTDContext();
  assert(context != nullptr);
  actual_output_length = ZSTD_decompress_usingDict(
-      context, output.get(), output_len, input_data, input_length,
-      ctx.dict().data(), ctx.dict().size());
+      context, output, output_len, input_data, input_length, ctx.dict().data(),
+      ctx.dict().size());
 #else   // up to v0.4.x
  actual_output_length =
-      ZSTD_decompress(output.get(), output_len, input_data, input_length);
+      ZSTD_decompress(output, output_len, input_data, input_length);
 #endif  // ZSTD_VERSION_NUMBER >= 500
  assert(actual_output_length == output_len);
  *decompress_size = static_cast<int>(actual_output_length);
@@ -1059,7 +1058,6 @@ inline CacheAllocationPtr ZSTD_Uncompress(const UncompressionContext& ctx,
  (void)input_data;
  (void)input_length;
  (void)decompress_size;
-  (void)allocator;
  return nullptr;
 #endif
 }
@@ -5,8 +5,8 @@
 * Copyright 2013 Facebook
 */

-#ifndef ROCKSDB_LITE
 #pragma once
+#ifndef ROCKSDB_LITE
 #include <exception>

 namespace rocksdb {
@@ -816,6 +816,7 @@ TEST_P(WritePreparedTransactionTest, CheckAgainstSnapshotsTest) {
  std::vector<SequenceNumber> snapshots = {100l, 200l, 300l, 400l, 500l,
                                           600l, 700l, 800l, 900l};
  const size_t snapshot_cache_bits = 2;
+  const uint64_t cache_size = 1ul << snapshot_cache_bits;
  // Safety check to express the intended size in the test. Can be adjusted if
  // the snapshots lists changed.
  assert((1ul << snapshot_cache_bits) * 2 + 1 == snapshots.size());
@@ -843,6 +844,57 @@ TEST_P(WritePreparedTransactionTest, CheckAgainstSnapshotsTest) {
                         commit_entry.prep_seq <= snapshots.back();
    ASSERT_EQ(expect_update, !wp_db->old_commit_map_empty_);
  }
+
+  // Test that search will include multiple snapshot from snapshot cache
+  {
+    // exclude first and last item in the cache
+    CommitEntry commit_entry = {snapshots.front() + 1,
+                                snapshots[cache_size - 1] - 1};
+    wp_db->old_commit_map_empty_ = true;  // reset
+    wp_db->old_commit_map_.clear();
+    wp_db->CheckAgainstSnapshots(commit_entry);
+    ASSERT_EQ(wp_db->old_commit_map_.size(), cache_size - 2);
+  }
+
+  // Test that search will include multiple snapshot from old snapshots
+  {
+    // include two in the middle
+    CommitEntry commit_entry = {snapshots[cache_size] + 1,
+                                snapshots[cache_size + 2] + 1};
+    wp_db->old_commit_map_empty_ = true;  // reset
+    wp_db->old_commit_map_.clear();
+    wp_db->CheckAgainstSnapshots(commit_entry);
+    ASSERT_EQ(wp_db->old_commit_map_.size(), 2);
+  }
+
+  // Test that search will include both snapshot cache and old snapshots
+  // Case 1: includes all in snapshot cache
+  {
+    CommitEntry commit_entry = {snapshots.front() - 1, snapshots.back() + 1};
+    wp_db->old_commit_map_empty_ = true;  // reset
+    wp_db->old_commit_map_.clear();
+    wp_db->CheckAgainstSnapshots(commit_entry);
+    ASSERT_EQ(wp_db->old_commit_map_.size(), snapshots.size());
+  }
+
+  // Case 2: includes all snapshot caches except the smallest
+  {
+    CommitEntry commit_entry = {snapshots.front() + 1, snapshots.back() + 1};
+    wp_db->old_commit_map_empty_ = true;  // reset
+    wp_db->old_commit_map_.clear();
+    wp_db->CheckAgainstSnapshots(commit_entry);
+    ASSERT_EQ(wp_db->old_commit_map_.size(), snapshots.size() - 1);
+  }
+
+  // Case 3: includes only the largest of snapshot cache
+  {
+    CommitEntry commit_entry = {snapshots[cache_size - 1] - 1,
+                                snapshots.back() + 1};
+    wp_db->old_commit_map_empty_ = true;  // reset
+    wp_db->old_commit_map_.clear();
+    wp_db->CheckAgainstSnapshots(commit_entry);
+    ASSERT_EQ(wp_db->old_commit_map_.size(), snapshots.size() - cache_size + 1);
+  }
 }

 // This test is too slow for travis
@@ -572,14 +572,16 @@ void WritePreparedTxnDB::ReleaseSnapshotInternal(
    bool need_gc = false;
    {
      WPRecordTick(TXN_OLD_COMMIT_MAP_MUTEX_OVERHEAD);
-      ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead");
+      ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead for %" PRIu64,
+                     snap_seq);
      ReadLock rl(&old_commit_map_mutex_);
      auto prep_set_entry = old_commit_map_.find(snap_seq);
      need_gc = prep_set_entry != old_commit_map_.end();
    }
    if (need_gc) {
      WPRecordTick(TXN_OLD_COMMIT_MAP_MUTEX_OVERHEAD);
-      ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead");
+      ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead for %" PRIu64,
+                     snap_seq);
      WriteLock wl(&old_commit_map_mutex_);
      old_commit_map_.erase(snap_seq);
      old_commit_map_empty_.store(old_commit_map_.empty(),
@@ -654,13 +656,20 @@ void WritePreparedTxnDB::CheckAgainstSnapshots(const CommitEntry& evicted) {
  // place before gets overwritten the reader that reads bottom-up will
  // eventully see it.
  const bool next_is_larger = true;
-  SequenceNumber snapshot_seq = kMaxSequenceNumber;
+  // We will set to true if the border line snapshot suggests that.
+  bool search_larger_list = false;
  size_t ip1 = std::min(cnt, SNAPSHOT_CACHE_SIZE);
  for (; 0 < ip1; ip1--) {
-    snapshot_seq = snapshot_cache_[ip1 - 1].load(std::memory_order_acquire);
+    SequenceNumber snapshot_seq =
+        snapshot_cache_[ip1 - 1].load(std::memory_order_acquire);
    TEST_IDX_SYNC_POINT("WritePreparedTxnDB::CheckAgainstSnapshots:p:",
                        ++sync_i);
    TEST_IDX_SYNC_POINT("WritePreparedTxnDB::CheckAgainstSnapshots:s:", sync_i);
+    if (ip1 == SNAPSHOT_CACHE_SIZE) {  // border line snapshot
+      // snapshot_seq < commit_seq => larger_snapshot_seq <= commit_seq
+      // then later also continue the search to larger snapshots
+      search_larger_list = snapshot_seq < evicted.commit_seq;
+    }
    if (!MaybeUpdateOldCommitMap(evicted.prep_seq, evicted.commit_seq,
                                 snapshot_seq, !next_is_larger)) {
      break;
@@ -675,17 +684,20 @@ void WritePreparedTxnDB::CheckAgainstSnapshots(const CommitEntry& evicted) {
 #endif
  TEST_SYNC_POINT("WritePreparedTxnDB::CheckAgainstSnapshots:p:end");
  TEST_SYNC_POINT("WritePreparedTxnDB::CheckAgainstSnapshots:s:end");
-  if (UNLIKELY(SNAPSHOT_CACHE_SIZE < cnt && ip1 == SNAPSHOT_CACHE_SIZE &&
-               snapshot_seq < evicted.prep_seq)) {
+  if (UNLIKELY(SNAPSHOT_CACHE_SIZE < cnt && search_larger_list)) {
    // Then access the less efficient list of snapshots_
    WPRecordTick(TXN_SNAPSHOT_MUTEX_OVERHEAD);
-    ROCKS_LOG_WARN(info_log_, "snapshots_mutex_ overhead");
+    ROCKS_LOG_WARN(info_log_,
+                   "snapshots_mutex_ overhead for <%" PRIu64 ",%" PRIu64
+                   "> with %" ROCKSDB_PRIszt " snapshots",
+                   evicted.prep_seq, evicted.commit_seq, cnt);
    ReadLock rl(&snapshots_mutex_);
    // Items could have moved from the snapshots_ to snapshot_cache_ before
    // accquiring the lock. To make sure that we do not miss a valid snapshot,
    // read snapshot_cache_ again while holding the lock.
    for (size_t i = 0; i < SNAPSHOT_CACHE_SIZE; i++) {
-      snapshot_seq = snapshot_cache_[i].load(std::memory_order_acquire);
+      SequenceNumber snapshot_seq =
+          snapshot_cache_[i].load(std::memory_order_acquire);
      if (!MaybeUpdateOldCommitMap(evicted.prep_seq, evicted.commit_seq,
                                   snapshot_seq, next_is_larger)) {
        break;
@@ -713,7 +725,10 @@ bool WritePreparedTxnDB::MaybeUpdateOldCommitMap(
  // then snapshot_seq < commit_seq
  if (prep_seq <= snapshot_seq) {  // overlapping range
    WPRecordTick(TXN_OLD_COMMIT_MAP_MUTEX_OVERHEAD);
-    ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead");
+    ROCKS_LOG_WARN(info_log_,
+                   "old_commit_map_mutex_ overhead for %" PRIu64
+                   " commit entry: <%" PRIu64 ",%" PRIu64 ">",
+                   snapshot_seq, prep_seq, commit_seq);
    WriteLock wl(&old_commit_map_mutex_);
    old_commit_map_empty_.store(false, std::memory_order_release);
    auto& vec = old_commit_map_[snapshot_seq];
@@ -222,7 +222,6 @@ class WritePreparedTxnDB : public PessimisticTransactionDB {
      // rare case and it is ok to pay the cost of mutex ReadLock for such old,
      // reading transactions.
      WPRecordTick(TXN_OLD_COMMIT_MAP_MUTEX_OVERHEAD);
-      ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead");
      ReadLock rl(&old_commit_map_mutex_);
      auto prep_set_entry = old_commit_map_.find(snapshot_seq);
      bool found = prep_set_entry != old_commit_map_.end();
@@ -352,14 +352,16 @@ class WBWIIteratorImpl : public WBWIIterator {
  }

  virtual void SeekToFirst() override {
-    WriteBatchIndexEntry search_entry(WriteBatchIndexEntry::kFlagMin,
-                                      column_family_id_, 0, 0);
+    WriteBatchIndexEntry search_entry(
+        nullptr /* search_key */, column_family_id_,
+        true /* is_forward_direction */, true /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
  }

  virtual void SeekToLast() override {
-    WriteBatchIndexEntry search_entry(WriteBatchIndexEntry::kFlagMin,
-                                      column_family_id_ + 1, 0, 0);
+    WriteBatchIndexEntry search_entry(
+        nullptr /* search_key */, column_family_id_ + 1,
+        true /* is_forward_direction */, true /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
    if (!skip_list_iter_.Valid()) {
      skip_list_iter_.SeekToLast();
@@ -369,12 +371,16 @@ class WBWIIteratorImpl : public WBWIIterator {
  }

  virtual void Seek(const Slice& key) override {
-    WriteBatchIndexEntry search_entry(&key, column_family_id_);
+    WriteBatchIndexEntry search_entry(&key, column_family_id_,
+                                      true /* is_forward_direction */,
+                                      false /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
  }

  virtual void SeekForPrev(const Slice& key) override {
-    WriteBatchIndexEntry search_entry(&key, column_family_id_);
+    WriteBatchIndexEntry search_entry(&key, column_family_id_,
+                                      false /* is_forward_direction */,
+                                      false /* is_seek_to_first */);
    skip_list_iter_.SeekForPrev(&search_entry);
  }

@@ -85,6 +85,20 @@ Status ReadableWriteBatch::GetEntryFromDataOffset(size_t data_offset,
  return Status::OK();
 }

+// If both of `entry1` and `entry2` point to real entry in write batch, we
+// compare the entries as following:
+// 1. first compare the column family, the one with larger CF will be larger;
+// 2. Inside the same CF, we first decode the entry to find the key of the entry
+//    and the entry with larger key will be larger;
+// 3. If two entries are of the same CF and offset, the one with larger offset
+//    will be larger.
+// Some times either `entry1` or `entry2` is dummy entry, which is actually
+// a search key. In this case, in step 2, we don't go ahead and decode the
+// entry but use the value in WriteBatchIndexEntry::search_key.
+// One special case is WriteBatchIndexEntry::key_size is kFlagMinInCf.
+// This indicate that we are going to seek to the first of the column family.
+// Once we see this, this entry will be smaller than all the real entries of
+// the column family.
 int WriteBatchEntryComparator::operator()(
    const WriteBatchIndexEntry* entry1,
    const WriteBatchIndexEntry* entry2) const {
@@ -94,9 +108,10 @@ int WriteBatchEntryComparator::operator()(
    return -1;
  }

-  if (entry1->offset == WriteBatchIndexEntry::kFlagMin) {
+  // Deal with special case of seeking to the beginning of a column family
+  if (entry1->is_min_in_cf()) {
    return -1;
-  } else if (entry2->offset == WriteBatchIndexEntry::kFlagMin) {
+  } else if (entry2->is_min_in_cf()) {
    return 1;
  }

@@ -31,21 +31,52 @@ struct WriteBatchIndexEntry {
        key_offset(ko),
        key_size(ksz),
        search_key(nullptr) {}
-  WriteBatchIndexEntry(const Slice* sk, uint32_t c)
-      : offset(0),
-        column_family(c),
+  // Create a dummy entry as the search key. This index entry won't be backed
+  // by an entry from the write batch, but a pointer to the search key. Or a
+  // special flag of offset can indicate we are seek to first.
+  // @_search_key: the search key
+  // @_column_family: column family
+  // @is_forward_direction: true for Seek(). False for SeekForPrev()
+  // @is_seek_to_first: true if we seek to the beginning of the column family
+  //                    _search_key should be null in this case.
+  WriteBatchIndexEntry(const Slice* _search_key, uint32_t _column_family,
+                       bool is_forward_direction, bool is_seek_to_first)
+      // For SeekForPrev(), we need to make the dummy entry larger than any
+      // entry who has the same search key. Otherwise, we'll miss those entries.
+      : offset(is_forward_direction ? 0 : port::kMaxSizet),
+        column_family(_column_family),
        key_offset(0),
-        key_size(0),
-        search_key(sk) {}
+        key_size(is_seek_to_first ? kFlagMinInCf : 0),
+        search_key(_search_key) {
+    assert(_search_key != nullptr || is_seek_to_first);
+  }

-  // If this flag appears in the offset, it indicates a key that is smaller
-  // than any other entry for the same column family
-  static const size_t kFlagMin = port::kMaxSizet;
+  // If this flag appears in the key_size, it indicates a
+  // key that is smaller than any other entry for the same column family.
+  static const size_t kFlagMinInCf = port::kMaxSizet;

-  size_t offset;           // offset of an entry in write batch's string buffer.
-  uint32_t column_family;  // column family of the entry.
+  bool is_min_in_cf() const {
+    assert(key_size != kFlagMinInCf ||
+           (key_offset == 0 && search_key == nullptr));
+    return key_size == kFlagMinInCf;
+  }
+
+  // offset of an entry in write batch's string buffer. If this is a dummy
+  // lookup key, in which case search_key != nullptr, offset is set to either
+  // 0 or max, only for comparison purpose. Because when entries have the same
+  // key, the entry with larger offset is larger, offset = 0 will make a seek
+  // key small or equal than all the entries with the seek key, so that Seek()
+  // will find all the entries of the same key. Similarly, offset = MAX will
+  // make the entry just larger than all entries with the search key so
+  // SeekForPrev() will see all the keys with the same key.
+  size_t offset;
+  uint32_t column_family;  // c1olumn family of the entry.
  size_t key_offset;       // offset of the key in write batch's string buffer.
-  size_t key_size;         // size of the key.
+  size_t key_size;         // size of the key. kFlagMinInCf indicates
+                           // that this is a dummy look up entry for
+                           // SeekToFirst() to the beginning of the column
+                           // family. We use the flag here to save a boolean
+                           // in the struct.

  const Slice* search_key;  // if not null, instead of reading keys from
                            // write batch, use it to compare. This is used
@@ -621,7 +621,7 @@ TEST_F(WriteBatchWithIndexTest, TestRandomIteraratorWithBase) {
    for (int i = 0; i < 128; i++) {
      // Random walk and make sure iter and result_iter returns the
      // same key and value
-      int type = rnd.Uniform(5);
+      int type = rnd.Uniform(6);
      ASSERT_OK(iter->status());
      switch (type) {
        case 0:
@@ -642,7 +642,15 @@ TEST_F(WriteBatchWithIndexTest, TestRandomIteraratorWithBase) {
          result_iter->Seek(key);
          break;
        }
-        case 3:
+        case 3: {
+          // SeekForPrev to random key
+          auto key_idx = rnd.Uniform(static_cast<int>(source_strings.size()));
+          auto key = source_strings[key_idx];
+          iter->SeekForPrev(key);
+          result_iter->SeekForPrev(key);
+          break;
+        }
+        case 4:
          // Next
          if (is_valid) {
            iter->Next();
@@ -652,7 +660,7 @@ TEST_F(WriteBatchWithIndexTest, TestRandomIteraratorWithBase) {
          }
          break;
        default:
-          assert(type == 4);
+          assert(type == 5);
          // Prev
          if (is_valid) {
            iter->Prev();
Author	SHA1	Message	Date
Maysam Yabandeh	2b39d125f3	WritePrepared: Fix bug in searching in non-cached snapshots (#4639 ) Summary: When evicting an entry form the commit_cache, it is verified against the list of old snapshots to see if it overlaps with any. The list of old snapshots is split into two lists: an efficient concurrent cache and an slow vector protected by a lock. The patch fixes a bug that would stop the search in the cache if it finds any and yet would not include the larger snapshots in the slower list. An extra info log entry is also removed. The condition to trigger that although very rare is still feasible and should not spam the LOG when that happens. Fixes #4621 Pull Request resolved: https://github.com/facebook/rocksdb/pull/4639 Differential Revision: D12934989 Pulled By: maysamyabandeh fbshipit-source-id: 4e0fe8147ba292b554ae78e94c21c2ef31e03e2d	2018-11-06 13:49:35 -08:00
Fosco Marotto	516c7df05d	Revert "Introduce CacheAllocator, a custom allocator for cache blocks (#4437 )" This reverts commit `bb3b2eb960`.	2018-10-31 14:55:11 -07:00
Igor Canadi	bb3b2eb960	Introduce CacheAllocator, a custom allocator for cache blocks (#4437 ) Summary: This is a conceptually simple change, but it touches many files to pass the allocator through function calls. We introduce CacheAllocator, which can be used by clients to configure custom allocator for cache blocks. Our motivation is to hook this up with folly's `JemallocNodumpAllocator` (https://github.com/facebook/folly/blob/f43ce6d6866b7b994b3019df561109afae050ebc/folly/experimental/JemallocNodumpAllocator.h), but there are many other possible use cases. Additionally, this commit cleans up memory allocation in `util/compression.h`, making sure that all allocations are wrapped in a unique_ptr as soon as possible. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4437 Differential Revision: D10132814 Pulled By: yiwu-arbug fbshipit-source-id: be1343a4b69f6048df127939fea9bbc96969f564	2018-10-31 13:02:53 -07:00
Yi Wu	ae5305ee70	Fix compile error with jemalloc (#4488 ) Summary: The "je_" prefix of jemalloc APIs presents only when the macro `JEMALLOC_NO_RENAME` from jemalloc.h presents. With the patch I'm also adding -DROCKSDB_JEMALLOC flag in buck TARGETS. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4488 Differential Revision: D10355971 Pulled By: yiwu-arbug fbshipit-source-id: 03a2d69790a44ac89219c7525763fa937a63d95a	2018-10-30 12:37:06 -07:00
Anand Ananthabhotla	f37ea82191	Update HISTORY.md Summary: Test Plan: Reviewers: Subscribers: Tasks: Tags:	2018-10-19 16:38:03 -07:00
Anand Ananthabhotla	c954445ac8	Handle mixed slowdown/no_slowdown writer properly (#4475 ) Summary: There is a bug when the write queue leader is blocked on a write delay/stop, and the queue has writers with WriteOptions::no_slowdown set to true. They are not woken up until the write stall is cleared. The fix introduces a dummy writer inserted at the tail to indicate a write stall and prevent further inserts into the queue, and a condition variable that writers who can tolerate slowdown wait on before adding themselves to the queue. The leader calls WriteThread::BeginWriteStall() to add the dummy writer and then walk the queue to fail any writers with no_slowdown set. Once the stall clears, the leader calls WriteThread::EndWriteStall() to remove the dummy writer and signal the condition variable. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4475 Differential Revision: D10285827 Pulled By: anand1976 fbshipit-source-id: 747465e5e7f07a829b1fb0bc1afcd7b93f4ab1a9	2018-10-19 16:29:57 -07:00
Siying Dong	619f754816	Fix WriteBatchWithIndex's SeekForPrev() (#4559 ) Summary: WriteBatchWithIndex's SeekForPrev() has a bug that we internally place the position just before the seek key rather than after. This makes the iterator to miss the result that is the same as the seek key. Fix it by position the iterator equal or smaller. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4559 Differential Revision: D10468534 Pulled By: siying fbshipit-source-id: 2fb371ae809c561b60a1c11cef71e1c66fea1f19	2018-10-19 15:31:49 -07:00
Andrew Kryczka	f81fe96eea	update HISTORY.md and version number	2018-10-16 15:43:44 -07:00
anand1976	7d3eb006ff	Properly determine a truncated CompactRange stop key (#4496 ) Summary: When a CompactRange() call for a level is truncated before the end key is reached, because it exceeds max_compaction_bytes, we need to properly set the compaction_end parameter to indicate the stop key. The next CompactRange will use that as the begin key. We set it to the smallest key of the next file in the level after expanding inputs to get a clean cut. Previously, we were setting it before expanding inputs. So we could end up recompacting some files. In a pathological case, where a single key has many entries spanning all the files in the level (possibly due to merge operands without a partial merge operator, thus resulting in compaction output identical to the input), this would result in an endless loop over the same set of files. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4496 Differential Revision: D10395026 Pulled By: anand1976 fbshipit-source-id: f0c2f89fee29b4b3be53b6467b53abba8e9146a9	2018-10-16 15:42:06 -07:00
Andrew Kryczka	02d47c1ff4	Avoid per-key linear scan over snapshots in compaction (#4495 ) Summary: `CompactionIterator::snapshots_` is ordered by ascending seqnum, just like `DBImpl`'s linked list of snapshots from which it was copied. This PR exploits this ordering to make `findEarliestVisibleSnapshot` do binary search rather than linear scan. This can make flush/compaction significantly faster when many snapshots exist since that function is called on every single key. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4495 Differential Revision: D10386470 Pulled By: ajkr fbshipit-source-id: 29734991631227b6b7b677e156ac567690118a8b	2018-10-16 13:11:32 -07:00
Fosco Marotto	0f17fb9622	Update version macro for 5.17 (#4472 ) Summary: Forgot this in previous commit. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4472 Differential Revision: D10244227 Pulled By: gfosco fbshipit-source-id: ba0cf7a2f5271f0d9f9443004e2620887cd5fd11	2018-10-08 16:26:12 -07:00
Fosco Marotto	9552660ecc	Update HISTORY.md to current status (#4471 ) Summary: 5.16.x status wasn't tracked, and also updated for pending 5.17 release. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4471 Differential Revision: D10240925 Pulled By: gfosco fbshipit-source-id: 95ab368a04a65b201d2518097af69edf2402f544	2018-10-08 11:21:55 -07:00
Andrew Gallagher	016437d1cd	rocksdb: put `#pragma once` before `#ifdef` Summary: Work around upstream bug with modules: https://bugs.llvm.org/show_bug.cgi?id=39184. Reviewed By: yiwu-arbug Differential Revision: D10209569 fbshipit-source-id: 696853a02a3869e9c33d0e61168ad4b0436fa3c0	2018-10-05 10:30:11 -07:00
Fosco Marotto	b6b72687c6	Revert "Introduce CacheAllocator, a custom allocator for cache blocks (#4437 )" This reverts commit `1cf5deb8fd`.	2018-10-04 15:17:30 -07:00