You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "HolyLow (via GitHub)" <gi...@apache.org> on 2025/11/19 03:08:07 UTC
[PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
HolyLow opened a new pull request, #3543:
URL: https://github.com/apache/celeborn/pull/3543
### What changes were proposed in this pull request?
This PR supports PushDataCallback in CppClient.
### Why are the changes needed?
PushDataCallback is the building block of PushData logic of CppClient's writing procedure.
### Does this PR resolve a correctness bug?
No.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Compilation and UTs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#issuecomment-3578958977
@SteNicholas Thanks for your comments, refactored as suggested. Any further suggestion is greatly appreciated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562722841
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ default: {
+ // This is treated as success.
+ LOG(WARNING) << "unhandled PushData success protocol::StatusCode: "
+ << reason;
+ }
+ }
+}
+
+void PushDataCallback::onFailure(std::unique_ptr<std::exception> exception) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnFailure, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (pushState_->exceptionExists()) {
+ return;
+ }
+
+ LOG(ERROR) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed for shuffle " << shuffleId_ << " map " << mapId_
+ << " attempt " << attemptId_ << " partition " << partitionId_
+ << " batch " << batchId_ << ", remain revive times "
+ << remainingReviveTimes_;
+
+ if (remainingReviveTimes_ <= 0) {
+ // TODO: set more specific exception.
+ pushState_->setException(std::move(exception));
+ return;
+ }
+
+ if (sharedClient->mapperEnded(shuffleId_, mapId_)) {
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ LOG(INFO) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed but mapper already ended for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_
+ << ", remain revive times " << remainingReviveTimes_ << ".";
+ return;
+ }
+ remainingReviveTimes_--;
+ // TODO: the cause should be extracted from error msg.
+ protocol::StatusCode cause =
+ protocol::StatusCode::PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY;
+ reviveAndRetryPushData(*sharedClient, cause);
+}
+
+void PushDataCallback::updateLatestLocation(
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ pushState_->addBatch(batchId_, latestLocation->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ latestLocation_ = latestLocation;
+}
+
+void PushDataCallback::reviveAndRetryPushData(
+ ShuffleClientImpl& shuffleClient,
+ protocol::StatusCode cause) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ cause);
+ VLOG(1) << "addRequest to reviveManager, shuffleId "
+ << reviveRequest->shuffleId << " mapId " << reviveRequest->mapId
+ << " attemptId " << reviveRequest->attemptId << " partitionId "
+ << reviveRequest->partitionId << " batchId " << batchId_ << " epoch "
+ << reviveRequest->epoch;
+ shuffleClient.addRequestToReviveManager(reviveRequest);
+ long dueTimeMs = utils::currentTimeMillis() +
+ shuffleClient.conf_->clientRpcRequestPartitionLocationRpcAskTimeout() /
+ utils::MS(1);
+ shuffleClient.addPushDataRetryTask(
+ [weakClient = this->weakClient_,
+ shuffleId = this->shuffleId_,
+ body = this->databody_->clone(),
+ batchId = this->batchId_,
+ callback = shared_from_this(),
+ pushState = this->pushState_,
+ reviveRequest,
+ remainingReviveTimes = this->remainingReviveTimes_,
+ dueTimeMs]() {
+ auto sharedClient = weakClient.lock();
Review Comment:
Ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562721736
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ default: {
+ // This is treated as success.
+ LOG(WARNING) << "unhandled PushData success protocol::StatusCode: "
+ << reason;
+ }
+ }
+}
+
+void PushDataCallback::onFailure(std::unique_ptr<std::exception> exception) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnFailure, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (pushState_->exceptionExists()) {
+ return;
+ }
+
+ LOG(ERROR) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed for shuffle " << shuffleId_ << " map " << mapId_
+ << " attempt " << attemptId_ << " partition " << partitionId_
+ << " batch " << batchId_ << ", remain revive times "
+ << remainingReviveTimes_;
+
+ if (remainingReviveTimes_ <= 0) {
+ // TODO: set more specific exception.
+ pushState_->setException(std::move(exception));
+ return;
+ }
+
+ if (sharedClient->mapperEnded(shuffleId_, mapId_)) {
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ LOG(INFO) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed but mapper already ended for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_
+ << ", remain revive times " << remainingReviveTimes_ << ".";
+ return;
+ }
+ remainingReviveTimes_--;
+ // TODO: the cause should be extracted from error msg.
+ protocol::StatusCode cause =
Review Comment:
Does this need to check whether is `PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY ` or `PUSH_DATA_CONNECTION_EXCEPTION_REPLICA`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562908552
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ default: {
+ // This is treated as success.
+ LOG(WARNING) << "unhandled PushData success protocol::StatusCode: "
+ << reason;
+ }
+ }
+}
+
+void PushDataCallback::onFailure(std::unique_ptr<std::exception> exception) {
+ auto sharedClient = weakClient_.lock();
Review Comment:
For protection purpose I think it is more proper to keep the checking lines verbose. Besides, if we extract the checking logic to some other function, the function still needs to return the shared_ptr, and we still needs to check if the returned value is valid or not then exit the function accordingly. In this way, no code line is saved, and the logic becomes even more verbose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3543: [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient
URL: https://github.com/apache/celeborn/pull/3543
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562949013
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ default: {
+ // This is treated as success.
+ LOG(WARNING) << "unhandled PushData success protocol::StatusCode: "
+ << reason;
+ }
+ }
+}
+
+void PushDataCallback::onFailure(std::unique_ptr<std::exception> exception) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnFailure, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (pushState_->exceptionExists()) {
+ return;
+ }
+
+ LOG(ERROR) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed for shuffle " << shuffleId_ << " map " << mapId_
+ << " attempt " << attemptId_ << " partition " << partitionId_
+ << " batch " << batchId_ << ", remain revive times "
+ << remainingReviveTimes_;
+
+ if (remainingReviveTimes_ <= 0) {
+ // TODO: set more specific exception.
+ pushState_->setException(std::move(exception));
+ return;
+ }
+
+ if (sharedClient->mapperEnded(shuffleId_, mapId_)) {
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ LOG(INFO) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed but mapper already ended for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_
+ << ", remain revive times " << remainingReviveTimes_ << ".";
+ return;
+ }
+ remainingReviveTimes_--;
+ // TODO: the cause should be extracted from error msg.
+ protocol::StatusCode cause =
+ protocol::StatusCode::PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY;
+ reviveAndRetryPushData(*sharedClient, cause);
+}
+
+void PushDataCallback::updateLatestLocation(
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ pushState_->addBatch(batchId_, latestLocation->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ latestLocation_ = latestLocation;
+}
+
+void PushDataCallback::reviveAndRetryPushData(
+ ShuffleClientImpl& shuffleClient,
+ protocol::StatusCode cause) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ cause);
+ VLOG(1) << "addRequest to reviveManager, shuffleId "
+ << reviveRequest->shuffleId << " mapId " << reviveRequest->mapId
+ << " attemptId " << reviveRequest->attemptId << " partitionId "
+ << reviveRequest->partitionId << " batchId " << batchId_ << " epoch "
+ << reviveRequest->epoch;
+ shuffleClient.addRequestToReviveManager(reviveRequest);
+ long dueTimeMs = utils::currentTimeMillis() +
+ shuffleClient.conf_->clientRpcRequestPartitionLocationRpcAskTimeout() /
+ utils::MS(1);
+ shuffleClient.addPushDataRetryTask(
+ [weakClient = this->weakClient_,
+ shuffleId = this->shuffleId_,
+ body = this->databody_->clone(),
+ batchId = this->batchId_,
+ callback = shared_from_this(),
+ pushState = this->pushState_,
+ reviveRequest,
+ remainingReviveTimes = this->remainingReviveTimes_,
+ dueTimeMs]() {
+ auto sharedClient = weakClient.lock();
Review Comment:
The calling here should not be extracted to a standalone function as the calling is wrapped within a lambda. We must control the parameters in detail to avoid any possible memory leaking problem, and the check calling should be direct, rather than by a function.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562712908
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
Review Comment:
```suggestion
<< " replicate congestion required for shuffle " << shuffleId_
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562716629
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ default: {
+ // This is treated as success.
+ LOG(WARNING) << "unhandled PushData success protocol::StatusCode: "
+ << reason;
+ }
+ }
+}
+
+void PushDataCallback::onFailure(std::unique_ptr<std::exception> exception) {
+ auto sharedClient = weakClient_.lock();
Review Comment:
Could this extract common method for `sharedClient` check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562706730
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
Review Comment:
Is there any test case to verify expire case? Meanwhile, does the expired shuffle client mean that `PushDataCallbackOnSuccess` has already been executed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#issuecomment-3550497577
@SteNicholas @RexXiong @FMX Could you kindly help review this PR? Thanks a lot.
Any suggestion would be greatly appreciated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562943393
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnSuccess, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (response->remainingSize() <= 0) {
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ return;
+ }
+ protocol::StatusCode reason =
+ static_cast<protocol::StatusCode>(response->read<uint8_t>());
+ switch (reason) {
+ case protocol::StatusCode::MAP_ENDED: {
+ auto mapperEndSet = sharedClient->mapperEndSets().computeIfAbsent(
+ shuffleId_,
+ []() { return std::make_shared<utils::ConcurrentHashSet<int>>(); });
+ mapperEndSet->insert(mapId_);
+ break;
+ }
+ case protocol::StatusCode::SOFT_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " soft split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ if (!ShuffleClientImpl::newerPartitionLocationExists(
+ sharedClient->getPartitionLocationMap(shuffleId_).value(),
+ partitionId_,
+ latestLocation_->epoch)) {
+ auto reviveRequest = std::make_shared<protocol::ReviveRequest>(
+ shuffleId_,
+ mapId_,
+ attemptId_,
+ partitionId_,
+ latestLocation_->epoch,
+ latestLocation_,
+ protocol::StatusCode::SOFT_SPLIT);
+ sharedClient->addRequestToReviveManager(reviveRequest);
+ }
+ pushState_->onSuccess(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::HARD_SPLIT: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " hard split required for shuffle " << shuffleId_ << " map "
+ << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ reviveAndRetryPushData(*sharedClient, protocol::StatusCode::HARD_SPLIT);
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_PRIMARY_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ case protocol::StatusCode::PUSH_DATA_SUCCESS_REPLICA_CONGESTED: {
+ VLOG(1) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " primary congestion required for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_ << ".";
+ pushState_->onCongestControl(latestLocation_->hostAndPushPort());
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ break;
+ }
+ default: {
+ // This is treated as success.
+ LOG(WARNING) << "unhandled PushData success protocol::StatusCode: "
+ << reason;
+ }
+ }
+}
+
+void PushDataCallback::onFailure(std::unique_ptr<std::exception> exception) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
+ "PushDataCallbackOnFailure, ignored, shuffle "
+ << shuffleId_ << " map " << mapId_ << " attempt " << attemptId_
+ << " partition " << partitionId_ << " batch " << batchId_
+ << ".";
+ return;
+ }
+ if (pushState_->exceptionExists()) {
+ return;
+ }
+
+ LOG(ERROR) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed for shuffle " << shuffleId_ << " map " << mapId_
+ << " attempt " << attemptId_ << " partition " << partitionId_
+ << " batch " << batchId_ << ", remain revive times "
+ << remainingReviveTimes_;
+
+ if (remainingReviveTimes_ <= 0) {
+ // TODO: set more specific exception.
+ pushState_->setException(std::move(exception));
+ return;
+ }
+
+ if (sharedClient->mapperEnded(shuffleId_, mapId_)) {
+ pushState_->removeBatch(batchId_, latestLocation_->hostAndPushPort());
+ LOG(INFO) << "Push data to " << latestLocation_->hostAndPushPort()
+ << " failed but mapper already ended for shuffle " << shuffleId_
+ << " map " << mapId_ << " attempt " << attemptId_ << " partition "
+ << partitionId_ << " batch " << batchId_
+ << ", remain revive times " << remainingReviveTimes_ << ".";
+ return;
+ }
+ remainingReviveTimes_--;
+ // TODO: the cause should be extracted from error msg.
+ protocol::StatusCode cause =
Review Comment:
In java end's code, the cause is extracted from the error message, which might have dozens of cases. But in cpp side the error message is not aligned, so it might bring problem when extracting the error message. For simplicity here we use PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY as the dummy value, and the TODO is added here for further alignment and debugging purpose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2206][CIP-14] Support PushDataCallback in CppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on code in PR #3543:
URL: https://github.com/apache/celeborn/pull/3543#discussion_r2562900993
##########
cpp/celeborn/client/writer/PushDataCallback.cpp:
##########
@@ -0,0 +1,264 @@
+/*
+ * 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 "celeborn/client/writer/PushDataCallback.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace celeborn {
+namespace client {
+
+std::shared_ptr<PushDataCallback> PushDataCallback::create(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation) {
+ return std::shared_ptr<PushDataCallback>(new PushDataCallback(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ numMappers,
+ numPartitions,
+ mapKey,
+ batchId,
+ std::move(databody),
+ pushState,
+ weakClient,
+ remainingReviveTimes,
+ latestLocation));
+}
+
+PushDataCallback::PushDataCallback(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ int numMappers,
+ int numPartitions,
+ const std::string& mapKey,
+ int batchId,
+ std::unique_ptr<memory::ReadOnlyByteBuffer> databody,
+ std::shared_ptr<PushState> pushState,
+ std::weak_ptr<ShuffleClientImpl> weakClient,
+ int remainingReviveTimes,
+ std::shared_ptr<const protocol::PartitionLocation> latestLocation)
+ : shuffleId_(shuffleId),
+ mapId_(mapId),
+ attemptId_(attemptId),
+ partitionId_(partitionId),
+ numMappers_(numMappers),
+ numPartitions_(numPartitions),
+ mapKey_(mapKey),
+ batchId_(batchId),
+ databody_(std::move(databody)),
+ pushState_(pushState),
+ weakClient_(weakClient),
+ remainingReviveTimes_(remainingReviveTimes),
+ latestLocation_(latestLocation) {}
+
+void PushDataCallback::onSuccess(
+ std::unique_ptr<memory::ReadOnlyByteBuffer> response) {
+ auto sharedClient = weakClient_.lock();
+ if (!sharedClient) {
+ LOG(WARNING) << "ShuffleClientImpl has expired when "
Review Comment:
The expired shuffleClient means that the shuffleClient object itself is destructed and should no longer be accessed. In cpp there is no GC mechanism like JVM, so we must carefully avoid dangling pointer problem (which causes program to crash) and cyclic referencing problem (which causes memory leakage). The solution is by carefully using shared_ptr and weak_ptr. Typically this case comes from releasing resources in abnormal cases, like the shuffle failed for some reason, or the shuffle is cancelled.
The expiration checking is for protecting the function from the pointer dangling problem, and it is not meaningful to mimic the dangling case in UT, though it is easy to do so.
What do you think? I am open to further changes if you like.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org