[CELEBORN-2229][CIP-14] Add support for celeborn.client.push.maxBytesSizeInFlight in CppClient #3568

afterincomparableyum · 2025-12-16T11:06:33Z

What changes were proposed in this pull request?

Add support for celeborn.client.push.maxBytesSizeInFlight in CppClient, similar to InFlightRequestTracker.java

Does this PR resolve a correctness bug?

No

How was this patch tested?

Compile locally, CI/CD will run unit tests

afterincomparableyum · 2025-12-21T23:13:23Z

@HolyLow @SteNicholas @RexXiong @FMX Could you please help review this PR? Thanks a lot for helping improve this as needed!

HolyLow · 2025-12-22T08:41:03Z

cpp/celeborn/conf/CelebornConf.cpp

          NUM_PROP(kShuffleCompressionZstdCompressLevel, 1),
+          STR_PROP(kClientPushBufferMaxSize, "64k"),
+          BOOL_PROP(kClientPushMaxBytesSizeInFlightEnabled, false),
+          NONE_PROP(kClientPushMaxBytesSizeInFlightTotal),


Why is this NONE_PROP?

my understanding is that NONE_PROP is correct. It matches Java's .createOptional behavior.

the property is optional, and the default is computed dynamically when not set.

Do you believe it should be something else other than NONE_PROP? I feel like NONE_PROP is good because these parameters are optional.

HolyLow · 2025-12-22T08:41:41Z

cpp/celeborn/client/writer/PushState.cpp

  pushStrategy_->clear();
+
+  if (maxInFlightBytesSizeEnabled_) {
+    LOG(INFO) << "Cleanup " << totalInflightBytes_.load() << " bytes in flight.";


Pls do not use LOG(INFO) here which is too verbose.

HolyLow · 2025-12-22T08:42:53Z

cpp/celeborn/client/writer/PushState.h

+      int>>
+      inflightBatchBytesSizes_;
  folly::Synchronized<std::unique_ptr<std::exception>> exception_;
+  volatile bool cleaned_{false};


Why should this be volatile? The volatile keyword in CPP has different meaning from Java.

HolyLow · 2025-12-22T08:44:04Z

cpp/celeborn/conf/CelebornConf.cpp

              kShuffleCompressionCodec,
              protocol::toString(protocol::CompressionCodec::NONE)),
          NUM_PROP(kShuffleCompressionZstdCompressLevel, 1),
+          STR_PROP(kClientPushBufferMaxSize, "64k"),


How will this be translated? Is this option tested?

HolyLow · 2025-12-22T08:44:45Z

cpp/celeborn/conf/CelebornConf.cpp

      optionalProperty(kShuffleCompressionZstdCompressLevel).value());
 }
 } // namespace conf
-} // namespace celeborn


Why is this change necessary?

HolyLow · 2025-12-22T08:46:10Z

cpp/celeborn/client/tests/PushStateTest.cpp

+    }
+
+    std::unique_ptr<PushState> pushState_;
+    static constexpr int pushTimeoutMs_ = 100;


It might be more consistent if we change the naming of static constexpr namings to kXxxXxx_

HolyLow · 2025-12-22T08:46:55Z

cpp/celeborn/client/tests/PushStateTest.cpp

+  pushState_->cleanup();
+
+  EXPECT_FALSE(pushState_->limitMaxInFlight(hostAndPushPort));
+}


Should be an empty line at end of file.

HolyLow · 2025-12-22T08:47:43Z

cpp/celeborn/client/writer/PushState.h

      std::string,
      std::shared_ptr<utils::ConcurrentHashSet<int>>>
      inflightBatchesPerAddress_;
+  std::optional<utils::ConcurrentHashMap<


If we already have flags, why should we use std::optional here?

HolyLow · 2025-12-22T08:48:27Z

cpp/celeborn/conf/CelebornConf.cpp

+long CelebornConf::clientPushMaxBytesSizeInFlightPerWorker() const {
+  auto optionalValue =
+      optionalProperty(kClientPushMaxBytesSizeInFlightPerWorker);
+  long maxBytesSizeInFlight =


Why should this be that complicated?

any suggestions for how to simplify it @HolyLow ? I see in the java client, this is how the function is:

def clientPushMaxBytesSizeInFlightPerWorker: Long = { val maxBytesSizeInFlight = get(CLIENT_PUSH_MAX_BYTES_SIZE_IN_FLIGHT_PERWORKER).getOrElse(0L) if (clientPushMaxBytesSizeInFlightEnabled && maxBytesSizeInFlight > 0L) { maxBytesSizeInFlight } else { clientPushMaxReqsInFlightPerWorker * clientPushBufferMaxSize } }

The logic is similar

HolyLow · 2025-12-22T08:48:45Z

cpp/celeborn/conf/CelebornConf.cpp

+}
+
+bool CelebornConf::clientPushMaxBytesSizeInFlightEnabled() const {
+  return optionalProperty(kClientPushMaxBytesSizeInFlightEnabled).value() ==


I think this is not consistent to existing code.

HolyLow · 2025-12-22T08:49:13Z

cpp/celeborn/conf/CelebornConf.cpp

+
+long CelebornConf::clientPushMaxBytesSizeInFlightTotal() const {
+  auto optionalValue = optionalProperty(kClientPushMaxBytesSizeInFlightTotal);
+  long maxBytesSizeInFlight =


ditto. Why is this so complicated?

It's exactly how the java client function is:

def clientPushMaxBytesSizeInFlightTotal: Long = { val maxBytesSizeInFlight = get(CLIENT_PUSH_MAX_BYTES_SIZE_IN_FLIGHT_TOTAL).getOrElse(0L) if (clientPushMaxBytesSizeInFlightEnabled && maxBytesSizeInFlight > 0L) { maxBytesSizeInFlight } else { clientPushMaxReqsInFlightTotal * clientPushBufferMaxSize } }

Any suggestions for how to simplify it @HolyLow ?

HolyLow

Thanks for your contribution. I've left some comments, and I think we should fix all the CI tests firstly.

afterincomparableyum · 2025-12-24T02:16:49Z

Thank you for comments @HolyLow wiil address them soon

afterincomparableyum · 2025-12-24T05:34:32Z

most recent commit should fix C++ unit test errors and clang formatting issues, next commit will address remaining comments.

afterincomparableyum · 2025-12-25T16:31:47Z

@HolyLow are you able to approve the github workflows again please? I've addressed all comments/responded to ones I did not make any code changes for. C++ unit tests pass, and I've applied clang format.

afterincomparableyum · 2025-12-26T14:46:24Z

@HolyLow is it safe to ignore the CI/CD Failures for Celeborn CI. All of them are due to:

curl: (28) Failed to connect to archive.apache.org port 443 after 135159 ms: Connection timed out

As for the Celeborn SBT CI failures, I will look into those.

afterincomparableyum · 2025-12-29T21:56:30Z

After taking a closer look it looks like the CI/CD failures for Celeborn SBT CI are not related to my C++ Code changes. It looks flaky to me.

I also see the exact workflows passed in my previous commit it ran for: 41ac021

I also ran the tests locally:

[info] Passed: Total 10, Failed 0, Errors 0, Passed 10
[info] Passed: Total 11, Failed 0, Errors 0, Passed 11
[info] ScalaTest
[info] Run completed in 1 hour, 16 seconds.
[info] Total number of tests run: 1
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 1, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
[info] Passed: Total 26, Failed 0, Errors 0, Passed 26
[info] Run completed in 1 hour, 1 second.
[info] Total number of tests run: 111
[info] Suites: completed 28, aborted 0
[info] Tests: succeeded 111, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
[success] Total time: 3737 s (01:02:17), completed Dec 29, 2025 9:50:22 PM

@HolyLow let me know if there are any other concerns with my PR that you would like me to address. Thank you for your review!

Add support for celeborn.client.push.maxBytesSizeInFlight in CppClient

c7d9409

github-actions bot added the module:cpp label Dec 16, 2025

HolyLow self-requested a review December 17, 2025 08:59

Fix compile errors on PushState.cpp

41ac021

afterincomparableyum marked this pull request as ready for review December 21, 2025 23:11

HolyLow reviewed Dec 22, 2025

View reviewed changes

Fix unit test errors/logic as well as clang format issues

85a0dc2

afterincomparableyum added 2 commits December 25, 2025 10:07

addressing comments

ac4d699

fix clang formatting

26724ba

afterincomparableyum mentioned this pull request Dec 28, 2025

[WIP][CELEBORN-2221][CIP-14] Add support for write compression in CppClient #3575

Draft

[CELEBORN-2229][CIP-14] Add support for celeborn.client.push.maxBytesSizeInFlight in CppClient #3568

Are you sure you want to change the base?

[CELEBORN-2229][CIP-14] Add support for celeborn.client.push.maxBytesSizeInFlight in CppClient #3568

Uh oh!

Conversation

afterincomparableyum commented Dec 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Does this PR resolve a correctness bug?

How was this patch tested?

Uh oh!

afterincomparableyum commented Dec 21, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

afterincomparableyum Dec 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HolyLow left a comment

Choose a reason for hiding this comment

Uh oh!

afterincomparableyum commented Dec 24, 2025

Uh oh!

afterincomparableyum commented Dec 24, 2025

Uh oh!

afterincomparableyum commented Dec 25, 2025

Uh oh!

afterincomparableyum commented Dec 26, 2025

Uh oh!

afterincomparableyum commented Dec 29, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

afterincomparableyum commented Dec 16, 2025 •

edited

Loading

afterincomparableyum Dec 25, 2025 •

edited

Loading