StackableDB optionally take shared ownership of the underlying DB

Summary:
Allow StackableDB optionally takes a shared_ptr on construction and thus hold shared ownership of the underlying DB.
Closes https://github.com/facebook/rocksdb/pull/3423

Differential Revision: D6824163

Pulled By: yiwu-arbug

fbshipit-source-id: dbdc30c42e007533a987ef413785e192340f03eb

db/db_impl.cc

@@ -1682,12 +1682,6 @@ void DBImpl::ReleaseSnapshot(const Snapshot* s) {
   delete casted_s;
 }
 
-bool DBImpl::HasActiveSnapshotInRange(SequenceNumber lower_bound,
-                                      SequenceNumber upper_bound) {
-  InstrumentedMutexLock l(&mutex_);
-  return snapshots_.HasSnapshotInRange(lower_bound, upper_bound);
-}
-
 #ifndef ROCKSDB_LITE
 Status DBImpl::GetPropertiesOfAllTables(ColumnFamilyHandle* column_family,
                                         TablePropertiesCollection* props) {

db/db_impl.h

@@ -229,10 +229,6 @@ class DBImpl : public DB {
 
   virtual bool SetPreserveDeletesSequenceNumber(SequenceNumber seqnum) override;
 
-  // Whether there is an active snapshot in range [lower_bound, upper_bound).
-  bool HasActiveSnapshotInRange(SequenceNumber lower_bound,
-                                SequenceNumber upper_bound);
-
 #ifndef ROCKSDB_LITE
   using DB::ResetStats;
   virtual Status ResetStats() override;

db/db_write_test.cc

@@ -39,52 +39,6 @@ TEST_P(DBWriteTest, SyncAndDisableWAL) {
   ASSERT_TRUE(dbfull()->Write(write_options, &batch).IsInvalidArgument());
 }
 
-// Sequence number should be return through input write batch.
-TEST_P(DBWriteTest, ReturnSeuqneceNumber) {
-  Random rnd(4422);
-  Open();
-  for (int i = 0; i < 100; i++) {
-    WriteBatch batch;
-    batch.Put("key" + ToString(i), test::RandomHumanReadableString(&rnd, 10));
-    ASSERT_OK(dbfull()->Write(WriteOptions(), &batch));
-    ASSERT_EQ(dbfull()->GetLatestSequenceNumber(),
-              WriteBatchInternal::Sequence(&batch));
-  }
-}
-
-TEST_P(DBWriteTest, ReturnSeuqneceNumberMultiThreaded) {
-  constexpr size_t kThreads = 16;
-  constexpr size_t kNumKeys = 1000;
-  Open();
-  ASSERT_EQ(0, dbfull()->GetLatestSequenceNumber());
-  // Check each sequence is used once and only once.
-  std::vector<std::atomic_flag> flags(kNumKeys * kThreads + 1);
-  for (size_t i = 0; i < flags.size(); i++) {
-    flags[i].clear();
-  }
-  auto writer = [&](size_t id) {
-    Random rnd(4422 + static_cast<uint32_t>(id));
-    for (size_t k = 0; k < kNumKeys; k++) {
-      WriteBatch batch;
-      batch.Put("key" + ToString(id) + "-" + ToString(k),
-                test::RandomHumanReadableString(&rnd, 10));
-      ASSERT_OK(dbfull()->Write(WriteOptions(), &batch));
-      SequenceNumber sequence = WriteBatchInternal::Sequence(&batch);
-      ASSERT_GT(sequence, 0);
-      ASSERT_LE(sequence, kNumKeys * kThreads);
-      // The sequence isn't consumed by someone else.
-      ASSERT_FALSE(flags[sequence].test_and_set());
-    }
-  };
-  std::vector<port::Thread> threads;
-  for (size_t i = 0; i < kThreads; i++) {
-    threads.emplace_back(writer, i);
-  }
-  for (size_t i = 0; i < kThreads; i++) {
-    threads[i].join();
-  }
-}
-
 TEST_P(DBWriteTest, IOErrorOnWALWritePropagateToWriteThreadFollower) {
   constexpr int kNumThreads = 5;
   std::unique_ptr<FaultInjectionTestEnv> mock_env(

db/snapshot_impl.h

@@ -108,22 +108,6 @@ class SnapshotList {
     return ret;
   }
 
-  // Whether there is an active snapshot in range [lower_bound, upper_bound).
-  bool HasSnapshotInRange(SequenceNumber lower_bound,
-                          SequenceNumber upper_bound) {
-    if (empty()) {
-      return false;
-    }
-    const SnapshotImpl* s = &list_;
-    while (s->next_ != &list_) {
-      if (s->next_->number_ >= lower_bound) {
-        return s->next_->number_ < upper_bound;
-      }
-      s = s->next_;
-    }
-    return false;
-  }
-
   // get the sequence number of the most recent snapshot
   SequenceNumber GetNewest() {
     if (empty()) {

include/rocksdb/statistics.h

@@ -415,7 +415,7 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
     {BLOB_DB_WRITE_BLOB, "rocksdb.blobdb.write.blob"},
     {BLOB_DB_WRITE_BLOB_TTL, "rocksdb.blobdb.write.blob.ttl"},
     {BLOB_DB_BLOB_FILE_BYTES_WRITTEN, "rocksdb.blobdb.blob.file.bytes.written"},
-    {BLOB_DB_BLOB_FILE_BYTES_READ, "rocksdb.blobdb.blob.file,bytes.read"},
+    {BLOB_DB_BLOB_FILE_BYTES_READ, "rocksdb.blobdb.blob.file.bytes.read"},
     {BLOB_DB_BLOB_FILE_SYNCED, "rocksdb.blobdb.blob.file.synced"},
     {BLOB_DB_BLOB_INDEX_EXPIRED, "rocksdb.blobdb.blob.index.expired"},
     {BLOB_DB_GC_NUM_FILES, "rocksdb.blobdb.gc.num.files"},

include/rocksdb/utilities/stackable_db.h

@@ -4,6 +4,7 @@
 
 #pragma once
 #include <map>
+#include <memory>
 #include <string>
 #include "rocksdb/db.h"
 
@@ -18,11 +19,20 @@ namespace rocksdb {
 // This class contains APIs to stack rocksdb wrappers.Eg. Stack TTL over base d
 class StackableDB : public DB {
  public:
-  // StackableDB is the owner of db now!
+  // StackableDB take sole ownership of the underlying db.
   explicit StackableDB(DB* db) : db_(db) {}
 
+  // StackableDB take shared ownership of the underlying db.
+  explicit StackableDB(std::shared_ptr<DB> db)
+      : db_(db.get()), shared_db_ptr_(db) {}
+
   ~StackableDB() {
-    delete db_;
+    if (shared_db_ptr_ == nullptr) {
+      delete db_;
+    } else {
+      assert(shared_db_ptr_.get() == db_);
+    }
+    db_ = nullptr;
   }
 
   virtual DB* GetBaseDB() {
@@ -373,6 +383,7 @@ class StackableDB : public DB {
 
  protected:
   DB* db_;
+  std::shared_ptr<DB> shared_db_ptr_;
 };
 
 } //  namespace rocksdb

utilities/blob_db/blob_db.cc

@@ -73,6 +73,8 @@ void BlobDBOptions::Dump(Logger* log) const {
                    blob_dir_size);
   ROCKS_LOG_HEADER(log, "           blob_db_options.ttl_range_secs: %" PRIu32,
                    ttl_range_secs);
+  ROCKS_LOG_HEADER(log, "            blob_db_options.min_blob_size: %" PRIu64,
+                   min_blob_size);
   ROCKS_LOG_HEADER(log, "           blob_db_options.bytes_per_sync: %" PRIu64,
                    bytes_per_sync);
   ROCKS_LOG_HEADER(log, "           blob_db_options.blob_file_size: %" PRIu64,

utilities/blob_db/blob_db.h

@@ -56,8 +56,10 @@ struct BlobDBOptions {
   // will be inlined in base DB together with the key.
   uint64_t min_blob_size = 0;
 
-  // at what bytes will the blob files be synced to blob log.
-  uint64_t bytes_per_sync = 0;
+  // Allows OS to incrementally sync blob files to disk for every
+  // bytes_per_sync bytes written. Users shouldn't rely on it for
+  // persistency guarantee.
+  uint64_t bytes_per_sync = 512 * 1024;
 
   // the target size of each blob file. File will become immutable
   // after it exceeds that size
@@ -76,6 +78,13 @@ struct BlobDBOptions {
   // blob files will be cleanup based on TTL.
   bool enable_garbage_collection = false;
 
+  // Time interval to trigger garbage collection, in seconds.
+  uint64_t garbage_collection_interval_secs = 60;
+
+  // If garbage collection is enabled, blob files with deleted size no less
+  // than this ratio will become candidates to be cleanup.
+  double garbage_collection_deletion_size_threshold = 0.75;
+
   // Disable all background job. Used for test only.
   bool disable_background_tasks = false;
 
@@ -203,6 +212,8 @@ class BlobDB : public StackableDB {
 
   virtual BlobDBOptions GetBlobDBOptions() const = 0;
 
+  virtual Status SyncBlobFiles() = 0;
+
   virtual ~BlobDB() {}
 
  protected:

utilities/blob_db/blob_db_impl.cc

@@ -111,7 +111,6 @@ BlobDBImpl::BlobDBImpl(const std::string& dbname,
       cf_options_(cf_options),
       env_options_(db_options),
       statistics_(db_options_.statistics.get()),
-      dir_change_(false),
       next_file_number_(1),
       epoch_of_(0),
       shutdown_(false),
@@ -124,6 +123,7 @@ BlobDBImpl::BlobDBImpl(const std::string& dbname,
   blob_dir_ = (bdb_options_.path_relative)
                   ? dbname + "/" + bdb_options_.blob_dir
                   : bdb_options_.blob_dir;
+  env_options_.bytes_per_sync = blob_db_options.bytes_per_sync;
 }
 
 BlobDBImpl::~BlobDBImpl() {
@@ -202,6 +202,7 @@ Status BlobDBImpl::Open(std::vector<ColumnFamilyHandle*>* handles) {
   }
 
   ROCKS_LOG_INFO(db_options_.info_log, "BlobDB pointer %p", this);
+  bdb_options_.Dump(db_options_.info_log.get());
   return s;
 }
 
@@ -210,7 +211,8 @@ void BlobDBImpl::StartBackgroundTasks() {
   tqueue_.add(
       kReclaimOpenFilesPeriodMillisecs,
       std::bind(&BlobDBImpl::ReclaimOpenFiles, this, std::placeholders::_1));
-  tqueue_.add(kGCCheckPeriodMillisecs,
+  tqueue_.add(static_cast<int64_t>(
+                  bdb_options_.garbage_collection_interval_secs * 1000),
               std::bind(&BlobDBImpl::RunGC, this, std::placeholders::_1));
   if (bdb_options_.enable_garbage_collection) {
     tqueue_.add(
@@ -225,8 +227,6 @@ void BlobDBImpl::StartBackgroundTasks() {
       std::bind(&BlobDBImpl::DeleteObsoleteFiles, this, std::placeholders::_1));
   tqueue_.add(kSanityCheckPeriodMillisecs,
               std::bind(&BlobDBImpl::SanityCheck, this, std::placeholders::_1));
-  tqueue_.add(kFSyncFilesPeriodMillisecs,
-              std::bind(&BlobDBImpl::FsyncFiles, this, std::placeholders::_1));
   tqueue_.add(
       kCheckSeqFilesPeriodMillisecs,
       std::bind(&BlobDBImpl::CheckSeqFiles, this, std::placeholders::_1));
@@ -472,7 +472,6 @@ std::shared_ptr<BlobFile> BlobDBImpl::SelectBlobFile() {
     return nullptr;
   }
 
-  dir_change_.store(true);
   blob_files_.insert(std::make_pair(bfile->BlobFileNumber(), bfile));
   open_non_ttl_file_ = bfile;
   return bfile;
@@ -547,7 +546,6 @@ std::shared_ptr<BlobFile> BlobDBImpl::SelectBlobFileTTL(uint64_t expiration) {
     return nullptr;
   }
 
-  dir_change_.store(true);
   blob_files_.insert(std::make_pair(bfile->BlobFileNumber(), bfile));
   open_ttl_files_.insert(bfile);
   epoch_of_++;
@@ -571,18 +569,14 @@ class BlobDBImpl::BlobInserter : public WriteBatch::Handler {
   const WriteOptions& options_;
   BlobDBImpl* blob_db_impl_;
   uint32_t default_cf_id_;
-  SequenceNumber sequence_;
   WriteBatch batch_;
 
  public:
   BlobInserter(const WriteOptions& options, BlobDBImpl* blob_db_impl,
-               uint32_t default_cf_id, SequenceNumber seq)
+               uint32_t default_cf_id)
       : options_(options),
         blob_db_impl_(blob_db_impl),
-        default_cf_id_(default_cf_id),
-        sequence_(seq) {}
-
-  SequenceNumber sequence() { return sequence_; }
+        default_cf_id_(default_cf_id) {}
 
   WriteBatch* batch() { return &batch_; }
 
@@ -597,8 +591,7 @@ class BlobDBImpl::BlobInserter : public WriteBatch::Handler {
     uint64_t expiration =
         blob_db_impl_->ExtractExpiration(key, value, &value_slice, &new_value);
     Status s = blob_db_impl_->PutBlobValue(options_, key, value_slice,
-                                           expiration, sequence_, &batch_);
-    sequence_++;
+                                           expiration, &batch_);
     return s;
   }
 
@@ -609,7 +602,6 @@ class BlobDBImpl::BlobInserter : public WriteBatch::Handler {
           "Blob DB doesn't support non-default column family.");
     }
     Status s = WriteBatchInternal::Delete(&batch_, column_family_id, key);
-    sequence_++;
     return s;
   }
 
@@ -621,7 +613,6 @@ class BlobDBImpl::BlobInserter : public WriteBatch::Handler {
     }
     Status s = WriteBatchInternal::DeleteRange(&batch_, column_family_id,
                                                begin_key, end_key);
-    sequence_++;
     return s;
   }
 
@@ -643,12 +634,8 @@ Status BlobDBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
   RecordTick(statistics_, BLOB_DB_NUM_WRITE);
   uint32_t default_cf_id =
       reinterpret_cast<ColumnFamilyHandleImpl*>(DefaultColumnFamily())->GetID();
-  // TODO(yiwu): In case there are multiple writers the latest sequence would
-  // not be the actually sequence we are writting. Need to get the sequence
-  // from write batch after DB write instead.
-  SequenceNumber current_seq = GetLatestSequenceNumber() + 1;
   Status s;
-  BlobInserter blob_inserter(options, this, default_cf_id, current_seq);
+  BlobInserter blob_inserter(options, this, default_cf_id);
   {
     // Release write_mutex_ before DB write to avoid race condition with
     // flush begin listener, which also require write_mutex_ to sync
@@ -693,6 +680,8 @@ Status BlobDBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
 
   if (bdb_options_.enable_garbage_collection) {
     // add deleted key to list of keys that have been deleted for book-keeping
+    SequenceNumber current_seq =
+        WriteBatchInternal::Sequence(blob_inserter.batch());
     DeleteBookkeeper delete_bookkeeper(this, current_seq);
     s = updates->Iterate(&delete_bookkeeper);
   }
@@ -761,11 +750,7 @@ Status BlobDBImpl::PutUntil(const WriteOptions& options, const Slice& key,
     // flush begin listener, which also require write_mutex_ to sync
     // blob files.
     MutexLock l(&write_mutex_);
-    // TODO(yiwu): In case there are multiple writers the latest sequence would
-    // not be the actually sequence we are writting. Need to get the sequence
-    // from write batch after DB write instead.
-    SequenceNumber sequence = GetLatestSequenceNumber() + 1;
-    s = PutBlobValue(options, key, value, expiration, sequence, &batch);
+    s = PutBlobValue(options, key, value, expiration, &batch);
   }
   if (s.ok()) {
     s = db_->Write(options, &batch);
@@ -776,7 +761,7 @@ Status BlobDBImpl::PutUntil(const WriteOptions& options, const Slice& key,
 
 Status BlobDBImpl::PutBlobValue(const WriteOptions& options, const Slice& key,
                                 const Slice& value, uint64_t expiration,
-                                SequenceNumber sequence, WriteBatch* batch) {
+                                WriteBatch* batch) {
   Status s;
   std::string index_entry;
   uint32_t column_family_id =
@@ -817,7 +802,6 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& options, const Slice& key,
     }
 
     if (s.ok()) {
-      bfile->ExtendSequenceRange(sequence);
       if (expiration != kNoExpiration) {
         bfile->ExtendExpirationRange(expiration);
       }
@@ -876,6 +860,9 @@ std::shared_ptr<BlobFile> BlobDBImpl::GetOldestBlobFile() {
   CopyBlobFiles(&blob_files, [](const std::shared_ptr<BlobFile>& f) {
     return !f->Obsolete() && f->Immutable();
   });
+  if (blob_files.empty()) {
+    return nullptr;
+  }
   blobf_compare_ttl compare;
   return *std::min_element(blob_files.begin(), blob_files.end(), compare);
 }
@@ -898,7 +885,7 @@ bool BlobDBImpl::EvictOldestBlobFile() {
                    total_blob_space_.load(), bdb_options_.blob_dir_size,
                    oldest_file->BlobFileNumber(), expiration_range.first,
                    expiration_range.second);
-    oldest_file->MarkObsolete(oldest_file->GetSequenceRange().second);
+    oldest_file->MarkObsolete(GetLatestSequenceNumber());
     obsolete_files_.push_back(oldest_file);
     oldest_file_evicted_.store(true);
     RecordTick(statistics_, BLOB_DB_FIFO_NUM_FILES_EVICTED);
@@ -906,6 +893,7 @@ bool BlobDBImpl::EvictOldestBlobFile() {
                oldest_file->BlobCount());
     RecordTick(statistics_, BLOB_DB_FIFO_BYTES_EVICTED,
                oldest_file->GetFileSize());
+    TEST_SYNC_POINT("BlobDBImpl::EvictOldestBlobFile:Evicted");
     return true;
   }
 
@@ -1072,58 +1060,58 @@ Status BlobDBImpl::GetBlobValue(const Slice& key, const Slice& index_entry,
   std::shared_ptr<RandomAccessFileReader> reader =
       GetOrOpenRandomAccessReader(bfile, env_, env_options_);
 
-  std::string* valueptr = value->GetSelf();
-  std::string value_c;
-  if (bdb_options_.compression != kNoCompression) {
-    valueptr = &value_c;
-  }
+  assert(blob_index.offset() > key.size() + sizeof(uint32_t));
+  uint64_t record_offset = blob_index.offset() - key.size() - sizeof(uint32_t);
+  uint64_t record_size = sizeof(uint32_t) + key.size() + blob_index.size();
 
   // Allocate the buffer. This is safe in C++11
-  // Note that std::string::reserved() does not work, since previous value
-  // of the buffer can be larger than blob_index.size().
-  valueptr->resize(blob_index.size());
-  char* buffer = &(*valueptr)[0];
+  std::string buffer_str(record_size, static_cast<char>(0));
+  char* buffer = &buffer_str[0];
 
-  Slice blob_value;
+  // A partial blob record contain checksum, key and value.
+  Slice blob_record;
   {
     StopWatch read_sw(env_, statistics_, BLOB_DB_BLOB_FILE_READ_MICROS);
-    s = reader->Read(blob_index.offset(), blob_index.size(), &blob_value,
-                     buffer);
-    RecordTick(statistics_, BLOB_DB_BLOB_FILE_BYTES_READ, blob_value.size());
+    s = reader->Read(record_offset, record_size, &blob_record, buffer);
+    RecordTick(statistics_, BLOB_DB_BLOB_FILE_BYTES_READ, blob_record.size());
   }
-  if (!s.ok() || blob_value.size() != blob_index.size()) {
-    if (debug_level_ >= 2) {
-      ROCKS_LOG_ERROR(db_options_.info_log,
-                      "Failed to read blob from file: %s blob_offset: %" PRIu64
-                      " blob_size: %" PRIu64 " read: %d key: %s status: '%s'",
-                      bfile->PathName().c_str(), blob_index.offset(),
-                      blob_index.size(), static_cast<int>(blob_value.size()),
-                      key.data(), s.ToString().c_str());
-    }
-    return Status::NotFound("Blob Not Found as couldnt retrieve Blob");
+  if (!s.ok()) {
+    ROCKS_LOG_DEBUG(db_options_.info_log,
+                    "Failed to read blob from blob file %" PRIu64
+                    ", blob_offset: %" PRIu64 ", blob_size: %" PRIu64
+                    ", key_size: " PRIu64 ", read " PRIu64
+                    "bytes, status: '%s'",
+                    bfile->BlobFileNumber(), blob_index.offset(),
+                    blob_index.size(), key.size(), s.ToString().c_str());
+    return s;
   }
+  if (blob_record.size() != record_size) {
+    ROCKS_LOG_DEBUG(db_options_.info_log,
+                    "Failed to read blob from blob file %" PRIu64
+                    ", blob_offset: %" PRIu64 ", blob_size: %" PRIu64
+                    ", key_size: " PRIu64 ", read " PRIu64
+                    "bytes, status: '%s'",
+                    bfile->BlobFileNumber(), blob_index.offset(),
+                    blob_index.size(), key.size(), s.ToString().c_str());
 
-  // TODO(yiwu): Add an option to skip crc checking.
-  Slice crc_slice;
+    return Status::Corruption("Failed to retrieve blob from blob index.");
+  }
+  Slice crc_slice(blob_record.data(), sizeof(uint32_t));
+  Slice blob_value(blob_record.data() + sizeof(uint32_t) + key.size(),
+                   blob_index.size());
   uint32_t crc_exp;
-  std::string crc_str;
-  crc_str.resize(sizeof(uint32_t));
-  char* crc_buffer = &(crc_str[0]);
-  s = reader->Read(blob_index.offset() - (key.size() + sizeof(uint32_t)),
-                   sizeof(uint32_t), &crc_slice, crc_buffer);
-  if (!s.ok() || !GetFixed32(&crc_slice, &crc_exp)) {
-    if (debug_level_ >= 2) {
-      ROCKS_LOG_ERROR(db_options_.info_log,
-                      "Failed to fetch blob crc file: %s blob_offset: %" PRIu64
-                      " blob_size: %" PRIu64 " key: %s status: '%s'",
-                      bfile->PathName().c_str(), blob_index.offset(),
-                      blob_index.size(), key.data(), s.ToString().c_str());
-    }
-    return Status::NotFound("Blob Not Found as couldnt retrieve CRC");
+  if (!GetFixed32(&crc_slice, &crc_exp)) {
+    ROCKS_LOG_DEBUG(db_options_.info_log,
+                    "Unable to decode CRC from blob file %" PRIu64
+                    ", blob_offset: %" PRIu64 ", blob_size: %" PRIu64
+                    ", key size: %" PRIu64 ", status: '%s'",
+                    bfile->BlobFileNumber(), blob_index.offset(),
+                    blob_index.size(), key.size(), s.ToString().c_str());
+    return Status::Corruption("Unable to decode checksum.");
   }
 
-  uint32_t crc = crc32c::Value(key.data(), key.size());
-  crc = crc32c::Extend(crc, blob_value.data(), blob_value.size());
+  uint32_t crc = crc32c::Value(blob_record.data() + sizeof(uint32_t),
+                               blob_record.size() - sizeof(uint32_t));
   crc = crc32c::Mask(crc);  // Adjust for storage
   if (crc != crc_exp) {
     if (debug_level_ >= 2) {
@@ -1136,7 +1124,9 @@ Status BlobDBImpl::GetBlobValue(const Slice& key, const Slice& index_entry,
     return Status::Corruption("Corruption. Blob CRC mismatch");
   }
 
-  if (bfile->compression() != kNoCompression) {
+  if (bfile->compression() == kNoCompression) {
+    value->PinSelf(blob_value);
+  } else {
     BlockContents contents;
     auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(DefaultColumnFamily());
     {
@@ -1147,11 +1137,9 @@ Status BlobDBImpl::GetBlobValue(const Slice& key, const Slice& index_entry,
           kBlockBasedTableVersionFormat, Slice(), bfile->compression(),
           *(cfh->cfd()->ioptions()));
     }
-    *(value->GetSelf()) = contents.data.ToString();
+    value->PinSelf(contents.data);
   }
 
-  value->PinSelf();
-
   return s;
 }
 
@@ -1271,9 +1259,26 @@ Status BlobDBImpl::CloseBlobFileIfNeeded(std::shared_ptr<BlobFile>& bfile) {
 bool BlobDBImpl::VisibleToActiveSnapshot(
     const std::shared_ptr<BlobFile>& bfile) {
   assert(bfile->Obsolete());
-  SequenceNumber first_sequence = bfile->GetSequenceRange().first;
+
+  // We check whether the oldest snapshot is no less than the last sequence
+  // by the time the blob file become obsolete. If so, the blob file is not
+  // visible to all existing snapshots.
+  //
+  // If we keep track of the earliest sequence of the keys in the blob file,
+  // we could instead check if there's a snapshot falls in range
+  // [earliest_sequence, obsolete_sequence). But doing so will make the
+  // implementation more complicated.
   SequenceNumber obsolete_sequence = bfile->GetObsoleteSequence();
-  return db_impl_->HasActiveSnapshotInRange(first_sequence, obsolete_sequence);
+  SequenceNumber oldest_snapshot = 0;
+  {
+    // Need to lock DBImpl mutex before access snapshot list.
+    InstrumentedMutexLock l(db_impl_->mutex());
+    auto snapshots = db_impl_->snapshots();
+    if (!snapshots.empty()) {
+      oldest_snapshot = snapshots.oldest()->GetSequenceNumber();
+    }
+  }
+  return oldest_snapshot < obsolete_sequence;
 }
 
 bool BlobDBImpl::FindFileAndEvictABlob(uint64_t file_number, uint64_t key_size,
@@ -1431,14 +1436,6 @@ std::pair<bool, int64_t> BlobDBImpl::CheckSeqFiles(bool aborted) {
   return std::make_pair(true, -1);
 }
 
-std::pair<bool, int64_t> BlobDBImpl::FsyncFiles(bool aborted) {
-  if (aborted || shutdown_) {
-    return std::make_pair(false, -1);
-  }
-  SyncBlobFiles();
-  return std::make_pair(true, -1);
-}
-
 Status BlobDBImpl::SyncBlobFiles() {
   MutexLock l(&write_mutex_);
 
@@ -1446,35 +1443,30 @@ Status BlobDBImpl::SyncBlobFiles() {
   {
     ReadLock rl(&mutex_);
     for (auto fitr : open_ttl_files_) {
-      if (fitr->NeedsFsync(true, bdb_options_.bytes_per_sync))
-        process_files.push_back(fitr);
+      process_files.push_back(fitr);
     }
-
-    if (open_non_ttl_file_ != nullptr &&
-        open_non_ttl_file_->NeedsFsync(true, bdb_options_.bytes_per_sync)) {
+    if (open_non_ttl_file_ != nullptr) {
       process_files.push_back(open_non_ttl_file_);
     }
   }
 
   Status s;
-
   for (auto& blob_file : process_files) {
-    if (blob_file->NeedsFsync(true, bdb_options_.bytes_per_sync)) {
-      s = blob_file->Fsync();
-      if (!s.ok()) {
-        ROCKS_LOG_ERROR(db_options_.info_log,
-                        "Failed to sync blob file %" PRIu64 ", status: %s",
-                        blob_file->BlobFileNumber(), s.ToString().c_str());
-        return s;
-      }
+    s = blob_file->Fsync();
+    if (!s.ok()) {
+      ROCKS_LOG_ERROR(db_options_.info_log,
+                      "Failed to sync blob file %" PRIu64 ", status: %s",
+                      blob_file->BlobFileNumber(), s.ToString().c_str());
+      return s;
     }
   }
 
-  bool expected = true;
-  if (dir_change_.compare_exchange_weak(expected, false)) {
-    s = dir_ent_->Fsync();
+  s = dir_ent_->Fsync();
+  if (!s.ok()) {
+    ROCKS_LOG_ERROR(db_options_.info_log,
+                    "Failed to sync blob directory, status: %s",
+                    s.ToString().c_str());
   }
-
   return s;
 }
 
@@ -1729,7 +1721,6 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
 
       WriteLock wl(&mutex_);
 
-      dir_change_.store(true);
       blob_files_.insert(std::make_pair(newfile->BlobFileNumber(), newfile));
     }
 
@@ -1757,8 +1748,6 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
                                                    &rewrite_batch, &callback);
     }
     if (rewrite_status.ok()) {
-      newfile->ExtendSequenceRange(
-          WriteBatchInternal::Sequence(&rewrite_batch));
       gc_stats->num_keys_relocated++;
       gc_stats->bytes_relocated += record.record_size();
     } else if (rewrite_status.IsBusy()) {
@@ -1775,10 +1764,7 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
   }  // end of ReadRecord loop
 
   if (s.ok()) {
-    SequenceNumber obsolete_sequence =
-        newfile == nullptr ? bfptr->GetSequenceRange().second + 1
-                           : newfile->GetSequenceRange().second;
-    bfptr->MarkObsolete(obsolete_sequence);
+    bfptr->MarkObsolete(GetLatestSequenceNumber());
     if (!first_gc) {
       WriteLock wl(&mutex_);
       obsolete_files_.push_back(bfptr);
@@ -1788,8 +1774,9 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
   ROCKS_LOG_INFO(
       db_options_.info_log,
       "%s blob file %" PRIu64 ". Total blob records: %" PRIu64
-      ", Expired: %" PRIu64 " keys/%" PRIu64 " bytes, Overwritten: %" PRIu64
-      " keys/%" PRIu64 " bytes.",
+      ", expired: %" PRIu64 " keys/%" PRIu64
+      " bytes, updated or deleted by user: %" PRIu64 " keys/%" PRIu64
+      " bytes, rewrite to new file: %" PRIu64 " keys/%" PRIu64 " bytes.",
       s.ok() ? "Successfully garbage collected" : "Failed to garbage collect",
       bfptr->BlobFileNumber(), gc_stats->blob_count, gc_stats->num_keys_expired,
       gc_stats->bytes_expired, gc_stats->num_keys_overwritten,
@@ -1843,14 +1830,9 @@ bool BlobDBImpl::ShouldGCFile(std::shared_ptr<BlobFile> bfile, uint64_t now,
       return true;
     }
 
-    if (bdb_options_.ttl_range_secs < kPartialExpirationGCRangeSecs) {
-      *reason = "has ttl but partial expiration not turned on";
-      return false;
-    }
-
     ReadLock lockbfile_r(&bfile->mutex_);
-    bool ret = ((bfile->deleted_size_ * 100.0 / bfile->file_size_.load()) >
-                kPartialExpirationPercentage);
+    bool ret = ((bfile->deleted_size_ / bfile->file_size_.load()) >
+                bdb_options_.garbage_collection_deletion_size_threshold);
     if (ret) {
       *reason = "deleted blobs beyond threshold";
     } else {
@@ -1869,8 +1851,8 @@ bool BlobDBImpl::ShouldGCFile(std::shared_ptr<BlobFile> bfile, uint64_t now,
   ReadLock lockbfile_r(&bfile->mutex_);
 
   if (bdb_options_.enable_garbage_collection) {
-    if ((bfile->deleted_size_ * 100.0 / bfile->file_size_.load()) >
-        kPartialExpirationPercentage) {
+    if ((bfile->deleted_size_ / bfile->file_size_.load()) >
+        bdb_options_.garbage_collection_deletion_size_threshold) {
       *reason = "deleted simple blobs beyond threshold";
       return true;
     }

utilities/blob_db/blob_db_impl.h

@@ -144,23 +144,6 @@ class BlobDBImpl : public BlobDB {
   // how many random access open files can we tolerate
   static constexpr uint32_t kOpenFilesTrigger = 100;
 
-  // how many periods of stats do we keep.
-  static constexpr uint32_t kWriteAmplificationStatsPeriods = 24;
-
-  // we will garbage collect blob files in
-  // which entire files have expired. However if the
-  // ttl_range of files is very large say a day, we
-  // would have to wait for the entire day, before we
-  // recover most of the space.
-  static constexpr uint32_t kPartialExpirationGCRangeSecs = 4 * 3600;
-
-  // this should be based on allowed Write Amplification
-  // if 50% of the space of a blob file has been deleted/expired,
-  static constexpr uint32_t kPartialExpirationPercentage = 75;
-
-  // how often should we schedule a job to fsync open files
-  static constexpr uint32_t kFSyncFilesPeriodMillisecs = 10 * 1000;
-
   // how often to schedule reclaim open files.
   static constexpr uint32_t kReclaimOpenFilesPeriodMillisecs = 1 * 1000;
 
@@ -228,7 +211,7 @@ class BlobDBImpl : public BlobDB {
 
   Status Open(std::vector<ColumnFamilyHandle*>* handles);
 
-  Status SyncBlobFiles();
+  Status SyncBlobFiles() override;
 
 #ifndef NDEBUG
   Status TEST_GetBlobValue(const Slice& key, const Slice& index_entry,
@@ -284,7 +267,7 @@ class BlobDBImpl : public BlobDB {
 
   Status PutBlobValue(const WriteOptions& options, const Slice& key,
                       const Slice& value, uint64_t expiration,
-                      SequenceNumber sequence, WriteBatch* batch);
+                      WriteBatch* batch);
 
   Status AppendBlob(const std::shared_ptr<BlobFile>& bfile,
                     const std::string& headerbuf, const Slice& key,
@@ -313,9 +296,6 @@ class BlobDBImpl : public BlobDB {
   // Major task to garbage collect expired and deleted blobs
   std::pair<bool, int64_t> RunGC(bool aborted);
 
-  // asynchronous task to fsync/fdatasync the open blob files
-  std::pair<bool, int64_t> FsyncFiles(bool aborted);
-
   // periodically check if open blob files and their TTL's has expired
   // if expired, close the sequential writer and make the file immutable
   std::pair<bool, int64_t> CheckSeqFiles(bool aborted);
@@ -420,8 +400,6 @@ class BlobDBImpl : public BlobDB {
   // pointer to directory
   std::unique_ptr<Directory> dir_ent_;
 
-  std::atomic<bool> dir_change_;
-
   // Read Write Mutex, which protects all the data structures
   // HEAVILY TRAFFICKED
   mutable port::RWMutex mutex_;

utilities/blob_db/blob_db_test.cc

@@ -901,20 +901,18 @@ TEST_F(BlobDBTest, SnapshotAndGarbageCollection) {
       ASSERT_EQ(1, gc_stats.blob_count);
       if (delete_key) {
         ASSERT_EQ(0, gc_stats.num_keys_relocated);
-        ASSERT_EQ(bfile->GetSequenceRange().second + 1,
-                  bfile->GetObsoleteSequence());
       } else {
         ASSERT_EQ(1, gc_stats.num_keys_relocated);
-        ASSERT_EQ(blob_db_->GetLatestSequenceNumber(),
-                  bfile->GetObsoleteSequence());
       }
+      ASSERT_EQ(blob_db_->GetLatestSequenceNumber(),
+                bfile->GetObsoleteSequence());
       if (i == 3) {
         snapshot = blob_db_->GetSnapshot();
       }
       size_t num_files = delete_key ? 3 : 4;
       ASSERT_EQ(num_files, blob_db_impl()->TEST_GetBlobFiles().size());
       blob_db_impl()->TEST_DeleteObsoleteFiles();
-      if (i == 0 || i == 3 || (i == 2 && delete_key)) {
+      if (i == 3) {
         // The snapshot shouldn't see data in bfile
         ASSERT_EQ(num_files - 1, blob_db_impl()->TEST_GetBlobFiles().size());
         blob_db_->ReleaseSnapshot(snapshot);
@@ -1063,15 +1061,19 @@ TEST_F(BlobDBTest, OutOfSpace) {
 }
 
 TEST_F(BlobDBTest, EvictOldestFileWhenCloseToSpaceLimit) {
-  // Use mock env to stop wall clock.
-  Options options;
   BlobDBOptions bdb_options;
   bdb_options.blob_dir_size = 270;
   bdb_options.blob_file_size = 100;
-  bdb_options.disable_background_tasks = true;
   bdb_options.is_fifo = true;
+  bdb_options.disable_background_tasks = true;
   Open(bdb_options);
 
+  std::atomic<int> evict_count{0};
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobDBImpl::EvictOldestBlobFile:Evicted",
+      [&](void *) { evict_count++; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
   // Each stored blob has an overhead of 32 bytes currently.
   // So a 100 byte blob should take up 132 bytes.
   std::string value(100, 'v');
@@ -1095,6 +1097,28 @@ TEST_F(BlobDBTest, EvictOldestFileWhenCloseToSpaceLimit) {
   bdb_impl->TEST_DeleteObsoleteFiles();
   obsolete_files = bdb_impl->TEST_GetObsoleteFiles();
   ASSERT_TRUE(obsolete_files.empty());
+  ASSERT_EQ(1, evict_count);
+}
+
+TEST_F(BlobDBTest, NoOldestFileToEvict) {
+  Options options;
+  BlobDBOptions bdb_options;
+  bdb_options.blob_dir_size = 1000;
+  bdb_options.blob_file_size = 5000;
+  bdb_options.is_fifo = true;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+
+  std::atomic<int> evict_count{0};
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobDBImpl::EvictOldestBlobFile:Evicted",
+      [&](void *) { evict_count++; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  std::string value(2000, 'v');
+  ASSERT_OK(Put("foo", std::string(2000, 'v')));
+  ASSERT_OK(Put("bar", std::string(2000, 'v')));
+  ASSERT_EQ(0, evict_count);
 }
 
 TEST_F(BlobDBTest, InlineSmallValues) {
@@ -1111,10 +1135,6 @@ TEST_F(BlobDBTest, InlineSmallValues) {
   Open(bdb_options, options);
   std::map<std::string, std::string> data;
   std::map<std::string, KeyVersion> versions;
-  SequenceNumber first_non_ttl_seq = kMaxSequenceNumber;
-  SequenceNumber first_ttl_seq = kMaxSequenceNumber;
-  SequenceNumber last_non_ttl_seq = 0;
-  SequenceNumber last_ttl_seq = 0;
   for (size_t i = 0; i < 1000; i++) {
     bool is_small_value = rnd.Next() % 2;
     bool has_ttl = rnd.Next() % 2;
@@ -1134,15 +1154,6 @@ TEST_F(BlobDBTest, InlineSmallValues) {
     versions[key] =
         KeyVersion(key, value, sequence,
                    (is_small_value && !has_ttl) ? kTypeValue : kTypeBlobIndex);
-    if (!is_small_value) {
-      if (!has_ttl) {
-        first_non_ttl_seq = std::min(first_non_ttl_seq, sequence);
-        last_non_ttl_seq = std::max(last_non_ttl_seq, sequence);
-      } else {
-        first_ttl_seq = std::min(first_ttl_seq, sequence);
-        last_ttl_seq = std::max(last_ttl_seq, sequence);
-      }
-    }
   }
   VerifyDB(data);
   VerifyBaseDB(versions);
@@ -1159,11 +1170,7 @@ TEST_F(BlobDBTest, InlineSmallValues) {
     ttl_file = blob_files[1];
   }
   ASSERT_FALSE(non_ttl_file->HasTTL());
-  ASSERT_EQ(first_non_ttl_seq, non_ttl_file->GetSequenceRange().first);
-  ASSERT_EQ(last_non_ttl_seq, non_ttl_file->GetSequenceRange().second);
   ASSERT_TRUE(ttl_file->HasTTL());
-  ASSERT_EQ(first_ttl_seq, ttl_file->GetSequenceRange().first);
-  ASSERT_EQ(last_ttl_seq, ttl_file->GetSequenceRange().second);
 }
 
 TEST_F(BlobDBTest, CompactionFilterNotSupported) {

utilities/blob_db/blob_dump_tool.cc

@@ -142,8 +142,6 @@ Status BlobDumpTool::DumpBlobLogFooter(uint64_t file_size,
   fprintf(stdout, "  Blob count       : %" PRIu64 "\n", footer.blob_count);
   fprintf(stdout, "  Expiration Range : %s\n",
           GetString(footer.expiration_range).c_str());
-  fprintf(stdout, "  Sequence Range   : %s\n",
-          GetString(footer.sequence_range).c_str());
   return s;
 }
 

utilities/blob_db/blob_file.cc

@@ -43,7 +43,6 @@ BlobFile::BlobFile()
       obsolete_(false),
       gc_once_after_open_(false),
       expiration_range_({0, 0}),
-      sequence_range_({kMaxSequenceNumber, 0}),
       last_access_(-1),
       last_fsync_(0),
       header_valid_(false),
@@ -67,7 +66,6 @@ BlobFile::BlobFile(const BlobDBImpl* p, const std::string& bdir, uint64_t fn,
       obsolete_(false),
       gc_once_after_open_(false),
       expiration_range_({0, 0}),
-      sequence_range_({kMaxSequenceNumber, 0}),
       last_access_(-1),
       last_fsync_(0),
       header_valid_(false),
@@ -116,12 +114,11 @@ std::string BlobFile::DumpState() const {
            " file_size: %" PRIu64 " deleted_count: %" PRIu64
            " deleted_size: %" PRIu64
            " closed: %d obsolete: %d expiration_range: (%" PRIu64 ", %" PRIu64
-           ") sequence_range: (%" PRIu64 " %" PRIu64 "), writer: %d reader: %d",
+           "), writer: %d reader: %d",
            path_to_dir_.c_str(), file_number_, blob_count_.load(),
            gc_epoch_.load(), file_size_.load(), deleted_count_, deleted_size_,
            closed_.load(), obsolete_.load(), expiration_range_.first,
-           expiration_range_.second, sequence_range_.first,
-           sequence_range_.second, (!!log_writer_), (!!ra_file_reader_));
+           expiration_range_.second, (!!log_writer_), (!!ra_file_reader_));
   return str;
 }
 
@@ -144,8 +141,6 @@ Status BlobFile::WriteFooterAndCloseLocked() {
     footer.expiration_range = expiration_range_;
   }
 
-  footer.sequence_range = sequence_range_;
-
   // this will close the file and reset the Writable File Pointer.
   Status s = log_writer_->AppendFooter(footer);
   if (s.ok()) {
@@ -185,7 +180,6 @@ Status BlobFile::SetFromFooterLocked(const BlobLogFooter& footer) {
   last_fsync_.store(file_size_);
   blob_count_ = footer.blob_count;
   expiration_range_ = footer.expiration_range;
-  sequence_range_ = footer.sequence_range;
   closed_ = true;
   return Status::OK();
 }

utilities/blob_db/blob_file.h

@@ -84,8 +84,6 @@ class BlobFile {
 
   ExpirationRange expiration_range_;
 
-  SequenceRange sequence_range_;
-
   // Sequential/Append writer for blobs
   std::shared_ptr<Writer> log_writer_;
 
@@ -177,17 +175,6 @@ class BlobFile {
     expiration_range_.second = std::max(expiration_range_.second, expiration);
   }
 
-  SequenceRange GetSequenceRange() const { return sequence_range_; }
-
-  void SetSequenceRange(SequenceRange sequence_range) {
-    sequence_range_ = sequence_range;
-  }
-
-  void ExtendSequenceRange(SequenceNumber sequence) {
-    sequence_range_.first = std::min(sequence_range_.first, sequence);
-    sequence_range_.second = std::max(sequence_range_.second, sequence);
-  }
-
   bool HasTTL() const { return has_ttl_; }
 
   void SetHasTTL(bool has_ttl) { has_ttl_ = has_ttl; }

utilities/blob_db/blob_log_format.cc

@@ -67,8 +67,6 @@ void BlobLogFooter::EncodeTo(std::string* dst) {
   PutFixed64(dst, blob_count);
   PutFixed64(dst, expiration_range.first);
   PutFixed64(dst, expiration_range.second);
-  PutFixed64(dst, sequence_range.first);
-  PutFixed64(dst, sequence_range.second);
   crc = crc32c::Value(dst->c_str(), dst->size());
   crc = crc32c::Mask(crc);
   PutFixed32(dst, crc);
@@ -82,14 +80,12 @@ Status BlobLogFooter::DecodeFrom(Slice src) {
                               "Unexpected blob file footer size");
   }
   uint32_t src_crc = 0;
-  src_crc = crc32c::Value(src.data(), BlobLogFooter::kSize - 4);
+  src_crc = crc32c::Value(src.data(), BlobLogFooter::kSize - sizeof(uint32_t));
   src_crc = crc32c::Mask(src_crc);
   uint32_t magic_number;
   if (!GetFixed32(&src, &magic_number) || !GetFixed64(&src, &blob_count) ||
       !GetFixed64(&src, &expiration_range.first) ||
-      !GetFixed64(&src, &expiration_range.second) ||
-      !GetFixed64(&src, &sequence_range.first) ||
-      !GetFixed64(&src, &sequence_range.second) || !GetFixed32(&src, &crc)) {
+      !GetFixed64(&src, &expiration_range.second) || !GetFixed32(&src, &crc)) {
     return Status::Corruption(kErrorMessage, "Error decoding content");
   }
   if (magic_number != kMagicNumber) {

utilities/blob_db/blob_log_format.h

@@ -24,7 +24,6 @@ constexpr uint32_t kVersion1 = 1;
 constexpr uint64_t kNoExpiration = std::numeric_limits<uint64_t>::max();
 
 using ExpirationRange = std::pair<uint64_t, uint64_t>;
-using SequenceRange = std::pair<uint64_t, uint64_t>;
 
 // Format of blob log file header (30 bytes):
 //
@@ -53,24 +52,23 @@ struct BlobLogHeader {
   Status DecodeFrom(Slice slice);
 };
 
-// Format of blob log file footer (48 bytes):
+// Format of blob log file footer (32 bytes):
 //
-//    +--------------+------------+-------------------+-------------------+------------+
-//    | magic number | blob count | expiration range  |  sequence range   | footer CRC |
-//    +--------------+------------+-------------------+-------------------+------------+
-//    |   Fixed32    |  Fixed64   | Fixed64 + Fixed64 | Fixed64 + Fixed64 |   Fixed32  |
-//    +--------------+------------+-------------------+-------------------+------------+
+//    +--------------+------------+-------------------+------------+
+//    | magic number | blob count | expiration range  | footer CRC |
+//    +--------------+------------+-------------------+------------+
+//    |   Fixed32    |  Fixed64   | Fixed64 + Fixed64 |   Fixed32  |
+//    +--------------+------------+-------------------+------------+
 //
 // The footer will be presented only when the blob file is properly closed.
 //
 // Unlike the same field in file header, expiration range in the footer is the
 // range of smallest and largest expiration of the data in this file.
 struct BlobLogFooter {
-  static constexpr size_t kSize = 48;
+  static constexpr size_t kSize = 32;
 
   uint64_t blob_count = 0;
   ExpirationRange expiration_range = std::make_pair(0, 0);
-  SequenceRange sequence_range = std::make_pair(0, 0);
   uint32_t crc = 0;
 
   void EncodeTo(std::string* dst);