Make add and remove pluggable compaction service thread-safe

Summary: We just need to add and remove under the lock, as well as make a copy of remote compaction service under the lock during creation of a CompactionJob

Test Plan: unit test still passes

Reviewers: dhruba, #platform

Reviewed By: dhruba, #platform

Subscribers: joe

Differential Revision: https://rockset.phacility.com/D5827

Author: Duc Hieu Pham

db/db_impl/db_impl.h

@@ -828,14 +828,19 @@ class DBImpl : public DB {
       bool sanitize) override;
 
   // This registered service will be called to do a remote compaction
-  virtual Status RegisterPluggableCompactionService(std::unique_ptr<PluggableCompactionService> rservice) override {
+  virtual Status RegisterPluggableCompactionService(
+      std::unique_ptr<PluggableCompactionService> rservice) override {
+    mutex_.Lock();
     remote_compaction_service_ = std::move(rservice);
+    mutex_.Unlock();
     return Status::OK();
   }
 
   // Clearoff any registered pluggable compaction service
   virtual void UnRegisterPluggableCompactionService() override {
-    remote_compaction_service_.reset(nullptr);
+    mutex_.Lock();
+    remote_compaction_service_ = nullptr;
+    mutex_.Unlock();
   }
 
   // Print information of all tombstones of all iterators to the std::string
@@ -1973,7 +1978,8 @@ class DBImpl : public DB {
   InstrumentedCondVar atomic_flush_install_cv_;
 
   // The pluggable compaction service, if registered.
-  std::unique_ptr<PluggableCompactionService> remote_compaction_service_;
+  // Protected by mutex_
+  std::shared_ptr<PluggableCompactionService> remote_compaction_service_;
 
   Status doCompact(const CompactionOptions& compact_options,
       ColumnFamilyData* cfd,

db/db_impl/db_impl_compaction_flush.cc

@@ -2750,15 +2750,15 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
     NotifyOnCompactionBegin(c->column_family_data(), c.get(), status,
                             compaction_job_stats, job_context->job_id);
 
+    // Make a copy of remote compaction service here, because it could be
+    // cleared once mutex_ is unlocked.
+    auto remote_compaction_service = remote_compaction_service_;
     mutex_.Unlock();
     TEST_SYNC_POINT_CALLBACK(
         "DBImpl::BackgroundCompaction:NonTrivial:BeforeRun", nullptr);
 
-    // This check is not thread safe. XXX
-    bool remote = (remote_compaction_service_ != nullptr);
-
-    if (remote) {
-      compaction_job.RunRemote(remote_compaction_service_.get());
+    if (remote_compaction_service) {
+      compaction_job.RunRemote(remote_compaction_service.get());
     } else {
       compaction_job.Run();
     }