Deprecate transfer capacity for shared arbitrator (#11084)

Summary: Pull Request resolved: #11084

Reviewed By: xiaoxmeng

Differential Revision: D63472474

Pulled By: tanjialiang

fbshipit-source-id: 47041957eb41069f1435fa035a4f63bd1134c62c

velox/common/memory/SharedArbitrator.cpp

@@ -109,16 +109,6 @@ uint64_t SharedArbitrator::ExtraConfig::getMemoryPoolReservedCapacity(
       config::CapacityUnit::BYTE);
 }
 
-uint64_t SharedArbitrator::ExtraConfig::getMemoryPoolTransferCapacity(
-    const std::unordered_map<std::string, std::string>& configs) {
-  return config::toCapacity(
-      getConfig<std::string>(
-          configs,
-          kMemoryPoolTransferCapacity,
-          std::string(kDefaultMemoryPoolTransferCapacity)),
-      config::CapacityUnit::BYTE);
-}
-
 uint64_t SharedArbitrator::ExtraConfig::getMemoryReclaimMaxWaitTimeMs(
     const std::unordered_map<std::string, std::string>& configs) {
   return std::chrono::duration_cast<std::chrono::milliseconds>(
@@ -182,8 +172,6 @@ SharedArbitrator::SharedArbitrator(const Config& config)
           ExtraConfig::getMemoryPoolInitialCapacity(config.extraConfigs)),
       memoryPoolReservedCapacity_(
           ExtraConfig::getMemoryPoolReservedCapacity(config.extraConfigs)),
-      memoryPoolTransferCapacity_(
-          ExtraConfig::getMemoryPoolTransferCapacity(config.extraConfigs)),
       memoryReclaimWaitMs_(
           ExtraConfig::getMemoryReclaimMaxWaitTimeMs(config.extraConfigs)),
       globalArbitrationEnabled_(
@@ -542,7 +530,7 @@ uint64_t SharedArbitrator::getCapacityGrowthTarget(
     const MemoryPool& pool,
     uint64_t requestBytes) const {
   if (fastExponentialGrowthCapacityLimit_ == 0 && slowCapacityGrowPct_ == 0) {
-    return std::max(requestBytes, memoryPoolTransferCapacity_);
+    return requestBytes;
   }
   uint64_t targetBytes{0};
   const auto capacity = pool.capacity();
@@ -551,8 +539,7 @@ uint64_t SharedArbitrator::getCapacityGrowthTarget(
   } else {
     targetBytes = capacity * slowCapacityGrowPct_;
   }
-  return std::max(
-      std::max(requestBytes, targetBytes), memoryPoolTransferCapacity_);
+  return std::max(requestBytes, targetBytes);
 }
 
 bool SharedArbitrator::growCapacity(MemoryPool* pool, uint64_t requestBytes) {
@@ -938,9 +925,8 @@ uint64_t SharedArbitrator::reclaim(
     MemoryPool* pool,
     uint64_t targetBytes,
     bool isLocalArbitration) noexcept {
-  int64_t bytesToReclaim = std::min<uint64_t>(
-      std::max(targetBytes, memoryPoolTransferCapacity_),
-      maxReclaimableCapacity(*pool, true));
+  int64_t bytesToReclaim =
+      std::min<uint64_t>(targetBytes, maxReclaimableCapacity(*pool, true));
   if (bytesToReclaim == 0) {
     return 0;
   }

velox/common/memory/SharedArbitrator.h

@@ -61,15 +61,6 @@ class SharedArbitrator : public memory::MemoryArbitrator {
     static uint64_t getMemoryPoolReservedCapacity(
         const std::unordered_map<std::string, std::string>& configs);
 
-    /// The minimal memory capacity to transfer out of or into a memory pool
-    /// during the memory arbitration.
-    static constexpr std::string_view kMemoryPoolTransferCapacity{
-        "memory-pool-transfer-capacity"};
-    static constexpr std::string_view kDefaultMemoryPoolTransferCapacity{
-        "128MB"};
-    static uint64_t getMemoryPoolTransferCapacity(
-        const std::unordered_map<std::string, std::string>& configs);
-
     /// Specifies the max time to wait for memory reclaim by arbitration. The
     /// memory reclaim might fail if the max time has exceeded. This prevents
     /// the memory arbitration from getting stuck when the memory reclaim waits
@@ -231,11 +222,7 @@ class SharedArbitrator : public memory::MemoryArbitrator {
     // The adjusted grow bytes based on 'requestBytes'. This 'targetBytes' is a
     // best effort target, and hence will not be guaranteed. The adjustment is
     // based on 'SharedArbitrator::fastExponentialGrowthCapacityLimit_'
-    // 'SharedArbitrator::slowCapacityGrowPct_' and
-    // 'MemoryArbitrator::memoryPoolTransferCapacity_'.
-    //
-    // TODO: deprecate 'MemoryArbitrator::memoryPoolTransferCapacity_' once
-    // exponential growth works well in production.
+    // 'SharedArbitrator::slowCapacityGrowPct_'
     const std::optional<uint64_t> targetBytes;
 
     // The start time of this arbitration operation.
@@ -516,7 +503,6 @@ class SharedArbitrator : public memory::MemoryArbitrator {
   const uint64_t reservedCapacity_;
   const uint64_t memoryPoolInitialCapacity_;
   const uint64_t memoryPoolReservedCapacity_;
-  const uint64_t memoryPoolTransferCapacity_;
   const uint64_t memoryReclaimWaitMs_;
   const bool globalArbitrationEnabled_;
   const bool checkUsageLeak_;

velox/common/memory/tests/MemoryArbitratorTest.cpp

@@ -144,6 +144,7 @@ TEST_F(MemoryArbitrationTest, queryMemoryCapacity) {
     ASSERT_FALSE(manager.arbitrator()->growCapacity(rootPool.get(), 6 << 20));
     ASSERT_EQ(rootPool->capacity(), 1 << 20);
     ASSERT_TRUE(manager.arbitrator()->growCapacity(rootPool.get(), 2 << 20));
+    ASSERT_TRUE(manager.arbitrator()->growCapacity(rootPool.get(), 1 << 20));
     ASSERT_EQ(rootPool->capacity(), 4 << 20);
     ASSERT_EQ(manager.arbitrator()->stats().freeCapacityBytes, 2 << 20);
     ASSERT_EQ(manager.arbitrator()->stats().freeReservedCapacityBytes, 2 << 20);
@@ -154,19 +155,19 @@ TEST_F(MemoryArbitrationTest, queryMemoryCapacity) {
         "Exceeded memory pool capacity after attempt to grow capacity through "
         "arbitration. Requestor pool name 'leaf-1.0', request size 7.00MB, "
         "memory pool capacity 4.00MB, memory pool max capacity 8.00MB");
-    ASSERT_EQ(manager.arbitrator()->shrinkCapacity(rootPool.get(), 0), 1 << 20);
+    ASSERT_EQ(manager.arbitrator()->shrinkCapacity(rootPool.get(), 0), 0);
     ASSERT_EQ(manager.arbitrator()->shrinkCapacity(leafPool.get(), 0), 0);
     ASSERT_EQ(manager.arbitrator()->shrinkCapacity(leafPool.get(), 1), 0);
     ASSERT_EQ(manager.arbitrator()->shrinkCapacity(rootPool.get(), 1), 0);
-    ASSERT_EQ(rootPool->capacity(), 3 << 20);
+    ASSERT_EQ(rootPool->capacity(), 4 << 20);
     static_cast<MemoryPoolImpl*>(rootPool.get())->testingSetReservation(0);
     ASSERT_EQ(
         manager.arbitrator()->shrinkCapacity(leafPool.get(), 1 << 20), 1 << 20);
     ASSERT_EQ(
         manager.arbitrator()->shrinkCapacity(rootPool.get(), 1 << 20), 1 << 20);
-    ASSERT_EQ(rootPool->capacity(), 1 << 20);
-    ASSERT_EQ(leafPool->capacity(), 1 << 20);
-    ASSERT_EQ(manager.arbitrator()->shrinkCapacity(leafPool.get(), 0), 1 << 20);
+    ASSERT_EQ(rootPool->capacity(), 2 << 20);
+    ASSERT_EQ(leafPool->capacity(), 2 << 20);
+    ASSERT_EQ(manager.arbitrator()->shrinkCapacity(leafPool.get(), 0), 2 << 20);
     ASSERT_EQ(rootPool->capacity(), 0);
     ASSERT_EQ(leafPool->capacity(), 0);
   }

velox/common/memory/tests/MockSharedArbitratorTest.cpp

@@ -65,8 +65,6 @@ constexpr int64_t MB = 1024L * KB;
 constexpr uint64_t kMemoryCapacity = 512 * MB;
 constexpr uint64_t kReservedMemoryCapacity = 128 * MB;
 constexpr uint64_t kMemoryPoolInitCapacity = 16 * MB;
-// TODO(jtan6): Remove after complete transfer capacity deprecation
-constexpr uint64_t kMemoryPoolTransferCapacity = 0;
 constexpr uint64_t kMemoryPoolReservedCapacity = 8 * MB;
 constexpr uint64_t kFastExponentialGrowthCapacityLimit = 32 * MB;
 constexpr double kSlowCapacityGrowPct = 0.25;
@@ -429,7 +427,6 @@ class MockSharedArbitrationTest : public testing::Test {
       int64_t reservedMemoryCapacity = kReservedMemoryCapacity,
       uint64_t memoryPoolInitCapacity = kMemoryPoolInitCapacity,
       uint64_t memoryPoolReserveCapacity = kMemoryPoolReservedCapacity,
-      uint64_t memoryPoolTransferCapacity = kMemoryPoolTransferCapacity,
       uint64_t fastExponentialGrowthCapacityLimit =
           kFastExponentialGrowthCapacityLimit,
       double slowCapacityGrowPct = kSlowCapacityGrowPct,
@@ -450,8 +447,6 @@ class MockSharedArbitrationTest : public testing::Test {
          folly::to<std::string>(memoryPoolInitCapacity) + "B"},
         {std::string(ExtraConfig::kMemoryPoolReservedCapacity),
          folly::to<std::string>(memoryPoolReserveCapacity) + "B"},
-        {std::string(ExtraConfig::kMemoryPoolTransferCapacity),
-         folly::to<std::string>(memoryPoolTransferCapacity) + "B"},
         {std::string(ExtraConfig::kFastExponentialGrowthCapacityLimit),
          folly::to<std::string>(fastExponentialGrowthCapacityLimit) + "B"},
         {std::string(ExtraConfig::kSlowCapacityGrowPct),
@@ -555,10 +550,6 @@ TEST_F(MockSharedArbitrationTest, extraConfigs) {
   ASSERT_EQ(
       SharedArbitrator::ExtraConfig::getMemoryPoolInitialCapacity(emptyConfigs),
       256 << 20);
-  ASSERT_EQ(
-      SharedArbitrator::ExtraConfig::getMemoryPoolTransferCapacity(
-          emptyConfigs),
-      128 << 20);
   ASSERT_EQ(
       SharedArbitrator::ExtraConfig::getMemoryReclaimMaxWaitTimeMs(
           emptyConfigs),
@@ -578,8 +569,6 @@ TEST_F(MockSharedArbitrationTest, extraConfigs) {
       SharedArbitrator::ExtraConfig::kMemoryPoolInitialCapacity)] = "512MB";
   configs[std::string(
       SharedArbitrator::ExtraConfig::kMemoryPoolReservedCapacity)] = "200B";
-  configs[std::string(
-      SharedArbitrator::ExtraConfig::kMemoryPoolTransferCapacity)] = "256MB";
   configs[std::string(
       SharedArbitrator::ExtraConfig::kMemoryReclaimMaxWaitTime)] = "5000ms";
   configs[std::string(
@@ -593,9 +582,6 @@ TEST_F(MockSharedArbitrationTest, extraConfigs) {
   ASSERT_EQ(
       SharedArbitrator::ExtraConfig::getMemoryPoolReservedCapacity(configs),
       200);
-  ASSERT_EQ(
-      SharedArbitrator::ExtraConfig::getMemoryPoolTransferCapacity(configs),
-      256 << 20);
   ASSERT_EQ(
       SharedArbitrator::ExtraConfig::getMemoryReclaimMaxWaitTimeMs(configs),
       5000);
@@ -610,8 +596,6 @@ TEST_F(MockSharedArbitrationTest, extraConfigs) {
       SharedArbitrator::ExtraConfig::kMemoryPoolInitialCapacity)] = "invalid";
   configs[std::string(
       SharedArbitrator::ExtraConfig::kMemoryPoolReservedCapacity)] = "invalid";
-  configs[std::string(
-      SharedArbitrator::ExtraConfig::kMemoryPoolTransferCapacity)] = "invalid";
   configs[std::string(
       SharedArbitrator::ExtraConfig::kMemoryReclaimMaxWaitTime)] = "invalid";
   configs[std::string(
@@ -627,9 +611,6 @@ TEST_F(MockSharedArbitrationTest, extraConfigs) {
   VELOX_ASSERT_THROW(
       SharedArbitrator::ExtraConfig::getMemoryPoolReservedCapacity(configs),
       "Invalid capacity string 'invalid'");
-  VELOX_ASSERT_THROW(
-      SharedArbitrator::ExtraConfig::getMemoryPoolTransferCapacity(configs),
-      "Invalid capacity string 'invalid'");
   VELOX_ASSERT_THROW(
       SharedArbitrator::ExtraConfig::getMemoryReclaimMaxWaitTimeMs(configs),
       "Invalid duration 'invalid'");
@@ -678,7 +659,7 @@ TEST_F(MockSharedArbitrationTest, arbitrationStateCheck) {
     ASSERT_TRUE(RE2::FullMatch(pool.name(), re)) << pool.name();
     ++checkCount;
   };
-  setupMemory(memCapacity, 0, 0, 0, 0, 0, 0, 0, 0, checkCountCb);
+  setupMemory(memCapacity, 0, 0, 0, 0, 0, 0, 0, checkCountCb);
 
   const int numTasks{5};
   std::vector<std::shared_ptr<MockTask>> tasks;
@@ -703,7 +684,7 @@ TEST_F(MockSharedArbitrationTest, arbitrationStateCheck) {
   MemoryArbitrationStateCheckCB badCheckCb = [&](MemoryPool& /*unused*/) {
     VELOX_FAIL("bad check");
   };
-  setupMemory(memCapacity, 0, 0, 0, 0, 0, 0, 0, 0, badCheckCb);
+  setupMemory(memCapacity, 0, 0, 0, 0, 0, 0, 0, badCheckCb);
   std::shared_ptr<MockTask> task = addTask(kMemoryCapacity);
   ASSERT_EQ(task->capacity(), 0);
   MockMemoryOperator* memOp = task->addMemoryOp();
@@ -1238,7 +1219,6 @@ DEBUG_ONLY_TEST_F(
       0,
       memoryPoolInitCapacity,
       0,
-      0,
       kFastExponentialGrowthCapacityLimit,
       kSlowCapacityGrowPct,
       0,
@@ -1331,7 +1311,6 @@ DEBUG_ONLY_TEST_F(
       0,
       memoryPoolInitCapacity,
       0,
-      0,
       kFastExponentialGrowthCapacityLimit,
       kSlowCapacityGrowPct,
       0,
@@ -1615,7 +1594,6 @@ DEBUG_ONLY_TEST_F(MockSharedArbitrationTest, globalArbitrationEnableCheck) {
         0,
         memoryPoolInitCapacity,
         0,
-        0,
         kFastExponentialGrowthCapacityLimit,
         kSlowCapacityGrowPct,
         kMemoryPoolMinFreeCapacity,
@@ -1652,7 +1630,6 @@ DEBUG_ONLY_TEST_F(
       0,
       memoryPoolInitCapacity,
       0,
-      0,
       kFastExponentialGrowthCapacityLimit,
       kSlowCapacityGrowPct,
       0,
@@ -1871,7 +1848,6 @@ TEST_F(MockSharedArbitrationTest, singlePoolShrinkWithoutArbitration) {
               0,
               0,
               0,
-              0,
               testParam.memoryPoolMinFreeCapacity,
               testParam.memoryPoolMinFreeCapacityPct),
           "both need to be set (non-zero) at the same time to enable shrink "
@@ -1885,7 +1861,6 @@ TEST_F(MockSharedArbitrationTest, singlePoolShrinkWithoutArbitration) {
           0,
           0,
           0,
-          0,
           testParam.memoryPoolMinFreeCapacity,
           testParam.memoryPoolMinFreeCapacityPct);
     }
@@ -1931,7 +1906,6 @@ TEST_F(MockSharedArbitrationTest, singlePoolGrowWithoutArbitration) {
         0,
         memoryPoolInitCapacity,
         0,
-        0,
         testParam.fastExponentialGrowthCapacityLimit,
         testParam.slowCapacityGrowPct);
 
@@ -2146,7 +2120,6 @@ TEST_F(MockSharedArbitrationTest, ensureMemoryPoolMaxCapacity) {
         0,
         poolInitCapacity,
         0,
-        0,
         kFastExponentialGrowthCapacityLimit,
         kSlowCapacityGrowPct,
         0,
@@ -2681,7 +2654,7 @@ DEBUG_ONLY_TEST_F(MockSharedArbitrationTest, failedToReclaimFromRequestor) {
        0}};
   for (const auto& testData : testSettings) {
     SCOPED_TRACE(testData.debugString());
-    setupMemory(kMemoryCapacity, 0, kMemoryPoolInitCapacity, 0, 0, 0, 0, 0, 0);
+    setupMemory(kMemoryCapacity, 0, kMemoryPoolInitCapacity, 0, 0, 0, 0, 0);
 
     std::vector<std::shared_ptr<MockTask>> otherTasks;
     std::vector<MockMemoryOperator*> otherTaskOps;
@@ -2865,7 +2838,7 @@ DEBUG_ONLY_TEST_F(MockSharedArbitrationTest, failedToReclaimFromOtherTask) {
        nonFailTaskMemoryCapacity}};
   for (const auto& testData : testSettings) {
     SCOPED_TRACE(testData.debugString());
-    setupMemory(kMemoryCapacity, 0, kMemoryPoolInitCapacity, 0, 0, 0, 0, 0, 0);
+    setupMemory(kMemoryCapacity, 0, kMemoryPoolInitCapacity, 0, 0, 0, 0, 0);
 
     std::vector<std::shared_ptr<MockTask>> nonFailedTasks;
     std::vector<MockMemoryOperator*> nonFailedTaskOps;
@@ -3015,7 +2988,6 @@ TEST_F(MockSharedArbitrationTest, memoryPoolAbortThrow) {
       0,
       kMemoryPoolInitCapacity,
       0,
-      0,
       kFastExponentialGrowthCapacityLimit,
       kSlowCapacityGrowPct,
       0,
@@ -3065,7 +3037,7 @@ TEST_F(MockSharedArbitrationTest, memoryPoolAbortThrow) {
 
 // This test makes sure the memory capacity grows as expected.
 DEBUG_ONLY_TEST_F(MockSharedArbitrationTest, concurrentArbitrationRequests) {
-  setupMemory(kMemoryCapacity, 0, 0, 0, 128 << 20);
+  setupMemory(kMemoryCapacity, 0, 0, 0);
   std::shared_ptr<MockTask> task = addTask();
   MockMemoryOperator* op1 = addMemoryOp(task);
   MockMemoryOperator* op2 = addMemoryOp(task);

velox/exec/tests/HashJoinTest.cpp

@@ -7506,7 +7506,7 @@ DEBUG_ONLY_TEST_F(HashJoinTest, taskWaitTimeout) {
 
   for (uint64_t timeoutMs : {0, 1'000, 30'000}) {
     SCOPED_TRACE(fmt::format("timeout {}", succinctMillis(timeoutMs)));
-    auto memoryManager = createMemoryManager(512 << 20, 0, 0, timeoutMs);
+    auto memoryManager = createMemoryManager(512 << 20, 0, timeoutMs);
     auto queryCtx =
         newQueryCtx(memoryManager.get(), executor_.get(), queryMemoryCapacity);
 

velox/exec/tests/utils/ArbitratorTestUtil.cpp

@@ -45,7 +45,6 @@ std::shared_ptr<core::QueryCtx> newQueryCtx(
 std::unique_ptr<memory::MemoryManager> createMemoryManager(
     int64_t arbitratorCapacity,
     uint64_t memoryPoolInitCapacity,
-    uint64_t memoryPoolTransferCapacity,
     uint64_t maxReclaimWaitMs,
     uint64_t fastExponentialGrowthCapacityLimit,
     double slowCapacityGrowPct) {
@@ -59,8 +58,6 @@ std::unique_ptr<memory::MemoryManager> createMemoryManager(
   options.extraArbitratorConfigs = {
       {std::string(ExtraConfig::kMemoryPoolInitialCapacity),
        folly::to<std::string>(memoryPoolInitCapacity) + "B"},
-      {std::string(ExtraConfig::kMemoryPoolTransferCapacity),
-       folly::to<std::string>(memoryPoolTransferCapacity) + "B"},
       {std::string(ExtraConfig::kMemoryReclaimMaxWaitTime),
        folly::to<std::string>(maxReclaimWaitMs) + "ms"},
       {std::string(ExtraConfig::kGlobalArbitrationEnabled), "true"},

velox/exec/tests/utils/ArbitratorTestUtil.h

@@ -33,7 +33,6 @@ constexpr int64_t MB = 1024L * KB;
 
 constexpr uint64_t kMemoryCapacity = 512 * MB;
 constexpr uint64_t kMemoryPoolInitCapacity = 16 * MB;
-constexpr uint64_t kMemoryPoolTransferCapacity = 8 * MB;
 
 class FakeMemoryReclaimer : public exec::MemoryReclaimer {
  public:
@@ -95,7 +94,6 @@ std::shared_ptr<core::QueryCtx> newQueryCtx(
 std::unique_ptr<memory::MemoryManager> createMemoryManager(
     int64_t arbitratorCapacity = kMemoryCapacity,
     uint64_t memoryPoolInitCapacity = kMemoryPoolInitCapacity,
-    uint64_t memoryPoolTransferCapacity = kMemoryPoolTransferCapacity,
     uint64_t maxReclaimWaitMs = 0,
     uint64_t fastExponentialGrowthCapacityLimit = 0,
     double slowCapacityGrowPct = 0);