Fix the titan invalid ratelimiter in Flush/GC (#210)

Specify IOPriority when creating blob files. Fix #209.
Signed-off-by: BaronStack <2689496754@qq.com>
Co-authored-by: Xinye Tao <xy.tao@outlook.com>

src/blob_file_manager.h

@@ -25,8 +25,14 @@ class BlobFileManager {
 
   // Creates a new file. The new file should not be accessed until
   // FinishFile() has been called.
-  // If successful, sets "*handle* to the new file handle.
-  virtual Status NewFile(std::unique_ptr<BlobFileHandle>* handle) = 0;
+  // If successful, sets "*handle* to the new file handle with given
+  // IOPriority.
+  //
+  // The reason why set the io priority for WritableFile in Flush,
+  // Compaction and GC is that the ratelimiter will use the default
+  // priority IO_TOTAL which won't be limited in ratelimiter.
+  virtual Status NewFile(std::unique_ptr<BlobFileHandle>* handle,
+                         Env::IOPriority pri = Env::IOPriority::IO_TOTAL) = 0;
 
   // Finishes the file with the provided metadata. Stops writting to
   // the file anymore.

src/blob_gc_job.cc

@@ -210,7 +210,8 @@ Status BlobGCJob::DoRunGC() {
         blob_file_builders_.emplace_back(std::make_pair(
             std::move(blob_file_handle), std::move(blob_file_builder)));
       }
-      s = blob_file_manager_->NewFile(&blob_file_handle);
+      s = blob_file_manager_->NewFile(&blob_file_handle,
+                                      Env::IOPriority::IO_LOW);
       if (!s.ok()) {
         break;
       }

src/blob_gc_job_test.cc

@@ -294,7 +294,8 @@ TEST_P(BlobGCJobTest, GCLimiter) {
     size_t RequestToken(size_t bytes, size_t alignment,
                         Env::IOPriority io_priority, Statistics* stats,
                         RateLimiter::OpType op_type) override {
-      if (IsRateLimited(op_type)) {
+      // Just the same condition with the rocksdb's RequestToken
+      if (io_priority < Env::IO_TOTAL && IsRateLimited(op_type)) {
         if (op_type == RateLimiter::OpType::kRead) {
           read = true;
         } else {

src/db_impl.cc

@@ -29,7 +29,8 @@ class TitanDBImpl::FileManager : public BlobFileManager {
  public:
   FileManager(TitanDBImpl* db) : db_(db) {}
 
-  Status NewFile(std::unique_ptr<BlobFileHandle>* handle) override {
+  Status NewFile(std::unique_ptr<BlobFileHandle>* handle,
+                 Env::IOPriority pri) override {
     auto number = db_->blob_file_set_->NewFileNumber();
     auto name = BlobFileName(db_->dirname_, number);
 
@@ -39,6 +40,8 @@ class TitanDBImpl::FileManager : public BlobFileManager {
       std::unique_ptr<WritableFile> f;
       s = db_->env_->NewWritableFile(name, &f, db_->env_options_);
       if (!s.ok()) return s;
+
+      f->SetIOPriority(pri);
       file.reset(new WritableFileWriter(std::move(f), name, db_->env_options_));
     }
 

src/table_builder.cc

@@ -146,7 +146,11 @@ void TitanTableBuilder::AddBlob(const ParsedInternalKey& ikey,
 
   // Init blob_builder_ first
   if (!blob_builder_) {
-    status_ = blob_manager_->NewFile(&blob_handle_);
+    // Set the Flush's blob file with a high_io pri  and the Compaction's
+    // blob file with a low_io pri in ratelimiter.
+    status_ = blob_manager_->NewFile(
+        &blob_handle_,
+        target_level_ > 0 ? Env::IOPriority::IO_LOW : Env::IOPriority::IO_HIGH);
     if (!ok()) return;
     ROCKS_LOG_INFO(db_options_.info_log,
                    "Titan table builder created new blob file %" PRIu64 ".",

src/table_builder_test.cc

@@ -24,7 +24,8 @@ class FileManager : public BlobFileManager {
         number_(kTestFileNumber),
         blob_file_set_(blob_file_set) {}
 
-  Status NewFile(std::unique_ptr<BlobFileHandle>* handle) override {
+  Status NewFile(std::unique_ptr<BlobFileHandle>* handle,
+                 Env::IOPriority pri = Env::IOPriority::IO_TOTAL) override {
     auto number = number_.fetch_add(1);
     auto name = BlobFileName(db_options_.dirname, number);
     std::unique_ptr<WritableFileWriter> file;
@@ -32,6 +33,7 @@ class FileManager : public BlobFileManager {
       std::unique_ptr<WritableFile> f;
       Status s = env_->NewWritableFile(name, &f, env_options_);
       if (!s.ok()) return s;
+      f->SetIOPriority(pri);
       file.reset(new WritableFileWriter(std::move(f), name, env_options_));
     }
     handle->reset(new FileHandle(number, name, std::move(file)));