This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 69200e209a [CELEBORN-2220][CIP-14] Support Slow Start Push in C++
Client
69200e209a is described below
commit 69200e209a21168d29faa17be03dadfb5a6f259f
Author: afterincomparableyum
<[email protected]>
AuthorDate: Tue Oct 20 11:23:09 2026 +0800
[CELEBORN-2220][CIP-14] Support Slow Start Push in C++ Client
### What changes were proposed in this pull request?
Add `SlowStartPushStrategy` to the C++ client, a port of the Java client's
TCP-like push congestion control:
- Keeps per-worker congestion state: the in-flight request limit grows by
one per successful push until it reaches
`celeborn.client.push.maxReqsInFlight.perWorker`, then grows linearly. The
limit halves when a worker responds with
`PUSH_DATA_SUCCESS_{PRIMARY,REPLICA}_CONGESTED`, and pushes sleep while the
limit is below the maximum.
- Selected with `celeborn.client.push.limit.strategy=SLOWSTART`
(case-insensitive, as in Java). Before this change, the C++ client rejected
`SLOWSTART`.
- Adds `celeborn.client.push.slowStart.initialSleepTime` (default `500ms`)
and `celeborn.client.push.slowStart.maxSleepTime` (default `2s`), with the same
keys and defaults as Java.
The C++ push callbacks and `PushState` already call `onSuccess`,
`onCongestControl`, `limitPushSpeed` and `getCurrentMaxReqsInFlight`, so this
PR only adds the strategy itself.
#3730 is closed as stale. That PR's review comments are addressed here.
### Why are the changes needed?
To close a feature gap between the C++ and Java clients (CIP-14).
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
C++ client users can set `celeborn.client.push.limit.strategy=SLOWSTART`
and the two `celeborn.client.push.slowStart.*` configs. The default strategy is
still `SIMPLE`. Like the Java client, SLOWSTART only backs off when workers
have `celeborn.worker.congestionControl.enabled=true`.
### How was this patch tested?
- New `PushStrategyTest`: strategy selection, sleep-time calculation, slow
start and congestion avoidance, and isolation between hosts. Its assertions
mirror Java's `SlowStartPushStrategyTest`.
- `CelebornConfTest`: defaults and overrides for the new configs.
Closes #3878 from afterincomparableyum/CELEBORN-2220.
Authored-by: afterincomparableyum
<[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/194
AI-Contributed/UT: 0/175
---
cpp/celeborn/client/tests/CMakeLists.txt | 1 +
cpp/celeborn/client/tests/PushStrategyTest.cpp | 161 +++++++++++++++++++++++++
cpp/celeborn/client/writer/PushStrategy.cpp | 101 ++++++++++++++++
cpp/celeborn/client/writer/PushStrategy.h | 65 +++++++++-
cpp/celeborn/conf/CelebornConf.cpp | 16 +++
cpp/celeborn/conf/CelebornConf.h | 12 ++
cpp/celeborn/conf/tests/CelebornConfTest.cpp | 13 ++
7 files changed, 368 insertions(+), 1 deletion(-)
diff --git a/cpp/celeborn/client/tests/CMakeLists.txt
b/cpp/celeborn/client/tests/CMakeLists.txt
index 08623486b2..fa50f2e879 100644
--- a/cpp/celeborn/client/tests/CMakeLists.txt
+++ b/cpp/celeborn/client/tests/CMakeLists.txt
@@ -22,6 +22,7 @@ add_executable(
CelebornInputStreamTest.cpp
CelebornInputStreamRetryTest.cpp
PushStateTest.cpp
+ PushStrategyTest.cpp
ReviveManagerTest.cpp
Lz4DecompressorTest.cpp
ZstdDecompressorTest.cpp
diff --git a/cpp/celeborn/client/tests/PushStrategyTest.cpp
b/cpp/celeborn/client/tests/PushStrategyTest.cpp
new file mode 100644
index 0000000000..094cf1ef3d
--- /dev/null
+++ b/cpp/celeborn/client/tests/PushStrategyTest.cpp
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <gtest/gtest.h>
+
+#include "celeborn/client/writer/PushStrategy.h"
+
+using namespace celeborn;
+using namespace celeborn::client;
+
+namespace {
+std::unique_ptr<SlowStartPushStrategy> createSlowStartStrategy(
+ int maxReqsInFlightPerWorker,
+ const std::string& maxSleepTime) {
+ conf::CelebornConf conf;
+ conf.registerProperty(
+ conf::CelebornConf::kClientPushMaxReqsInFlightPerWorker,
+ std::to_string(maxReqsInFlightPerWorker));
+ conf.registerProperty(
+ conf::CelebornConf::kClientPushSlowStartMaxSleepTime, maxSleepTime);
+ // Strategy selection through PushStrategy::create is covered by the
+ // createStrategy test.
+ return std::make_unique<SlowStartPushStrategy>(conf);
+}
+} // namespace
+
+TEST(PushStrategyTest, createStrategy) {
+ conf::CelebornConf conf;
+ // Default strategy is SIMPLE.
+ EXPECT_NE(
+ dynamic_cast<SimplePushStrategy*>(PushStrategy::create(conf).get()),
+ nullptr);
+
+ // The strategy name is case-insensitive, as the Java config entry
+ // uppercases the configured value.
+ conf.registerProperty(
+ conf::CelebornConf::kClientPushLimitStrategy, "slowstart");
+ EXPECT_NE(
+ dynamic_cast<SlowStartPushStrategy*>(PushStrategy::create(conf).get()),
+ nullptr);
+
+ conf.registerProperty(
+ conf::CelebornConf::kClientPushLimitStrategy, "UNKNOWN");
+ EXPECT_THROW(PushStrategy::create(conf), std::exception);
+}
+
+TEST(PushStrategyTest, sleepTime) {
+ auto strategy = createSlowStartStrategy(32, "3s");
+ const std::string dummyHostPort = "test:9087";
+ auto context = strategy->getCongestControlContextByAddress(dummyHostPort);
+
+ // If the currentReq is 1, should sleep 440 ms
+ EXPECT_EQ(440, strategy->getSleepTime(*context));
+
+ // If the currentReq is 8, should sleep 20 ms
+ for (int i = 0; i < 7; i++) {
+ strategy->onSuccess(dummyHostPort);
+ }
+ EXPECT_EQ(20, strategy->getSleepTime(*context));
+
+ // If the currentReq is 16, should sleep 0 ms
+ for (int i = 0; i < 8; i++) {
+ strategy->onSuccess(dummyHostPort);
+ }
+ EXPECT_EQ(0, strategy->getSleepTime(*context));
+
+ // Congest the requests, the currentReq reduced to 8, should sleep 20 ms
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(20, strategy->getSleepTime(*context));
+
+ // Congest the requests, the currentReq reduced to 4, should sleep 260 ms
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(260, strategy->getSleepTime(*context));
+ // Congest the requests, the currentReq reduced to 2, should sleep 380 ms
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(380, strategy->getSleepTime(*context));
+ // Congest the requests, the currentReq reduced to 1, should sleep 440 ms
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(440, strategy->getSleepTime(*context));
+ // Keep congest the requests even the currentReq reduced to 1, will increase
+ // the sleep time 1s
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(1440, strategy->getSleepTime(*context));
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(2440, strategy->getSleepTime(*context));
+
+ // Cannot exceed the max sleep time
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(3000, strategy->getSleepTime(*context));
+
+ // If start to return success, the currentReq is increased to 2, should
+ // sleep 380 ms
+ strategy->onSuccess(dummyHostPort);
+ EXPECT_EQ(380, strategy->getSleepTime(*context));
+}
+
+TEST(PushStrategyTest, congestStrategy) {
+ auto strategy = createSlowStartStrategy(5, "4s");
+ const std::string dummyHostPort = "test:9087";
+ // Slow start, should exponentially increase the currentReq
+ strategy->onSuccess(dummyHostPort);
+ EXPECT_EQ(2, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ EXPECT_EQ(4, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+
+ // Will linearly increase the currentReq if meet the maxReqsInFlight
+ EXPECT_EQ(5, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ strategy->onSuccess(dummyHostPort);
+ EXPECT_EQ(6, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+
+ // Congest controlled, should half the currentReq
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(3, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(1, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+
+ // Cannot lower than 1
+ strategy->onCongestControl(dummyHostPort);
+ EXPECT_EQ(1, strategy->getCurrentMaxReqsInFlight(dummyHostPort));
+}
+
+TEST(PushStrategyTest, multiHosts) {
+ auto strategy = createSlowStartStrategy(3, "3s");
+ const std::string dummyHostPort1 = "test1:9087";
+ const std::string dummyHostPort2 = "test2:9087";
+ auto context1 = strategy->getCongestControlContextByAddress(dummyHostPort1);
+ auto context2 = strategy->getCongestControlContextByAddress(dummyHostPort2);
+
+ EXPECT_EQ(440, strategy->getSleepTime(*context1));
+ EXPECT_EQ(440, strategy->getSleepTime(*context2));
+
+ // Control the dummyHostPort1, should not affect dummyHostPort2
+ for (int i = 0; i < 3; i++) {
+ strategy->onSuccess(dummyHostPort1);
+ }
+ EXPECT_EQ(0, strategy->getSleepTime(*context1));
+ EXPECT_EQ(440, strategy->getSleepTime(*context2));
+}
diff --git a/cpp/celeborn/client/writer/PushStrategy.cpp
b/cpp/celeborn/client/writer/PushStrategy.cpp
index 66dad6c4e4..ab4ce08e30 100644
--- a/cpp/celeborn/client/writer/PushStrategy.cpp
+++ b/cpp/celeborn/client/writer/PushStrategy.cpp
@@ -17,18 +17,119 @@
#include "celeborn/client/writer/PushStrategy.h"
+#include <algorithm>
+#include <cctype>
+#include <thread>
+
namespace celeborn {
namespace client {
std::unique_ptr<PushStrategy> PushStrategy::create(
const conf::CelebornConf& conf) {
auto strategyName = conf.clientPushLimitStrategy();
+ // The Java config entry uppercases the configured value.
+ std::transform(
+ strategyName.begin(),
+ strategyName.end(),
+ strategyName.begin(),
+ [](unsigned char c) { return static_cast<char>(std::toupper(c)); });
if (strategyName == conf::CelebornConf::kSimplePushStrategy) {
return std::make_unique<SimplePushStrategy>(conf);
+ } else if (strategyName == conf::CelebornConf::kSlowStartPushStrategy) {
+ return std::make_unique<SlowStartPushStrategy>(conf);
} else {
CELEBORN_FAIL("unsupported pushStrategy: " + strategyName);
}
}
+void SlowStartPushStrategy::CongestControlContext::increaseCurrentMaxReqs() {
+ std::lock_guard<std::mutex> lock(mutex_);
+ continueCongestedNumber_.store(0);
+ if (currentMaxReqsInFlight_.load() >= reqsInFlightBlockThreshold_) {
+ // Congestion avoidance
+ congestionAvoidanceFlag_++;
+ if (congestionAvoidanceFlag_ >= currentMaxReqsInFlight_.load()) {
+ currentMaxReqsInFlight_.fetch_add(1);
+ congestionAvoidanceFlag_ = 0;
+ }
+ } else {
+ // Slow start
+ currentMaxReqsInFlight_.fetch_add(1);
+ }
+}
+
+void SlowStartPushStrategy::CongestControlContext::decreaseCurrentMaxReqs() {
+ std::lock_guard<std::mutex> lock(mutex_);
+ if (currentMaxReqsInFlight_.load() <= 1) {
+ currentMaxReqsInFlight_.store(1);
+ continueCongestedNumber_.fetch_add(1);
+ } else {
+ currentMaxReqsInFlight_.store(currentMaxReqsInFlight_.load() / 2);
+ }
+ reqsInFlightBlockThreshold_ = currentMaxReqsInFlight_.load();
+ congestionAvoidanceFlag_ = 0;
+}
+
+std::shared_ptr<SlowStartPushStrategy::CongestControlContext>
+SlowStartPushStrategy::getCongestControlContextByAddress(
+ const std::string& hostAndPushPort) {
+ return congestControlInfoPerAddress_.computeIfAbsent(hostAndPushPort, [&]() {
+ return std::make_shared<CongestControlContext>(maxInFlightPerWorker_);
+ });
+}
+
+void SlowStartPushStrategy::onSuccess(const std::string& hostAndPushPort) {
+ getCongestControlContextByAddress(hostAndPushPort)->increaseCurrentMaxReqs();
+}
+
+void SlowStartPushStrategy::onCongestControl(
+ const std::string& hostAndPushPort) {
+ getCongestControlContextByAddress(hostAndPushPort)->decreaseCurrentMaxReqs();
+}
+
+long SlowStartPushStrategy::getSleepTime(CongestControlContext& context) const
{
+ int currentMaxReqs = context.getCurrentMaxReqsInFlight();
+ if (currentMaxReqs >= maxInFlightPerWorker_) {
+ return 0;
+ }
+
+ long sleepInterval = initialSleepMills_ - 60L * currentMaxReqs;
+
+ if (currentMaxReqs == 1) {
+ return std::min(
+ sleepInterval + context.getContinueCongestedNumber() * 1000L,
+ maxSleepMills_);
+ }
+
+ return std::max(sleepInterval, 0L);
+}
+
+void SlowStartPushStrategy::limitPushSpeed(
+ PushState& pushState,
+ const std::string& hostAndPushPort) {
+ auto exceptionMsg = pushState.getExceptionMsg();
+ if (exceptionMsg.has_value()) {
+ CELEBORN_FAIL(exceptionMsg.value());
+ }
+ auto congestControlContext =
+ getCongestControlContextByAddress(hostAndPushPort);
+ long sleepInterval = getSleepTime(*congestControlContext);
+ if (sleepInterval > 0L) {
+ VLOG(1) << "Will sleep " << sleepInterval
+ << " ms to control the push speed to " << hostAndPushPort << ".";
+ std::this_thread::sleep_for(utils::MS(sleepInterval));
+ }
+}
+
+int SlowStartPushStrategy::getCurrentMaxReqsInFlight(
+ const std::string& hostAndPushPort) {
+ return getCongestControlContextByAddress(hostAndPushPort)
+ ->getCurrentMaxReqsInFlight();
+}
+
+void SlowStartPushStrategy::clear() {
+ congestControlInfoPerAddress_.clear();
+}
+
} // namespace client
} // namespace celeborn
diff --git a/cpp/celeborn/client/writer/PushStrategy.h
b/cpp/celeborn/client/writer/PushStrategy.h
index 60a40f5934..7e7b35d026 100644
--- a/cpp/celeborn/client/writer/PushStrategy.h
+++ b/cpp/celeborn/client/writer/PushStrategy.h
@@ -17,10 +17,13 @@
#pragma once
+#include <atomic>
#include <memory>
+#include <mutex>
#include "celeborn/client/writer/PushState.h"
#include "celeborn/conf/CelebornConf.h"
+#include "celeborn/utils/CelebornUtils.h"
namespace celeborn {
namespace client {
@@ -73,7 +76,67 @@ class SimplePushStrategy : public PushStrategy {
const int maxInFlightPerWorker_;
};
-// TODO: support SlowStartPushStrategy
+class SlowStartPushStrategy : public PushStrategy {
+ public:
+ class CongestControlContext {
+ public:
+ CongestControlContext(int reqsInFlightBlockThreshold)
+ : reqsInFlightBlockThreshold_(reqsInFlightBlockThreshold) {}
+
+ void increaseCurrentMaxReqs();
+
+ void decreaseCurrentMaxReqs();
+
+ int getCurrentMaxReqsInFlight() const {
+ return currentMaxReqsInFlight_.load();
+ }
+
+ int getContinueCongestedNumber() const {
+ return continueCongestedNumber_.load();
+ }
+
+ private:
+ std::mutex mutex_;
+ std::atomic<int> currentMaxReqsInFlight_{1};
+ // Indicate the number of congested times even after the in flight
+ // requests reduced to 1.
+ std::atomic<int> continueCongestedNumber_{0};
+ int congestionAvoidanceFlag_{0};
+ int reqsInFlightBlockThreshold_;
+ };
+
+ SlowStartPushStrategy(const conf::CelebornConf& conf)
+ : maxInFlightPerWorker_(conf.clientPushMaxReqsInFlightPerWorker()),
+ initialSleepMills_(conf.clientPushSlowStartInitialSleepTimeMs()),
+ maxSleepMills_(conf.clientPushSlowStartMaxSleepTimeMs()) {}
+
+ ~SlowStartPushStrategy() override = default;
+
+ void onSuccess(const std::string& hostAndPushPort) override;
+
+ void onCongestControl(const std::string& hostAndPushPort) override;
+
+ void clear() override;
+
+ void limitPushSpeed(PushState& pushState, const std::string& hostAndPushPort)
+ override;
+
+ int getCurrentMaxReqsInFlight(const std::string& hostAndPushPort) override;
+
+ // Visible for testing.
+ std::shared_ptr<CongestControlContext> getCongestControlContextByAddress(
+ const std::string& hostAndPushPort);
+
+ // Visible for testing.
+ long getSleepTime(CongestControlContext& context) const;
+
+ private:
+ const int maxInFlightPerWorker_;
+ const long initialSleepMills_;
+ const long maxSleepMills_;
+ utils::ConcurrentHashMap<std::string, std::shared_ptr<CongestControlContext>>
+ congestControlInfoPerAddress_;
+};
} // namespace client
} // namespace celeborn
diff --git a/cpp/celeborn/conf/CelebornConf.cpp
b/cpp/celeborn/conf/CelebornConf.cpp
index c8ca58d1f9..20257b62a0 100644
--- a/cpp/celeborn/conf/CelebornConf.cpp
+++ b/cpp/celeborn/conf/CelebornConf.cpp
@@ -147,6 +147,8 @@ CelebornConf::defaultProperties() {
NUM_PROP(kClientPushReviveBatchSize, 2048),
NUM_PROP(kClientPushMaxReviveTimes, 5),
STR_PROP(kClientPushLimitStrategy, kSimplePushStrategy),
+ STR_PROP(kClientPushSlowStartInitialSleepTime, "500ms"),
+ STR_PROP(kClientPushSlowStartMaxSleepTime, "2s"),
NUM_PROP(kClientPushMaxReqsInFlightPerWorker, 32),
NUM_PROP(kClientPushMaxReqsInFlightTotal, 256),
NUM_PROP(kClientPushLimitInFlightTimeoutMs, 240000),
@@ -258,6 +260,20 @@ std::string CelebornConf::clientPushLimitStrategy() const {
return optionalProperty(kClientPushLimitStrategy).value();
}
+long CelebornConf::clientPushSlowStartInitialSleepTimeMs() const {
+ return utils::toTimeout(
+ toDuration(optionalProperty(kClientPushSlowStartInitialSleepTime)
+ .value()))
+ .count();
+}
+
+long CelebornConf::clientPushSlowStartMaxSleepTimeMs() const {
+ return utils::toTimeout(
+ toDuration(
+ optionalProperty(kClientPushSlowStartMaxSleepTime).value()))
+ .count();
+}
+
int CelebornConf::clientPushMaxReqsInFlightPerWorker() const {
return std::stoi(
optionalProperty(kClientPushMaxReqsInFlightPerWorker).value());
diff --git a/cpp/celeborn/conf/CelebornConf.h b/cpp/celeborn/conf/CelebornConf.h
index 860911eb41..ac9684c79c 100644
--- a/cpp/celeborn/conf/CelebornConf.h
+++ b/cpp/celeborn/conf/CelebornConf.h
@@ -77,6 +77,14 @@ class CelebornConf : public BaseConf {
static constexpr std::string_view kSimplePushStrategy{"SIMPLE"};
+ static constexpr std::string_view kSlowStartPushStrategy{"SLOWSTART"};
+
+ static constexpr std::string_view kClientPushSlowStartInitialSleepTime{
+ "celeborn.client.push.slowStart.initialSleepTime"};
+
+ static constexpr std::string_view kClientPushSlowStartMaxSleepTime{
+ "celeborn.client.push.slowStart.maxSleepTime"};
+
static constexpr std::string_view kClientPushMaxReqsInFlightPerWorker{
"celeborn.client.push.maxReqsInFlight.perWorker"};
@@ -185,6 +193,10 @@ class CelebornConf : public BaseConf {
std::string clientPushLimitStrategy() const;
+ long clientPushSlowStartInitialSleepTimeMs() const;
+
+ long clientPushSlowStartMaxSleepTimeMs() const;
+
int clientPushMaxReqsInFlightPerWorker() const;
int clientPushMaxReqsInFlightTotal() const;
diff --git a/cpp/celeborn/conf/tests/CelebornConfTest.cpp
b/cpp/celeborn/conf/tests/CelebornConfTest.cpp
index 3686660330..f541b454a1 100644
--- a/cpp/celeborn/conf/tests/CelebornConfTest.cpp
+++ b/cpp/celeborn/conf/tests/CelebornConfTest.cpp
@@ -54,6 +54,9 @@ void testDefaultValues(CelebornConf* conf) {
EXPECT_EQ(conf->clientFetchMaxRetriesForEachReplica(), 3);
EXPECT_EQ(conf->networkIoRetryWait(), SECOND(5));
EXPECT_FALSE(conf->clientPushReplicateEnabled());
+ EXPECT_EQ(conf->clientPushLimitStrategy(),
CelebornConf::kSimplePushStrategy);
+ EXPECT_EQ(conf->clientPushSlowStartInitialSleepTimeMs(), 500);
+ EXPECT_EQ(conf->clientPushSlowStartMaxSleepTimeMs(), 2000);
EXPECT_FALSE(conf->clientFetchExcludeWorkerOnFailureEnabled());
EXPECT_EQ(conf->clientFetchExcludedWorkerExpireTimeout(), SECOND(60));
EXPECT_FALSE(conf->clientAdaptiveOptimizeSkewedPartitionReadEnabled());
@@ -101,6 +104,16 @@ TEST(CelebornConfTest, setValues) {
EXPECT_TRUE(conf->clientPushReplicateEnabled());
conf->registerProperty(CelebornConf::kClientPushReplicateEnabled, "false");
EXPECT_FALSE(conf->clientPushReplicateEnabled());
+ conf->registerProperty(
+ CelebornConf::kClientPushLimitStrategy,
+ std::string(CelebornConf::kSlowStartPushStrategy));
+ EXPECT_EQ(
+ conf->clientPushLimitStrategy(), CelebornConf::kSlowStartPushStrategy);
+ conf->registerProperty(
+ CelebornConf::kClientPushSlowStartInitialSleepTime, "1s");
+ EXPECT_EQ(conf->clientPushSlowStartInitialSleepTimeMs(), 1000);
+ conf->registerProperty(CelebornConf::kClientPushSlowStartMaxSleepTime, "3s");
+ EXPECT_EQ(conf->clientPushSlowStartMaxSleepTimeMs(), 3000);
conf->registerProperty(
CelebornConf::kClientFetchExcludeWorkerOnFailureEnabled, "true");
EXPECT_TRUE(conf->clientFetchExcludeWorkerOnFailureEnabled());
