Improve OutputBufferManager initialization

Starting from C++11, the C++ standard guarantees that the initialization
of function-local static variables is thread-safe. This is better than
using a global mutex, especially for subsequent getInstance() calls. This
is because the overhead of using a static local variable only needs to do
a simple check to see if the variable has already been initialized, while
for the global mutex case, all getInstance() calls need to aquire this
lock exclusively.

velox/exec/Exchange.h

@@ -54,7 +54,7 @@ class Exchange : public SourceOperator {
         processSplits_{operatorCtx_->driverCtx()->driverId == 0},
         exchangeClient_{std::move(exchangeClient)} {
     options_.compressionKind =
-        OutputBufferManager::getInstance().lock()->compressionKind();
+        OutputBufferManager::getInstance()->compressionKind();
   }
 
   ~Exchange() override {

velox/exec/OutputBufferManager.cpp

@@ -17,21 +17,12 @@
 #include "velox/exec/Task.h"
 
 namespace facebook::velox::exec {
-// static
-void OutputBufferManager::initialize(const Options& options) {
-  std::lock_guard<std::mutex> l(initMutex_);
-  VELOX_CHECK(
-      instance_ == nullptr, "May initialize OutputBufferManager only once");
-  instance_ = std::make_shared<OutputBufferManager>(options);
-}
 
 // static
-std::weak_ptr<OutputBufferManager> OutputBufferManager::getInstance() {
-  std::lock_guard<std::mutex> l(initMutex_);
-  if (!instance_) {
-    instance_ = std::make_shared<OutputBufferManager>(Options());
-  }
-  return instance_;
+const std::shared_ptr<OutputBufferManager>& OutputBufferManager::getInstance() {
+  static const std::shared_ptr<OutputBufferManager> instance =
+      std::make_shared<OutputBufferManager>(Options());
+  return instance;
 }
 
 std::shared_ptr<OutputBuffer> OutputBufferManager::getBuffer(

velox/exec/OutputBufferManager.h

@@ -98,11 +98,7 @@ class OutputBufferManager {
 
   void removeTask(const std::string& taskId);
 
-  /// Initializes singleton with 'options'. May be called once before
-  /// getInstance().
-  static void initialize(const Options& options);
-
-  static std::weak_ptr<OutputBufferManager> getInstance();
+  static const std::shared_ptr<OutputBufferManager>& getInstance();
 
   uint64_t numBuffers() const;
 
@@ -157,8 +153,5 @@ class OutputBufferManager {
 
   std::function<std::unique_ptr<OutputStreamListener>()> listenerFactory_{
       nullptr};
-
-  inline static std::shared_ptr<OutputBufferManager> instance_;
-  inline static std::mutex initMutex_;
 };
 } // namespace facebook::velox::exec

velox/exec/PartitionedOutput.cpp

@@ -56,7 +56,7 @@ BlockingReason Destination::advance(
     auto rowType = asRowType(output->type());
     serializer::presto::PrestoVectorSerde::PrestoOptions options;
     options.compressionKind =
-        OutputBufferManager::getInstance().lock()->compressionKind();
+        OutputBufferManager::getInstance()->compressionKind();
     options.minCompressionRatio = PartitionedOutput::minCompressionRatio();
     current_->createStreamTree(rowType, rowsInCurrent_, &options);
   }

velox/exec/tests/ExchangeClientTest.cpp

@@ -47,7 +47,7 @@ class ExchangeClientTest : public testing::Test,
     if (!isRegisteredVectorSerde()) {
       velox::serializer::presto::PrestoVectorSerde::registerVectorSerde();
     }
-    bufferManager_ = OutputBufferManager::getInstance().lock();
+    bufferManager_ = OutputBufferManager::getInstance();
 
     common::testutil::TestValue::enable();
   }

velox/exec/tests/GroupedExecutionTest.cpp

@@ -287,7 +287,7 @@ TEST_F(GroupedExecutionTest, groupedExecutionWithOutputBuffer) {
 
   // 'Delete results' from output buffer triggers 'set all output consumed',
   // which should finish the task.
-  auto outputBufferManager = exec::OutputBufferManager::getInstance().lock();
+  auto outputBufferManager = exec::OutputBufferManager::getInstance();
   outputBufferManager->deleteResults(task->taskId(), 0);
 
   // Task must be finished at this stage.
@@ -471,7 +471,7 @@ DEBUG_ONLY_TEST_F(
 
     // 'Delete results' from output buffer triggers 'set all output consumed',
     // which should finish the task.
-    auto outputBufferManager = exec::OutputBufferManager::getInstance().lock();
+    auto outputBufferManager = exec::OutputBufferManager::getInstance();
     outputBufferManager->deleteResults(task->taskId(), 0);
 
     // Task must be finished at this stage.
@@ -627,7 +627,7 @@ TEST_F(GroupedExecutionTest, groupedExecutionWithHashAndNestedLoopJoin) {
 
     // 'Delete results' from output buffer triggers 'set all output consumed',
     // which should finish the task.
-    auto outputBufferManager = exec::OutputBufferManager::getInstance().lock();
+    auto outputBufferManager = exec::OutputBufferManager::getInstance();
     outputBufferManager->deleteResults(task->taskId(), 0);
 
     // Task must be finished at this stage.

velox/exec/tests/LimitTest.cpp

@@ -121,7 +121,7 @@ TEST_F(LimitTest, partialLimitEagerFlush) {
     params.planNode = builder.partitionedOutput({}, 1).planNode();
     auto cursor = TaskCursor::create(params);
     ASSERT_FALSE(cursor->moveNext());
-    auto bufferManager = exec::OutputBufferManager::getInstance().lock();
+    auto bufferManager = exec::OutputBufferManager::getInstance();
     auto [numPagesPromise, numPagesFuture] = folly::makePromiseContract<int>();
     ASSERT_TRUE(bufferManager->getData(
         cursor->task()->taskId(),

velox/exec/tests/MultiFragmentTest.cpp

@@ -197,7 +197,7 @@ class MultiFragmentTest : public HiveConnectorTestBase {
   std::vector<std::shared_ptr<TempFilePath>> filePaths_;
   std::vector<RowVectorPtr> vectors_;
   std::shared_ptr<OutputBufferManager> bufferManager_{
-      OutputBufferManager::getInstance().lock()};
+      OutputBufferManager::getInstance()};
 };
 
 TEST_F(MultiFragmentTest, aggregationSingleKey) {
@@ -1940,7 +1940,7 @@ class DataFetcher {
   int64_t totalBytes_{0};
 
   std::shared_ptr<OutputBufferManager> bufferManager_{
-      OutputBufferManager::getInstance().lock()};
+      OutputBufferManager::getInstance()};
 };
 
 /// Verify that POBM::getData() honors maxBytes parameter roughly at 1MB

velox/exec/tests/OutputBufferManagerTest.cpp

@@ -43,7 +43,7 @@ class OutputBufferManagerTest : public testing::Test {
 
   void SetUp() override {
     pool_ = facebook::velox::memory::memoryManager()->addLeafPool();
-    bufferManager_ = OutputBufferManager::getInstance().lock();
+    bufferManager_ = OutputBufferManager::getInstance();
     if (!isRegisteredVectorSerde()) {
       facebook::velox::serializer::presto::PrestoVectorSerde::
           registerVectorSerde();

velox/exec/tests/PartitionedOutputTest.cpp

@@ -78,7 +78,7 @@ class PartitionedOutputTest : public OperatorTestBase {
 
  private:
   const std::shared_ptr<OutputBufferManager> bufferManager_{
-      OutputBufferManager::getInstance().lock()};
+      OutputBufferManager::getInstance()};
 };
 
 TEST_F(PartitionedOutputTest, flush) {

velox/exec/tests/TaskTest.cpp

@@ -1044,7 +1044,7 @@ TEST_F(TaskTest, updateBroadCastOutputBuffers) {
                   .project({"c0 % 10"})
                   .partitionedOutputBroadcast({})
                   .planFragment();
-  auto bufferManager = OutputBufferManager::getInstance().lock();
+  auto bufferManager = OutputBufferManager::getInstance();
   {
     auto task = Task::create(
         "t0",

velox/exec/tests/utils/LocalExchangeSource.cpp

@@ -52,7 +52,7 @@ class LocalExchangeSource : public exec::ExchangeSource {
 
     promise_ = std::move(promise);
 
-    auto buffers = OutputBufferManager::getInstance().lock();
+    auto buffers = OutputBufferManager::getInstance();
     VELOX_CHECK_NOT_NULL(buffers, "invalid OutputBufferManager");
     VELOX_CHECK(requestPending_);
     auto requestedSequence = sequence_;
@@ -164,7 +164,7 @@ class LocalExchangeSource : public exec::ExchangeSource {
   void pause() override {
     common::testutil::TestValue::adjust(
         "facebook::velox::exec::test::LocalExchangeSource::pause", nullptr);
-    auto buffers = OutputBufferManager::getInstance().lock();
+    auto buffers = OutputBufferManager::getInstance();
     VELOX_CHECK_NOT_NULL(buffers, "invalid OutputBufferManager");
     int64_t ackSequence;
     {
@@ -177,7 +177,7 @@ class LocalExchangeSource : public exec::ExchangeSource {
   void close() override {
     checkSetRequestPromise();
 
-    auto buffers = OutputBufferManager::getInstance().lock();
+    auto buffers = OutputBufferManager::getInstance();
     buffers->deleteResults(taskId_, destination_);
   }
 

velox/tool/trace/PartitionedOutputReplayer.h

@@ -59,7 +59,7 @@ class PartitionedOutputReplayer final : public OperatorReplayerBase {
 
   const core::PartitionedOutputNode* const originalNode_;
   const std::shared_ptr<exec::OutputBufferManager> bufferManager_{
-      exec::OutputBufferManager::getInstance().lock()};
+      exec::OutputBufferManager::getInstance()};
   const std::unique_ptr<folly::Executor> executor_{
       std::make_unique<folly::CPUThreadPoolExecutor>(
           std::thread::hardware_concurrency(),

velox/tool/trace/tests/PartitionedOutputReplayerTest.cpp

@@ -111,7 +111,7 @@ class PartitionedOutputReplayerTest : public HiveConnectorTestBase {
   }
 
   const std::shared_ptr<OutputBufferManager> bufferManager_{
-      exec::OutputBufferManager::getInstance().lock()};
+      exec::OutputBufferManager::getInstance()};
 };
 
 TEST_F(PartitionedOutputReplayerTest, defaultConsumer) {