Lock publisher before obtaining pointer to it

Summary:
This happened during an on-diff link test:
```
ThreadSanitizer:DEADLYSIGNAL
==3804651==ERROR: ThreadSanitizer: SEGV on unknown address 0x0000003a0deb (pc 0x56253d46afc2 bp 0x7f5fceda87e0 sp 0x7f5fceda87d0 T3804675)
==3804651==The signal is caused by a READ memory access.
    #0 facebook::fboss::fsdb::FsdbPubSubManager::publishState(facebook::fboss::fsdb::OperDelta&&) <null> (setup_qsfp_test_bin-credo-0.7.2+0x4b04fc2)
    #1 facebook::fboss::fsdb::FsdbComponentSyncer::publishDelta(facebook::fboss::fsdb::OperDelta&&, bool) <null> (setup_qsfp_test_bin-credo-0.7.2+0x4afb231)
    #2 facebook::fboss::fsdb::FsdbStateComponentSyncer<facebook::fboss::cfg::QsfpServiceConfig>::publisherStateChanged(facebook::fboss::fsdb::FsdbStreamClient::State, facebook::fboss::fsdb::FsdbStreamClient::State)::'lambda'()::operator()() const <null> (setup_qsfp_test_bin-credo-0.7.2+0x4af553e)
    #3 void folly::detail::function::FunctionTraits<void ()>::callSmall<facebook::fboss::fsdb::FsdbStateComponentSyncer<facebook::fboss::cfg::QsfpServiceConfig>::publisherStateChanged(facebook::fboss::fsdb::FsdbStreamClient::State, facebook::fboss::fsdb::FsdbStreamClient::State)::'lambda'()>(folly::detail::function::Data&) <null> (setup_qsfp_test_bin-credo-0.7.2+0x4af5389)
    #4 void folly::detail::function::FunctionTraits<void ()>::callBig<folly::EventBase::runInEventBaseThreadAndWait(folly::Function<void ()>)::$_4>(folly::detail::function::Data&) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1dda4a)
    #5 bool folly::AtomicNotificationQueue<folly::Function<void ()> >::drive<folly::EventBase::FuncRunner&>(folly::EventBase::FuncRunner&) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1dcd22)
    #6 folly::EventBaseAtomicNotificationQueue<folly::Function<void ()>, folly::EventBase::FuncRunner>::execute() <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1d722c)
    #7 non-virtual thunk to folly::EventBaseAtomicNotificationQueue<folly::Function<void ()>, folly::EventBase::FuncRunner>::handlerReady(unsigned short) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1dedfd)
    #8 folly::EventHandler::libeventCallback(int, short, void*) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1e287c)
    #9 event_base_loop <null> (setup_qsfp_test_bin-credo-0.7.2+0xcd43578)
    #10 (anonymous namespace)::EventBaseBackend::eb_event_base_loop(int) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1d95aa)
    #11 folly::EventBase::loopMain(int, bool) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1d6127)
    #12 folly::EventBase::loopForever() <null> (setup_qsfp_test_bin-credo-0.7.2+0xc1d9fa4)
    #13 folly::run(folly::EventBaseManager*, folly::EventBase*, folly::Baton<true, std::atomic>*, folly::Range<char const*> const&) <null> (setup_qsfp_test_bin-credo-0.7.2+0xc140edf)
    #14 std::thread::_State_impl<std::thread::_Invoker<std::tuple<void (*)(folly::EventBaseManager*, folly::EventBase*, folly::Baton<true, std::atomic>*, folly::Range<char const*> const&), folly::EventBaseManager*, folly::EventBase*, folly::Baton<true, std::atomic>*, folly::Range<char const*> > > >::_M_run() <null> (setup_qsfp_test_bin-credo-0.7.2+0xc14190b)
    #15 execute_native_thread_routine <null> (libstdc++.so.6+0xdf2e4)
    #16 __tsan_thread_start_func <null> (setup_qsfp_test_bin-credo-0.7.2+0xd17efae)
    #17 start_thread <null> (libc.so.6+0x979be)
    #18 clone <null> (libc.so.6+0x127b9f)
```

Publishers are currently protected by publisherMutex_. But here we have publishState()->publishImpl(). We lock in publishImpl(). But publishState() does this without lock:
```
  publishImpl(statePathPublisher_.get(), std::move(pubUnit));
```

If the other thread is calling createStateDeltaPublisher(), this pointer could be half written. We obtain this garbage pointer. Then obtain the lock while trying to use this garbage.

Also, removeStateDeltaPublisher() is completely unprotected. It seems to be used only in tests. But adding locks there for completeness.

Reviewed By: jasmeetbagga

Differential Revision: D38337566

fbshipit-source-id: 08c2f83bfaae649baa591549a53fa1a94d22e313

fboss/fsdb/client/FsdbPubSubManager.cpp

@@ -139,15 +139,19 @@ void FsdbPubSubManager::createStatPathPublisher(
 }
 
 void FsdbPubSubManager::removeStateDeltaPublisher() {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   stateDeltaPublisher_.reset();
 }
 void FsdbPubSubManager::removeStatePathPublisher() {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   statePathPublisher_.reset();
 }
 void FsdbPubSubManager::removeStatDeltaPublisher() {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   statDeltaPublisher_.reset();
 }
 void FsdbPubSubManager::removeStatPathPublisher() {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   statPathPublisher_.reset();
 }
 
@@ -156,23 +160,26 @@ void FsdbPubSubManager::publishImpl(PublisherT* publisher, PubUnitT&& pubUnit) {
   if (!publisher) {
     throw std::runtime_error("Publisher must be created before publishing");
   }
-  std::lock_guard<std::mutex> lk(publisherMutex_);
   publisher->write(std::forward<PubUnitT>(pubUnit));
 }
 
 void FsdbPubSubManager::publishState(OperDelta&& pubUnit) {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   publishImpl(stateDeltaPublisher_.get(), std::move(pubUnit));
 }
 
 void FsdbPubSubManager::publishState(OperState&& pubUnit) {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   publishImpl(statePathPublisher_.get(), std::move(pubUnit));
 }
 
 void FsdbPubSubManager::publishStat(OperDelta&& pubUnit) {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   publishImpl(statDeltaPublisher_.get(), std::move(pubUnit));
 }
 
 void FsdbPubSubManager::publishStat(OperState&& pubUnit) {
+  std::lock_guard<std::mutex> lk(publisherMutex_);
   publishImpl(statPathPublisher_.get(), std::move(pubUnit));
 }