Introduce a new storage specific Env API (#5761)

Summary:
The current Env API encompasses both storage/file operations, as well as OS related operations. Most of the APIs return a Status, which does not have enough metadata about an error, such as whether its retry-able or not, scope (i.e fault domain) of the error etc., that may be required in order to properly handle a storage error. The file APIs also do not provide enough control over the IO SLA, such as timeout, prioritization, hinting about placement and redundancy etc.

This PR separates out the file/storage APIs from Env into a new FileSystem class. The APIs are updated to return an IOStatus with metadata about the error, as well as to take an IOOptions structure as input in order to allow more control over the IO.

The user can set both ```options.env``` and ```options.file_system``` to specify that RocksDB should use the former for OS related operations and the latter for storage operations. Internally, a ```CompositeEnvWrapper``` has been introduced that inherits from ```Env``` and redirects individual methods to either an ```Env``` implementation or the ```FileSystem``` as appropriate. When options are sanitized during ```DB::Open```, ```options.env``` is replaced with a newly allocated ```CompositeEnvWrapper``` instance if both env and file_system have been specified. This way, the rest of the RocksDB code can continue to function as before.

This PR also ports PosixEnv to the new API by splitting it into two - PosixEnv and PosixFileSystem. PosixEnv is defined as a sub-class of CompositeEnvWrapper, and threading/time functions are overridden with Posix specific implementations in order to avoid an extra level of indirection.

The ```CompositeEnvWrapper``` translates ```IOStatus``` return code to ```Status```, and sets the severity to ```kSoftError``` if the io_status is retryable. The error handling code in RocksDB can then recover the DB automatically.
Pull Request resolved: facebook/rocksdb#5761

Differential Revision: D18868376

Pulled By: anand1976

fbshipit-source-id: 39efe18a162ea746fabac6360ff529baba48486f
Signed-off-by: Changlong Chen <levisonchen@live.cn>

file/delete_scheduler.cc

@@ -19,12 +19,13 @@
 
 namespace rocksdb {
 
-DeleteScheduler::DeleteScheduler(Env* env, int64_t rate_bytes_per_sec,
-                                 Logger* info_log,
+DeleteScheduler::DeleteScheduler(Env* env, FileSystem* fs,
+                                 int64_t rate_bytes_per_sec, Logger* info_log,
                                  SstFileManagerImpl* sst_file_manager,
                                  double max_trash_db_ratio,
                                  uint64_t bytes_max_delete_chunk)
     : env_(env),
+      fs_(fs),
       total_trash_size_(0),
       rate_bytes_per_sec_(rate_bytes_per_sec),
       pending_files_(0),
@@ -61,7 +62,7 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
     // Rate limiting is disabled or trash size makes up more than
     // max_trash_db_ratio_ (default 25%) of the total DB size
     TEST_SYNC_POINT("DeleteScheduler::DeleteFile");
-    s = env_->DeleteFile(file_path);
+    s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
     if (s.ok()) {
       sst_file_manager_->OnDeleteFile(file_path);
     }
@@ -74,7 +75,7 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
 
   if (!s.ok()) {
     ROCKS_LOG_ERROR(info_log_, "Failed to mark %s as trash", file_path.c_str());
-    s = env_->DeleteFile(file_path);
+    s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
     if (s.ok()) {
       sst_file_manager_->OnDeleteFile(file_path);
     }
@@ -83,7 +84,7 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
 
   // Update the total trash size
   uint64_t trash_file_size = 0;
-  env_->GetFileSize(trash_file, &trash_file_size);
+  fs_->GetFileSize(trash_file, IOOptions(), &trash_file_size, nullptr);
   total_trash_size_.fetch_add(trash_file_size);
 
   // Add file to delete queue
@@ -165,10 +166,10 @@ Status DeleteScheduler::MarkAsTrash(const std::string& file_path,
   int cnt = 0;
   InstrumentedMutexLock l(&file_move_mu_);
   while (true) {
-    s = env_->FileExists(*trash_file);
+    s = fs_->FileExists(*trash_file, IOOptions(), nullptr);
     if (s.IsNotFound()) {
       // We found a path for our file in trash
-      s = env_->RenameFile(file_path, *trash_file);
+      s = fs_->RenameFile(file_path, *trash_file, IOOptions(), nullptr);
       break;
     } else if (s.ok()) {
       // Name conflict, generate new random suffix
@@ -262,7 +263,7 @@ Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
                                         uint64_t* deleted_bytes,
                                         bool* is_complete) {
   uint64_t file_size;
-  Status s = env_->GetFileSize(path_in_trash, &file_size);
+  Status s = fs_->GetFileSize(path_in_trash, IOOptions(), &file_size, nullptr);
   *is_complete = true;
   TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:DeleteFile");
   if (s.ok()) {
@@ -273,17 +274,19 @@ Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
       // file after the number of file link check and ftruncte because
       // the file is now in trash and no hardlink is supposed to create
       // to trash files by RocksDB.
-      Status my_status = env_->NumFileLinks(path_in_trash, &num_hard_links);
+      Status my_status = fs_->NumFileLinks(path_in_trash, IOOptions(),
+                                           &num_hard_links, nullptr);
       if (my_status.ok()) {
         if (num_hard_links == 1) {
-          std::unique_ptr<WritableFile> wf;
-          my_status =
-              env_->ReopenWritableFile(path_in_trash, &wf, EnvOptions());
+          std::unique_ptr<FSWritableFile> wf;
+          my_status = fs_->ReopenWritableFile(path_in_trash, FileOptions(),
+                                              &wf, nullptr);
           if (my_status.ok()) {
-            my_status = wf->Truncate(file_size - bytes_max_delete_chunk_);
+            my_status = wf->Truncate(file_size - bytes_max_delete_chunk_,
+                                     IOOptions(), nullptr);
             if (my_status.ok()) {
               TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:Fsync");
-              my_status = wf->Fsync();
+              my_status = wf->Fsync(IOOptions(), nullptr);
             }
           }
           if (my_status.ok()) {
@@ -312,14 +315,14 @@ Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
     }
 
     if (need_full_delete) {
-      s = env_->DeleteFile(path_in_trash);
+      s = fs_->DeleteFile(path_in_trash, IOOptions(), nullptr);
       if (!dir_to_sync.empty()) {
-        std::unique_ptr<Directory> dir_obj;
+        std::unique_ptr<FSDirectory> dir_obj;
         if (s.ok()) {
-          s = env_->NewDirectory(dir_to_sync, &dir_obj);
+          s = fs_->NewDirectory(dir_to_sync, IOOptions(), &dir_obj, nullptr);
         }
         if (s.ok()) {
-          s = dir_obj->Fsync();
+          s = dir_obj->Fsync(IOOptions(), nullptr);
           TEST_SYNC_POINT_CALLBACK(
               "DeleteScheduler::DeleteTrashFile::AfterSyncDir",
               reinterpret_cast<void*>(const_cast<std::string*>(&dir_to_sync)));

file/delete_scheduler.h

@@ -15,6 +15,7 @@
 #include "monitoring/instrumented_mutex.h"
 #include "port/port.h"
 
+#include "rocksdb/file_system.h"
 #include "rocksdb/status.h"
 
 namespace rocksdb {
@@ -32,8 +33,8 @@ class SstFileManagerImpl;
 // case DeleteScheduler will delete files immediately.
 class DeleteScheduler {
  public:
-  DeleteScheduler(Env* env, int64_t rate_bytes_per_sec, Logger* info_log,
-                  SstFileManagerImpl* sst_file_manager,
+  DeleteScheduler(Env* env, FileSystem* fs, int64_t rate_bytes_per_sec,
+                  Logger* info_log, SstFileManagerImpl* sst_file_manager,
                   double max_trash_db_ratio, uint64_t bytes_max_delete_chunk);
 
   ~DeleteScheduler();
@@ -91,6 +92,8 @@ class DeleteScheduler {
   void BackgroundEmptyTrash();
 
   Env* env_;
+  FileSystem* fs_;
+
   // total size of trash files
   std::atomic<uint64_t> total_trash_size_;
   // Maximum number of bytes that should be deleted per second

file/delete_scheduler_test.cc

@@ -93,8 +93,10 @@ class DeleteSchedulerTest : public testing::Test {
     // Tests in this file are for DeleteScheduler component and dont create any
     // DBs, so we need to set max_trash_db_ratio to 100% (instead of default
     // 25%)
+    std::shared_ptr<FileSystem>
+                fs(std::make_shared<LegacyFileSystemWrapper>(env_));
     sst_file_mgr_.reset(
-        new SstFileManagerImpl(env_, nullptr, rate_bytes_per_sec_,
+        new SstFileManagerImpl(env_, fs, nullptr, rate_bytes_per_sec_,
                                /* max_trash_db_ratio= */ 1.1, 128 * 1024));
     delete_scheduler_ = sst_file_mgr_->delete_scheduler();
   }

file/file_util.cc

@@ -17,28 +17,28 @@
 namespace rocksdb {
 
 // Utility function to copy a file up to a specified length
-Status CopyFile(Env* env, const std::string& source,
+Status CopyFile(FileSystem* fs, const std::string& source,
                 const std::string& destination, uint64_t size, bool use_fsync) {
-  const EnvOptions soptions;
+  const FileOptions soptions;
   Status s;
   std::unique_ptr<SequentialFileReader> src_reader;
   std::unique_ptr<WritableFileWriter> dest_writer;
 
   {
-    std::unique_ptr<SequentialFile> srcfile;
-    s = env->NewSequentialFile(source, &srcfile, soptions);
+    std::unique_ptr<FSSequentialFile> srcfile;
+    s = fs->NewSequentialFile(source, soptions, &srcfile, nullptr);
     if (!s.ok()) {
       return s;
     }
-    std::unique_ptr<WritableFile> destfile;
-    s = env->NewWritableFile(destination, &destfile, soptions);
+    std::unique_ptr<FSWritableFile> destfile;
+    s = fs->NewWritableFile(destination, soptions, &destfile, nullptr);
     if (!s.ok()) {
       return s;
     }
 
     if (size == 0) {
       // default argument means copy everything
-      s = env->GetFileSize(source, &size);
+      s = fs->GetFileSize(source, IOOptions(), &size, nullptr);
       if (!s.ok()) {
         return s;
       }
@@ -69,14 +69,14 @@ Status CopyFile(Env* env, const std::string& source,
 }
 
 // Utility function to create a file with the provided contents
-Status CreateFile(Env* env, const std::string& destination,
+Status CreateFile(FileSystem* fs, const std::string& destination,
                   const std::string& contents, bool use_fsync) {
   const EnvOptions soptions;
   Status s;
   std::unique_ptr<WritableFileWriter> dest_writer;
 
-  std::unique_ptr<WritableFile> destfile;
-  s = env->NewWritableFile(destination, &destfile, soptions);
+  std::unique_ptr<FSWritableFile> destfile;
+  s = fs->NewWritableFile(destination, soptions, &destfile, nullptr);
   if (!s.ok()) {
     return s;
   }

file/file_util.h

@@ -9,17 +9,18 @@
 #include "file/filename.h"
 #include "options/db_options.h"
 #include "rocksdb/env.h"
+#include "rocksdb/file_system.h"
 #include "rocksdb/status.h"
 #include "rocksdb/types.h"
 
 namespace rocksdb {
 // use_fsync maps to options.use_fsync, which determines the way that
 // the file is synced after copying.
-extern Status CopyFile(Env* env, const std::string& source,
+extern Status CopyFile(FileSystem* fs, const std::string& source,
                        const std::string& destination, uint64_t size,
                        bool use_fsync);
 
-extern Status CreateFile(Env* env, const std::string& destination,
+extern Status CreateFile(FileSystem* fs, const std::string& destination,
                          const std::string& contents, bool use_fsync);
 
 extern Status DeleteDBFile(const ImmutableDBOptions* db_options,

file/random_access_file_reader.cc

@@ -61,8 +61,8 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
         }
         {
           IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, env_);
-          s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, &tmp,
-                          buf.Destination());
+          s = file_->Read(aligned_offset + buf.CurrentSize(), allowed,
+                          IOOptions(), &tmp, buf.Destination(), nullptr);
         }
         if (ShouldNotifyListeners()) {
           auto finish_ts = std::chrono::system_clock::now();
@@ -110,7 +110,8 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
 #endif
         {
           IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, env_);
-          s = file_->Read(offset + pos, allowed, &tmp_result, scratch + pos);
+          s = file_->Read(offset + pos, allowed, IOOptions(), &tmp_result,
+                          scratch + pos, nullptr);
         }
 #ifndef ROCKSDB_LITE
         if (ShouldNotifyListeners()) {
@@ -145,7 +146,7 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
   return s;
 }
 
-Status RandomAccessFileReader::MultiRead(ReadRequest* read_reqs,
+Status RandomAccessFileReader::MultiRead(FSReadRequest* read_reqs,
                                          size_t num_reqs) const {
   Status s;
   uint64_t elapsed = 0;
@@ -165,7 +166,7 @@ Status RandomAccessFileReader::MultiRead(ReadRequest* read_reqs,
 #endif  // ROCKSDB_LITE
     {
       IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, env_);
-      s = file_->MultiRead(read_reqs, num_reqs);
+      s = file_->MultiRead(read_reqs, num_reqs, IOOptions(), nullptr);
     }
     for (size_t i = 0; i < num_reqs; ++i) {
 #ifndef ROCKSDB_LITE

file/random_access_file_reader.h

@@ -13,12 +13,12 @@
 #include <string>
 #include "port/port.h"
 #include "rocksdb/env.h"
+#include "rocksdb/file_system.h"
 #include "rocksdb/listener.h"
 #include "rocksdb/rate_limiter.h"
 #include "util/aligned_buffer.h"
 
 namespace rocksdb {
-
 class Statistics;
 class HistogramImpl;
 
@@ -48,7 +48,7 @@ class RandomAccessFileReader {
 
   bool ShouldNotifyListeners() const { return !listeners_.empty(); }
 
-  std::unique_ptr<RandomAccessFile> file_;
+  std::unique_ptr<FSRandomAccessFile> file_;
   std::string file_name_;
   Env* env_;
   Statistics* stats_;
@@ -59,7 +59,7 @@ class RandomAccessFileReader {
 
  public:
   explicit RandomAccessFileReader(
-      std::unique_ptr<RandomAccessFile>&& raf, std::string _file_name,
+      std::unique_ptr<FSRandomAccessFile>&& raf, std::string _file_name,
       Env* env = nullptr, Statistics* stats = nullptr, uint32_t hist_type = 0,
       HistogramImpl* file_read_hist = nullptr,
       RateLimiter* rate_limiter = nullptr,
@@ -105,13 +105,13 @@ class RandomAccessFileReader {
   Status Read(uint64_t offset, size_t n, Slice* result, char* scratch,
               bool for_compaction = false) const;
 
-  Status MultiRead(ReadRequest* reqs, size_t num_reqs) const;
+  Status MultiRead(FSReadRequest* reqs, size_t num_reqs) const;
 
   Status Prefetch(uint64_t offset, size_t n) const {
-    return file_->Prefetch(offset, n);
+    return file_->Prefetch(offset, n, IOOptions(), nullptr);
   }
 
-  RandomAccessFile* file() { return file_.get(); }
+  FSRandomAccessFile* file() { return file_.get(); }
 
   std::string file_name() const { return file_name_; }
 

file/read_write_util.cc

@@ -13,15 +13,16 @@
 #include "test_util/sync_point.h"
 
 namespace rocksdb {
-Status NewWritableFile(Env* env, const std::string& fname,
-                       std::unique_ptr<WritableFile>* result,
-                       const EnvOptions& options) {
-  Status s = env->NewWritableFile(fname, result, options);
+
+IOStatus NewWritableFile(FileSystem* fs, const std::string& fname,
+                         std::unique_ptr<FSWritableFile>* result,
+                         const FileOptions& options) {
+  IOStatus s = fs->NewWritableFile(fname, options, result, nullptr);
   TEST_KILL_RANDOM("NewWritableFile:0", rocksdb_kill_odds * REDUCE_ODDS2);
   return s;
 }
 
-bool ReadOneLine(std::istringstream* iss, SequentialFile* seq_file,
+bool ReadOneLine(std::istringstream* iss, FSSequentialFile* seq_file,
                  std::string* output, bool* has_data, Status* result) {
   const int kBufferSize = 8192;
   char buffer[kBufferSize + 1];
@@ -39,7 +40,8 @@ bool ReadOneLine(std::istringstream* iss, SequentialFile* seq_file,
       // if we're not sure whether we have a complete line,
       // further read from the file.
       if (*has_data) {
-        *result = seq_file->Read(kBufferSize, &input_slice, buffer);
+        *result = seq_file->Read(kBufferSize, IOOptions(),
+                                 &input_slice, buffer, nullptr);
       }
       if (input_slice.size() == 0) {
         // meaning we have read all the data

file/read_write_util.h

@@ -10,6 +10,7 @@
 #pragma once
 #include <atomic>
 #include "rocksdb/env.h"
+#include "rocksdb/file_system.h"
 
 namespace rocksdb {
 // Returns a WritableFile.
@@ -18,12 +19,12 @@ namespace rocksdb {
 // fname   : the file name.
 // result  : output arg. A WritableFile based on `fname` returned.
 // options : the Env Options.
-extern Status NewWritableFile(Env* env, const std::string& fname,
-                              std::unique_ptr<WritableFile>* result,
-                              const EnvOptions& options);
+extern IOStatus NewWritableFile(FileSystem* fs, const std::string& fname,
+                                std::unique_ptr<FSWritableFile>* result,
+                                const FileOptions& options);
 
 // Read a single line from a file.
-bool ReadOneLine(std::istringstream* iss, SequentialFile* seq_file,
+bool ReadOneLine(std::istringstream* iss, FSSequentialFile* seq_file,
                  std::string* output, bool* has_data, Status* result);
 
 #ifndef NDEBUG