You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/02/09 06:57:22 UTC
[incubator-celeborn] branch main updated: [CELEBORN-277] PushDataHandle callback could miss soft split status (#1212)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new adb6592d3 [CELEBORN-277] PushDataHandle callback could miss soft split status (#1212)
adb6592d3 is described below
commit adb6592d316b28fdf26cc0364779b64de96c9e6a
Author: Rex(Hui) An <an...@126.com>
AuthorDate: Thu Feb 9 14:57:18 2023 +0800
[CELEBORN-277] PushDataHandle callback could miss soft split status (#1212)
---
.../service/deploy/worker/PushDataHandler.scala | 39 ++++++++++++++--------
1 file changed, 26 insertions(+), 13 deletions(-)
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
index d1ef7998e..74937abd8 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
@@ -285,6 +285,10 @@ class PushDataHandler extends BaseMessageHandler with Logging {
resp.flip()
callbackWithTimer.onSuccess(resp)
} else if (softSplit.get()) {
+ // TODO Currently if the worker is in soft split status, given the guess that the client
+ // will fast stop pushing data to the worker, we won't return congest status. But
+ // in the long term, especially if this issue could frequently happen, we may need to return
+ // congest&softSplit status together
callbackWithTimer.onSuccess(
ByteBuffer.wrap(Array[Byte](StatusCode.SOFT_SPLIT.getValue)))
} else {
@@ -357,22 +361,31 @@ class PushDataHandler extends BaseMessageHandler with Logging {
// The codes here could be executed if
// 1. the client doesn't enable push data to the replica, the master worker could hit here
// 2. the client enables push data to the replica, and the replica worker could hit here
- Option(CongestionController.instance()) match {
- case Some(congestionController) =>
- if (congestionController.isUserCongested(fileWriter.getFileInfo.getUserIdentifier)) {
- if (isMaster) {
- callbackWithTimer.onSuccess(
- ByteBuffer.wrap(
- Array[Byte](StatusCode.PUSH_DATA_SUCCESS_MASTER_CONGESTED.getValue)))
+ // TODO Currently if the worker is in soft split status, given the guess that the client
+ // will fast stop pushing data to the worker, we won't return congest status. But
+ // in the long term, especially if this issue could frequently happen, we may need to return
+ // congest&softSplit status together
+ if (softSplit.get()) {
+ callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte](StatusCode.SOFT_SPLIT.getValue)))
+ } else {
+ Option(CongestionController.instance()) match {
+ case Some(congestionController) =>
+ if (congestionController.isUserCongested(fileWriter.getFileInfo.getUserIdentifier)) {
+ if (isMaster) {
+ callbackWithTimer.onSuccess(
+ ByteBuffer.wrap(
+ Array[Byte](StatusCode.PUSH_DATA_SUCCESS_MASTER_CONGESTED.getValue)))
+ } else {
+ callbackWithTimer.onSuccess(
+ ByteBuffer.wrap(
+ Array[Byte](StatusCode.PUSH_DATA_SUCCESS_SLAVE_CONGESTED.getValue)))
+ }
} else {
- callbackWithTimer.onSuccess(
- ByteBuffer.wrap(Array[Byte](StatusCode.PUSH_DATA_SUCCESS_SLAVE_CONGESTED.getValue)))
+ callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte]()))
}
- } else {
+ case None =>
callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte]()))
- }
- case None =>
- callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte]()))
+ }
}
}