Use FSRandomRWFilePtr Object to call underlying file system. (#7198)

Summary:
Replace FSRandomRWFile pointer with FSRandomRWFilePtr object in the rocksdb internal code.
This new object wraps FSRandomRWFile pointer.

Objective: If tracing is enabled, FSRandomRWFile object returns FSRandomRWFileTracingWrapper pointer that includes all necessary information in IORecord and calls underlying FileSystem and invokes IOTracer to dump that record in a binary file. If tracing is disabled then, underlying FileSystem pointer is returned directly.
FSRandomRWFilePtr wrapper class is added to bypass the FSRandomRWFileWrapper when
tracing is disabled.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/7198

Test Plan: make check -j64

Reviewed By: anand1976

Differential Revision: D23421116

Pulled By: akankshamahajan15

fbshipit-source-id: 8a5ba0e7d9c1ba34c3a6f29829b107c5f09ab6a3

db/external_sst_file_ingestion_job.cc

@@ -791,13 +791,14 @@ Status ExternalSstFileIngestionJob::AssignGlobalSeqnoForIngestedFile(
         fs_->NewRandomRWFile(file_to_ingest->internal_file_path, env_options_,
                              &rwfile, nullptr);
     if (status.ok()) {
+      FSRandomRWFilePtr fsptr(std::move(rwfile), io_tracer_);
       std::string seqno_val;
       PutFixed64(&seqno_val, seqno);
-      status = rwfile->Write(file_to_ingest->global_seqno_offset, seqno_val,
-                             IOOptions(), nullptr);
+      status = fsptr->Write(file_to_ingest->global_seqno_offset, seqno_val,
+                            IOOptions(), nullptr);
       if (status.ok()) {
         TEST_SYNC_POINT("ExternalSstFileIngestionJob::BeforeSyncGlobalSeqno");
-        status = SyncIngestedFile(rwfile.get());
+        status = SyncIngestedFile(fsptr.get());
         TEST_SYNC_POINT("ExternalSstFileIngestionJob::AfterSyncGlobalSeqno");
         if (!status.ok()) {
           ROCKS_LOG_WARN(db_options_.info_log,

env/file_system_tracer.h

@@ -337,26 +337,32 @@ class FSRandomRWFileTracingWrapper : public FSRandomRWFileWrapper {
 // FSRandomRWFileTracingWrapper when tracing is disabled.
 class FSRandomRWFilePtr {
  public:
-  FSRandomRWFilePtr(FSRandomRWFile* fs, std::shared_ptr<IOTracer> io_tracer)
-      : fs_(fs),
+  FSRandomRWFilePtr(std::unique_ptr<FSRandomRWFile>&& fs,
+                    std::shared_ptr<IOTracer> io_tracer)
+      : fs_(std::move(fs)),
         io_tracer_(io_tracer),
-        fs_tracer_(new FSRandomRWFileTracingWrapper(fs_, io_tracer_)) {}
-
-  explicit FSRandomRWFilePtr(FSRandomRWFile* fs)
-      : fs_(fs), io_tracer_(nullptr), fs_tracer_(nullptr) {}
+        fs_tracer_(fs_.get(), io_tracer_) {}
 
   FSRandomRWFile* operator->() const {
     if (io_tracer_ && io_tracer_->is_tracing_enabled()) {
-      return fs_tracer_;
+      return const_cast<FSRandomRWFileTracingWrapper*>(&fs_tracer_);
     } else {
-      return fs_;
+      return fs_.get();
+    }
+  }
+
+  FSRandomRWFile* get() const {
+    if (io_tracer_ && io_tracer_->is_tracing_enabled()) {
+      return const_cast<FSRandomRWFileTracingWrapper*>(&fs_tracer_);
+    } else {
+      return fs_.get();
     }
   }
 
  private:
-  FSRandomRWFile* fs_;
+  std::unique_ptr<FSRandomRWFile> fs_;
   std::shared_ptr<IOTracer> io_tracer_;
-  FSRandomRWFileTracingWrapper* fs_tracer_;
+  FSRandomRWFileTracingWrapper fs_tracer_;
 };
 
 }  // namespace ROCKSDB_NAMESPACE